企业官网建设流程全解析

LangGraph 智能体实战：从环境搭建到条件路由与记忆机制，这一篇讲透

如果你正在学习 Agent 开发，大概率已经听过 LangGraph，甚至已经看过几篇教程。但多数教程只停留在“画个流程图”“跑个 demo”的阶段，真正到了自己写业务逻辑时，节点怎么设计、状态怎么传递、条件分支怎么跳、记忆怎么保存、工具调用怎么兜底，这些问题全冒出来了。

这篇文章不打算重复那些官方案例的翻译版。我会从工程落地角度，把 LangGraph 从环境准备、核心概念、第一个可运行示例，到条件路由、子图、记忆机制、Human-in-the-loop 真实场景完整拆一遍。读完之后，你能独立搭建一个带工具调用和对话记忆的 LangGraph Agent，并且知道每一步为什么这样设计。

先说一个明确判断：LangGraph 真正降低的不是“写一个 AI 应用”的门槛，而是“把 AI 应用做成有状态、可控制、可维护的工程系统”的门槛。它和普通 LangChain chain 的本质区别，在于它把 Agent 的整个执行过程显式建模成了图结构，让开发者对流程有完全的掌控力。这一点理解到位了，后面的代码才不算白写。

1. 为什么需要 LangGraph：从 Chain 到 Graph 的必然演进

很多刚接触 LangChain 的开发者会有个困惑：LangChain 已经能调模型、调工具、做 RAG 了，为什么还需要 LangGraph？

答案是：LangChain 的 Chain 是线性管道，而真实 Agent 是带分支、循环、状态和人工干预的复杂流程。

举个例子，一个最简单的客服 Agent 流程可能是：

1. 接收用户提问。
2. 判断是否需要查询订单系统。
3. 如果需要，调用订单查询工具。
4. 根据工具返回结果生成回答。
5. 如果回答过程中用户又追问了，可能需要回到第 2 步再次判断。

这种流程用 Chain 表达会非常别扭。因为 Chain 的假设是“输入一次，走完一条固定管道，输出结果”。而 Agent 的实际执行是“走一步，看一步，根据当前状态决定下一步去哪”。

LangGraph 把这个过程建模成一张有向图：

• 每个处理步骤是一个Node（节点）。
• 节点之间通过Edge（边）连接。
• 边的走向可以由条件函数（conditional_edge）动态决定。
• 整张图共享一个State（状态对象），任何节点都可以读写这个状态。

这意味着，你可以精确控制 Agent 的每一步：哪些步骤必须顺序执行，哪些步骤需要根据模型输出决定走向，哪些步骤允许并行，哪些步骤需要停下来等用户确认。

另外，LangGraph 早期版本的核心定位是“低阶编排”，它把 LangChain Agent 那些封装好的高层接口拆开，暴露底层控制能力。后来 LangChain 官方明确推荐，新项目优先使用 LangGraph 进行 Agent 开发，甚至可以完全不依赖 LangChain 的 Agent 封装，只用 LangChain 的模型封装和工具封装。

一句话总结：

如果你只需要一个固定流程的提示词管道，用 LangChain Chain 就行。如果你需要一个可以自主决策、多轮交互、可人工干预的 Agent，用 LangGraph。

2. LangGraph 的核心概念与工作原理

在写代码之前，先把 LangGraph 的五个核心概念彻底理清。不理解这五个概念，代码就是死记硬背。

2.1 State：全局共享状态

State 是 LangGraph 的灵魂。它定义了整个图在任意时刻的数据快照。所有节点都接收当前 State 作为输入，返回值会合并进 State，供下一个节点使用。

State 通常是一个 TypedDict 或 Pydantic 模型。每个字段可以配置不同的归约操作，比如覆盖、追加、合并。

from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): messages: Annotated[List[dict], operator.add] # 消息列表，新消息追加 next_step: str # 下一步走向，默认覆盖 tool_result: str # 工具返回结果

这里operator.add是一个归约器（reducer）。当节点返回新的messages时，LangGraph 会执行两段列表相加而不是直接覆盖。这个设计非常关键，它让多个节点可以往同一个状态字段追加内容，而不会互相覆盖。

2.2 Node：处理节点

