多智能体主管系统

多智能体最容易做成「几个聊天框互相转述」。这个教程用结构化契约和共享事实表，把协调成本压下来，并用同一批任务对比单 Agent 与多 Agent 的成功率、成本和延迟。

目标

输入一个研究问题，输出一份带引用的简报：

bash

python -m team "对比 ACME 与 Beta 的退款政策，并说明对跨境订单的影响"

流程：

text

Supervisor
  ├─ Researcher A：ACME 政策
  ├─ Researcher B：Beta 政策
  ├─ Researcher C：跨境条款
  ├─ Analyst：对比与冲突
  ├─ Writer：成稿
  └─ Reviewer：事实与引用校验

项目结构

text

team/
├─ __init__.py
├─ __main__.py
├─ contracts.py     # 交接契约
├─ state.py         # 共享状态
├─ agents.py        # 专家实现
├─ supervisor.py    # 编排
└─ prompts.py
data/
└─ docs/            # 本地资料，教程用关键词检索模拟

Step 1：定义交接契约

contracts.py：

python

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any


@dataclass
class Contract:
    from_agent: str
    to_agent: str
    task: str
    reason: str
    context: dict[str, Any] = field(default_factory=dict)
    constraints: list[str] = field(default_factory=list)
    output_schema: dict[str, Any] = field(default_factory=dict)

    def to_prompt(self) -> str:
        return f"""你收到了来自 {self.from_agent} 的任务。

任务：{self.task}
原因：{self.reason}
背景：{self.context}
约束：{self.constraints}

只输出符合以下 Schema 的 JSON：
{self.output_schema}
"""

契约里必须有的五样东西：

任务：要做什么；
原因：为什么交给你，避免重复判断；
背景：已经知道什么，避免重复检索；
约束：不能做什么；
输出 Schema：下游怎么消费。

Step 2：共享状态

state.py：

python

from __future__ import annotations

from dataclasses import dataclass, field


@dataclass
class Fact:
    claim: str
    quote: str
    source: str
    confidence: float = 0.0


@dataclass
class CaseState:
    question: str
    facts: list[Fact] = field(default_factory=list)
    notes: list[str] = field(default_factory=list)
    draft: str = ""
    review: dict | None = None
    tokens: int = 0
    cost: float = 0.0

    def add_facts(self, items: list[dict], source: str) -> None:
        for item in items:
            if not item.get("quote"):
                continue
            self.facts.append(
                Fact(
                    claim=item["claim"],
                    quote=item["quote"],
                    source=source,
                    confidence=float(item.get("confidence", 0.5)),
                )
            )

    def facts_digest(self, limit: int = 12) -> str:
        lines = []
        for i, f in enumerate(self.facts[:limit], start=1):
            lines.append(f"[F{i}] {f.claim} | 原文：{f.quote[:40]} | 来源：{f.source}")
        return "\n".join(lines)

只共享事实与决策，不共享全部对话。原始文档放在外部，需要时按来源回查。

Step 3：专家 Agent

agents.py：

python

from __future__ import annotations

from dataclasses import dataclass

from .contracts import Contract
from .llm import LLM
from .state import CaseState


@dataclass
class Specialist:
    name: str
    system: str
    llm: LLM

    def run(self, contract: Contract, state: CaseState) -> dict:
        prompt = contract.to_prompt() + "\n\n已知事实：\n" + state.facts_digest()
        result = self.llm.complete_json(
            system=self.system,
            user=prompt,
            schema=contract.output_schema,
        )
        state.tokens += result["tokens"]
        state.notes.append(f"{self.name}: {contract.task}")
        return result["data"]

LLM.complete_json 是一个薄封装，负责：

强制 JSON 输出；
解析失败时修复一次；
返回 token 用量。

专家不需要完整的 ReAct 循环——如果它需要工具，可以在 Specialist 里加一个受限的检索工具，但不要让每个专家都能写数据库。

Step 4：主管编排

supervisor.py：

python

from __future__ import annotations

import asyncio

from .agents import Specialist
from .contracts import Contract
from .state import CaseState


