网上关于 LangGraph 的教程很多但大多数要么只讲概念要么贴一段跑不通的示例。这次我们直接从一个可运行的多智能体架构入手拆清楚 LangGraph 怎么编排多个 Agent、条件路由怎么写、子图怎么抽、并行分支怎么挂最后给出一套能接入 HTTP 服务和批量任务的完整代码骨架。LangGraph 是 LangChain 团队开源的多智能体编排框架它的核心思路是把 Agent 工作流建模成一张有向图节点Node是执行单元边Edge决定流转方向状态State在节点之间传递。相比 LangChain 传统的线性 ChainLangGraph 天然支持条件路由、循环、递归、子图和并行分支这些正是多智能体应用最需要的控制能力。本文会覆盖 LangGraph 的多智能体设计模式、核心组件、条件路由、循环检测、子图与并行分支并在代码实战部分给出一个客服多智能体示例。然后我会把它封装成 FastAPI 接口演示带 thread_id 的批量任务调用方式最后聊聊资源占用、常见问题和工程化建议。无论你准备接入 OpenAI 这类云端模型还是用 Ollama 做本地部署流程是一样的。1. 核心能力速览能力项说明项目类型多智能体编排框架基于图结构管理 Agent 工作流开源团队LangChain 团队主要功能多智能体协作、条件路由、分支控制、循环检测、子图、并行分支、状态持久化支持语言Python、JavaScript / TypeScript模型接入支持 OpenAI、Anthropic、Ollama、通义千问等任何有 SDK 的大模型启动方式代码调用invoke / streamlanggraph dev可视化调试API 能力编译后的 graph 对象是 Runnable可通过 FastAPI 或 LangServe 暴露为 HTTP 接口批量任务支持推荐用异步 astream 按 thread_id 隔离会话显存占用编排层几乎为 0显存取决于所接的本地模型以实际环境为准适合场景客服系统、内容生产、数据分析、复杂工作流编排、Agent 协作平台2. LangGraph 与 LangChain 的区别很多人第一次接触 LangGraph 会问已经有 LangChain 了为什么还要这个框架一句话回答LangChain 擅长“链式调用”适合线性流程LangGraph 擅长“图式编排”适合有分支、有循环、有并行的复杂流程。传统 LangChain 的 Chain 结构是线性的A 处理完给 BB 处理完给 C。一旦业务中需要“根据条件走不同分支”“某一步失败要重试”“多个 Agent 并行处理再汇总”Chain 就会变得很别扭。LangGraph 的做法是把整个流程抽象成一张图用状态机驱动执行。节点只关心输入输出边决定怎么走。这样无论流程多复杂结构始终清晰。从 2026 年前后的社区趋势看多智能体应用已经不再只是演示 demo而是要真正落到客服、办公自动化、数据分析这些场景里。LangGraph 正好补齐了编排层的能力可控、可追踪、可恢复。3. 适用场景与使用边界LangGraph 适合这几类场景客服系统意图识别后路由到不同的专业 Agent比如订单、退款、售后。内容生产一个 Agent 负责生成一个负责审查一个负责改写按拓扑顺序执行。数据分析自然语言转 SQL 查询再由结果 Agent 汇总。办公自动化多 Agent 分别处理邮件、文档、日程最后汇总输出。复杂决策需要循环推理、尝试多种方案并选出最优结果。不太适合的场景简单的一问一答不需要额外编排直接用模型即可。对响应延迟极度敏感且流程固定不变的场景路由和状态管理会带来额外开销。团队没有基本工程化能力只想拖一个界面就出结果的场景LangGraph 仍需要写代码。使用边界必须明确多智能体系统往往会让不同 Agent 访问不同工具和数据库不要给每个 Agent 都授予最高权限。涉及用户隐私、人脸声音数据、版权素材时必须确认授权范围。在测试环境先跑通再上生产。4. 环境准备与前置条件LangGraph 是纯 Python 库依赖模型 SDK不直接运行模型所以对硬件没有硬性要求。你只需要考虑所选 LLM 的部署方式。4.1 基础环境建议使用 Python 3.10 及以上版本并创建独立虚拟环境python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate4.2 安装 LangGraph 与模型 SDKpip install langgraph langgraph-cli langchain-openai如果计划接本地模型还需要安装 Ollama 的集成包pip install langchain-ollamaLangGraph 需要模型 SDK 来调用大模型但它本身不绑定具体厂商。OpenAI、Anthropic、通义、Ollama 都可以。4.3 准备模型访问方式云端模型方案配置环境变量export OPENAI_API_KEYyour-openai-key本地模型方案先安装 Ollama 并拉取模型然后在代码中通过ChatOllama接入from langchain_ollama import ChatOllama llm ChatOllama( modelqwen2.5:7b, base_urlhttp://localhost:11434 )5. LangGraph 核心组件拆解LangGraph 的核心组件不算多但每个都很关键。理解这几个概念基本就能看懂大部分多智能体项目。5.1 StateGraph 与 StateStateGraph是图的构建器State是流经整个图的数据结构。State 必须声明结构常用TypedDict定义from typing import TypedDict class AgentState(TypedDict): user_input: str intent: str response: str history: list[str]每个节点函数都接收当前 State返回一个字典LangGraph 会把返回的字段合并到 State 中。5.2 Node 与 EdgeNode 是执行单元可以是普通函数也可以是 LangChain Runnable。Edge 是节点之间的连线决定流转方向。from langgraph.graph import StateGraph, START, END def node_a(state: AgentState) - dict: return {response: processed by A} builder StateGraph(AgentState) builder.add_node(node_a, node_a) builder.add_edge(START, node_a) builder.add_edge(node_a, END)5.3 ConditionalEdge 条件边条件边是多智能体路由的核心。它根据当前 State 的内容动态决定下一步走向def route_by_intent(state: AgentState): if state[intent] refund: return refund_agent return human_agent builder.add_conditional_edges( parse_intent, route_by_intent, { refund_agent: refund, human_agent: human } )5.4 Checkpointer 状态持久化Checkpointer 用于保存每个节点的执行快照。它的作用有两个一是支持断点续跑二是通过 thread_id 保留会话记忆。LangGraph 内置了InMemorySaver适合开发和测试from langgraph.checkpoint.memory import InMemorySaver checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer) result graph.invoke( {user_input: 我要退款}, config{configurable: {thread_id: session-001}} )每次传入相同的thread_id图可以读到之前保存的 State。想从内存恢复上下文核心就是这个机制。6. 多智能体的四种交互模式从社区讨论看多智能体通常有四种交互模式。LangGraph 都能表达只是实现方式不同。6.1 顺序流水线模式多个 Agent 按固定顺序执行前一个的输出作为后一个的输入。适合内容生成的“生成 - 审查 - 改写”流程。在 LangGraph 中就是一条直线builder.add_edge(START, generator) builder.add_edge(generator, reviewer) builder.add_edge(reviewer, rewriter) builder.add_edge(rewriter, END)6.2 监督者模式一个中心 Supervisor 负责接收任务判断应该交给哪个子 Agent。这是目前多智能体系统里最常见的架构。LangGraph 用条件路由实现中心节点就是意图分发器。6.3 层级模式Supervisor 下面再挂 Supervisor形成树状结构。比如总控 Agent 分派给技术组和业务组技术组再分派给前端 Agent、后端 Agent。LangGraph 用子图嵌套实现每个子图内部可以有自己的路由逻辑。6.4 网络协作模式多个 Agent 通过共享 State 或消息系统自由协作互相调用。LangGraph 中最直接的方式是设置一个共享的 state 字段各个节点往里面追加消息并允许多条边形成循环。7. 条件路由与分支控制实战条件路由是多智能体系统的灵魂。下面用完整代码演示“意图识别 多 Agent 路由”的实现。7.1 完整代码示例from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver class AgentState(TypedDict): user_input: str intent: str response: str history: list[str] def parse_intent(state: AgentState) - dict: text state[user_input] if 退款 in text or 退 in text: return {intent: refund} if 订单 in text or 查 in text: return {intent: order} return {intent: human} def refund_agent(state: AgentState) - dict: return {response: [退款智能体] 已收到您的退款申请我们将在 1-3 个工作日内完成审核。} def order_agent(state: AgentState) - dict: return {response: [订单智能体] 您的订单正在加急配送预计明天送达。} def human_agent(state: AgentState) - dict: return {response: [人工客服] 请提供订单号人工客服将为您处理。} def route_by_intent(state: AgentState) - Literal[refund, order, human]: return state[intent] builder StateGraph(AgentState) builder.add_node(parse_intent, parse_intent) builder.add_node(refund, refund_agent) builder.add_node(order, order_agent) builder.add_node(human, human_agent) builder.add_edge(START, parse_intent) builder.add_conditional_edges( parse_intent, route_by_intent, {refund: refund, order: order, human: human} ) builder.add_edge(refund, END) builder.add_edge(order, END) builder.add_edge(human, END) checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer) # 测试调用 result graph.invoke( {user_input: 我要退款, history: []}, config{configurable: {thread_id: session-001}} ) print(result[response])运行后输出[退款智能体] 已收到您的退款申请我们将在 1-3 个工作日内完成审核。7.2 循环检测与递归限制多智能体系统中经常出现“Agent A 觉得需要问 BB 又觉得需要问 A”的死循环。LangGraph 对图执行深度有默认限制超过会抛RecursionLimit。两种处理方式在调用时提高限制result graph.invoke( state, {recursion_limit: 50, configurable: {thread_id: session-001}} )在路由函数中显式设置终止条件比如当某个字段达到阈值时返回END。7.3 子图 Subgraph子图是把一个完整的StateGraph编译后作为节点挂到父图里。适合把复杂的专业流程拆成独立模块也便于单独测试。from typing import TypedDict class SubState(TypedDict): x: int y: int def double_x(state: SubState) - dict: return {x: state[x] * 2} sub_builder StateGraph(SubState) sub_builder.add_node(double_x, double_x) sub_builder.add_edge(START, double_x) sub_builder.add_edge(double_x, END) sub_graph sub_builder.compile() class ParentState(TypedDict): x: int result: int def call_sub(state: ParentState) - dict: out sub_graph.invoke({x: state[x]}) return {result: out[x]} parent_builder StateGraph(ParentState) parent_builder.add_node(call_sub, call_sub) parent_builder.add_edge(START, call_sub) parent_builder.add_edge(call_sub, END) parent_graph parent_builder.compile()7.4 并行分支LangGraph 的 fan-out / fan-in 可以让多个 Agent 并行执行。比如同时让三个 Agent 分别处理不同任务再汇总结果。可以用SendAPIfrom langgraph.types import Send def spawn_workers(state): return [Send(worker, {task: task}) for task in state[tasks]]这里Send会为每个 task 动态创建一个 worker 节点的独立执行实例。注意并行执行会同时发起多个 LLM 调用要留意模型的并发限流。8. 监督者多智能体示例下面把条件路由和 LLM 判断结合起来实现一个真正由大模型做路由决策的监督者模式。from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver from langchain_openai import ChatOpenAI class SupervisorState(TypedDict): task: str decision: str result: str llm ChatOpenAI(modelgpt-4o-mini, temperature0) def supervisor(state: SupervisorState) - dict: prompt f根据用户任务决定交给 research 还是 code\n任务{state[task]} decision llm.invoke(prompt).content.strip().lower() if code in decision: return {decision: code} return {decision: research} def research_agent(state: SupervisorState) - dict: return {result: f[研究] 已完成{state[task]}} def code_agent(state: SupervisorState) - dict: return {result: f[编程] 已完成{state[task]}} def route_after_supervisor(state: SupervisorState) - Literal[research, code]: return state[decision] builder StateGraph(SupervisorState) builder.add_node(supervisor, supervisor) builder.add_node(research, research_agent) builder.add_node(code, code_agent) builder.add_edge(START, supervisor) builder.add_conditional_edges( supervisor, route_after_supervisor, {research: research, code: code} ) builder.add_edge(research, END) builder.add_edge(code, END) graph builder.compile(checkpointerInMemorySaver()) result graph.invoke( {task: 帮我查一下 LangGraph 最新版本}, config{configurable: {thread_id: s-001}} ) print(result[result])实际项目中不要把原始输入直接拼进 prompt 就完事要做提示词清洗和注入防护。尤其是多智能体场景每一个 Agent 的输入都可能来自另一个 Agent 的输出存在 prompt injection 风险。9. 启动方式与可视化调试LangGraph 提供了两个常见的启动路径。9.1 直接启动 FastAPI 应用如果把 graph 包装成 FastAPI 接口可以直接用 uvicorn 启动uvicorn app:app --host 127.0.0.1 --port 8000这种方式适合生产环境接口完全由自己控制。9.2 使用 langgraph dev 开发模式langgraph dev是官方开发服务器会启动一个可视化调试界面可以看到图结构、节点执行顺序和 State 变化排查路由问题时非常有用。先安装 CLIpip install langgraph-cli在项目根目录创建langgraph.json{ dependencies: [.], graphs: { agent: ./my_graph.py:graph }, env: { OPENAI_API_KEY: your-key } }然后启动langgraph dev注意langgraph dev与uvicorn app的定位不同。前者是开发调试工具后者是常规应用服务。调试阶段用langgraph dev发布阶段封装成自己的 FastAPI 服务。10. 接口 API 与批量任务把 graph 编译结果封装成 HTTP 接口非常简单。下面给出一个 FastAPI 示例。10.1 FastAPI 封装from fastapi import FastAPI from pydantic import BaseModel # 假设 graph 已经编译好并且是上一节的客服多智能体 app FastAPI() class ChatRequest(BaseModel): message: str thread_id: str default class ChatResponse(BaseModel): reply: str app.post(/chat, response_modelChatResponse) async def chat(req: ChatRequest): result await graph.ainvoke( {user_input: req.message, history: []}, config{configurable: {thread_id: req.thread_id}} ) return ChatResponse(replyresult[response])启动后可以用 curl 验证curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {message: 我要退款, thread_id: session-001}返回示例{ reply: [退款智能体] 已收到您的退款申请我们将在 1-3 个工作日内完成审核。 }10.2 批量任务调用批量调用时最重要的原则是每个会话用独立的 thread_id互不干扰。可以用asyncio.gather并发import asyncio async def batch_chat(messages: list[str], base_thread: str): tasks [ graph.ainvoke( {user_input: msg, history: []}, config{configurable: {thread_id: f{base_thread}-{i}}} ) for i, msg in enumerate(messages) ] results await asyncio.gather(*tasks) return [r[response] for r in results] async def main(): replies await batch_chat( [我要退款, 查一下订单, 转人工], batch-001 ) for r in replies: print(r) asyncio.run(main())批量任务建议加上失败重试和结果记录。特别是涉及大量 LLM 调用时手动记录每个 thread_id 的输入输出方便回溯。11. 资源占用与性能观察11.1 框架本身开销LangGraph 的编排层本身非常轻量几乎不占 GPU 显存。真正的资源消耗来自你接入的 LLM云端模型主要消耗 API 调用量和 token 费用。本地模型显存占用由模型规模和量化精度决定比如常见的 7B 量化模型通常需要数 GB 显存具体以本机测试为准。11.2 观察方法开发阶段建议在节点函数里打日志或使用graph.stream()观察每一步输出for step in graph.stream( {user_input: 我要退款, history: []}, config{configurable: {thread_id: session-001}}, stream_modeupdates ): print(节点输出:, step)这样可以精确看到每个节点的产出定位是哪个环节变慢或返回异常。11.3 降低开销的建议State 中不要塞大文件、长文本原始数据尽量放摘要或引用。对话历史要截断按 token 数限制或者用摘要 Agent 压缩旧对话。并行分支会同时请求多个 LLM注意模型限流必要时在代码层加信号量控制并发数。本地模型优先选量化版本用小模型跑通流程后再切换更大模型。12. 常见问题与排查方法问题现象可能原因排查方式解决方案编译时报错 InvalidNodeReferenceadd_edge 引用了未添加的节点检查所有节点名是否一致先 add_node 再 add_edge注意名称拼写条件路由返回错误route 函数返回的 key 与映射表不一致打印 route 函数返回值确认 mapping 中存在对应 key执行不结束报 RecursionLimit图中存在循环且无终止条件观察路由逻辑设置 recursion_limit并增加终止分支Checkpointer 报序列化错误State 中包含不可序列化对象检查 State 中字段类型将自定义对象转为字符串或使用 SqliteSaverlanggraph dev 启动失败端口被占用或 langgraph.json 错误查看启动日志换端口检查 graphs 路径LLM 调用超时网络问题或 API key 失效单独测试模型 SDK检查 key、base_url、超时参数本地模型显存不足模型规模超出显存用nvidia-smi查看显存换小模型或低精度量化版上下文越来越贵history 无限增长检查 State 中 history 长度做截断或摘要压缩13. 最佳实践与使用建议第一State 是唯一事实来源。不要把状态藏到全局变量或类成员里多个节点共享的修改都通过返回字典合并到 State这样图的执行才是可追踪、可恢复的。第二先跑通最小图再扩展。第一次写多智能体时不要一上来就画十几个节点。先用一个路由加两个子 Agent 验证链路再逐步添加子图和并行分支。第三给每个会话分配独立的 thread_id。无论是单用户对话还是批量任务thread_id 都是隔离状态的关键。出问题时凭 thread_id 可以精确还原当时的状态。第四多智能体场景要控制工具权限。不同的 Agent 可能访问不同的数据和工具不要把高权限能力暴露给所有 Agent。涉及用户隐私、肖像、声音、版权素材时必须先确认授权。第五生产环境不要裸跑内存 Checkpointer。InMemorySaver适合开发和测试服务重启后状态会丢失。生产环境建议使用 SQLite 或 Postgres 这类持久化方案。第六如果要接入本地部署的大模型先把 Ollama 或 vLLM 的服务跑通再用 LangGraph 连接。不要把模型启动和 LangGraph 调试混在一起会很难定位问题。14. 总结与下一步LangGraph 值得最先验证的功能是条件路由也就是让一个中心节点根据 LLM 或规则判断任务走向。这个能力能撑起大多数多智能体应用。第二个要验证的是子图和并行分支它决定了复杂流程的可维护性。最容易踩的坑是递归调用没有终止条件导致RecursionLimit报错。下一步可以这样扩展先把你自己的业务拆成“意图识别 - 专业 Agent - 汇总输出”三层结构用本文的代码骨架跑通然后接入 LangSmith 做全链路追踪最后把内存 Checkpointer 换成 SQLite部署成 FastAPI 服务。多智能体系统并不神秘本质就是给大模型套上一层可控的图执行框架。控制好状态和路由剩下的交给模型。