agno 多智能体质量把关Quality Review实战Labeler–Reviewer–Adjudicator 质检工作流解析【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇技术指南以 agno 数据标注 Cookbook 中_18_quality_review目录为核心讲解如何在任意抽取原语文本、图片、音频、文档抽取之上叠加一套双标注 分歧审查 仲裁的多智能体质检流水线两个使用不同模型提供商的标注 Agent 并行独立抽取Reviewer Agent 逐字段比对找出分歧Adjudicator Agent 仅在存在分歧时基于原始输入裁决。读完本篇你将掌握这套工作流在 basic.py 中的完整实现、agnoWorkflow的Parallel/Condition原语用法以及如何把它复用到你自己的高价值标注任务上。一、什么是 Quality Review给标注加一道质检闸门质量把关Quality Review是数据标注流水线中一种多智能体质控模式两个标注器labeler使用不同模型提供商独立抽取同一输入一个审查器reviewer逐字段比对找出分歧一个仲裁器adjudicator在出现分歧时对照原始输入给出最终结论。在 agno 的标注 Cookbook 中该模式被表达为一个 agnoWorkflow拓扑为Parallel(labeler_a, labeler_b) → reviewer → Condition(adjudicator)两个 labeler 运行在Parallel(...)中并发执行、互不干扰reviewer 对二者的输出逐字段 diff一个Condition步骤仅在 reviewer 标记存在分歧时才运行 adjudicator。它被设计为叠加在任何抽取原语之上的通用质检层文本抽取_03_text_extraction的 Contact schema、图片、音频、文档抽取均可复用同一形态见 README.md。适用场景When to use原文档 README.md 明确给出了三类典型场景高价值标签错误答案代价高昂wrong answer is expensive的任务多智能体互相校验值得付出额外算力构建评测/训练集分歧本身即信号——标注器在哪些字段上不一致恰好暴露了输入文本的歧义点受监管工作负载需要可审计的裁决轨迹auditable resolution trail每一步谁抽了什么、谁判了什么、最终如何裁决都有记录。如果只需要单遍抽取直接用对应的*_extraction/Cookbook 即可如果要做的是面向评测的模型集成provider ensembling for evals而非生产标签则应参考_17_llm_as_judge/。二、工作流拓扑与核心设计2.1 三阶段流水线┌─────────────────┐ 原始输入 ────► │ Parallel 标注 │ │ Labeler A (G) │──┐ │ Labeler B (C) │──┤ └─────────────────┘ │ ▼ ┌─────────────────────┐ │ Reviewer 逐字段diff │ │ needs_adjudication? │ └─────────────────────┘ │ needs_adjudicationFalse needs_adjudicationTrue │ │ ▼ ▼ 跳过仲裁步骤 ┌──────────────────┐ │ Adjudicator 仲裁 │ │ 对照原始输入裁决 │ └──────────────────┘关键设计决策模型提供商差异化两个 labeler 分别跑在google:gemini-3.5-flash与anthropic:claude-opus-4-7上以获得集成多样性ensemble diversity——不同模型对同一文本的解读盲区不完全重合分歧更有信息量分歧才仲裁adjudicator 由Condition门控无分歧时整条流水线只花两次标注 一次审查的代价仲裁只在真正必要时触发兼顾质量与吞吐全程落库workflow 挂载SqliteDb每次运行都持久化构成可审计的裁决轨迹。2.2 运行时测试依据对应测试记录 TEST_LOG.md 记载2026-07-18 测试agno 2.7.4干净输入两个 labeler 返回完全一致的ContactnameLiam Ortega、emailliammeadow.io、companyMeadow、titleSupport Engineerreviewer 判定needs_adjudicationFalsestep trail 显示 Condition 跳过——Condition Adjudicate not met - skipped 1 steps冲突输入两个 labeler 在 nameDr. Sarah Chen-Watanabe vs Sarah Chen与 titlePrincipal Scientist and acting Head of Platform vs Principal Scientist上产生分歧reviewer 逐字段给出原因adjudicator 介入并输出最终FinalLabel(contactContact(nameSarah Chen, emails.chennova-labs.io, phone1-415-555-0177, companyNovaLabs, titlePrincipal Scientist and acting Head of Platform))。测试同时注明哪些字段会分歧因运行而异但冲突输入被构造为至少有一个字段可靠地产生分歧从而保证两条路径skip 路径与仲裁路径都能被稳定覆盖。三、数据结构设计四个 Pydantic Schema整个流水线的契约由 basic.py 中四个 Pydantic 模型定义Schema字段职责Contactname / email / phone / company / title均可空标注目标本体继承自_03_text_extraction的 Contact 结构FieldDisagreementfield、value_a、value_b、reason单个字段的分歧描述reason说明为何需要仲裁DisagreementReportdisagreements 列表、needs_adjudicationreviewer 的输出needs_adjudicationTrue表示存在任意字段分歧FinalLabelcontact、notes可选仲裁器的最终标签class Contact(BaseModel): name: Optional[str] None email: Optional[str] None phone: Optional[str] None company: Optional[str] None title: Optional[str] None class FieldDisagreement(BaseModel): field: str Field(..., descriptionTop-level Contact field name) value_a: Optional[str] None value_b: Optional[str] None reason: str Field(..., descriptionWhy this field needs adjudication) class DisagreementReport(BaseModel): disagreements: List[FieldDisagreement] Field(default_factorylist) needs_adjudication: bool Field(..., descriptionTrue if any field disagrees) class FinalLabel(BaseModel): contact: Contact notes: Optional[str] None这些模型同时充当智能体输出契约每个 Agent 的output_schema指向对应模型模型输出会被强制解析为结构化对象这为 reviewer 的逐字段 diff和 adjudicator 的对照原始输入裁决提供了机器可读的数据基础。四、Agent 与指令设计四个角色各司其职4.1 标注器Labeler A / BLABELER_INSTRUCTIONS \ Extract contact information from the input. Use exactly what the text shows. If a field is missing, leave it null. Do not guess. labeler_a Agent( nameLabeler A, modelgoogle:gemini-3.5-flash, instructionsLABELER_INSTRUCTIONS, output_schemaContact, ) labeler_b Agent( nameLabeler B, modelanthropic:claude-opus-4-7, instructionsLABELER_INSTRUCTIONS, output_schemaContact, )指令要点是原文照录、缺失置空、禁止猜测——这保证了分歧不是凭空捏造而是真实文本歧义如昵称、改名、复合头衔、过时联系方式的投影。运行前提需要同时配置GOOGLE_API_KEY与ANTHROPIC_API_KEY见 README.md。4.2 审查器Reviewerreviewer Agent( nameReviewer, modelanthropic:claude-opus-4-7, instructions\ You are given two labelers Contact outputs. Compare them field by field. A field needs adjudication when both labelers report non-null but different values. Emit one FieldDisagreement per such field. Set needs_adjudicationtrue if any field needs adjudication. , output_schemaDisagreementReport, )注意分歧判定规则的精确定义只有双方都非空non-null但取值不同的字段才需要仲裁——单方缺失null不算分歧这避免了把一方漏抽误判为需要仲裁。测试日志也验证了该规则干净输入双方一致时 reviewer 输出needs_adjudicationFalse。4.3 仲裁器Adjudicatoradjudicator Agent( nameAdjudicator, modelanthropic:claude-opus-4-7, instructions\ Re-read the original input text and resolve every reported disagreement. Return a FinalLabel.contact populated with the correct values for all fields (use the agreed values for fields not in dispute). , output_schemaFinalLabel, )仲裁器拥有最完整的上下文原始输入 两个 labeler 输出 reviewer 报告。对争议字段对照原文裁决对无争议字段沿用双方一致值。这正是测试日志中冲突输入裁决结果 nameSarah Chen采用短名偏好而 title 保留完整复合头衔的原因。五、Step 与 Workflow 组装从 Agent 到可运行流水线5.1 普通 Agent 步骤两个 labeler 直接封装为普通步骤label_a Step(nameLabeler A, agentlabeler_a) label_b Step(nameLabeler B, agentlabeler_b)5.2 自定义执行器Executor把多步骤输出拼进 Promptreviewer 与 adjudicator 都需要不止上一步的输出因此用自定义executor函数手动装配 prompt。关键在于StepInput.get_step_output()def run_reviewer(step_input: StepInput) - StepOutput: a step_input.get_step_output(Labeler A).content b step_input.get_step_output(Labeler B).content prompt ( fLabeler A:\n{a.model_dump_json(indent2)}\n\n fLabeler B:\n{b.model_dump_json(indent2)} ) report reviewer.run(prompt).content return StepOutput(contentreport)源码层面StepInput.get_step_output()在 workflow/types.py 中实现先按步骤名直接查找若失败则递归下钻嵌套步骤Parallel / Condition / Router / Loop / Steps这正是它能找到位于Parallel块内部的Labeler A/Labeler B输出的原因。adjudicator 的 executor 进一步把三路信息组装进 promptdef run_adjudicator(step_input: StepInput) - StepOutput: a step_input.get_step_output(Labeler A).content b step_input.get_step_output(Labeler B).content report step_input.get_step_output(Reviewer).content prompt ( fOriginal input:\n{step_input.input}\n\n fLabeler A:\n{a.model_dump_json(indent2)}\n\n fLabeler B:\n{b.model_dump_json(indent2)}\n\n fReviewer report:\n{report.model_dump_json(indent2)} ) final adjudicator.run(prompt).content return StepOutput(contentfinal)5.3 Condition只有分歧才仲裁def has_disagreement(step_input: StepInput) - bool: report step_input.previous_step_content return bool(report and getattr(report, needs_adjudication, False)) adjudicate Step(nameAdjudicator, executorrun_adjudicator)随后在 Workflow 中用Condition(evaluatorhas_disagreement, steps[adjudicate])门控。源码层面workflow/condition.pyCondition的evaluator支持三种形态可调用函数本示例所用返回 bool布尔字面量True/FalseCEL 表达式字符串可访问input、previous_step_content、previous_step_outputs、additional_data、session_state等变量例如previous_step_outputs.research.contains(error)。当条件不满足且未提供else_steps时Condition 会返回一条not met消息并跳过其子步骤见 condition.py——测试日志中的Condition Adjudicate not met - skipped 1 steps正是这条路径。5.4 Workflow 组装与持久化workflow Workflow( nameQuality review labeling, dbSqliteDb(db_filetmp/labeling.db), # every run is persisted steps[ Parallel(label_a, label_b, nameLabel), # labelers run concurrently review, # diff them field by field Condition( # adjudicate only on disagreement nameAdjudicate, evaluatorhas_disagreement, steps[adjudicate], ), ], )Parallel原语workflow/parallel.py接受可变参数强调各子步骤独立且无序支持Parallel(step1, step2, namemy_parallel)或名称置首的调用约定。SqliteDb(db_filetmp/labeling.db)让每次运行含每个步骤的输入输出持久化到本地 SQLite为受监管场景提供审计依据。六、运行方式与两条路径的实测解读6.1 运行命令python cookbook/data_labeling/_18_quality_review/basic.py前置条件配置GOOGLE_API_KEY与ANTHROPIC_API_KEY两个 labeler 使用不同提供商以保证集成多样性。6.2 两条测试输入脚本构造了两个对照输入# 干净输入双方预期一致 clean ( Liam Ortega is a Support Engineer at Meadow and can be reached at liammeadow.io. ) # 冲突输入刻意埋入歧义 conflicting ( Forwarded note: you can reach Dr. Sarah Chen-Watanabe (she goes by Sarah Chen) about the platform work. Sarah is Principal Scientist and acting Head of Platform at NovaLabs, which recently rebranded from Nova Biotech. Email s.chennova-labs.io. The 555-0123 number on the website is stale; her direct line is 1-415-555-0177. )冲突输入按构造产生分歧两个电话号码、改名后的公司、复合头衔、偏好的短名——独立标注器至少会在一个字段上产生不同序列化结果从而稳定触发仲裁路径。6.3 结果打印与 step traildef print_step_outputs(step_results) - None: Walk the step trail, descending into Parallel / Condition children. for step in step_results: if step.steps: print_step_outputs(step.steps) elif step.content is not None: pprint({step: step.step_name, output: step.content})该递归函数会下钻 Parallel / Condition 的子步骤把整条流水线的执行轨迹完整打印出来——这在质检场景中就是可审计的裁决轨迹的可视化形式。依据 TEST_LOG.md 的实测skip 路径干净输入下双方 Contact 完全一致 → reviewerneeds_adjudicationFalse→ trail 出现Condition Adjudicate not met - skipped 1 steps仲裁步骤未执行仲裁路径冲突输入下 name / title 分歧被 reviewer 逐字段标记 → adjudicator 基于原始输入裁决出FinalLabel保留复合头衔、采用短名、采用直拨电话trail 完整记录每一步。七、复用与扩展把质检闸门装到任意抽取原语上这套工作流的复用方式非常直接换掉 labeler agent 与 schema即可把同样的质检闸门套到本目录下任意其他原语图片、音频、文档抽取上README.md。具体步骤替换 schema将Contact换成目标任务的输出模型如_03_text_extraction之外的其他抽取结果结构并同步调整FieldDisagreement的field描述与 reviewer / adjudicator 的指令替换 labeler保持不同提供商原则把model换成你拥有的 API 组合如 OpenAI Anthropic、OpenAI Gemini并保留原文照录、缺失置空的指令风格保持拓扑Parallel双标注 → reviewer 逐字段 diff →Condition门控仲裁 的三段式结构无需改动按需升级门控若某类输入必须仲裁可将Condition的evaluator换成 CEL 表达式或布尔值需要人工复核时Condition还支持requires_confirmationTrue的 HITL 模式见 condition.py。八、底层原理小结这篇代码为什么这样组织从 basic.py 到 agno 源码可以梳理出四条支撑本工作流的底层机制机制源码位置作用Step统一抽象workflow/step.pyAgent / 自定义 executor 均可成为工作流单元且支持嵌套Parallel并发workflow/parallel.py强调步骤独立无序两个标注器并发执行Condition门控workflow/condition.pyevaluator 支持函数 / 布尔 / CEL条件不满足时输出 not met 并跳过StepInput.get_step_output递归查找workflow/types.py跨嵌套层级按名取任意前序步骤输出支撑 reviewer / adjudicator 的自定义 prompt 装配SqliteDb持久化agno.db.sqlite.SqliteDb每次运行全量落库形成可审计轨迹这套组合回答了一个核心工程问题当质量比吞吐更重要时如何用最少的架构成本获得可审计、可仲裁的高置信标签——双标注并行摊薄延迟审查只在必要时仲裁仲裁始终回到原始证据全程落库可追溯。补充说明本文基于当前仓库cookbook/data_labeling/_18_quality_review/的 README、测试日志与源码编写。其中涉及的模型gemini-3.5-flash、claude-opus-4-7与 agno 版本2.7.4为仓库文档与测试日志记载的当时配置实际运行请以你自己配置的 API 与 agno 安装版本为准。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考