多智能体（Multi-Agent）编排实战：用LangGraph构建生产级AI系统

程序员小橙

898人浏览 · 2026-03-19 15:46:38

程序员小橙 · 2026-03-19 15:46:38 发布

一、引言

过去两年，大语言模型（LLM）驱动的AI Agent已经从概念验证走向了生产应用。然而，随着业务场景复杂度的提升，单一Agent的局限性逐渐暴露：上下文窗口有限、职责过于宽泛导致推理质量下降、难以处理需要多步协作的复杂任务流。

多智能体（Multi-Agent）系统应运而生。其核心思想并不新颖——"分而治之"在软件工程中早已被反复验证。将一个复杂任务拆解为多个专注于特定子任务的Agent，通过编排层进行协调，不仅能提升单项任务的执行质量，还能带来更好的可观测性和可维护性。

在众多多智能体框架中，LangGraph凭借其基于有向图的状态机模型、对人机交互（Human-in-the-Loop）的原生支持以及与LangChain生态的深度集成，成为构建生产级多智能体系统的优选方案。本文将从架构模式出发，逐步深入到LangGraph的核心机制，最终通过一个完整的代码审查流水线实战，帮助读者掌握多智能体系统的设计与实现。

二、多智能体架构模式

在设计多智能体系统之前，理解主流的架构模式至关重要。不同的模式适用于不同的场景，选错模式往往意味着后期大量的返工。

2.1 Supervisor模式（主管模式）

Supervisor模式是最直观也最常用的多智能体架构。一个中心化的"主管"Agent负责接收用户请求，将任务分发给下属的Worker Agent，收集结果并进行汇总。

其工作流程如下：

用户请求进入Supervisor Agent
Supervisor分析任务，决定调用哪个Worker
Worker执行具体任务并返回结果
Supervisor根据结果决定是否需要继续分发，或汇总输出

这种模式的优势在于控制流清晰，Supervisor掌握全局上下文，便于做全局决策。劣势则是Supervisor本身可能成为瓶颈——当Worker数量增多时，Supervisor需要理解所有Worker的能力边界，提示词会变得臃肿，推理质量也会随之下降。

适用场景：Worker数量在5个以内、任务分发逻辑相对明确的场景，例如客服系统中的意图识别与分发。

2.2 Hierarchical模式（层级模式）

层级模式可以看作Supervisor模式的递归扩展。当系统中Agent数量较多时，引入多层级的管理结构：顶层Supervisor管理若干中层Supervisor，中层Supervisor再管理各自的Worker Agent。

这种树状结构天然地解决了单一Supervisor的能力瓶颈问题。每一层只需要关注其直接下属的能力，极大降低了单个Agent的认知负荷。

适用场景：大型复杂系统，例如端到端的软件开发流水线——顶层编排器协调"需求分析组"、"开发组"、"测试组"，每个组内部又有各自的Agent分工。

2.3 Swarm模式（群体智能模式）

Swarm模式打破了中心化的控制结构。在这种模式下，没有固定的"主管"，Agent之间可以直接进行对等通信，通过"移交"（Handoff）机制将控制权从一个Agent转移到另一个Agent。

OpenAI在2024年开源的Swarm框架让这种模式广为人知。其核心机制非常简洁：每个Agent定义自己的指令和工具，其中某些工具的返回值是另一个Agent对象，触发控制权的转移。

Swarm模式的优势在于灵活性极高，Agent之间的协作关系可以动态变化。但也正因如此，系统的行为变得难以预测和调试，不适合对确定性要求较高的生产场景。

适用场景：探索性任务、开放域对话等对灵活性要求高于确定性的场景。

2.4 模式选型建议

维度	Supervisor	Hierarchical	Swarm
控制流确定性	高	高	低
扩展性	中	高	高
实现复杂度	低	中	中
可观测性	好	好	差
推荐Agent数量	2-5	5-20	3-10

在实际项目中，混合使用多种模式是常见的做法。例如，顶层采用Supervisor模式保证控制流的确定性，某个子模块内部采用Swarm模式保留灵活性。

三、LangGraph核心概念

LangGraph是LangChain团队推出的Agent编排框架，其核心抽象是状态图（StateGraph）。理解以下几个核心概念，是使用LangGraph构建多智能体系统的前提。

3.1 State（状态）

State是贯穿整个图执行过程的共享数据结构。所有Node读取和修改的都是同一份State。在LangGraph中，State通常使用TypedDict或Pydantic模型定义：

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

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    current_file: str
    review_results: list[dict]
    final_report: str

注意Annotated[list, add_messages]这个写法——add_messages是一个reducer函数，它告诉LangGraph当多个节点都向messages字段写入时，应该采用追加而非覆盖的策略。Reducer机制是LangGraph处理并发写入冲突的核心手段。

3.2 Node（节点）

Node是图中的计算单元，通常是一个Python函数或一个可调用对象。每个Node接收当前State作为输入，返回需要更新的State字段：

