多智能体系统实战:基于LangGraph的角色分工与协作机制
2026/9/24 22:13:26 网站建设 项目流程

1. 从单兵作战到团队协同:为什么需要多智能体

1.1 单智能体的天花板在哪里

刚开始接触 Agent 开发的时候,我也是从单智能体入手的。一个 LLM 加上几个工具,套一个 ReAct 循环,就能做出挺像样的东西——查资料、写代码、调 API,一个人全包。但项目稍微复杂一点,问题就全冒出来了。

最典型的场景:你让一个 Agent 同时负责需求分析、代码生成和测试验证。它会在一个上下文窗口里塞进三种完全不同的思维模式,结果就是——需求分析做得不够深入,代码生成时忘了前面的约束,测试环节又因为上下文太长开始丢信息。这不是模型能力的问题,是架构的问题。

单智能体的核心瓶颈有三个。第一是上下文污染,不同任务的信息互相干扰,前面聊的需求细节会干扰后面写代码的判断。第二是工具过载,当一个 Agent 挂载了十几个工具,它在选择工具时的准确率会明显下降,实测下来工具超过 8 个之后选错率就开始飙升。第三是无法并行,所有事情串行处理,一个环节卡住整个流程就停了。

打个比方,单智能体就像一个全能但只有一个人的创业公司,什么都要自己干,看着灵活,实际上每件事都做不到极致。

1.2 多智能体到底解决了什么问题

多智能体的思路很直接:既然一个人干不好所有事,那就拆成几个人,每人负责自己擅长的部分,通过协作完成整体目标。

具体来说,多智能体架构带来三个核心收益。专业化分工让每个 Agent 的系统提示词更聚焦,工具集更精简,在自己领域内的表现明显优于通才 Agent。上下文隔离意味着每个 Agent 维护自己的对话历史,不会互相污染,代码 Agent 不需要知道需求讨论过程中的所有细节,只需要拿到最终的需求文档就行。并行执行则让没有依赖关系的任务可以同时跑,比如前端代码和后端代码可以两个 Agent 同时写。

但这里有个关键认知:多智能体不是银弹。它引入了新的复杂度——通信开销、状态同步、错误传播。如果任务本身很简单,硬上多智能体就是过度设计。我的经验判断标准是:当你的单 Agent 系统提示词超过 2000 字、工具超过 6 个、或者流程中有明显的阶段划分时,就该考虑拆成多智能体了。

1.3 什么样的项目适合多智能体架构

不是所有项目都需要多智能体。我踩过的坑是:早期有个简单的文档问答项目,硬拆成了检索 Agent、总结 Agent、格式化 Agent 三个角色,结果延迟翻了三倍,效果还不如一个 Agent 加好一点的提示词。

适合多智能体的场景通常有这几个特征:任务可以自然分解为多个专业领域,比如软件开发中的需求、设计、编码、测试;不同阶段需要不同的工具集,比如调研阶段需要搜索工具,编码阶段需要代码执行工具;流程中有明确的检查点或审核环节,比如代码写完需要 review 才能合并。

反过来,如果你的任务是一个连续的、不可分割的推理过程,比如数学证明或者单轮问答,那单智能体反而是更好的选择。多智能体的价值在于协作,没有协作需求就不要硬造协作。

2. 角色分工的设计方法论

2.1 怎么拆角色才合理

角色拆分是多智能体设计中最关键的一步,拆得好事半功倍,拆得不好就是灾难。我总结了一个原则:按职责边界拆,不按功能点拆

什么意思?举个例子,你要做一个竞品分析系统。按功能点拆可能是:搜索 Agent、爬取 Agent、分析 Agent、写报告 Agent。但这样拆的问题是,搜索和爬取本质上是一件事的两个步骤,硬拆开反而增加了通信成本。按职责边界拆应该是:信息收集 Agent(负责搜索和爬取)、数据分析 Agent(负责整理和分析)、报告撰写 Agent(负责输出)。

判断职责边界是否清晰的三个标准:每个 Agent 的输入输出是否明确、Agent 之间是否需要频繁来回沟通、一个 Agent 的失败是否会影响其他 Agent 的核心逻辑。如果两个 Agent 之间需要来回沟通超过三次才能完成一个任务,那大概率它们应该合并。

2.2 常见角色类型与职责定义

