企业官网建设流程全解析

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_adjudication=False needs_adjudication=True │ │ ▼ ▼ （跳过仲裁步骤） ┌──────────────────┐ │ Adjudicator 仲裁 │ │ 对照原始输入裁决 │ └──────────────────┘

关键设计决策：

1. 模型提供商差异化：两个 labeler 分别跑在google:gemini-3.5-flash与anthropic:claude-opus-4-7上，以获得"集成多样性（ensemble diversity）"——不同模型对同一文本的解读盲区不完全重合，分歧更有信息量；
2. 分歧才仲裁：adjudicator 由Condition门控，无分歧时整条流水线只花两次标注 + 一次审查的代价，仲裁只在真正必要时触发，兼顾质量与吞吐；
3. 全程落库：workflow 挂载SqliteDb，每次运行都持久化，构成可审计的裁决轨迹。

2.2 运行时测试依据

对应测试记录 TEST_LOG.md 记载（2026-07-18 测试，agno 2.7.4）：

• 干净输入：两个 labeler 返回完全一致的Contact（name='Liam Ortega'、email='liam@meadow.io'、company='Meadow'、title='Support Engineer'），reviewer 判定needs_adjudication=False，step trail 显示 Condition 跳过——'Condition Adjudicate not met - skipped 1 steps'；
• 冲突输入：两个 labeler 在 name（'Dr. Sarah Chen-Watanabe' vs 'Sarah Chen'）与 title（'Principal Scientist and acting Head of Platform' vs 'Principal Scientist'）上产生分歧，reviewer 逐字段给出原因；adjudicator 介入并输出最终FinalLabel(contact=Contact(name='Sarah Chen', email='s.chen@nova-labs.io', phone='+1-415-555-0177', company='NovaLabs', title='Principal Scientist and acting Head of Platform'))。

测试同时注明：哪些字段会分歧因运行而异，但冲突输入被构造为"至少有一个字段可靠地产生分歧"，从而保证两条路径（skip 路径与仲裁路径）都能被稳定覆盖。

三、数据结构设计：四个 Pydantic Schema

整个流水线的契约由 basic.py 中四个 Pydantic 模型定义：

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(..., description="Top-level Contact field name") value_a: Optional[str] = None value_b: Optional[str] = None reason: str = Field(..., description="Why this field needs adjudication") class DisagreementReport(BaseModel): disagreements: List[FieldDisagreement] = Field(default_factory=list) needs_adjudication: bool = Field(..., description="True if any field disagrees") class FinalLabel(BaseModel): contact: Contact notes: Optional[str] = None

这些模型同时充当智能体输出契约：每个 Agent 的output_schema指向对应模型，模型输出会被强制解析为结构化对象，这为 reviewer 的"逐字段 diff"和 adjudicator 的"对照原始输入裁决"提供了机器可读的数据基础。

四、Agent 与指令设计：四个角色各司其职

4.1 标注器（Labeler A / B）

LABELER_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( name="Labeler A", model="google:gemini-3.5-flash", instructions=LABELER_INSTRUCTIONS, output_schema=Contact, ) labeler_b = Agent( name="Labeler B", model="anthropic:claude-opus-4-7", instructions=LABELER_INSTRUCTIONS, output_schema=Contact, )

指令要点是"原文照录、缺失置空、禁止猜测"——这保证了分歧不是凭空捏造，而是真实文本歧义（如昵称、改名、复合头衔、过时联系方式）的投影。运行前提：需要同时配置GOOGLE_API_KEY与ANTHROPIC_API_KEY（见 README.md）。

4.2 审查器（Reviewer）

reviewer = Agent( name="Reviewer", model="anthropic: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_adjudication=true if any field needs adjudication. """, output_schema=DisagreementReport, )

注意分歧判定规则的精确定义：只有双方都非空（non-null）但取值不同的字段才需要仲裁——单方缺失（null）不算分歧，这避免了把"一方漏抽"误判为需要仲裁。测试日志也验证了该规则：干净输入双方一致时 reviewer 输出needs_adjudication=False。

4.3 仲裁器（Adjudicator）

