多手一支笔:协调AI智能体的七条规则
编排人工智能智能体的七条规则:何时普通工作流更优、为何编排器无需工具、带类型标注的子智能体返回以及书面的停止条件。
来源: Many Hands, One Pen: Orchestrating Agents Without Losing the Plot
想象一下1942年的美国海军,他们面临着协调上的巨大难题,这恰恰反映了现代软件所面临的挑战。战舰配备了雷达、声纳、无线电截获设备以及人工瞭望员,各种报告同时以随机顺序传送到指挥桥上。试图指挥战斗的舰长被海量未处理的信号淹没。
海军的解决方案是一个没有窗户的房间,被称为作战信息中心。所有传感器报告都按照统一的格式标注在同一个图表上。操作人员负责读取仪器数据,但他们并不直接下达指令,其职责只是确保图表的准确性,而舰队的下一步行动则由一名军官独自决定。信息从各个方向汇集而来,决策则逐渐聚焦,而标准的报告格式使得各项数据能够相互对比。
多智能体系统不断重新探索那个“空间”。那种认为“增加一个智能体”就能解决所有问题的阶段已经结束,2026年大部分时间都用于衡量这种做法的代价。最终保留下来的模式正是海军所采用的:并行收集信息,以规范格式汇报,统一决策,并明确界定任务完成的标准。
一个有用的视角:大多数多智能体问题其实并非真正的智能体问题,而是披着智能体外衣的上下文问题与控制流问题。任务出错通常是因为两个智能体做出了彼此无法察觉的互斥决策,或者信息在跨越边界时失去了意义,又或者没人明确写出任务完成的标准是什么。
以下是七种实践方法,均源自已投入使用的系统中的失败案例。它们共同帮助你判断何时普通工作流优于智能体,何时需要为协调器分配令牌,如何根据上下文边界划分任务,为何协调器不应直接干预工具使用,如何为子智能体设定可验证的回报契约,以及如何有意识地结束任务,而非任由模型自行决定。
优先使用普通工作流;自主权需通过努力获得
出问题所在。以一个有四种结果的退款流程为例,每种结果都有相应的书面规则。某个团队将其封装在一个自主循环中,使用六种工具并设定一个目标字符串。结果却是,周二处理的工单与周一的路径不同,而且每次运行所需的计算量是旧代码的十到五十倍。由于流程中的任何环节都足够明确,无需依赖智能系统,因此准确率依然保持不变。
解决办法。首先绘制流程图。所有能提前规划的内容都应属于普通控制流,这样简单的代码就能处理所有可命名的分支。只有在输入确实存在多种解释可能的步骤才调用模型。只有当某个分支无法用文字描述时,才应让智能代理来处理它。
REFUND_LIMIT = 50.00
def llm_step(prompt: str) -> str:
"""Called only where the written rules run out."""
return model.complete(prompt).strip().lower()
def handle_refund(ticket):
if ticket.days_since_purchase > 30:
return deny(ticket, reason="outside window")
if ticket.amount <= REFUND_LIMIT:
return approve(ticket) # a rule, not a judgement call
intent = llm_step(
f"Classify as fraud, defect, or remorse:\n{ticket.body}"
)
if intent == "fraud":
return escalate(ticket, queue="risk")
if intent == "defect":
return approve(ticket)
return route_to_human(ticket)
问题所在。你可能会陷入层层嵌套的条件判断中,这些条件实际上无法体现真正的判断规则。当你要为自由文本编写第五个条件时,这就是该模型存在的模糊性所在。
当问题已明确时无需使用协调器
出错的原因。一个四步的报告处理流程被包裹在最多进行十五次迭代的协调器循环中。每次迭代都会花费时间来决定下一步该做什么,而每次决策都会得出你早已确定的答案:获取、提取、总结、格式化。这样就需要进行十五次推理,每次处理几千个标记,只为重新得到本可以用五行文字写出的列表。
解决办法。在编写任何代码之前先使用快速规则。如果在运行开始前就已知道所有子任务的完整列表,就以固定调度方式的简单循环来执行它们,并将令牌留作其他用途。模型驱动编排适用于那些任务分解本身未知,或需要根据前期步骤的发现而动态变化的情况。
# Before: up to 15 planning calls to rediscover a fixed list
def run_orchestrated(doc):
state = {"doc": doc}
for _ in range(15):
decision = orchestrator.plan(state) # 1 LLM call per pass
if decision.action == "finalize":
break
state = WORKERS[decision.worker](state)
return state
# After: 0 planning calls, same four workers, same result
PIPELINE = [fetch, extract, summarize, format_report]
def run_static(doc):
state = {"doc": doc}
for step in PIPELINE: # 0 LLM calls here
state = step(state)
return state
问题所在。这条建议也有其反面。如果一个流程每月都会产生新的特例,或者第二步的输出真正决定了第三步是否执行,那才算是真正的规划。在此类情况下移除规划器,只会把分支逻辑转移到某个没人会去维护的配置文件中。
按上下文而非职位划分智能体
出错的原因。多智能体的组织结构图看起来很整齐:规划者、实现者、测试者、审核者,依次处理一个功能。但实际上,每次任务交接都会带来一些未被明文记录的假设。测试者会检查实现者从未打算过的行为,而审核者则会反对规划者刻意做出的选择。你因此付出了四倍的成本,而让一个智能体承担所有四项工作本会更好。
解决办法。沿着上下文边界进行划分。在添加第二个智能体之前,先向设计层面提出一个问题:这个子任务能否在与其相邻子任务毫无关联的上下文窗口内完成?如果答案是否定的,那么这两个子任务就应该归属于同一个智能体,这也正是为什么负责构建某个功能的智能体也应同时为其编写测试的原因。
from dataclasses import dataclass
@dataclass
class Subtask:
name: str
needs: set[str] # facts this subtask must read
produces: set[str] # facts this subtask decides
def should_split(a: Subtask, b: Subtask) -> bool:
"""Split only when neither side needs what the other decides."""
shared = (a.needs & b.produces) | (b.needs & a.produces)
return not shared
implement = Subtask("implement", {"spec"}, {"api_shape", "error_semantics"})
test = Subtask("test", {"spec", "api_shape", "error_semantics"}, {"cases"})
assert not should_split(implement, test) # one agent writes both
问题所在。真正完全独立的子任务比组织结构图所显示的更为罕见,因此测试中常常会提到“一名处理员”,而这正是正确答案,而非缺乏想象力。对于需要广泛处理的任务,比如让五名处理员分别阅读五个无关的资料,拆分任务会更有成效,因为其中一人的决定不会影响另一人所需的内容。
让协调器输出决策,切勿直接调用工具
出错的原因。协调器获得了用于紧急情况的搜索工具。很快它就不再分配任务,而是开始自行回答子任务。原始工具的输出结果不断堆积在它的界面中,因此随着任务时长增加——而此时规划最为重要——它的规划能力反而会下降。可追溯性也会受到影响,因为无法明确是哪名处理员得出了哪项结果。
解决方案。任何计划都不应自行生成,因此应不给协调器提供任何工具。它仅会输出一个简洁的、已定义好的决策:委托、重新规划或最终确定,同时还会指定要使用的执行者以及任务内容。周围的循环会读取该决策,运行指定的执行者,并将结果存储在共享状态中,以便做出下一个决策。
from typing import Literal
from pydantic import BaseModel
class OrchestratorDecision(BaseModel):
next_action: Literal["delegate", "replan", "finalize"]
target_worker: str | None = None
task_description: str | None = None
reasoning: str
def orchestrator_step(state):
d = decide(state) # model has zero tools attached
if d.next_action == "finalize":
return finalize(state, d.reasoning)
if d.next_action == "replan":
return state.reset_plan(d.reasoning)
worker = WORKERS[d.target_worker] # workers own every tool
result = worker.run(d.task_description)
return state.record(d.target_worker, result)
问题所在。在没有工具的情况下,协调者只能看到工作者返回的信息。如果工作者只返回简略的总结,就会让规划工作缺乏依据。应当提供对共享状态的只读访问权限,而非实时的工具访问权限,同时要确保工作者返回的信息结构清晰,便于进行规划。
要求每个子代理遵守规范化的返回格式
出错的原因。假设某个研究子代理会返回一篇2,500字的日记,详细记录它尝试过的所有方法以及失败的路径。如果有七个这样的工作者,那么在开始任何综合处理之前,就有大约20,000个标记会被放入父代理的窗口中。委托本应起到过滤这些信息的作用,但实际上探索工作被重复付出了成本,而父代理在最需要准确执行指令的时候反而表现更差。
解决方案。可将返回结果视为 API 边界,每个子代理都需遵循相应的响应格式规范。其结束消息为 JSON 格式,包含命名字段:发现结果、数据来源、未解答的问题以及处理完成状态。在消息到达时立即进行检查,将任何格式错误或过长的消息退回以便重新处理,确保其不会传递给上层节点。
from pydantic import BaseModel, Field, ValidationError
class SubagentResult(BaseModel):
findings: list[str] = Field(max_length=5) # hard cap, one sentence each
sources: list[str]
open_questions: list[str] = []
completion_status: str # complete | partial | blocked
def parse_or_retry(worker, task, attempts=2):
for _ in range(attempts):
raw = worker.run(task, response_format=SubagentResult)
try:
return SubagentResult.model_validate_json(raw)
except ValidationError as err:
task = f"{task}\n\nRejected: {err}\nReturn only JSON in the schema."
raise RuntimeError(f"{worker.name} returned no valid result")
问题所在。模式只设定了最低要求,没有更高限制。工作人员仍可能将400字的段落塞进一个结果字符串中,再以JSON的形式呈现同样臃肿的数据。应限制条目数量及每个字段的长度,并在拒绝信息中注明超出限制的字段,这样重试时就能知道需要修正什么。
并行读取并通过单个代理处理写入操作
可能出现的问题。两个代理同时处理同一个功能,且都报告成功。一个代理生成了超级马里奥风格的背景,另一个则生成了飞鸟游戏的精灵图,而两者都通过了针对各自设计版本的测试。由于合并过程无法解决这两个代理都未意识到的差异,最终要么一个结果被丢弃,要么两个结果都被舍弃。
解决方案。阅读可以并行进行,而写作则不能。将每个工作节点限制在只读工具的允许列表中,这样收集阶段就可以根据需要扩展到更大规模,同时绝不要给任何工作节点提供写入工具。最终,通过一次无条件调用即可整合所有工作节点的结果来生成最终产物。
import asyncio
READ_TOOLS = ["search_repo", "read_file", "fetch_docs"]
async def gather_context(subtasks):
workers = [Agent(name=t.name, tools=READ_TOOLS) for t in subtasks]
return await asyncio.gather(
*(w.run(t.prompt) for w, t in zip(workers, subtasks))
)
async def build_feature(spec, subtasks):
findings = await gather_context(subtasks) # wide, parallel, read-only
writer = Agent(name="writer", tools=["write_file", "apply_patch"])
return await writer.run(spec, context=findings) # one writer, one pass
问题所在。那唯一的写入器成了瓶颈,必须同时处理所有工作者的成果,因此早在达到逻辑限制之前就会耗尽令牌预算。如果输出结果真的能够像三个没有接口交互的文件那样被清晰分割,那就为每个文件分配独立的写入器,并在每个写入器内部保持相同的处理规范。
明确写入终止条件
出错的原因。运行结束后没有出现错误,但任务仍未完成。对七个框架中1,600条多智能体轨迹的分析显示,12.4%的失败是由于智能体不知道自身的终止条件,另有6.2%的智能体过早停止。因此,几乎五分之一的失败表现为以未完成的结果正常退出。
解决办法。将终点线转化为具体的事物而非抽象的感觉。为每项任务制定明确的完成检查清单,并在通往最终状态的唯一路径上设置依据初始目标设定的审批节点。同一项研究显示,验证机制使ChatDev的成功率提升了9.4%,而检查高层目标则使其提升了15.6%。
CHECKLIST = [
"every requested section exists",
"each claim cites a retrieved source",
"open questions are listed, or explicitly none",
]
def verify_complete(objective, checklist, output) -> bool:
for item in checklist:
verdict = judge(f"Objective: {objective}\nCheck: {item}\n\n{output}")
if not verdict.passed:
log.info("termination blocked by: %s", item)
return False
return judge(f"Does this satisfy the objective?\n{objective}\n\n{output}").passed
def finish(state):
if verify_complete(state.objective, CHECKLIST, state.draft):
return state.done()
return state.keep_working(reason="checklist not satisfied")
关键问题。如果让生成模型自行评估检查清单,它很可能会直接通过审核。尽可能使用机械式的验证项,比如文件是否存在、字段是否已填写或引用是否有效。永远不打开的验证关卡本身也是一种故障,因此应限制尝试次数,用尽后立即抛出错误。
从何处开始
如果只能改变一件事,那就选择最简单的方法:以书面形式明确“完成”的标准。一个检查清单加上进入最终状态前的验证步骤,只需一个下午就能完成,既适用于单个代理也适用于整个系统,还能避免一类隐蔽的故障。把这当作本周的任务吧。
接下来是移除步骤。检查你的协调器,测试其决策是否可简化为固定列表与聚合函数。若是如此,就将其移除;这样令牌消耗和延迟都会下降,且结果依然稳定。由于这种操作很容易撤销,因此是安全的首选方案。
尽量推迟使用复杂工具,直到实在无法避免为止。当工作者数量达到五人左右时,带类型返回的契约才会显现出优势;而当有两个代理真正同时进行写入操作时,只读白名单就显得非常重要。例外情况是无工具协调器:初期几乎无需成本,但之后添加功能则会十分麻烦。
在2026年仍能保持稳定的架构,并不会使用过多的代理。在这些架构中,代理之间的连接部分是设计中最简单且最明确的环节。更少的参与者、更严格的契约规则,再加上统一的操作方式即可。
相关阅读
- 结合 Git Worktrees、Conductor 与 Antigravity 2.0 的并行编码代理——协调代理、各代理独立的 Git Worktrees 以及基于问题的任务跟踪机制如何帮助在大约五小时内完成十余项 SRE 功能的开发,同时涵盖有关 HITL 问题及合并规范的要点。
- Agno Teams 搭配的代理式 RAG:查询重写、评估循环与 Qdrant——协调团队负责查询重写,通过评估机制循环进行信息检索,并仅从经过验证的上下文中给出答案,该系统在 Qdrant 中运用了混合密集索引、BM25 索引及 ColBERT 模型进行搜索。