芥末
发布于 2026-01-05 / 0 阅读
0
0

多 Agent 系统中的协作模式、Handoff 与稳定性设计

多 Agent 系统的难点不在于“让多个 Agent 跑起来”,而在于让它们稳定地协作:谁负责决策,谁持有上下文,什么时候交接控制权,交接失败后怎么恢复,状态冲突怎么处理。

如果只是在一个任务里调用多个工具,很多时候单 Agent 就够了。多 Agent 真正解决的是异构协作问题,也就是不同任务阶段需要不同角色、不同提示词、不同工具权限,甚至不同模型能力。

一个典型的软件研发 Agent 系统可能是这样:

flowchart LR
    U[用户需求] --> P[产品分析 Agent]
    P --> C[代码生成 Agent]
    C --> R[代码审查 Agent]
    R --> T[测试 Agent]
    T --> O[最终交付结果]

这里的几个 Agent 不只是“分工不同”,而是能力边界不同:

Agent核心职责需要的工具输出形式
产品分析 Agent澄清需求、拆任务需求模板、历史 PRD需求规格
代码生成 Agent编写代码代码仓库、编辑器、依赖管理源码变更
代码审查 Agent检查质量和风险静态分析、规范检查Review 结论
测试 Agent生成并执行测试测试框架、CI 环境测试报告

如果这些职责用同一个 System Prompt、同一批工具、同一种输出格式就能完成,那就没有必要拆成多个 Agent。拆分会带来额外成本:调用链变长、上下文传递更复杂、错误更容易级联。


什么时候应该使用多 Agent

多 Agent 适合解决“角色、工具、模型、上下文”明显不同的问题,而不是单纯解决“步骤多”的问题。

场景是否适合多 Agent原因
一个 Agent 调用搜索、数据库、计算器等多个工具不一定适合只是工具多,单 Agent + Function Calling 通常足够
先检索资料,再写报告,再做事实核查适合检索、写作、核查的提示词和输出标准不同
代码生成、代码审查、测试执行协作适合角色目标不同,工具权限也不同
辩论、红蓝对抗、多角度评估适合Agent 需要持有不同立场
固定表单填写流程不一定适合状态机或普通工作流可能更可靠
企业审批、风控、审计流程适合但要谨慎需要角色隔离、日志追踪和权限控制

判断是否该拆 Agent,可以看四个信号:

  1. 提示词是否明显不同
    如果每个阶段都要用完全不同的角色设定和约束,就有拆分价值。

  2. 工具权限是否不同
    例如代码 Agent 可以写仓库,审查 Agent 只能读仓库,测试 Agent 可以访问 CI(持续集成)环境。

  3. 输出契约是否不同
    一个 Agent 输出 Markdown 报告,另一个 Agent 输出 JSON 测试结果,下游消费方式完全不同。

  4. 模型选择是否不同
    有的阶段需要强推理模型,有的阶段只需要低成本模型做格式整理。

多 Agent 系统在特定研究和基准中出现过较高失败率,范围可达到 41% 到 86.7%。这个数字不应该被理解成“多 Agent 一定不能用”,而是说明:只要拆成多个 Agent,就必须认真处理交接、状态、失败恢复和可观测性。


Supervisor 与 Swarm:两种常见协作模式

多 Agent 系统最常见的两种编排方式是 Supervisor 和 Swarm。

Supervisor:集中式调度

Supervisor 模式里有一个中心 Agent,负责理解用户请求、拆分任务、分配工作、收集结果和生成最终回复。其他 Agent 通常不直接面对用户。

flowchart TD
    U[用户] --> S[Supervisor Agent]
    S --> A1[检索 Agent]
    S --> A2[分析 Agent]
    S --> A3[写作 Agent]
    A1 --> S
    A2 --> S
    A3 --> S
    S --> U

这种模式的优势是控制清晰。所有关键决策都经过 Supervisor,因此容易做日志、审计、权限控制和失败恢复。

但它也有明显代价:Supervisor 每次路由都要消耗一次模型调用或规则判断,还要在不同 Agent 之间转换上下文。对于强交互场景,这部分开销可能占据较大比例。

Supervisor 适合这些场景:

  • 企业级流程,需要审计和追踪;
  • 工作流结构比较明确;
  • 需要统一权限控制;
  • 下游 Agent 不应该直接暴露给用户;
  • 每一步都需要中心节点判断是否继续。