adjudicator = Agent( name="Adjudicator", model="anthropic: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_schema=FinalLabel, )

仲裁器拥有最完整的上下文：原始输入 + 两个 labeler 输出 + reviewer 报告。对争议字段对照原文裁决，对无争议字段沿用双方一致值。这正是测试日志中冲突输入裁决结果 name='Sarah Chen'（采用短名偏好）而 title 保留完整复合头衔的原因。

五、Step 与 Workflow 组装：从 Agent 到可运行流水线

5.1 普通 Agent 步骤

两个 labeler 直接封装为普通步骤：

label_a = Step(name="Labeler A", agent=labeler_a) label_b = Step(name="Labeler B", agent=labeler_b)

5.2 自定义执行器（Executor）：把多步骤输出拼进 Prompt

reviewer 与 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 = ( f"Labeler A:\n{a.model_dump_json(indent=2)}\n\n" f"Labeler B:\n{b.model_dump_json(indent=2)}" ) report = reviewer.run(prompt).content return StepOutput(content=report)

源码层面，StepInput.get_step_output()在 workflow/types.py 中实现：先按步骤名直接查找，若失败则递归下钻嵌套步骤（Parallel / Condition / Router / Loop / Steps），这正是它能找到位于Parallel块内部的Labeler A/Labeler B输出的原因。

adjudicator 的 executor 进一步把三路信息组装进 prompt：

def 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 = ( f"Original input:\n{step_input.input}\n\n" f"Labeler A:\n{a.model_dump_json(indent=2)}\n\n" f"Labeler B:\n{b.model_dump_json(indent=2)}\n\n" f"Reviewer report:\n{report.model_dump_json(indent=2)}" ) final = adjudicator.run(prompt).content return StepOutput(content=final)

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(name="Adjudicator", executor=run_adjudicator)

随后在 Workflow 中用Condition(evaluator=has_disagreement, steps=[adjudicate])门控。源码层面（workflow/condition.py），Condition的evaluator支持三种形态：

• 可调用函数（本示例所用）：返回 bool；
• 布尔字面量True/False；
• CEL 表达式字符串：可访问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( name="Quality review labeling", db=SqliteDb(db_file="tmp/labeling.db"), # every run is persisted steps=[ Parallel(label_a, label_b, name="Label"), # labelers run concurrently review, # diff them field by field Condition( # adjudicate only on disagreement name="Adjudicate", evaluator=has_disagreement, steps=[adjudicate], ), ], )

Parallel原语（workflow/parallel.py）接受可变参数，强调各子步骤独立且无序，支持Parallel(step1, step2, name="my_parallel")或名称置首的调用约定。SqliteDb(db_file="tmp/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 liam@meadow.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.chen@nova-labs.io. The " "555-0123 number on the website is stale; her direct line is " "+1-415-555-0177." )

冲突输入"按构造产生分歧"：两个电话号码、改名后的公司、复合头衔、偏好的短名——独立标注器至少会在一个字段上产生不同序列化结果，从而稳定触发仲裁路径。

6.3 结果打印与 step trail

def 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_adjudication=False→ trail 出现Condition Adjudicate not met - skipped 1 steps，仲裁步骤未执行；
• 仲裁路径：冲突输入下 name / title 分歧被 reviewer 逐字段标记 → adjudicator 基于原始输入裁决出FinalLabel（保留复合头衔、采用短名、采用直拨电话），trail 完整记录每一步。

七、复用与扩展：把质检闸门装到任意抽取原语上

这套工作流的复用方式非常直接：换掉 labeler agent 与 schema，即可把同样的质检闸门套到本目录下任意其他原语（图片、音频、文档抽取）上（README.md）。

具体步骤：

1. 替换 schema：将Contact换成目标任务的输出模型（如_03_text_extraction之外的其他抽取结果结构），并同步调整FieldDisagreement的field描述与 reviewer / adjudicator 的指令；
2. 替换 labeler：保持"不同提供商"原则，把model换成你拥有的 API 组合（如 OpenAI + Anthropic、OpenAI + Gemini），并保留"原文照录、缺失置空"的指令风格；
3. 保持拓扑：Parallel双标注 → reviewer 逐字段 diff →Condition门控仲裁 的三段式结构无需改动；
4. 按需升级门控：若某类输入必须仲裁，可将Condition的evaluator换成 CEL 表达式或布尔值；需要人工复核时，Condition还支持requires_confirmation=True的 HITL 模式（见 condition.py）。

八、底层原理小结：这篇代码为什么这样组织

从 basic.py 到 agno 源码，可以梳理出四条支撑本工作流的底层机制：

这套组合回答了一个核心工程问题：当"质量比吞吐更重要"时，如何用最少的架构成本获得可审计、可仲裁的高置信标签——双标注并行摊薄延迟，审查只在必要时仲裁，仲裁始终回到原始证据，全程落库可追溯。

补充说明：本文基于当前仓库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