LangGraph 不是又一个“套着 LangChain 壳”的编排工具而是真正把 Agent 流程当成一张图来跑的执行引擎。它适合处理条件路由、循环调用、并行分支、子图嵌套、长期记忆这类复杂状态流。本文不是概念搬运直接按“核心能力 - 环境搭建 - StateGraph 最小实例 - conditional_edge 深度解析 - 子图 - 状态修改 - 持久化 - API 封装 - 性能观察 - 排错”的顺序展开你可以照着跑通一整套 LangGraph 实战链路。1. LangGraph 核心能力速览在动手之前先把关键信息放在前面方便判断这个框架适不适合你的项目。能力项说明项目定位面向有状态、多参与者 LLM 应用编排框架核心是 StateGraph开源情况由 LangChain 团队开源Python 与 JS/TS 双语言支持核心模型StateGraph、Node、Edge、conditional_edge、State、Checkpointer主要功能条件路由、图结构循环、并行分支、子图嵌套、状态持久化、流式输出、人工介入断点与 LangChain 关系不是替代关系LangGraph 可独立使用也可直接调用 LangChain 生态组件硬件需求极低LangGraph 本身是纯 Python 编排层不执行模型推理CPU 即可运行显存占用取决于节点内运行的 LLMGraph 框架本身不占用显存显式循环控制图结构允许节点间构成有环依赖运行时通过 recursion_limit 控制最大执行步数批量任务支持可用 Send API 做 Map-Reduce 并行扇出也可在节点内接收批量输入接口 API可结合 FastAPI / LangGraph Server 快速暴露 HTTP 接口适合场景多步工具调用 Agent、客服多轮流程、RAG 后处理链路、数据清洗管线、人工审核流程一句话总结LangGraph 的入门门槛不在硬件而在理解“状态图”的执行思维。先跑通最小实例再逐步加入条件路由、子图和持久化。2. LangGraph 与 LangChain 的区别LangChain 和 LangGraph 经常被一起讨论但两者解决的问题层次不同。2.1 设计哲学差异LangChain 的核心抽象是 Chain也就是一个线性调用链输入 - Prompt - 模型 - 输出。虽然 LCELLangChain Expression Language可以让多个组件组合成管道但本质上仍是 DAG 的线性或者简单分支结构面对“模型反复调用工具直到完成任务”这类循环场景会比较别扭。LangGraph 的核心抽象是 StateGraph。它把应用执行过程建模成一张有向图State 是贯穿整张图的数据对象Node 是实际执行逻辑的函数Edge 决定数据从哪个节点流向哪个节点conditional_edge 根据运行时状态决定下一步走向因此天然支持循环和分支。LangGraph 还提供了 Checkpointer 机制可以把每一步的 state 持久化保存。这意味着一个 Agent 应用可以在多次调用之间保持上下文而不是每次调用都从零开始。2.2 选型建议什么时候用什么场景建议简单链式调用Prompt - LLM - 输出解析直接用 LangChain LCEL 更轻单轮带工具调用的 AgentLangChain AgentExecutor 可用但深入改造困难多步工具循环、条件分支、子任务并行选 LangGraph需要断点续跑、人工审核、多轮长期记忆选 LangGraph需要全量状态可视化、重新执行历史步骤选 LangGraph从团队成本考虑如果项目当前只有线性链路不要为了“新”而引入 LangGraph一旦出现循环、人工介入、多 Agent 协作LangGraph 的优势就会体现出来。3. 适用场景与使用边界LangGraph 适合这些项目客服 Agent多个工具节点、条件转人工、对话状态持久化自动化数据分析先判断用户意图再决定调用哪个分析工具结果不满足要求时循环重试报告生成流水线检索资料 - 生成大纲 - 并行生成多个段落 - 最终合并审校多 Agent 协作主 Agent 根据任务动态调度子 Agent业务流程自动化涉及审批、人工审核、异常回滚。使用边界也要明确LangGraph 不负责模型推理你需要自己接 OpenAI、Anthropic、Ollama 或任意云模型 API它不解决模型输出质量问题图结构只能帮助你控制流程不能提升单次生成质量对强数学调度、分布式计算场景Graph 框架本身不是最优解在涉及用户隐私、人脸、声音、版权素材的流程中必须确认数据来源和模型调用符合授权要求不要将未脱敏数据直接塞进外部模型。4. 环境准备与安装部署4.1 环境要求LangGraph 是纯 Python 库部署要求非常低。检查项推荐要求操作系统Windows 10 / macOS / Linux三者均可Python3.9 及以上建议 3.10 或 3.11包管理工具pip 或 poetry / uv本文用 pipGPU非必须仅当你在本地跑 LLM 才需要磁盘空间安装依赖约占用几百 MB不含模型权重端口如果启动 API 服务默认预留 8000 或其他自定义端口如果你的环境已经有 Python 和 pip直接进入安装。4.2 安装 LangGraph新建虚拟环境避免和系统环境冲突python -m venv .venv # Windows .venv\Scripts\activate # macOS / Linux source .venv/bin/activate安装核心依赖pip install -U langgraph langchain-openai如果后续需要持久化到 SQLite 或 PostgreSQLpip install langgraph-checkpoint-sqlite # 或者 pip install langgraph-checkpoint-postgres验证安装是否成功python -c import langgraph; print(langgraph.__version__)说明以上命令中的langgraph为 PyPI 包名不同版本 API 有细微差异。执行成功后就可以开始构建第一个图。4.3 推荐项目目录结构langgraph-demo/ ├── .venv/ ├── graphs/ │ ├── __init__.py │ ├── state.py │ ├── nodes.py │ ├── router.py │ └── build_graph.py ├── api/ │ └── main.py ├── tests/ │ └── test_graph.py ├── .env └── requirements.txt把 state、node、graph 构建逻辑分开后续维护会轻松很多。5. 第一个 LangGraph 应用从零构建状态图直接写代码。下面是一个最小 StateGraph包含两个节点先给消息列表追加一段文本再追加另一段文本。from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): messages: list def node_a(state: State) - dict: return {messages: state[messages] [A]} def node_b(state: State) - dict: return {messages: state[messages] [B]} # 1. 创建状态图 graph StateGraph(State) # 2. 注册节点 graph.add_node(a, node_a) graph.add_node(b, node_b) # 3. 连接边 graph.add_edge(START, a) graph.add_edge(a, b) graph.add_edge(b, END) # 4. 编译 app graph.compile() # 5. 运行 result app.invoke({messages: []}) print(result)运行结果{messages: [A, B]}这里有一个关键点节点函数返回的是一个字典LangGraph 会把返回的 key 与当前的 State 合并。如果节点函数返回None表示不修改任何状态。6. 条件路由与分支控制conditional_edge 深度解析这是 LangGraph 最核心的进阶能力。条件路由的作用是根据当前 state 动态决定下一步执行哪个节点。6.1 基础条件路由假设我们需要判断输入文本属于“紧急”还是“常规”然后走不同节点。from typing import TypedDict from langgraph.graph import StateGraph, START, END class QueryState(TypedDict): user_input: str level: str result: str def classify(state: QueryState) - dict: if 紧急 in state[user_input] or 加急 in state[user_input]: return {level: urgent} return {level: normal} def route_by_level(state: QueryState) - str: if state[level] urgent: return fast_handler return normal_handler def fast_handler(state: QueryState) - dict: return {result: 已进入快速处理通道} def normal_handler(state: QueryState) - dict: return {result: 已进入常规处理队列} graph StateGraph(QueryState) graph.add_node(classify, classify) graph.add_node(fast_handler, fast_handler) graph.add_node(normal_handler, normal_handler) graph.add_edge(START, classify) # 条件边classify 节点执行完根据 route_by_level 结果走向不同目标 graph.add_conditional_edges( classify, route_by_level, { fast_handler: fast_handler, normal_handler: normal_handler, }, ) graph.add_edge(fast_handler, END) graph.add_edge(normal_handler, END) app graph.compile() print(app.invoke({user_input: 这个问题很紧急请加急处理}))运行结果{user_input: 这个问题很紧急请加急处理, level: urgent, result: 已进入快速处理通道}add_conditional_edges的参数有三个第一个是源节点名第二个是路由函数接收当前 state返回一个字符串第三个是映射字典key 是路由函数返回值value 是目标节点名。也可以省略第三个参数让路由函数直接返回目标节点名graph.add_conditional_edges(classify, route_by_level)6.2 循环检测Agent 工具调用循环条件路由天然可以实现循环。下图逻辑是Agent 节点调用模型如果模型返回工具调用请求就进入 tools 节点否则直接结束。由于agent和tools之间存在环LangGraph 需要限制最大执行步数避免无限循环。import os from typing import Annotated, TypedDict from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] llm ChatOpenAI( modelgpt-4o-mini, api_keyos.getenv(OPENAI_API_KEY), ) def agent(state: AgentState) - dict: response llm.invoke(state[messages]) return {messages: [response]} def tools(state: AgentState) - dict: return {messages: [{role: tool, content: 模拟工具返回结果}]} def should_continue(state: AgentState) - str: last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return END graph StateGraph(AgentState) graph.add_node(agent, agent) graph.add_node(tools, tools) graph.add_edge(START, agent) # 关键条件边可能回到 agent形成环 graph.add_conditional_edges( agent, should_continue, { tools: tools, END: END, }, ) graph.add_edge(tools, agent) app graph.compile() result app.invoke( {messages: [{role: user, content: 调用一次工具并结束}]}, config{recursion_limit: 50}, )这里的recursion_limit就是循环运行的最大步数上限。若没有合理设置或者模型反复请求调用工具LangGraph 会抛出GraphRecursionError。生产环境建议把日志加上观察每个循环轮次的 tool_calls。6.3 并行分支与 Send API当多个子任务互相独立时可以用并行分支加快执行。LangGraph 的经典的 Map-Reduce 模式通过 Send API 实现动态扇出。from typing import Annotated, TypedDict from langgraph.constants import Send from langgraph.graph import StateGraph, START, END import operator class RepoState(TypedDict): topics: list sections: Annotated[list, operator.add] def plan_sections(state: RepoState) - dict: return {topics: state[topics]} def continue_to_write(state: RepoState) - list: # 为每个 topic 创建一个并行任务 return [Send(write_section, {topic: topic}) for topic in state[topics]] def write_section(state: dict) - dict: # 单篇生成放入 sections return {sections: [f### {state[topic]}\n内容草稿]} graph StateGraph(RepoState) graph.add_node(plan_sections, plan_sections) graph.add_node(write_section, write_section) graph.add_edge(START, plan_sections) # 从 plan_sections 扇出到多个 write_section graph.add_conditional_edges(plan_sections, continue_to_write, [write_section]) graph.add_edge(write_section, END) app graph.compile() result app.invoke({topics: [LangGraph, LangChain, RAG]}) print(result)Send的作用是动态生成多个“虚拟边”和“子任务启动”LangGraph 会尽量并行执行。如果你的流程是“先把一批文件切片再逐个 embedding”这种模式非常好用。7. 子图 Subgraph复杂流程模块化子图就是“图里的图”。当主流程包含多个可复用的业务子流程时可以把子流程封装成独立的 StateGraph在主节点中调用。from typing import TypedDict from langgraph.graph import StateGraph, START, END class SubState(TypedDict): sub_input: str sub_output: str def sub_node_1(state: SubState) - dict: return {sub_output: state[sub_input] processed} sub_graph StateGraph(SubState) sub_graph.add_node(sub_node_1, sub_node_1) sub_graph.add_edge(START, sub_node_1) sub_graph.add_edge(sub_node_1, END) sub_app sub_graph.compile() class ParentState(TypedDict): text: str final_output: str def parent_node(state: ParentState) - dict: # 直接调用子图 sub_result sub_app.invoke({sub_input: state[text]}) return {final_output: sub_result[sub_output]} parent_graph StateGraph(ParentState) parent_graph.add_node(parent_node, parent_node) parent_graph.add_edge(START, parent_node) parent_graph.add_edge(parent_node, END) parent_app parent_graph.compile() print(parent_app.invoke({text: hello}))运行结果{text: hello, final_output: hello processed}子图的好处是让业务逻辑边界清晰子图可以独立测试子图可以有自己的 Checkpointer 和配置父节点只需要关心子图的输入输出结构不关心内部细节。需要注意子图的输入通常需要从父节点 state 中取字段并把子图输出映射回父节点返回的字典。不要让子图直接修改父节点的 state而是通过返回值向上合并。8. 状态管理与长期记忆8.1 如何在节点函数改变 state 状态值很多初学者在“如何修改 state”上踩坑。LangGraph 的规则很简单每个节点函数接收当前 state函数的返回值变成一个 dict键值与 state 合并不返回的字段保持原样返回None表示不修改状态带 reducer 注解的字段使用 reducer 规则合并未注解字段直接覆盖。看一个字段覆盖与 reducer 的对比from typing import Annotated, TypedDict import operator class CounterState(TypedDict): count: int history: Annotated[list, operator.add] messages: Annotated[list, add_messages] def increment(state: CounterState) - dict: # count 没有 reducer返回的新值直接覆盖 return {count: state[count] 1} def add_history(state: CounterState) - dict: # history 有 operator.add reducer返回的列表会追加而不是覆盖 return {history: [step]}实际项目中最常用的 reducer 是add_messages。当节点返回{messages: [新消息]}时新消息会被追加进消息列表而不是覆盖整个列表。这个特性在构建多轮 Agent 时非常关键。8.2 Checkpointer 持久化LangGraph 的状态默认只存在于单次invoke调用中。如果想在多次调用之间保留状态就需要引入 Checkpointer。from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import StateGraph, START, END checkpointer InMemorySaver() # 编译时传入 checkpointer app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: user-001}} # 第一次调用 app.invoke({messages: [{role: user, content: 你好}]}, config) # 第二次调用同一 thread_id 下可以读取之前的状态 app.invoke({messages: [{role: user, content: 你能记住我吗}]}, config)thread_id就是“会话标识”。同一个thread_id的多次调用共享一份 Checkpoint 状态不同thread_id之间相互隔离。生产环境建议使用持久化后端例如 SQLiteimport sqlite3 from langgraph.checkpoint.sqlite import SqliteSaver conn sqlite3.connect(checkpoints.sqlite, check_same_threadFalse) checkpointer SqliteSaver(conn) app graph.compile(checkpointercheckpointer)8.3 长期记忆与 StoreCheckpointer 解决的是“对话轮次之间”的短期记忆。如果要跨用户长期保存用户偏好、永久知识LangGraph 在较新版本中提供了BaseStore接口。实践中也可以先使用 Redis、PostgreSQL 保存用户画像在节点中手动读取写入这样不依赖于特定框架 API更稳定。9. 将 LangGraph 封装为 API 服务LangGraph 是库不是服务。项目落地时通常要包一层 HTTP 接口。下面用 FastAPI 做一个最小封装。9.1 FastAPI 接口服务import os from typing import TypedDict from fastapi import FastAPI from pydantic import BaseModel from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END app FastAPI() class QueryState(TypedDict): user_input: str answer: str llm ChatOpenAI(api_keyos.getenv(OPENAI_API_KEY)) async def reply(state: QueryState) - dict: # 简化直接调用模型 resp await llm.ainvoke(state[user_input]) return {answer: resp.content} graph_builder StateGraph(QueryState) graph_builder.add_node(reply, reply) graph_builder.add_edge(START, reply) graph_builder.add_edge(reply, END) graph_app graph_builder.compile() class ChatRequest(BaseModel): message: str app.post(/chat) def chat(req: ChatRequest): result graph_app.invoke({user_input: req.message}) return {answer: result[answer]}启动服务uvicorn api:app --host 0.0.0.0 --port 8000注意FastAPI 的路由函数名如果叫app可能会和 LangGraph 实例名冲突建议区分命名。9.2 Python 客户端调用示例import requests url http://127.0.0.1:8000/chat payload {message: 帮我写一封周报邮件} response requests.post(url, jsonpayload, timeout60) print(response.json())9.3 批量任务与并发批量任务有两种常见做法客户端并发请求用ThreadPoolExecutor并发调用 HTTP 接口服务端批处理在节点内部接收一个批量输入列表循环或并行处理。用 Send API 做 Map-Reduce 更符合 LangGraph 的并行模式def continue_to_process(state: BatchState) - list: return [Send(process_item, {item: item}) for item in state[items]]生产环境建议给请求加超时和重试机制。比如用tenacity包装节点函数当模型 API 临时报错时自动重试 3 次。10. 资源占用与性能观察10.1 LangGraph 本身不消耗 GPU很多读者会问“LangGraph 需要多大显存”。答案很明确LangGraph 这个框架本身就是纯 Python 的图执行引擎它不运行 LLM所以几乎不消耗显存。显存占用完全取决于你在节点里接入了什么模型。如果使用云端 API本地只需要几百 MB 内存如果使用本地大模型显存需求根据模型大小而定这部分与框架本身无关。10.2 观察 LLM 调用耗时性能瓶颈通常在模型调用而不是图调度。可以在节点函数里加入耗时统计import time def llm_node(state: State) - dict: start time.perf_counter() response llm.invoke(state[user_input]) elapsed time.perf_counter() - start print(f[timing] llm_node: {elapsed:.3f}s) return {answer: response.content}对于复杂图可以用 LangGraph 自带的流式接口观察每个中间节点的输出for step in app.stream({user_input: test}, config): print(step)10.3 降低延迟与成本先做小参数测试减少模型调用次数、缩短输入上下文合理设置recursion_limit避免无意义循环多个模型调用能合并成一次调用的先合并可以并行执行的节点用 Send API 或异步节点不能并行的步骤不要强行拆节点对重复出现的中间结果考虑缓存或持久化。11. 常见问题与排查方法问题现象可能原因排查方式解决方案安装langgraph失败网络源不可用查看 pip 报错切换国内 pip 镜像源导入MemorySaver报错不同版本 API 不同检查langgraph.checkpoint模块新版用InMemorySaver节点函数修改的 state 不生效节点返回了None或键名拼写错误在节点内打印 state 再返回确认返回 dict 的键与 State 字段名一致字段被整组覆盖而不是追加该字段没有配置 reducer查看 State 定义使用Annotated[list, operator.add]或add_messages出现死循环GraphRecursionError条件路由始终返回同一个节点打印should_continue返回值提高recursion_limit更重要的是检查路由判断逻辑API 服务启动端口被占用8000 端口被其他进程占用netstat -ano查看端口更换端口--port 8001请求外部模型经常超时网络不稳定或模型响应慢在节点函数加timeout参数使用重试策略如tenacity多用户同时调用时状态串线没有使用不同thread_id检查配置对象每个用户请求设置唯一的thread_idCheckpointer 重启后状态丢失使用了内存型 Checkpointer检查存储介质换 SQLite 或 PostgreSQL CheckpointerLangGraph 可以和 LangChain 混用吗可以详见官方文档节点函数内直接调用 LangChain 模型、Retriever、Tool 是标准用法另外关于“LangGraph 是否有 Rust 版本”官方目前提供 Python 和 JS/TS 两个实现没有官方 Rust 版。如果项目需要 Rust 生态接入通常是内部封装 Python 子进程或者用 HTTP 调用 LangGraph 服务。关于“ECharts 与 LangGraph 能否实现节点可视化”LangGraph 官方支持把编译后的图导出为 Mermaid 格式可以用app.get_graph().draw_mermaid()获取 Mermaid 文本再放到支持 Mermaid 的文档工具中展示。ECharts 本身不支持 Mermaid 解析若一定要在 ECharts 中渲染需要自己把节点和边数据解析为 ECharts graph 类型的 JSON 结构。12. 最佳实践与工程化建议12.1 小参数先行逐步加复杂功能第一次跑 LangGraph先构建一个只有两个节点的线性图。确认状态读写和invoke调用没问题后再加入条件路由、循环和 Checkpointer。复杂图调试起来很痛苦不要一次性堆所有功能。12.2 保留一套最小可运行配置把最小图示例单独存成脚本例如quickstart.py。后续每次改动只在业务图目录里进行最小示例始终作为“环境是否正常”的检测基准。12.3 日志与可观测性生产环境建议在图中加入关键步骤日志每个节点进入、离开的时间每次模型调用的 token 数条件路由的实际走向每次invoke的thread_id和状态更新。这些数据对排查“为什么 Agent 不按预期行动”非常关键。12.4 模型文件、输入素材、输出结果分目录管理data/ ├── inputs/ ├── checkpoints/ ├── outputs/ └── logs/模型权重、缓存文件、临时文件不要混在一起避免磁盘占满后难以清理。12.5 接口服务要限制访问范围如果封装成 HTTP 服务关注以下几点服务绑定到127.0.0.1而不是0.0.0.0除非确实需要外部访问加接口鉴权至少是静态 Token限制单用户请求频率请求体大小和超时时间需要显式配置。12.6 合规与授权提醒LangGraph 本身只是编排框架但其中的数据流可能涉及敏感内容。使用用户数据、第三方内容、人脸、声音、版权素材时请确认数据来源合法、模型调用符合服务商条款、输出内容经过人工审核。商用场景尤其要对最终输出做复核不要盲目全自动化。13. 总结与学习路线LangGraph 最值得尝试的点是它的条件路由 循环 Checkpointer 组合。先用最小的 Agent 工具调用循环跑通效果再叠加并行子任务和长期记忆最后包成 FastAPI 服务。这里有一个最容易踩的坑不要按 LangChain 的“线性 Chain”思维去写 LangGraph遇到“我想在节点之间跳转”的需求直接想清楚 State 的状态流转然后把它拆成节点和条件边描述。后续可以继续扩展的方向多 Agent 协作图一个主图调度多个子图人工介入节点遇到审核需求时暂停图执行等待人工确认后继续流式输出把流式 token 传递给前端可视化监控把图执行状态导出到外部监控面板。如果是要从零学习 LangGraph建议按这条路线走先读官方文档中的概念部分再跑一遍本文的环境与最小示例然后自己改一个带条件路由的小 Demo最后用 FastAPI 包一个接口。不要把精力花在记 API 名上多用print(step)和调试小脚本观察状态流转比死记文档更有效。建议收藏备用后面做 Agent 项目时可以直接回来对照。