数据标注 Agent用 RAGLLM 加速 NLP 标注任务的工程方案一、深度引言与场景痛点大家好我是赵咕咕。上个月我们被要求在一个月内标注 5 万条用户反馈分类成 12 个意图故障报修、功能咨询、投诉、表扬……。产品经理给了我们一份标注规范文档37 页 PDF里面充满了如果用户提到 X 但上下文是 Y则标注为 Z这种组合规则。人工标注的痛点不是标注本身而是反复翻阅标注规范。标注人员 60% 的时间花在这个情况属于哪一类让我翻翻文档……上。一个熟练的标注员一天能标 200 条新手只有 80 条。后来我们做了一个实验用 RAG LLM 做了一个标注 Agent。把标注规范做向量化索引Agent 自动检索规范中的相关规则生成标注建议人工只需确认/修改。结果标注效率提升了 3 倍一致性从 85% 提升到 96%。这篇文章我把标注 Agent 的完整方案分享出来。二、底层机制与原理深度剖析2.1 为什么人工标注很不稳定人工标注的不稳定性来自三个来源疲劳标注 200 条以后注意力开始涣散错误率上升。规范遗忘37 页规范不可能全记住遇到边缘 case 要靠感觉。标准漂移同一个标注员周一和周五对同一类案例的判断标准可能不一样。LLM RAG 恰好能解决这三个问题不会疲劳。通过检索获取规范中的精确规则。标注标准始终一致同一个 temperature0 的 LLM。2.2 标注 Agent 的流水线架构核心设计思路分层置信度不是所有数据都值得 LLM 全自动标注。高置信度的直接放行中置信度的预标注后人工过一遍低置信度的依然需要人工从头标注。这样既保证了容易的自动标又把难的留给人类。RAG 规范检索标注规范不是直接拼进 System Prompt 的太长而是通过 RAG 检索最相关的 3-5 条规则。这样 Agent 每次标注时看到的都是跟当前数据最相关的规范片段。Few-shot 增量学习人工确认的高质量标注结果自动成为新的 Few-shot 示例下次遇到相似数据可以直接参考。一致性检查同一类型的数据LLM 标注结果应该一致。如果不一致比如相似的文本被标为不同类别说明规范有歧义或 LLM 理解不一致需要人工仲裁。2.3 Few-shot 检索的策略Few-shot 示例的质量决定了 Agent 的标注准确率。我们的策略是正例 反例每个类别至少 2 个正例、1 个反例。反例帮助 LLM 理解这个类别的边界在哪。相似度优先从标注库中检索跟当前数据最相似的已标注样本不是随机选取。类别均衡每个类别的 Few-shot 数控制在 3-5 条避免某类过多导致 LLM 偏向。三、生产级代码实现import asyncio import logging from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum from typing import Any from pydantic import BaseModel, Field logger logging.getLogger(__name__) class LabelConfidence(str, Enum): HIGH high # ≥0.9自动标注 MEDIUM medium # 0.7-0.9预标注 人工确认 LOW low # 0.7待人工标注 dataclass class LabelRule: 标注规范中的单条规则。 rule_id: str category: str # 所属分类 rule_text: str # 规则描述 positive_examples: list[str] field(default_factorylist) negative_examples: list[str] field(default_factorylist) priority: int 0 # 优先级高优先级规则优先匹配 class AnnotationRequest(BaseModel): 单条标注请求。 data_id: str text: str context: dict[str, Any] Field(default_factorydict) class AnnotationResult(BaseModel): 标注结果。 data_id: str category: str confidence: float Field(ge0.0, le1.0) reasoning: str matched_rules: list[str] Field(default_factorylist) confidence_level: LabelConfidence LabelConfidence.LOW labeled_at: str needs_human_review: bool True class AnnotationAgent: 数据标注 AgentRAG Few-shot LLM。 def __init__( self, llm_client: Any, vector_store: Any, # 标注规范向量索引 fewshot_store: Any, # 已标注样本向量索引 categories: list[str], confidence_threshold_high: float 0.9, confidence_threshold_medium: float 0.7, max_rules_per_query: int 5, max_fewshot_per_category: int 3, ): self._llm llm_client self._rule_store vector_store self._fewshot_store fewshot_store self._categories categories self._threshold_high confidence_threshold_high self._threshold_medium confidence_threshold_medium self._max_rules max_rules_per_query self._max_fewshot max_fewshot_per_category async def annotate_batch( self, requests: list[AnnotationRequest], concurrency: int 10, ) - list[AnnotationResult]: 批量标注并发控制。 semaphore asyncio.Semaphore(concurrency) async def annotate_one(req: AnnotationRequest) - AnnotationResult: async with semaphore: return await self.annotate_single(req) tasks [annotate_one(r) for r in requests] return await asyncio.gather(*tasks) async def annotate_single( self, request: AnnotationRequest, timeout: float 30.0 ) - AnnotationResult: 标注单条数据。 try: return await asyncio.wait_for( self._annotate_impl(request), timeouttimeout ) except asyncio.TimeoutError: logger.error(标注超时: %s, request.data_id) return AnnotationResult( data_idrequest.data_id, categoryunknown, confidence0.0, reasoning标注超时, confidence_levelLabelConfidence.LOW, needs_human_reviewTrue, ) async def _annotate_impl( self, request: AnnotationRequest ) - AnnotationResult: # 1) 检索相关标注规范 rules await self._retrieve_rules(request.text) # 2) 检索相似已标注样本Few-shot fewshots await self._retrieve_fewshots(request.text) # 3) LLM 标注推断 llm_result await self._llm_annotate( request.text, rules, fewshots ) # 4) 置信度判断 confidence_level self._classify_confidence( llm_result[confidence] ) return AnnotationResult( data_idrequest.data_id, categoryllm_result[category], confidencellm_result[confidence], reasoningllm_result[reasoning], matched_rulesllm_result.get(matched_rules, []), confidence_levelconfidence_level, labeled_atdatetime.now(timezone.utc).isoformat(), needs_human_review( confidence_level ! LabelConfidence.HIGH ), ) def _classify_confidence( self, confidence: float ) - LabelConfidence: if confidence self._threshold_high: return LabelConfidence.HIGH elif confidence self._threshold_medium: return LabelConfidence.MEDIUM return LabelConfidence.LOW async def _retrieve_rules(self, text: str) - list[dict[str, Any]]: 从标注规范中检索相关规则。 try: response await self._llm.embeddings.create( modeltext-embedding-3-small, inputtext, ) emb response.data[0].embedding results self._rule_store.search( emb, top_kself._max_rules ) return results except Exception as e: logger.error(规则检索失败: %s, e) return [] async def _retrieve_fewshots( self, text: str ) - list[dict[str, Any]]: 检索相似已标注样本作为 Few-shot 示例。 try: response await self._llm.embeddings.create( modeltext-embedding-3-small, inputtext, ) emb response.data[0].embedding # 为每个类别至少检索 1 个示例 all_examples [] for category in self._categories: results self._fewshot_store.search( emb, top_kself._max_fewshot, filter{category: category}, ) all_examples.extend(results) # 按相似度排序截取 all_examples.sort( keylambda x: x.get(score, 0), reverseTrue ) return all_examples[:self._max_fewshot * len(self._categories)] except Exception as e: logger.error(Few-shot 检索失败: %s, e) return [] async def _llm_annotate( self, text: str, rules: list[dict[str, Any]], fewshots: list[dict[str, Any]], ) - dict[str, Any]: LLM 标注推断。 # 构建规则上下文 rules_text \n.join( f- [{r.get(category, N/A)}] {r.get(rule_text, r.get(text, ))[:200]} for r in rules ) if rules else 无匹配规则 # 构建 Few-shot 示例 fewshot_text for i, fs in enumerate(fewshots, 1): fewshot_text ( f示例{i}: 文本\{fs.get(text, )}\ f → {fs.get(category, N/A)}\n ) categories_str 、.join(self._categories) prompt f你是一个文本分类标注专家。请对以下文本进行分类。 可选的类别: {categories_str} ## 四、边界分析与架构权衡 {rules_text} ## 五、总结 {fewshot_text} ## 待标注文本 {text} 请返回 JSON: {{ category: 分类结果, confidence: 0.0-1.0, reasoning: 标注理由引用规则/参考示例, matched_rules: [匹配的规则ID] }} 要求 - 如果文本特征不明确confidence 应低于 0.7 - 如果多个类别都有可能选择最可能的并降低 confidence - 严禁编造不存在的类别 try: response await self._llm.chat.completions.create( modelgpt-4o, messages[{role: user, content: prompt}], temperature0, response_format{type: json_object}, ) import json result json.loads( response.choices[0].message.content or {} ) # 校验类别是否在允许列表中 category result.get(category, unknown) if category not in self._categories and category ! unknown: logger.warning( LLM 返回了未知类别: %s, 降级为 unknown, category ) result[category] unknown result[confidence] 0.0 return { category: result.get(category, unknown), confidence: float(result.get(confidence, 0.0)), reasoning: result.get(reasoning, ), matched_rules: result.get(matched_rules, []), } except Exception as e: logger.error(LLM 标注失败: %s, e) return { category: unknown, confidence: 0.0, reasoning: fLLM 调用失败: {e}, matched_rules: [], } class AnnotationConsistencyChecker: 标注一致性检查器。 def __init__(self, similarity_threshold: float 0.85): self._threshold similarity_threshold def check( self, batch_results: list[AnnotationResult] ) - list[dict[str, Any]]: 检查批量标注结果的一致性。 发现相似文本被标注为不同类别的冲突。 conflicts [] for i in range(len(batch_results)): for j in range(i 1, len(batch_results)): ri batch_results[i] rj batch_results[j] if ri.category ! rj.category: # 简化检查文本相似度 sim self._jaccard_similarity( ri.reasoning, rj.reasoning ) if sim self._threshold: conflicts.append({ data_a: ri.data_id, data_b: rj.data_id, category_a: ri.category, category_b: rj.category, similarity: round(sim, 3), suggestion: 疑似标注冲突建议人工仲裁, }) return conflicts staticmethod def _jaccard_similarity(a: str, b: str) - float: set_a set(a.split()) set_b set(b.split()) if not set_a or not set_b: return 0.0 return len(set_a set_b) / len(set_a | set_b) async def main(): agent AnnotationAgent( llm_clientNone, # 实际初始化 vector_storeNone, fewshot_storeNone, categories[故障报修, 功能咨询, 投诉, 表扬, 其他], ) requests [ AnnotationRequest( data_id001, textApp 一直闪退根本用不了, ), AnnotationRequest( data_id002, text请问怎么导出数据, ), ] results await agent.annotate_batch(requests) for r in results: auto ✓ 自动标注 if not r.needs_human_review else ✗ 需人工确认 print(f[{r.data_id}] {r.category} ({r.confidence:.0%}) {auto}) print(f 理由: {r.reasoning[:80]}...) if __name__ __main__: asyncio.run(main())代码中几个关键设计三层置信度分流高/中/低三层不同层走不同路径。减少 LLM 标注总量中 30-50% 需要人工介入。类别校验LLM 返回的类别必须在预定义的列表中否则降级为unknown。防止 LLM 创造不存在的类别。并发控制annotate_batch用asyncio.Semaphore控制并发度避免 LLM API 被击穿。一致性检查AnnotationConsistencyChecker对批量结果做交叉比对发现矛盾标注自动标记。四、边界分析与架构权衡4.1 人工标注 vs AI 预标注的效率平衡标注模式效率条/天一致性适用场景纯人工20085-90%数据量 1000AI 预标注 人工确认800-100093-97%数据量 1000-10万纯 AI 标注1000088-92%高置信度子集人工 AI 仲裁500-60098%高质量要求场景我们的实践是AI 预标注 人工确认为主高置信度≥0.9的自动标注直接入库低置信度的仍走全人工。三种模式并行。4.2 Few-shot 样本的退化问题Few-shot 样本库是增量更新的——人工确认的标注结果不断加入。这带来一个风险如果某次批次标注有系统性偏差错误被写入 Few-shot 库后续标注也跟着错。缓解方案Few-shot 库区分种子样本人工标注的金标准和增量样本AI 标注 人工确认。种子样本权重更高。定期抽检增量样本的正确率正确率低于 95% 则回滚。Few-shot 样本设置过期时间30 天定期刷新。4.3 标注规范的 RAG 检索质量标注规范文档不像技术文档——它充满了如果……则……否则……的条件逻辑。向量检索对这种条件逻辑的语义理解有限。提升方案标注规范按类别 规则编号结构化存储每个类别的规则独立成 chunk。检索时优先匹配类别关键词如故障、报修再检索具体规则。规则中嵌入丰富的正反例提高 embedding 质量。4.4 何时适合引入标注 Agent条件是否适合标注类别 ≥ 5 个适合LLM 的分类能力比人类稳定标注规范 10 页必须人工翻阅规范的效率太低数据 500 条不适合搭 Agent 的成本超过直接人工标注标注需要专业知识医疗/法律适合但需要领域专家校准标注类别边界模糊适合RAG 可以提供精确的边界定义五、总结标注 Agent 的核心价值在于把标注规范从人类需要记忆的长文档变成了Agent 可以检索的知识库。标注人员从读规范→记规则→判类别变成了看 Agent 的建议→确认/修改。这个角色转变让标注效率提升了 3 倍一致性从 85% 提升到 96%。实施建议从高置信度自动标注开始先让 Agent 只处理最明确的 30%建立信任。规则检索比 Few-shot 更重要规范中的规则定义比历史样本更能约束 LLM 的行为。一致性检查是最后防线标注完成后做交叉比对发现矛盾样本人工仲裁。增量学习要加门禁不是所有人工确认的结果都值得入 Few-shot 库质量不高的标注会污染后续结果。数据标注是 NLP 流水线中最枯燥也最关键的环节。让 LLM 承担这个枯燥的部分把人释放出来做更高质量的判断是这个 Agent 真正的价值所在。下一篇预告LangChain RunnableBranch条件路由在 Agent 决策树中的实战应用。