在值得并行的地方使用多 Agent

先画出依赖，再决定 Agent 数量

Repo 助手要比较三个独立组件的超时设置，可以让三个 worker 分别调查。若任务是先确定实际入口，再沿调用链查覆盖规则，后一步依赖前一步，强行拆开容易让各个 worker 猜测前提。适合并行的单位通常具有明确输入、独立证据和可合并输出。

同一模型的多个实例也能组成多 Agent 系统，关键在于它们有各自的状态与任务。一次模型调用并行执行几个工具，则是工具并行，两者不能混为一谈。先试单 Agent 配合并行读取，只有独立的推理任务确实受益时，再增加多个模型循环。

委派消息要让接收者可以直接行动

不要只说“你负责配置”。给出目标、范围、允许修改的位置、输入版本、返回格式、停止条件和预算。例如“只调查 prod 目录的请求超时；返回值、文件行号、覆盖关系与未解决冲突；最多读取五个文件；只读”。清楚的契约可以减少重复调查与漏项。

推荐结果对象包含 task_id、status、claims、evidence 和 open_questions。worker 的自然语言结论是候选信息，协调者仍要检查证据是否对应当前版本。向每个 worker 广播完整历史既贵又容易造成注意力分散，通常传递它完成子任务需要的上下文即可。

子任务交接可以用一个具体限制检查：worker 返回的 claims 每一条必须指向 evidence 中的一个 id；协调者发现悬空引用就拒绝合并。若只需要比较数值，worker 不必写长篇报告，但应保留无法确定的原因。这让结果既紧凑又可追查，避免主 Agent 在汇总时把“尚未找到”润色成“已经证实不存在”。

任务的输入版本也要一致。若一个 worker 调查提交 A，另一个调查提交 B，合并出的项目状态可能从未真实存在。为本次调查固定仓库快照，或者在最终输出明确标注各来源版本和无法直接比较的部分。

共享状态需要所有权

多个 worker 同时修改同一个文件，可能覆盖彼此的更改。可以把文件所有权分给不同 worker，或者让它们只返回补丁，由协调者统一合并。即使改的是不同文件，接口也要先约定，例如都以 timeout_seconds 为字段名。隔离写入位置不能自动解决语义冲突。

协调者维护任务表，记录待开始、运行中、成功、失败和取消状态。只有依赖满足的任务才能启动。最终答案必须区分已完成与未完成部分，不能把“worker 已返回”视为“它负责的问题已经解决”。检查缺失的 task_id 比阅读一大段完成感很强的总结更可靠。

动手：并发收集结果，统一核对来源

下面是无需 API 的 Python 标准库演示。worker 是固定函数，模拟子任务的执行协议，不是模型。它展示并发提交、逐项收集异常和合并证据；把 worker 替换成完整 Agent 循环，协调层仍可使用同一种结果契约。

from concurrent.futures import ThreadPoolExecutor, as_completed

FILES = {"prod.cfg": "timeout=30", "test.cfg": "timeout=5"}
tasks = [{"id": "prod", "path": "prod.cfg"},
         {"id": "test", "path": "test.cfg"},
         {"id": "legacy", "path": "missing.cfg"}]

def worker(task):
    text = FILES[task["path"]]
    return {"task_id": task["id"], "status": "ok",
            "value": int(text.split("=", 1)[1]),
            "evidence": {"path": task["path"], "line": 1, "text": text}}

results = {}
with ThreadPoolExecutor(max_workers=2) as pool:
    pending = {pool.submit(worker, task): task for task in tasks}
    for future in as_completed(pending):
        task = pending[future]
        try:
            result = future.result()
            evidence = result["evidence"]
            if FILES.get(evidence["path"]) != evidence["text"]:
                raise ValueError("source_changed")
            results[task["id"]] = result
        except Exception as error:
            results[task["id"]] = {"task_id": task["id"],
                                   "status": "failed", "error": type(error).__name__}

assert set(results) == {task["id"] for task in tasks}
for task_id in sorted(results):
    print(task_id, results[task_id])

结果应有两个成功项和一个明确失败项。完成顺序不保证与提交顺序相同，因此依靠 task_id 对应，最后排序只为便于展示。这里对来源文本的核对很简单；真实合并还要检查值、环境、行号与结论之间的关系。

并行缩短等待，但不自动减少工作量

假设三个独立调查各需四秒，串行共十二秒；理想并行只需四秒，但若拆分与汇总再花两秒，总时长是六秒。这是教学演算，不是性能承诺。多 Agent 往往增加重复上下文和汇总调用，因此延迟下降时，token 与费用反而可能上升。

先限制最大并发数，再设置全局调用预算，不能让每个 worker 都独享原任务的全部预算。真实网络请求要设置请求级超时；取消一个 future 不保证能停止已经运行的线程。对长任务还要设计协作式取消和清理行为，不能只在协调者一侧结束等待。Python futures 文档说明了这些执行语义。

分歧需要证据，不需要简单投票

两个 worker 都说 30 秒，另一个说 45 秒，票数不能决定配置。先检查是否读了不同环境、不同提交，或一个读到了覆盖项。共享同一份错误输入的多个模型可能一致犯错，结果一致并不代表独立验证。让最终结论回到版本明确的证据上。

保留单 Agent 基线，在相同任务、质量标准和总预算下比较。只有成功率或延迟收益足够抵偿协调成本时，才保留多 Agent 结构。Anthropic 的研究系统文章给出了协调者与 worker 的实践，但其中具体收益取决于其任务与模型，不能直接外推到我们的文件助手。阅读工程案例。

合上书，检查一下

先作答，再看解释。答错的地方，正好是值得再读一遍的地方。

配合本章看视频

下面按本站章节主题选配。为什么没有采用原课录播 →

继续读原始资料

• Anthropic：Multi-agent research system ↗
• Python concurrent.futures ↗