多 Agent 协作架构实战：分工、通信、冲突，别让你的 Agent 们打起来

上篇写 Agent 可观测性的时候，我提到过一句话：调试多个 Agent 协作比单个 Agent 难了一个数量级。当时不是随便说的——我手头刚好在重构一个多 Agent 系统，三个 Agent 各管一摊：一个负责搜索、一个负责分析、一个负责生成。结果呢？A 刚处理完的数据被 B 覆盖，C 拿到的上下文是过期的，监控日志满天飞但定位不到根因。一地鸡毛。

这篇就来填那个坑。我从自己的实战经历出发，分享多 Agent 协作的几种模式、通信机制、以及摔出来的几条经验。

什么时候该拆成多 Agent？

先泼盆冷水：不是所有场景都适合多 Agent。我自己有三个判断标准：

判断条件	适合拆多 Agent	单 Agent 就够了
任务是否需要不同专业技能	需要不同角色（搜索 vs 分析 vs 写作）	一个角色能搞定
子任务是否可以并行	明显可并行处理	串行依赖强，必须按顺序
是否需要隔离的上下文/记忆	各 Agent 有自己的状态空间	共享一个上下文就行

我踩过的坑：一上来就把一个简单的信息提取任务拆成”阅读 Agent”+”总结 Agent”+”校验 Agent”三个角色，结果 80% 的开销花在了 Agent 之间传消息，比单 Agent 慢了两倍。拆之前一定先问自己：这个问题真的需要多角色协作吗？

三种主流协作模式

我实践下来，大部分多 Agent 场景逃不出下面三种模式：

Pipeline 流水线模式

最简单的模式——Agent A 到 Agent B 再到 Agent C，数据像流水线一样依次传递。适合有明确处理阶段的场景，比如文档处理：先分块、再提取、再格式化。

用户输入 -> 预处理 Agent -> 分析 Agent -> 格式化 Agent -> 输出

优点是简单直接，每步输出都能单独检查。缺点是不能并行，中间一个节点挂了整个链都断。所以我在 Pipeline 里会给每个节点加超时和降级策略——超时了就跳过去用默认值。

Supervisor 模式（我最常用的）

一个监督 Agent（Supervisor）负责理解整体任务、拆解成子任务、分派给不同 Worker、收集结果、做最终汇总。Worker 各管一摊，不需要知道系统全貌。

            +--- Worker A (搜索/检索)
            |
用户 -> Supervisor ---- Worker B (数据分析)
            |           |
            +--------- Worker C (综合生成)
                 |
              汇总结果 -> 用户

这个模式我用得最多。关键设计点：Supervisor 的决策逻辑要尽量轻——它只负责”派活”和”汇总”，不参与具体处理。如果 Supervisor 本身变成了瓶颈，就说明拆分的粒度不对。

Swarm 平等模式

多个 Agent 地位平等，各有专长，通过交换信息逐步达成共识。典型场景是”多 Agent 辩论”——让两个 Agent 扮演不同角色讨论同一个问题，反复迭代后取最优解。

LangGraph 的 Send API 就是为这种模式设计的——它允许一个节点有条件地向多个下游节点并行发送消息。我今年在做一个方案评审系统时就用到了这个：一个 Agent 写方案，另一个 Agent 挑刺，来回三轮后给出最终版本，效果比单 Agent 自评自改好不少。

通信协议：统一消息格式

这是多 Agent 系统最容易翻车的地方。我一开始让每个 Agent 自己想传什么就传什么，结果很快变成了灾难——A 传的是 Markdown 字符串，B 传的是 JSON 数组，C 又传的是 Pydantic 对象。调试的时候脑子都快炸了。

我的做法：统一一个消息结构，所有 Agent 都遵守同一套协议。

from dataclasses import dataclass, field
from typing import Any

@dataclass
class AgentMessage:
    sender: str           # 发送者
    receiver: str         # 接收者
    msg_type: str         # request | response | error | status
    payload: dict         # 实际数据
    trace_id: str = ""    # 链路追踪 ID
    timestamp: float = 0.0

就这么一个简单的 dataclass，把日志和追踪的复杂度降了一大截。每个消息都带 trace_id，配合上篇讲的可观测性工具，一条任务经过的所有 Agent 步骤都能串起来看。

冲突与死锁：三招避坑

多 Agent 最隐形的坑是死锁和竞态冲突。我吃过几次亏后总结了三条原则：

1. 每个 Agent 调用设超时。 不管是 LLM 调用还是工具调用，都设一个合理的 timeout。超过时间就降级（返回”查不到”）而不是死等。Supervisor 根据超时结果决定重试还是跳过。

2. 幂等性设计。 Worker 处理任务要设计成幂等的——同一个任务执行一次和两次结果一样。这样重试就安全了，不会因为网络抖动导致重复处理出问题。

