这次我们不聊某一个具体模型来聊一个把大模型应用从“顺序调用链”升级成“可控状态机”的编排框架——LangGraph。过去半年LangGraph 在 Agent 开发圈子里热度很高原因是它解决了 LangChain 时代最头疼的问题多步任务怎么管理状态、分支怎么走、循环怎么跳、对话记忆怎么持久化。如果你写过多轮 Agent 应用大概率遇到过“流程一复杂代码就变成一坨 if-else”的情况。LangGraph 的核心价值就是把这些流程控制从业务代码里抽出来交给图结构去管理。这篇文章会把 LangGraph 从安装部署、核心概念、条件路由、子图、并行分支、长期记忆到接口 API 包装和批量任务完整过一遍。全文不是概念罗列而是按“先跑通再深入最后排错”的思路来写。每一步都有可复制的代码读完之后你能独立搭出一个带分支、带记忆、可对外提供 API 的 LangGraph 应用。如果你最近在学 LangGraph或者正打算把已有的 LangChain 应用改造成图状态管理这篇可以直接收藏。1. LangGraph 核心能力速览先把最关键的信息放在前面方便你快速判断这个框架适不适合现阶段去学。能力项说明项目类型大模型应用编排框架基于图结构管理 Agent 工作流开源来源LangChain 团队开源属于 LangChain 生态核心组件核心功能状态管理、节点编排、条件路由、循环控制、子图嵌套、并行分支、检查点记忆运行环境Python 3.9 及以上纯 Python 实现不依赖特定 GPU显存占用框架本身几乎不占显存实际开销取决于节点内调用的大模型是 API 还是本地模型是否支持 CPU支持。LangGraph 只是编排层CPU 上也能跑只是模型推理速度由模型决定支持平台Windows / macOS / Linux 均可启动方式以 Python 库方式安装通过代码定义图后编译调用是否支持 API可以。编译后的图可以包装成 FastAPI 服务对外提供接口是否支持批量任务可以。用循环或异步方式批量 invoke配合重试和日志管理与 LangChain 关系LangChain 负责模型调用和工具集成LangGraph 负责流程、状态和分支控制适合人群有 Python 基础想开发多步骤 Agent 应用的后端工程师、算法工程师、AI 应用开发者需要说明的是LangGraph 目前官方主推的是 Python 版本也有 TypeScript 版本但绝大部分教程、案例和公司落地场景都集中在 Python。热搜里提到的“Rust 版本”目前没有官方实现不需要等用 Python 就够了。2. 适用场景与使用边界LangGraph 解决的问题很具体当一个任务需要多个模型调用、多个工具调用、或者需要根据中间结果决定下一步走向时用图结构来组织远比手写代码维护起来更清晰。适合用 LangGraph 的场景包括多步骤 Agent 工作流比如“先分析用户意图再决定调哪个工具最后汇总结果”。需要条件分支的流程比如“如果文章超过 5000 字就走长文本摘要分支否则走普通摘要分支”。需要循环和自省能力的 Agent比如“生成结果后让大模型自己检查一遍不满意的生成次数上限以内重新生成”。需要对话记忆的应用LangGraph 的检查点机制可以把每一轮状态保存下来下一轮直接接着跑。多 Agent 协作比如一个 Agent 负责生成另一个 Agent 负责审核两者通过图节点串起来。不适合用 LangGraph 的场景也很明显单次简单的模型调用比如只做一次文本分类直接调 API 就行不需要上框架。不需要状态管理的一次性任务加图反而增加理解成本。团队里完全没有 Python 工程化经验纯给大模型写 Prompt 的团队建议先从简单的 LangChain 链开始。使用边界方面需要提醒三点LangGraph 本身不会替你解决模型合规问题节点里调用什么模型、模型服务是否允许商用需要你自己看对应 API 的服务条款。不要把敏感数据直接塞进 State 再拿去调第三方大模型接口。本地开发测试可以生产环境要评估数据合规风险。如果应用中涉及读取用户人脸、声音、身份证、手机号等隐私信息必须在用户明确授权的前提下进行并做好脱敏处理。3. LangGraph 环境准备与安装LangGraph 的安装非常轻量本质上就是装一个 Python 库。这里给出一套完整的环境准备流程建议在虚拟环境里操作避免污染全局 Python 环境。3.1 环境检查先确认本机 Python 版本。在终端执行python --version如果输出是 Python 3.9 及以上就可以继续。低于 3.9 建议先升级 Python。3.2 创建虚拟环境Linux / macOS / Windows PowerShell 通用做法python -m venv langgraph_env激活虚拟环境Windows PowerShelllanggraph_env\Scripts\Activate.ps1macOS / Linuxsource langgraph_env/bin/activate3.3 安装 LangGraphpip install -U langgraph如果你打算在节点里调用 OpenAI 兼容接口的模型需要再装 LangChain 的模型集成包pip install langchain-openai如果你想把图应用包装成 API 服务需要 FastAPI 和 uvicornpip install fastapi uvicorn为了后面方便调试建议把常用依赖一次性装好pip install -U langgraph langchain-openai fastapi uvicorn3.4 配置大模型 API KeyLangGraph 只负责编排真正干活的是节点里的模型。你需要准备一个可用的模型 API Key比如 OpenAI 兼容接口的 Key或者本地部署模型的接口地址。在终端设置环境变量export OPENAI_API_KEY你的-api-keyWindows PowerShell 下用$env:OPENAI_API_KEY你的-api-key如果你用的是本地模型或第三方兼容接口通过OpenAI客户端的base_url指定接口地址即可LangGraph 本身不关心你接的是什么模型。3.5 验证安装执行以下命令不报错即安装成功from langgraph.graph import StateGraph, START, END print(LangGraph import success)到这里环境就准备好了。安装过程中最常见的坑是网络超时如果 pip 下载慢可以换国内镜像源再装。4. 从零构建第一个 LangGraph 应用环境装好之后先不要急着看复杂概念直接跑通一个最小图。这个例子包含两个节点每个节点对 State 做一次数值累加最终输出结果。4.1 定义 StateLangGraph 的 State 本质是一个 TypedDict所有节点读写的共享数据都在这里面from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, START, END class State(TypedDict): total: int messages: list[str]这里定义了total和messages两个字段分别存计算值和过程日志。4.2 定义节点函数每个节点是一个普通函数接收当前 State返回一个字典。返回的字典会合并到 State 里def add_one(state: State) - dict: next_total state[total] 1 return { total: next_total, messages: state[messages] [fadd_one - {next_total}], } def add_two(state: State) - dict: next_total state[total] 2 return { total: next_total, messages: state[messages] [fadd_two - {next_total}], }4.3 构建图并编译graph StateGraph(State) graph.add_node(add_one, add_one) graph.add_node(add_two, add_two) graph.add_edge(START, add_one) graph.add_edge(add_one, add_two) graph.add_edge(add_two, END) app graph.compile()4.4 运行图result app.invoke({total: 0, messages: []}) print(result)预期输出{total: 3, messages: [add_one - 1, add_two - 3]}流程非常直观0 1 1然后1 2 3。两个节点按图结构依次执行状态被自动传递。这一步跑通之后你就已经理解了 LangGraph 最核心的四个要素State 存数据、Node 处理数据、Edge 决定流向、compile 之后可以 invoke 执行。5. LangGraph 核心机制深入节点、状态与条件路由上一步只是串行执行没有体现 LangGraph 真正的价值。这一节重点讲状态更新规则和条件路由这是 LangGraph 和普通函数调用链最大的区别。5.1 节点返回值如何合并到 State每个节点函数返回一个 dictLangGraph 会把它更新到 State 中。默认行为是“覆盖”如果返回的 key 在 State 中已存在就替换掉旧值。上面例子里messages用了state[messages] [msg]这是手动做追加。真实项目中消息列表通常需要自动追加而不是覆盖。LangGraph 提供了Annotated和 reducer 机制来解决这个问题。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] total: intadd_messages是 LangGraph 内置的 reducer节点返回新消息时会自动追加到messages列表而不是覆盖旧消息。这在对话类 Agent 中非常常用。你可以同时使用两种字段需要覆盖的用普通类型需要追加的用Annotated加 reducer。5.2 条件路由与分支控制条件路由解决的是“根据当前状态决定下一步走哪个节点”的问题。这是 LangGraph 实战中最常写的代码。先看一个综合示例假设有一个工作流第一节点根据用户输入生成一个数字然后判断这个数字是否大于等于 10决定走“直接结束”还是“再加一次”。class State(TypedDict): user_input: int total: int messages: Annotated[list, add_messages] def generate_number(state: State) - dict: return {total: state[user_input], messages: [generate_number 完成]} def add_one(state: State) - dict: return {total: state[total] 1, messages: [add_one 完成]} def too_small(state: State) - dict: return {messages: [f当前值 {state[total]} 仍小于 10流程结束]} def route_by_total(state: State) - Literal[add_one, too_small, END]: if state[total] 10: return add_one if state[total] 10: return END return too_small构建图graph StateGraph(State) graph.add_node(generate_number, generate_number) graph.add_node(add_one, add_one) graph.add_node(too_small, too_small) graph.add_edge(START, generate_number) graph.add_conditional_edges( generate_number, route_by_total, { add_one: add_one, too_small: too_small, END: END, }, ) graph.add_edge(add_one, generate_number) app graph.compile()测试print(app.invoke({user_input: 8, total: 0, messages: []})) # 8 10走 add_one 变为 9重新进入 generate_number此时 9 10继续 add_one 变为 10然后流程重新到 generate_number10 不小于 10 且等于 10走 END这个例子展示了条件路由最典型的三种走向分支到一个普通节点。分支回自身之前的节点形成循环。分支直接到 END 结束流程。热搜里提到的“conditional_edge 深度解析、循环检测”指的就是这里。LangGraph 允许图中有循环因为generate_number - add_one - generate_number的路径是合法的。但要注意循环必须有退出条件否则会无限执行。LangGraph 默认每个节点最多执行一定次数超出后会抛出GraphRecursionError需要在compile(recursion_limit...)或invoke(config{recursion_limit: 50})中调整上限。6. 高级特性子图、并行分支与长期记忆基础的路由和状态管理掌握之后LangGraph 的高阶能力可以大幅简化复杂系统的开发。6.1 子图嵌套子图就是把一个编译好的图当作另一个图的节点来使用。适用于“先跑一个独立子流程再继续主流程”的场景。# 子图定义 sub_graph StateGraph(State) sub_graph.add_node(sub_add, add_one) sub_graph.add_edge(START, sub_add) sub_graph.add_edge(sub_add, END) sub_app sub_graph.compile() # 主图使用子图 main_graph StateGraph(State) main_graph.add_node(main_gen, generate_number) main_graph.add_node(sub, sub_app) # 直接把编译后的子图作为节点 main_graph.add_edge(START, main_gen) main_graph.add_edge(main_gen, sub) main_graph.add_edge(sub, END) main_app main_graph.compile()子图的好处是每个子图可以独立测试、独立复用主图的逻辑不膨胀。多 Agent 协作时可以把每个 Agent 封装成子图在主图里统一调度。6.2 并行分支LangGraph 支持从一个节点出发同时进入多个节点最后在某个节点汇合。适合做“同时调用多个工具最后汇总结果”的场景。def tool_a(state: State) - dict: return {messages: [工具 A 完成]} def tool_b(state: State) - dict: return {messages: [工具 B 完成]} def aggregate(state: State) - dict: return {messages: [汇总完成]} graph StateGraph(State) graph.add_node(tools_a, tool_a) graph.add_node(tools_b, tool_b) graph.add_node(aggregate, aggregate) graph.add_edge(START, tools_a) graph.add_edge(START, tools_b) graph.add_edge(tools_a, aggregate) graph.add_edge(tools_b, aggregate) graph.add_edge(aggregate, END) app graph.compile()这里需要理解一点LangGraph 的并行是异步并发执行的但最终结果仍然会汇总到同一个 State。汇合节点aggregate会在所有上游节点都完成后才执行这种机制叫 Fan-out / Fan-in。6.3 长期记忆与检查点LangGraph 的“长期记忆”能力来自 Checkpointer。每次invoke结束后可以把这一轮的状态保存到内存或数据库里下一轮invoke时把历史状态加载出来继续。这就是热搜里“langgraph 长期记忆”的核心机制。先用最简单的内存检查点from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory) # 第一轮调用thread_id 用于区分会话 config {configurable: {thread_id: user-001}} result1 app.invoke({user_input: 10, total: 0, messages: []}, configconfig) print(result1) # 第二轮调用带上同一个 thread_id可以接着上一轮的状态继续 result2 app.invoke({user_input: 0, total: 0, messages: []}, configconfig) print(result2)生产环境中不要用内存检查点服务一重启就没了。可以使用 SQLite 或 Postgres 作为检查点存储后端LangGraph 官方提供对应实现具体安装方式以官方文档为准。检查点机制意味着每个用户会话可以拥有独立的记忆空间。只需要在请求时传入不同的thread_idLangGraph 就会自动隔离不同会话的状态。这对做聊天机器人、客服系统非常关键。7. 与 LangChain 集成模型调用和工具编排这一节解决一个高频疑问LangGraph 和 LangChain 到底是什么关系实际项目中怎么配合使用LangChain 是一套模型调用和工具集成的抽象层。你可以用langchain-openai的ChatOpenAI在节点里调用大模型用langchain_core.tools定义工具让模型决定是否调用工具。LangGraph 则是状态和工作流编排层。它不关心模型是哪家的也不关心工具的具体实现只管理节点的执行顺序、状态传递和分支逻辑。实际代码中可以这样配合from langchain_openai import ChatOpenAI from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询城市天气 # 这里做真实的天气查询 return f{city} 天气晴 def call_model(state: State) - dict: llm ChatOpenAI(modelgpt-4o-mini) llm_with_tools llm.bind_tools([get_weather]) result llm_with_tools.invoke(state[messages]) return {messages: [result]} def call_tool(state: State) - dict: # 解析模型返回的 tool_call执行工具 return {messages: [get_weather.invoke(北京)]} def route_by_tool(state: State) - Literal[call_tool, END]: last_message state[messages][-1] if getattr(last_message, tool_calls, None): return call_tool return END这个结构就是 ReAct Agent 的原型模型决定要不要调工具调完工具把结果写回消息队列再次交给模型判断。这种“思考-行动-观察”的循环是 LangGraph 最擅长处理的场景。给一个选型建议如果你的应用只是单个模型调用不需要 LangGraph如果你已经用了 LangChain 写工具调用但分支和状态控制靠手写代码强烈建议迁移到 LangGraph。它不是取代 LangChain而是补上 LangChain 最不擅长的流程控制能力。8. 可观测性与调试如何看清每一步状态图节点一多最怕的是不知道流程走到哪里、状态变成什么样。LangGraph 提供了两种调试手段。8.1 打印中间状态最简单的方法是每个节点里打印关键信息def add_one(state: State) - dict: next_total state[total] 1 print(f[add_one] total from {state[total]} to {next_total}) return {total: next_total}这种方式的缺点是日志散落在各节点里图大了以后不好统一管理。建议在节点函数里用logging模块统一输出。8.2 可视化图结构LangGraph 可以把编译后的图导出为 Mermaid 格式方便你在查看图结构时使用。在 Jupyter 或 Python 脚本里执行# 导出 Mermaid 文本可以复制到支持 Mermaid 的工具中查看 print(app.get_graph().draw_mermaid())这里说明一下draw_mermaid()返回的是图结构的 Mermaid 文本你可以复制到支持 Mermaid 渲染的工具比如 draw.io、Typora、语雀等中查看节点流向。这比用 echarts 自己实现节点可视化要省事得多。如果你想做更精细的 Web 端可视化调试可以考虑 LangSmith。它是 LangChain 团队提供的全链路追踪平台能看到每一步节点的输入输出、耗时、token 消耗。使用方式是在环境变量里配置 LangSmith 的 API Keyexport LANGCHAIN_TRACING_V2true export LANGCHAIN_API_KEY你的-langsmith-api-key配置之后每次invoke都会自动上报整条链路的执行记录。不过需要注意使用 LangSmith 会上传少量追踪数据到官方平台如果对数据敏感关闭 tracing 只用本地日志即可。9. 使用 FastAPI 包装接口服务与批量任务LangGraph 本身不是一个服务但编译后的app可以非常方便地嵌入到 FastAPI 服务中对外提供 HTTP 接口。9.1 创建 FastAPI 服务示例代码from fastapi import FastAPI from pydantic import BaseModel from langgraph.graph import StateGraph, START, END from typing import TypedDict class State(TypedDict): total: int messages: list[str] def add_one(state: State) - dict: next_total state[total] 1 return {total: next_total, messages: state[messages] [fadd_one - {next_total}]} graph StateGraph(State) graph.add_node(add_one, add_one) graph.add_edge(START, add_one) graph.add_edge(add_one, END) app graph.compile(checkpointerMemorySaver()) fastapi_app FastAPI() class RunRequest(BaseModel): thread_id: str total: int 0 messages: list[str] [] fastapi_app.post(/run) def run(req: RunRequest): config {configurable: {thread_id: req.thread_id}} result app.invoke( {total: req.total, messages: req.messages}, configconfig, ) return result fastapi_app.get(/health) def health(): return {status: ok}启动服务uvicorn main:fastapi_app --host 127.0.0.1 --port 80009.2 curl 调用接口curl -X POST http://127.0.0.1:8000/run \ -H Content-Type: application/json \ -d {thread_id: u-001, total: 0, messages: []}返回结果自带上一次thread_id的历史状态实现多轮会话记忆。9.3 批量任务处理LangGraph 的invoke是同步阻塞的直接在循环里跑即可但建议加上错误隔离和重试import time import logging def run_batch(inputs: list[dict], max_retries: int 3) - list: results [] for item in inputs: for attempt in range(max_retries): try: result app.invoke( {total: item[total], messages: []}, config{configurable: {thread_id: item[thread_id]}}, ) results.append(result) break except Exception as exc: logging.error(ftask {item[thread_id]} failed attempt {attempt 1}: {exc}) if attempt max_retries - 1: results.append({error: str(exc)}) time.sleep(2 ** attempt) # 指数退避重试 return results批量任务的关键点有三个失败重试、任务日志、结果隔离。不要让一个失败任务拖垮整批任务。10. 资源占用与性能观察LangGraph 本身是纯 Python 编排层CPU 和内存开销非常低。真正的资源瓶颈在节点内调用的模型。分两种情况看调用云端 API 模型本机只承担 HTTP 请求和状态管理显存基本不占用。CPU 完全够用瓶颈在 API 响应延迟和 Rate Limit。调用本地模型显存占用完全取决于你加载的模型大小。比如 7B 模型通常需要 6G 到 8G 显存13B 模型需要 12G 到 16G 显存。具体数字以本地部署工具的实测为准。性能观察建议用time.perf_counter()统计每个节点的耗时定位哪一步最慢。如果节点内部有多个模型调用优先做并行化配合 LangGraph 的 Fan-out 机制。给每个节点加超时控制。避免某次 API 调用卡住整个流程。如果相同请求频繁执行考虑加缓存。LangGraph 允许在节点内部做结果缓存但缓存 key 需要自己设计。需要强调一点不要把 LangGraph 的编排延迟和模型推理延迟混在一起。大多数情况下一次图编排的纯框架开销在毫秒级真正的耗时是大模型生成 token 的时间。11. 常见问题与排查方法下面是 LangGraph 开发中比较高频的问题和排查思路。问题现象可能原因排查方式解决方案pip 安装失败网络问题或依赖冲突查看 pip 报错日志换国内镜像源或升级 pip 后重试导入 langgraph 报错Python 版本过低检查python --version升级到 Python 3.9 以上invoke 报 API Key 错误环境变量未设置或 Key 无效打印os.environ.get(OPENAI_API_KEY)重新配置环境变量节点返回的值没有更新到 State返回 key 与 State 字段不一致检查节点返回值字段名确保返回字典的 key 与 State 定义一致messages 一直覆盖而不是追加没配置 reducer检查字段是否用了Annotated消息列表字段加add_messagesreducer条件路由报“找不到节点”条件边映射到不存在的节点名检查add_conditional_edges的映射表确保分支名与add_node名称一致图递归次数超限循环没有退出条件查看是否出现GraphRecursionError给循环节点加退出条件或调大recursion_limit多轮对话状态串了thread_id 没有隔离检查每次请求是否传了不同的 thread_id每个用户会话使用独立 thread_id服务部署后接口不稳定第三方 API 限流查看 API 返回的限流报错增加重试和退避策略批量任务卡住单个任务 API 超时给 invoke 加超时在节点内设置请求超时或异步并发执行如果你的问题不在表里优先看三样东西报错堆栈、节点返回值的类型、State 里字段的当前值。绝大多数 LangGraph 问题都能通过这三样定位。12. 学习路线与实战建议如果你计划在七天左右从入门到能写项目可以参考下面这个节奏但核心是“先跑通再深入”第 1 天安装环境跑通最小图理解 State、Node、Edge 三个概念。第 2 天写条件路由掌握add_conditional_edges的用法。第 3 天写带循环的 Agent理解recursion_limit。第 4 天给图加 Checkpointer实现多轮对话记忆。第 5 天把 LangChain 的模型和工具集成到节点里。第 6 天用 FastAPI 包装接口跑通批量任务。第 7 天选一个真实小项目比如“PDF 问答 Agent”把前面所有能力串起来。工程实践上有几条建议第一版图不要设计得太复杂先用最少的节点跑通主流程再逐步加分支。所有节点函数保持单一职责一个节点只做一件事。模型相关参数模型名、温度、超时时间放到配置里不要写死在节点代码中。批量任务必须加日志和错误隔离否则上游一个 API 抖动整批任务都会失败。接口服务部署到公网前务必加鉴权。不要在公网裸奔一个能让你花钱调模型的接口。涉及用户隐私数据、版权素材时先确认授权再上线。13. 总结LangGraph 最值得尝试的点是它把复杂 Agent 应用的流程控制从手写 if-else 中解放出来让状态、分支、循环、记忆都变成图结构的一部分。而且它和 LangChain 天然互补团队如果已经在用 LangChain迁移成本并不高。拿到一个 LangGraph 项目最先应该验证三件事最小图能不能跑通、条件路由是否正确走到目标分支、跨轮记忆是否按 thread_id 隔离。这三件事验证通过框架的骨架就算掌握了。最容易踩的坑也有三个第一个是节点返回值覆盖了 State 里的旧字段导致信息丢失第二个是条件路由分支名写错GraphRecursionError反复出现第三个是生产环境用了内存检查点服务重启后对话记忆全部丢失。后续可以继续扩展的方向是把本地模型接入 LangGraph做多 Agent 协作系统以及结合 LangSmith 做更精细的链路分析。LangGraph 的生态还在快速迭代社区里已经有很多基于它搭建的 Agent 产品值得跟进。建议先跑通第 4 节的最小图再花半天时间把条件路由和记忆两个机制玩熟。这套东西学会之后写复杂 Agent 的维护成本会明显下降。
LangGraph实战：从状态管理到条件路由，构建可控Agent工作流
这次我们不聊某一个具体模型来聊一个把大模型应用从“顺序调用链”升级成“可控状态机”的编排框架——LangGraph。过去半年LangGraph 在 Agent 开发圈子里热度很高原因是它解决了 LangChain 时代最头疼的问题多步任务怎么管理状态、分支怎么走、循环怎么跳、对话记忆怎么持久化。如果你写过多轮 Agent 应用大概率遇到过“流程一复杂代码就变成一坨 if-else”的情况。LangGraph 的核心价值就是把这些流程控制从业务代码里抽出来交给图结构去管理。这篇文章会把 LangGraph 从安装部署、核心概念、条件路由、子图、并行分支、长期记忆到接口 API 包装和批量任务完整过一遍。全文不是概念罗列而是按“先跑通再深入最后排错”的思路来写。每一步都有可复制的代码读完之后你能独立搭出一个带分支、带记忆、可对外提供 API 的 LangGraph 应用。如果你最近在学 LangGraph或者正打算把已有的 LangChain 应用改造成图状态管理这篇可以直接收藏。1. LangGraph 核心能力速览先把最关键的信息放在前面方便你快速判断这个框架适不适合现阶段去学。能力项说明项目类型大模型应用编排框架基于图结构管理 Agent 工作流开源来源LangChain 团队开源属于 LangChain 生态核心组件核心功能状态管理、节点编排、条件路由、循环控制、子图嵌套、并行分支、检查点记忆运行环境Python 3.9 及以上纯 Python 实现不依赖特定 GPU显存占用框架本身几乎不占显存实际开销取决于节点内调用的大模型是 API 还是本地模型是否支持 CPU支持。LangGraph 只是编排层CPU 上也能跑只是模型推理速度由模型决定支持平台Windows / macOS / Linux 均可启动方式以 Python 库方式安装通过代码定义图后编译调用是否支持 API可以。编译后的图可以包装成 FastAPI 服务对外提供接口是否支持批量任务可以。用循环或异步方式批量 invoke配合重试和日志管理与 LangChain 关系LangChain 负责模型调用和工具集成LangGraph 负责流程、状态和分支控制适合人群有 Python 基础想开发多步骤 Agent 应用的后端工程师、算法工程师、AI 应用开发者需要说明的是LangGraph 目前官方主推的是 Python 版本也有 TypeScript 版本但绝大部分教程、案例和公司落地场景都集中在 Python。热搜里提到的“Rust 版本”目前没有官方实现不需要等用 Python 就够了。2. 适用场景与使用边界LangGraph 解决的问题很具体当一个任务需要多个模型调用、多个工具调用、或者需要根据中间结果决定下一步走向时用图结构来组织远比手写代码维护起来更清晰。适合用 LangGraph 的场景包括多步骤 Agent 工作流比如“先分析用户意图再决定调哪个工具最后汇总结果”。需要条件分支的流程比如“如果文章超过 5000 字就走长文本摘要分支否则走普通摘要分支”。需要循环和自省能力的 Agent比如“生成结果后让大模型自己检查一遍不满意的生成次数上限以内重新生成”。需要对话记忆的应用LangGraph 的检查点机制可以把每一轮状态保存下来下一轮直接接着跑。多 Agent 协作比如一个 Agent 负责生成另一个 Agent 负责审核两者通过图节点串起来。不适合用 LangGraph 的场景也很明显单次简单的模型调用比如只做一次文本分类直接调 API 就行不需要上框架。不需要状态管理的一次性任务加图反而增加理解成本。团队里完全没有 Python 工程化经验纯给大模型写 Prompt 的团队建议先从简单的 LangChain 链开始。使用边界方面需要提醒三点LangGraph 本身不会替你解决模型合规问题节点里调用什么模型、模型服务是否允许商用需要你自己看对应 API 的服务条款。不要把敏感数据直接塞进 State 再拿去调第三方大模型接口。本地开发测试可以生产环境要评估数据合规风险。如果应用中涉及读取用户人脸、声音、身份证、手机号等隐私信息必须在用户明确授权的前提下进行并做好脱敏处理。3. LangGraph 环境准备与安装LangGraph 的安装非常轻量本质上就是装一个 Python 库。这里给出一套完整的环境准备流程建议在虚拟环境里操作避免污染全局 Python 环境。3.1 环境检查先确认本机 Python 版本。在终端执行python --version如果输出是 Python 3.9 及以上就可以继续。低于 3.9 建议先升级 Python。3.2 创建虚拟环境Linux / macOS / Windows PowerShell 通用做法python -m venv langgraph_env激活虚拟环境Windows PowerShelllanggraph_env\Scripts\Activate.ps1macOS / Linuxsource langgraph_env/bin/activate3.3 安装 LangGraphpip install -U langgraph如果你打算在节点里调用 OpenAI 兼容接口的模型需要再装 LangChain 的模型集成包pip install langchain-openai如果你想把图应用包装成 API 服务需要 FastAPI 和 uvicornpip install fastapi uvicorn为了后面方便调试建议把常用依赖一次性装好pip install -U langgraph langchain-openai fastapi uvicorn3.4 配置大模型 API KeyLangGraph 只负责编排真正干活的是节点里的模型。你需要准备一个可用的模型 API Key比如 OpenAI 兼容接口的 Key或者本地部署模型的接口地址。在终端设置环境变量export OPENAI_API_KEY你的-api-keyWindows PowerShell 下用$env:OPENAI_API_KEY你的-api-key如果你用的是本地模型或第三方兼容接口通过OpenAI客户端的base_url指定接口地址即可LangGraph 本身不关心你接的是什么模型。3.5 验证安装执行以下命令不报错即安装成功from langgraph.graph import StateGraph, START, END print(LangGraph import success)到这里环境就准备好了。安装过程中最常见的坑是网络超时如果 pip 下载慢可以换国内镜像源再装。4. 从零构建第一个 LangGraph 应用环境装好之后先不要急着看复杂概念直接跑通一个最小图。这个例子包含两个节点每个节点对 State 做一次数值累加最终输出结果。4.1 定义 StateLangGraph 的 State 本质是一个 TypedDict所有节点读写的共享数据都在这里面from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, START, END class State(TypedDict): total: int messages: list[str]这里定义了total和messages两个字段分别存计算值和过程日志。4.2 定义节点函数每个节点是一个普通函数接收当前 State返回一个字典。返回的字典会合并到 State 里def add_one(state: State) - dict: next_total state[total] 1 return { total: next_total, messages: state[messages] [fadd_one - {next_total}], } def add_two(state: State) - dict: next_total state[total] 2 return { total: next_total, messages: state[messages] [fadd_two - {next_total}], }4.3 构建图并编译graph StateGraph(State) graph.add_node(add_one, add_one) graph.add_node(add_two, add_two) graph.add_edge(START, add_one) graph.add_edge(add_one, add_two) graph.add_edge(add_two, END) app graph.compile()4.4 运行图result app.invoke({total: 0, messages: []}) print(result)预期输出{total: 3, messages: [add_one - 1, add_two - 3]}流程非常直观0 1 1然后1 2 3。两个节点按图结构依次执行状态被自动传递。这一步跑通之后你就已经理解了 LangGraph 最核心的四个要素State 存数据、Node 处理数据、Edge 决定流向、compile 之后可以 invoke 执行。5. LangGraph 核心机制深入节点、状态与条件路由上一步只是串行执行没有体现 LangGraph 真正的价值。这一节重点讲状态更新规则和条件路由这是 LangGraph 和普通函数调用链最大的区别。5.1 节点返回值如何合并到 State每个节点函数返回一个 dictLangGraph 会把它更新到 State 中。默认行为是“覆盖”如果返回的 key 在 State 中已存在就替换掉旧值。上面例子里messages用了state[messages] [msg]这是手动做追加。真实项目中消息列表通常需要自动追加而不是覆盖。LangGraph 提供了Annotated和 reducer 机制来解决这个问题。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] total: intadd_messages是 LangGraph 内置的 reducer节点返回新消息时会自动追加到messages列表而不是覆盖旧消息。这在对话类 Agent 中非常常用。你可以同时使用两种字段需要覆盖的用普通类型需要追加的用Annotated加 reducer。5.2 条件路由与分支控制条件路由解决的是“根据当前状态决定下一步走哪个节点”的问题。这是 LangGraph 实战中最常写的代码。先看一个综合示例假设有一个工作流第一节点根据用户输入生成一个数字然后判断这个数字是否大于等于 10决定走“直接结束”还是“再加一次”。class State(TypedDict): user_input: int total: int messages: Annotated[list, add_messages] def generate_number(state: State) - dict: return {total: state[user_input], messages: [generate_number 完成]} def add_one(state: State) - dict: return {total: state[total] 1, messages: [add_one 完成]} def too_small(state: State) - dict: return {messages: [f当前值 {state[total]} 仍小于 10流程结束]} def route_by_total(state: State) - Literal[add_one, too_small, END]: if state[total] 10: return add_one if state[total] 10: return END return too_small构建图graph StateGraph(State) graph.add_node(generate_number, generate_number) graph.add_node(add_one, add_one) graph.add_node(too_small, too_small) graph.add_edge(START, generate_number) graph.add_conditional_edges( generate_number, route_by_total, { add_one: add_one, too_small: too_small, END: END, }, ) graph.add_edge(add_one, generate_number) app graph.compile()测试print(app.invoke({user_input: 8, total: 0, messages: []})) # 8 10走 add_one 变为 9重新进入 generate_number此时 9 10继续 add_one 变为 10然后流程重新到 generate_number10 不小于 10 且等于 10走 END这个例子展示了条件路由最典型的三种走向分支到一个普通节点。分支回自身之前的节点形成循环。分支直接到 END 结束流程。热搜里提到的“conditional_edge 深度解析、循环检测”指的就是这里。LangGraph 允许图中有循环因为generate_number - add_one - generate_number的路径是合法的。但要注意循环必须有退出条件否则会无限执行。LangGraph 默认每个节点最多执行一定次数超出后会抛出GraphRecursionError需要在compile(recursion_limit...)或invoke(config{recursion_limit: 50})中调整上限。6. 高级特性子图、并行分支与长期记忆基础的路由和状态管理掌握之后LangGraph 的高阶能力可以大幅简化复杂系统的开发。6.1 子图嵌套子图就是把一个编译好的图当作另一个图的节点来使用。适用于“先跑一个独立子流程再继续主流程”的场景。# 子图定义 sub_graph StateGraph(State) sub_graph.add_node(sub_add, add_one) sub_graph.add_edge(START, sub_add) sub_graph.add_edge(sub_add, END) sub_app sub_graph.compile() # 主图使用子图 main_graph StateGraph(State) main_graph.add_node(main_gen, generate_number) main_graph.add_node(sub, sub_app) # 直接把编译后的子图作为节点 main_graph.add_edge(START, main_gen) main_graph.add_edge(main_gen, sub) main_graph.add_edge(sub, END) main_app main_graph.compile()子图的好处是每个子图可以独立测试、独立复用主图的逻辑不膨胀。多 Agent 协作时可以把每个 Agent 封装成子图在主图里统一调度。6.2 并行分支LangGraph 支持从一个节点出发同时进入多个节点最后在某个节点汇合。适合做“同时调用多个工具最后汇总结果”的场景。def tool_a(state: State) - dict: return {messages: [工具 A 完成]} def tool_b(state: State) - dict: return {messages: [工具 B 完成]} def aggregate(state: State) - dict: return {messages: [汇总完成]} graph StateGraph(State) graph.add_node(tools_a, tool_a) graph.add_node(tools_b, tool_b) graph.add_node(aggregate, aggregate) graph.add_edge(START, tools_a) graph.add_edge(START, tools_b) graph.add_edge(tools_a, aggregate) graph.add_edge(tools_b, aggregate) graph.add_edge(aggregate, END) app graph.compile()这里需要理解一点LangGraph 的并行是异步并发执行的但最终结果仍然会汇总到同一个 State。汇合节点aggregate会在所有上游节点都完成后才执行这种机制叫 Fan-out / Fan-in。6.3 长期记忆与检查点LangGraph 的“长期记忆”能力来自 Checkpointer。每次invoke结束后可以把这一轮的状态保存到内存或数据库里下一轮invoke时把历史状态加载出来继续。这就是热搜里“langgraph 长期记忆”的核心机制。先用最简单的内存检查点from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory) # 第一轮调用thread_id 用于区分会话 config {configurable: {thread_id: user-001}} result1 app.invoke({user_input: 10, total: 0, messages: []}, configconfig) print(result1) # 第二轮调用带上同一个 thread_id可以接着上一轮的状态继续 result2 app.invoke({user_input: 0, total: 0, messages: []}, configconfig) print(result2)生产环境中不要用内存检查点服务一重启就没了。可以使用 SQLite 或 Postgres 作为检查点存储后端LangGraph 官方提供对应实现具体安装方式以官方文档为准。检查点机制意味着每个用户会话可以拥有独立的记忆空间。只需要在请求时传入不同的thread_idLangGraph 就会自动隔离不同会话的状态。这对做聊天机器人、客服系统非常关键。7. 与 LangChain 集成模型调用和工具编排这一节解决一个高频疑问LangGraph 和 LangChain 到底是什么关系实际项目中怎么配合使用LangChain 是一套模型调用和工具集成的抽象层。你可以用langchain-openai的ChatOpenAI在节点里调用大模型用langchain_core.tools定义工具让模型决定是否调用工具。LangGraph 则是状态和工作流编排层。它不关心模型是哪家的也不关心工具的具体实现只管理节点的执行顺序、状态传递和分支逻辑。实际代码中可以这样配合from langchain_openai import ChatOpenAI from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询城市天气 # 这里做真实的天气查询 return f{city} 天气晴 def call_model(state: State) - dict: llm ChatOpenAI(modelgpt-4o-mini) llm_with_tools llm.bind_tools([get_weather]) result llm_with_tools.invoke(state[messages]) return {messages: [result]} def call_tool(state: State) - dict: # 解析模型返回的 tool_call执行工具 return {messages: [get_weather.invoke(北京)]} def route_by_tool(state: State) - Literal[call_tool, END]: last_message state[messages][-1] if getattr(last_message, tool_calls, None): return call_tool return END这个结构就是 ReAct Agent 的原型模型决定要不要调工具调完工具把结果写回消息队列再次交给模型判断。这种“思考-行动-观察”的循环是 LangGraph 最擅长处理的场景。给一个选型建议如果你的应用只是单个模型调用不需要 LangGraph如果你已经用了 LangChain 写工具调用但分支和状态控制靠手写代码强烈建议迁移到 LangGraph。它不是取代 LangChain而是补上 LangChain 最不擅长的流程控制能力。8. 可观测性与调试如何看清每一步状态图节点一多最怕的是不知道流程走到哪里、状态变成什么样。LangGraph 提供了两种调试手段。8.1 打印中间状态最简单的方法是每个节点里打印关键信息def add_one(state: State) - dict: next_total state[total] 1 print(f[add_one] total from {state[total]} to {next_total}) return {total: next_total}这种方式的缺点是日志散落在各节点里图大了以后不好统一管理。建议在节点函数里用logging模块统一输出。8.2 可视化图结构LangGraph 可以把编译后的图导出为 Mermaid 格式方便你在查看图结构时使用。在 Jupyter 或 Python 脚本里执行# 导出 Mermaid 文本可以复制到支持 Mermaid 的工具中查看 print(app.get_graph().draw_mermaid())这里说明一下draw_mermaid()返回的是图结构的 Mermaid 文本你可以复制到支持 Mermaid 渲染的工具比如 draw.io、Typora、语雀等中查看节点流向。这比用 echarts 自己实现节点可视化要省事得多。如果你想做更精细的 Web 端可视化调试可以考虑 LangSmith。它是 LangChain 团队提供的全链路追踪平台能看到每一步节点的输入输出、耗时、token 消耗。使用方式是在环境变量里配置 LangSmith 的 API Keyexport LANGCHAIN_TRACING_V2true export LANGCHAIN_API_KEY你的-langsmith-api-key配置之后每次invoke都会自动上报整条链路的执行记录。不过需要注意使用 LangSmith 会上传少量追踪数据到官方平台如果对数据敏感关闭 tracing 只用本地日志即可。9. 使用 FastAPI 包装接口服务与批量任务LangGraph 本身不是一个服务但编译后的app可以非常方便地嵌入到 FastAPI 服务中对外提供 HTTP 接口。9.1 创建 FastAPI 服务示例代码from fastapi import FastAPI from pydantic import BaseModel from langgraph.graph import StateGraph, START, END from typing import TypedDict class State(TypedDict): total: int messages: list[str] def add_one(state: State) - dict: next_total state[total] 1 return {total: next_total, messages: state[messages] [fadd_one - {next_total}]} graph StateGraph(State) graph.add_node(add_one, add_one) graph.add_edge(START, add_one) graph.add_edge(add_one, END) app graph.compile(checkpointerMemorySaver()) fastapi_app FastAPI() class RunRequest(BaseModel): thread_id: str total: int 0 messages: list[str] [] fastapi_app.post(/run) def run(req: RunRequest): config {configurable: {thread_id: req.thread_id}} result app.invoke( {total: req.total, messages: req.messages}, configconfig, ) return result fastapi_app.get(/health) def health(): return {status: ok}启动服务uvicorn main:fastapi_app --host 127.0.0.1 --port 80009.2 curl 调用接口curl -X POST http://127.0.0.1:8000/run \ -H Content-Type: application/json \ -d {thread_id: u-001, total: 0, messages: []}返回结果自带上一次thread_id的历史状态实现多轮会话记忆。9.3 批量任务处理LangGraph 的invoke是同步阻塞的直接在循环里跑即可但建议加上错误隔离和重试import time import logging def run_batch(inputs: list[dict], max_retries: int 3) - list: results [] for item in inputs: for attempt in range(max_retries): try: result app.invoke( {total: item[total], messages: []}, config{configurable: {thread_id: item[thread_id]}}, ) results.append(result) break except Exception as exc: logging.error(ftask {item[thread_id]} failed attempt {attempt 1}: {exc}) if attempt max_retries - 1: results.append({error: str(exc)}) time.sleep(2 ** attempt) # 指数退避重试 return results批量任务的关键点有三个失败重试、任务日志、结果隔离。不要让一个失败任务拖垮整批任务。10. 资源占用与性能观察LangGraph 本身是纯 Python 编排层CPU 和内存开销非常低。真正的资源瓶颈在节点内调用的模型。分两种情况看调用云端 API 模型本机只承担 HTTP 请求和状态管理显存基本不占用。CPU 完全够用瓶颈在 API 响应延迟和 Rate Limit。调用本地模型显存占用完全取决于你加载的模型大小。比如 7B 模型通常需要 6G 到 8G 显存13B 模型需要 12G 到 16G 显存。具体数字以本地部署工具的实测为准。性能观察建议用time.perf_counter()统计每个节点的耗时定位哪一步最慢。如果节点内部有多个模型调用优先做并行化配合 LangGraph 的 Fan-out 机制。给每个节点加超时控制。避免某次 API 调用卡住整个流程。如果相同请求频繁执行考虑加缓存。LangGraph 允许在节点内部做结果缓存但缓存 key 需要自己设计。需要强调一点不要把 LangGraph 的编排延迟和模型推理延迟混在一起。大多数情况下一次图编排的纯框架开销在毫秒级真正的耗时是大模型生成 token 的时间。11. 常见问题与排查方法下面是 LangGraph 开发中比较高频的问题和排查思路。问题现象可能原因排查方式解决方案pip 安装失败网络问题或依赖冲突查看 pip 报错日志换国内镜像源或升级 pip 后重试导入 langgraph 报错Python 版本过低检查python --version升级到 Python 3.9 以上invoke 报 API Key 错误环境变量未设置或 Key 无效打印os.environ.get(OPENAI_API_KEY)重新配置环境变量节点返回的值没有更新到 State返回 key 与 State 字段不一致检查节点返回值字段名确保返回字典的 key 与 State 定义一致messages 一直覆盖而不是追加没配置 reducer检查字段是否用了Annotated消息列表字段加add_messagesreducer条件路由报“找不到节点”条件边映射到不存在的节点名检查add_conditional_edges的映射表确保分支名与add_node名称一致图递归次数超限循环没有退出条件查看是否出现GraphRecursionError给循环节点加退出条件或调大recursion_limit多轮对话状态串了thread_id 没有隔离检查每次请求是否传了不同的 thread_id每个用户会话使用独立 thread_id服务部署后接口不稳定第三方 API 限流查看 API 返回的限流报错增加重试和退避策略批量任务卡住单个任务 API 超时给 invoke 加超时在节点内设置请求超时或异步并发执行如果你的问题不在表里优先看三样东西报错堆栈、节点返回值的类型、State 里字段的当前值。绝大多数 LangGraph 问题都能通过这三样定位。12. 学习路线与实战建议如果你计划在七天左右从入门到能写项目可以参考下面这个节奏但核心是“先跑通再深入”第 1 天安装环境跑通最小图理解 State、Node、Edge 三个概念。第 2 天写条件路由掌握add_conditional_edges的用法。第 3 天写带循环的 Agent理解recursion_limit。第 4 天给图加 Checkpointer实现多轮对话记忆。第 5 天把 LangChain 的模型和工具集成到节点里。第 6 天用 FastAPI 包装接口跑通批量任务。第 7 天选一个真实小项目比如“PDF 问答 Agent”把前面所有能力串起来。工程实践上有几条建议第一版图不要设计得太复杂先用最少的节点跑通主流程再逐步加分支。所有节点函数保持单一职责一个节点只做一件事。模型相关参数模型名、温度、超时时间放到配置里不要写死在节点代码中。批量任务必须加日志和错误隔离否则上游一个 API 抖动整批任务都会失败。接口服务部署到公网前务必加鉴权。不要在公网裸奔一个能让你花钱调模型的接口。涉及用户隐私数据、版权素材时先确认授权再上线。13. 总结LangGraph 最值得尝试的点是它把复杂 Agent 应用的流程控制从手写 if-else 中解放出来让状态、分支、循环、记忆都变成图结构的一部分。而且它和 LangChain 天然互补团队如果已经在用 LangChain迁移成本并不高。拿到一个 LangGraph 项目最先应该验证三件事最小图能不能跑通、条件路由是否正确走到目标分支、跨轮记忆是否按 thread_id 隔离。这三件事验证通过框架的骨架就算掌握了。最容易踩的坑也有三个第一个是节点返回值覆盖了 State 里的旧字段导致信息丢失第二个是条件路由分支名写错GraphRecursionError反复出现第三个是生产环境用了内存检查点服务重启后对话记忆全部丢失。后续可以继续扩展的方向是把本地模型接入 LangGraph做多 Agent 协作系统以及结合 LangSmith 做更精细的链路分析。LangGraph 的生态还在快速迭代社区里已经有很多基于它搭建的 Agent 产品值得跟进。建议先跑通第 4 节的最小图再花半天时间把条件路由和记忆两个机制玩熟。这套东西学会之后写复杂 Agent 的维护成本会明显下降。