在实际项目中,我遇到过和设计过的角色类型大致可以归为几类。规划者负责拆解任务、制定执行计划,它通常不直接干活,而是输出一个结构化的任务列表。执行者是真正干活的角色,比如代码生成、文档撰写、数据查询。审核者负责检查执行者的输出质量,发现问题后打回重做。协调者在多 Agent 需要并行工作时负责汇总结果、解决冲突。

以 LangGraph 为例,一个典型的软件开发多智能体系统可能包含这些角色:

角色职责核心工具输出物
需求分析师理解用户需求,输出需求文档无特殊工具结构化需求文档
架构师根据需求设计技术方案知识库检索技术方案文档
编码 Agent按方案写代码代码执行、文件读写代码文件
测试 Agent验证代码正确性代码执行、测试框架测试报告
审核 Agent最终质量把关无特殊工具审核意见

这张表的关键在于:每个角色的输出物是下一个角色的输入,形成清晰的流水线。工具分配也遵循最小权限原则,需求分析师不需要代码执行工具,编码 Agent 不需要搜索工具。

2.3 角色提示词的设计要点

每个 Agent 的系统提示词决定了它的行为边界。我见过最常见的错误是提示词写得太泛,比如“你是一个有帮助的助手”——这种提示词在多智能体系统里等于没写。

好的角色提示词应该包含四个要素。身份定义要具体到专业领域,不是“你是程序员”而是“你是一个专注于 Python 后端开发的工程师,擅长 FastAPI 和 PostgreSQL”。职责边界要明确说什么该做什么不该做,“你只负责生成代码,不负责运行测试,测试由测试 Agent 完成”。输出格式要严格约定,因为下游 Agent 需要解析你的输出,“以 JSON 格式返回,包含 code 和 explanation 两个字段”。协作协议要说明遇到问题怎么办,“如果需求不明确,在输出中标记 NEED_CLARIFICATION 并说明具体问题”。

这里有个实操技巧:提示词里的输出格式约定,最好用具体的 schema 而不是自然语言描述。比如用 Pydantic 模型定义输出结构,然后在提示词里直接引用这个 schema,这样下游解析的成功率会高很多。

3. 协作机制的核心实现

3.1 通信模式:消息传递还是共享状态

多智能体之间的通信有两种基本模式。消息传递是 Agent A 直接把结果发给 Agent B,像打电话。共享状态是所有 Agent 读写同一个状态对象,像白板。

LangGraph 采用的是共享状态模式,所有 Agent 节点读写同一个 State。这种模式的好处是解耦——Agent A 不需要知道谁会消费它的输出,只需要把结果写到 State 的对应字段就行。坏处是 State 的结构设计变得很关键,字段太多会导致每个 Agent 都要处理大量无关信息。

我的经验是:流程固定的用共享状态,流程动态的用消息传递。比如软件开发流水线,需求→设计→编码→测试,流程是确定的,用共享状态最清晰。但如果是客服系统,用户可能问任何问题,需要动态路由到不同 Agent,那消息传递加路由更合适。

在 LangGraph 里,State 通常用 TypedDict 定义:

from typing import TypedDict, Annotated from langgraph.graph import add_messages class DevState(TypedDict): messages: Annotated[list, add_messages] requirements: str design_doc: str code: str test_report: str current_phase: str

注意Annotated[list, add_messages]这个写法,它告诉 LangGraph 这个字段是追加而不是覆盖。这个细节很关键,我当初没注意,导致消息历史一直被覆盖,调试了半天才发现。

3.2 编排模式:流水线、路由还是层级

协作的编排方式决定了 Agent 之间的调用关系。常见的三种模式各有适用场景。

流水线模式最简单,Agent 按固定顺序依次执行,A 的输出是 B 的输入。适合流程确定的场景,比如内容生产:选题→写作→编辑→发布。实现上用 LangGraph 的线性图就行,每个节点是一个 Agent。

路由模式需要一个路由 Agent 根据当前状态决定下一步调用哪个 Agent。适合分支逻辑多的场景,比如客服系统根据用户意图路由到技术支持、退款处理或投诉建议。路由 Agent 本身通常不干活,只做决策。

层级模式是树状结构,顶层 Agent 负责拆解任务,把子任务分配给下层 Agent,下层 Agent 还可以继续往下分。适合复杂项目,比如一个完整的软件项目,顶层是项目经理 Agent,下面是前端、后端、测试三个组长 Agent,每个组长再管理具体的执行 Agent。

实际项目中这三种模式经常混用。我做过的一个系统就是:顶层用层级模式做任务拆解,每个子团队内部用流水线模式执行,团队之间用路由模式做协调。

