这次我们来看 LangGraph。如果你最近在调研 Agent 编排方案大概率已经被这个框架刷屏了。LangChain 团队在 LangChain 之上单独做了这个编排框架核心思路非常直白把 Agent 的执行流程画成一张有方向的图节点Node里跑具体逻辑边Edge决定下一步怎么走状态State在所有节点之间传递。相比传统链式调用它的优势是有状态、可循环、支持条件分支、支持并行、支持多 Agent 协作还能把整张图部署成 API 服务。这篇文章按“零基础实战”的顺序来写不堆概念直接带你从安装环境开始用可以运行的代码把 State、Node、Edge 跑通再做条件分支、子图和并行编排最后落到多 Agent 协作和 API 部署。看完你能知道LangGraph 到底解决了什么问题、在自己的项目里怎么用、哪些坑要先避开。1. LangGraph 核心能力速览先把最关心的能力项列出来方便判断这东西是否适合你。能力项说明项目类型基于状态图StateGraph的 Agent 编排框架开源来源LangChain 团队维护核心概念State状态、Node节点、Edge边、StateGraph状态管理通过 TypedDict 定义状态结构节点函数返回部分状态完成更新条件分支支持 Conditional Edge根据节点返回值动态路由到不同下游节点循环控制允许边回到上游节点天然支持循环不受 DAG 限制并行分支多个节点可配置为并行执行再由汇总节点合并结果子图支持一个图作为节点嵌入另一个图适合模块化编排多 Agent 协作支持 supervisor worker 的主从模式subagent 可看作特殊 tool长期记忆通过 checkpointer 保存状态支持跨会话恢复部署方式可本地运行也可通过 LangGraph CLI 部署为 API 服务API 接口支持把编译后的图暴露为 REST 接口方便业务系统调用硬件门槛框架本身 CPU 即可运行实际资源消耗取决于所接的大模型语言支持官方以 Python 为主也有 JavaScript/TypeScript SDK从表格能看出来LangGraph 不是替代大模型推理的工具它的定位是“流程编排层”。真正消耗 GPU 的是你接的本地模型或云端模型 APILangGraph 本身只是调度和执行图逻辑。2. 适用场景与使用边界LangGraph 适合的场景核心是“流程复杂、状态需要跟着走”的 Agent 应用。2.1 适合谁用正在用 LangChain 写 Agent但发现chain | chain这种线性方式不够用的人。需要多轮对话中保持上下文而不是每次单独调模型 API 的开发者。要做“判断-执行-再判断”这类复杂流程的人比如客服分流、内容审核、任务拆解。想把多个 Agent 组织成主从协作模式的工程团队。需要把 Agent 流程做成可观测、可恢复服务的后端开发者。2.2 能解决什么问题LangGraph 解决的核心问题有两个。一是状态传递。传统 LangChain 链里每一步的中间结果要靠手动拼接 prompt图一复杂就乱。LangGraph 把状态定义成全局 schema每个节点只返回自己要更新的字段框架自动合并。二是流程控制。用条件边可以做出“如果……就走 A否则走 B”的逻辑用循环边可以让流程反复执行直到满足条件。2.3 不适合什么场景极简单的单次问答一个model.invoke()就结束了套图反而是多余设计。对响应延迟极其敏感的服务端接口图编排有状态合并和节点调度开销虽然不大但比直接调模型 API 要多一层。不需要状态持久化的纯函数调用链用普通 Python 函数组合反而更轻。2.4 合规边界需要特别提醒。LangGraph 本身是中立的编排工具但你在上面跑什么业务直接影响合规边界。调用云端 LLM 时注意不要把用户隐私数据、业务密钥直接拼进 prompt。做客服、审核类应用时对输入输出要做敏感信息过滤。涉及人脸、声音、肖像、版权素材的内容必须先确认授权。部署 API 服务时不要裸奔到公网至少要加鉴权。3. 环境准备与前置条件LangGraph 是纯 Python 框架安装门槛不高。下面是通用的环境检查清单具体版本以你本机实际情况为准。3.1 基础环境操作系统Windows 10/11、macOS、Linux 均可。Python 版本建议 3.9 及以上优先 3.10 或 3.11。包管理工具pip 或 conda。大模型访问方式云端 API KeyOpenAI、Anthropic、国内大模型厂商等或本地模型服务Ollama、vLLM 等。3.2 安装依赖创建一个新的虚拟环境避免和已有项目冲突python -m venv langgraph-env source langgraph-env/bin/activate # Windows 下执行 langgraph-env\Scripts\activate然后安装核心依赖pip install -U langgraph langchain-openai如果要用本地模型可以额外装pip install -U langchain-ollama如果要把图部署成 API 服务还需要安装命令行工具pip install -U langgraph-cli判断安装是否成功python -c import langgraph; print(langgraph.__version__)能正常打印版本号环境就通了。3.3 关于 LangGraph 和 LangChain 的区别安装时容易混淆。LangChain 是一个包含模型调用、提示词模板、向量存储、工具封装等能力的大型生态库。LangGraph 则专注于“把多个调用步骤编排成图”。两者的关系是互补而不是替代你可以在 LangGraph 的节点里调用 LangChain 的 ChatModel、Agent 或 Tool。实际项目中经常是 LangChain 提供组件LangGraph 负责流程控制。4. 核心组件实战State、Node、Edge 打通第一个 Graph这一章直接从代码开始。先不用接大模型用纯 Python 函数把 LangGraph 的执行机制跑通。4.1 State图的状态结构State 是整个图共享的数据结构。通常用TypedDict定义。字段类型是普通类型时节点返回的新值会覆盖旧值字段声明为Annotated[list, add_messages]时新消息会追加而不是覆盖。from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END, add_messages # 定义图的状态结构 class State(TypedDict): messages: Annotated[list, add_messages] step: int这里messages用add_messages做合并器step就是普通字段后续节点返回step时直接覆盖。4.2 Node执行逻辑的节点函数节点就是一个普通函数。入参是整个 State返回值是一个字典只包含你要更新的字段。LangGraph 会自动把返回值合并到全局状态里。def node_a(state: State): return {messages: [来自节点 A 的消息], step: state[step] 1} def node_b(state: State): return {messages: [来自节点 B 的消息], step: state[step] 1}注意“在节点函数里改状态”并不是修改传入的 state 对象而是返回一个局部更新字典。如果你前面的处理逻辑比较复杂也可以读取 state 后计算再返回增量更新。4.3 Edge连接节点的边Edge 定义执行顺序。START是图入口END是出口。# 创建状态图 graph StateGraph(State) # 注册节点 graph.add_node(a, node_a) graph.add_node(b, node_b) # 连接边 graph.add_edge(START, a) graph.add_edge(a, b) graph.add_edge(b, END) # 编译图 app graph.compile()4.4 运行验证# 第一次运行 result app.invoke({messages: [], step: 0}, {configurable: {thread_id: demo-001}}) print(result[messages]) print(result[step])预期输出[来自节点 A 的消息, 来自节点 B 的消息] 2从输出可以看出三个关键点messages字段没有被覆盖而是追加这是add_messages合并器生效。step字段被累加说明每个节点的返回值都正确合并到了状态。整个图是按a - b的顺序执行的。到这里State、Node、Edge 三个核心组件的最小闭环就跑通了。如果结果符合预期说明你已经理解了 LangGraph 的基础执行模型。5. 条件路由、循环、并行与子图实现实际业务很少是一条直线跑到底的。这一章处理四个常见需求条件分支、循环检测、并行分支、子图复用。5.1 条件路由用 Conditional Edge 做动态分流条件边的作用是根据某个节点函数的返回值决定下一步进入哪个节点。下面的例子实现一个简单分流逻辑用户输入包含“人工”就走人工节点否则走自动问答节点。from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class RouterState(TypedDict): user_input: str route: str def process_input(state: RouterState): # 模拟预处理实际可以在这里做实体提取、意图识别等 return {route: pending} def decide_route(state: RouterState) - Literal[qa, human]: if 人工 in state[user_input]: return human return qa def qa_node(state: RouterState): return {route: qa} def human_node(state: RouterState): return {route: human} router_graph StateGraph(RouterState) router_graph.add_node(process, process_input) router_graph.add_node(qa, qa_node) router_graph.add_node(human, human_node) router_graph.add_edge(START, process) # 在 process 节点后添加条件边 router_graph.add_conditional_edges( process, decide_route, { qa: qa, human: human, } ) router_graph.add_edge(qa, END) router_graph.add_edge(human, END) router_app router_graph.compile() # 测试自动问答路径 result router_app.invoke({user_input: 这件商品怎么退款}) print(result[route]) # qa # 测试转人工路径 result router_app.invoke({user_input: 我想转接人工客服}) print(result[route]) # human条件判断函数decide_route的返回值必须是一个字符串并且这个字符串要能在映射表中找到键。如果找不到对应键LangGraph 会报错。这个设计约束了行为减少了拼写错误导致的静默失败。真实项目中decide_route内部往往不是关键词判断而是调用 LLM 让模型输出一个分类标签。这里先不给 LLM 路由的完整代码你先理解机制。5.2 循环用边实现“回到上游节点”LangGraph 的图允许出现环。这个能力很有用比如“生成内容 - 质量检查 - 不合格就重新生成”。下面是一个最少循环示例from typing import TypedDict from langgraph.graph import StateGraph, START, END class LoopState(TypedDict): count: int def generate(state: LoopState): return {count: state[count] 1} def check(state: LoopState) - str: # 模拟质量检查重试 3 次后强制通过 if state[count] 3: return retry return pass loop_graph StateGraph(LoopState) loop_graph.add_node(generate, generate) loop_graph.add_node(check, check) loop_graph.add_edge(START, generate) loop_graph.add_edge(generate, check) # 根据 check 的返回值决定是回到 generate 还是结束 loop_graph.add_conditional_edges( check, check, { retry: generate, pass: END, } ) loop_app loop_graph.compile() result loop_app.invoke({count: 0}) print(result[count]) # 3在这种循环结构里最怕的是“死循环”。解决办法是在状态里增加一个重试计数字段在条件判断中限制最大循环次数。这也是为什么 State 字段设计很重要循环控制需要的计数器必须提前定义好。5.3 并行分支多路执行再合并LangGraph 里实现并行并不需要特殊 API。只要让两个 Node 都从同一个前置节点出发再共同汇入一个汇总节点运行时就会自动并行执行。from typing import TypedDict from langgraph.graph import StateGraph, START, END class ParallelState(TypedDict): left_output: str right_output: str merged: str def left_branch(state: ParallelState): return {left_output: 左路处理完成} def right_branch(state: ParallelState): return {right_output: 右路处理完成} def merge(state: ParallelState): return {merged: state[left_output] state[right_output]} parallel_graph StateGraph(ParallelState) parallel_graph.add_node(left, left_branch) parallel_graph.add_node(right, right_branch) parallel_graph.add_node(merge, merge) # START 同时连接到 left 和 right parallel_graph.add_edge(START, left) parallel_graph.add_edge(START, right) # 两路都汇入 merge parallel_graph.add_edge(left, merge) parallel_graph.add_edge(right, merge) parallel_graph.add_edge(merge, END) parallel_app parallel_graph.compile() result parallel_app.invoke({left_output: , right_output: , merged: }) print(result[merged])使用并行分支时有一个隐含约定merge节点必须在这两个分支都执行完后才会被调用。LangGraph 内部会等所有指向merge的上游边完成再执行merge。如果某个分支需要长时间调用模型并行能显著缩短总耗时。5.4 子图一个编译好的图作为节点当流程模块化之后可以把编译好的子图当成一个普通节点嵌入父图。子图的输入输出会按照父图的 State schema 进行映射。from typing import TypedDict from langgraph.graph import StateGraph, START, END class SharedState(TypedDict): count: int # 先构建一个子图 def sub_increment(state: SharedState): return {count: state[count] 1} sub_graph StateGraph(SharedState) sub_graph.add_node(sub_increment, sub_increment) sub_graph.add_edge(START, sub_increment) sub_graph.add_edge(sub_increment, END) sub_app sub_graph.compile() # 父图中直接引用编译后的子图 def main_node(state: SharedState): return {count: state[count] * 2} main_graph StateGraph(SharedState) main_graph.add_node(subgraph, sub_app) main_graph.add_node(main, main_node) main_graph.add_edge(START, subgraph) main_graph.add_edge(subgraph, main) main_graph.add_edge(main, END) main_app main_graph.compile() result main_app.invoke({count: 1}) print(result[count]) # (11)*2 4子图适合做“公共能力复用”。比如登录校验、内容过滤、结果打分都可以单独做成子图在不同父图中复用。6. 多 Agent 协作主从模式与 subagent 实战多 Agent 是热词里出现频率很高的方向。当前社区讨论比较集中的设计有两种主从模式supervisor workers以及“把 subagent 当作特殊 tool 调用”的思路。本质上两者并不冲突主从模式是架构关系subagent 当 tool 是调用方式。6.1 主从模式一个 supervisor 调度多个 worker主从模式里有一个总控节点负责接收任务、决定交给哪个 worker 执行然后根据 worker 的结果决定是否结束或继续。下面用纯函数模拟一个简化版调度流程。from typing import Annotated, TypedDict, Literal from langgraph.graph import StateGraph, START, END, add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] active_agent: str def research_agent(state: AgentState): return { messages: [(assistant, 检索完成返回相关资料)], active_agent: research, } def write_agent(state: AgentState): return { messages: [(assistant, 文案撰写完成返回成稿)], active_agent: writer, } def supervisor(state: AgentState) - Literal[research, write, end]: # 实际情况这里应该调用 LLM让模型决策下一步交给谁 last_message state[messages][-1].content if 查一下 in last_message: return research if 写一段 in last_message: return write return end agent_graph StateGraph(AgentState) agent_graph.add_node(supervisor, supervisor) agent_graph.add_node(research, research_agent) agent_graph.add_node(write, write_agent) agent_graph.add_edge(START, supervisor) # supervisor 根据意图路由 agent_graph.add_conditional_edges( supervisor, supervisor, { research: research, write: write, end: END, } ) # worker 执行完后回到 supervisor进入下一轮判断 agent_graph.add_edge(research, supervisor) agent_graph.add_edge(write, supervisor) agent_app agent_graph.compile() result agent_app.invoke({ messages: [(user, 先查一下资料再写一段产品介绍)], }) print(result[active_agent])这个例子里supervisor节点先用规则判断执行完一个 worker 后又回到supervisor从而实现“计划-执行-再计划”的循环。你在真实项目里要做的事有两个在 supervisor 里换成 LLM 调用让它返回 JSON 格式的下一步动作在 worker 里接入真正的工具或模型。6.2 把 subagent 当作特殊 tool 调用另一种常见设计思路是父 Agent 不直接管理多个 worker而是把某个子 Agent 封装成一个函数在需要时调用。这种模式的本质就是把 subagent 当作“更复杂的 tool”。在 LangGraph 里实现起来也很直接# 假设已经有一个编译好的子图 sub_app def subagent_tool_call(state: AgentState): # 调用子图相当于调用一个 tool sub_result sub_app.invoke({ input_text: state[messages][-1].content }) return {messages: [(assistant, f子 Agent 返回: {sub_result.get(output)})]}把 subagent 当 tool 的好处是解耦。父 Agent 只关心“调用什么函数、传什么参数、拿什么结果”不关心子 Agent 内部跑了多少个模型、多少个工具。缺点是子 Agent 内部细节对外不可见排错时需要在子图里单独加日志。6.3 多 Agent 场景的注意点控制循环次数。主从模式容易出现 supervisor 反复调度 worker 的死循环必须给轮次计数加上限。明确每个 agent 的职责边界。角色越清晰supervisor 的决策准确率越高。每个 worker 的输入输出要结构化。尽量返回 JSON而不是大段自然语言方便 supervisor 解析。7. 长期记忆与状态持久化LangGraph 的 Checkpointer 机制可以保存每次运行的完整状态。最常用的内存版本是MemorySaver生产环境可以换 SQLite 或 Postgres 实现。from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START, END from typing import TypedDict class MemoryState(TypedDict): messages: list def user_node(state: MemoryState): return {messages: state[messages]} memory_graph StateGraph(MemoryState) memory_graph.add_node(user, user_node) memory_graph.add_edge(START, user) memory_graph.add_edge(user, END) checkpointer MemorySaver() memory_app memory_graph.compile(checkpointercheckpointer) # 第一次调用thread_id 相同 result1 memory_app.invoke( {messages: [第一轮]}, {configurable: {thread_id: user-1001}} ) # 第二次调用状态能从 checkpointer 恢复 result2 memory_app.invoke( {messages: [第二轮]}, {configurable: {thread_id: user-1001}} ) print(result2[messages])这里最关键的是thread_id。同一个thread_id的不同调用会共享同一份图状态不同thread_id之间相互隔离。这个机制就是 LangGraph 实现“长期记忆”的基础会话状态不在代码里临时保存而是由 checkpointer 持久化。注意MemorySaver只适合开发和单机测试。进程重启后数据就丢了。生产环境要换成 SQLite 或 Postgres 这类外部存储否则无法实现真正的长期记忆。8. 接口 API 调用与批量任务LangGraph 图编译后可以嵌入 FastAPI 等服务也可以借助 LangGraph CLI 启动开发服务。先介绍通用做法自己写一个 FastAPI 接口把图暴露出去。8.1 用 FastAPI 包装图服务from fastapi import FastAPI from pydantic import BaseModel from langgraph.graph import StateGraph, START, END from typing import TypedDict class State(TypedDict): input_text: str output_text: str def echo_node(state: State): return {output_text: f处理结果: {state[input_text]}} graph StateGraph(State) graph.add_node(echo, echo_node) graph.add_edge(START, echo) graph.add_edge(echo, END) app_graph graph.compile() # FastAPI 服务 api FastAPI() class InvokeRequest(BaseModel): input_text: str thread_id: str default api.post(/invoke) def invoke(req: InvokeRequest): config {configurable: {thread_id: req.thread_id}} result app_graph.invoke({input_text: req.input_text}, config) return {output_text: result[output_text]}启动服务uvicorn main:api --host 0.0.0.0 --port 8000注意如果你要把服务暴露到内网或公网必须加鉴权。上面的示例没有鉴权仅限本机调试。8.2 curl 调用示例curl -X POST http://127.0.0.1:8000/invoke \ -H Content-Type: application/json \ -d {input_text: 测试一条消息, thread_id: batch-001}预期返回{ output_text: 处理结果: 测试一条消息 }8.3 Python 批量调用批量任务的核心是控制并发和频率。一个简单的批量调用模板import requests from concurrent.futures import ThreadPoolExecutor API_URL http://127.0.0.1:8000/invoke items [ {input_text: 任务1, thread_id: batch-001}, {input_text: 任务2, thread_id: batch-002}, ] def call_api(item): response requests.post(API_URL, jsonitem, timeout120) response.raise_for_status() return response.json() with ThreadPoolExecutor(max_workers2) as executor: results list(executor.map(call_api, items)) for r in results: print(r)批量任务最容易出的问题有三个并发超出模型 API 限流阈值、某个任务长时间挂起、中间失败后缺少重试机制。工程化一点的做法是把任务列表先落地成文件或数据库记录每个任务记录状态pending / running / success / failed成功后保留输出失败后记录错误信息并支持重跑。LangGraph 本身解决“流程编排”批量任务队列最好是结合你熟悉的任务队列系统来做。8.4 使用 LangGraph CLI 部署如果不想自己写 FastAPI 服务官方提供了langgraph-cli来启动开发服务器。安装后在项目根目录执行langgraph dev启动后终端会输出访问地址。不同版本的 CLI 端口和路由会变以实际输出为准。这种方式的优点是自带调试界面适合把图内部的执行过程可视化排查缺点是需要额外适配项目结构。9. 资源占用与性能观察LangGraph 本身是纯 Python 框架CPU 就能跑。它的常驻资源占用很小主要消耗发生在两个地方一是图内部状态序列化和合并二是节点里调用的大模型。9.1 怎么观察资源占用如果调用的是云端 LLM API主要观察网络延迟、token 消耗、API 限流情况。如果调用的是本地 Ollama 或 vLLM主要观察显存占用和 GPU 利用率。具体显存取决于本地模型的参数量、量化格式和并发数。LangGraph 节点并发执行时多线程会占用 CPU。机器核心数不够时并行分支不一定会更快。9.2 影响性能的关键因素因素影响状态字段数量字段越多合并和序列化开销越大并行节点数并行度越高对 CPU 和模型 API 并发压力越大模型输入上下文长度越长token 消耗和响应延迟越高循环次数未加限制的循环会显著增加耗时checkpointer 类型MemorySaver 快但不可恢复外部存储偏慢但可靠9.3 降低资源占用的建议只把需要跨节点传递的字段放进 State不需要的中间结果不要往 State 里塞。对模型输出做缓存。同一输入在同一时间段内可以复用结果降低 API 压力和成本。本地模型优先选量化版本并用流式输出避免整段生成完才返回。批量任务限制并发数避免打满模型服务。10. 常见问题与排查方法问题现象可能原因排查方式解决方案安装 langgraph 失败Python 版本过低或依赖冲突检查python --version看 pip 报错升级到 Python 3.9使用虚拟环境重装节点返回的值没有更新到状态里State schema 中字段拼写不一致或返回了未声明的 key检查节点函数返回的 dict 与 TypedDict 字段名统一字段名返回的 key 必须在 State 中声明messages字段被覆盖而不是追加没有给该字段配置add_messages合并器检查 State 定义中 messages 字段改为Annotated[list, add_messages]条件边报错找不到路由键add_conditional_edges映射表中缺少函数返回值打印条件函数实际返回值检查映射表是否覆盖全部返回值图死循环不结束循环边缺少终止条件检查状态中是否有计数字段增加最大循环次数超出后强制走 END调用 API 返回 404服务未启动、端口错误或路由路径不匹配查看 uvicorn 启动日志用浏览器访问文档地址修改请求路径或重新启动服务批量任务中途失败某个输入触发异常、模型 API 限流、网络超时查看日志中失败任务的输入和错误信息增加重试机制和失败任务隔离服务重启后历史会话丢失使用了 MemorySaver检查 checkpointer 类型换成 SQLite/Postgres 持久化 checkpointer想用 Rust 或 Node 调用 LangGraph官方主要提供 Python 和 JS/TS SDK查看官方仓库文档目前 Rust 没有官方版本建议用 Python 或 JS/TS另外补充一个大家常纠结的问题LangGraph 能不能做可视化监控。直接用get_graph()方法可以导出图结构供可视化工具读取如果要用 ECharts 这类前端库做自定义可视化可以监听每次节点执行的事件把节点名、耗时、状态变化推给前端渲染。这里不展开具体代码思路是“事件 - 数据 - 前端图表”。11. 最佳实践与使用建议做了大量 LangGraph 项目后我建议你从一开始就建立这几个习惯。11.1 先设计 State Schema再写节点这是 LangGraph 项目里最容易忽略的事。State 的字段设计决定整个流程的耦合度。先问自己几个问题哪些数据需要跨节点传递哪些中间结果只是某个节点内部用哪些字段需要合并哪些直接覆盖把答案写进 TypedDict再开始写节点函数。11.2 节点函数保持纯函数风格每个节点只做一件事输入来自 State输出是局部更新字典。不要在一个节点里既调用模型又写数据库又发通知否则状态管理优势会被抹平。11.3 第一次运行前做干跑测试图写完之后先用最小输入跑一遍。不用急着接真实模型先用手工构造的假数据验证节点连接是否正确。这一步能排除大部分低级错误。11.4 用 thread_id 隔离会话多用户场景下每个用户必须用独立的 thread_id否则不同会话会串数据。建议以业务用户 ID 或请求 ID 作为 thread_id并在日志中完整保留。11.5 批量任务要加日志和重试批量任务不要只看最终结果。每个任务执行前后都打日志记录输入摘要、耗时、输出路径、错误信息。失败任务要有独立的重试队列而不是全部重新跑。11.6 接口服务必须限制访问范围FastAPI 示例里的/invoke接口没有鉴权只适合本机调试。部署到服务器上时加一层 API Key 或 OAuth 校验至少限制内网访问。11.7 涉及敏感数据时先过滤用户在对话里可能输入身份证、银行卡、密码等敏感信息。这些内容一旦进入 prompt 发给云端模型隐私风险很大。建议在进入 LangGraph 之前做一次输入过滤或脱敏。11.8 输出要做复核模型输出不一定正确。涉及客服话术、法律建议、医疗建议等场景必须加人工复核节点或规则校验节点。12. 总结与下一步LangGraph 最值得尝试的点是把 Agent 应用从“单向调用”改成“有状态、可分支、可循环”的图。这篇文章里你跑通了第一个 StateGraph掌握了状态传递机制做了条件路由、循环、并行、子图也看到了多 Agent 主从模式和 API 部署的通用方式。建议你接下来按这个顺序做实验先用第一节的最小 Graph 跑通环境。加一个条件边尝试用关键词或 LLM 做路由判断。把两个 worker Agent 挂到一个 supervisor 下面测试最简单的多 Agent 协作。给图加上 checkpointer验证同一个 thread_id 跨轮次的记忆能力。最后套一层 FastAPI 接口让业务系统可以调用。最容易踩的坑是三个State 字段名写错导致节点返回不生效messages字段没用add_messages导致历史消息被覆盖循环边没加终止条件导致任务挂死。先把这三个坑避开LangGraph 的体验会顺畅很多。后续可以继续扩展的方向包括接入 LangGraph Platform 做托管部署、把图运行事件输出到监控系统、用 ECharts 等前端库做流程可视化、结合向量库给 Agent 增加长期事实记忆。建议先把这篇文章里的代码模板保存下来做一个最小可运行的 LangGraph 项目骨架之后所有新流程都在这个骨架上迭代。 
长河君 · 2026/8/30 2:23:34 · 阅读 1.2k LangGraph实战：从零搭建有状态可循环的多Agent编排流程
这次我们来看 LangGraph。如果你最近在调研 Agent 编排方案大概率已经被这个框架刷屏了。LangChain 团队在 LangChain 之上单独做了这个编排框架核心思路非常直白把 Agent 的执行流程画成一张有方向的图节点Node里跑具体逻辑边Edge决定下一步怎么走状态State在所有节点之间传递。相比传统链式调用它的优势是有状态、可循环、支持条件分支、支持并行、支持多 Agent 协作还能把整张图部署成 API 服务。这篇文章按“零基础实战”的顺序来写不堆概念直接带你从安装环境开始用可以运行的代码把 State、Node、Edge 跑通再做条件分支、子图和并行编排最后落到多 Agent 协作和 API 部署。看完你能知道LangGraph 到底解决了什么问题、在自己的项目里怎么用、哪些坑要先避开。1. LangGraph 核心能力速览先把最关心的能力项列出来方便判断这东西是否适合你。能力项说明项目类型基于状态图StateGraph的 Agent 编排框架开源来源LangChain 团队维护核心概念State状态、Node节点、Edge边、StateGraph状态管理通过 TypedDict 定义状态结构节点函数返回部分状态完成更新条件分支支持 Conditional Edge根据节点返回值动态路由到不同下游节点循环控制允许边回到上游节点天然支持循环不受 DAG 限制并行分支多个节点可配置为并行执行再由汇总节点合并结果子图支持一个图作为节点嵌入另一个图适合模块化编排多 Agent 协作支持 supervisor worker 的主从模式subagent 可看作特殊 tool长期记忆通过 checkpointer 保存状态支持跨会话恢复部署方式可本地运行也可通过 LangGraph CLI 部署为 API 服务API 接口支持把编译后的图暴露为 REST 接口方便业务系统调用硬件门槛框架本身 CPU 即可运行实际资源消耗取决于所接的大模型语言支持官方以 Python 为主也有 JavaScript/TypeScript SDK从表格能看出来LangGraph 不是替代大模型推理的工具它的定位是“流程编排层”。真正消耗 GPU 的是你接的本地模型或云端模型 APILangGraph 本身只是调度和执行图逻辑。2. 适用场景与使用边界LangGraph 适合的场景核心是“流程复杂、状态需要跟着走”的 Agent 应用。2.1 适合谁用正在用 LangChain 写 Agent但发现chain | chain这种线性方式不够用的人。需要多轮对话中保持上下文而不是每次单独调模型 API 的开发者。要做“判断-执行-再判断”这类复杂流程的人比如客服分流、内容审核、任务拆解。想把多个 Agent 组织成主从协作模式的工程团队。需要把 Agent 流程做成可观测、可恢复服务的后端开发者。2.2 能解决什么问题LangGraph 解决的核心问题有两个。一是状态传递。传统 LangChain 链里每一步的中间结果要靠手动拼接 prompt图一复杂就乱。LangGraph 把状态定义成全局 schema每个节点只返回自己要更新的字段框架自动合并。二是流程控制。用条件边可以做出“如果……就走 A否则走 B”的逻辑用循环边可以让流程反复执行直到满足条件。2.3 不适合什么场景极简单的单次问答一个model.invoke()就结束了套图反而是多余设计。对响应延迟极其敏感的服务端接口图编排有状态合并和节点调度开销虽然不大但比直接调模型 API 要多一层。不需要状态持久化的纯函数调用链用普通 Python 函数组合反而更轻。2.4 合规边界需要特别提醒。LangGraph 本身是中立的编排工具但你在上面跑什么业务直接影响合规边界。调用云端 LLM 时注意不要把用户隐私数据、业务密钥直接拼进 prompt。做客服、审核类应用时对输入输出要做敏感信息过滤。涉及人脸、声音、肖像、版权素材的内容必须先确认授权。部署 API 服务时不要裸奔到公网至少要加鉴权。3. 环境准备与前置条件LangGraph 是纯 Python 框架安装门槛不高。下面是通用的环境检查清单具体版本以你本机实际情况为准。3.1 基础环境操作系统Windows 10/11、macOS、Linux 均可。Python 版本建议 3.9 及以上优先 3.10 或 3.11。包管理工具pip 或 conda。大模型访问方式云端 API KeyOpenAI、Anthropic、国内大模型厂商等或本地模型服务Ollama、vLLM 等。3.2 安装依赖创建一个新的虚拟环境避免和已有项目冲突python -m venv langgraph-env source langgraph-env/bin/activate # Windows 下执行 langgraph-env\Scripts\activate然后安装核心依赖pip install -U langgraph langchain-openai如果要用本地模型可以额外装pip install -U langchain-ollama如果要把图部署成 API 服务还需要安装命令行工具pip install -U langgraph-cli判断安装是否成功python -c import langgraph; print(langgraph.__version__)能正常打印版本号环境就通了。3.3 关于 LangGraph 和 LangChain 的区别安装时容易混淆。LangChain 是一个包含模型调用、提示词模板、向量存储、工具封装等能力的大型生态库。LangGraph 则专注于“把多个调用步骤编排成图”。两者的关系是互补而不是替代你可以在 LangGraph 的节点里调用 LangChain 的 ChatModel、Agent 或 Tool。实际项目中经常是 LangChain 提供组件LangGraph 负责流程控制。4. 核心组件实战State、Node、Edge 打通第一个 Graph这一章直接从代码开始。先不用接大模型用纯 Python 函数把 LangGraph 的执行机制跑通。4.1 State图的状态结构State 是整个图共享的数据结构。通常用TypedDict定义。字段类型是普通类型时节点返回的新值会覆盖旧值字段声明为Annotated[list, add_messages]时新消息会追加而不是覆盖。from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END, add_messages # 定义图的状态结构 class State(TypedDict): messages: Annotated[list, add_messages] step: int这里messages用add_messages做合并器step就是普通字段后续节点返回step时直接覆盖。4.2 Node执行逻辑的节点函数节点就是一个普通函数。入参是整个 State返回值是一个字典只包含你要更新的字段。LangGraph 会自动把返回值合并到全局状态里。def node_a(state: State): return {messages: [来自节点 A 的消息], step: state[step] 1} def node_b(state: State): return {messages: [来自节点 B 的消息], step: state[step] 1}注意“在节点函数里改状态”并不是修改传入的 state 对象而是返回一个局部更新字典。如果你前面的处理逻辑比较复杂也可以读取 state 后计算再返回增量更新。4.3 Edge连接节点的边Edge 定义执行顺序。START是图入口END是出口。# 创建状态图 graph StateGraph(State) # 注册节点 graph.add_node(a, node_a) graph.add_node(b, node_b) # 连接边 graph.add_edge(START, a) graph.add_edge(a, b) graph.add_edge(b, END) # 编译图 app graph.compile()4.4 运行验证# 第一次运行 result app.invoke({messages: [], step: 0}, {configurable: {thread_id: demo-001}}) print(result[messages]) print(result[step])预期输出[来自节点 A 的消息, 来自节点 B 的消息] 2从输出可以看出三个关键点messages字段没有被覆盖而是追加这是add_messages合并器生效。step字段被累加说明每个节点的返回值都正确合并到了状态。整个图是按a - b的顺序执行的。到这里State、Node、Edge 三个核心组件的最小闭环就跑通了。如果结果符合预期说明你已经理解了 LangGraph 的基础执行模型。5. 条件路由、循环、并行与子图实现实际业务很少是一条直线跑到底的。这一章处理四个常见需求条件分支、循环检测、并行分支、子图复用。5.1 条件路由用 Conditional Edge 做动态分流条件边的作用是根据某个节点函数的返回值决定下一步进入哪个节点。下面的例子实现一个简单分流逻辑用户输入包含“人工”就走人工节点否则走自动问答节点。from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class RouterState(TypedDict): user_input: str route: str def process_input(state: RouterState): # 模拟预处理实际可以在这里做实体提取、意图识别等 return {route: pending} def decide_route(state: RouterState) - Literal[qa, human]: if 人工 in state[user_input]: return human return qa def qa_node(state: RouterState): return {route: qa} def human_node(state: RouterState): return {route: human} router_graph StateGraph(RouterState) router_graph.add_node(process, process_input) router_graph.add_node(qa, qa_node) router_graph.add_node(human, human_node) router_graph.add_edge(START, process) # 在 process 节点后添加条件边 router_graph.add_conditional_edges( process, decide_route, { qa: qa, human: human, } ) router_graph.add_edge(qa, END) router_graph.add_edge(human, END) router_app router_graph.compile() # 测试自动问答路径 result router_app.invoke({user_input: 这件商品怎么退款}) print(result[route]) # qa # 测试转人工路径 result router_app.invoke({user_input: 我想转接人工客服}) print(result[route]) # human条件判断函数decide_route的返回值必须是一个字符串并且这个字符串要能在映射表中找到键。如果找不到对应键LangGraph 会报错。这个设计约束了行为减少了拼写错误导致的静默失败。真实项目中decide_route内部往往不是关键词判断而是调用 LLM 让模型输出一个分类标签。这里先不给 LLM 路由的完整代码你先理解机制。5.2 循环用边实现“回到上游节点”LangGraph 的图允许出现环。这个能力很有用比如“生成内容 - 质量检查 - 不合格就重新生成”。下面是一个最少循环示例from typing import TypedDict from langgraph.graph import StateGraph, START, END class LoopState(TypedDict): count: int def generate(state: LoopState): return {count: state[count] 1} def check(state: LoopState) - str: # 模拟质量检查重试 3 次后强制通过 if state[count] 3: return retry return pass loop_graph StateGraph(LoopState) loop_graph.add_node(generate, generate) loop_graph.add_node(check, check) loop_graph.add_edge(START, generate) loop_graph.add_edge(generate, check) # 根据 check 的返回值决定是回到 generate 还是结束 loop_graph.add_conditional_edges( check, check, { retry: generate, pass: END, } ) loop_app loop_graph.compile() result loop_app.invoke({count: 0}) print(result[count]) # 3在这种循环结构里最怕的是“死循环”。解决办法是在状态里增加一个重试计数字段在条件判断中限制最大循环次数。这也是为什么 State 字段设计很重要循环控制需要的计数器必须提前定义好。5.3 并行分支多路执行再合并LangGraph 里实现并行并不需要特殊 API。只要让两个 Node 都从同一个前置节点出发再共同汇入一个汇总节点运行时就会自动并行执行。from typing import TypedDict from langgraph.graph import StateGraph, START, END class ParallelState(TypedDict): left_output: str right_output: str merged: str def left_branch(state: ParallelState): return {left_output: 左路处理完成} def right_branch(state: ParallelState): return {right_output: 右路处理完成} def merge(state: ParallelState): return {merged: state[left_output] state[right_output]} parallel_graph StateGraph(ParallelState) parallel_graph.add_node(left, left_branch) parallel_graph.add_node(right, right_branch) parallel_graph.add_node(merge, merge) # START 同时连接到 left 和 right parallel_graph.add_edge(START, left) parallel_graph.add_edge(START, right) # 两路都汇入 merge parallel_graph.add_edge(left, merge) parallel_graph.add_edge(right, merge) parallel_graph.add_edge(merge, END) parallel_app parallel_graph.compile() result parallel_app.invoke({left_output: , right_output: , merged: }) print(result[merged])使用并行分支时有一个隐含约定merge节点必须在这两个分支都执行完后才会被调用。LangGraph 内部会等所有指向merge的上游边完成再执行merge。如果某个分支需要长时间调用模型并行能显著缩短总耗时。5.4 子图一个编译好的图作为节点当流程模块化之后可以把编译好的子图当成一个普通节点嵌入父图。子图的输入输出会按照父图的 State schema 进行映射。from typing import TypedDict from langgraph.graph import StateGraph, START, END class SharedState(TypedDict): count: int # 先构建一个子图 def sub_increment(state: SharedState): return {count: state[count] 1} sub_graph StateGraph(SharedState) sub_graph.add_node(sub_increment, sub_increment) sub_graph.add_edge(START, sub_increment) sub_graph.add_edge(sub_increment, END) sub_app sub_graph.compile() # 父图中直接引用编译后的子图 def main_node(state: SharedState): return {count: state[count] * 2} main_graph StateGraph(SharedState) main_graph.add_node(subgraph, sub_app) main_graph.add_node(main, main_node) main_graph.add_edge(START, subgraph) main_graph.add_edge(subgraph, main) main_graph.add_edge(main, END) main_app main_graph.compile() result main_app.invoke({count: 1}) print(result[count]) # (11)*2 4子图适合做“公共能力复用”。比如登录校验、内容过滤、结果打分都可以单独做成子图在不同父图中复用。6. 多 Agent 协作主从模式与 subagent 实战多 Agent 是热词里出现频率很高的方向。当前社区讨论比较集中的设计有两种主从模式supervisor workers以及“把 subagent 当作特殊 tool 调用”的思路。本质上两者并不冲突主从模式是架构关系subagent 当 tool 是调用方式。6.1 主从模式一个 supervisor 调度多个 worker主从模式里有一个总控节点负责接收任务、决定交给哪个 worker 执行然后根据 worker 的结果决定是否结束或继续。下面用纯函数模拟一个简化版调度流程。from typing import Annotated, TypedDict, Literal from langgraph.graph import StateGraph, START, END, add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] active_agent: str def research_agent(state: AgentState): return { messages: [(assistant, 检索完成返回相关资料)], active_agent: research, } def write_agent(state: AgentState): return { messages: [(assistant, 文案撰写完成返回成稿)], active_agent: writer, } def supervisor(state: AgentState) - Literal[research, write, end]: # 实际情况这里应该调用 LLM让模型决策下一步交给谁 last_message state[messages][-1].content if 查一下 in last_message: return research if 写一段 in last_message: return write return end agent_graph StateGraph(AgentState) agent_graph.add_node(supervisor, supervisor) agent_graph.add_node(research, research_agent) agent_graph.add_node(write, write_agent) agent_graph.add_edge(START, supervisor) # supervisor 根据意图路由 agent_graph.add_conditional_edges( supervisor, supervisor, { research: research, write: write, end: END, } ) # worker 执行完后回到 supervisor进入下一轮判断 agent_graph.add_edge(research, supervisor) agent_graph.add_edge(write, supervisor) agent_app agent_graph.compile() result agent_app.invoke({ messages: [(user, 先查一下资料再写一段产品介绍)], }) print(result[active_agent])这个例子里supervisor节点先用规则判断执行完一个 worker 后又回到supervisor从而实现“计划-执行-再计划”的循环。你在真实项目里要做的事有两个在 supervisor 里换成 LLM 调用让它返回 JSON 格式的下一步动作在 worker 里接入真正的工具或模型。6.2 把 subagent 当作特殊 tool 调用另一种常见设计思路是父 Agent 不直接管理多个 worker而是把某个子 Agent 封装成一个函数在需要时调用。这种模式的本质就是把 subagent 当作“更复杂的 tool”。在 LangGraph 里实现起来也很直接# 假设已经有一个编译好的子图 sub_app def subagent_tool_call(state: AgentState): # 调用子图相当于调用一个 tool sub_result sub_app.invoke({ input_text: state[messages][-1].content }) return {messages: [(assistant, f子 Agent 返回: {sub_result.get(output)})]}把 subagent 当 tool 的好处是解耦。父 Agent 只关心“调用什么函数、传什么参数、拿什么结果”不关心子 Agent 内部跑了多少个模型、多少个工具。缺点是子 Agent 内部细节对外不可见排错时需要在子图里单独加日志。6.3 多 Agent 场景的注意点控制循环次数。主从模式容易出现 supervisor 反复调度 worker 的死循环必须给轮次计数加上限。明确每个 agent 的职责边界。角色越清晰supervisor 的决策准确率越高。每个 worker 的输入输出要结构化。尽量返回 JSON而不是大段自然语言方便 supervisor 解析。7. 长期记忆与状态持久化LangGraph 的 Checkpointer 机制可以保存每次运行的完整状态。最常用的内存版本是MemorySaver生产环境可以换 SQLite 或 Postgres 实现。from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START, END from typing import TypedDict class MemoryState(TypedDict): messages: list def user_node(state: MemoryState): return {messages: state[messages]} memory_graph StateGraph(MemoryState) memory_graph.add_node(user, user_node) memory_graph.add_edge(START, user) memory_graph.add_edge(user, END) checkpointer MemorySaver() memory_app memory_graph.compile(checkpointercheckpointer) # 第一次调用thread_id 相同 result1 memory_app.invoke( {messages: [第一轮]}, {configurable: {thread_id: user-1001}} ) # 第二次调用状态能从 checkpointer 恢复 result2 memory_app.invoke( {messages: [第二轮]}, {configurable: {thread_id: user-1001}} ) print(result2[messages])这里最关键的是thread_id。同一个thread_id的不同调用会共享同一份图状态不同thread_id之间相互隔离。这个机制就是 LangGraph 实现“长期记忆”的基础会话状态不在代码里临时保存而是由 checkpointer 持久化。注意MemorySaver只适合开发和单机测试。进程重启后数据就丢了。生产环境要换成 SQLite 或 Postgres 这类外部存储否则无法实现真正的长期记忆。8. 接口 API 调用与批量任务LangGraph 图编译后可以嵌入 FastAPI 等服务也可以借助 LangGraph CLI 启动开发服务。先介绍通用做法自己写一个 FastAPI 接口把图暴露出去。8.1 用 FastAPI 包装图服务from fastapi import FastAPI from pydantic import BaseModel from langgraph.graph import StateGraph, START, END from typing import TypedDict class State(TypedDict): input_text: str output_text: str def echo_node(state: State): return {output_text: f处理结果: {state[input_text]}} graph StateGraph(State) graph.add_node(echo, echo_node) graph.add_edge(START, echo) graph.add_edge(echo, END) app_graph graph.compile() # FastAPI 服务 api FastAPI() class InvokeRequest(BaseModel): input_text: str thread_id: str default api.post(/invoke) def invoke(req: InvokeRequest): config {configurable: {thread_id: req.thread_id}} result app_graph.invoke({input_text: req.input_text}, config) return {output_text: result[output_text]}启动服务uvicorn main:api --host 0.0.0.0 --port 8000注意如果你要把服务暴露到内网或公网必须加鉴权。上面的示例没有鉴权仅限本机调试。8.2 curl 调用示例curl -X POST http://127.0.0.1:8000/invoke \ -H Content-Type: application/json \ -d {input_text: 测试一条消息, thread_id: batch-001}预期返回{ output_text: 处理结果: 测试一条消息 }8.3 Python 批量调用批量任务的核心是控制并发和频率。一个简单的批量调用模板import requests from concurrent.futures import ThreadPoolExecutor API_URL http://127.0.0.1:8000/invoke items [ {input_text: 任务1, thread_id: batch-001}, {input_text: 任务2, thread_id: batch-002}, ] def call_api(item): response requests.post(API_URL, jsonitem, timeout120) response.raise_for_status() return response.json() with ThreadPoolExecutor(max_workers2) as executor: results list(executor.map(call_api, items)) for r in results: print(r)批量任务最容易出的问题有三个并发超出模型 API 限流阈值、某个任务长时间挂起、中间失败后缺少重试机制。工程化一点的做法是把任务列表先落地成文件或数据库记录每个任务记录状态pending / running / success / failed成功后保留输出失败后记录错误信息并支持重跑。LangGraph 本身解决“流程编排”批量任务队列最好是结合你熟悉的任务队列系统来做。8.4 使用 LangGraph CLI 部署如果不想自己写 FastAPI 服务官方提供了langgraph-cli来启动开发服务器。安装后在项目根目录执行langgraph dev启动后终端会输出访问地址。不同版本的 CLI 端口和路由会变以实际输出为准。这种方式的优点是自带调试界面适合把图内部的执行过程可视化排查缺点是需要额外适配项目结构。9. 资源占用与性能观察LangGraph 本身是纯 Python 框架CPU 就能跑。它的常驻资源占用很小主要消耗发生在两个地方一是图内部状态序列化和合并二是节点里调用的大模型。9.1 怎么观察资源占用如果调用的是云端 LLM API主要观察网络延迟、token 消耗、API 限流情况。如果调用的是本地 Ollama 或 vLLM主要观察显存占用和 GPU 利用率。具体显存取决于本地模型的参数量、量化格式和并发数。LangGraph 节点并发执行时多线程会占用 CPU。机器核心数不够时并行分支不一定会更快。9.2 影响性能的关键因素因素影响状态字段数量字段越多合并和序列化开销越大并行节点数并行度越高对 CPU 和模型 API 并发压力越大模型输入上下文长度越长token 消耗和响应延迟越高循环次数未加限制的循环会显著增加耗时checkpointer 类型MemorySaver 快但不可恢复外部存储偏慢但可靠9.3 降低资源占用的建议只把需要跨节点传递的字段放进 State不需要的中间结果不要往 State 里塞。对模型输出做缓存。同一输入在同一时间段内可以复用结果降低 API 压力和成本。本地模型优先选量化版本并用流式输出避免整段生成完才返回。批量任务限制并发数避免打满模型服务。10. 常见问题与排查方法问题现象可能原因排查方式解决方案安装 langgraph 失败Python 版本过低或依赖冲突检查python --version看 pip 报错升级到 Python 3.9使用虚拟环境重装节点返回的值没有更新到状态里State schema 中字段拼写不一致或返回了未声明的 key检查节点函数返回的 dict 与 TypedDict 字段名统一字段名返回的 key 必须在 State 中声明messages字段被覆盖而不是追加没有给该字段配置add_messages合并器检查 State 定义中 messages 字段改为Annotated[list, add_messages]条件边报错找不到路由键add_conditional_edges映射表中缺少函数返回值打印条件函数实际返回值检查映射表是否覆盖全部返回值图死循环不结束循环边缺少终止条件检查状态中是否有计数字段增加最大循环次数超出后强制走 END调用 API 返回 404服务未启动、端口错误或路由路径不匹配查看 uvicorn 启动日志用浏览器访问文档地址修改请求路径或重新启动服务批量任务中途失败某个输入触发异常、模型 API 限流、网络超时查看日志中失败任务的输入和错误信息增加重试机制和失败任务隔离服务重启后历史会话丢失使用了 MemorySaver检查 checkpointer 类型换成 SQLite/Postgres 持久化 checkpointer想用 Rust 或 Node 调用 LangGraph官方主要提供 Python 和 JS/TS SDK查看官方仓库文档目前 Rust 没有官方版本建议用 Python 或 JS/TS另外补充一个大家常纠结的问题LangGraph 能不能做可视化监控。直接用get_graph()方法可以导出图结构供可视化工具读取如果要用 ECharts 这类前端库做自定义可视化可以监听每次节点执行的事件把节点名、耗时、状态变化推给前端渲染。这里不展开具体代码思路是“事件 - 数据 - 前端图表”。11. 最佳实践与使用建议做了大量 LangGraph 项目后我建议你从一开始就建立这几个习惯。11.1 先设计 State Schema再写节点这是 LangGraph 项目里最容易忽略的事。State 的字段设计决定整个流程的耦合度。先问自己几个问题哪些数据需要跨节点传递哪些中间结果只是某个节点内部用哪些字段需要合并哪些直接覆盖把答案写进 TypedDict再开始写节点函数。11.2 节点函数保持纯函数风格每个节点只做一件事输入来自 State输出是局部更新字典。不要在一个节点里既调用模型又写数据库又发通知否则状态管理优势会被抹平。11.3 第一次运行前做干跑测试图写完之后先用最小输入跑一遍。不用急着接真实模型先用手工构造的假数据验证节点连接是否正确。这一步能排除大部分低级错误。11.4 用 thread_id 隔离会话多用户场景下每个用户必须用独立的 thread_id否则不同会话会串数据。建议以业务用户 ID 或请求 ID 作为 thread_id并在日志中完整保留。11.5 批量任务要加日志和重试批量任务不要只看最终结果。每个任务执行前后都打日志记录输入摘要、耗时、输出路径、错误信息。失败任务要有独立的重试队列而不是全部重新跑。11.6 接口服务必须限制访问范围FastAPI 示例里的/invoke接口没有鉴权只适合本机调试。部署到服务器上时加一层 API Key 或 OAuth 校验至少限制内网访问。11.7 涉及敏感数据时先过滤用户在对话里可能输入身份证、银行卡、密码等敏感信息。这些内容一旦进入 prompt 发给云端模型隐私风险很大。建议在进入 LangGraph 之前做一次输入过滤或脱敏。11.8 输出要做复核模型输出不一定正确。涉及客服话术、法律建议、医疗建议等场景必须加人工复核节点或规则校验节点。12. 总结与下一步LangGraph 最值得尝试的点是把 Agent 应用从“单向调用”改成“有状态、可分支、可循环”的图。这篇文章里你跑通了第一个 StateGraph掌握了状态传递机制做了条件路由、循环、并行、子图也看到了多 Agent 主从模式和 API 部署的通用方式。建议你接下来按这个顺序做实验先用第一节的最小 Graph 跑通环境。加一个条件边尝试用关键词或 LLM 做路由判断。把两个 worker Agent 挂到一个 supervisor 下面测试最简单的多 Agent 协作。给图加上 checkpointer验证同一个 thread_id 跨轮次的记忆能力。最后套一层 FastAPI 接口让业务系统可以调用。最容易踩的坑是三个State 字段名写错导致节点返回不生效messages字段没用add_messages导致历史消息被覆盖循环边没加终止条件导致任务挂死。先把这三个坑避开LangGraph 的体验会顺畅很多。后续可以继续扩展的方向包括接入 LangGraph Platform 做托管部署、把图运行事件输出到监控系统、用 ECharts 等前端库做流程可视化、结合向量库给 Agent 增加长期事实记忆。建议先把这篇文章里的代码模板保存下来做一个最小可运行的 LangGraph 项目骨架之后所有新流程都在这个骨架上迭代。