Node 就是一个普通的 Python 函数（也可以是异步函数），接收 State 参数，返回一个字典，字典的键是要更新的状态字段。

def call_model(state: AgentState): # 调用大模型的逻辑 response = llm.invoke(state["messages"]) return {"messages": [{"role": "assistant", "content": response.content}]}

Node 里可以写任何逻辑：调用大模型、调用工具、查数据库、写日志、做业务校验。LangGraph 不限制节点内部做什么，它只负责调度和状态流转。

2.3 Edge：静态边和条件边

Edge 分两种。

静态边（Edge）：节点 A 执行完，无条件走向节点 B。

graph.add_edge("node_a", "node_b")

条件边（Conditional Edge）：节点 A 执行完后，根据一个路由函数的返回值，决定走向哪个节点。这是 LangGraph 最核心的亮点之一。

def route_after_tool(state: AgentState): if state["next_step"] == "call_model": return "call_model" else: return "end" graph.add_conditional_edges( "tool_node", route_after_tool, {"call_model": "call_model", "end": END} )

条件边让图具备了“循环”能力：当 Agent 判断还需要继续调用工具时，可以沿着条件边回到前面的节点，而不是线性结束。

2.4 Checkpointer：记忆与断点

Checkpointer 是 LangGraph 的持久化层。它把每一步执行后的 State 快照保存到存储后端，让 Agent 具备多轮对话记忆和断点恢复能力。

如果你不配置 Checkpointer，LangGraph 每次执行完图之后 State 就丢了，下一次对话从零开始。这是很多初学 LangGraph 的开发者容易忽略的点。

LangGraph 官方提供基于 SQLite 的SqliteSaver和基于内存的MemorySaver。生产环境可以对接 Redis 或 Postgres。

2.5 编译与执行

LangGraph 图需要先编译（.compile()）再执行。编译后得到的是一个CompiledGraph对象，可以直接调用invoke()或stream()方法。

app = graph.compile(checkpointer=checkpointer) result = app.invoke( {"messages": [{"role": "user", "content": "帮我查一下订单状态"}]}, config={"configurable": {"thread_id": "user-001"}} )

thread_id是对话会话的标识，同一个thread_id的多轮调用会共享 Checkpointer 里保存的状态。

3. 环境准备与安装配置

LangGraph 的环境准备并不复杂，但有几个版本匹配的细节值得注意。如果你在安装阶段就把环境配错了，后面跑示例时会遇到莫名其妙的报错。

3.1 环境要求

• Python 3.9 及以上版本（3.10、3.11 更稳定）。
• 操作系统：Windows、macOS、Linux 均可。
• 包管理工具：推荐使用pip或poetry，不要直接在系统全局环境里装，建议使用虚拟环境。

3.2 创建虚拟环境

以当前教程使用的 Python 虚拟环境为例：

python -m venv langgraph-env source langgraph-env/bin/activate # Windows 下执行 langgraph-env\Scripts\activate

3.3 安装核心依赖

pip install --upgrade langgraph langchain langchain-openai

如果没有 API Key，想先本地体验，可以安装 Ollama 和对应依赖：

pip install ollama

注意：langgraph、langchain、langchain-openai这三个包的版本兼容性直接影响运行结果。推荐的稳妥做法是安装最新稳定版，如果出现 API 不兼容报错，优先检查langgraph与langchain-core的版本匹配关系。

3.4 关于 MCP 的补充说明

MCP（Model Context Protocol）最近在 Agent 生态中非常火，它的目标是统一大模型与外部工具、数据源的接入协议。LangGraph 的 Agent 在工程实践中通常会结合 MCP Server 来接入数据库、代码仓库、设计稿等外部资源。

但是请注意：MCP 是工具接入层协议，LangGraph 是编排层框架，两者解决的问题不冲突。更准确的理解是，通过 MCP Server 暴露工具，LangGraph 负责将这些工具编排进 Agent 的决策流程。后面会有具体示例。

4. 核心组件拆解：Node、State、Edge 的工程语义

把概念映射到工程语义，你才能真正理解为什么 LangGraph 这么设计。

4.1 Node 的工程语义

在真实的智能体项目里，Node 通常对应一个“可复用的处理单元”。一个常见的划分方式：