Swarm:去中心化交接

Swarm 模式没有固定的中心调度者。当前活跃 Agent 可以根据任务需要,把控制权直接交给另一个 Agent。

flowchart LR
    U[用户] --> A[接待 Agent]
    A --> B[技术支持 Agent]
    B --> C[账单 Agent]
    C --> B
    B --> U

Swarm 的核心是 Agent 之间可以直接 Handoff。它更像一个专家网络:当前 Agent 发现自己不适合继续处理,就把上下文和控制权交给更合适的 Agent。

这种方式的优势是交互灵活、延迟较低。不必每一步都回到中心节点,适合动态对话和低延迟任务。

代价是全局控制变弱。Agent 之间如果没有清晰的交接协议,容易出现循环交接、上下文丢失、责任不清等问题。

Swarm 适合这些场景:

  • 对话式任务,用户问题会频繁变化;
  • Agent 数量可能动态增加或减少;
  • 更关注响应速度,而不是严格流程;
  • 每个 Agent 都能独立判断下一步应该交给谁。

混合架构:上层集中,下层灵活

复杂系统经常采用混合架构:顶层用 Supervisor 控制主流程,局部任务内部用 Swarm 做灵活协作。

flowchart TD
    U[用户] --> S[顶层 Supervisor]

    S --> G1[研发子系统]
    S --> G2[数据分析子系统]

    subgraph 研发子系统 Swarm
        C[代码 Agent] --> R[Review Agent]
        R --> T[测试 Agent]
        T --> C
    end

    subgraph 数据分析子系统 Swarm
        Q[查询 Agent] --> A[分析 Agent]
        A --> V[校验 Agent]
    end

    G1 --> S
    G2 --> S
    S --> U

选型可以按下面这张表判断:

需求更适合的模式
需要审计、追踪、权限控制Supervisor
流程固定,步骤清晰Supervisor
对话动态变化,路由不固定Swarm
希望减少中心调度开销Swarm
既要控制主流程,又要局部灵活混合架构

Handoff 不是传消息,而是控制权、上下文和状态的交接

Handoff 是多 Agent 系统里最容易被低估的机制。它不是简单地把一句话传给下一个 Agent,而是包含三件事:

  1. 控制权转移:下一个由谁继续处理?
  2. 上下文迁移:应该传哪些历史信息和中间结果?
  3. 状态同步:共享状态如何更新,冲突如何处理?

完整的 Handoff 过程可以表示成这样:

sequenceDiagram
    participant A as 当前 Agent
    participant S as 状态存储
    participant B as 目标 Agent
    participant L as 日志系统

    A->>A: 判断无法继续处理
    A->>S: 写入阶段结果和交接原因
    A->>B: 发送精简上下文
    B->>S: 读取必要状态
    B->>B: 校验输入契约
    B-->>A: 接收控制权
    A->>L: 记录 Handoff 事件

一个可靠的 Handoff 至少要回答这些问题:

问题如果不处理会怎样
为什么要交接?下游 Agent 不知道任务背景
交给谁?可能路由到错误角色
传什么上下文?传太少会丢信息,传太多会干扰判断
当前状态是否已经写入?下游读到旧状态或半成品
交接失败怎么办?任务卡住或重复执行
是否允许交回?可能出现无限循环

LangGraph 中的两种 Handoff 写法

LangGraph 适合构建多 Agent 工作流,因为它把 Agent 调用建模成图,把共享数据建模成状态。节点是执行单元,边决定下一步去哪里。

条件边:适合固定流程

如果流程比较固定,可以用条件边根据状态选择下一个节点。

from typing import Literal, TypedDict
from langgraph.graph import StateGraph, END

class State(TypedDict):
    task: str
    route: Literal["researcher", "writer", "finish"]
    research_notes: str
    final_answer: str

def router(state: State) -> State:
    # 实际项目里可以由规则或 LLM 决定 route
    if not state.get("research_notes"):
        return {**state, "route": "researcher"}
    if not state.get("final_answer"):
        return {**state, "route": "writer"}
    return {**state, "route": "finish"}

def choose_next(state: State):
    return state["route"]

builder = StateGraph(State)

builder.add_node("router", router)
builder.add_node("researcher", researcher_agent)
builder.add_node("writer", writer_agent)

builder.set_entry_point("router")

builder.add_conditional_edges(
    "router",
    choose_next,
    {
        "researcher": "researcher",
        "writer": "writer",
        "finish": END,
    },
)

