心法利器

本栏目主要和大家一起讨论近期自己学习的心得和体会。具体介绍：仓颉专项：飞机大炮我都会，利器心法我还有。

2024年新的文章合集已经发布，获取方式看这里：再添近20万字-CS的陋室2024年文章合集更新，更有历史文章合集，欢迎下载。

往期回顾

• 心法利器[136] | 算法开荒：大模型版本下高效应对碎片化需求的实战指南

• 心法利器[138] | 大模型如何避免“千人一面”？个性化开发的破局之道

• 心法利器[139] | 算法方案设计：我的规范化框架与避坑要点

• 心法利器[140] | 致校招算法新人：从完成任务到独当一面的成长路径

• 心法利器[141] | 大模型学习的下一站

随着大模型能力逐渐变得可靠，应用也逐步成熟，我们对大模型的信任度也在逐渐变高，苦标注久矣的我们，可以开始相信大模型，通过大模型标注的方式来帮我们完成一些数据的标注，大幅提升我们模型开发迭代的速度。

今天来给大家讲一讲，大模型的自动化标注该怎么做，有什么值得注意的细节。

• 核心流程及要点

• 方案优缺点分析

• 加点代码来讲讲

• • 预标注

  • 标注复核

  • 数据集整理和训练

  • 批跑测试

  • 小结

• 总结

核心流程及要点

这里，我先把整体关键流程先列举一下，然后再来讲思路。

为了简单，我这里以一个简单的文本分类问题为例，来讲具体的操作。

1. 类目体系梳理。梳理好整个类目体系，每个类目体系最好有简单的概念和解释，甚至建议给一些典型的例子。

2. 预标注。构造合适的prompt，然后找一个比较靠谱的大模型，建议最好14B以上的模型，一般而言是越大越好，我自己的经验是比较喜欢用“Qwen3-235B-A22B-Instruct-2507”，当然小一些的类似“Qwen3-30B-A3B-Instruct-2507”也是不错的。很建议用这种MoE的模式，性价比还挺高的。

3. 大模型复核。再用大模型判断，前面预标注的内容是否正确，并让大模型给出新的结果。

4. 人工复核。对预标注和复核结果，再进行一次人工标注看预标注和复核后结果的实际准确情况，如果准确情况不达标，可以对2和3的prompt进行调整，甚至把大模型标注的bad case作为例子加入（预标注不那么怕过拟合），反复2-4步，直到最终标注结果达标。

5. 4的结果进行数据集构造，为大模型训练做准备。

6. 大模型训练。

7. 推理，进行结果评估，如果4的结果足够可靠，那这一步可以直接用标注结果评估，如果信不过，还是得走一遍人工评估。

文字是相对简单的，但是这里的操作细节很多，我来对这里的步骤进行详细解析，包括里面的思路和经验，我个人还是非常建议大家仔细看，吸收消化的，毕竟这里我只是举了个比较简单的例子，后续大家的任务大概率比这个复杂，所以需要大家在此基础上进行修改。

首先，类目梳理这一步，本质上是对文本分类这个任务本身的梳理，毕竟类目做不好，后面所有事都没啥用，而大家在具体的任务场景下，也需要对具体任务有明确的认识，实现最好借助数据多了解了解，具体要做什么，预期结果是什么，给一些例子，这样大模型能更好地理解，这事本质是一个信息传递的过程，把大模型当做人，你要把任务交给他，那就要把任务给他解释清楚。

然后是预标注，这个就比较简单了，把前面的内容梳理完，配合输入信息，都交给大模型就好了。这里一个细节是对大模型的选型，这里一定要选比较靠谱，效果稳定的大模型，资源充足的话最好能大一些，下限高一些的，这里的标注好坏很大程度就靠这个大模型了，所以，嗯，对自己好点。

复核。这一步很容易漏掉，但很重要，不能漏。我说三个原因让大家意识到这个的重要性吧。

• 这个复核，能让你知道，前面的预标注，大概是个什么效果，有个快速的把握，简单的说，差的很远，先别往下整了。千万不要过分相信大模型的所有结果，自动化也不能放任，还是需要监控起来。

• 复核能及时发现标注阶段的问题，包括prompt写的不规范、模糊等的问题，及时发现更新，大幅提升标注质量，毕竟标注效果不好，后面真的什么训练都白搭，甚至可能因为标注不对，最后的准确指标也是假的，上线有极大的风险。

• 第三个是人工复核的。数据我们仍旧是要看的，就借此机会一举多得吧，通过看数据我们能理解具体业务以及我们可能会遇到的难点，这些观察能让我们后续的调优更有目的性。

