大家好我是你们的老朋友。这段时间在给一个知识库问答类项目做 Agent 升级发现真正难的不是让大模型“会说话”而是如何把一个完整的业务链路比如检索、推理、调用工具、记忆上下文、人工确认串成一个可控可回滚、可以观测和调试的流程。最开始试着用纯 LangChain 硬写代码越写越散维护成本很高后来切换到 LangGraph思路才一下子清晰起来。网上关于 LangGraph 的教程不少但要么停留在概念搬运要么代码是早期版本跑不起来了完整成体系的确实难找。这里把最近一段时间的实战经验整理成一篇系统教程从智能体架构讲起把核心组件逐个拆开再给一个能直接复制运行的真实对话 Agent 案例。不管你是刚接触 Agent 开发的新手还是已经在用 LangChain 想继续进阶的开发者这篇文章都能用得上。1. 背景与核心概念1.1 LangGraph 是什么LangGraph 是一个面向大语言模型应用的工作流编排框架它的核心思路是把原本隐藏在 Prompt 和多次 LLM 调用里的流程逻辑显式地建构成一张“图”。在这张图里每个节点Node是一个处理步骤可以是调用大模型、执行工具、读取数据库、做规则判断边Edge定义了步骤之间的流转方向状态State则是所有节点共享的一块“黑板”用来存消息、存中间结果、存业务标志。流程跑起来以后LLM 只是其中一部分其他逻辑由开发者自己掌控。如果你之前用过 LangChain 的Chain可以这么理解两者的差异LangChain Chain偏向线性执行数据在链中按顺序传递逻辑相对固定。LangGraph Graph几乎所有环节都能自定义支持分支、循环、条件跳转、人工介入适合做复杂智能体。说白了LangGraph 把 Agent 开发从“靠提示词碰运气”变成了“靠流程设计保下限”。以前我们反复调 Prompt 希望模型“按规矩办事”现在我们可以直接在图上规定它必须先做什么、再做什么、什么情况下停止。这在小任务里好像区别不大一旦进入多步骤、多工具的复杂流程优势会非常明显。1.2 LangGraph 解决什么问题在实际开发中用普通方式实现一个 Agent 通常会遇到下面几个问题。第一个问题是流程不可见。大模型返回什么、调用了什么工具、中间经历了哪些判断全部黑盒。一旦线上出问题排查只能靠猜。第二个问题是状态难以管理。一个复杂的 Agent 往往需要多轮对话、多次工具调用这些中间变量散落在代码各处。如果还要支持不同用户会话之间的隔离和持久化实现起来非常痛苦。第三个问题是控制力不足。很多时候我们并不希望 Agent 完全自主行动比如执行支付操作前需要人工确认、读取敏感数据前需要检查用户权限。纯 LLM 驱动的循环很难在关键节点“暂停”。LangGraph 从设计上兼顾了这些问题图本身就是可观测的流程图State 统一管理上下文Checkpointer 支持会话记忆和断点续跑Interrupt 机制能让流程停在指定节点等人来处理。整个 Agent 开发就从一个“不可控的黑盒”变成了一张“可干预、可审计、可回滚的工作流图”。1.3 与 LangChain、LangSmith 的关系这三个项目同属 LangChain 生态但定位完全不同。LangChain提供模型封装、Prompt 管理、文档加载、向量库集成等基础组件相当于工具箱。LangGraph负责把这些组件编排成状态图解决流程编排和状态管理问题相当于流水线。LangSmith负责对 Agent 运行过程做 Trace 和评测记录每一轮的输入输出、Token 消耗、耗时适合排查线上问题。在官方推荐的技术栈里三者经常配合使用但 LangGraph 并不是 LangChain 的替代品它更像 LangChain 能力的组合器和调度器。本文核心聚焦在 LangGraph 本身的架构与代码实践用到了少量 LangChain 的ChatOpenAI模型接口。1.4 适用场景LangGraph 适合需要明确流程和多步骤控制的场景客服机器人先识别用户意图再查知识库必要时调用订单查询工具。数据分析助手根据问题选图表、写代码、执行代码、根据结果修正分析。自动化运维 Agent读取监控数据、定位问题、生成修复建议并等待管理员确认后执行。多级编排工作流内容审核、智能写作、报告生成等。不过也要提醒一句如果只是“单轮问答 简单工具调用”直接用单次模型调用就够了没必要为了图而图。LangGraph 的收益在复杂流程中才明显。2. 环境准备与版本说明2.1 运行环境要求LangGraph 是一个纯 Python 库核心依赖相对轻量但会间接依赖langchain-core、langchain等包。在动手之前建议先创建一个干净的环境python -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate建议使用 Python 3.9 及以上版本。至于 LangGraph 具体版本号不同时间段的最新版差异比较大网上教程里很多代码跑不通就是因为版本 API 变了。我这里统一建议安装最新稳定版并在成功后查看版本pip install --upgrade langgraph langchain langchain-openai langchain-corepython -c import langgraph; print(langgraph.__version__)本文代码重点演示的是 0.2 及之后版本中稳定存在的一套 API包括StateGraph、START、END、MemorySaver、ToolNode、tools_condition。如果你的环境是更新的大版本API 名称可能微调过按官方文档对应调整即可核心思路完全一致。2.2 配置大模型服务LangGraph 本身不关心底层大模型是谁它通过 LangChain 的标准模型接口对接各家模型。最省事的做法是使用兼容 OpenAI 协议的模型服务。安装好依赖后在项目中创建.env文件OPENAI_API_KEY你的Key OPENAI_BASE_URLhttps://api.openai.com/v1如果你使用的是国内模型服务或本地部署的模型服务只需要把OPENAI_BASE_URL改成对应地址即可。尽量优先选一个稳定的模型服务否则后面的工具调用效果会受影响。示例中我会用ChatOpenAI作为模型入口。如果要用 Anthropic 的模型就安装langchain-anthropic然后把模型对象换成ChatAnthropic图中的逻辑完全不变。2.3 推荐的调试工具langgraph devLangGraph 自带的本地开发服务可以可视化流程。LangSmith需要注册并配置环境变量LANGCHAIN_TRACING_V2true。Python 内置logging在节点函数里打日志简单直接。对于刚入门的读者我建议先把核心代码跑通再逐步引入可视化、追踪等能力避免一上来信息量太大。3. LangGraph 核心组件原理拆解LangGraph 的 API 虽然不少但核心概念其实只有几张“牌”State、Node、Edge、Conditional Edge、Checkpointer。把这几个搞明白大部分业务工作流都能搭建出来。3.1 State全流程共享的“黑板”State 是 LangGraph 里最重要的设计。每个节点执行后返回一个字典这个字典会刷新全局状态。其他节点再从状态里取自己需要的字段继续处理。定义状态通常有两种方式。3.1.1 简单 TypedDict 定义from typing import TypedDict class State(TypedDict): question: str answer: str steps: int这种定义方式最简单每个节点只需要返回{answer: ...}就可以覆盖上游写入的旧值。理解起来非常直观。3.1.2 复杂字段用 Reducer业务中经常碰到“追加而非覆盖”的需求比如我们要保存完整的对话历史而不希望每次模型调用都把上一轮消息冲掉。这就要用到 LangGraph 的 Reducer 机制。最经典的 Reducer 是add_messagesfrom typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]当多个节点都想往messages里写入新消息时add_messages会自动把新消息和旧消息合并成列表而不是用新值覆盖旧值。配合 LangChain 的HumanMessage、AIMessage、ToolMessage一套完整的对话历史就维护起来了。简单总结一下节点返回字典。普通字段默认覆盖。Annotated字段可以自定义合并逻辑。状态是整个图共享的节点之间通过它传递信息。3.2 Node图中的最小执行单元Node 可以是一个普通 Python 函数也可以是一个可调用对象。它接收当前 State做一些业务处理然后返回一个字典来更新状态。下面是一个最简单的节点def call_model(state: AgentState): answer 这是一个固定的回答 return {answer: answer}真正在工作流中节点往往做更多的事调用模型、查询数据库、执行工具、做权限校验。为了好维护建议每个函数只做一件事函数名就代表它在图里的用途。节点加入到图中用add_nodefrom langgraph.graph import StateGraph builder StateGraph(AgentState) builder.add_node(call_model, call_model)节点名字是图里的标识符随便取但建议用可读性强的英文名。3.3 Edge 与条件边定义流程走向边回答了“下一步去哪里”的问题。LangGraph 里有三种常见边。普通边从一个节点无条件走到另一个节点builder.add_edge(start_node, next_node)START 和 ENDSTART是图的入口END是图的结束。任何一个可运行的工作流都必须有入口和出口from langgraph.graph import START, END builder.add_edge(START, start_node) builder.add_edge(next_node, END)条件边条件边相当于流程里的if-else。它接收当前状态返回一个字符串决定下一步去哪def should_continue(state: AgentState): if state[steps] 3: return continue return end builder.add_conditional_edges( call_model, should_continue, { continue: call_model, end: END, } )条件边是实现 Agent 循环的核心武器。在 ReAct 模式里模型决定是否调用工具就对应一个条件边如果模型返回工具调用指令就进入工具节点否则结束。3.4 Checkpointer给工作流加入记忆默认情况下每次graph.invoke()执行结束后整张图的状态就丢弃了。第二次重新调用时又从空白状态开始。如果希望 Agent 记住上一轮聊过什么就需要 Checkpointer。它可以捕获每一步的状态变化并按thread_id保存起来。最简单的内存版from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app builder.compile(checkpointercheckpointer)调用时带上thread_id同一个线程内的多轮会话就会共享状态config {configurable: {thread_id: user-001}} result app.invoke({messages: [(user, 你好)]}, config)从MemorySaver换成SqliteSaver或 PostgresSaver就把记忆从内存持久化到了磁盘。这样即使服务重启Agent 也不会遗忘上下文。3.5 ToolNode预置工具执行节点大多数 Agent 都需要调用工具。如果不做封装工具调用需要我们手动写解析模型返回的 tool_calls执行对应的 Python 函数再把结果包装成 ToolMessage 放回消息列表。这套流程非常固定LangGraph 直接用ToolNode帮我们完成了。from langgraph.prebuilt import ToolNode def get_weather(city: str) - str: 查询指定城市的天气。 return f{city}晴25℃东南风2级 tools [get_weather] tool_node ToolNode(tools) builder.add_node(tools, tool_node)结合条件边就得到了经典的 ReAct 图模型节点判断是否需要工具如果需要就进tools节点否则结束工具执行完把结果写回消息再回到模型节点继续推理直到答案完整。3.6 Interrupt人工介入与审批有些场景下 Flow 不能自动跑完。例如删除数据库前要管理员确认ToolNode 不应该直接执行。LangGraph 提供了interrupt函数流程执行到该节点时会暂停把状态保存下来等待外部输入确认后再继续。from langgraph.types import interrupt def human_approval(state: AgentState): action state[action] result interrupt({action: action, tips: 请人工确认是否执行}) return {approved: result}调用方在第一次 invoke 后拿到暂停状态用户确认后携带Command(resume...)继续执行。这个机制让 Agent 在关键节点停下来而不是一股脑把所有操作跑完非常适合集成到需要审批的工作流里。Interrupt 的 API 在不同版本里变化不小如果你的版本不支持langgraph.types.interrupt多看看官方手册里的迁移说明原理是一致的。4. 完整实战搭建一个带记忆的检索和工具调用 Agent前面讲了这么多概念接下来用一个可运行的例子把整张图串起来。4.1 项目结构与需求我们做一个“个人助手 Agent”功能是记住用户在多轮对话中的上下文。根据用户问题判断是否需要调用工具。提供一个城市天气查询工具。当模型发出工具调用请求时自动执行工具并把结果返回给模型。项目结构如下langgraph_agent_project/ ├── .env ├── requirements.txt └── agent.pyrequirements.txt 内容如下langgraph0.2.40 langchain0.3.0 langchain-openai0.2.0 python-dotenv1.0.04.2 编写基础代码在agent.py中先完成环境加载、模型初始化和工具定义import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() model ChatOpenAI( modelgpt-4o-mini, temperature0, )如果用的是 DeepSeek、通义千问等模型通常只需要修改base_url和modelmodel ChatOpenAI( modeldeepseek-chat, api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL), )定义工具def get_weather(city: str) - str: 查询指定城市的当天天气。 weather_map { 北京: 晴25℃, 上海: 多云28℃, 广州: 小雨26℃, 深圳: 雷阵雨29℃, } return weather_map.get(city, 暂不支持该城市查询)这里特别注意工具函数的文档字符串会被模型读取用来判断该工具该何时调用。所以一定要写清楚函数用途参数名也要有清晰的含义。4.3 手动构建状态机版 ReAct Agent先不要急着用高级封装把手动方式跑通这样你就能真正理解 Agent 内部发生了什么。首先定义状态from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]然后定义模型调用节点和工具调用节点from langchain_core.messages import ToolMessage def call_model(state: AgentState): # 只有 model 节点自己可以决定是否生成 tool_calls response model.invoke(state[messages]) return {messages: [response]} def call_tool(state: AgentState): last_message state[messages][-1] tool_results [] for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] # 这里可以根据工具名做一个简单的分发 if tool_name get_weather: result get_weather(**tool_args) else: result f未找到工具: {tool_name} tool_results.append( ToolMessage(contentresult, tool_call_idtool_call[id]) ) return {messages: tool_results}接着定义条件边函数决定是否继续调用工具def should_continue(state: AgentState): last_message state[messages][-1] if last_message.tool_calls: return tools return END最后组装成图from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver builder StateGraph(AgentState) builder.add_node(model, call_model) builder.add_node(tools, call_tool) builder.add_edge(START, model) builder.add_conditional_edges(model, should_continue, {tools: tools, END: END}) builder.add_edge(tools, model) checkpointer MemorySaver() app builder.compile(checkpointercheckpointer)这个图已经是一个完整的 ReAct Agent模型节点输出判断是否要调工具调完工具回到模型节点继续分析直到模型不再生成tool_calls流程走向 END。4.4 调用运行写一个测试入口if __name__ __main__: config {configurable: {thread_id: demo-001}} # 第一轮 result app.invoke( {messages: [(user, 北京今天天气怎么样)]}, config, ) print(result[messages][-1].content) # 第二轮此时模型拥有上一轮的记忆 result app.invoke( {messages: [(user, 那上海呢)]}, config, ) print(result[messages][-1].content)预期输出类似北京今天晴25℃适合外出活动。 上海今天多云28℃体感温度稍热注意补水。可以看到第二轮问“那上海呢”时模型结合了 thread 中第一轮的历史知道是在对比城市天气。4.5 使用 Prebuilt ToolNode 简化代码如果觉得手动写call_tool太繁琐可以直接用langgraph.prebuilt里的ToolNode。手动版适合理解底层原理实际项目我更推荐简洁版。from langgraph.prebuilt import ToolNode, tools_condition tools [get_weather] tool_node ToolNode(tools) builder StateGraph(AgentState) builder.add_node(model, call_model) builder.add_node(tools, tool_node) builder.add_edge(START, model) builder.add_conditional_edges(model, tools_condition, {tools: tools, END: END}) builder.add_edge(tools, model)tools_condition就是 LangGraph 官方预置的“是否调用工具”判断函数返回tools或END。整个流程和使用 ToolNode 的注意事项不用再手写代码简洁多了。如果你的需求高度模板化还可以直接用create_react_agent一行创建from langgraph.prebuilt import create_react_agent app create_react_agent( modelmodel, toolstools, state_modifier你是一个友好的助手回答要简洁必要时才调用工具。, ) app app.compile(checkpointerMemorySaver())这个方法适合快速搭建原型但自定义能力会弱一些。正式项目为了应对复杂业务逻辑通常还是基于StateGraph自行设计节点和边。5. 常见问题与排查思路LangGraph 实际跑起来之后最容易踩到的坑集中在状态类型、节点返回、工具调用三个方面。下面整理几个高频问题。5.1 报错Invalid node name ...原因add_edge或add_conditional_edges里写了一个不存在的节点名。排查步骤用builder.get_graph().print_ascii()打印当前图结构。核对接下来的add_edge参数是否与图中节点一致。检查是否忘了用add_node注册节点。builder.get_graph().print_ascii()5.2 Agent 不调用工具只输出文字原因之一模型没被绑定工具。如果直接model.invoke()ChatOpenAI 不会知道外面有工具存在必须使用model.bind_tools(tools)或者在create_react_agent中传入 tools。原因之二工具文档和参数描述不够清晰模型判断不需要调用。解决方案model_with_tools model.bind_tools(tools)然后 Node 里调用的是model_with_tools不是原model。5.3 工具调用后模型报错ToolMessage与tool_call_id不匹配原因工具返回消息时没有带上tool_call_id模型无法把工具结果和原始调用关联起来。解决方案用ToolMessage(contentresult, tool_call_idtool_call[id])确保 ID 原样返回。如果使用ToolNode它会自动完成这一步。5.4 记忆失效对话隔了几轮就“失忆”原因没有配置 Checkpointer或者每次调用使用不同的thread_id。排查print(config[configurable][thread_id])如果使用了MemorySaver注意服务重启后内容会丢失。需要长期记忆时换成SqliteSaver。5.5 状态字段被覆盖不是追加原因定义状态时没用Annotated[T, add_messages]于是后一个节点的返回值把前一个节点写的数据冲掉了。解决方案消息列表字段使用add_messagesreducer。问题现象常见原因解决思路启动报无效节点边引用了未注册的节点名检查add_node与add_edge模型不调用工具没有bind_tools创建model.bind_tools(tools)工具结果丢失ToolMessage 缺 ID使用ToolNode或补全tool_call_id多轮对话失忆缺少 Checkpointer编译时传入 MemorySaver消息覆盖旧值状态未配置 reducer使用Annotated[list, add_messages]6. 最佳实践与工程建议看完代码没问题只是第一步把 Agent 真正稳定地部署到生产环境还需要关注很多工程细节。6.1 图设计要小而清晰理想情况下每个节点只做一件事。一个节点里既查数据库、又调模型、还更新业务状态以后排错非常痛苦。宁可多拆几个节点也不要让单个函数超过 50 行。节点命名建议统一使用动词短语例如retrieve_docs、generate_answer、check_permission这样画流程图和查日志都直观。6.2 控制 State 体积State 是全流程共享的很多中间结果不需要全局保留。比如“生成的 Prompt 原文”可能只在模型调用节点用到没有后续消费者存下来只会浪费 Token 和内存。建议在节点返回时只写必要字段或者定期清理大的中间字段。一个经验法则是先记录流程必须依赖的数据再考虑做追踪和审计的数据。6.3 必须有超时和重试LLM 服务可能出故障、限流、超时。建议对模型节点做两层保护外层langchain的with_retry或团队封装的装饰器处理偶发网络错误。内层给整次invoke设置总超时时间避免用户等待无限。model model.with_retry( retry_if_exception_type(TimeoutError, ConnectionError), stop_after_attempt3, )6.4 安全边界要前置Agent 能调用工具意味着代码、“系统指令”、“数据库权限”都可能成为攻击面。生产环境必须做到给工具函数做参数校验和长度限制。禁止工具拼接任意 shell 命令除非在隔离容器里。涉及用户隐私、支付、删除类操作时一律走人工确认或二次身份验证。把工具权限与前端用户权限统一收敛到一个权限服务中。这里特别说一下工具不是越细越好。能提供 3 个高度内聚的工具比提供 20 个单点工具更容易控制行为边界。6.5 日志与追踪建议同时保留两条日志通道应用日志记录节点进出、耗时、异常信息。LangSmith 追踪记录模型输入输出、工具调用细节、Token 消耗。上线第一位排查问题时先看 LangSmith Trace再定位到具体 Node。如果没法接 LangSmith至少要保证每个节点开始前和结束后都打一条结构化日志。6.6 人工审批要放在最安全的环节使用interrupt时审批节点越是靠近真正执行操作的节点越好。比如审批确认没问题后直接进入执行节点如果在早早就审批了后面节点又把数据改了那审批就失去了意义。6.7 测试策略对普通代码我们习惯写单测对 Agent建议按层级测节点级单测准备一个 State调用一个节点函数检查返回字段。图级集成测试跑完整工作流验证最终结果。回归测试固定一组 Prompt 与工具排列比较每次输出是否稳定。固定模型参数temperature0可以减少随机性但不要指望输出完全一致。引入自动化测评比看一两次输出更可靠。7. 总结与学习路线这篇教程把 LangGraph 的智能体架构拆成了三部分用 State 管理上下文用节点封装处理步骤用边控制流程走向再配合 Checkpointer 实现记忆。然后通过一个带工具调用和会话记忆的天气问答 Agent把从模型配置、工具定义、手动 ReAct 构建到ToolNode简化封装完整走了一遍。对于想深入学习的同学建议按下面顺序继续把本文的手动 ReAct 示例改写为create_react_agent感受两者差异。为项目配置SqliteSaver把记忆从内存改成持久化。尝试再加一个工具比如计算器或数据库查询把ToolNode跑通第二种工具。学习使用interrupt给删除、支付类操作加人工审批。接入 LangSmith学会看 Trace 和耗时统计排查线上问题。在实际项目中最优先要注意的风险不是大模型回答错而是工具被错误调用、状态被污染、上下文膨胀、以及权限边界失效。只要流程图设计清楚、超时重试和安全校验到位LangGraph 能很好地从 Demo 转成生产级应用。如果这篇文章对你有帮助建议先点收藏按着代码敲一遍再进入下一步。遇到报错也欢迎在评论区带图和版本号提问我会尽量帮你定位。后面我还会更新关于StateGraph多分支、基于interrupt的审批型 Agent 长文欢迎持续关注。