尧图建网站 尧图建网站 YAOTU WEB BUILD 免费咨询
ARTICLE DETAIL

资讯详情

深耕网站建设与建站编程的一线实战洞察。

LangGraph实战:用状态图构建可控的Agent工作流

LangGraph实战:用状态图构建可控的Agent工作流 从今年开始很多团队在落地 LLM Agent 时已经不满足于“调用一次模型返回结果”这种简单交互而是希望让模型能够自主规划、调用工具、多轮反思、按不同分支执行任务。每当这时候LangChain 和 LangGraph 就会同时出现在技术方案里。不少人会把它们当成同一个东西实际在工程上它们是互相配合的两层LangChain 提供模型封装、Prompt 管理、工具抽象LangGraph 则负责把整个 Agent 的流程做成一张可控制、可暂停、可回溯的状态图。这篇文章会从 LangGraph 核心概念讲起通过一个可运行的实战项目把 State、Node、Edge、条件路由、循环控制、子图、MCP 工具接入这些关键点一次讲清楚。无论你是刚接触 LangChain 的新手还是已经被 Agent 流程编排折磨过的后端开发都可以按文章步骤复现整个项目。文章涉及到的代码较多建议打开 IDE 跟着敲一遍。1. LangGraph 是什么先搞清楚它和 LangChain 的关系1.1 从一次性问答到多步 Agent 工作流先回顾一个最简单的 AI 应用场景用户输入一个问题程序把问题拼到 Prompt 里调用大模型把模型返回的文字展示给用户。这个流程用普通 Python 写也能完成不需要任何框架。但真实业务往往是下面这样的用户询问“帮我查一下最近一周的订单总量并和上上周比较生成一份分析报告”。Agent 需要先理解任务拆成“查数据库”“数据对比”“生成报告”等多步。每一步都可能要调用不同工具比如 SQL 查询、代码执行、表格渲染。如果某一步失败Agent 可能还要重试或者换一条路径执行。整个过程中状态数据要在多步之间传递比如用户 ID、查询结果、历史消息。这类需求如果全部用if/else和while手写代码会非常脆弱。因为你不仅要写业务流程还要处理模型输出格式不稳定、工具调用参数错误、多轮状态同步等一堆问题。LangGraph 解决的正是“如何把 Agent 的工作流编排成一张可控的状态图”。专业一点说LangGraph 是一个基于图结构的大模型应用编排框架。它把系统中的每一步操作建模为“节点”把节点之间的流转关系建模为“边”。每一个节点执行完毕后会更新一个全局的State对象然后根据当前状态决定下一步去哪个节点。1.2 LangChain 与 LangGraph 到底有什么区别这是搜索引擎里被问得最多的问题之一。很多人以为 LangGraph 是 LangChain 的升级版或者 LangChain 已经过时了。实际并不是两者的定位完全不同。LangChain 提供的是大模型应用的“零件库”比如ChatPromptTemplate负责管理提示词ChatOpenAI负责封装模型接口Tool负责定义工具描述Retriever负责处理向量检索。你可以把它理解为“函数工具箱”。LangGraph 提供的是“流水线”它不关心你用的是哪家模型也不关心 Tool 内部怎么实现它只关心节点之间的依赖关系、状态如何流转、流程何时结束。用一个表格来对比更直观对比维度LangChainLangGraph核心定位模型与工具的工具箱多智能体工作流编排典型组件Prompt、Model、Tool、RetrieverState、Node、Edge、Checkpointer是否管理流程只做基础串联显式状态图支持分支和循环适合场景简单问答、RAG 工具类应用复杂 Agent、多步骤、人机协同、持久化对话与对方关系LangGraph 可调用 LangChain 组件使用 LangChain 作为模型/工具层在实际项目里LangGraph 节点内部经常会写model.invoke(...)、tool.invoke(...)这种 LangChain 风格的代码。可以这么理解LangChain 负责“干活”LangGraph 负责“指挥谁去干活”。1.3 MCP、Skill 和 LangChain 工具的边界MCPModel Context Protocol是近两年很热的一个词它解决的是“模型如何标准地访问外部数据与工具”的问题。简单理解MCP 定义了一套统一协议让大模型应用可以通过标准客户端连接各种服务端从而获得工具列表、执行工具调用、读取资源。很多开发者会混淆 MCP 和 LangChain Tool原因在于它们最终都能让模型调用外部能力。区别如下LangChain Tool 是框架内部的一种工具抽象它包含名称、描述、参数 Schema 和执行函数。开发者可以手动注册任意函数。MCP 是一种跨框架、跨服务的协议。它强调“服务端把能力暴露出来客户端动态发现并调用”。Skill 更像一种“提示词技能包”它通常由指令文本、示例和策略组成目的是让 Agent 学会某种工作模式。它不一定有可执行代码。在 LangGraph 中MCP 工具等外部能力通常会被适配成 LangChain 的 Tool 格式再注册进节点所以三者并不是互斥关系。后面实战部分会演示这个接入过程。1.4 为什么仍然值得学习 LangGraphLangGraph 的官方文档和视频教程确实很多但大部分内容只停留在“画图解释概念”阶段很少有一个项目把状态管理、条件路由、工具调用、持久化串起来。对于开发者来说真正有价值的是掌握“如何把一个业务需求拆成状态图”然后落成代码。LangGraph 另一个优点在于它天然支持“人在回路”和“持久化”。Agent 在执行过程中可以暂停由人工审核后继续运行进程重启后还能从某个状态节点恢复。这对企业级生产环境非常重要。下面进入环境准备阶段。2. 环境准备安装与最小项目结构2.1 版本与环境要求本文章示例是基于 Python 环境。建议使用 Python 3.10 及以上版本因为新版 LangGraph / LangChain 对高版本 Python 的支持更完善且一些类型语法需要新版本解释器。关于 LangGraph、LangChain、MCP 相关包的版本当前迭代速度很快。不同版本的 API 可能有细微差异比如MemorySaver的导入路径、ToolNode的参数形式。建议不要盲用旧教程里的代码以你安装的官方文档为准。我演示的环境如下你可以按实际项目调整操作系统Windows / macOS / Linux 均可Python3.10包管理pip 或 poetryLLM 接口OpenAI 兼容接口可以用 OpenAI、通义千问、DeepSeek、本地 vLLM 服务等2.2 安装依赖创建一个虚拟环境然后安装核心依赖。python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate pip install langchain langgraph langchain-openai如果要用 MCP 和 LangGraph 集成通常还需要 MCP 适配层。社区常用做法是安装langchain-mcp-adapters然后通过适配器把 MCP Server 暴露的工具变成 LangChain Toolpip install langchain-mcp-adaptersMCP 服务端本身可以使用官方 MCP Python SDK 来开发。安装方式一般是pip install mcp以上包名和安装命令在版本快速迭代期可能变化如果你在安装时遇到命令找不到或包不存在的情况请优先查阅官方 README。2.3 最小项目结构为了让后面的实战更清晰建议先按下面的目录结构组织代码langgraph-mcp-demo/ ├── .env # 存放 API Key 等敏感配置 ├── requirements.txt # 依赖列表 ├── server/ │ └── math_server.py # 一个简单的 MCP Server ├── agent/ │ ├── state.py # 定义 LangGraph State │ ├── nodes.py # 定义节点函数 │ ├── tools.py # 把 MCP 工具注册成 LangChain Tool │ ├── graph.py # 构建并编译 LangGraph │ └── main.py # 入口脚本这种分层的结构在项目变大后优势很明显状态、节点、工具、图构建互相解耦排错时可以单独测试某一个模块。3. 核心概念拆解State、Node、Edge在写完整项目之前必须先把 LangGraph 的几个基本概念彻底搞懂。很多新手代码跑不通不是因为 API 不会写而是没有理解数据在图中是怎么流动的。3.1 State节点之间的唯一通信渠道LangGraph 中的State可以理解为一个“全局上下文对象”。每个节点执行结束后会返回一个字典LangGraph 会把返回值合并到全局 State 中。下一个节点读取 State就能拿到前一个节点的计算结果。State 的类型一般用TypedDict或 PydanticBaseModel定义。这里用TypedDict举例# 文件路径agent/state.py from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_input: str tool_results: dict step_count: int注意messages字段上的Annotated[list, add_messages]。这个写法表示当多个节点都返回messages时不是简单地覆盖而是用add_messages函数把消息追加到列表里。这是 LangGraph 里比较核心的一个机制叫做“消息归约器”。如果不加归约器节点返回同名 key 时后面的值会直接覆盖前面的值。对于消息列表这类需要累加的数据这显然不是我们想要的。3.2 Node真正执行业务逻辑的地方Node 是 LangGraph 中的“执行单元”。它可以是一个普通 Python 函数也可以是任何可调用对象。函数接收的参数是当前全局 State返回值是一个字典表示对 State 的增量更新。# 文件路径agent/nodes.py from agent.state import AgentState def call_model(state: AgentState) - dict: user_input state[user_input] # 实际项目中这里会调用 LLM response f我收到你的输入{user_input} return {messages: [{role: assistant, content: response}], step_count: state[step_count] 1}这里有一个容易被忽略的细节节点函数不要直接修改传入的 state 对象而是通过返回值声明“要对 state 做哪些更新”。这样 LangGraph 可以跟踪每次变更也便于后期实现时间旅行、断点恢复等功能。3.3 Edge定义下一步怎么走Edge 是连接节点的“线”。最简单的边是无条件边表示执行完 A 节点后一定去 B 节点。from langgraph.graph import StateGraph, START, END builder StateGraph(AgentState) builder.add_node(call_model, call_model) builder.add_edge(START, call_model) builder.add_edge(call_model, END) graph builder.compile()START是虚拟入口节点END是虚拟结束节点。上面这段代码表达的意思很直白图从入口开始进入call_model节点执行完毕后直接结束。3.4 条件路由与分支控制conditional_edge 实战无条件的线性执行只适合最简单的流水线真实 Agent 往往需要根据模型输出决定走哪条路。条件边通过add_conditional_edges实现。需求场景模型节点输出一个finish字段当它为True时流程结束当它为False时进入工具调用节点。def should_continue(state: AgentState) - str: if state.get(finish): return end return continue_tool builder.add_node(call_model, call_model) builder.add_node(call_tool, call_tool) builder.add_conditional_edges( call_model, should_continue, { continue_tool: call_tool, end: END, } ) builder.add_edge(call_tool, call_model)这段代码是 Agent 循环的经典写法。模型先决定是否需要调用工具如果需要进入call_tool节点执行工具执行完后再回到call_model让模型观察工具结果。这个“模型—工具—模型”的循环会一直持续直到模型认为任务已完成。条件路由本质上是“在节点执行完后调用一个路由函数根据返回值查表决定下一个节点”。理解这个机制后你就能自由实现多分支、异常重试、人工审核跳转等逻辑。3.5 循环与终止条件Agent 的循环不能无限跑下去。LangGraph 内部有一个recursion_limit参数用来限制一次运行最多执行多少个节点。当执行步数超过限制时会抛出异常。实际项目中除了依赖这个兜底限制还应该在节点函数里增加业务层面的终止条件。比如只允许模型最多调用 5 次工具超过后强制结束if state[step_count] 5: return {finish: True, error: too many steps}不要只依赖框架底层的递归限制否则生产环境很容易出现“模型陷入死循环、任务长时间不结束”的情况。3.6 子图和并行分支让流程可复用当图结构变大后把整张图写在一个文件里几乎没法维护。LangGraph 支持把一张已经compile()的图当作另一个图的普通节点来使用这种图就叫子图Subgraph。子图的典型应用场景是多个业务分支共享同一条“工具调用 模型总结”流程。你可以把这段流程封装成子图然后在父图的不同节点中调用它。LangGraph 还支持并行执行多个节点。当你把多个节点的入边都指定为同一个上游节点时这些节点会被视为可以并行执行的分支。并行分支对于“同时查询订单系统、库存系统、用户系统”这类场景特别有价值。不过要注意并行分支同时更新 State 时容易发生字段覆盖设计 State 时要提前规划好每个分支写入哪些字段。4. 完整实战LangGraph MCP 工具接入的 Agent理论讲再多不如一个能跑通的例子。这一节我们用 LangGraph 构建一个“数学助手 Agent”它通过 MCP Server 暴露两个计算工具然后由模型自动判断是否调用工具最终给出答案。4.1 需求分析我们要做一个 Agent用户可以输入类似这样的问题请计算 23 乘以 17 的结果然后加上 56 再除以 2。按照传统程序实现我们需要写解析逻辑但用 Agent 实现模型负责拆解任务工具负责执行计算LangGraph 负责整个流程控制。为了简化问题我们定义两个工具add两个整数相加multiply两个整数相乘MCP Server 用 Python 的mcpSDK 实现。LangGraph 端通过 MCP 适配层拿到工具列表注册为 LangChain Tool然后构建 Agent 循环。4.2 定义 State 与工具先定义 State。因为 Agent 要与模型多轮交互消息列表用add_messages做追加合并。# 文件路径agent/state.py from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] finish: bool step_count: int再用pydantic定义工具参数结构。这里没有直接使用 LangChain 的tool装饰器而是让 MCP 适配层自动生成工具避免工具定义分散在多个地方。4.3 构建 MCP Server 与工具加载先写一个最简单的 MCP Server。mcpSDK 的 API 在快速迭代下面的写法是当前社区比较常见的风格如果 API 变化了请以官方文档为准。# 文件路径server/math_server.py from mcp.server.fastmcp import FastMCP mcp FastMCP(math-server) mcp.tool() def add(a: int, b: int) - int: 两个数字相加。 return a b mcp.tool() def multiply(a: int, b: int) - int: 两个数字相乘。 return a * b if __name__ __main__: mcp.run(transportstdio)在这个 Server 中add和multiply就是暴露给客户端的能力。MCP 协议会自动把函数名、参数 Schema、描述信息序列化给客户端这样大模型就能“看见”有哪些工具可用。接下来在 LangGraph 侧加载 MCP 工具。当前常用的适配方式是langchain-mcp-adapters大致流程如下# 文件路径agent/tools.py from contextlib import asynccontextmanager from langchain_mcp_adapters.client import MultiServerMCPClient asynccontextmanager async def load_mcp_tools(): async with MultiServerMCPClient( { math: { command: python, args: [server/math_server.py], transport: stdio, } } ) as client: tools await client.get_tools() yield tools这里使用stdio传输方式意思是 MCP Server 通过标准输入输出与客户端通信。生产环境如果服务部署在不同机器上通常会用sse或streamable-http传输方式连接地址变成http://localhost:8000/mcp这种形式。4.4 构建 Agent 工作流LangGraph 中实现 Agent 有两种常见方式。第一种是完全手写节点自己控制“模型→工具→模型”的循环第二种是使用langgraph.prebuilt里的create_react_agent快速搭建。为了让原理更清楚这篇文章使用手写节点的方式。先定义两个节点call_model和call_tool。# 文件路径agent/nodes.py from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolNode from agent.state import AgentState llm ChatOpenAI(modelgpt-4o-mini, temperature0) def call_model(state: AgentState) - dict: messages state[messages] # 绑定工具后模型返回的消息中会带有 tool_calls 信息 response llm.bind_tools(tools).invoke(messages) # 如果模型没有请求调用工具说明它已经得到最终答案 finish not getattr(response, tool_calls, None) return { messages: [response], finish: finish, step_count: state[step_count] 1, } def call_tool(state: AgentState) - dict: tool_node ToolNode(tools) tool_message tool_node.invoke(state[messages]) return {messages: tool_message}上面的代码中有几个点需要解释。第一llm.bind_tools(tools)的作用是告诉模型“当前任务允许使用哪些工具”。模型在推理时如果发现需要计算会在返回的消息里附加tool_calls字段字段中包含了工具名和参数。第二ToolNode会读取当前消息列表中的tool_calls自动执行对应工具并把执行结果封装成ToolMessage返回。第三finish字段用来标记任务是否结束。这个字段会在条件路由中被读取。然后把这些节点组装成图# 文件路径agent/graph.py from langgraph.graph import StateGraph, START, END from agent.state import AgentState from agent.nodes import call_model, call_tool, tools def build_graph(): builder StateGraph(AgentState) builder.add_node(call_model, call_model) builder.add_node(call_tool, call_tool) builder.add_edge(START, call_model) builder.add_conditional_edges( call_model, lambda state: call_tool if not state[finish] else END, { call_tool: call_tool, END: END, }, ) builder.add_edge(call_tool, call_model) return builder.compile()需要注意上面的代码中tools是全局变量实际项目中建议把工具列表作为图构建函数的参数传入避免模块循环依赖。4.5 运行与验证入口脚本如下# 文件路径agent/main.py import asyncio from agent.graph import build_graph from agent.tools import load_mcp_tools async def main(): async with load_mcp_tools() as loaded_tools: global tools tools loaded_tools graph build_graph() result await graph.ainvoke( { messages: [ {role: user, content: 请计算 23 乘以 17然后再加上 56 的结果} ], finish: False, step_count: 0, } ) print(result[messages][-1].content) if __name__ __main__: asyncio.run(main())运行python agent/main.py预期结果应该是模型先调用multiply(23, 17)得到 391再调用add(391, 56)得到 447最后输出完整计算过程和分析。这里有一个细节要注意create_react_agent等高级封装会自己处理工具绑定但手写节点时必须在call_model里调用bind_tools否则模型永远不可能返回tool_calls。4.6 增加记忆与持久化上面实现的 Agent 是无状态的。每次执行都是从 0 开始历史对话不会保留。要让 Agent 支持多轮上下文LangGraph 官方推荐使用 Checkpointer检查点机制。简单理解Checkpointer 会把图每次执行后的状态持久化到存储中。下一次运行时只要传入同一个thread_id就能从上次的状态继续执行。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() def build_graph(): # ... 前面的节点与边定义 ... return builder.compile(checkpointercheckpointer)运行时传入配置config {configurable: {thread_id: user-001}} result await graph.ainvoke(input_state, configconfig)MemorySaver是内存版检查点适合开发和测试。生产环境建议使用 SQLite、PostgreSQL 等持久化存储方案。提供持久化后Agent 才能真正地支持“跨请求恢复对话状态”。5. 常见问题与排查思路LangGraph 的报错信息在国产模型中经常比较抽象很多问题看起来是代码报错实际原因是状态设计不合理或工具调用格式不对。下面梳理几个高频问题。问题现象常见原因解决思路运行超过几十步后报 Recursion Limit 相关错误Agent 陷入工具调用死循环在节点函数中设置 step_count 上限超限强制 finish节点返回的字段没有更新到全局 State返回的 key 在 State 中未定义检查 TypedDict 是否定义了对应 keymessages最后只剩最后一条消息同名 key 覆盖缺少归约器在 State 定义中使用Annotated[list, add_messages]模型始终不调用工具没有调用bind_tools或工具描述不准确检查节点代码是否绑定了 toolsMCP Server 启动成功但客户端拿不到工具传输方式不匹配或 Server 启动路径错误先用命令行测试 MCP Server 的 stdio 通信工具调用报参数解析错误模型生成的参数与工具 Schema 不一致给工具参数写清晰描述统一使用 JSON Schema 类型并行节点同时写入同一 keyState 字段冲突不同分支写入不同字段或用归约器合并LangGraph 报错时建议先看两个地方一是节点返回值里是否包含非法 key二是recursion_limit是否设置得过于激进。大多数运行时报错都能通过打印每次状态的step_count和finish快速定位。6. 工程最佳实践6.1 把 State 当作数据库表来设计State 设计是整个 LangGraph 应用的“数据结构设计”。不要随手往 State 里塞字段。我的经验是每个字段都要有唯一职责能推导出来的数据不要重复存储。比如step_count和messages长度有相关性但step_count用于业务终止条件messages用于模型上下文职责不同所以保留两个字段是合理的。反之如果只是为了展示而存冗余字段就应该删掉。6.2 节点要小、职责要单一一个节点只做一件事。比如“模型调用”节点只负责生成模型回复“工具调用”节点只负责执行工具“结果校验”节点只负责判断是否结束。不要在一个节点里既调用模型又执行多个工具否则出现问题后很难定位。节点命名要可读。LangGraph 的调试面板和日志都会显示节点名使用generate_reply、execute_sql、human_review这样的名字比node1、node2可读性强得多。6.3 终止条件要多层防御Agent 在生产环境最怕失控。建议建立多层终止机制业务规则层明确限定最大工具调用次数、最大 LLM 调用次数。图配置层设置合理的recursion_limit不要为了方便直接设成 100 甚至 1000。超时控制外层函数用asyncio.wait_for或分布式任务框架设置超时时间。安全工具边界对 Agent 可执行的工具做白名单控制尤其是涉及写操作、删除操作、支付操作的接口必须加入人工审批节点。6.4 配置隔离与密钥管理不要在生产代码里硬编码模型名称、API Key、MCP Server 地址。建议用.env文件管理本地配置用部署平台的 Secret 管理生产密钥。MCP Server 的地址也应放在配置中心避免每次发布代码都要改地址。6.5 日志与可观测性Agent 应用比普通接口难调试因为你很难复现某次模型输出的tool_calls。建议在关键节点输出结构化日志至少包含节点名称当前 step_count输入 State 的关键字段摘要模型返回的 tool_calls 原始内容工具执行结果将这些日志接入 ELK 或 Jaeger 等链路追踪系统出现线上问题时可以快速回放一次 Agent 任务执行过程。6.6 MCP 接入的安全边界MCP 让 Agent 获得外部工具能力的同时也放大了安全风险。接入任何 MCP Server 前要确认这是可信来源。生产环境建议遵循最小权限原则不要给 Agent 一个拥有数据库所有权限的连接串而应创建一个只读账户或者只允许白名单 SQL 模板不要给 Agent 直接执行任意 shell 命令的能力只暴露封装后的目标函数。7. 总结与学习路线通过这篇文章你应该已经掌握 LangGraph 的完整学习路径从理解 State、Node、Edge 三大核心概念到用条件路由实现模型与工具的循环交互再到通过 MCP 把外部服务接入 Agent。LangGraph 看起来 API 不多但它对流程控制的设计思路是通用的。很多企业级 Agent 框架比如 Dify、Coze 的可视化编排底层也是类似的“节点 边 状态”模型。下一步可以按这个顺序继续深入先做一个小项目用 LangGraph 实现一个带记忆的客服机器人重点练习 Checkpointer。再做中等复杂度项目实现一个支持多工具调用的数据分析 Agent包含 SQL 查询、代码执行、图表生成等节点。然后尝试多人协作场景通过interrupt和Command实现人工审核后再继续执行。最后再研究更复杂的多 Agent 协作比如管理节点、执行节点分离或使用子图组织大型任务。学习 LangGraph 最大的捷径不是看更多的视频而是亲手修一个RecursionLimit报错。当你把状态流转的每一环都摸清楚后就能真正理解 Agent 的工程化问题了。如果这篇教程对你搭建 Agent 流程有帮助可以先收藏备用动手跑通后你会对接下来的进阶学习更有底气。
返回列表