这里，有大模型复核和人工复核，前者是给后者减轻压力的，虽说标注和复核都是大模型，但是复核和标注在大模型视角看是割裂的，所以相对而言还比较中立，因此质量一般不会太低。

人工复核则不能绕开，不能绕开，不能绕开。即使是大模型标注、复核，也可能存在一些理解的偏差或者偏见，一旦出现这种偏差，标注会出现成片的错误，后续很少有机会会发现，因此，需要自己拿100-200个case再看看，弄一个人工标注的预标注准确率，留意有没有很统一的错误，确认预标注质量，只有达到够高的标准（例如文本分类大概是90%+），才能说这个标注数据集是可靠的（注意，这里哪怕全都是人工标，达到90%都是不简单的）。

有关标注大模型的选型上，有这几个建议。

• 建议用更大的模型，效果真的会更好，标注这个任务对准确程度要求高。

• 我这里推荐的都MoE模型，主要是因为MoE模型的效果相比非MoE模型效果差距小，然而推理性能又比非MoE强，所以很适合。

• 我这里选的都是Instruct模型，并非thinking，主要原因是前者更快，没有别的原因了。这里还有个选择的参考思路，Instruct模型的实际召回率会偏高，而thinking的准确率会偏高，这个是经验值，大家可以根据自己的实际情况进行选择。

• 如果担心效果不好，可以多选几个大模型，分别跑比对着看。

后面的数据集构造、模型训练、测试就是基本功了，并非本文重点就不赘述了，大家在后面的代码大概看看理解一下就行。

方案优缺点分析

本来这章是打算放最后的，但是担心大家代码看不下去就走了，这章又很重要，因此我还是往前提一提。

先说优点。

• 最容易想到的是，就是减少人力成本。从时间上，大模型标注的效率是很高的，尤其是任务困难的（如类目混淆程度高）、长输入的（如对话类）等的问题，大模型优势更为明显，还可以通过提升并发来进一步提速。经验上看，顺利的标注，大概能把一周的标注压缩到1天左右。

• 大模型的标准稳定。人工标注有个很大的缺点，就是随着标注，有一些比较模糊的标准，可能会出现偏离，但大模型不存在，毕竟他没有上下文，每一条数据都是独立的。

• 专业知识稳定。有一些任务需要专业的知识，人工标注可能需要经过一些培训，然而大模型标注不需要，有可能需要在prompt中解释，但相比之下肯定后者简单很多。

• 纠错能力强。标注有一个很痛苦的情况，标到一半才发现某些标准有问题，前面的数需要重标，无论是时间还标注质量都会受到严重影响，人的话需要沟通，还会有情绪，然而大模型只需要改prompt重标即可。

• 有时候我们可能会用一些规则之类的方式标注，这种标注准确率高，但是召回率低，大模型的泛化能力能有效解决这点。

• 代码自动化，开始后就不花费人力了，人力可以释放出来干别的事，提升效率。

然后是缺点。

• 标注要足够质量，基本避不开足够大的模型，成本肯定高，要用更大的模型，那成本就更高。

• 留意召回率相关的指标。前文虽有提到Instruct模型的实际召回率会偏高，但这个偏高是相对thinking而言的，实际召回率仍旧偏低，建议大家多检查召回率，这个指标平时很容易忽略。

• 人工标注量少了以后，大家对数据和实际任务的理解会大幅降低，后续的调优会出现一些隐性的困难，所以我仍旧建议大家还是要做人工复核，借此机会减少这种情况的出现。

加点代码来讲讲

为了更清楚地讲清上面的流程，我用AI写了几段代码方便大家理解。（PS：现在的代码质量说实话已经不低了，大胆用，但如果业务逻辑很复杂，要频繁修改的话，自己写肯定更合适了）

预标注

首先是预标注，就是读文件、拼prompt、请求大模型、存储，完事了。

import json
import os
from openai import OpenAI

# 初始化OpenAI客户端，请替换为您的API密钥和基座URL
client = OpenAI(api_key=os.getenv("API_KEY"), base_url=os.getenv("BASE_URL"))

def classify_query(query, categories):
    """
    使用大模型对单个query进行意图分类
    """
    system_prompt = """你是一名专业的搜索意图分类员。请将用户的搜索Query精确分类到以下类别之一。
类别列表：{}
请只返回类别名称，不要任何解释。""".format("、".join(categories))

    user_prompt = f"需要分类的Query：{query}"

    try:
        response = client.chat.completions.create(
            model="qwen-max",  # 或您选择的模型
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            temperature=0.1,  # 低温度保证输出稳定性
            max_tokens=50,
        )
        result = response.choices[0].message.content.strip()
        # 简单清洗结果，确保返回类别在预设列表中
        for cat in categories:
            if cat in result:
                return cat
        return"其他"# 兜底类别
    except Exception as e:
        print(f"分类出错: {e}，Query: {query}")
        return"其他"

