这次我们来看 LangGraph。它不是一个用来写文案的模型也不是某个一键启动包而是 LangChain 团队开源的图编排框架专门解决多步骤、有状态、带分支和循环的 LLM 应用问题。很多人学 LangChain 学到一半会卡在“怎么让模型自己决定下一步走哪个节点”LangGraph 就是来补这块短板的定义 State注册 Node用 Edge 把流程串起来再通过条件边让流程自己“拐弯”。这篇文章的核心不是列 API 文档而是直接带你跑通三个实战例子先写一个最简单的状态管理工作流再做一个带条件分支和循环检测的流程控制图最后实现一个多 Agent 协作模式——其中一种主从模式本质上就是把 Subagent 当成一个 Tool 来调用。整个过程我会把环境准备、代码结构、运行结果、常见报错都拆开讲。如果你想在项目里正式用 LangGraph还会看到接口 API、批量任务和资源占用的落地思路。文中代码示例基于当前稳定版 LangChain / LangGraph 的常见写法具体路径和 API 以你安装的版本为准。适合的读者是有 Python 基础、知道 LangChain 是什么但还没把 LangGraph 用起来的开发者或者已经在做 Agent 应用但被复杂流程控制折磨过的朋友。1. LangGraph 核心能力速览在写代码之前先给一张速览表看完你就知道这个东西值不值得学、能塞进什么场景。能力项说明项目类型开源 LLM 应用编排框架围绕 State / Node / Edge 构建图计算开源来源LangChain 团队维护属于 LangChain 生态核心组件State共享状态、Node节点函数、Edge普通边与条件边主要功能状态管理、条件分支、循环控制、子图嵌套、并行分支、多 Agent 协作运行方式Python 代码构建图编译后调用 invoke / stream / batch硬件要求本身不需要 GPUCPU 即可运行实际算力取决于图中调用的 LLM是否支持 API官方提供服务化部署方案编译后的图可暴露为 HTTP 接口是否支持批量任务支持可通过批量输入或任务队列并发跑图实例记忆能力支持 Checkpointer 与长期记忆设计可把状态持久化到外部存储适合场景需要流程控制、多角色协作、可审计步骤的复杂 Agent 应用从这张表能明显看到一个结论LangGraph 不是一个模型而是模型之上的“流程骨架”。它的价值在于把零散的 LLM 调用变成有状态、可控制、可观测的图。你不需要为它配显卡主要关注的是 Python 环境、模型接口和状态设计。2. LangGraph 核心组件拆解State、Node、Edge很多教程一上来就丢图看得人头晕。其实 LangGraph 的核心就三样东西State、Node、Edge。2.1 State所有节点共享的数据容器State 是整张图的“共享内存”。每个节点函数可以读取 State也可以返回新的字段来更新 State。LangGraph 里 State 通常用一个 TypedDict 定义字段可以是基础类型、列表、字典也可以是 Pydantic 模型。from typing_extensions import TypedDict, Annotated import operator class State(TypedDict): messages: Annotated[list, operator.add] # 列表字段默认追加合并 step: int # 普通字段直接覆盖 final_answer: str这里有一个关键点State 的字段更新策略可以自定义。用Annotated[list, operator.add]每个节点返回的列表会被追加合并如果直接写step: int则后返回的值覆盖前值。这个设计解决了多个节点写同一个字段时的冲突问题。2.2 Node一个普通的 Python 函数Node 在 LangGraph 里就是一个函数。输入是 State输出是更新后的 State 字典。每个 Node 只做一件事尽量保持职责单一这样后面的条件分支和复用才方便。def call_llm(state: State) - State: # 这里一般是调用模型此处用伪代码表示 new_message 收到用户问题开始处理 return {messages: [new_message], step: state[step] 1}节点函数不一定必须调用 LLM。它可以是普通计算、数据库查询、外部工具调用、子图调用也可以是一个 Agent。这是 LangGraph 灵活的地方——图里每个节点都是可替换的积木。2.3 Edge连接与条件路由Edge 决定节点之间的流转方向。最简单的是普通边从节点 A 无条件走到节点 B。更常用的是条件边节点结束后根据某个函数或模型返回的结果选择不同的下游节点。builder.add_edge(START, call_llm) builder.add_edge(call_llm, format_answer) builder.add_conditional_edges( router, route_by_score, {high: node_high, low: node_low, END: END} )普通边适合固定流程条件边适合“模型判断走哪条路”的场景。条件分支是 LangGraph 比普通 LangChain Chain 强很多的地方后面专门用一节实战演示。2.4 LangGraph 和 LangChain 到底是什么关系这是搜索量非常高的问题一句话回答LangChain 是组件库LangGraph 是流程编排器。LangChain 提供模型封装、提示词、工具、检索器等“零件”。LangGraph 用图的方式把这些“零件”组织成可控的流程。在 LangGraph 的节点里可以继续使用 LangChain 的ChatPromptTemplate、LCEL、Tool封装等功能组件。两者不是二选一而是搭配使用。很多人纠结“我有 LangChain 了还要不要学 LangGraph”如果你只是单次调用模型或简单的链式调用LangChain 够用。一旦出现“根据前一步结果决定走哪条分支”“需要循环直到满足条件”“多个 Agent 角色协作”这类需求LangChain 的 Chain 会变得很别扭LangGraph 才是合适的选择。3. LangGraph 本地部署环境准备LangGraph 是纯 Python 库环境要求不高重点是把 Python 版本和依赖隔离做好。3.1 基础环境清单检查项建议操作系统Windows / Linux / macOS 均可Python 版本3.9 以上推荐 3.10 或 3.11包管理pip、uv 或 poetryGPU非必需CPU 即可跑图逻辑模型接口OpenAI 协议兼容接口或本地部署的模型服务磁盘空间依赖安装约 1-2 GB具体看虚拟环境LangGraph 不依赖 CUDA。真正的显存消耗来自你节点里调用的 LLM 或嵌入模型。如果你用纯本地小模型CPU 也能跑。3.2 安装依赖建议新建一个虚拟环境避免污染全局 Python。python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate安装核心依赖pip install --upgrade langgraph langchain langchain-openai如果需要持久化状态可能还要装langgraph-checkpoint以及对应数据库驱动例如 SQLite 或 Postgres 驱动。不需要时先不装保持最小依赖。安装完成后快速验证一下版本python -c import langgraph; print(langgraph.__version__)只要不报错环境就通了。有些环境会因为 Python 版本过旧或依赖冲突失败优先升级 Python再用虚拟环境安装。3.3 模型接口配置LangGraph 本身不包含模型。我建议先用 OpenAI 协议兼容的接口做测试这样后面切换本地模型或云端模型都不需要改节点逻辑。export OPENAI_API_KEY你的密钥或本地服务密钥 export OPENAI_API_BASEhttp://127.0.0.1:8000/v1如果只是测试图结构甚至可以在节点里写死返回值不调用真实模型。调试完再接入模型这个习惯可以帮你快速定位问题出在图结构还是模型调用。4. 实战一手写一个状态管理工作流第一个实战目标是用 LangGraph 实现一个带状态累积的多步工作流。这个例子能回答另一个高频问题“LangGraph 如何在节点函数里改变 State 状态值”。4.1 定义状态假设我们要做一个小型问答流程用户输入问题系统记录历史然后调用一个处理节点最后返回结论。状态里有消息列表和步数。from typing_extensions import TypedDict, Annotated import operator class QAState(TypedDict): messages: Annotated[list, operator.add] step: int user_question: str answer: strmessages字段用operator.add意思是节点每次返回messages都会把新消息追加到已有列表后面而不是覆盖旧消息。step每次由节点手动加一。4.2 编写节点函数每个节点是一个普通 Python 函数接收 State返回需要更新的字段。def receive_question(state: QAState) - QAState: print(收到问题:, state[user_question]) return { messages: [已收到用户问题], step: state[step] 1 } def process_step(state: QAState) - QAState: # 模拟一次中间计算 interim f正在分析: {state[user_question]} return { messages: [interim], step: state[step] 1 } def make_answer(state: QAState) - QAState: answer f这是针对『{state[user_question]}』的最终回答 return { messages: [answer], answer: answer, step: state[step] 1 }看到规律没有节点函数不直接修改传入的 State而是返回一个字典LangGraph 会把返回的字典合并到全局 State 中。这个设计非常清晰避免了多个节点并发写同一个变量带来的混乱。4.3 构建图和编译from langgraph.graph import StateGraph, START, END builder StateGraph(QAState) builder.add_node(receive, receive_question) builder.add_node(process, process_step) builder.add_node(answer, make_answer) builder.add_edge(START, receive) builder.add_edge(receive, process) builder.add_edge(process, answer) builder.add_edge(answer, END) graph builder.compile()编译之后graph是一个可调用对象。LangGraph 的底层通过 Pregel 模型执行图调用方式类似普通 Runnable。4.4 执行并查看状态变化result graph.invoke({ messages: [], step: 0, user_question: LangGraph 的 State 是怎么更新的, answer: }) print(result)预期结果收到问题: LangGraph 的 State 是怎么更新的 {messages: [已收到用户问题, 正在分析: LangGraph 的 State 是怎么更新的, 这是针对『LangGraph 的 State 是怎么更新的』的最终回答], step: 3, user_question: LangGraph 的 State 是怎么更新的, answer: 这是针对『LangGraph 的 State 是怎么更新的』的最终回答}判断成功标准step 3messages按顺序累积了三段内容。这说明每条边都走了而且状态更新策略生效了。如果想看每一步执行过程可以用stream替代invokefor chunk in graph.stream({ messages: [], step: 0, user_question: 测试流式输出, answer: }): print(chunk)这种状态下能很直观地看到每个节点产生的增量。5. 实战二条件分支与循环检测第二个实战实现一个带条件路由的流程。这是 LangGraph 被搜索最多的功能点conditional_edge深度解析、循环检测、子图和并行分支。5.1 场景设计假设我们有一个内容审核流程用户提交一段文本先判断文本主题类型再走对应处理流程。这特别适合用条件边实现。5.2 路由函数条件边需要一个路由函数输入是 State输出是一个字符串。这个字符串会被拿去条件边映射表里匹配下一个节点。def classify_topic(state: QAState) - str: text state[user_question] if 代码 in text or python in text.lower(): return tech if 价格 in text or 购买 in text: return sales return other注意路由函数不要写复杂逻辑尽量只做分类判断。它是在图执行过程中被同步调用的太重的计算会影响整体性能。5.3 构建带条件边的图这里为了演示分支结果每个分支都放一个简单节点。def handle_tech(state: QAState) - QAState: return {messages: [进入技术处理分支], step: state[step] 1} def handle_sales(state: QAState) - QAState: return {messages: [进入销售处理分支], step: state[step] 1} def handle_other(state: QAState) - QAState: return {messages: [进入通用处理分支], step: state[step] 1} builder StateGraph(QAState) builder.add_node(classify, classify_topic) builder.add_node(tech, handle_tech) builder.add_node(sales, handle_sales) builder.add_node(other, handle_other) builder.add_edge(START, classify) builder.add_conditional_edges( classify, classify_topic, { tech: tech, sales: sales, other: other, } ) builder.add_edge(tech, END) builder.add_edge(sales, END) builder.add_edge(other, END) graph builder.compile()执行时classify 节点会根据输入内容自动选择下游分支。result graph.invoke({ messages: [], step: 0, user_question: python 代码怎么写, answer: }) print(result[messages]) # 预期输出[进入技术处理分支]5.4 循环与递归限制LangGraph 的图结构天然支持循环节点 A 连到节点 B节点 B 根据条件又连回节点 A。这种模式经常用于“继续追问直到必要条件满足”。但是循环需要终止条件。LangGraph 默认有递归限制防止图无限循环耗尽资源。result graph.invoke( state, config{recursion_limit: 10} )如果循环次数超过recursion_limit会抛出类似RecursionError或提示达到最大递归深度的异常。排错时遇到这个错误优先检查路由函数是否在某种输入下永远返回“回到上一个节点”的结果。5.5 并行分支多个独立节点可以并行执行。LangGraph 提供了SendAPI可以动态向多个节点分发任务。from langgraph.types import Send def fan_out(state: QAState) - list[Send]: return [ Send(process_item, {messages: [fitem {i}], step: 0}) for i in range(3) ]注意并行分支会提高吞吐但也会在短时间内创建多个子任务。如果每个子任务内部都要调用外部 LLM 服务要注意接口限流和资源峰值。5.6 子图把一个图塞进另一个图LangGraph 支持子图。你可以把一张已经编译好的图作为一个节点注册到另一张图里。builder.add_node(sub_workflow, sub_graph)这样做的好处子图可以内部测试再作为完整工作流的一部分复用。多 Agent 协作中每个 Agent 都可以是自己的子图。6. 实战三多 Agent 协作主从模式与 Subagent 即 Tool多 Agent 是当前最热的方向。网络上有一种典型设计“主从模式其实本质上将 Subagent 视作另一种 Tool 进行调用”。这个描述很准确也是 LangGraph 里最实用的一种多 Agent 集成方式。6.1 为什么把 Subagent 当 Tool 用如果你让一个主 Agent 直接面对所有工具和输入它会变得很难控制。主从模式的做法是主 Agent 负责理解用户意图、做任务规划然后把具体子任务交给 Subagent。Subagent 本质上就是一个被包装成 Tool 的可调用图。这样做有几个好处主 Agent 不需要知道每个子任务的实现细节。每个 Subagent 可以专注一个领域提示词和工具都更纯净。主 Agent 的工具列表就是一行“调用子任务”降低了模型选择工具的复杂度。每个 Subagent 可以有自己的 State 和内部流程职责隔离。6.2 实现思路首先把每个 Subagent 编译成一个图from langgraph.graph import StateGraph, START, END class SubTaskState(TypedDict): topic: str result: str def research_node(state: SubTaskState) - SubTaskState: # 替换为真实调用检索或 LLM 的逻辑 return {result: f已完成资料收集: {state[topic]}} sub_builder StateGraph(SubTaskState) sub_builder.add_node(research, research_node) sub_builder.add_edge(START, research) sub_builder.add_edge(research, END) research_agent sub_builder.compile()然后把这个子图包装成一个 Tool。LangChain 生态里自定义 Tool 最简单的写法如下注意不同版本 API 差异from langchain_core.tools import tool tool def research_tool(topic: str) - str: 当你需要收集某个主题的资料时调用。输入为主题名称。 result research_agent.invoke({topic: topic, result: }) return result.get(result, )这样主 Agent 的工具列表里就多了一个research_tool。接下来用普通 Agent 模式把工具交给主模型。6.3 主 Agent 与多工具组合主 Agent 的节点里可以使用create_react_agent或手动编写 Agent 节点。为了不过度依赖某个实现版本这里给出手动节点的方式核心思路是主 Agent 决定调用哪个工具工具执行完把结果写回 State。def supervisor_node(state: QAState) - QAState: # 伪代码模型判断是否需要调用 research_tool然后执行工具 if 资料 in state[user_question]: tool_result research_tool.invoke(state[user_question]) return {messages: [fSubagent 返回: {tool_result}], step: state[step] 1} return {messages: [主 Agent 直接回答], step: state[step] 1}真实项目里主 Agent 会用 LLM 的 tool calling 机制动态决定是否调用 Subagent而不是这种硬编码判断。上面这个例子是为了让你先看清调用链主 Agent - Tool - Subagent 子图 - 返回结果。6.4 多 Agent 协作的扩展方向主从模式一个 Supervisor 统一调度多个 Worker。网络模式Agent 之间互相传递消息图中存在多条环状边。层级模式多个子图嵌套每个子图内部还可以有自己的多 Agent 结构。从工程角度顺序建议是先做主从模式因为最可控。等状态设计、权限边界、日志追踪都成熟了再考虑更复杂的网络模式。6.5 多 Agent 协作的使用边界多 Agent 场景下工具调用权限和内容安全必须提前规划给每个 Subagent 最小化的工具权限不要让它能调用不相关的系统接口。涉及读取用户隐私数据、人脸、声音、文件内容时必须先获得合法授权。所有 Agent 的工具调用步骤建议记录日志方便审计。不要让 Agent 直接执行业务系统的高风险操作先走人工确认流程。输出内容在对外发布前要做复核避免模型生成的结果被直接采用。7. 接口 API 与批量任务接入LangGraph 图编译后就是一个 Python 对象可以直接在项目里调用。但如果要做服务化部署、对外提供 API、批量处理任务还需要把图暴露成接口。7.1 服务化部署思路LangGraph 官方提供了服务化部署方案可以把编译后的图包装成一个 API 服务。具体启动命令和 Docker 镜像请以当前官方文档为准。通用做法是用 FastAPI 写一个薄服务层内部调用 LangGraph 图然后暴露 HTTP 接口。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class GraphInput(BaseModel): user_question: str class GraphOutput(BaseModel): result: dict app.post(/run, response_modelGraphOutput) def run_graph(payload: GraphInput): result graph.invoke({ messages: [], step: 0, user_question: payload.user_question, answer: }) return GraphOutput(resultresult)启动服务uvicorn app:app --host 127.0.0.1 --port 8000这里我写的是通用 FastAPI 模板实际项目里你要根据自己图中的 State 结构调整初始化参数。7.2 客户端调用示例部署好服务后客户端用 requests 调用import requests url http://127.0.0.1:8000/run payload {user_question: 帮我分析一段日志文件中的错误} response requests.post(url, jsonpayload, timeout120) print(response.json())如果服务返回超时优先检查图内部是否有循环、递归限制和外部模型调用时长。7.3 批量任务设计LangGraph 图对象本身支持批量风格调用但更稳妥的批量处理是在应用层做。建议流程是把批量输入写入任务队列数据库表或消息队列。工作进程逐条读取任务调用图处理。记录每条任务的输入、耗时、输出和状态。失败任务进入重试队列设置最大重试次数。完成后写入结果表或通知下游系统。伪代码示例def process_batch(tasks): results [] for task in tasks: try: result graph.invoke(task[state]) results.append({ task_id: task[id], status: success, result: result }) except Exception as exc: results.append({ task_id: task[id], status: failed, error: str(exc) }) return results批量任务最常见的坑不是 LangGraph 本身而是对模型服务的并发压力。每个任务都要调用 LLM批量并发很容易触发限流建议加信号量控制并发数并设置合理的超时时间。8. 资源占用与性能观察LangGraph 本身是 CPU 上的图执行引擎不直接吃显存。性能瓶颈通常在节点里调用的模型、检索器和外部服务上。8.1 需要观察哪些指标图执行时间每个节点的耗时LangGraph 的 stream 和日志可以看出节点级别耗时。LLM 调用耗时模型接口的响应时间、token 数、并发延迟。内存占用State 中累积的消息列表会随着流程变长而增大长时间运行的图要注意清理历史消息。外部工具调用耗时数据库查询、HTTP 请求、文件读写的耗时。观察方式最简单的是在节点函数里打时间戳import time def timed_node(state: QAState) - QAState: start time.time() # 节点逻辑 elapsed time.time() - start print(f节点耗时: {elapsed:.2f}s) return {messages: [fdone], step: state[step] 1}8.2 如何降低资源占用几个有效优化手段限制 State 里的消息长度只保留最近 N 轮对话。不要一个 Node 里做所有事拆细后更容易定位瓶颈。对长文本先做压缩或检索再送入模型。合理设置recursion_limit防止循环失控。批量任务加并发限制避免瞬间打满接口配额。如果 State 很大考虑用 Checkpointer 做持久化而不是长期驻留内存。8.3 端口冲突与进程残留如果你用 FastAPI 或 LangGraph 服务化方式启动会遇到端口被占用的问题。Windows 下常见netstat -ano | findstr :8000 taskkill /PID 进程号 /FLinux 下lsof -i :8000 kill 进程号服务化部署时要格外注意进程生命周期管理不要测试完留下多个后台进程占住端口。9. LangGraph 常见问题与排查方法这一节把新手最容易踩的坑汇总成一张表格。真遇到问题优先看对应行不要盲目重启。问题现象可能原因排查方式解决方案pip 安装依赖失败Python 版本过低或依赖冲突检查 python --version查看报错依赖名升级到 Python 3.10使用虚拟环境重新安装节点执行后 State 没更新节点函数忘记 return 字典或返回了未定义的字段打印节点返回值检查 State 定义让每个节点返回一个字典只包含需要更新的字段条件分支报错找不到映射值路由函数返回了映射表里不存在的字符串打印路由函数返回值检查add_conditional_edges的第三个参数映射是否覆盖所有可能返回循环执行超时或 RecursionError缺少退出条件或 recursion_limit 设置过低检查路由逻辑观察循环是否按预期退出增加退出条件或调大 recursion_limit图执行很慢节点内 LLM 调用频繁或消息列表过长逐节点打印耗时压缩历史消息、减少单图内模型调用次数调用 API 服务超时图中某个 LLM 调用超时或负载过高查服务日志看日志停在哪个节点延长请求超时时间并在节点重试逻辑中加入退避多个进程启动后端口被占用上一次服务没退出查看端口占用进程结束残留进程或更换端口模型返回格式不稳定模型 tool calling 能力弱或提示词不明确单独测试工具调用观察返回 JSON更换模型、补充 few-shot 示例或做输出校验重试还有一个很常见的坑在节点函数里直接修改传入的 State 对象而不是返回新字典。LangGraph 的 State 合并机制是基于返回值的你直接改对象可能不会触发正确的状态更新甚至会在并发执行时产生脏数据。规范做法永远是“读取 State返回更新”。10. LangGraph 最佳实践与下一步最后给一套可以直接沿用到项目里的实践经验尤其适合从零开始接 LangGraph 的团队。10.1 先小参数测试第一次跑通图不要接太多真实模型调用。先用写死返回值的节点验证图结构再逐步替换成真实 LLM 和工具。这样可以把“图结构问题”和“模型调用问题”分开。10.2 保留一套最小可运行配置给你的项目建一个examples/quickstart.py放一份最小的 StateGraph 示例。以后任何人接手只需要跑这个文件就能确认环境可用。10.3 模型文件、输入素材、输出结果分目录管理如果图涉及本地模型、输入文件或生成结果建议用清晰目录结构project/ ├── app.py ├── graph/ │ ├── state.py │ ├── nodes.py │ └── builder.py ├── inputs/ ├── outputs/ ├── logs/ └── models/分工明确后面做批量任务和日志排查都会轻松很多。10.4 批量任务要加日志和失败重试批量处理不要一条命令跑到底。每条任务都要有task_id、开始时间、结束时间、状态、错误信息。失败任务要进入重试队列并设置最大重试次数和退避时间防止一条坏数据拖垮整个队列。10.5 接口服务要限制访问范围对外暴露 LangGraph 接口时至少要加一层接口鉴权不要直接把图的能力裸奔到公网。同时设置请求体大小限制、超时时间和并发上限避免被恶意或异常请求打挂。10.6 安全与合规红线再强调一次LangGraph 只是流程编排工具安全边界在应用层。凡是涉及人脸、声音、隐私文件、版权素材、内部系统操作都要先确认授权再做最小权限隔离最后保留审计日志。工具调用的每一步都要可追溯不然一旦出错你根本不知道是哪个 Agent 哪一步出了问题。10.7 接下来的学习方向如果这篇文章你已经跑通了前两个实战下一步建议按这个顺序深入学习 Checkpointer把 State 持久化到数据库实现真正可恢复的长时间运行任务。研究子图嵌套把复杂的业务拆成多个可独立维护的小图。实现一个真实的主从多 Agent 系统把不同领域的 Subagent 当 Tool 用。尝试事件流和流式输出把 LangGraph 接到 Web 前端。探索图结构可视化工具结合 ECharts 等前端方案把节点状态渲染成实时视图。第一次用 LangGraph重点不是把 API 背熟而是把“状态怎么流动、分支怎么走、循环怎么退出”这三个问题想清楚。先能从零写出一张能跑通的图后面所有复杂功能都是在这张图上加节点和边。这篇文章值得收藏下次搭 Agent 流程时可以直接对着抄。