def code_review_node(state: AgentState) -> dict:
    """代码审查节点：调用LLM对当前文件进行审查"""
    messages = state["messages"]
    current_file = state["current_file"]
    # ... 调用LLM进行审查
    return {
        "messages": [AIMessage(content=review)],
        "review_results": [{"file": current_file, "review": review}]
    }

3.3 Edge（边）

Edge定义了Node之间的转移关系。LangGraph支持两种边：

5.4 成本控制

多智能体系统的LLM调用次数是单Agent的数倍，成本控制不可忽视：

5.5 测试策略

多智能体系统的测试需要分层进行：

多智能体系统并非银弹，它带来能力提升的同时也引入了额外的复杂度。在决定是否采用多智能体架构之前，建议先问自己两个问题：单一Agent是否真的无法满足需求？拆分后的收益是否大于引入的复杂度？

如果答案是肯定的，那么本文介绍的方法论可以作为参考：

多智能体编排仍然是一个快速演进的领域。LangGraph在2025年的多次重大更新中引入了对长时运行任务、远程Agent（A2A协议）的支持，进一步拓展了多智能体系统的应用边界。保持对新模式和新工具的关注，在实践中不断迭代，才是构建生产级AI系统的正确姿态。

普通边：无条件转移，A执行完毕后必定执行B
条件边：根据当前State动态决定下一个要执行的Node

def route_after_review(state: AgentState) -> str:
    """根据审查结果决定下一步"""
    results = state["review_results"]
    if any(r["severity"] == "critical" for r in results):
        return "human_review"  # 转入人工审核
    return "generate_report"   # 直接生成报告

graph.add_conditional_edges(
    "code_review",
    route_after_review,
    {"human_review": "human_review", "generate_report": "report_generator"}
)

3.4 子图（Subgraph）

LangGraph支持将一个完整的StateGraph作为另一个图的Node嵌入，这是实现Hierarchical模式的基础。子图拥有自己的独立状态空间，通过输入/输出映射与父图通信：

# 定义子图
review_subgraph = StateGraph(ReviewState)
# ... 添加节点和边

# 将子图作为节点嵌入主图
main_graph.add_node("review_pipeline", review_subgraph.compile())

四、实战：构建代码审查流水线

接下来我们用LangGraph构建一个多智能体代码审查流水线。该系统包含三个专业Agent：代码风格检查Agent、安全漏洞扫描Agent和逻辑审查Agent，由一个Supervisor Agent负责编排。

4.1 定义状态与Agent

import operator
from typing import TypedDict, Annotated, Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langgraph.graph import StateGraph, END, START

# 定义全局状态
class ReviewState(TypedDict):
    messages: Annotated[list, operator.add]
    code_snippet: str
    style_review: str
    security_review: str
    logic_review: str
    final_report: str
    next_agent: str

# 初始化LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 代码风格检查Agent
def style_checker(state: ReviewState) -> dict:
    prompt = f"""你是一位资深的代码风格审查专家。请审查以下代码的风格问题，
包括但不限于：命名规范、代码格式、注释质量、函数长度等。
仅输出风格相关的问题，不要涉及安全或逻辑问题。

代码：
{state['code_snippet']}"""

    response = llm.invoke([SystemMessage(content="你是代码风格审查专家。"),
                           HumanMessage(content=prompt)])
    return {
        "style_review": response.content,
        "messages": [AIMessage(content=f"[风格检查完成] {response.content}")]
    }

# 安全漏洞扫描Agent
def security_scanner(state: ReviewState) -> dict:
    prompt = f"""你是一位安全审计专家。请扫描以下代码中的安全漏洞，
包括但不限于：SQL注入、XSS、敏感信息泄露、不安全的依赖等。
给出风险等级（高/中/低）和修复建议。

代码：
{state['code_snippet']}"""

    response = llm.invoke([SystemMessage(content="你是安全审计专家。"),
                           HumanMessage(content=prompt)])
    return {
        "security_review": response.content,
        "messages": [AIMessage(content=f"[安全扫描完成] {response.content}")]
    }

# 逻辑审查Agent
def logic_reviewer(state: ReviewState) -> dict:
    prompt = f"""你是一位资深软件工程师。请审查以下代码的逻辑正确性，
包括但不限于：边界条件处理、异常处理、算法正确性、性能隐患等。

代码：
{state['code_snippet']}"""

    response = llm.invoke([SystemMessage(content="你是代码逻辑审查专家。"),
                           HumanMessage(content=prompt)])
    return {
        "logic_review": response.content,
        "messages": [AIMessage(content=f"[逻辑审查完成] {response.content}")]
    }

4.2 构建Supervisor与编排图

# Supervisor：汇总所有审查结果，生成最终报告
def supervisor(state: ReviewState) -> dict:
    prompt = f"""你是代码审查团队的负责人。以下是三位专家的审查结果，
请汇总生成一份结构化的代码审查报告，包含：
1. 总体评价（通过/需修改/拒绝）
2. 关键问题列表（按严重程度排序）
3. 改进建议

风格审查结果：
{state.get('style_review', '未完成')}

安全审查结果：
{state.get('security_review', '未完成')}

逻辑审查结果：
{state.get('logic_review', '未完成')}"""

    response = llm.invoke([SystemMessage(content="你是代码审查团队负责人。"),
                           HumanMessage(content=prompt)])
    return {
        "final_report": response.content,
        "messages": [AIMessage(content=f"[最终报告] {response.content}")]
    }