• call_model：负责调用大模型，生成回复或决策。
• call_tool：负责执行具体工具，比如查数据库、调 API。
• check_hallucination：负责校验模型输出是否可靠。
• save_to_db：负责把结果持久化。

这种划分的好处是，每个 Node 都可以单独测试、单独替换、单独复用。比如你想把模型从 GPT-4 换成本地模型，只需要改call_model这一个 Node 的内部实现，图结构完全不用动。

4.2 State 的工程语义

State 在设计时需要区分两类信息：

• 对话数据：用户消息、AI 消息、工具消息。这类数据通常是追加式的，用operator.add。
• 控制数据：当前步骤标识、重试次数、错误信息、分支标记。这类数据通常是覆盖式的，用默认 reducer。

如果在设计 State 时混淆了这两类字段，后续调试会陷入“不知道状态为什么变成这样了”的困境。建议每个 Node 里只修改自己负责的字段，并在节点开头打印关键状态，方便追踪。

4.3 Edge 的工程语义

Edge 的最重要价值是显式化流程控制。相比让模型自由发挥（纯 ReAct），通过条件边把一些“固定规则”和“模型决策”分开，可以让系统更稳定：

• 固定规则走静态边，比如“工具执行完必须回到模型”。
• 模型决策走条件边，比如“模型判断是否继续调用工具”。
• 异常分支必须走条件边，比如“工具调用报错时重试还是结束”。

这个“规则与模型分离”的设计思路，是 LangGraph Agent 在生产环境中比纯 AutoGPT 方案更可靠的重要原因。

5. 第一个 LangGraph 示例：带工具调用的智能客服 Agent

现在开始写真实的代码。这里用一个“智能客服 Agent”作为示例，它能根据用户问题决定是否调用订单查询工具，然后基于工具结果生成回答。

5.1 定义 State

from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): # 对话消息列表，新消息自动追加 messages: Annotated[List[dict], operator.add] # 是否需要调用工具，控制字段，默认覆盖 should_call_tool: str # 工具返回结果，默认覆盖 tool_result: str

5.2 定义工具

工具函数用@tool装饰器包装，便于 LangChain 自动生成工具描述和参数 schema。

from langchain_core.tools import tool @tool def query_orders(user_id: str) -> str: """根据用户ID查询订单信息""" # 这里是模拟实现，真实项目中换成数据库查询即可 orders = { "1001": "订单A：已发货，预计明天送达", "1002": "订单B：待付款", } return orders.get(user_id, "未查询到订单信息")

5.3 定义 Node

三个核心 Node：

• call_model：调用大模型，让它判断下一步动作（直接回答还是调用工具）。
• execute_tool：执行工具调用。
• respond：基于工具结果生成最终回答。

from langchain_openai import ChatOpenAI # 这里使用 OpenAI 兼容接口，你可以替换成任何 OpenAI 兼容服务 llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) def call_model(state: AgentState): last_message = state["messages"][-1]["content"] # 简单判断：如果消息中包含“订单”，则标记需要调用工具 # 真实项目中应让模型判断，这里为了演示条件边，先简化 if "订单" in last_message: should_call_tool = "yes" else: should_call_tool = "no" # 仍然调用一次模型，生成中间回复 response = llm.invoke(state["messages"]) return { "messages": [{"role": "assistant", "content": response.content}], "should_call_tool": should_call_tool, } def execute_tool(state: AgentState): """执行订单查询工具""" # 生产环境应从用户资料中取 user_id，这里写死演示 result = query_orders.invoke({"user_id": "1001"}) return { "tool_result": result, "should_call_tool": "no", # 执行完工具后，重置标记 } def respond(state: AgentState): """基于工具结果生成最终回复""" tool_result = state.get("tool_result", "") if tool_result: answer = f"根据系统查询结果：{tool_result}" else: answer = "抱歉，我暂时无法处理这个问题，请稍后再试。" return {"messages": [{"role": "assistant", "content": answer}]}

5.4 定义条件路由

这是 LangGraph 条件边的核心逻辑：

from langgraph.graph import StateGraph, START, END def route_after_model(state: AgentState): """根据模型判断结果路由到工具节点或回答节点""" if state["should_call_tool"] == "yes": return "execute_tool" return "respond"

5.5 构建并编译图

