这次我们来看一个在 AI Agent 开发里讨论度非常高的开源框架LangGraph。它是 LangChain 团队推出的智能体编排框架解决的问题很具体——大模型单次问答做不了复杂任务但多步调用、工具选择、状态流转、人工审核、断点续跑这些事手写代码又容易写成一团乱麻。LangGraph 用图结构把整个流程组织起来节点干什么、边按什么条件走、中间状态存在哪全部显式定义代码可读性比传统 ReAct 手写循环高一个量级。这个项目的核心特点可以概括为五条第一图状态编排智能体的每一步用节点和边来表达第二内置状态管理与 checksum 持久化支持断点续跑第三支持人工介入流程可以在指定节点暂停确认后再继续第四支持流式输出与并发分支适合复杂工作流第五框架本身不依赖显卡CPU 也能跑显存占用完全取决于你接在哪一个大模型服务上。本文会从环境准备、安装部署、核心组件、两套代码实战、服务化封装到性能排查完整走一遍 Agent 开发流程。先说明门槛。LangGraph 本身是 Python 框架不直接推理大模型所以不需要高配显卡如果你接 OpenAI 或国内大模型 API只需要网络和 Key想完全本地化也可以接 Ollama 这类本地模型服务。本文演示用的设备是普通开发机重点观察的是代码能不能跑通、流程可读性如何、接口好不好接而不是显卡参数。所以写这篇教程时不会出现“4060 占用 7G”这类结论实际资源占用以你选择的大模型服务为准。适合的读者很明确已经会调 Prompt但想从“单轮对话”升级到“多步骤智能体”的开发者RAG 项目里流程已经复杂到不知道如何继续维护的人需要在团队里快速搭建一套可解释、可审计、可人工介入的 Agent 工作流的人。下面直接进入正题。1. LangGraph 核心能力速览先把最关键的参数和规格放在前面方便你快速判断这个框架适不适合你的项目。能力项说明项目类型开源智能体编排框架由 LangChain 团队维护主要功能Agent 构建、图工作流、状态管理、持久化、流式输出、人工介入、函数工具调用编程语言Python建议 3.9 及以上版本硬件要求框架本身 CPU 可运行大模型推理需求由所选模型服务决定显存占用不固定取决于接入的本地模型或在线 API支持平台macOS / Linux / WindowsWindows 上个别组件依赖需要额外处理启动方式Python 脚本编译图可配合 FastAPI / LangGraph Server 暴露为 HTTP 服务是否支持 API支持可以把编译后的图封装成接口服务前面再套业务逻辑是否支持批量任务支持同一张图可以接收多组输入也可在节点内做并发 map 处理适合场景AI Agent 开发、工作流搭建、工具调用编排、需要人审的长流程是否需要付费框架开源免费大模型 API 按服务商计费从材料看LangGraph 最大的竞争力不是某一个组件而是“把流程画成图让每一条路径都可解释”。构建出的图可以直接用invoke同步执行也可以astream异步流式输出这给了工程化集成的空间。2. 智能体架构与核心组件LangGraph 的架构本质上是用图论组织智能体的执行流程。一个智能体不再是一个黑盒 Prompt而是多个节点组成的拓扑结构。节点负责计算边负责跳转状态对象在节点之间传递。2.1 State整个流程的数据模型State 定义了这个工作流“每一刻都拿着什么数据”。你可以把它理解成一张放在所有节点中间的消息总线。LangGraph 允许多个节点往 State 里追加字段也支持用 reducer 控制合并逻辑。最常用的就是对话消息列表。LangGraph 提供的add_messagesreducer 会自动把新消息追加到已有消息列表中而不是覆盖掉这对 Agent 的上下文累积至关重要。State 设计直接影响后续所有节点的输入输出是整个图的“地基”。2.2 Node节点就是 Python 函数Node 是图的最小执行单元本质就是一个普通函数接收 State返回 State 的一部分更新。你不需要继承任何基类不需要写装饰器标记这降低了入门门槛。通常一个节点只做一件事比如“调用大模型生成回复”“解析工具结果”“检查文本质量”。做多件事的节点会让图不可读也不利于复用。这就是为什么 LangGraph 智能体需要拆节点而不是在函数里堆逻辑。2.3 Edge 与条件边决定下一步去哪边分为普通边和条件边。普通边表示“执行完 A 之后一定去 B”条件边表示“根据当前 State 的计算结果动态决定下一步”。条件边是 Agent 和固定流水线最大的区别模型决定是否需要调用工具工具调用结果决定是继续调用还是输出最终回答。条件边的返回结果会映射到不同节点名称。例如tools_condition返回tools图就进入工具执行节点返回END图就结束。2.4 Checkpointer 与持久化Checkpointer 是 LangGraph 支持断点续跑的关键组件。它把每一步的 State 快照保存下来每个会话用thread_id标识。当流程被人工暂停或中断时同一thread_id可以通过传入相同配置继续执行。默认常用MemorySaver把快照放在内存里适合本地开发和测试生产环境建议换成数据库存储的 checkpointer。这个机制让“人审 Agent 自动执行”的组合落地非常自然。核心组件对照表如下组件作用常见配置示例StateGraph构建状态图StateGraph(State)State全局数据模型TypedDictAnnotatedNode执行单元add_node(agent, fn)Edge固定跳转add_edge(a, b)Conditional Edge动态跳转add_conditional_edges(agent, tools_condition)START / END特殊起止节点add_edge(START, agent)ToolNode工具执行节点ToolNode([tool1, tool2])Checkpointer状态持久化MemorySaver()3. 适用场景与使用边界LangGraph 适合的典型场景包括多步骤任务拆解、RAG 问答后需要自动执行操作的流程、根据用户输入动态选择工具的 Agent、需要加入人工确认环节的内容生产流水线、跨轮次任务状态保持的客服系统。使用边界要拎清楚。它不适合单轮 Prompt 直出任务这类任务直接用大模型 API 更轻它也不适合承载重型训练或数据处理图编排不负责计算加速它更不适合替代业务系统Agent 只是其中一层决策和执行框架。需要特别强调的是接入在线大模型 API 时会把文本内容发送到模型服务商涉及敏感数据要确定脱敏方案调用工具时要考虑该工具是否允许高频访问。本文所有演示均在本地测试环境完成输出结果用于功能验证不做公网生产部署。4. 环境准备与前置条件动手之前先检查环境这套清单适用于大部分 LangGraph 项目。Python 3.9建议 3.10 或 3.11避免部分依赖在未来版本出现兼容问题。建议使用虚拟环境例如venv或conda不要让依赖污染系统 Python。需要一个可访问的大模型服务地址和 API Key。如果接本地模型需要额外配置 Ollama 或 vLLM 等推理服务。Windows 用户建议确认终端使用的是 PowerShell 或 Git Bash个别依赖编译报错需要先装对应版本的 Visual C 运行库。防火墙或代理策略需要允许访问你选择的模型服务域名。虚拟环境创建示例python -m venv .venv # Windows .venv\Scripts\activate # macOS / Linux source .venv/bin/activate安装依赖这一步在下一章统一说明。第一次接触 LangGraph 的同学不要跳步虚拟环境隔离能省掉大量“为什么别人能跑我不能跑”的问题。5. 安装部署与项目初始化LangGraph 的安装非常简单。以下是基础依赖实际项目按需增减。这里以常见组合为例pip install langgraph langchain-openai langchain-core如果你接的是其他厂商的模型也可以安装对应的 LangChain 适配包比如langchain-anthropic、langchain-community等。部分国内模型服务支持 OpenAI 兼容接口可以直接用langchain-openai指定base_url。项目目录建议这样组织langgraph-demo/ ├── .venv/ # 虚拟环境 ├── config.yaml # 模型配置 ├── tools/ # 自定义工具 │ └── device_tools.py ├── graphs/ # 图定义 │ ├── basic_agent.py │ └── workflow.py └── server.py # API 服务比较关键的是模型配置。不要把 API Key 直接写死在代码里用环境变量或者.env文件管理更稳妥OPENAI_API_BASEhttps://your-endpoint.example.com/v1 OPENAI_API_KEYyour-api-key MODEL_NAMEgpt-4o-mini代码里读取环境变量import os from langchain_openai import ChatOpenAI model ChatOpenAI( base_urlos.getenv(OPENAI_API_BASE), api_keyos.getenv(OPENAI_API_KEY), modelos.getenv(MODEL_NAME, gpt-4o-mini), )到这里LangGraph 项目已经初始化完成可以开始写第一张图。6. 代码实战一基础 Agent 与工具调用这一节直接演示一个带工具调用的 ReAct 风格智能体。场景是“设备状态查询 Agent”用户提供设备编号Agent 判断需要调工具工具返回状态Agent 汇总回答。6.1 定义工具from langchain_core.tools import tool tool def get_device_status(device_id: str) - str: 查询设备在线状态输入设备编号返回正常或离线。 status_map { A003: 正常, B204: 离线, C999: 未知, } return status_map.get(device_id, 没有该设备的记录)工具函数最上面必须写清楚描述因为大模型靠这个描述决定要不要调用它。返回字符串是 LangChain 工具的标准做法。6.2 定义 State 与节点from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode, tools_condition class State(TypedDict): messages: Annotated[list, add_messages] model_with_tools model.bind_tools([get_device_status]) def agent_node(state: State): response model_with_tools.invoke(state[messages]) return {messages: [response]} def build_basic_agent(model): builder StateGraph(State) builder.add_node(agent, agent_node) builder.add_node(tools, ToolNode([get_device_status])) builder.add_edge(START, agent) builder.add_conditional_edges(agent, tools_condition, {tools: tools, END: END}) builder.add_edge(tools, agent) return builder.compile()这段代码一共做了四件事定义全局消息状态把大模型绑定工具让模型具备工具调用能力注册 agent 节点和 tools 节点用条件边让模型自行决定是调工具还是直接结束。tools_condition是 LangGraph 预置逻辑检查模型返回的内容中是否包含工具调用请求无需自己判断 message 结构。6.3 运行测试graph build_basic_agent(model) result graph.invoke({ messages: [(user, 帮我查一下设备 A003 的状态)] }) print(result[messages][-1].content)预期输出大致是这样一句话设备 A003 的状态是正常。如果模型判断需要工具但工具结果为空则回答可能包含查询失败提示。判断标准进程正常结束控制台能看到回答工具相关日志显示节点从 agent 走到 tools 再回到 agent。第一次跑如果报错先确认三件事模型地址是否正确、API Key 是否有权限、模型名称是否与服务商提供的一致。容易忽略的是部分模型服务商不兼容工具调用格式导致tools_condition永远返回 END这类问题要查模型文档确认是否支持 OpenAI 风格的 function calling。7. 代码实战二工作流、分支循环与人工确认基础 Agent 演示了动态循环这一节看两个在工作中更常用的能力条件分支和人工确认。场景是内容生产流水线生成草稿 - 质量检查 - 人工审核 - 发布。没有通过质量检查的稿子直接重新生成。7.1 条件分支与循环import random def generate_draft_node(state: State): # 实际项目中这里替换为具体的生成逻辑 content model_with_tools.invoke([(system, 写一段产品介绍200字以内), (user, state[topic])]) return {messages: [content], draft: content.content} def quality_check_node(state: State): # 简化判断随机返回 pass 或 retry score random.randint(0, 10) return {quality_score: score} def should_publish(state: State): if state.get(quality_score, 0) 7: return publish return retryfrom langgraph.checkpoint.memory import MemorySaver builder StateGraph(State) builder.add_node(generate, generate_draft_node) builder.add_node(check, quality_check_node) builder.add_node(publish, lambda state: {final_output: 已发布 state[draft]}) builder.add_edge(START, generate) builder.add_edge(generate, check) builder.add_conditional_edges(check, should_publish, {publish: publish, retry: generate}) builder.add_edge(publish, END) saver MemorySaver() compiled_workflow builder.compile(checkpointersaver)这里 retry 边指回 generate 节点就构成了循环。你要限制重试次数的话可以在 State 里加一个retry_count字段在条件函数里判断次数超过上限就走到人工节点。7.2 人工确认节点LangGraph 的人工介入不是写一个特殊节点而是用interrupt_before参数让图跑到指定节点前停下config {configurable: {thread_id: daily-task-20260101}} graph_paused builder.compile( checkpointersaver, interrupt_before[publish], ) partial_result graph_paused.invoke( {topic: 智能硬件设备的产品介绍}, configconfig, ) print(流程已在 publish 前暂停等待人工确认) print(partial_result[draft])此时进程控制权交还给你。确认没问题后用同一个thread_id再次调用图会从暂停点继续final_result graph_paused.invoke(None, configconfig) print(final_result[final_output])注意第二次invoke的第一个参数传None意思是“不新增输入继续执行剩余节点”。这是 LangGraph 中断续跑的标准用法。人工确认内容正确后信息流会继续进入 publish 节点否则你可以在代码里选择重新生成或者丢弃本次任务。7.3 判断效果判断工作流改造是否成功主要看三点图能够停在 publish 前第二次调用可以继续执行而不是从头开始连贯执行时条件分支在多次运行中表现出差异化路径。运行时可以在控制台打印节点名称确认每一步。使用 MemorySaver 时重启进程后快照会丢失这只是测试用法生产环境要把 checkpointer 持久化到数据库。8. 接口 API、批量任务与服务化扩展图编译好之后下一步就是把它暴露成服务给上游系统调用。最直接的方式是包一层 FastAPI。8.1 用 FastAPI 包装 Agentfrom fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleLangGraph Agent Demo) class AgentRequest(BaseModel): prompt: str thread_id: str default-thread class AgentResponse(BaseModel): status: int message: str thread_id: str app.post(/agent/run, response_modelAgentResponse) async def run_agent(req: AgentRequest): try: config {configurable: {thread_id: req.thread_id}} result graph.invoke( {messages: [(user, req.prompt)]}, configconfig, ) return AgentResponse( status0, messageresult[messages][-1].content, thread_idreq.thread_id, ) except Exception as exc: return AgentResponse(status1, messagestr(exc), thread_idreq.thread_id)启动服务uvicorn server:app --host 127.0.0.1 --port 8000这里要多说一句同一个thread_id会对应同一个 checkpointer 快照状态所以多轮会话只要复用thread_idAgent 就能自动记住上下文。这是接口设计里最值得利用的一个点。8.2 curl 调用curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d {prompt: 查一下设备 A003 的状态, thread_id: demo-001}后端会返回 JSON结构包含status、message和thread_id。接口能跑通就可以接入其他工具链了。8.3 批量任务设计import csv import time tasks [ {id: t001, prompt: 查设备 A003}, {id: t002, prompt: 查设备 B204}, {id: t003, prompt: 查设备 C999}, ] results [] for task in tasks: try: config {configurable: {thread_id: fbatch-{task[id]}}} result graph.invoke({messages: [(user, task[prompt])]}, configconfig) results.append({ task_id: task[id], answer: result[messages][-1].content, }) print(f任务 {task[id]} 完成) except Exception as exc: results.append({task_id: task[id], error: str(exc)}) print(f任务 {task[id]} 失败: {exc}) with open(batch_results.csv, w, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnames[task_id, answer, error]) writer.writeheader() writer.writerows(results)批量任务的要点是每条任务用独立thread_id互不污染任务处理要记录日志失败任务单独标记并发量如果大可以用concurrent.futures.ThreadPoolExecutor控制并发数但要注意模型服务的限流这里不展开讲具体限流数值以实际服务商为准。9. 性能观察与常见问题排查LangGraph 本身是编排层节点之间的跳转开销很小瓶颈几乎都被大模型推理占走。评判性能时不要看框架微秒级调度要看以下几个维度单次模型调用的延迟、工具调用次数、图在数据库中持久化的读写耗时、以及用户在感知上的等待时间。流式输出是改善观感的一个重要手段。LangGraph 提供了astream方法可以让前端在模型生成过程中逐 token 收到内容而不是等整段完成后一次返回async for chunk in graph.astream( {messages: [(user, 查一下 A003 状态)]}, config{configurable: {thread_id: stream-001}}, ): print(chunk)为了让排查效率更高把常见问题整理成了这张表问题现象可能原因排查方式解决方案启动时报依赖安装失败Python 版本过旧或缺少编译环境检查python --version与报错尾部信息用 3.10/3.11 新建虚拟环境重新安装接口报认证失败API Key 错误或模型名不正确先单独测试模型客户端核对 Key、base_url、模型名模型输出不被 tools_condition 识别模型服务不支持工具调用格式打印模型返回 message 内容更换支持 function calling 的模型图执行进入无限循环条件边总是返回 retry 且无次数上限打印条件函数返回值在 State 中增加重试计数并在分支中判断二次 invoke 没有从暂停点继续thread_id 不一致或 checkpointer 丢失确认 config 中的 thread_id 完全相同复用相同 thread_id生产环境换持久化存储接口返回慢LLM 首字延迟较高观察模型 API 耗时启用流式输出或换更快的模型规格批量任务部分失败模型限流或网络抖动查看任务日志中的 exception加入退避重试例如指数间隔重试 3 次条件分支行为不符合预期State 中字段被覆盖打印每个节点后 State 快照用 reducer 合并字段避免简单覆盖排查时建议先打印节点执行顺序。在编译后的图对象上可以通过graph.get_graph()打印出图结构的文本表示确认拓扑是否符合预期。10. 最佳实践与后续扩展最后这部分给出几条工程化建议都是从实际项目里沉淀出来的经验。第一先小后大。任何新项目第一次运行都先用最小图验证只有 start、一个写死输出的节点、end。模型都先不接确认图能跑通再逐步加节点和工具。这一步能省掉大量反复排查的精力。第二State 设计要克制。不要把状态对象膨胀成“什么都有”的数据库表节点之间传参与收参遵循最小化原则。需要持久化的业务字段单独设计不要全部塞进 messages 里。第三人工确认只放在关键节点。用户不可能每次都审核整条链路把interrupt_before用在“发布前”“确认执行外部动作前”这类真正需要人把关的位置即可。自动化能覆盖的检查优先交给条件节点判断。第四日志和可观测性提前做。每个节点函数入口打印出当前执行节点名和时间失败时记录完整 State 快照。生产环境里的 Agent 调试问题九成发生在日志缺失的角落。第五接口服务要控制访问范围。FastAPI 示例里用的 127.0.0.1只允许本机调用跨机器调用需要配合认证鉴权。涉及人脸、声音、版权素材识别与生成时必须在确认授权后才纳入工作流调用第三方工具接口时要遵守对方平台的使用条款工具执行的每一步都应该在日志里留痕。后续扩展方向可以考虑三个方面。一是多 Agent 协作在同一个图里注册多个 agent 节点让不同节点负责不同领域二是并行分支在节点返回send指令实现 fan-out 并行处理适合 RAG 多路召回三是生产部署把 MemorySaver 换成数据库 checkpointer并考虑 LangGraph Platform 或自建服务。本文没有编造任何具体部署平台参数生产环境的架构选择需要你按实际项目规模和成本来评估。最后说一个比较容易踩的坑版本。LangGraph 迭代很快网上很多教程的 API 写法在你安装的版本里可能已经变了。遇到报错时不必纠结旧代码优先打开本地安装的langgraph包看实际方法名和参数。本文所有代码以常见 0.7.x 版本 API 风格为基准运行时要结合你本机安装版本微调。最有价值的第一个验证目标是把基础工具调用 Agent 跑通。这个目标实现之后你对 State、Node、条件边和工具绑定的理解就串起来了后面再学工作流、持久化和服务化都会快很多。先把最小图跑起来比背十张架构图都有用。