builder.add_edge("researcher", "router")
builder.add_edge("writer", "router")

graph = builder.compile()

条件边的好处是可控、可视化、容易测试。缺点是灵活性有限,流程变化多时,边会越来越复杂。

Command:适合动态交接

如果希望 Agent 自己决定下一站,可以返回 Command,同时更新状态并指定目标节点。

from typing import TypedDict
from langgraph.types import Command

class State(TypedDict):
    task: str
    research_notes: str
    draft: str

def researcher_agent(state: State) -> Command:
    notes = "整理后的检索结论..."

    return Command(
        update={
            "research_notes": notes,
        },
        goto="writer",
    )

def writer_agent(state: State) -> Command:
    draft = f"基于资料生成回答:{state['research_notes']}"

    return Command(
        update={
            "draft": draft,
        },
        goto="reviewer",
    )

Command 的优势是交接逻辑可以靠近 Agent 本身,Agent 执行完后直接声明“我已经完成了什么,应该交给谁”。在 Swarm 风格的系统中,这种写法更自然。


ToolMessage 与 tool_call_id:容易踩的工程细节

当大语言模型通过工具调用触发 Handoff 时,对话历史里通常会出现一条 tool call。这个 tool call 必须有对应的 ToolMessage,而且 tool_call_id 要匹配。

from langchain_core.messages import ToolMessage

def handoff_to_writer(tool_call):
    # 执行控制权转移
    target_agent = "writer"

    return ToolMessage(
        content=f"Handoff accepted. Next agent: {target_agent}",
        name=tool_call["name"],
        tool_call_id=tool_call["id"],
    )

如果模型消息里声明了工具调用,但历史记录里没有对应的 ToolMessage,后续模型调用可能会认为对话结构损坏。表现出来的问题可能是:

  • 模型重复调用同一个 Handoff 工具;
  • 下一个 Agent 无法理解上一轮工具调用结果;
  • 框架校验消息历史时报错;
  • 日志里出现“工具调用未闭合”的异常。

多 Agent 调试时,不能只看最终输出,还要检查消息历史是否成对出现:AIMessage.tool_callsToolMessage.tool_call_id 必须对齐。


上下文迁移:不要把所有历史都塞给下一个 Agent

Handoff 时常见错误是把完整聊天记录、Supervisor 的路由过程、所有工具返回结果一起传给下游 Agent。这样做看似保险,实际会引入噪声。

下游 Agent 需要的是“完成自己任务所需的信息”,不是系统内部所有历史。

一个更稳妥的做法是构造标准化交接包:

{
  "task_id": "task-20260607-001",
  "goal": "根据检索资料生成一份技术方案",
  "handoff_from": "researcher",
  "handoff_to": "writer",
  "handoff_reason": "资料收集完成,需要生成结构化方案",
  "constraints": [
    "输出必须是 Markdown",
    "不要包含未验证的数据"
  ],
  "artifacts": {
    "research_notes": "核心资料摘要...",
    "references": [
      "https://example.com/a",
      "https://example.com/b"
    ]
  },
  "decisions": [
    "采用 Supervisor + 局部 Swarm 架构"
  ],
  "open_questions": [
    "是否需要加入人工审批节点"
  ]
}

上下文迁移要遵守三个原则:

原则说明
只传业务信息路由日志、内部推理草稿、无关聊天历史不应该传给下游
保留关键决策下游需要知道已经做过哪些判断,避免重复推理
明确输出契约交接包里要写清楚下游应该产出什么格式

如果两个 Agent 的输入格式不同,中间还需要做格式转换。例如检索 Agent 输出的是引用列表,写作 Agent 需要的是按主题聚合后的资料摘要,不能直接把原始搜索结果扔过去。


状态同步:共享状态不是共享草稿纸

多 Agent 系统往往会维护一个共享状态,例如任务目标、消息历史、中间结果、文件路径、执行状态等。共享状态如果没有约束,很容易出现冲突。

常见状态问题有三类:

问题例子后果
写冲突两个 Agent 同时修改 final_answer后写入覆盖先写入
脏读一个 Agent 读取到另一个 Agent 未完成的中间结果下游基于半成品继续执行
状态污染审查 Agent 把自己的评论写进代码生成字段后续 Agent 误判数据含义