3.3 状态管理与上下文传递

多智能体系统里,状态管理是最容易出 bug 的地方。核心问题是:每个 Agent 需要看到多少上下文

全量共享是最简单的做法,所有 Agent 都能看到完整的 State。但这样会导致两个问题:上下文窗口浪费,每个 Agent 都要处理大量无关信息;信息泄露,比如审核 Agent 看到了编码 Agent 的中间思考过程,可能产生偏见。

我的做法是按需裁剪。每个 Agent 节点在执行前,从全局 State 中提取自己需要的字段,组装成自己的上下文。比如编码 Agent 只需要 requirements 和 design_doc,不需要看到需求分析阶段的原始对话记录。

def coding_agent(state: DevState): # 只提取需要的字段 context = { "requirements": state["requirements"], "design": state["design_doc"] } # 执行编码逻辑 result = llm.invoke(build_prompt(context)) # 只写回自己负责的字段 return {"code": result, "current_phase": "coding_done"}

这种做法的另一个好处是,当某个 Agent 需要升级或替换时,只要保持输入输出接口不变,内部实现随便改。

4. 基于 LangGraph 的实操落地

4.1 环境搭建与项目结构

先说环境。LangGraph 的依赖很轻量,核心就是langgraphlangchain-core,但如果要用到具体的模型和工具,还需要装对应的包。

pip install langgraph langchain-core langchain-openai

项目结构我建议这样组织,这是踩过坑之后总结的:

project/ ├── agents/ │ ├── __init__.py │ ├── planner.py # 规划 Agent │ ├── coder.py # 编码 Agent │ ├── tester.py # 测试 Agent │ └── reviewer.py # 审核 Agent ├── graph/ │ ├── __init__.py │ ├── state.py # State 定义 │ └── workflow.py # 图编排 ├── tools/ │ └── __init__.py ├── config.py └── main.py

每个 Agent 一个文件,State 定义单独放,图编排单独放。这样当 Agent 数量增多时不会乱。我见过把所有东西塞一个文件的,超过三个 Agent 之后基本没法维护。

4.2 定义 State 与 Agent 节点

State 的设计原则是:字段对应流程中的关键产物,而不是 Agent 的内部状态。比如 requirements、design_doc、code 这些是产物,而“编码 Agent 当前在第几轮迭代”这种是内部状态,不应该放进全局 State。

from typing import TypedDict, Annotated, Literal from langgraph.graph import add_messages class TeamState(TypedDict): messages: Annotated[list, add_messages] task: str plan: str code: str test_result: str review_status: Literal["approved", "rejected", "pending"] iteration: int

每个 Agent 节点本质上就是一个函数,接收 State 返回 State 的更新部分:

