游乐游手机版
首页/AI教程/文章详情

多智能体编排实战：使用LangGraph打造生产级AI系统

时间：2026-06-08 16:22
多智能体系统通过分而治之策略拆分复杂任务，LangGraph基于有向图状态机模型支持人机交互与LangChain生态，提供Supervisor、层级、Swarm等架构模式。以代码审查流水线为例，展示多智能体协作实现风格、安全与逻辑审查的编排流程。

一、引言

过去两年，大语言模型驱动的AI Agent已经从概念验证走向了生产应用。但说实话，随着业务场景越来越复杂，单一Agent的短板开始暴露：上下文窗口有限、职责太宽导致推理质量下降，遇到需要多步协作的复杂任务流时更是力不从心。

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

于是多智能体系统应运而生。它的核心思想说白了并不新鲜——"分而治之"在软件工程里早就被反复验证过。把一个复杂任务拆成多个专注特定子任务的Agent，通过编排层来协调，不仅能提升单项任务的执行质量，还能让整个系统更可观测、更可维护。

在众多多智能体框架中，LangGraph之所以能脱颖而出，靠的是它基于有向图的状态机模型、对人机交互（Human-in-the-Loop）的原生支持，以及和LangChain生态的深度集成。这篇文章会从架构模式切入，逐步深入到LangGraph的核心机制，最后通过一个完整的代码审查流水线实战，帮你掌握多智能体系统的设计与实现。

二、多智能体架构模式

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

2.1 Supervisor模式（主管模式）

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

工作流程大致是这样：

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

这种模式的优势在于控制流清晰，Supervisor掌握全局上下文，做全局决策很方便。缺点也很明显——Supervisor本身可能成为瓶颈：Worker一多，它就得理解所有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 模式选型建议

维度SupervisorHierarchicalSwarm
控制流确定性高高低
扩展性中高高
实现复杂度低中中
可观测性好好差
推荐Agent数量2-55-203-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支持两种边：

  • 普通边：无条件转移，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 PostgresSa ver
checkpointer = PostgresSa ver.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的延迟突然升高时，往往意味着提示词需要优化或上游数据质量出了问题。

5.4 成本控制

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

  • 模型分级：不是所有Agent都需要使用最强的模型。风格检查等相对简单的任务可以使用gpt-4o-mini，将gpt-4o留给逻辑审查等需要深度推理的任务。
  • 缓存策略：对于相同或相似的输入，利用语义缓存避免重复调用。LangChain生态中的GPTCache等工具可以在这方面提供帮助。
  • 并行执行：如上文示例所示，相互独立的Agent应当并行执行，以减少端到端延迟。LangGraph的图调度器会自动识别可并行的节点。

5.5 测试策略

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

  • 单元测试：为每个Agent编写独立的测试用例，mock掉LLM调用，验证提示词模板和输出解析逻辑。
  • 集成测试：验证Agent之间的通信和状态传递是否正确，重点测试边界条件和异常路径。
  • 评估测试（Eval）：准备一批标注好的测试用例，定期运行端到端评估，跟踪系统整体质量的变化趋势。LangSmith提供了完善的Eval工具链。

六、总结

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

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

  1. 根据任务特征选择合适的架构模式（Supervisor / Hierarchical / Swarm）
  2. 用LangGraph的StateGraph建模Agent之间的协作关系
  3. 充分利用并行执行、状态持久化、Human-in-the-Loop等机制
  4. 在可观测性、成本控制和测试策略上投入足够的工程资源

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

本文代码基于LangGraph 0.3.x版本，完整示例代码已托管至GitHub。如有疑问，欢迎在评论区交流。

来源：https://juejin.cn/post/7618405987397976104
AI系统
上一篇免费AI公文生成软件推荐，助力数字化转型 下一篇AI简历生成指南：用AI提升简历质量的详细方法与范文
本站内容用于信息整理与展示，如有侵权或内容问题请及时联系处理。

相关推荐

补充同频道和同主题内容，方便继续浏览更多相关内容。

AI系统源码应用指南：高效生成专业文档与PPT

同类最新

继续查看同栏目最近更新的文章。

更多
Kimi App手机电脑联动下载安装及浏览器兼容教程
AI教程 · 2026-06-09

Kimi App手机电脑联动下载安装及浏览器兼容教程

本文介绍了Kimi智能助手从手机端到电脑端的下载与安装方法，重点阐述了不同平台（包括iOS、Android、Windows、macOS）的获取途径。同时，详细说明了如何通过浏览器直接访问网页版，并针对主流浏览器的兼容性进行了分析，旨在帮助用户根据自身设备选择最便捷、稳定的使用方式。

 HeyGen稳定安装步骤：先配置创意团队环境再注册开通
AI教程 · 2026-06-09

HeyGen稳定安装步骤：先配置创意团队环境再注册开通

HeyGen的稳定安装与高效使用，关键在于前期团队环境的统一规划与后期账号流程的顺畅完成。团队需明确设计规范、素材管理及权限分工，为工具运行打下基础。随后，通过官方渠道完成注册、验证及订阅开通，确保服务稳定。最后进行基础功能测试与团队培训，即可快速投入实际创作流程。

 Mochi 1从零搭建本地服务与工作流导入指南
AI教程 · 2026-06-09

Mochi 1从零搭建本地服务与工作流导入指南

本文介绍了在成功完成Mochi1本地服务的基础搭建后，如何继续处理工作流导入这一关键后续步骤。内容涵盖工作流文件准备、导入操作的具体流程、常见问题的排查与解决，以及导入后的配置优化与测试验证，旨在帮助用户将预设的自动化流程顺利集成到本地环境中，确保工具发挥完整效能。

 InvokeAI Linux用户安装配置与节点处理指南
AI教程 · 2026-06-09

InvokeAI Linux用户安装配置与节点处理指南

本文详细介绍了在Linux系统上安装和配置InvokeAI的完整流程。内容涵盖从环境准备、依赖安装到模型下载与加载的关键步骤，并重点解析了核心组件“处理节点”的安装与使用方法。指南旨在帮助用户顺利完成部署，并理解其工作流程，以便更好地利用这一AI图像生成工具进行创作。

 Dify保姆级部署指南：服务安装与模型接入下载
AI教程 · 2026-06-09

Dify保姆级部署指南：服务安装与模型接入下载

本文详细介绍了开源AI应用开发平台Dify的部署流程。内容涵盖从服务器环境准备、Docker安装、Dify核心服务启动，到如何接入OpenAI、Azure等云端大模型API，以及如何配置Ollama等本地模型。最后，还提供了使用ModelScope社区下载特定模型文件并集成到本地环境中的具体操作方法，旨在帮助用户快速搭建属于自己的AI应用开发与测试平台。