这次我们直接看 LangGraph。它不是一个概念 Demo而是 LangChain 团队开源的智能体编排框架用来把 Agent 从“模型调用一次就结束”的线性流程改造成“可以循环调用工具、可以分支、可以暂停等人工确认、还可以断点续跑”的图结构工作流。很多人学 Agent 开发卡在两点一是组件太多不知道先看哪个二是照抄代码后一跑就报错、不知道去哪查。这篇文章把智能体架构、核心组件、代码实战、接口批量任务一次性拆开讲所有示例都按最小可运行的标准给出来看完可以直接在本地跑通。先列几个最值得关注的点LangGraph 支持有状态多轮会话、支持条件分支和循环、支持人工审批打断点、支持执行状态持久化、支持流式输出并且它有很强的服务化能力可以包装成本地接口或接入批量任务队列。从算力角度看LangGraph 本身只是编排层跑纯逻辑演示 CPU 就够了真正吃显存的是底层 LLM 和向量检索部分这一点我们在后面资源占用章节单独说。这篇文章的实操路线是先讲架构和核心组件再给环境准备清单然后依次跑通“最小 StateGraph”“带工具调用的 ReAct Agent”“人工审批与流式输出”“API 与批量任务”。适合正在学 Agent 开发、想把 LLM 接进业务流程、或者被“结果不可控、多轮对话断上下文、没法人工干预”这类问题卡住的人。下面直接进入正文。1. LangGraph 智能体核心能力速览能力项说明框架类型开源 LLM 应用编排框架基于图结构建模智能体开源来源LangChain 团队维护核心建模方式把 Agent 流程表达为 State状态 Node节点 Edge边有状态能力多轮对话、会话隔离、断点续跑支持按 thread_id 隔离会话循环与分支原生支持条件边Agent 可以反复调用工具直到任务完成人机协作支持 interrupt 打断点敏感步骤等待人工审批后继续持久化Checkpointer 保存执行状态可恢复会话MemorySaver、文件、数据库均有社区方案流式输出支持按节点更新流式输出也支持 Token 级流式输出服务化能力可自行封装 FastAPI/Flask 接口或使用官方 LangGraph Server 方式提供 REST API支持语言Python 生态最全也有 LangGraph.js 的对应实现硬件需求编排层本身很轻纯 CPU 可跑演示实际显存占用取决于底层 LLM 是本地模型还是 API 调用批量任务通过 thread_id 隔离不同会话可以循环分发批量任务也可以接入消息队列适用场景客服 Agent、工具调用、RPA 审批流、文档问答、工作流编排、企业内部自动化补充一点LangGraph 不是替代 LangChain 的存在而是继承了 LangChain 的组件生态比如 Tool、Message、模型接口等同时把“执行流程”的控制权从框架手里拿回用户手里。你完全可以只学 LangGraph 不碰 LangChain 的高阶封装但通常还是建议把基础概念过一遍。2. 智能体架构拆解从 ReAct 到图编排早期用 LangChain 写 Agent很多方案落在 AgentExecutor 上。它的逻辑可以理解为“模型决定下一步 → 执行工具 → 把结果放回消息列表 → 再次交给模型”这个循环在内部执行用户能干预的点非常有限。遇到复杂业务比如步骤多、需要人工审批、需要不同分支策略AgentExecutor 的线性封装就开始别扭。LangGraph 的思路是把 Agent 画成一张图。每个“操作”是一个节点比如“模型推理”“调用工具”“输出最终回答”节点之间用边连接普通边表示必然跳转条件边表示根据状态决定下一步走向哪里。这样设计之后三个最大优势是对比维度传统线性 AgentExecutorLangGraph 图编排流程表达隐式循环难以控制显式建图节点与边一目了然分支能力弱靠模型硬猜条件边显式路由人工干预困难interrupt 打断点审批后继续状态持久化会话内有效重启丢失Checkpointer 可保存恢复调试方式看日志靠猜图结构可视化 流式事件输出LangGraph 智能体的核心组件可以浓缩为一张表组件作用关键理解State全局共享状态所有节点读写用 TypedDict 或 Pydantic 定义字段Node一个执行单元Python 函数签名是 (state) - dictEdge节点之间的确定性跳转无条件执行完就跳到下一个节点Conditional Edge根据返回值或状态选择跳转目标函数返回节点名或 ENDReducer定义状态的合并规则用 Annotated 声明解决字段覆盖问题Checkpointer会话状态持久化支持多轮问答断点恢复ToolNode工具的统一执行入口所有工具挂在一个节点里执行interrupt打断图执行等用户输入实现人工确认、信息补全这套组件说起来抽象但实际写代码只有几个动作定义状态、写节点函数、画边、编译图、调用。接下来我们把环境准备好然后开始跑最小示例。3. 环境准备与依赖安装LangGraph 是一个 Python 库安装不复杂但建议不要直接装在全局 Python 环境里避免和项目依赖冲突。推荐用 venv 或 conda 隔离环境。先从基础环境开始Python 版本建议 Python 3.10 以上具体以你使用的 LangGraph 版本说明为准。操作系统Windows / macOS / Linux 都可以Linux 部署最省心。基础依赖langgraph、langchain-core、langchain-community可选、transformers 或 ollama 等模型接入方式。安装命令示例python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install -U langgraph langchain-core如果后续要用到工具调用和人机协作再装可选组件pip install langchain-community langchain-openai # 如果希望打印图结构用于调试可额外安装 pip install grandalf模型接入这一块需要单独说明LangGraph 本身不带模型。你可以接 OpenAI、Ollama、HuggingFace Transformers、vLLM 等多种后端。在纯本地演示阶段最省钱的做法是用一个固定逻辑的“假 LLM”代替真实模型先把图和节点跑通。等确认流程没有问题了再接真实模型。这样安排的好处是不用先付 API 费用也不用担心本地显存不够。后面实战示例我会用注释里的方式说明如何替换成真实模型。4. 最小可运行示例先跑通一个 StateGraph先不搞复杂 Agent我们从最小状态图开始。目的是理解三个概念State 怎么定义、Node 怎么执行、Edge 怎么跳转。from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END def merge_logs(left: list, right: list) - list: 将多条日志合并追加避免新返回覆盖旧状态。 return left right class State(TypedDict): counter: int log: Annotated[list, merge_logs] def node_a(state: State) - dict: current state.get(counter, 0) 1 return {counter: current, log: [fnode_a 执行当前计数: {current}]} def node_b(state: State) - dict: current state.get(counter, 0) 10 return {counter: current, log: [fnode_b 执行当前计数: {current}]} graph StateGraph(State) graph.add_node(a, node_a) graph.add_node(b, node_b) graph.set_entry_point(a) graph.add_edge(a, b) graph.add_edge(b, END) app graph.compile() result app.invoke({counter: 0, log: []}) print(result)这段代码做了什么定义了一个State它有两个字段counter和log。log字段用Annotated指定了 reducer 为merge_logs这样每一次节点返回的日志不会覆盖之前的值而是追加。图从a节点开始执行完后跳到b再从b结束。compile()将图编译成可调用对象invoke()传入初始状态执行。执行后预期输出类似{counter: 11, log: [node_a 执行当前计数: 1, node_b 执行当前计数: 11]}这里最需要理解的是 reducer 的作用。如果不加Annotated[list, merge_logs]节点b返回的log会直接覆盖node_a写入的日志最后你只能看到node_b的那一条。这个规则对 Agent 的消息列表尤其重要因为每一轮模型输出和工具结果都要追加到历史里而不是互相覆盖。跑通最小示例后可以试一个简单改动把graph.add_edge(a, b)换成条件边让node_a自己决定下一步是去b还是直接结束。def choose_next(state: State) - str: if state[counter] 5: return b return END graph.add_conditional_edges(a, choose_next, {b: b, END: END})条件边是 LangGraph 智能体最重要的能力之一因为“是否继续调用工具”本质上就是一个条件判断。下一节我们进入实战把条件边和工具调用结合起来。5. 实战构建带工具调用的 ReAct Agent这一节的示例会模拟一个真正 Agent 的思考循环输入用户问题 → 判断是否需要调用工具 → 调用工具 → 把工具结果放回消息列表 → 再次判断是否结束。我们用一个简单的天气查询工具做演示。为了让你跑起来不依赖任何外部 API示例里先用固定逻辑代替模型决策代码注释里也给了替换成真实模型的方式。from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage from langchain_core.tools import tool from langgraph.prebuilt import ToolNode tool def get_weather(city: str) - str: 查询指定城市当前的天气情况。 return f{city} 当前 24℃多云有微风。 class AgentState(TypedDict): messages: Annotated[list, add_messages] def agent_node(state: AgentState) - dict: # 演示环境不接外部模型用固定规则模拟模型决策。 # 真实开发中替换为类似 # model chat_ollama(...) 或 model ChatOpenAI(...) # message model.bind_tools(tools).invoke(state[messages]) # return {messages: [message]} last_content state[messages][-1].content if 天气 in last_content: # 模拟模型返回一个 tool_call要求执行 get_weather 工具 return { messages: [ AIMessage( content, tool_calls[ { id: call_demo_001, name: get_weather, args: {city: 上海}, } ], ) ] } return {messages: [AIMessage(content我没有理解你的意图当前演示只支持天气查询。)]} def should_continue(state: AgentState) - str: last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return END # 初始化图 tools [get_weather] tool_node ToolNode(tools) graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, tool_node) graph.set_entry_point(agent) graph.add_edge(tools, agent) graph.add_conditional_edges(agent, should_continue, {tools: tools, END: END}) app graph.compile() result app.invoke( {messages: [HumanMessage(content上海天气怎么样)]}, ) for message in result[messages]: print(message)运行后你会看到完整的事件链用户消息 → 带tool_calls的 AIMessage → 工具执行结果 ToolMessage → 再次回到 agent 节点。这里的关键点add_messages是 LangGraph 内置的消息合并 reducer专门用于给消息列表追加新消息。ToolNode会自动读取 AIMessage 中的tool_calls字段执行对应工具并把结果包装成 ToolMessage 写回状态。should_continue通过判断最后一条消息是否带tool_calls来决定是否继续调用工具。如果模型还在要工具图就继续循环直到模型返回纯文本答案。如果把示例中的问题改成“你好”流程会走另一个分支should_continue发现没有工具调用图直接结束。这就是条件边的作用一个图里可以同时存在“继续循环”和“结束”两条路径。替换成真实模型时只需要把agent_node中间的模拟逻辑替换为from langchain_ollama import ChatOllama # 以本地 Ollama 为例 model ChatOllama(modelqwen3:8b, temperature0) model_with_tools model.bind_tools(tools) def agent_node(state: AgentState) - dict: response model_with_tools.invoke(state[messages]) return {messages: [response]}这样你就得到了一个能真正调用本地模型的 LangGraph Agent。Ollama 等本地模型服务只要按各自方式启动即可调用接口不占用 LangGraph 本身的资源。6. 人工审批、流式输出与持久化跑通基础 Agent 后下一步解决企业场景里最常见的三个问题怎么在敏感操作前插入人审、怎么实现打字机式输出、怎么保存多轮会话状态。6.1 人工审批节点早期 Agent 最大的风险是“不可控”模型可能直接执行危险工具。LangGraph 的interrupt机制可以把流程停在某个节点等服务端拿到人工输入后才继续。from langgraph.types import interrupt from langgraph.checkpoint.memory import MemorySaver def approval_node(state: AgentState) - dict: user_decision interrupt( { type: approval, content: 是否批准执行工具调用请回复 yes 或 no。, } ) if user_decision.get(approve) is True: return {messages: [HumanMessage(content审批通过继续执行。)]} return {messages: [HumanMessage(content审批拒绝流程终止。)]} # 在图中加入审批节点并把它放在工具调用之前 graph.add_node(approval, approval_node) graph.add_edge(approval, tools) graph.add_edge(agent, approval) # 实际按你的流程调整边的连接注意interrupt的导入路径在不同版本中可能有差异旧版本可能从langgraph.constants导入。如果你照代码写报错 ModuleNotFoundError先检查当前安装的 langgraph 版本再按对应的包路径调整导入。6.2 检查点持久化多轮对话要求每次请求都能读写同一个会话上下文LangGraph 的compile(checkpointer...)就是干这个的。最简单的内存版检查点适合熟悉机制checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: user-001}} # 第一轮对话 app.invoke({messages: [HumanMessage(content上海天气怎么样)]}, config) # 第二轮对话同一 thread_id 下状态延续 app.invoke({messages: [HumanMessage(content那明天呢)]}, config)同一个thread_id下的消息历史会一直保留不同thread_id的会话彼此隔离。这也是 LangGraph 实现批量任务的基础。生产环境建议把检查点换成数据库或文件存储方案避免进程重启后数据丢失MemorySaver 适合验证不适合正式用。6.3 流式输出Llama、ChatGPT 式的打字机输出在 LangGraph 里也支持。最常用的方式是按节点更新来流式读取inputs {messages: [HumanMessage(content上海天气怎么样)]} for event in app.stream(inputs, config, stream_modeupdates): for node_name, node_output in event.items(): print(node_name, node_output)stream_modeupdates会在每个节点执行完后返回该节点的输出。如果你接的是真实 LLM 接口还有更细粒度的 token 流式模式可以在推理阶段拿到 token 逐步输出。流式输出的价值不仅仅是体验好更重要的是长任务里用户可以尽早看到进度而不是干等着接口超时。7. LangGraph 接口 API 与批量任务LangGraph 跑通之后要接到真实业务里大多数情况下需要包一层接口服务。 LangGraph 本身没有规定你是用 FastAPI、Flask 还是官方 Server社区常见的两种思路都可用。下面给一个用 FastAPI 包装的最小示例最简单也最容易控制权限和扩展。from fastapi import FastAPI from pydantic import BaseModel from langchain_core.messages import HumanMessage server FastAPI() class ChatRequest(BaseModel): thread_id: str message: str class ChatResponse(BaseModel): answer: str server.post(/chat, response_modelChatResponse) def chat(req: ChatRequest): config {configurable: {thread_id: req.thread_id}} result app.invoke({messages: [HumanMessage(contentreq.message)]}, config) return ChatResponse(answerresult[messages][-1].content)启动接口服务uvicorn main:server --host 0.0.0.0 --port 8000启动后访问http://127.0.0.1:8000/docs可以直接调试接口。用 curl 测试请求curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {thread_id: user-001, message: 上海天气怎么样}返回结果类似{answer:上海 当前 24℃多云有微风。}接口跑通之后批量任务的思路就简单了每个任务分配一个独立thread_id循环调用同一个接口收集返回结果。生产级别要注意三件事并发限流、失败重试、日志记录。import time questions [ 北京天气怎么样, 广州天气怎么样, 深圳天气怎么样, ] def batch_run(questions: list[str]) - list[dict]: results [] for idx, question in enumerate(questions): try: config {configurable: {thread_id: fbatch-{idx}}} result app.invoke({messages: [HumanMessage(contentquestion)]}, config) results.append({question: question, answer: result[messages][-1].content}) except Exception as exc: # 单条失败不影响整批任务 results.append({question: question, error: str(exc)}) time.sleep(0.1) return results批量任务的工程要点每个线程一个thread_id不要复用否则上下文串话。重试策略放在外层循环超过 2 次失败的单独记录避免死循环。用队列服务承接大批量任务比 for 循环更稳比如 Redis RQ、Celery、Arq 都可以。每次调用要记录输入输出的 token 数和耗时方便成本核算。如果你想用官方服务化方式跑可以按官方文档安装langgraph-cli并把graph.py放在指定目录启动本地开发服务器它会自动生成 REST API 和可视化面板。不过不同版本配置差异较大首次上手更推荐上面这段 FastAPI 代码它完全可控且不依赖云服务。8. 资源占用、性能观察与问题排查8.1 资源占用怎么观察LangGraph 编排层本身占用的内存和 CPU 很少。真正影响资源的是底层模型和检索组件资源项影响来源观察方式显存本地模型推理、上下文长度nvidia-smi、Ollama ps 或模型服务自带状态内存长对话消息列表、Checkpointer 缓存系统任务管理器、Python memory_profilerCPU节点函数计算量、工具执行、文档解析观察宿主机 CPU 使用率磁盘模型文件、检查点快照、日志统计模型目录和 checkpoint 目录大小本地跑模型时显存占用不是固定数字它会随上下文长度增加而上涨。输入越长、并发越多显存和延迟越高。如果你的业务是几十个会话并行建议认真做上下文裁剪不要把所有历史永远堆在 messages 里。调用云端 API 模型时客户端基本无显存问题但每个请求的 token 和成本都要记录。编排层可以做的优化是控制单轮工具调用次数、裁剪历史消息、避免每次 reconstruct 整个图。8.2 性能观察方法给节点计时是一个简单而有效的做法import time start time.perf_counter() result app.invoke(inputs, config) print(total time:, time.perf_counter() - start)想要更细就在每个节点函数里打点。配合app.stream可以观察每一步的间隔找出卡在模型调用还是卡在工具执行。8.3 常见问题排查表问题现象可能原因排查方式解决方案节点没有执行没有设置入口点或边没连上打印图结构检查 set_entry_point设置正确入口补全边返回的状态字段报错节点返回的 key 不在 State 定义中看异常信息里的 key扩展 State 定义或调整 return条件边没有命中预期节点should_continue 返回值不是注册的路由 key打印 should_continue 返回值检查路由表确认 END 写法Agent 循环停不下来模型反复生成 tool_calls结束分支没触发打印每一步 message设置最大轮数完善条件边或添加 recursion_limit工具没有被执行AIMessage 里没有 tool_calls打印 agent 输出的最后一条消息检查 bind_tools 是否正确模型版本是否支持函数调用interrupt 报错或导入失败版本差异导致导入路径不同查看 langgraph 版本按当前版本文档调整 import多轮对话上下文丢失调用时没有传 config或 thread_id 每次变化检查 config 和 checkpointer 是否传入 compile固定 thread_id使用同一 checkpointer接口并发请求卡死同步阻塞 排队无超时观察服务日志加异步队列、连接池超时、并发数限制批量任务跑到一半失败单一 task 异常中断了循环查看异常堆栈单条失败捕获重试策略放外层这里额外提一个高频问题用MemorySaver跑多轮测试时第一次正常、第二次重启后上下文没了。这不是 bug而是内存版检查点重启即清空。正式环境换用持久化后端。9. 最佳实践与合规边界9.1 工程实践建议先假后真。先用固定逻辑跑通图结构再接入真实 LLM能节约大量调试时间。State 字段尽量少。只放节点真正需要的数据削掉中间缓存字段否则每个节点都要处理无关状态。工具权限最小化。不要一股脑把“文件删除”“数据库执行”“发邮件”等高风险工具开放给模型先只挂必要工具。敏感步骤加 interrupt。涉及转账、删除、外发数据、权限变更的节点必须人工确认。所有工具函数加清晰 docstring模型通过 docstring 理解工具用途写清楚参数含义能明显提升工具调用准确率。上下文要裁剪。长期会话会给 prompt 塞入大量历史要按 token 阈值合并或摘要避免成本爆炸。每次调用记录 thread_id、模型名、token 数、耗时。出问题的时候才能快速定位。9.2 合规与安全边界智能体框架把模型从“回答问题”扩展成“操作业务系统”这带来明显的能力提升也带来责任。开发时建议注意以下几点输入用户数据和 prompt 内容涉及个人信息时先评估数据出境和存储要求不能直接把敏感信息抛给第三方模型服务。模型生成结果必须人工复核后再进入正式业务环节尤其是通知类、审批类、对外发布类场景。工具调用要有白名单、权限校验、操作日志。模型只能调用当前会话授权范围内的工具。涉及人脸、声音、版权素材等场景必须先确认授权涉及自动化操作外部系统要确认使用边界避免绕过平台限制或产生误操作。开源组件要保留许可证信息第三方库升级前先看 breaking change避免生产环境静默升级。10. 总结与下一步LangGraph 最值得尝试的点是它让 Agent 的流程真正变得可见、可控、可恢复。你不再需要在一个黑盒循环里赌模型能不能完成整个任务而是可以把“意图判断”“工具执行”“人工审批”“最终回复”拆成一条条带名字的路径。状态由你定义分支由你画复杂流程也能拆成一个一个小节点来调试。拿到这篇文章后建议按顺序做三件事先跑通第四节中最小的 StateGraph理解状态合并再把第五节中的 ReAct Agent 粘到本地把问句从“上海天气”换成你自己的场景看条件边和工具节点如何联动最后用第七节的 FastAPI 包装一层接口接一个批量任务测试稳定性。最容易踩的坑不是语法而是条件边路由写错、状态字段没有定义、interrupt 导入路径对不上版本遇到报错先去看路由表和 state schema。想继续深入下一步可以尝试给 Agent 挂上知识库检索节点、接入长时记忆存储、或者把同一个图部署到多 worker 上做水平扩展。