1. 引言为什么需要多智能体编排随着大语言模型LLM能力的持续提升单一智能体在复杂业务场景中逐渐暴露出局限性上下文窗口有限、工具调用链路过长、职责边界模糊、错误难以隔离。多智能体Multi-Agent架构通过将复杂任务拆解为多个专业智能体各自负责特定子任务再通过编排层协调协作从而提升系统的可维护性、可扩展性和鲁棒性。LangGraph 是 LangChain 团队推出的低层级编排框架专为构建有状态、可循环、可细粒度控制的智能体工作流而设计。与 LangChain 的 Chain 相比LangGraph 更贴近图计算模型支持条件分支、循环、人工介入、持久化等生产级特性是多智能体编排的理想底座。本文将从零开始围绕一个「智能客服工单处理系统」的实战案例逐步讲解如何用 LangGraph 构建生产级多智能体系统涵盖状态管理、图编排、智能体间通信、人工审批、持久化与可观测性等核心主题。2. LangGraph 核心概念在动手编码之前先厘清 LangGraph 的五个核心抽象它们是理解后续所有代码的基础。State状态贯穿整个图执行过程的共享数据结构可以是 TypedDict、Pydantic 模型或自定义类。每个节点读写 State 的指定字段实现数据流转。Node节点图中的一个执行单元通常是一个 Python 函数接收 State 并返回 State 的部分更新。Edge边连接节点的有向边定义执行顺序。普通边表示无条件流转条件边根据 State 内容动态选择下一个节点。Graph图由节点和边组成的执行拓扑LangGraph 支持 StateGraph 和 MessageGraph 两种构建方式。Checkpointer检查点持久化 State 快照的组件支持断点续跑、时间旅行和人工介入。下图展示了本文实战案例的整体编排拓扑flowchart TD A[入口: 工单接收] -- B{意图分类} B --|技术问题| C[技术专家 Agent] B --|账单问题| D[账单 Agent] B --|通用咨询| E[通用客服 Agent] C -- F{是否需要升级} D -- F E -- G[直接回复] F --|是| H[人工审批节点] F --|否| I[生成最终回复] H -- I I -- J[结束] G -- J3. 环境准备与项目结构首先创建虚拟环境并安装依赖。建议使用 Python 3.10 及以上版本。python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install langgraph langchain langchain-openai langchain-community pip install redis # 用于持久化 Checkpointer pip install langsmith # 用于可观测性追踪项目目录结构如下multi_agent_system/ ├── main.py # 入口脚本 ├── state.py # State 定义 ├── agents/ │ ├── __init__.py │ ├── triage.py # 意图分类 Agent │ ├── technical.py # 技术专家 Agent │ ├── billing.py # 账单 Agent │ └── general.py # 通用客服 Agent ├── graph/ │ ├── __init__.py │ ├── builder.py # 图构建逻辑 │ └── edges.py # 条件边路由函数 ├── tools/ │ ├── __init__.py │ └── knowledge_base.py # 知识库检索工具 └── config.py # 全局配置4. 定义全局状态StateState 是多智能体协作的「共享黑板」。我们使用 TypedDict 定义字段并通过 Annotated 指定消息列表的合并策略。# state.py from typing import Annotated, TypedDict, List, Optional from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage class AgentState(TypedDict): 全局共享状态 # 原始工单内容 ticket_id: str customer_name: str ticket_text: str # 消息历史使用 add_messages 自动合并 messages: Annotated[List[BaseMessage], add_messages] 意图分类结果 intent: Optional[str] 当前处理 Agent 的名称 current_agent: Optional[str] 是否需要人工审批 needs_human_approval: bool 最终回复 final_response: Optional[str]关键点说明messages字段使用add_messages归约器LangGraph 会自动将各节点返回的新消息追加到历史列表避免手动拼接。intent、current_agent等字段用于条件边路由决策。needs_human_approval是人工介入的开关当技术 Agent 判定问题复杂时置为 True。5. 构建基础工具知识库检索为了让 Agent 具备领域知识我们实现一个简单的知识库检索工具。生产环境可替换为向量数据库如 Pinecone、Milvus或 Elasticsearch。# tools/knowledge_base.py from typing import List, Dict 模拟知识库数据生产环境替换为向量检索 KNOWLEDGE_BASE [ { id: KB001, category: technical, keywords: [登录, 密码, 认证失败], content: 如果用户遇到登录失败请先引导其检查网络连接然后尝试重置密码。若仍无法登录可能是账号被锁定需联系安全团队解锁。 }, { id: KB002, category: billing, keywords: [退款, 账单, 扣费], content: 退款申请需在账单生成后 30 天内提交。确认订单状态为已支付后财务团队将在 3-5 个工作日内处理退款。 }, { id: KB003, category: general, keywords: [使用, 教程, 入门], content: 新用户可参考官方快速入门文档包含账号注册、项目创建和第一个 API 调用的完整指引。 } ] def search_knowledge_base(query: str, category: str None, top_k: int 2) - List[Dict]: 基于关键词匹配的知识库检索 results [] for item in KNOWLEDGE_BASE: if category and item[category] ! category: continue # 简单关键词命中评分 score sum(1 for kw in item[keywords] if kw in query) if score 0: results.append({score: score, **item}) results.sort(keylambda x: x[score], reverseTrue) return results[:top_k]在实际生产系统中建议使用 Embedding 模型将查询和知识条目向量化通过余弦相似度检索召回效果远优于关键词匹配。6. 实现意图分类 Agent意图分类 Agent 是整个工作流的入口负责判断工单属于技术问题、账单问题还是通用咨询。这里使用结构化输出Structured Output确保返回结果可被程序解析。# agents/triage.py from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage from langchain_core.pydantic_v1 import BaseModel, Field from state import AgentState class IntentOutput(BaseModel): 意图分类结构化输出 intent: str Field(description工单意图类别: technical / billing / general) confidence: float Field(description置信度 0-1) reasoning: str Field(description分类依据简述) def triage_agent(state: AgentState) - AgentState: 意图分类节点 llm ChatOpenAI(modelgpt-4o, temperature0) structured_llm llm.with_structured_output(IntentOutput) system_prompt 你是工单意图分类专家。请根据用户工单内容判断其属于以下哪一类 - technical: 技术故障、报错、功能异常 - billing: 账单、扣费、退款、发票 - general: 产品咨询、使用教程、其他 只输出 JSON 格式结果。 result structured_llm.invoke([ SystemMessage(contentsystem_prompt), HumanMessage(contentf工单内容: {state[ticket_text]}) ]) return { intent: result.intent, messages: [HumanMessage(contentf[意图分类] 判定为 {result.intent}置信度 {result.confidence:.2f})] }这里使用with_structured_output让模型直接输出 Pydantic 对象避免手写 JSON 解析的脆弱性。生产环境建议增加置信度阈值判断低于阈值时直接转人工。7. 实现专业 Agent技术 / 账单 / 通用三个专业 Agent 结构相似均接收工单内容结合知识库检索结果生成回复。以技术专家 Agent 为例# agents/technical.py from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage from state import AgentState from tools.knowledge_base import search_knowledge_base def technical_agent(state: AgentState) - AgentState: 技术专家 Agent llm ChatOpenAI(modelgpt-4o, temperature0.3) # 检索相关知识 kb_results search_knowledge_base(state[ticket_text], categorytechnical) kb_context \n\n.join( f[知识条目 {r[id]}] {r[content]} for r in kb_results ) if kb_results else 知识库中未找到直接匹配条目。 system_prompt f 你是资深技术支持工程师。请基于以下知识库内容结合你的技术经验为用户工单提供解决方案。 知识库参考 {kb_context} 要求 先复述用户问题确认理解正确 给出分步骤的解决方案 如果问题超出知识库范围且涉及账号安全、数据丢失等高风险场景将 needs_human_approval 置为 True response llm.invoke([ SystemMessage(contentsystem_prompt), HumanMessage(contentf用户工单: {state[ticket_text]}) ]) 简单判断是否需要人工介入生产环境可用 LLM 判断或规则引擎 needs_human any(kw in state[ticket_text] for kw in [数据丢失, 账号被盗, 无法恢复]) return { current_agent: technical, needs_human_approval: needs_human, messages: [HumanMessage(contentf[技术专家] {response.content})] }账单 Agent 和通用客服 Agent 的实现模式完全一致仅替换系统提示词和知识库分类参数。为节省篇幅这里不再重复贴出完整代码可在文末获取。8. 实现人工审批节点人工审批是多智能体系统生产化的关键能力。LangGraph 的interrupt机制允许图在特定节点暂停执行等待人工输入后恢复。# agents/human_approval.py from langgraph.types import interrupt, Command from state import AgentState def human_approval_node(state: AgentState) - AgentState: 人工审批节点暂停图执行等待人工决策 # 组装需要人工审核的信息 approval_request { ticket_id: state[ticket_id], customer: state[customer_name], ticket_text: state[ticket_text], agent_response: state[messages][-1].content if state[messages] else , question: 该工单涉及高风险操作请审核是否批准以下回复 } # 暂停执行将审批请求发送给前端 user_decision interrupt(approval_request) 根据人工决策更新状态 if user_decision.get(approved): return { needs_human_approval: False, messages: [HumanMessage(content[人工审批] 已批准该回复)] } else: # 人工驳回可附带修改意见 feedback user_decision.get(feedback, 请重新生成回复) return { needs_human_approval: True, messages: [HumanMessage(contentf[人工审批] 已驳回意见: {feedback})] }当图执行到interrupt时LangGraph 会保存检查点并暂停。外部系统通过graph.invoke(Command(resume...))恢复执行传入人工决策结果。9. 组装多智能体图现在将上述节点组装成完整的 StateGraph。这是整个系统的核心编排逻辑。# graph/builder.py from langgraph.graph import StateGraph, START, END from state import AgentState from agents.triage import triage_agent from agents.technical import technical_agent from agents.billing import billing_agent from agents.general import general_agent from agents.human_approval import human_approval_node def route_by_intent(state: AgentState) - str: 根据意图分类结果路由到对应 Agent intent state.get(intent) if intent technical: return technical elif intent billing: return billing else: return general def route_after_agent(state: AgentState) - str: 判断是否需要人工审批 if state.get(needs_human_approval): return human_approval return finalize def build_graph(): 构建多智能体图 graph StateGraph(AgentState) # 添加节点 graph.add_node(triage, triage_agent) graph.add_node(technical, technical_agent) graph.add_node(billing, billing_agent) graph.add_node(general, general_agent) graph.add_node(human_approval, human_approval_node) graph.add_node(finalize, finalize_node) 添加边 graph.add_edge(START, triage) graph.add_conditional_edges(triage, route_by_intent, { technical: technical, billing: billing, general: general }) graph.add_conditional_edges(technical, route_after_agent, { human_approval: human_approval, finalize: finalize }) graph.add_conditional_edges(billing, route_after_agent, { human_approval: human_approval, finalize: finalize }) graph.add_conditional_edges(general, route_after_agent, { human_approval: human_approval, finalize: finalize }) graph.add_edge(human_approval, finalize) graph.add_edge(finalize, END) return graph.compile()其中finalize_node负责从消息历史中提取最终回复并写入final_response字段# graph/builder.py 中补充 def finalize_node(state: AgentState) - AgentState: 汇总最终回复 # 取最后一个 Agent 的回复作为最终结果 for msg in reversed(state[messages]): if msg.content.startswith([技术专家]) or \ msg.content.startswith([账单专家]) or \ msg.content.startswith([通用客服]): return {final_response: msg.content} return {final_response: 抱歉暂时无法处理该工单已转交人工客服。}10. 持久化与断点续跑生产级系统必须支持持久化以便在进程重启后恢复执行状态。LangGraph 提供 Checkpointer 抽象这里以 Redis 为例# main.py 片段 from langgraph.checkpoint.redis import RedisSaver import redis 初始化 Redis 连接 redis_client redis.Redis(hostlocalhost, port6379, db0) checkpointer RedisSaver(redis_client) 编译图时传入 checkpointer app build_graph() app app.compile(checkpointercheckpointer)执行图并指定线程 ID实现会话级状态隔离# main.py 片段 config {configurable: {thread_id: ticket-1001}} initial_state { ticket_id: ticket-1001, customer_name: 张三, ticket_text: 我无法登录账号提示密码错误但重置密码后仍然无法登录怀疑账号被锁定。 } 首次执行 result app.invoke(initial_state, configconfig) print(意图:, result[intent]) print(最终回复:, result[final_response])当图在人工审批节点暂停后可通过Command(resume...)恢复# 恢复执行传入人工决策 from langgraph.types import Command resume_result app.invoke( Command(resume{approved: True}), configconfig ) print(审批后最终回复:, resume_result[final_response])11. 可观测性与追踪生产环境必须能观测每个 Agent 的输入输出、耗时和 Token 消耗。LangGraph 原生集成 LangSmith只需配置环境变量即可自动追踪export LANGCHAIN_TRACING_V2true export LANGCHAIN_API_KEYyour_langsmith_api_key export LANGCHAIN_PROJECTmulti_agent_system此外可以在节点函数中手动记录关键指标# 在节点中记录耗时 import time from langchain_core.callbacks import CallbackManager def technical_agent_with_metrics(state: AgentState) - AgentState: start time.time() result technical_agent(state) elapsed time.time() - start print(f[Metrics] technical_agent 耗时: {elapsed:.2f}s) return result12. 生产化最佳实践将多智能体系统部署到生产环境需要关注以下关键点超时与重试为每个节点设置超时时间LLM 调用失败时自动重试避免单点故障拖垮整个工作流。限流与配额对 LLM API 调用做限流防止并发过高导致成本失控。状态版本控制State 结构变更时做好迁移策略避免旧检查点无法恢复。安全与权限Agent 调用外部工具时必须做权限校验和敏感信息脱敏。灰度发布新 Agent 或新提示词上线前先在小流量灰度验证效果。成本控制为每个 Agent 设置 Token 预算上限超限自动降级到简单回复。13. 总结与扩展方向本文通过一个完整的智能客服工单处理系统系统讲解了如何用 LangGraph 构建生产级多智能体系统。核心要点总结如下使用StateGraph定义节点和边的执行拓扑通过条件边实现动态路由。利用Annotated归约器管理消息历史的自动合并。通过interrupt机制实现人工审批结合 Checkpointer 支持断点续跑。使用 Redis 等外部存储实现状态持久化保证系统可恢复性。集成 LangSmith 实现全链路可观测性。后续可扩展的方向包括引入 ReAct 循环让 Agent 自主调用工具、使用 Supervisor 模式实现层级化编排、结合向量数据库构建更强大的知识检索、以及通过 LangGraph Cloud 实现一键部署和自动扩缩容。多智能体编排是通往复杂 AI 应用的关键技术栈LangGraph 提供了灵活且生产友好的基础设施。建议读者在本文示例基础上结合自身业务场景逐步迭代构建真正可落地的智能体系统。