资讯动态

LangGraph实战:从状态图到智能体编排的核心指南

发布时间:2026/10/22 2:38:24 来源:尧图企业网站定制
如果你最近在做 Agent 开发大概率已经听过 LangGraph 的名字。但要搞清楚它到底解决了什么问题可能比想象中要费劲有人把它当成“又一个 LangChain 插件”有人觉得它不过是把对话循环包了一层还有人打开官方文档后被StateGraph、add_node、add_conditional_edges这些概念直接劝退。我的判断很简单LangGraph 真正改变的不是“调用大模型”的方式而是把 Agent 从“一段调模型的循环”变成了“一张可以设计、可以调试、可以挂载记忆的图”。如果你还在用几十行while True维护对话状态或者用一堆 if-else 判断工具调用又或者希望 Agent 流程能让团队成员一眼看懂那这篇实战教程就是为你准备的。本文会从应用场景出发先讲清楚 LangGraph 的智能体架构和核心组件然后给出最小可运行示例再带你实现条件分支、工具调用和多轮记忆。最后会整理我见过最频繁的几个踩坑点以及生产环境下的工程建议。建议先收藏配置环境时直接对照操作。1. LangGraph 到底是什么它解决了什么问题1.1 传统 Agent 开发的三个痛点在 LangGraph 出现之前自己做 Agent 通常有三种方案但每一种都有明显的痛点。第一种是直接用 LangChain 的AgentExecutor。它确实把 Agent 的“思考-行动-观察”循环封装好了但你很难控制它内部的具体流程。你希望“先查库存再查物流”“查询失败就尝试一次重试”这些精细化控制都需要绕开现有封装去 hack调试时很不直观。第二种是自己写循环。很多团队为了灵活性选择直接调模型 API然后自己维护工具调用逻辑。这种方案的问题在于状态散落在各个变量里多轮对话时要手动拼接历史消息工具调用结果和中间步骤没有统一记录。一旦流程复杂代码很快就会变成一坨“上下文变量接力赛”。第三种是纯用 Prompt 让模型记住步骤。比如在系统提示词里写“先分析再搜索最后回答”。这种方案成本最低但效果也最不稳定。模型可能跳步可能把中间结果和最终结论混在一起而且没有任何机制保证它能“按流程执行”。这三个痛点的本质是同一个Agent 的执行过程缺少一张明确的“图”。你无法定义节点、连接节点、控制分支也无法在任意节点挂载状态。1.2 LangGraph 与 LangChain 的关系很多人误以为 LangGraph 是 LangChain 的升级版其实它们是两个不同层次的东西。LangChain 提供的是“模型调用、Prompt 模板、输出解析、外部工具”这一层抽象。它负责的是“Agent 和组织外部世界打交道”的零件。LangGraph 则是站在更高一层负责“编排流程”它把每一步处理都看成图上的节点把节点之间的跳转看成边把整个执行过程中的数据看成 State。官方原话是“LangGraph 是一个用于构建有状态、有条件的多参与者应用的低层编排框架”。注意这几个词有状态、有条件、多参与者、低层编排。它并没有取代 LangChain 的模型封装和工具封装而是把 Agent 的总体执行逻辑从“黑盒循环”变成了“白盒图”。你可以只用 LangGraph不用 LangChain 的模型封装也可以在 LangGraph 里继续使用ChatOpenAI、ToolNode这类 LangChain 组件。两者是配合关系不是竞争关系。1.3 它适合谁不适合谁从实际使用体验来看LangGraph 更适合下面几类人需要构建复杂流程的开发者比如多步骤 RAG、审批流、多 Agent 协作。需要给流程加“可控性”的开发者比如某些步骤必须走、某些条件满足后才跳转。需要把多轮会话状态落盘的项目比如客服机器人、智能助手。希望流程能被可视化和监控的团队。但它并不适合所有场景。如果你的需求只是“大模型接一个工具问一句答一句”那 LangGraph 带来的收益很小直接用函数调用接口更轻。如果你的 Agent 流程只有固定三步且永远不会变化一张普通的顺序执行图也比 LangGraph 更简单。不要因为它名字里带 Graph 就觉得它是万能的。明确了这个边界后面学习核心组件时才不会学偏。2. 图形化理解 Agent核心概念与原理LangGraph 的核心概念并不多State、Node、Edge、Conditional Edge、Checkpointer。把五个概念串起来你就理解了 LangGraph 的全部骨架。2.1 State图的“全局变量”在 LangGraph 里State 代表整个 Agent 执行过程中的共享数据。它可以是字典、TypedDict也可以是 dataclass。每个节点都能读取 State也能向 State 写入新的字段或追加数据。这里特别容易误解的一点是State 不是“某一个节点内部的数据”而是“整张图执行期间的全局状态”。所有节点共享同一个 State节点返回值会被合并回 State 中。这样设计的好处是你不需要在每个节点之间手动传参数据流天然清晰。在实际工程中我建议把 State 字段控制在一个合理范围内。比如只放messages、current_step、tool_results、final_answer这几个关键字段不要把临时推理过程全部塞进去。State 字段越多图的可维护性越差。2.2 节点、边和条件边Node 是一个普通的函数或可调用对象。它接收当前的 State处理业务逻辑然后返回一个字典字典里的内容会被更新到 State。Edge 定义节点之间的跳转关系。在 LangGraph 里有两种边普通边直接跳到下一个节点和条件边根据当前 State 的内容决定跳转到哪个节点。条件边是 Agent 具备“判断力”的关键。同一个节点既可以连接到“工具调用节点”也可以连接到“直接回答节点”。到底走哪条路由你写的 route 函数决定。这种设计比在 Prompt 里写着“如果……就……”要可靠得多因为路由逻辑是完全确定的代码不依赖模型临场发挥。2.3 Checkpointer让 Agent 拥有“记忆”普通多轮对话怎么记忆历史最常见的做法是把历史消息拼到 Prompt 里。但这种方式有两个问题一是每次都要把全部历史拼进去浪费 Token二是中间状态无法恢复比如用户在第二轮中途想撤销重做传统方式很难处理。LangGraph 的解决方案是 Checkpointer。它会在每次图执行到一个节点时把当前 State 保存下来。下次执行相同线程的图时可以从保存的状态继续运行。这个“线程”由configurable.thread_id标识一个用户一个 ID他的多轮上下文就会被自动恢复。MemorySaver是最简单的内存版 Checkpointer。它适合开发调试和生产中的单实例场景但如果在多实例部署环境下需要把 Checkpointer 换成 Redis 或数据库持久化版本。2.4 和普通 if-else 的本质区别有开发者会问条件边不就是 if-else 吗为什么值得单独设计区别在于if-else 是分支结构而 LangGraph 的条件边支持任意复杂的图拓扑。你可以实现“工具节点失败后回到上一步重试”也可以实现“多个人类输入等待”还可以实现一个节点同时汇聚多个上游结果。这些在纯代码里需要引入状态机和回调体系而在 LangGraph 里只是一条边的定义。它把“过程控制”显式提升到了框架层面这是普通代码分支做不到的。3. 环境准备与依赖安装3.1 Python 版本与虚拟环境LangGraph 基于 Python 3.9 以上版本开发建议使用 3.10 或 3.11。我推荐先建一个独立的虚拟环境避免和系统 Python 或公司项目产生依赖冲突。mkdir langgraph-demo cd langgraph-demo python -m venv .venv source .venv/bin/activateWindows 环境下激活命令是.venv\Scripts\activate。看到命令行前缀出现(.venv)就代表环境已经激活。3.2 安装 LangGraph 及相关依赖最小安装只需要langgraph但要跑通完整示例还需要模型封装库和配置文件读取库。以 OpenAI 兼容接口为例pip install langgraph langchain-openai python-dotenv如果你的模型服务商不是 OpenAI 官方渠道而是国内云厂商或自建的 OpenAI 兼容服务只需要在初始化ChatOpenAI时指定base_url即可具体地址以服务商文档为准。注意这要求你通过合法合规的渠道获取 API 密钥和服务访问权限。如果安装时出现版本冲突常见原因是你本地已经安装了较早版本的langchain。建议先执行pip show langgraph langchain看一下版本再考虑升级或重装。3.3 配置模型 API不硬编码密钥任何生产项目都不应该把 API Key 写进代码。这里用.env文件统一管理# 文件路径.env MODEL_API_KEYyour-api-key-here MODEL_BASE_URLhttps://api.example.com/v1 MODEL_NAMEgpt-4o-mini然后在 Python 代码里通过dotenv读取import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY os.getenv(MODEL_API_KEY) OPENAI_BASE_URL os.getenv(MODEL_BASE_URL) MODEL_NAME os.getenv(MODEL_NAME)注意.env文件不要提交到 Git 仓库。项目根目录下的.gitignore里至少应该包含.env这一行。4. 最小可运行示例三节点图跑通流程这一节的目标只有一个用最少的代码把一张图跑起来。理解这张最小图之后再看复杂的工具调用和多轮记忆会轻松很多。4.1 定义 State建议使用TypedDict定义 State它能提供类型提示写节点函数时不容易出错。4.2 定义节点函数每个节点函数都接收一个state参数返回一个字典作为状态更新。4.3 构建图并调用# 文件路径demo_graph.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): messages: list step: str def node_llm(state: State) - dict: print(进入节点node_llm) return {step: llm_done, messages: state[messages] [模型已收到消息]} def node_tool(state: State) - dict: print(进入节点node_tool) return {step: tool_done, messages: state[messages] [工具已执行完成]} def node_output(state: State) - dict: print(进入节点node_output) return {step: output_done, messages: state[messages] [输出整理完成]} graph StateGraph(State) graph.add_node(llm, node_llm) graph.add_node(tool, node_tool) graph.add_node(output, node_output) graph.add_edge(START, llm) graph.add_edge(llm, tool) graph.add_edge(tool, output) graph.add_edge(output, END) app graph.compile() result app.invoke({messages: [], step: }) print(最终 State, result)运行python demo_graph.py预期你会看到三个节点的打印信息然后输出最终 State {messages: [模型已收到消息, 工具已执行完成, 输出整理完成], step: output_done}这个示例看起来简单但它已经把 LangGraph 的五个核心概念全部用到了State、Node、Edge、START、END。后面所有复杂功能都是在这五个概念之上扩展的。我这里刻意没有真的调用大模型目的是先让你理解“图的执行机制”。如果一开始就引入模型调用你会发现很难分辨某个报错是模型的问题还是图的问题。5. 条件分支与工具调用让 Agent 有判断力最小示例中的图是固定顺序的“一条道走到黑”。真实 Agent 需要根据用户提问来决定是否调用工具。下面模拟一个“天气查询 Agent”判断用户是否想查天气想查就进入工具节点不想查就直接回答。5.1 场景设计用户输入“北京今天天气怎么样”Agent 需要调用天气工具用户输入“你好”Agent 不需要调工具。关键在于判断逻辑。最可靠的方式是用大模型判断用户意图让模型输出结构化指令路由函数根据指令选择路径。为了不依赖某个特定模型是否支持 function calling我们先写一个“模型输出 JSON 指令”的版本逻辑直观适合理解条件边。5.2 路由节点逻辑路由函数返回的是下一个节点名字符串。LangGraph 会根据返回值自动跳转。这里我们需要两个可选目标tool_node和answer_node。5.3 完整代码示例# 文件路径conditional_demo.py import json import os from typing import TypedDict from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END load_dotenv() class State(TypedDict): user_input: str need_tool: bool tool_result: str answer: str def build_llm(): return ChatOpenAI( modelos.getenv(MODEL_NAME, gpt-4o-mini), api_keyos.getenv(MODEL_API_KEY), base_urlos.getenv(MODEL_BASE_URL), temperature0, ) def judge_node(state: State) - dict: llm build_llm() prompt f 请判断用户是否在询问天气信息。 如果询问天气只输出{{need_tool: true}} 如果没有询问天气只输出{{need_tool: false}} 用户输入{state[user_input]} resp llm.invoke(prompt) parsed json.loads(resp.content) return {need_tool: bool(parsed[need_tool])} def fake_weather_tool(state: State) - dict: city 北京 weather f{city}晴气温 24 度微风 return {tool_result: weather} def answer_node(state: State) - dict: if state.get(tool_result): return {answer: f根据天气工具查询结果{state[tool_result]}} return {answer: 你好我是天气助手你可以问我任意城市的天气。} def route_after_judge(state: State) - str: if state.get(need_tool): return tool_node return answer_node graph StateGraph(State) graph.add_node(judge, judge_node) graph.add_node(tool_node, fake_weather_tool) graph.add_node(answer_node, answer_node) graph.add_edge(START, judge) graph.add_conditional_edges(judge, route_after_judge, { tool_node: tool_node, answer_node: answer_node, }) graph.add_edge(tool_node, answer_node) graph.add_edge(answer_node, END) app graph.compile() print( 第一个问题询问天气 ) r1 app.invoke({user_input: 北京今天天气怎么样, need_tool: False, tool_result: , answer: }) print(r1[answer]) print( 第二个问题普通问候 ) r2 app.invoke({user_input: 你好, need_tool: False, tool_result: , answer: }) print(r2[answer])这里的关键在于route_after_judge函数。它读的是state[need_tool]而这个字段已经在judge_node里被模型更新过了。也就是说模型负责“理解意图”代码负责“选择路径”。这是 LangGraph 中很常见的“模型做决策、代码定规则”分工方式。运行后如果模型配置正确第一个问题应该走工具节点第二个问题直接走回答节点。如果第二个问题没有走回答节点说明judge_node输出的 JSON 格式不固定可以在 Prompt 里加入 few-shot 示例约束。5.4 生产环境推荐ToolNode上面的写法适合教学因为逻辑清楚不依赖额外组件。但生产环境里更推荐直接使用模型原生的 function calling 能力。LangGraph 的 prebuilt 节点已经封装好了ToolNode和tools_condition代码更短# 文件路径toolnode_demo.py from langgraph.prebuilt import ToolNode, tools_condition from typing import TypedDict, Annotated from langchain_core.messages import HumanMessage from langchain_core.tools import tool tool def get_weather(city: str) - str: 获取指定城市的天气 return f{city}晴气温 24 度 class State(TypedDict): messages: Annotated[list, lambda old, new: old new] llm build_llm().bind_tools([get_weather]) def assistant(state: State): return {messages: [llm.invoke(state[messages])]} graph StateGraph(State) graph.add_node(assistant, assistant) graph.add_node(tools, ToolNode([get_weather])) graph.add_edge(START, assistant) graph.add_conditional_edges(assistant, tools_condition) graph.add_edge(tools, assistant)这种方式把工具调用的判断完全交给模型和tools_condition完成代码量少逻辑也更贴近真实产品。如果项目刚起步建议从 5.3 的自定义路由开始如果模型支持工具调用再切换到 ToolNode。6. 多轮对话与状态持久化让 Agent 记住上下文上一节的示例每执行一次invoke图的状态都是从头开始的。实际产品中的客服机器人、AI 助手都有多轮对话需求。用户说“那上海呢”Agent 需要知道“那”指的是“上海天气怎么样”而不是把这句话当成新问题理解。6.1 为什么普通拼接不够最简单的多轮记忆是把所有历史消息拼成 Prompt。但这样会带来两个问题Token 越来越多成本快速上升Agent 无法区分哪些是历史决策过程、哪些是当前用户意图。LangGraph 的 Checkpointer 机制可以解决这个问题。6.2 使用 MemorySaver 保存状态下面是一个带 Checkpointer 的多轮示例每次调用都指定同一个thread_idLangGraph 会从历史状态继续执行# 文件路径memory_demo.py from typing import TypedDict from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START, END from langchain_core.messages import HumanMessage class State(TypedDict): messages: list def chat_node(state: State) - dict: # 实际项目中这里会用大模型生成回复 last_user_message state[messages][-1].content reply f我记住了你刚才说{last_user_message} return {messages: [{role: assistant, content: reply}]} graph StateGraph(State) graph.add_node(chat, chat_node) graph.add_edge(START, chat) graph.add_edge(chat, END) checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) config {configurable: {thread_id: user-001}} app.invoke({messages: [HumanMessage(content你好我叫小明)]}, config) app.invoke({messages: [HumanMessage(content我叫什么名字)]}, config)指定同一个thread_id后第二次执行时chat_node能看到第一次写入state[messages]的消息。实际项目中你需要在chat_node里把历史消息交给模型让模型基于完整上下文生成回复。MemorySaver是内存实现服务重启后状态会丢失。如果要在生产环境使用需要替换为基于 Redis 或数据库的 Checkpointer。6.3 消息追加的核心Reducer如果节点每次都直接覆盖state[messages]多轮上下文是存不住的。LangGraph 提供了一种 Reducer 机制在 State 字段上指定合并逻辑from typing import Annotated, TypedDict from langchain_core.messages import HumanMessage def merge_messages(old: list, new: list): return old new class State(TypedDict): messages: Annotated[list, merge_messages]Annotated的第二参数就是这个字段的 Reducer。以后的节点执行时返回值不会覆盖旧数据而是通过merge_messages把新消息追加到旧消息后面。这是 LangGraph 官方推荐的写法因为多轮消息列表本质上应该是追加而不是覆盖。7. 常见问题与排查思路我在实践过程中遇到的大部分问题集中在依赖版本、API 配置、状态写入和图构建这几类。下面整理成表格方便直接对照排查。问题现象可能原因排查方式解决方案ImportError: cannot import name STARTLangGraph 版本过旧pip show langgraph查看版本升级到新版pip install -U langgraph模型 API 报 401 / 403API Key 错误或未加载检查.env是否正确打印os.getenv结果重新配置 API Key确认服务商提供的信息invoke 返回结果没有写入新字段节点返回值与 State 字段不匹配检查节点返回字典的 key确保返回值 key 名称与 State 字段一致多轮对话上下文丢失没有传config或没有配置 Checkpointer检查compile()是否传入checkpointer使用MemorySaver并保证每次 invoke 使用同一个thread_id中文 JSON 解析报错模型输出前后有额外内容打印resp.content原始内容在 Prompt 中强调“只输出 JSON”或使用输出解析器图无法编译节点定义了未连接的边检查add_edge是否引用了不存在的节点确保边的起点和终点都已在add_node中注册节点执行重复或死循环条件边路由函数返回了错误节点名打印路由函数的返回值检查条件边第三个参数映射表是否完整调用工具报错工具函数参数签名与模型输出不匹配打印工具调用参数确认工具节点参数名与函数定义的参数一致遇到报错时第一条原则是先看完整堆栈不要只看最后一行。LangGraph 的报错信息通常包含节点名和当前 State这些信息比报错本身更有价值。8. 工程化最佳实践从“跑通 Demo”到“上线生产”中间还有很多工程细节。下面几条是我认为最重要、但新手最容易忽略的。8.1 节点职责要单一一个节点只做一件事。比如“判断意图”是一个节点“调用天气工具”是另一个节点“汇总回答”是第三个节点。不要为了少写几个函数而把三个逻辑塞进一个节点。节点职责清晰后条件边才能灵活组合出问题时也更容易定位。8.2 State 字段宁少勿多State 是全局共享的字段越多数据流越难追踪。建议按照“输入、中间结果、输出”三类来组织字段。例如只保留user_input、need_tool、tool_result、answer。不要把大段原始日志塞进 State日志应该走 logging 输出到监控系统。8.3 为模型调用设置超时与重试模型 API 有波动生产环境必须控制超时时间。调用模型时建议把timeout设置为合理值并使用重试策略。但要注意不是所有节点都适合自动重试。如果节点内部包含写入数据库等副作用操作重试前要考虑“是否已经执行成功过一次”。8.4 日志、监控与图可视化LangGraph 的图结构天然适合可视化。开发阶段可以打印每一步的 State 变化确认路由是否正确生产阶段建议把每次执行的thread_id、节点名、耗时、模型输出都记录到日志系统。这样用户反馈“回答不对”时你可以直接定位到是哪个节点出了问题。8.5 安全与数据合规在 LangGraph 中用户消息和工具结果都会进入 State可能还会落盘到 Checkpointer。使用消息数据时必须遵守数据安全规范不把密钥写入 State 或日志不把敏感用户数据保存到内存之外的地方如果必须持久化需要明确数据保留期限并做好脱敏。调用外部工具和模型 API 时确保使用合法合规的服务渠道不要通过非正规途径访问服务。8.6 版本锁定的重要性LangGraph 迭代速度较快不同版本之间 API 存在差异。建议在requirements.txt或pyproject.toml中锁定版本范围并在升级前查看官方变更日志。不要盲目升级到最新版否则很容易遇到START导入失败或tools_condition行为不一致这类兼容性问题。9. 总结与后续学习方向这篇教程从 Agent 开发的三个痛点讲起介绍了 LangGraph 的五个核心概念State、Node、Edge、Conditional Edge、Checkpointer。然后通过三个可运行示例覆盖了固定流程、条件分支、工具调用和多轮记忆。最核心的一条主线是不要让模型用 Prompt 控制流程而是用图结构把流程固定下来模型只负责它擅长的那部分——理解和生成。下一步建议你按这样的顺序练习先复现最小示例再给固定流程加一个条件分支然后用 ToolNode 接入一个真实工具最后加上持久化 Checkpointer。每走一步就打印一次 State你会直观感受到“图式 Agent”和“循环式 Agent”在设计上的差异。如果项目已经开始使用 LangGraph优先检查三件事State 字段是否精简、Checkpointer 是否按thread_id隔离用户、模型调用有没有超时和重试。这三件事做完大部分低级问题就不会出现了。未来值得深挖的方向有三个集成 Redis 做长期记忆、多 Agent 协作图、结合 LangSmith 做可观测性。这些都是基于本文基础概念的延伸。先把最小图跑通再逐个深入你会比绝大多数只收藏不练的人更快掌握这套框架。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