from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o", temperature=0) def planner_node(state: TeamState): prompt = f"""你是一个技术规划师。根据以下任务制定执行计划: 任务:{state['task']} 输出格式: 1. 任务拆解 2. 技术选型 3. 关键风险 """ response = llm.invoke(prompt) return {"plan": response.content, "iteration": 0}

这里有个细节:temperature=0对于需要稳定输出的 Agent 很重要,特别是规划者和审核者。编码 Agent 可以稍微高一点,0.2 左右,让它有一点灵活性。

4.3 构建协作图与条件路由

图编排是 LangGraph 的核心。先定义节点,再定义边,最后编译成可执行图。

from langgraph.graph import StateGraph, END workflow = StateGraph(TeamState) # 添加节点 workflow.add_node("planner", planner_node) workflow.add_node("coder", coder_node) workflow.add_node("tester", tester_node) workflow.add_node("reviewer", reviewer_node) # 设置入口 workflow.set_entry_point("planner") # 添加边 workflow.add_edge("planner", "coder") workflow.add_edge("coder", "tester") # 条件边:测试通过去审核,不通过回编码 def test_router(state: TeamState): if "PASS" in state["test_result"]: return "reviewer" return "coder" workflow.add_conditional_edges( "tester", test_router, {"reviewer": "reviewer", "coder": "coder"} ) # 审核条件边 def review_router(state: TeamState): if state["review_status"] == "approved": return END return "coder" workflow.add_conditional_edges( "reviewer", review_router, {"coder": "coder", END: END} ) app = workflow.compile()

这段代码里最关键的是两个条件路由。测试不通过回编码,审核不通过也回编码,形成一个带反馈的循环。但这里有个坑:必须设置最大迭代次数,否则可能无限循环。我一般会在 State 里加 iteration 字段,每次回退加一,超过阈值就强制结束并报错。

4.4 加入 Human-in-the-Loop

实际生产环境里,完全自动化的多智能体系统风险很高。LangGraph 提供了 interrupt 机制,可以在关键节点暂停,等人工确认后再继续。

from langgraph.checkpoint.memory import MemorySaver # 编译时加入 checkpointer memory = MemorySaver() app = workflow.compile( checkpointer=memory, interrupt_before=["reviewer"] # 审核前暂停 ) # 执行 config = {"configurable": {"thread_id": "task-001"}} result = app.invoke({"task": "写一个用户登录接口"}, config) # 人工检查后继续 app.invoke(None, config)

interrupt_before指定在哪些节点前暂停。我通常会在两个地方加人工介入:一是规划完成后,确认计划没问题再往下走;二是最终审核前,人工看一眼代码质量。这样既保留了自动化的效率,又避免了完全失控的风险。

5. 常见问题与排查技巧实录

5.1 Agent 之间信息传递丢失

这是最高频的问题。表现是下游 Agent 拿不到上游的输出,或者拿到的是空值。排查思路按这个顺序来。

先检查 State 字段名是否一致。上游写的是design_doc,下游读的是design,这种拼写不一致在 TypedDict 里不会报错,但运行时就是拿不到数据。我建议用常量定义字段名,避免手写字符串。

再检查返回值格式。LangGraph 的节点函数返回的字典会被合并到 State 里,但如果你返回的是{"design_doc": None},它会把原来的值覆盖成 None。所以返回前要确认值不为空。

最后检查条件路由。如果路由函数返回了未在映射中定义的 key,LangGraph 会静默失败,后续节点不会执行。这个坑我踩过,调试了半小时才发现是路由返回值写错了。

5.2 无限循环与死锁

多智能体系统里,Agent A 等 Agent B 的输出,Agent B 又等 Agent A 的输出,就死锁了。或者测试不通过回编码,编码改了测试还是不通过,来回循环。

防死锁的核心是设置迭代上限。在 State 里加一个计数器,每次循环加一,超过阈值就强制走异常出口。

def test_router(state: TeamState): if state["iteration"] > 5: return "force_end" if "PASS" in state["test_result"]: return "reviewer" return "coder"

另一个技巧是给回退加约束。不要让编码 Agent 无限制地重写,而是在回退时把测试报告作为反馈传给它,让它有针对性地修改。我实测下来,带反馈的回退比盲目重写,平均迭代次数从 4.2 次降到 1.8 次。

5.3 输出格式不一致导致解析失败

下游 Agent 需要解析上游的输出,但 LLM 的输出格式经常不稳定。今天返回 JSON,明天返回 Markdown,后天加了一段解释文字。

解决方案是结构化输出。LangChain 支持用 Pydantic 模型约束输出格式:

from pydantic import BaseModel, Field class CodeOutput(BaseModel): code: str = Field(description="生成的代码") explanation: str = Field(description="代码说明") dependencies: list[str] = Field(description="依赖列表") structured_llm = llm.with_structured_output(CodeOutput) result = structured_llm.invoke(prompt) # result 一定是 CodeOutput 类型,字段一定存在

用了结构化输出之后,解析失败率从大概 15% 降到了接近 0。代价是稍微增加了一点延迟,但完全值得。

5.4 常见问题速查表

问题现象可能原因排查方法解决方案
下游拿不到数据字段名不一致打印 State 内容统一用常量定义字段名
无限循环缺少迭代上限看 iteration 值加计数器强制退出
解析失败输出格式不稳定检查原始输出用结构化输出
Agent 选错工具工具描述不清看工具调用日志优化工具描述,减少工具数
响应太慢串行执行看各节点耗时无依赖的节点并行化
结果质量差提示词太泛检查系统提示词加具体身份和输出约束

6. 多智能体系统的进阶优化

6.1 记忆机制的设计

多智能体系统里,记忆分两层。短期记忆是当前任务的上下文,存在 State 里,任务结束就清空。长期记忆是跨任务的知识积累,比如用户偏好、历史决策、常见错误模式。

LangGraph 的 checkpointer 天然支持短期记忆,每个 thread_id 对应一个独立的会话。长期记忆需要自己实现,我通常用向量数据库存历史任务的摘要,在新任务开始时检索相关经验注入到规划 Agent 的提示词里。

这里有个经验:长期记忆不要存原始对话,要存结构化的经验条目。比如“用户 X 偏好用 TypeScript 而不是 JavaScript”、“上次类似任务在数据库连接池配置上出过问题”。原始对话太长,检索效率低,而且噪音大。

6.2 错误处理与降级策略

多智能体系统里,一个 Agent 失败不应该导致整个系统崩溃。我的做法是每个 Agent 节点都有 try-catch,失败时返回一个标记了错误的状态,由路由决定是重试、跳过还是终止

def coder_node(state: TeamState): try: result = structured_llm.invoke(prompt) return {"code": result.code, "error": None} except Exception as e: return {"code": "", "error": str(e)}

然后在路由里判断:

def after_coder(state: TeamState): if state.get("error"): if state["iteration"] < 3: return "retry_coder" return "fallback" return "tester"

降级策略也很重要。如果编码 Agent 连续失败,可以降级到用一个更简单的模板生成代码,或者直接转人工处理。关键是不要让系统卡死。

6.3 性能优化:并行与缓存

多智能体系统天然比单 Agent 慢,因为多了通信和编排开销。优化手段主要有两个。

并行执行:没有依赖关系的 Agent 可以同时跑。LangGraph 支持并行节点,比如前端代码和后端代码可以两个 Agent 同时生成。实测下来,并行化能把总耗时降低 40% 左右。

缓存:相同的输入不需要重复调用 LLM。我通常用 LangChain 的缓存机制,对规划、审核这类确定性高的节点开启缓存。编码节点因为需要多样性,一般不缓存。

from langchain_core.caches import InMemoryCache from langchain_core.globals import set_llm_cache set_llm_cache(InMemoryCache())

这个缓存对开发调试特别有用,改下游逻辑的时候不用每次都重新跑上游的 LLM 调用,省时省钱。

6.4 评估与迭代

多智能体系统上线后,怎么知道它好不好?我一般从三个维度评估。任务完成率是最基本的,多少任务最终成功完成。平均迭代次数反映效率,迭代次数越多说明 Agent 之间的配合越差。人工介入率反映自动化程度,需要人工干预的比例越低越好。

评估数据从 LangGraph 的 trace 里拿,每个节点的输入输出、耗时、状态变化都有记录。我习惯每周跑一次评估,看指标趋势,如果某个指标恶化就针对性优化对应的 Agent。

迭代的时候一次只改一个 Agent,改完跑评估对比。同时改多个 Agent 会导致无法定位是哪个改动带来的效果变化,这是血泪教训。

7. 我踩过的坑与实战心得

7.1 不要过早引入多智能体

这是我最大的教训。有个项目一开始就设计了五个 Agent,结果开发了两周发现,其中三个 Agent 的职责用提示词工程就能在一个 Agent 里解决。多智能体带来的通信开销和调试复杂度,远远超过了它带来的收益。

正确的做法是:先用单 Agent 跑通,遇到明确的瓶颈再拆。拆的时候也要一个一个拆,拆一个验证一个,不要一次性全拆。

7.2 提示词比架构更重要

我见过太多人花大量时间设计复杂的 Agent 拓扑,却用着敷衍的提示词。实际上,一个提示词写得好的单 Agent,效果往往超过提示词写得差的多 Agent 系统。

每个 Agent 的提示词都值得反复打磨。我的习惯是:先写一版,跑 20 个测试用例,看失败案例,针对性修改提示词,再跑。通常迭代 3-5 轮才能稳定。

7.3 日志和可观测性是生命线

多智能体系统的调试难度是单 Agent 的好几倍。没有完善的日志,出了问题根本不知道是哪个 Agent 的哪一步出了错。

我的做法是:每个 Agent 节点的输入输出都打日志,包括完整的提示词和原始响应。用 LangSmith 或者自己搭一个简单的日志系统都行。关键是出问题时能快速定位。

7.4 从小处着手,逐步扩展

最后一个心得:不要一上来就搞复杂的层级编排和动态路由。先从最简单的流水线开始,两个 Agent,A 的输出给 B。跑通了,再加第三个。加条件路由,加循环,加人工介入。每一步都验证稳定了再往下走。

我现在的习惯是:新项目先用一个周末搭一个两 Agent 的最小原型,验证核心流程能跑通,再花时间扩展。这样即使方向错了,损失也有限。

多智能体系统本质上是在用工程手段弥补单个 LLM 的能力边界。它不是越复杂越好,而是在合适的地方用合适的复杂度。理解每个 Agent 的能力边界,设计清晰的协作协议,做好错误处理和可观测性,剩下的就是迭代打磨了。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询