# 构建状态图
workflow = StateGraph(ReviewState)

# 添加节点
workflow.add_node("style_checker", style_checker)
workflow.add_node("security_scanner", security_scanner)
workflow.add_node("logic_reviewer", logic_reviewer)
workflow.add_node("supervisor", supervisor)

# 定义边：三个审查Agent并行执行，全部完成后进入Supervisor
workflow.add_edge(START, "style_checker")
workflow.add_edge(START, "security_scanner")
workflow.add_edge(START, "logic_reviewer")
workflow.add_edge("style_checker", "supervisor")
workflow.add_edge("security_scanner", "supervisor")
workflow.add_edge("logic_reviewer", "supervisor")
workflow.add_edge("supervisor", END)

# 编译图
app = workflow.compile()

4.3 运行流水线

```
# 待审查的代码
code = """
def get_user(request):
    user_id = request.GET.get('id')
    query = f"SELECT * FROM users WHERE id = {user_id}"
    result = db.execute(query)
    return JsonResponse({'user': result[0]})
"""

# 执行审查
result = app.invoke({
    "code_snippet": code,
    "messages": [HumanMessage(content="请审查以下代码")],
    "style_review": "",
    "security_review": "",
    "logic_review": "",
    "final_report": "",
    "next_agent": ""
})

print(result["final_report"])
```
上面这段待审查代码包含了一个典型的SQL注入漏洞（字符串拼接构造SQL查询）、缺少异常处理（直接访问result[0]可能越界）以及代码风格问题（缺少类型注解和文档字符串）。三个专业Agent会分别从各自的角度识别这些问题，最终由Supervisor汇总为一份完整的审查报告。

这种架构的核心价值在于关注点分离。每个Agent只需要精通自己的领域，提示词可以写得更加精确和专业，审查质量远高于让一个通用Agent处理所有维度。

五、生产部署注意事项

从Demo到生产，多智能体系统需要跨越几个关键的工程门槛。

5.1 状态持久化与容错

LangGraph内置了Checkpointer机制，支持将每一步的State快照持久化到外部存储（如PostgreSQL、Redis）。当某个Agent执行失败时，系统可以从最近的Checkpoint恢复，而不需要从头开始执行。
```
from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver.from_conn_string("postgresql://...")
app = workflow.compile(checkpointer=checkpointer)

# 执行时指定thread_id，用于状态隔离
config = {"configurable": {"thread_id": "review-001"}}
result = app.invoke(initial_state, config=config)
```
在生产环境中，务必为每次执行分配唯一的thread_id，确保不同请求之间的状态完全隔离。

5.2 Human-in-the-Loop

对于高风险决策，引入人工审核环节是必要的。LangGraph通过interrupt_before机制原生支持这一模式：

app = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["supervisor"]  # 在Supervisor执行前暂停
)

# 第一次执行：到达supervisor前暂停
result = app.invoke(initial_state, config)

# 人工审核中间结果后，继续执行
result = app.invoke(None, config)  # 传入None表示继续

5.3 可观测性

多智能体系统的调试难度远高于单Agent系统。建议从以下几个维度建设可观测性：

追踪（Tracing）：使用LangSmith或OpenTelemetry记录每个Agent的输入输出、Token消耗和延迟。LangGraph与LangSmith的集成开箱即用，只需设置环境变量即可。
日志分级：为不同层级的Agent设置不同的日志级别。Supervisor层使用INFO级别记录决策过程，Worker层使用DEBUG级别记录详细的推理步骤。
指标监控：关注端到端延迟、单Agent延迟、Token消耗、失败率等核心指标。当某个Agent的延迟突然升高时，往往意味着提示词需要优化或上游数据质量出了问题。
模型分级：不是所有Agent都需要使用最强的模型。风格检查等相对简单的任务可以使用gpt-4o-mini，将gpt-4o留给逻辑审查等需要深度推理的任务。
缓存策略：对于相同或相似的输入，利用语义缓存避免重复调用。LangChain生态中的GPTCache等工具可以在这方面提供帮助。
并行执行：如上文示例所示，相互独立的Agent应当并行执行，以减少端到端延迟。LangGraph的图调度器会自动识别可并行的节点。
单元测试：为每个Agent编写独立的测试用例，mock掉LLM调用，验证提示词模板和输出解析逻辑。
集成测试：验证Agent之间的通信和状态传递是否正确，重点测试边界条件和异常路径。
评估测试（Eval）：准备一批标注好的测试用例，定期运行端到端评估，跟踪系统整体质量的变化趋势。LangSmith提供了完善的Eval工具链。
根据任务特征选择合适的架构模式（Supervisor / Hierarchical / Swarm）
用LangGraph的StateGraph建模Agent之间的协作关系
充分利用并行执行、状态持久化、Human-in-the-Loop等机制
在可观测性、成本控制和测试策略上投入足够的工程资源