更安全的做法是做状态分区:每个 Agent 只能写自己负责的字段,共享字段只能通过 Reducer 合并。

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # 消息历史用专门的消息合并器
    messages: Annotated[list, add_messages]

    # 每个 Agent 写自己的字段
    research_notes: Annotated[list[str], add]
    code_changes: Annotated[list[str], add]
    review_comments: Annotated[list[str], add]
    test_reports: Annotated[list[dict], add]

    # 最终输出最好由固定节点生成,避免多个 Agent 同时写
    final_answer: str

这里的 add 表示列表追加,而不是覆盖。多个 Agent 同时返回更新时,Reducer 会把结果合并起来。

状态设计可以按所有权划分:

状态字段写入者读取者
research_notes检索 Agent写作 Agent、审查 Agent
code_changes代码 Agent审查 Agent、测试 Agent
review_comments审查 Agent代码 Agent、Supervisor
test_reports测试 AgentSupervisor、写作 Agent
final_answer汇总 Agent用户

最危险的是让所有 Agent 都能读写同一个大对象,例如 contextmemory。短期看起来方便,长期会变成不可调试的黑盒。


AutoGen、CrewAI、LangGraph 怎么选

多 Agent 框架不能只看热度,要看它解决的是哪类问题。

框架核心思路适合场景主要代价
LangGraph图 + 状态机复杂工作流、可控编排、需要状态管理学习曲线较陡
AutoGenAgent 之间通过对话协作企业应用、对话式协作、可靠性要求高配置和抽象较重
CrewAI角色、任务、流程编排快速原型、MVP、任务型协作复杂控制能力有限

LangGraph:适合复杂流程和精细控制

LangGraph 的优势是可以明确描述节点、边、状态和循环。它适合这些场景:

  • RAG(检索增强生成)和多工具混合;
  • 有条件分支、循环修正、人工审批;
  • 需要 checkpoint 和恢复;
  • 需要清晰地控制每一步状态更新。

如果任务里存在“生成 → 审查 → 修改 → 再审查”的循环,LangGraph 会比简单链式调用更合适。

AutoGen:适合对话式协作和企业场景

AutoGen 把多 Agent 协作建模成对话。它适合多个角色围绕一个问题来回讨论,例如规划、执行、审查、反馈。

它的优势在于企业级能力相对完整,包括日志、可观测性、错误处理等方面的支持。代价是抽象较重,配置也更复杂。

CrewAI:适合快速验证想法

CrewAI 的上手门槛低,角色、任务、工具的概念直观。它适合快速验证一个多 Agent 想法,例如:

  • 自动生成市场分析报告;
  • 多角色内容生产;
  • 简单研发辅助流程;
  • MVP 阶段的任务编排。

当流程开始出现复杂分支、循环、并发和强状态管理时,就需要评估是否迁移到 LangGraph 这类更可控的框架。


多 Agent 稳定性的核心:先约束,再重试

多 Agent 系统失败并不完全随机,常见问题有明确模式。公开研究对大量执行轨迹的分析显示,规格问题和协调失败占比较高,其中规格问题可达 41.77%,协调失败可达 36.94%。

规格问题:Agent 对输入输出理解不一致

规格问题通常表现为:

  • 上游输出 Markdown,下游却按 JSON 解析;
  • 字段名不一致,例如 summaryabstract 混用;
  • Agent 自由发挥,输出了额外解释文本;
  • 下游要求数组,上游返回字符串。

解决方式是强制输入输出结构化。

from pydantic import BaseModel, Field
from typing import Literal

class ReviewResult(BaseModel):
    status: Literal["approved", "need_changes", "rejected"]
    risk_level: Literal["low", "medium", "high"]
    comments: list[str] = Field(default_factory=list)
    required_changes: list[str] = Field(default_factory=list)

每个 Agent 都应该有明确契约:

def parse_review_result(raw_output: str) -> ReviewResult:
    return ReviewResult.model_validate_json(raw_output)

配套的契约测试也很重要:

def test_review_agent_output_contract():
    raw = run_review_agent(
        code_diff="修改了登录接口的鉴权逻辑",
        requirement="检查安全风险"
    )

    result = ReviewResult.model_validate_json(raw)

    assert result.status in {"approved", "need_changes", "rejected"}
    assert result.risk_level in {"low", "medium", "high"}
    assert isinstance(result.comments, list)

没有 Schema 的多 Agent 系统,很容易把错误推迟到下游才暴露。等到最后一步才发现格式不对,排查成本会很高。