class Supervisor:
    def __init__(self, researchers, analyst, writer, reviewer, max_rounds=2):
        self.researchers = researchers
        self.analyst = analyst
        self.writer = writer
        self.reviewer = reviewer
        self.max_rounds = max_rounds

    async def run(self, question: str) -> CaseState:
        state = CaseState(question=question)

        # 1. 三个独立研究任务并行
        tasks = [
            ("ACME 退款政策与例外", "docs/acme-refund.md"),
            ("Beta 退款政策与例外", "docs/beta-refund.md"),
            ("跨境订单的退款与时效", "docs/cross-border.md"),
        ]
        results = await asyncio.gather(
            *[
                asyncio.to_thread(
                    researcher.run,
                    Contract(
                        from_agent="supervisor",
                        to_agent=researcher.name,
                        task=task,
                        reason="需要该来源的原文证据",
                        context={"question": question, "source": source},
                        constraints=[
                            "每条事实必须附原文摘录，不得改写",
                            "没有来源的内容不要输出",
                            "最多 8 条事实",
                        ],
                        output_schema={
                            "type": "object",
                            "properties": {
                                "facts": {
                                    "type": "array",
                                    "items": {
                                        "type": "object",
                                        "properties": {
                                            "claim": {"type": "string"},
                                            "quote": {"type": "string"},
                                            "confidence": {"type": "number"},
                                        },
                                        "required": ["claim", "quote", "confidence"],
                                    },
                                }
                            },
                            "required": ["facts"],
                        },
                    ),
                    state,
                )
                for researcher, (task, source) in zip(self.researchers, tasks)
            ]
        )
        for (task, source), result in zip(tasks, results):
            state.add_facts(result.get("facts", []), source)

        # 2. 分析
        analysis = self.analyst.run(
            Contract(
                from_agent="supervisor",
                to_agent="analyst",
                task="对比两家政策，列出差异、冲突与对跨境订单的影响",
                reason="事实已收集，需要结构化对比",
                context={"question": question},
                constraints=["只使用已知事实", "冲突必须明确指出，不要擅自选边"],
                output_schema={
                    "type": "object",
                    "properties": {
                        "comparison": {"type": "array", "items": {"type": "string"}},
                        "conflicts": {"type": "array", "items": {"type": "string"}},
                        "implications": {"type": "array", "items": {"type": "string"}},
                    },
                    "required": ["comparison", "conflicts", "implications"],
                },
            ),
            state,
        )

        # 3. 写作 → 审查 → 最多修一轮
        for round_no in range(1, self.max_rounds + 1):
            draft = self.writer.run(
                Contract(
                    from_agent="supervisor",
                    to_agent="writer",
                    task="写一份 600 字以内、带引用的简报",
                    reason=f"第 {round_no} 稿",
                    context={"analysis": analysis, "facts": state.facts_digest()},
                    constraints=["每个事实后标注 [F#]", "不要引入事实表之外的内容"],
                    output_schema={"type": "object", "properties": {"draft": {"type": "string"}}, "required": ["draft"]},
                ),
                state,
            )
            state.draft = draft["draft"]

            review = self.reviewer.run(
                Contract(
                    from_agent="supervisor",
                    to_agent="reviewer",
                    task="检查引用、事实一致性与遗漏",
                    reason="成稿后必须独立校验",
                    context={"draft": state.draft, "facts": state.facts_digest()},
                    constraints=["只输出问题清单与结论", "不要重写全文"],
                    output_schema={
                        "type": "object",
                        "properties": {
                            "pass": {"type": "boolean"},
                            "issues": {"type": "array", "items": {"type": "string"}},
                        },
                        "required": ["pass", "issues"],
                    },
                ),
                state,
            )
            state.review = review
            if review["pass"]:
                break

        return state

真实代码还要加：

每个专家独立超时与重试；
主管的最大协调轮数；
某一路研究失败时的降级（少一个来源也能写）；
事实去重；
全链路 trace。

Step 5：提示词

prompts.py：

python

RESEARCHER = """你是资料研究员。只从给定文档中抽取事实。
每条事实必须包含：
- claim：一句话结论
- quote：原文摘录，最多 50 字，不得改写
- confidence：0 到 1
找不到相关信息时返回 {"facts": []}，不要编造。"""

ANALYST = """你是分析师。根据事实表做对比。
要求：
1. 只使用给定事实。
2. 明确列出冲突；如果两个来源数字不同，说明可能的口径差异。
3. 不要给投资、法律或医疗建议。"""

WRITER = """你是技术写作者。把分析写成简洁简报。
要求：
1. 先结论，后依据。
2. 每个事实性句子后标注 [F#]。
3. 不超过 600 字。
4. 信息不足时明确写「资料不足」。"""

REVIEWER = """你是严格的审稿人。检查：
1. 每个 [F#] 是否存在于事实表。
2. 引用是否支持该句结论。
3. 是否出现事实表之外的新事实。
4. 是否遗漏了关键冲突。
只输出问题清单与 pass 布尔值。"""

Step 6：成本对比

用同一批 20 个问题，分别跑单 Agent 和多 Agent：

指标	单 Agent	多 Agent
通过率	78%	86%
平均 token	9k	38k
P95 延迟	18s	47s
交接丢失	无	2 次
引用准确率	71%	89%

结论通常是：多 Agent 提升的是引用准确率和覆盖度，代价是 3–4 倍成本。如果业务不需要多源核对，单 Agent 更划算。

Step 7：故障演练

故意制造三种故障：

一个研究员超时：主管是否还能用另外两路写完？
研究员返回无来源的「事实」：审查是否能拦住？
写作引用了一个不存在的 [F9]：审查是否报错？

如果系统只是「三个 Agent 互相聊天」，这三种情况大概率会静默通过，最后输出一份看起来完整但不可信的报告。

验收清单

每个任务有结构化交接契约
子 Agent 只拿到必要背景，不共享全部对话
独立研究并行执行
事实带来源，冲突显式记录
审查能拦截无来源事实与无效引用
有最大协调轮数、超时与降级
能输出每个 Agent 的 token 与耗时
与单 Agent 做过同集对比

常见错误

把完整历史复制给每个子 Agent

上下文隔离失效，成本翻倍。只传任务、必要背景和输出要求。

用自然语言摘要交接

细节会丢。用结构化字段：已完成、发现、未决、约束。

没有独立审查

写作 Agent 自己检查自己，几乎总能通过。审查者要拿事实表逐条核对。

主管无限重派

设置最大轮数，并在无新信息时停止。

小结

多 Agent 的收益来自上下文隔离、专业化、并行和独立验证。
交接契约比「让它们聊」可靠得多。
事实表是共享的，原始材料放外部。
用数据证明拆分值得，而不是因为架构图好看。

下一篇把质量变成数字：从零搭一条评估流水线。

#目标

#项目结构

#Step 1：定义交接契约

#Step 2：共享状态

#Step 3：专家 Agent

#Step 4：主管编排

#Step 5：提示词

#Step 6：成本对比

#Step 7：故障演练

#验收清单

#常见错误

#小结

相关阅读

目标

项目结构

Step 1：定义交接契约

Step 2：共享状态

Step 3：专家 Agent

Step 4：主管编排

Step 5：提示词

Step 6：成本对比

Step 7：故障演练

验收清单

常见错误

小结