def batch_pre_annotation(input_file, output_file):
    """
    批量预标注主函数
    """
    # 定义分类体系
    intent_categories = ["导航类", "信息类", "交易类", "娱乐类", "教育类", "其他"]

    # 读取未标注数据（假设每行一个query）
    with open(input_file, 'r', encoding='utf-8') as f:
        queries = [line.strip() for line in f if line.strip()]

    results = []
    for i, query in enumerate(queries):
        print(f"Processing {i+1}/{len(queries)}: {query}")
        predicted_category = classify_query(query, intent_categories)
        results.append({
            "query": query,
            "pre_annotated_category": predicted_category,
            "review_status": "pending"# 标记为待复核
        })

    # 保存预标注结果
    with open(output_file, 'w', encoding='utf-8') as f:
        json.dump(results, f, ensure_ascii=False, indent=2)
    print(f"预标注完成！结果保存至：{output_file}")

if __name__ == "__main__":
    batch_pre_annotation("unlabeled_queries.txt", "pre_annotated.json")

标注复核

复核也比较简单，读取预标注结果，请求大模型，更新标注结果，保存。

import json

def review_annotation(item):
    """
    复核单条标注结果
    """
    query = item["query"]
    pre_annotated_category = item["pre_annotated_category"]

    review_prompt = f"""
请扮演一个数据标注质量审核员的角色。你需要审核以下搜索Query的意图分类结果。

原始Query：{query}
预标注类别：{pre_annotated_category}

请判断这个分类是否正确。
如果正确，请回复：正确
如果不正确，请回复：错误，并给出你认为正确的类别。

请严格按以下格式回复：
判断结果：正确/错误
建议类别：（如果错误则填写，否则留空）
"""
    try:
        response = client.chat.completions.create(
            model="qwen-max",
            messages=[{"role": "user", "content": review_prompt}],
            temperature=0.1,
        )
        review_text = response.choices[0].message.content.strip()
        
        # 解析大模型的回复
        if"判断结果：正确"in review_text:
            item["review_status"] = "correct"
            item["final_category"] = pre_annotated_category
        elif"判断结果：错误"in review_text:
            item["review_status"] = "incorrect"
            # 尝试提取建议类别
            lines = review_text.split('\n')
            for line in lines:
                if"建议类别："in line and line != "建议类别：":
                    suggested_cat = line.replace("建议类别：", "").strip()
                    item["suggested_category"] = suggested_cat
                    item["final_category"] = suggested_cat  # 采用建议
                    break
            # 如果未解析出建议类别，则保持原分类但标记为错误，需要人工介入
            if"suggested_category"notin item:
                item["final_category"] = pre_annotated_category
                item["needs_manual_review"] = True
        else:
            # 如果回复格式不符合预期，标记为需人工复核
            item["review_status"] = "unclear"
            item["final_category"] = pre_annotated_category
            item["needs_manual_review"] = True

        item["review_comment"] = review_text

    except Exception as e:
        print(f"复核出错: {e}")
        item["review_status"] = "error"
        item["final_category"] = pre_annotated_category

    return item

def batch_review():
    with open("pre_annotated.json", 'r', encoding='utf-8') as f:
        pre_annotated_data = json.load(f)

    reviewed_data = []
    for item in pre_annotated_data:
        reviewed_item = review_annotation(item)
        reviewed_data.append(reviewed_item)

    with open("reviewed_annotations.json", 'w', encoding='utf-8') as f:
        json.dump(reviewed_data, f, ensure_ascii=False, indent=2)
    print("标注复核完成！")

if __name__ == "__main__":
    batch_review()

人工复核的是就自己整理数据然后人工标吧，不涉及代码这里就不写了，但注意这个工作还是得做！

数据集整理和训练

这里，要把数据整理成llama-factory适用的数据，这个就非常简单了。

import json
from sklearn.model_selection import train_test_split