协调失败:一个错误被下游不断放大

协调失败通常不是单点错误,而是错误传播链。

flowchart LR
    A[检索 Agent 返回错误资料] --> B[分析 Agent 基于错误资料推理]
    B --> C[写作 Agent 生成错误结论]
    C --> D[审查 Agent 未发现问题]
    D --> E[最终输出错误结果]

防护手段包括:

手段作用
熔断某个 Agent 连续失败后停止调用,避免浪费成本
降级非关键 Agent 失败时使用备用流程
检查点每个关键阶段保存状态,失败后从最近检查点恢复
幂等设计重试不会造成重复写入或重复执行危险操作
超时控制防止某个 Agent 长时间占用流程
最大交接次数防止 Agent 之间无限 Handoff

例如可以给 Handoff 设置最大次数:

MAX_HANDOFFS = 5

def guard_handoff(state, target_agent: str):
    handoff_count = state.get("handoff_count", 0)

    if handoff_count >= MAX_HANDOFFS:
        return {
            **state,
            "status": "failed",
            "error": "handoff limit exceeded",
        }

    return {
        **state,
        "next_agent": target_agent,
        "handoff_count": handoff_count + 1,
    }

Swarm 模式尤其需要这个保护,否则两个 Agent 可能来回交接:

客服 Agent -> 技术 Agent -> 账单 Agent -> 客服 Agent -> 技术 Agent ...

运行时问题:工具、网络和外部系统失败

多 Agent 系统通常依赖外部工具,例如搜索 API、数据库、代码仓库、CI 系统。LLM(大语言模型)本身的不确定性之外,外部系统也会失败。

常见防护包括:

  • 工具调用设置超时;
  • 网络错误使用指数退避重试;
  • 写操作使用幂等键;
  • 高风险操作加入人工确认;
  • 每个工具返回结构化错误,而不是直接抛出裸异常。
import time

def call_tool_with_retry(fn, *, retries=3, base_delay=0.5):
    last_error = None

    for attempt in range(retries):
        try:
            return fn()
        except TimeoutError as e:
            last_error = e
            time.sleep(base_delay * (2 ** attempt))

    return {
        "ok": False,
        "error_type": "timeout",
        "message": str(last_error),
    }

返回结构化错误后,Agent 才能判断是重试、降级,还是交给人工处理。


可观测性:没有 Trace,就没有调试能力

多 Agent 系统的错误经常跨越多个节点。如果没有完整 Trace,只看最终回答,很难定位问题发生在哪一步。

每次 Agent 调用至少要记录这些信息:

字段说明
task_id一次用户任务的唯一 ID
agent_name当前执行的 Agent
input_schema_version输入契约版本
output_schema_version输出契约版本
handoff_from来源 Agent
handoff_to目标 Agent
handoff_reason交接原因
token_usageToken 消耗
latency_ms调用耗时
tool_calls调用过哪些工具
status成功、失败、降级、熔断

Trace 的价值不是“方便看日志”,而是能回答生产问题:

  • 哪个 Agent 最容易失败?
  • 哪个 Handoff 最常丢上下文?
  • 哪个工具调用耗时最高?
  • 哪类输入最容易触发格式错误?
  • 成本主要消耗在哪些节点?

没有这些数据,稳定性优化只能靠猜。


一个可落地的多 Agent 设计清单

设计多 Agent 系统时,可以按这份清单检查:

模块必须明确的问题
Agent 边界每个 Agent 负责什么,不负责什么
输入输出契约每个 Agent 的输入和输出 Schema
协作模式Supervisor、Swarm 还是混合架构
Handoff 协议交接原因、目标 Agent、上下文包、失败处理
状态管理哪些字段共享,哪些字段只能由特定 Agent 写入
错误处理重试、熔断、降级、超时、人工介入
可观测性Trace、日志、指标、成本统计
测试策略单 Agent 契约测试、端到端流程测试、失败注入测试

多 Agent 的关键不是把 Agent 数量堆起来,而是让每个 Agent 有清晰职责,并且让它们之间的交接可控、可验证、可恢复。

能用单 Agent + 多工具解决的问题,不必强行拆成多 Agent。确实需要多角色协作时,重点要放在 Handoff、状态同步、结构化输出和失败防护上。框架只是实现手段,系统边界和协作协议才决定稳定性。


评论