# 初始化图 graph = StateGraph(AgentState) # 添加节点 graph.add_node("call_model", call_model) graph.add_node("execute_tool", execute_tool) graph.add_node("respond", respond) # 添加入口边 graph.add_edge(START, "call_model") # 添加条件边 graph.add_conditional_edges( "call_model", route_after_model, { "execute_tool": "execute_tool", "respond": "respond", } ) # 工具执行完后回到模型，让模型基于工具结果再组织回答 graph.add_edge("execute_tool", "call_model") graph.add_edge("respond", END) # 编译图 app = graph.compile()

5.6 运行并验证

result = app.invoke({ "messages": [{"role": "user", "content": "你好，帮我查一下订单状态"}], }) for msg in result["messages"]: print(f"{msg['role']}: {msg['content']}")

预期执行流程是：

1. call_model判断用户输入包含“订单”，将should_call_tool设为"yes"。
2. 条件边路由到execute_tool。
3. execute_tool执行查询工具，把结果写入tool_result。
4. 静态边回到call_model。
5. 此时should_call_tool已经是"no"，条件边路由到respond。
6. 最终回答来自工具查询结果。

这个示例虽然简单，但它已经包含了 Agent 的基本骨架：模型决策、工具调用、条件路由、循环回退。

6. 进阶实战：给 Agent 加上记忆和人类确认环节

上面的示例有一个明显缺陷：它没有记忆。用户第二次说“那发货地址呢”，Agent 根本不知道用户是谁，也不知道之前查过哪笔订单。解决这个问题需要引入 Checkpointer 和thread_id。

6.1 配置 MemorySaver（内存级持久化）

from langgraph.checkpoint.memory import MemorySaver # 使用内存级 Checkpointer memory = MemorySaver() app = graph.compile(checkpointer=memory)

6.2 带 thread_id 的多轮调用

config = {"configurable": {"thread_id": "customer-001"}} # 第一轮 result1 = app.invoke( {"messages": [{"role": "user", "content": "查一下订单状态"}]}, config=config ) # 第二轮，同一 thread_id result2 = app.invoke( {"messages": [{"role": "user", "content": "再说一下发货时间"}]}, config=config )

加了 Checkpointer 之后，LangGraph 会在每次节点执行后保存 State 快照。第二轮调用时，模型能读到第一轮的历史消息，这是多轮对话记忆的基础。

6.3 Human-in-the-loop：打断和恢复

Human-in-the-loop（人类介入）是生产级 Agent 的必备能力。典型场景是：工具要执行高风险操作（比如退款、删数据、发邮件）前，需要暂停让用户确认。

LangGraph 里可以用interrupt_before实现：