def convert_to_llama_factory_format():
    with open("reviewed_annotations.json", 'r', encoding='utf-8') as f:
        reviewed_data = json.load(f)

    # 过滤出已复核且可用的数据
    usable_data = [item for item in reviewed_data if item.get("review_status") in ["correct", "incorrect"] andnot item.get("needs_manual_review")]

    formatted_data = []
    for item in usable_data:
        # 构建LLaMA-Factory所需的指令微调格式
        formatted_item = {
            "instruction": "请对以下搜索Query进行意图分类。",
            "input": item["query"],
            "output": item["final_category"]
        }
        formatted_data.append(formatted_item)

    # 数据集划分 (8:1:1)
    train_val, test = train_test_split(formatted_data, test_size=0.1, random_state=42, stratify=[d['output'] for d in formatted_data])
    train, val = train_test_split(train_val, test_size=0.1, random_state=42, stratify=[d['output'] for d in train_val])

    # 保存数据集
    with open('data/train.json', 'w', encoding='utf-8') as f:
        json.dump(train, f, ensure_ascii=False, indent=2)
    with open('data/val.json', 'w', encoding='utf-8') as f:
        json.dump(val, f, ensure_ascii=False, indent=2)
    with open('data/test.json', 'w', encoding='utf-8') as f:
        json.dump(test, f, ensure_ascii=False, indent=2)

    print(f"数据整理完成！训练集：{len(train)}条，验证集：{len(val)}条，测试集：{len(test)}条")

if __name__ == "__main__":
    convert_to_llama_factory_format()

整理完就可以开始进行训练了。这里我选用llama-factory，最简单的方式便是配置yaml，然后用命令行就能跑了，这里选的是lora，当然大家可以根据自己的情况改成别的方法。

# config.yaml 示例
model_name_or_path:Qwen/Qwen3-4B-Instruct-2507# 基座模型
dataset_dir:data
dataset:train.json,val.json# 数据文件
template:qwen
finetuning_type:lora
lora_target:q_proj,v_proj

output_dir:./output
per_device_train_batch_size:4
gradient_accumulation_steps:4
lr_scheduler_type:cosine
learning_rate:1.0e-4
num_train_epochs:3
fp16:true

logging_steps:10
eval_steps:100
save_steps:200

然后命令行跑起来。

# 在LLaMA-Factory项目根目录下执行
conda activate llama_factory
python src/train_bash.py --config config.yaml

批跑测试

批跑测试用的vllm。（省略了合并lora权重的过程，不是本文重点不赘述了）

from vllm import LLM, SamplingParams
import json
from sklearn.metrics import accuracy_score

# 初始化vLLM引擎，加载微调后的模型
# 假设您已使用LLaMA-Factory的export_model.py脚本将LoRA权重与基座模型合并，并保存到了 './merged_model' 路径
llm = LLM(model="./merged_model", tensor_parallel_size=1) # tensor_parallel_size根据GPU数量调整

# 配置生成参数
sampling_params = SamplingParams(temperature=0.01, top_p=0.9, max_tokens=50)

def create_prompt(input_text):
    """创建与训练时一致的提示模板"""
    returnf"""请对以下搜索Query进行意图分类。
    
Query: {input_text}
意图类别："""

def predict_category(query):
    """预测单个query的类别"""
    prompt = create_prompt(query)
    outputs = llm.generate([prompt], sampling_params)
    predicted_text = outputs[0].outputs[0].text.strip()
    return predicted_text

def evaluate_model():
    """在测试集上评估模型"""
    with open('data/test.json', 'r', encoding='utf-8') as f:
        test_data = json.load(f)

    true_labels = []
    predicted_labels = []

    for item in test_data:
        true_label = item['output']
        predicted_label = predict_category(item['input'])

        true_labels.append(true_label)
        predicted_labels.append(predicted_label)

        print(f"Query: {item['input']}")
        print(f"真实: {true_label} -> 预测: {predicted_label}")

    # 计算准确率
    accuracy = accuracy_score(true_labels, predicted_labels)
    print(f"\n=== 最终测试结果 ===")
    print(f"整体准确率 (ACC): {accuracy:.4f}")

    # 保存详细结果
    results = {
        "accuracy": accuracy,
        "details": [{"query": test_data[i]["input"], "true": true_labels[i], "predicted": predicted_labels[i]} for i in range(len(test_data))]
    }
    with open('test_results.json', 'w', encoding='utf-8') as f:
        json.dump(results, f, ensure_ascii=False, indent=2)

if __name__ == "__main__":
    evaluate_model()

小结

可以看到，整个流程基本上代码一条龙，还是非常方便的。

总结

大模型标注这个事，从大模型刚出来那会就已经有人尝试，后续也有了不少的打磨，应该已经有不少的人在用了，实用性也是挺不错的，这次的内容梳理，把核心点和通用的部分分享给大家，希望能对大家有所启发。