这次直接看 LangGraph。如果你最近在关注 AI Agent 开发这个名字大概率刷到过很多次。LangGraph 不是替代 LangChain而是 LangChain 生态里专门负责 Agent 编排的状态图框架。简单理解把一个多步骤、有循环、有条件分支的 AI 流程用一张“图”来表示每一步是一个节点节点之间的跳转是边数据和上下文全部通过 State 对象传递。为什么需要它很多人写 Agent 的开头是一个函数里反复调大模型遇到分支用 if-else需要重试就 while 循环。两三个工具还好一旦工具调多了、状态要回退、要跨会话记忆、要多个分支并行处理这种写法会很难维护。LangGraph 解决的就是这个问题它把控制流变成显式的图循环和分支可以由开发者自己控制并且支持持久化记忆、并行分支、子图嵌套、API 服务部署。这篇文章直接给出从零开始的完整路径安装环境、画第一张图、条件路由、循环 Agent、并行分支、子图、记忆恢复、API 部署和批量调用。所有代码都可以直接复制到本地跑。文章的结构是从简单到复杂前两节看明白 LangGraph 能干什么第三节开始上手写代码。1. LangGraph 核心能力速览能力项说明项目类型有状态图执行引擎 / Agent 编排框架核心功能状态管理、条件路由、循环控制、并行分支、子图、持久化记忆与 LangChain 关系LangChain 生态中的官方 Agent 编排库也可独立使用支持语言Python 为主官方也有 JS/TS 版本运行环境CPU 即可运行框架本身LLM 调用可走云端 API 或本地推理引擎启动方式Python SDK 直接调用 / langgraph CLI 启动开发服务接口能力通过 LangGraph Server 暴露 HTTP API支持同步调用、流式调用、历史会话批量任务可通过代码循环、异步并发、Server 状态化调用实现适合场景多步骤 Agent、工具调用循环、客服机器人、知识库问答、工作流编排这个表里最值得注意的一点是LangGraph 本身不是一个“模型”不参与大模型推理。它的资源开销主要在 Python 运行时、框架依赖和 State 数据管理上。真正的显存和内存消耗取决于你用的是云端模型还是本地模型。后面第 8 节会专门讲资源占用怎么看。2. LangGraph 与 LangChain、Agent 的关系这三个概念经常被放在一起讲但实际分工完全不同。2.1 LangChain 是什么LangChain 是面向大模型应用开发的框架提供了模型封装、Prompt 模板、工具调用、RAG 组件、记忆管理等基础能力。你可以用它快速做出一版“调用大模型 调用工具”的应用。2.2 传统 Agent 的问题LangChain 早期的 Agent 依赖AgentExecutor它在内部实现了 ReAct 循环模型思考、调用工具、观察结果、再思考。问题是这个循环是封装在内部的开发者很难干预“什么时候停”“哪个分支先执行”“中间状态如何恢复”。流程简单时还好复杂一点就没法精细控制。2.3 LangGraph 的定位LangGraph 把控制流显式拆成了节点和边。你可以清楚看到每一步执行了什么哪个节点返回了什么状态根据状态决定走哪条边循环什么时候结束状态如何持久化所以更准确的说法是LangChain 提供大模型应用的各类组件LangGraph 负责把这些组件按照可控的流程图编排起来。两者不是竞争关系LangGraph 是 LangChain 生态里目前最推荐的 Agent 编排层。2.4 新项目怎么选如果你的流程是线性的取输入 - 调模型 - 返回结果那用 LangChain 的 Chain 就够了没必要引入 LangGraph。但如果流程里有循环、分支、并行、多轮记忆、需要人工干预直接上 LangGraph。从当前 Agent 开发趋势看LangGraph 已经成为团队协作时比较普适的选择因为流程图本身就是设计文档团队成员看代码就能理解整体逻辑。3. 适用场景与使用边界3.1 适合什么场景场景说明客服系统多轮对话 查询订单 退换货流程状态需要跨会话保持知识库问答RAG 检索后需要判断是否补充问题、是否需要重新检索多工具 AgentAgent 在循环中自行决定调用哪个工具、调用几次内容审核流程初筛、人工复核、申诉多个环节按条件流转数据分析任务先选数再写查询结果不满足条件再修正3.2 不适合什么场景单次 Prompt 调用杀鸡用牛刀直接调模型接口更快。极低延迟的在线接口图编排会引入额外的状态管理开销不适合纯透传场景。复杂的数值计算或矩阵运算这类工作交给数据处理框架不是 LangGraph 的强项。3.3 使用边界LangGraph 只负责流程编排不负责模型输出质量。大模型可能产生幻觉工具调用可能返回脏数据条件路由可能因为 LLM 返回了预期之外的字符串而走错分支。因此所有面向用户的结果建议做一层校验或人工复核。会话记录、用户输入、工具返回内容属于敏感数据日志里不要记录完整内容。如果涉及人脸、声音、版权素材必须确认授权和遵守平台服务条款。批量任务要注意调用频率限制不要无控制地并发请求模型服务。4. 环境准备与前置条件4.1 基础环境项目建议操作系统Windows / macOS / Linux 均可Python3.9 及以上推荐 3.10 或 3.11虚拟环境建议使用 venv 或 conda网络能访问模型 API或本机已有推理服务磁盘空间框架占空间很小预留 1GB 完全足够本地模型另算4.2 需要哪个大模型LangGraph 本身不包含大模型。你可以选择OpenAI 兼容 APIgpt-4o-mini这类在线模型需要 API Key。本地模型使用 Ollama、vLLM、llama.cpp 启动一个 OpenAI 兼容服务本机调用。如果打算先跑通流程最省事的方式是用一个带 OpenAI 兼容接口的本地模型或者直接在环境变量里配置在线模型的 Key。整个过程只验证流程编排不要求模型有多强。4.3 安装依赖建议先创建虚拟环境python -m venv .venv source .venv/bin/activate # Windows 下用 .venv\Scripts\activate安装核心依赖pip install langgraph langchain-core langchain-openai如果后面需要启动本地 API 服务再确认命令行工具。部分版本把 CLI 放在langgraph包里部分版本需要单独装pip install langgraph-cli安装后检查python -c from langgraph.graph import StateGraph; print(ok)能看到ok说明环境基本可用。不同版本的 API 细节会有一点差异如果运行时提示某个符号找不到优先去 LangGraph 官方文档对应版本查一下。5. 安装部署与启动方式5.1 方式一Python SDK 直接调用这是最常见的用法。在代码里定义图编译后直接invoke。不需要启动任何额外服务适合写测试脚本和批量任务。5.2 方式二langgraph CLI 启动开发服务如果你希望可视化查看图结构、观察状态变化或者把图暴露成 API 给其他系统调用可以用 CLI 启动开发服务。项目根目录需要有一个langgraph.json文件。典型的配置文件结构如下字段以实际版本为准{ dependencies: [.], graphs: { agent: ./agent_graph.py:app }, env: .env }其中./agent_graph.py:app指的是agent_graph.py文件里导出的编译后的图对象。然后执行langgraph dev启动成功后终端会输出一个本地访问地址。打开/docs或/ui可以看到接口文档和可视化界面。这种方式适合调试也适合做团队内部工具。5.3 方式三Docker / 服务化部署LangGraph 的服务端通常内置在平台能力中本地方案一般通过 LangGraph CLI 或自建 API 包装层。如果团队内部需要稳定部署可以用 FastAPI 把编译后的图包一层 HTTP 服务这是最通用也最可控的做法。后面第 7 节会给出 API 调用示例。6. 功能测试与效果验证这一节是全文重点。下面每个小节都是独立可运行的例子建议按顺序执行先在本地跑通再改造成自己的业务。6.1 最小两节点图先建立一个最小流程节点 A 处理输入节点 B 继续处理最后输出结果。from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): message: str # 所有节点共享的状态 def node_a(state: State) - dict: return {message: state[message] | A} def node_b(state: State) - dict: return {message: state[message] | B} # 1. 创建状态图 graph StateGraph(State) # 2. 添加节点 graph.add_node(A, node_a) graph.add_node(B, node_b) # 3. 添加边START - A - B - END graph.add_edge(START, A) graph.add_edge(A, B) graph.add_edge(B, END) # 4. 编译并执行 app graph.compile() result app.invoke({message: start}) # 5. 查看结果 print(result)预期输出{message: start | A | B}这个例子解释了 LangGraph 的核心机制所有节点共享同一个 State 对象节点返回的 dict 会被合并到 State 中下一个节点可以读取到更新后的值。同名键默认是覆盖更新这个后面会经常用到。验证成功标准result[message]包含A和B两个节点的追加内容。6.2 条件路由 conditional_edges条件路由是 LangGraph 最常用的能力。节点 A 执行完后根据当前 State 的状态决定走节点 B 还是节点 C。from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): message: str def node_a(state: State) - dict: return {message: state[message] - A} def node_b(state: State) - dict: return {message: state[message] - B} def node_c(state: State) - dict: return {message: state[message] - C} def route_after_a(state: State) - str: # 根据消息内容决定走哪个节点 if b in state[message]: return B return C graph StateGraph(State) graph.add_node(A, node_a) graph.add_node(B, node_b) graph.add_node(C, node_c) graph.add_edge(START, A) # 关键条件边 graph.add_conditional_edges( A, route_after_a, { B: B, C: C, } ) graph.add_edge(B, END) graph.add_edge(C, END) app graph.compile() print(app.invoke({message: need b})) print(app.invoke({message: need c}))预期输出第一条结果进入 B第二条结果进入 C。这里最容易出错的是路由函数返回的字符串必须存在于第三参数的映射表里。如果返回B但映射表里没有B这个 key执行会直接报错。建议所有路由返回值都集中定义成常量不要散落在代码里。6.3 循环 Agent 与防死循环Agent 最常见形态就是循环大模型判断需要调用工具调用完工具再看结果直到满足结束条件。LangGraph 允许节点连回自身或前序节点形成循环。循环必须加最大次数保护否则模型输出异常时可能死循环。下面这个例子模拟一个最简单的循环 Agent。它调用模型如果模型回答包含“不知道”或“无法”就重新让模型回答一次直到回答合格或达到最大次数。from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI class AgentState(TypedDict): question: str answer: str step: int def llm_node(state: AgentState) - dict: # 实际使用时换成自己的模型配置 model ChatOpenAI(modelgpt-4o-mini, temperature0) response model.invoke(state[question] 如果不知道就回答不知道) return { answer: response.content, step: state[step] 1, } def should_repeat(state: AgentState) - str: # 最大循环保护最多执行 3 次 if state[step] 3: return end # 结果包含不明确表述继续循环 if 不知道 in state[answer] or 无法 in state[answer]: return agent return end graph StateGraph(AgentState) graph.add_node(llm, llm_node) graph.add_edge(START, llm) graph.add_conditional_edges( llm, should_repeat, { agent: llm, end: END, } ) app graph.compile() result app.invoke( { question: LangGraph 的核心概念有哪些, answer: , step: 0, } ) print(result)这个例子有两点需要注意节点返回{step: state[step] 1}因为 State 默认覆盖更新所以每轮循环 step 都会增加。should_repeat必须有两个出口一个是返回循环节点一个是返回END。缺少终止出口会让图无法结束。验证成功标准如果模型回答正常一次循环结束如果回答“不知道”会看到 step 依次增加直到达到 step3 强制结束。建议把这段代码复制本地跑一次观察循环次数和answer的变化。6.4 并行分支LangGraph 支持一个节点同时走向多个下游节点。比较典型的场景是一个结果同时做关键词提取、情感分类、摘要生成三个任务最后统一汇总。多个分支执行完成后再进入汇总节点。并行分支的数据合并依赖 State 的 reducer 机制。如果多个分支同时写一个字段就需要给字段配置合并函数。下面用日志字段演示追加合并from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END def append_log(existing: list, updates: list) - list: return existing updates class State(TypedDict): log: Annotated[list, append_log] def node_a(state: State) - dict: return {log: [A]} def node_b(state: State) - dict: return {log: [B]} def node_c(state: State) - dict: return {log: [C]} def node_d(state: State) - dict: return {log: [D]} graph StateGraph(State) graph.add_node(A, node_a) graph.add_node(B, node_b) graph.add_node(C, node_c) graph.add_node(D, node_d) graph.add_edge(START, A) # A 执行完后B 和 C 并行执行 graph.add_edge(A, B) graph.add_edge(A, C) # B 和 C 都完成后进入 D graph.add_edge(B, D) graph.add_edge(C, D) graph.add_edge(D, END) app graph.compile() result app.invoke({log: [start]}) print(result)这里最关键的就是Annotated[list, append_log]。LangGraph 会用append_log函数合并新旧值。如果不写 reducer并行分支的多个返回值互相覆盖最后只会保留最后一个。由于 B 和 C 是并行执行的输出顺序可能不同。可能的结果{log: [start, A, B, C, D]}也可能{log: [start, A, C, B, D]}这属于正常现象。业务逻辑不应该依赖并行分支的执行顺序。6.5 子图 Subgraph当业务流程变大时可以把一组节点封装成一个独立的子图再作为父图的一个节点使用。子图适合做模块化拆分。from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): log: list def child_node(state: State) - dict: return {log: state[log] [child]} def parent_node(state: State) - dict: return {log: state[log] [parent]} # 子图 child_graph StateGraph(State) child_graph.add_node(child, child_node) child_graph.add_edge(START, child) child_graph.add_edge(child, END) child_app child_graph.compile() # 父图子图作为节点 parent_graph StateGraph(State) parent_graph.add_node(parent, parent_node) parent_graph.add_node(child_subgraph, child_app) parent_graph.add_edge(START, parent) parent_graph.add_edge(parent, child_subgraph) parent_graph.add_edge(child_subgraph, END) app parent_graph.compile() result app.invoke({log: [start]}) print(result)预期结果{log: [start, parent, child]}子图使用时有几个判断标准子图的 State 类型必须和父图兼容否则会在运行时出现字段缺失。子图的编译结果可以被多个父图复用适合做通用能力沉淀。如果不需要在外面展示内部节点结构用子图可以降低父图复杂度。6.6 记忆恢复 Checkpointer带记忆的 Agent 是 LangGraph 的强项之一。通过 checkpointer 保存每次执行后的状态再用同一个thread_id恢复上下文。下面这个例子演示了一个最小记忆会话。from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver def append_message(left: list, right: list) - list: return left right class State(TypedDict): messages: Annotated[list, append_message] def chat_node(state: State) - dict: # 真实项目中这里应该调用大模型生成回复 new_msg { role: assistant, content: 已经收到你的消息当前历史 str(len(state[messages])) 条, } return {messages: [new_msg]} graph StateGraph(State) graph.add_node(chat, chat_node) graph.add_edge(START, chat) graph.add_edge(chat, END) # 使用内存 checkpointer memory MemorySaver() app graph.compile(checkpointermemory) # 同一个 thread_id 会共享记忆 config {configurable: {thread_id: session-001}} result1 app.invoke( {messages: [{role: user, content: 你好}]}, config, ) print(result1) result2 app.invoke( {messages: [{role: user, content: 继续}]}, config, ) print(result2)第二次调用时新消息会通过append_message合并到历史消息中所以模型看到的上下文是包含第一轮对话的。这里的核心是thread_id是会话的隔离单位不同 thread_id 之间互不影响。checkpointer 支持内存、SQLite、Postgres 等后端。生产环境不要用MemorySaver进程重启后数据就丢了。每次invoke的输入会被作为对当前状态的更新配合 reducer 实现消息追加。验证成功标准第二次调用后messages列表长度比第一次多说明上下文没有被重置。7. 接口 API 与批量任务LangGraph 写完之后不只是能在本地跑。通过 CLI 启动服务后编译好的图可以暴露成 HTTP API方便前端、后端或定时任务调用。7.1 启动 API 服务在项目根目录准备langgraph.json指定图入口和依赖{ dependencies: [.], graphs: { agent: ./agent_graph.py:app }, env: .env }然后启动langgraph dev启动成功后会输出访问地址。实际端口以启动日志为准不要默认写死某一个端口。打开/docs能看到接口文档。7.2 用 curl 调用下面是一个通用的请求示例请把PORT换成实际端口把thread_id换成自己的会话 ID。curl -X POST http://127.0.0.1:PORT/api/invoke \ -H Content-Type: application/json \ -d { thread_id: test-001, input: { question: LangGraph 怎么部署成 API } }7.3 用 Python 调用import requests url http://127.0.0.1:PORT/api/invoke payload { thread_id: test-001, input: { question: LangGraph 支持批量任务吗 } } resp requests.post(url, jsonpayload, timeout60) print(resp.status_code) print(resp.json())实际请求参数可能因为 LangGraph 版本不同而略有区别建议先打开/docs界面看具体字段定义。只要是编译好的图通常都有同步调用和流式调用两种接口。7.4 批量任务设计LangGraph 本身不内置任务队列批量任务可以用最简单的方式实现循环调用。questions [ 什么是 LangGraph, LangGraph 和 LangChain 有什么区别, LangGraph 如何做循环控制, ] for idx, q in enumerate(questions): config {configurable: {thread_id: fbatch-{idx}}} result app.invoke({question: q, step: 0}, config) print(idx, result.get(answer))如果并发需求高可以用ThreadPoolExecutor但要注意模型 API 可能有速率限制并发过高会被限流。每个任务尽量用独立的thread_id避免状态串扰。加上超时和重试。单条任务失败不能影响整个批次。from concurrent.futures import ThreadPoolExecutor, as_completed def run_one(idx, q): config {configurable: {thread_id: fbatch-{idx}}} result app.invoke({question: q, step: 0}, config) return idx, result.get(answer) with ThreadPoolExecutor(max_workers4) as pool: futures [pool.submit(run_one, idx, q) for idx, q in enumerate(questions)] for future in as_completed(futures): idx, answer future.result() print(idx, answer)8. 资源占用与性能观察8.1 框架本身的资源占用LangGraph 是纯 Python 编排框架本身不消耗显存。CPU 和内存开销主要来自Python 解释器和依赖包State 数据的大小checkpoint 保存的状态快照并行分支时同时执行的节点数量正常使用下框架本身占用很小。真正的大头是底层大模型调用。8.2 云端模型的情况如果用的是 OpenAI 兼容在线 API本地资源占用可以忽略。响应时间取决于模型服务和网络延迟。在 LangGraph 里每次节点调用模型都是一次完整的网络请求所以节点设计得越少整体延迟越低。8.3 本地大模型的情况如果使用本地推理引擎建议单独观察显存。可以用下面的命令查看nvidia-smi --query-gpuname,memory.used,memory.total --formatcsv本地大模型的显存占用由推理引擎和模型大小决定和 LangGraph 没有直接关系。不同模型、不同量化方式、不同上下文长度占用差异很大。判断基线的方法先不启动 LangGraph单独调用模型服务记录显存基线再通过 LangGraph 跑任务对比显存增量。这样定位问题是 LangGraph 还是模型推理造成的。8.4 几个容易影响性能的点大模型的messages历史越长每次请求的 token 数量越大响应越慢。checkpoint 保存的数据量过大会拖慢写入速度。并行分支如果同时调用多个模型请求可能被模型服务限流。循环节点如果设计不合理可能产生大量重复调用。建议在 State 里维护 step 或者记录已调用过的工具列表。9. 常见问题与排查方法下面这些问题是从 LangGraph 开发中比较常见的坑整理出来的按现象、原因、排查方式、解决方案四列给出。问题现象可能原因排查方式解决方案import langgraph失败环境未安装或 Python 版本过低执行pip list查看包列表升级 Python 到 3.9重新安装依赖条件路由报 key 不存在路由函数返回值没有出现在映射表中打印路由函数返回值把路由返回值定义成常量映射表补齐State 中字段被新值覆盖忘记给并行合并字段配置 reducer检查节点返回值是否来自多个分支用Annotated[list, merge_func]定义合并逻辑循环执行次数超出预期循环终止条件不严格查看 State 中 step 字段在终止判断中强制校验最大次数子图状态缺失子图和父图 State 类型不兼容查看错误信息中的字段名统一子图与父图的状态定义调用模型报 401API Key 未配置或已失效检查环境变量正确配置模型服务 Keylanggraph dev 启动失败langgraph.json路径或入口配置错误查看终端完整报错确认 graph 入口路径和导出变量名正确API 调用返回超时模型响应慢或接口并发过高增加日志查看模型耗时加长超时时间降低并发数并行分支输出顺序不稳定并行节点执行顺序不确定观察日志中的执行时序不要依赖并行结果顺序后续汇总节点统一处理如果遇到报错信息不明显的情况最直接的办法是在节点函数里加日志def llm_node(state: AgentState) - dict: print(llm_node state:, state) ...把每一步的 State 打出来能快速定位数据在哪一层出了问题。日志只需在调试阶段打开正式运行建议关闭。10. 最佳实践与使用建议10.1 先画图再写代码LangGraph 的优势是流程可视化。动手前先在纸上画出节点、边、条件分支、循环出口再照着图写代码。流程越复杂这个习惯越值钱。10.2 State 设计尽量简单State 字段不要放一些只在单个节点内部使用的临时变量。字段越多reducer 和 checkpoint 的维护成本越高。建议只放跨节点共享的数据例如用户问题、历史消息、最终答案、步骤计数。10.3 循环必须有终止条件所有带循环的图都要在