app = graph.compile( checkpointer=memory, interrupt_before=["execute_tool"], # 执行工具前打断 )

运行时会发现，程序停在execute_tool之前，不会继续往下走。此时你可以打印当前状态给用户确认，然后调用invoke(None, config=config)从断点处继续执行。

# 第一次调用，会在 execute_tool 前被打断 app.invoke( {"messages": [{"role": "user", "content": "查一下订单状态"}]}, config=config ) # 打印当前状态，人工确认 current_state = app.get_state(config) print("待执行的操作：", current_state.next) # 确认通过后，继续执行 app.invoke(None, config=config)

这个机制在真实项目中非常实用。比如销售智能体要执行“发送营销邮件”或“创建客户订单”之前，都需要人工确认。

6.4 子图拆分

当一个 Agent 的节点数量超过 10 个，图结构会变得难以维护。LangGraph 支持子图（Subgraph），可以把一组相关节点封装成一个子图，作为主图的一个节点。

一个典型场景：把“订单查询”的完整流程（查询、校验、格式化）封装成子图，然后主图只调用一个order_process节点。

def build_order_subgraph() -> StateGraph: # 这里定义子图内部节点和边 subgraph = StateGraph(AgentState) subgraph.add_node("query", execute_tool) subgraph.add_node("format", respond) subgraph.add_edge("query", "format") subgraph.add_edge("format", END) return subgraph.compile() # 在主图中添加子图作为节点 app = graph.compile(checkpointer=memory)

子图的价值是模块化。多个 Agent 共享同一个订单处理逻辑时，只需要复用这个子图即可。

7. LangGraph 与 MCP 的工程配合

7.1 MCP 是什么

MCP（Model Context Protocol）是一套开放的协议，用于让 AI 应用连接外部工具、数据源、文件系统等。你可以把它理解成“AI 应用时代的 USB-C 接口”。

7.2 LangGraph Agent 接入 MCP 服务器

在生产环境中，LangGraph 的 Node 内部可以调用 MCP Client 去连接远程的 MCP Server。例如，连接一个提供订单查询能力的 MCP Server。

关键代码示意：

import json def call_mcp_tool(state: AgentState): """Node 内部调用 MCP Server 暴露的工具""" # 这部分是伪代码，MCP SDK 的使用方式以官方文档为准 # client = MCPClient("http://localhost:8000/mcp") # result = client.call_tool("query_orders", {"user_id": "1001"}) result = {"status": "success", "data": "订单A：已发货"} return { "tool_result": json.dumps(result, ensure_ascii=False), "should_call_tool": "no" }

MCP 和 LangGraph 不冲突。LangGraph 更关注“流程怎么编”，MCP 更关注“工具怎么接”。一个完整的 Agent 研发链路是：MCP Server 负责统一暴露企业内部工具，LangGraph 负责这些工具的调用时序、状态管理和异常补偿。

8. 常见问题与排查思路

9. 最佳实践与工程建议

9.1 状态设计原则

• 对话数据用追加式 reducer（如operator.add）。
• 控制数据用覆盖式 reducer。
• 避免在 State 中存放大数据对象，比如完整文件内容，建议存引用或摘要。
• 将“用户 ID”“会话 ID”等上下文信息放在 config 的configurable中，而不是塞进 State。

9.2 节点设计原则

• 每个 Node 只做一件事。call_tool不要顺便写日志入库，拆成独立节点。
• 节点函数保持无副作用（幂等），便于重放和调试。
• 所有 Node 内部错误都应该被捕获，并通过控制字段路由到异常处理分支，而不是直接抛异常导致整个图崩溃。

9.3 工具调用安全边界

• 工具的参数校验、权限校验必须在工具内部完成，不能只靠模型自觉。
• 高风险工具（删除、退款、发送消息）必须配合interrupt_before实现人工确认。
• 工具调用要设计超时和重试机制，防止外部服务不可用时拖死整个 Agent。
• 日志中禁止打印用户的敏感字段，如密码、Token、身份证号。

9.4 日志与可观测性

LangGraph 的stream()方法可以逐步输出每个节点的输入输出，这在调试时非常有用。

for chunk in app.stream( {"messages": [{"role": "user", "content": "查订单"}]}, config=config, stream_mode="updates" ): print(chunk) # 每个节点的状态更新 print("-" * 50)

生产环境建议把stream_mode="updates"的输出接入日志系统，这样每次 Agent 执行的完整轨迹都能被还原。

10. 总结与后续学习规划

这篇文章从 LangGraph 的定位讲起，带你理清了 LangChain 与 LangGraph 的本质区别；接着拆解了 State、Node、Edge、Checkpointer 等核心概念；然后通过“带工具调用的客服 Agent”示例，跑通了第一版 LangGraph 应用；最后补充了记忆、人工确认、子图、MCP 配合等进阶能力。

现在你可以试着做一个完整的练习：把这个客服 Agent 扩展成“支持多用户、支持多个工具（查订单、查物流、查退款）、带人工确认、带日志持久化”的生产级智能体。这个过程会逼你把本文中提到的知识点全部串起来。

下一步值得深入的方向有三个：

1. Agents 抽象层：LangGraph 在langchain.agents里提供了高层封装，可以大幅简化代码。
2. 并行与 Fan-out 模式：LangGraph 支持一个节点并行分发到多个分支，最后汇总，这是复杂 Agent 的常见模式。
3. 与 RAG 结合：把 RAG 检索作为一个工具节点接入 LangGraph，构建“能查文档的 Agent”，这是目前企业落地最多的形态。

建议先把文中的示例代码在自己的环境里完整跑通，然后在 State 里增加一个retry_count字段，自己实现一版带重试机制的工具调用流程。这一步做完，你对 LangGraph 的理解就不是看过教程的程度了，而是真正能上手写项目的程度。