3. 版本号乐观锁。 多个 Agent 写同一个状态时，加一个 version 字段。更新前检查版本号，如果被其他 Agent 改过了就重新读取再更新。

我之前系统就吃过这个亏：两个 Worker 同时更新同一个任务的”状态”字段，互相覆盖，最后显示”已完成”但实际数据是空的。加了 version 乐观锁之后这个问题再没出现过。

实战代码：用 LangGraph 搭 Supervisor 模式

下面是一个用 LangGraph 实现 Supervisor 模式的简化示例。我让 Supervisor 决定把问题派给哪个 Worker，Worker 处理完后回到 Supervisor 做下一步决策。

from langgraph.graph import StateGraph, MessagesState
from langgraph.types import Command
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# 定义两个 Worker
search_agent = create_react_agent(
    ChatOpenAI(model="gpt-4o-mini"),
    tools=[],
    prompt="你是一个搜索专家，负责查找最新信息。"
)

analyze_agent = create_react_agent(
    ChatOpenAI(model="gpt-4o-mini"),
    tools=[],
    prompt="你是一个数据分析师，擅长分析技术方案。"
)

# Supervisor 节点：看完当前状态，决定下一步去哪
def supervisor(state: MessagesState):
    llm = ChatOpenAI(model="gpt-4o")
    messages = state["messages"]
    
    router_prompt = "Based on the conversation, who should go next? search / analyze / aggregate"
    
    response = llm.invoke([
        ("system", router_prompt),
        *messages
    ])
    
    content = response.content.lower()
    if "search" in content:
        return Command(goto="search_worker",
                       update={"messages": [response]})
    elif "analyze" in content:
        return Command(goto="analyze_worker",
                       update={"messages": [response]})
    else:
        return Command(goto="aggregator",
                       update={"messages": [response]})

# 构建状态图
builder = StateGraph(MessagesState)
builder.add_node("supervisor", supervisor)
builder.add_node("search_worker",
    lambda s: {"messages": search_agent.invoke(s)["messages"]})
builder.add_node("analyze_worker",
    lambda s: {"messages": analyze_agent.invoke(s)["messages"]})
builder.add_node("aggregator", 
    lambda s: {"messages": [("assistant", "汇总完成")]})

builder.set_entry_point("supervisor")
builder.add_edge("search_worker", "supervisor")
builder.add_edge("analyze_worker", "supervisor")
builder.add_edge("aggregator", "__end__")

graph = builder.compile()

# 运行
result = graph.invoke({
    "messages": [("user", "了解最新的 AI Agent 框架对比")]
})

这段代码需要安装 langgraph 和 langchain-openai：pip install langgraph langchain-openai，再加上有效的 OpenAI API Key 就能跑。核心思路是 Supervisor 做动态路由，每个 Worker 只处理自己的专业领域。

几点实战建议

先单后多：先用一个 Agent 把端到端逻辑跑通，再拆成多 Agent。不要一上来就画复杂的协作图。
独立测试每个 Worker：整合之前先确保每个 Worker 单独调用时表现符合预期。不然出了问题你不知道是协作逻辑的锅还是单个 Worker 的锅。
日志里带上角色名：每个 Agent 打印日志时带上自己的名字加 trace_id，不然多个 Agent 的日志混在一起根本没法看。

总结

多 Agent 协作不是银弹，但用对场景能大幅提升系统的灵活性和专业化。核心就三件事：选对协作模式（Pipeline / Supervisor / Swarm）、统一通信协议降低心智负担、做好冲突兜底（超时 + 幂等 + 版本锁）。

下期预告
Agent 能调外部工具、能访问内部数据之后，安全问题就绕不过去了。下一篇我写 Agent 安全与防护——prompt 注入怎么防、Agent 权限怎么管控、沙箱策略怎么设计。感兴趣的可以持续关注。

参考文献

LangGraph 官方文档 — Multi-Agent Systems: https://langchain-ai.github.io/langgraph/concepts/multi_agent/
LangGraph Send API: https://langchain-ai.github.io/langgraph/how-tos/map-reduce/
OpenAI Function Calling: https://platform.openai.com/docs/guides/function-calling
Andrew Ng — AI Agentic Design Patterns: https://www.deeplearning.ai/the-batch/how-agents-can-improve-llm-performance/
AutoGen — Multi-Agent Conversations: https://microsoft.github.io/autogen/stable/user-guide/core-user-guide/concepts/conversation-patterns.html

转载请注明：Falost的小窝 » 多 Agent 协作架构实战：分工、通信、冲突，别让你的 Agent 们打起来

如果你觉得这篇文章不错或者对你有帮助，想请我喝一杯咖啡，可以打赏

打赏

Falost的小窝一名热爱Hack技术的前端程序员，神棍自留地！