基于LangGraph 实现一个具备自我路径规划及结果验证的 Agent
·
目录
1. LangGraph 核心架构概览
1.1 状态图（StateGraph）基础
LangGraph 的核心是 StateGraph 类，它是一个构建有状态、基于事件的应用程序的框架。图由以下核心组件组成：
| 组件 | 说明 | 核心 API |
|---|---|---|
| State（状态） | 节点间共享的数据结构，使用 Annotation.Root 或 TypedDict 定义 |
Annotation.Root({...}) |
| Node（节点） | 执行具体逻辑的函数，签名通常为 (state) => Partial<State> |
add_node(name, action) |
| Edge（边） | 固定流转路径，无条件从 A 到 B | add_edge(start, end) |
| Conditional Edge（条件边） | 动态路由，由路由函数决定下一节点 | add_conditional_edges(source, path, pathMap) |
| Checkpointer | 状态持久化，支持断点续传 | compile(checkpointer=...) |
官方 API 参考：
-
StateGraph类文档：StateGraph | langgraph | LangChain Reference -
LangGraph.js API Reference：StateGraph | LangGraph.js API Reference
-
图 API 概述（中文）：Graph API概述 | LangChain 中文文档
1.2 核心 API 签名
# StateGraph 构造函数
new StateGraph(
state: AnnotationRoot<SD>,
options?: {
context?: C | AnnotationRoot<...>;
input?: I | AnnotationRoot<...>;
output?: O | AnnotationRoot<...>;
}
)
# 添加节点
addNode(key: string, action: NodeAction): StateGraph
# 添加条件边（核心路由机制）
addConditionalEdges(
source: string, # 源节点名称
path: RunnableLike, # 路由决策函数：接收 state，返回路由标签
pathMap?: Record<string, string> | string[] # 路径映射表
)
# 编译图
compile(
checkpointer?: BaseCheckpointSaver,
interruptBefore?: "*" | string[],
interruptAfter?: "*" | string[]
): CompiledStateGraph
出处：LangGraph.js API Reference — StateGraph 类详细签名 StateGraph | LangGraph.js API Reference
1.3 状态管理的关键设计
LangGraph 使用 Pregel 架构支持状态持久化，每个状态键可配置 reducer 函数用于聚合来自多个节点的值：
from langgraph.graph.message import add_messages class AgentState(TypedDict): # add_messages 会自动处理消息的追加，而非覆盖 messages: Annotated[List[BaseMessage], add_messages] iterations: int # 重试计数器，防止死循环 is_valid: bool # 校验标志位
出处：基于 LangGraph 构建企业级 Agent 技术文章 构建具备 Cyclic Loop（循环反思） 与 Self-Correction（自我修正） 能力的企业级 Agent-CSDN博客
2. 长短期记忆机制（Memory Systems）
LangGraph 采用双层记忆架构模拟人类记忆的层次结构：短期记忆（Short-Term Memory, STM）负责单轮/多轮对话上下文，长期记忆（Long-Term Memory, LTM）负责跨会话的用户画像与知识持久化。两者协同工作，使 Agent 既能理解当前对话的上下文，又能回忆用户的历史偏好和已知事实。
出处：LangGraph 记忆机制深度解析 — 短期与长期记忆的实现 LangGraph State记忆机制深度解析：短期与长期记忆的实现原理与实战-CSDN博客
2.1 双层记忆架构总览
LangGraph State 核心 ├── 短期记忆层 - Checkpointer │ ├── thread-scoped（线程作用域，按 thread_id 隔离） │ ├── 每个 Superstep 自动保存状态快照 │ ├── 支持中断恢复 / 时间旅行（time travel） │ └── 存储：对话历史、中间结果、临时变量 ├── 长期记忆层 - Store │ ├── 跨线程共享（namespace 组织） │ ├── 显式读写操作（put / get / search） │ ├── 语义搜索 / 向量检索（需配置 embedding） │ └── 存储：用户偏好、用户画像、知识库、事实三元组 └── State Graph（状态流转） ├── 节点 / 边执行 └── 消息传递
| 对比维度 | 短期记忆（Checkpointer） | 长期记忆（Store） |
|---|---|---|
| 核心组件 | Checkpointer |
BaseStore |
| 作用域标识 | thread_id |
namespace（通常含 user_id） |
| 操作方式 | 自动（每个 Superstep 触发） | 显式（put / get / search） |
| 存储内容 | 状态通道值（对话历史、变量、中间结果） | JSON 文档（用户偏好、知识库） |
| 生产后端 | PostgresSaver / MongoDBSaver / RedisSaver |
PostgresStore / RedisStore |
| 检索能力 | 按 thread_id 精确加载 |
精确匹配 + 语义向量搜索 |
| 生命周期 | 线程绑定 | 独立持久化，跨会话共享 |
| 本质问题 | "这次对话我们说了什么" | "关于这个用户我们都知道什么" |
出处：LangGraph 长期记忆（Cross-Session Memory Store） LangGraph教程（三）：记忆组件memory_langgraph memory-CSDN博客
2.2 短期记忆（Short-Term Memory）
2.2.1 核心机制：Checkpointer
Checkpointer 是 LangGraph 的自动化状态持久化机制。每当图执行一步（调用一个节点函数），系统会自动将当前完整状态保存为"检查点"（Checkpoint）。这使得 Agent 能够：
-
多轮对话：在同一线程内保持上下文连续性
-
断点续传：执行中断后从上次保存点恢复
-
时间旅行：回溯到任意历史状态重新执行
短期记忆执行流程： 用户输入 → 图执行 Step → 有 Checkpointer? ├── 是 → 自动保存状态快照 → 持久化到存储 → 继续下一步 └── 否 → 状态丢失（无法多轮对话）
出处：LangGraph 记忆机制深度解析 LangGraph State记忆机制深度解析：短期与长期记忆的实现原理与实战-CSDN博客
2.2.2 thread_id：会话隔离钥匙
thread_id 是短期记忆的核心隔离标识。不同的 thread_id 对应完全独立的对话上下文：
# 通过 configurable 字典传入 thread_id
config = {"configurable": {"thread_id": "user_123_session_001"}}
# 线程隔离示意：
# thread_id: user_001 → 状态1（对话历史A）【完全隔离】
# thread_id: user_002 → 状态2（对话历史B）【完全隔离】
出处：LangGraph 添加记忆官方教程 添加记忆 - LangChain 教程
2.2.3 Checkpointer 实现矩阵
| 实现类 | 存储介质 | 适用场景 | 安装命令 |
|---|---|---|---|
InMemorySaver |
内存 | 开发测试、单元测试 | 内置 |
SqliteSaver |
SQLite 数据库 | 小规模部署 | pip install langgraph |
PostgresSaver |
PostgreSQL | 生产环境、高可用 | pip install langgraph-checkpoint-postgres |
MongoDBSaver |
MongoDB | 文档型存储场景 | pip install langgraph-checkpoint-mongodb |
RedisSaver |
Redis 集群 | 分布式系统、低延迟 | pip install langgraph-checkpoint-redis |
出处：LangGraph 记忆机制深度解析 — Checkpointer 存储后端矩阵 LangGraph State记忆机制深度解析：短期与长期记忆的实现原理与实战-CSDN博客
2.2.4 基础用法：InMemorySaver
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState
# 1. 创建内存检查点记录器
checkpointer = InMemorySaver()
# 2. 编译图时启用 checkpointer
builder = StateGraph(MessagesState)
# ... 添加节点和边 ...
graph = builder.compile(checkpointer=checkpointer)
# 3. 第一次对话 - 指定 thread_id
config = {"configurable": {"thread_id": "user_123"}}
graph.invoke(
{"messages": [{"role": "user", "content": "我叫张三"}]},
config
)
# 4. 第二次对话 - 同一个 thread_id，可以访问之前的消息
graph.invoke(
{"messages": [{"role": "user", "content": "我刚才说我叫什么？"}]},
config
)
# Agent 可以回答："你刚才说你叫张三"
出处：LangGraph 添加记忆官方教程 添加记忆 - LangChain 教程
2.2.5 生产环境用法：PostgresSaver
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import StateGraph, MessagesState, START
DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
# checkpointer.setup() # 首次使用需要初始化表结构
def call_model(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "1"}}
for chunk in graph.stream(
{"messages": [{"role": "user", "content": "hi! I'm bob"}]},
config,
stream_mode="values"
):
chunk["messages"][-1].pretty_print()
异步版本：
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer: # await checkpointer.setup() # ... 异步节点定义和图编译 ... async for chunk in graph.astream(...): chunk["messages"][-1].pretty_print()
出处：LangGraph 添加记忆官方教程 — Postgres 完整示例 添加记忆 - LangChain 教程
2.2.6 在子图（Subgraphs）中使用短期记忆
如果父图包含子图，只需在编译父图时提供 checkpointer，LangGraph 将自动传播到所有子图：
from langgraph.checkpoint.memory import InMemorySaver
# 子图
subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node)
subgraph = subgraph_builder.compile()
# 父图
builder = StateGraph(State)
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
如果希望子图拥有独立的记忆（如在多智能体系统中）：
subgraph_builder = StateGraph(...) subgraph = subgraph_builder.compile(checkpointer=True) # 子图独立 checkpointer
出处：LangGraph 添加记忆官方教程 — 子图记忆传播 添加记忆 - LangChain 教程
2.2.7 消息历史的 Reducer 处理
短期记忆中消息列表的追加不是覆盖，而是通过 Reducer 函数实现的：
from typing import Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): # add_messages 会自动处理消息的追加合并，而非覆盖 messages: Annotated[list, add_messages] iterations: int
这意味着每次调用 graph.invoke() 传入新消息时，新消息会被追加到现有消息列表中，而不是替换整个列表。
出处：LangGraph 记忆机制深度解析 — Reducer 机制 LangGraph State记忆机制深度解析：短期与长期记忆的实现原理与实战-CSDN博客
2.3 长期记忆（Long-Term Memory）
2.3.1 核心机制：Store API
Store 是 LangGraph 暴露给图节点和工具的键值数据库，操作是显式和主动的。与 Checkpointer 的自动保存不同，Store 需要开发者在节点中显式调用 put() / get() / search() 进行读写。
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# 显式写入
store.put(("users", "user_123", "preferences"), "theme", "dark")
# 显式读取
value = store.get(("users", "user_123", "preferences"), "theme")
print(value.value) # "dark"
BaseStore 接口骨架：
class BaseStore(ABC): @abstractmethod def batch(self, ops: Iterable[Op]) -> List[Result]: ... # 语法糖，最终都调用 batch def get(self, namespace, key): ... def put(self, namespace, key, value, **kw): ... def search(self, namespace, query="", **kw): ...
出处：LangGraph 长期记忆（Cross-Session Memory Store） LangGraph教程（三）：记忆组件memory_langgraph memory-CSDN博客
2.3.2 Namespace：分层目录结构
Namespace 是 Store 的核心组织单元，使用元组表示层次化路径：
# 用户数据存储
namespace = ("users", "user_123", "preferences") # 用户偏好
namespace = ("users", "user_123", "history") # 用户历史
# 应用数据存储
namespace = ("app", "config", "version") # 应用配置
namespace = ("app", "knowledge", "faq") # 知识库 FAQ
# 组织级数据存储
namespace = ("orgs", "org_456", "policies") # 组织策略
命名空间树形结构：
根命名空间 ├── users │ ├── user_123 │ │ ├── preferences │ │ ├── history │ │ └── settings │ └── user_456 ├── app │ ├── config │ └── knowledge └── orgs └── org_456 ├── members └── policies
出处：LangGraph 记忆机制深度解析 — Namespace 设计 LangGraph State记忆机制深度解析：短期与长期记忆的实现原理与实战-CSDN博客
2.3.3 在节点内读写长期记忆
节点函数通过参数注入接收 store 实例：
from langgraph.store.base import BaseStore
from langchain_core.runnables import RunnableConfig
def call_model(
state: MessagesState, # 当前状态（自动注入）
config: RunnableConfig, # 运行时配置（自动注入）
*, # 关键字参数分隔
store: BaseStore, # Store 实例（自动注入）
):
user_id = config["configurable"]["user_id"]
namespace = ("memories", user_id)
# 读取长期记忆
memories = store.search(namespace, query=str(state["messages"][-1].content))
info = "\n".join([d.value["data"] for d in memories])
# 根据记忆构建系统提示
system_msg = f"You are a helpful assistant. User info: {info}"
# 如需更新记忆
if "remember" in state["messages"][-1].content.lower():
store.put(namespace, str(uuid.uuid4()), {"data": "新记忆内容"})
response = model.invoke([{"role": "system", "content": system_msg}] + state["messages"])
return {"messages": response}
官方推荐的三步曲范式：
-
声明 Store + 编译带 store 的图：
graph.compile(checkpointer=..., store=store) -
在节点里注入
store与config：节点函数签名中使用*, store: BaseStore -
调用时给
user_id：config = {"configurable": {"user_id": "u123"}}
出处：LangGraph 添加记忆官方教程 — 节点内存储访问 添加记忆 - LangChain 教程
2.3.4 跨会话用户画像持久化示例
以下示例演示了长期记忆的核心能力：不同 thread_id 的会话共享同一 user_id 的记忆。
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import InMemoryStore
from typing_extensions import TypedDict
# 1. 长期记忆仓（跨线程）
store = InMemoryStore()
class State(TypedDict):
messages: list
user_id: str
def bot(state: State):
user_text = state["messages"][-1]["content"]
# 2. 读长期记忆
fact = store.get((state["user_id"], "facts"), "pet")
if fact:
advice = f"记忆：{fact.value}，建议养猫。"
else:
advice = "暂无记忆，已记录。"
# 3. 硬编码写记忆
if "怕狗" in user_text:
store.put((state["user_id"], "facts"), "pet", "用户怕狗")
state["messages"].append({"role": "assistant", "content": advice})
return {"messages": state["messages"]}
graph = StateGraph(State).add_node("bot", bot).add_edge(START, "bot").add_edge("bot", END)
app = graph.compile(checkpointer=MemorySaver())
# 第一次对话（线程 A）
cfg1 = {"configurable": {"thread_id": "A"}}
app.invoke(
{"messages": [{"role": "user", "content": "我怕狗"}], "user_id": "u111"},
cfg1
)
# 第二次对话（线程 B，全新会话）
cfg2 = {"configurable": {"thread_id": "B"}}
result = app.invoke(
{"messages": [{"role": "user", "content": "推荐宠物"}], "user_id": "u111"},
cfg2
)
# Bot 回复会包含："记忆：用户怕狗，建议养猫。"
关键点：线程 A 和线程 B 使用不同的 thread_id（"A" 和 "B"），但共享相同的 user_id（"u111"）。线程 A 中存储的事实能被线程 B 读取，这正是跨会话长期记忆的核心机制。
出处：LangGraph 长期记忆（Cross-Session Memory Store） LangGraph教程（三）：记忆组件memory_langgraph memory-CSDN博客
2.3.5 语义搜索与向量检索
Store 支持配置 Embedding 模型实现语义搜索，让 Agent 能按语义相似性检索记忆：
from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore
# 创建启用语义搜索的存储
embeddings = init_embeddings("openai:text-embedding-3-small")
store = InMemoryStore(
index={
"embed": embeddings,
"dims": 1536,
}
)
# 存储记忆
store.put(("user_123", "memories"), "1", {"text": "I love pizza"})
store.put(("user_123", "memories"), "2", {"text": "I am a plumber"})
# 语义搜索
items = store.search(
("user_123", "memories"),
query="I'm hungry", # 语义匹配 "I love pizza"
limit=1
)
生产环境配置（PostgreSQL + pgvector）：
from langgraph.store.postgres import PostgresStore
store = PostgresStore(
connection_string="postgresql://user:pass@host:5432/db",
index={
"dims": 1536,
"embed": "openai:text-embedding-3-small"
}
)
出处：LangGraph 添加记忆官方教程 — 语义搜索 添加记忆 - LangChain 教程
2.3.6 在工具中访问长期记忆
使用 get_store() 函数在工具内访问存储，无需将 store 显式传入工具：
from langgraph.config import get_store
from langchain_core.runnables import RunnableConfig
from langgraph.prebuilt import create_react_agent
store = InMemoryStore()
# 预存用户数据
store.put(("users",), "user_123", {"name": "John Smith", "language": "English"})
def get_user_info(config: RunnableConfig) -> str:
"""Look up user info."""
store = get_store() # 在代码任意位置访问存储
user_id = config["configurable"].get("user_id")
user_info = store.get(("users",), user_id)
return str(user_info.value) if user_info else "Unknown user"
agent = create_react_agent(
model="anthropic:claude-3-7-sonnet-latest",
tools=[get_user_info],
store=store
)
Store API 核心操作汇总：
| 操作 | 同步方法 | 异步方法 | 说明 |
|---|---|---|---|
| 存储数据 | store.put(namespace, key, value) |
store.aput(...) |
写入命名空间中的键值对 |
| 检索数据 | store.get(namespace, key) |
— | 按命名空间和键精确检索 |
| 搜索数据 | store.search(namespace, query, limit) |
store.asearch(...) |
语义搜索（需配置 index） |
| 访问存储 | get_store() |
get_store() |
在任意位置获取存储实例 |
出处：LangGraph 添加记忆官方教程 — 工具中访问存储 添加记忆 - LangChain 教程
2.4 短期记忆与长期记忆的协同工作
在一个完整的 Agent 系统中，短期记忆和长期记忆通常按以下模式协同：
用户输入 ↓ ┌─────────────────────────────────────────────────────────────┐ │ 短期记忆层（Checkpointer） │ │ ├── 加载当前 thread_id 的历史状态（对话上下文） │ │ └── 将新消息追加到 messages 列表 │ └─────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────┐ │ 长期记忆层（Store） │ │ ├── 用 user_id 构建 namespace │ │ ├── search(namespace, query=用户输入) 检索相关历史记忆 │ │ └── 将检索到的记忆注入系统提示（System Prompt） │ └─────────────────────────────────────────────────────────────┘ ↓ Agent 推理（结合短期上下文 + 长期记忆） ↓ ┌─────────────────────────────────────────────────────────────┐ │ 记忆更新（可选） │ │ ├── 判断是否需要将新信息写入长期记忆 │ │ └── store.put(namespace, key, value) 持久化新事实 │ └─────────────────────────────────────────────────────────────┘ ↓ 输出回复 ↓ 短期记忆自动保存（Checkpointer 每个 Superstep 触发）
出处：LangGraph 记忆机制深度解析 — 完整实战架构 LangGraph State记忆机制深度解析：短期与长期记忆的实现原理与实战-CSDN博客
2.5 记忆在自我修正循环中的作用
在 Self-Correction 架构中，记忆机制发挥以下关键作用：
| 记忆类型 | 作用 |
|---|---|
| 短期记忆 | 保存当前修正循环中的迭代历史，使 Validator 能看到之前所有的尝试和错误 |
| 长期记忆 | 保存"错误模式"和"修正策略"，当同类问题再次出现时，Agent 可以直接应用已验证过的修正方案 |
| 消息 Reducer | 每次修正尝试产生的新消息自动追加到历史，避免覆盖之前的上下文 |
# 在自我修正循环中，iterations 和 messages 都存储在短期记忆中 class AgentState(TypedDict): messages: Annotated[list, add_messages] # 每次修正尝试自动追加 iterations: int # 当前循环次数 is_valid: bool # 校验结果 error_patterns: list # 可写入长期记忆的错误模式
出处：基于 LangGraph 构建企业级 Agent — 状态设计 构建具备 Cyclic Loop（循环反思） 与 Self-Correction（自我修正） 能力的企业级 Agent-CSDN博客
3. 意图识别（Intent Recognition）
2.1 实现方式概述
在 LangGraph 中，意图识别通过路由节点（Router Node）+ 条件边（Conditional Edges）实现。路由节点负责识别用户意图并输出结构化分类结果，条件边根据该结果将执行流导向不同的处理分支。
核心模式：先处理输入，再将其导向上下文相关的任务流。
出处：LangGraph 条件边与动态路由技术文章 https://juejin.cn/post/7600705962248978441
2.2 三种意图识别实现方案
方案一：规则路由（关键词/正则匹配）
适用场景：分类明确、需求简单、追求稳定性。
def router_rule(state: State) -> dict:
text = state["user_input"].lower()
if any(k in text for k in ["计算", "+", "-", "*", "/", "多少"]):
route = "math"
elif any(k in text for k in ["搜索", "查一下", "资料", "wiki"]):
route = "search"
else:
route = "chat"
return {"route": route}
def route_decision(state: State) -> str:
return state["route"]
builder.add_conditional_edges(
"router",
route_decision,
{
"math": "math",
"search": "search",
"chat": "chat",
},
)
出处：LangGraph 条件边与动态路由 — 5 种路由模式完整代码 https://juejin.cn/post/7600705962248978441
方案二：LLM 结构化输出路由（推荐）
适用场景：分类依赖语义理解、类别可能扩展。
核心做法：
-
使用
llm.with_structured_output(Route)让模型输出结构化字段 -
用条件边根据结构化字段跳到对应节点
from pydantic import BaseModel, Field
from typing_extensions import Literal
class Intent(BaseModel):
intent: Literal["pricing", "refund", "tech", "other"] = Field(
description="用户意图分类：pricing/refund/tech/other"
)
intent_router = llm.with_structured_output(Intent)
def classify_intent(state: State) -> dict:
result = intent_router.invoke([
SystemMessage(content="你是客服分流路由器。只做分类，不要回答用户问题。"),
HumanMessage(content=state["user_input"]),
])
return {"intent": result.intent}
def route_intent(state: State) -> str:
return state["intent"]
builder.add_conditional_edges(
"router",
route_intent,
{
"pricing": "pricing",
"refund": "refund",
"tech": "tech",
"other": "other",
},
)
出处：LangGraph 条件边与动态路由 — LLM 路由模式 https://juejin.cn/post/7600705962248978441
方案三：Embedding 语义路由
适用场景：意图类别多、需要语义相似度匹配。
实现思路：将用户输入和各意图描述转为 embedding，计算余弦相似度，选择最匹配的意图。此方案可作为节点函数嵌入 LangGraph 路由节点中使用。
2.3 意图识别的最佳实践
| 实践要点 | 说明 |
|---|---|
| 结构化输出 | 路由节点必须使用 with_structured_output 或 Pydantic 模型，禁止纯文本解析 |
| 路由标签与节点名分离 | route_fn 返回业务意图标签，mapping 决定标签对应哪个节点 |
| 可观测性 | 将 decision/intent 放入 state，便于 debug 和统计分析 |
| 兜底策略 | 路由输出不在 mapping 中时，统一 fallback 到 other 或 human_handoff |
| 分支数量控制 | 3～6 个一级分支为宜，更细分流放在二级路由 |
出处：LangGraph 条件边与动态路由 — 路由设计最佳实践 https://juejin.cn/post/7600705962248978441
4. 路径规划（Path Planning）
3.1 动态路径规划的核心机制
LangGraph 的路径规划完全依赖条件边（Conditional Edges）实现。与传统 DAG（有向无环图）不同，LangGraph 支持循环（Cyclic），允许 Agent 在执行过程中根据中间结果动态调整路径。
# 核心条件边函数：决定下一步走向
def should_continue(state: AgentState) -> str:
if state["is_valid"]:
return "end"
if state["iterations"] > 3:
return "end" # 防止死循环
return "continue"
# 循环边的关键连接：Validator -> END 或 Agent
workflow.add_conditional_edges(
"validator",
should_continue,
{
"end": END,
"continue": "agent" # Cyclic Loop 的关键
}
)
出处：基于 LangGraph 构建企业级 Agent — 循环边实现 构建具备 Cyclic Loop（循环反思） 与 Self-Correction（自我修正） 能力的企业级 Agent-CSDN博客
3.2 路径规划的五种模式
| 模式 | 适用场景 | 关键特征 | 出处 |
|---|---|---|---|
| 规则路由 | 分类明确、追求稳定 | if/else 判断 | https://juejin.cn/post/7600705962248978441 |
| LLM 路由 | 语义理解、类别扩展 | 结构化输出 | https://juejin.cn/post/7600705962248978441 |
| Prompt Chaining + Routing | 多步骤处理、质量要求高 | 先分类，再链式处理 | https://juejin.cn/post/7600705962248978441 |
| Parallelization + Routing | 独立子任务、性能要求高 | 先分类，再并行执行 | https://juejin.cn/post/7600705962248978441 |
| Cyclic Self-Correction | 需要试错和修正 | Validator → Agent 回路 | 构建具备 Cyclic Loop（循环反思） 与 Self-Correction（自我修正） 能力的企业级 Agent-CSDN博客 |
3.3 路径规划中的状态检查点
LangGraph 基于 Pregel 架构支持状态持久化，通过 checkpointer 实现断点续传：
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string(":memory:")
app_with_memory = workflow.compile(checkpointer=memory)
# 通过 thread_id 区分不同会话
config = {"configurable": {"thread_id": "user-123-session-abc"}}
出处：基于 LangGraph 构建企业级 Agent — 持久化与记忆机制 构建具备 Cyclic Loop（循环反思） 与 Self-Correction（自我修正） 能力的企业级 Agent-CSDN博客
5. 工具执行层：MCP 与 Skills 集成
4.1 MCP（Model Context Protocol）集成
LangGraph 通过 langchain-mcp-adapters 库实现与 MCP 服务器的集成。
核心 API：
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
# 配置多个 MCP 服务器
client = MultiServerMCPClient({
"math": {
"command": "python",
"args": ["/path/to/math_server.py"],
"transport": "stdio",
},
"weather": {
"url": "https://:8000/mcp",
"transport": "streamable_http",
}
})
# 获取工具并创建代理
tools = await client.get_tools()
agent = create_react_agent("anthropic:claude-3-7-sonnet-latest", tools)
支持的传输方式：
| 传输方式 | 说明 | 配置字段 |
|---|---|---|
stdio |
标准输入输出通信 | command, args, transport |
streamable_http |
可流式 HTTP 传输 | url, transport |
出处：LangGraph MCP 集成官方文档 https://langgraph.com.cn/agents/mcp/index.html
自定义 MCP 服务器示例：
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Math")
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
if __name__ == "__main__":
mcp.run(transport="stdio")
出处：LangGraph MCP 集成官方文档 https://langgraph.com.cn/agents/mcp/index.html
4.2 Skills（技能）模块化支持
除 MCP 外，LangGraph 支持通过自定义 Skill 基类实现模块化能力扩展。Skill 是可复用、可插拔的能力单元。
Skill 基类设计：
from abc import abstractmethod from langchain_core.tools import BaseTool from pydantic import BaseModel, Field class BaseSkill(BaseTool): """Skill基类，子类必须实现_run方法""" @abstractmethod def _run(self, *args, **kwargs): """Skill核心执行逻辑，子类必须实现""" pass
出处：基于 LangGraph 实现模块化 Skills 型 AI Agent 基于LangGraph实现模块化Skills型AI Agent_langgraph skill-CSDN博客
Skill 实现示例：
class SerperSearchSkill(BaseSkill): name: str = "serper_search" description: str = "用于联网搜索实时信息，输入搜索关键词，返回结构化搜索结果" class ArgsSchema(BaseModel): query: str = Field(description="搜索关键词") args_schema: type[BaseModel] = ArgsSchema def _run(self, query: str) -> str: search = GoogleSerperAPIWrapper(gl="cn", hl="zh-CN", k=5) search_results = search.results(query) return json.dumps(search_results, ensure_ascii=False, indent=2)
出处：基于 LangGraph 实现模块化 Skills 型 AI Agent 基于LangGraph实现模块化Skills型AI Agent_langgraph skill-CSDN博客
Skill 注册与绑定：
# 实例化所有 Skill search_skill = SerperSearchSkill() calculator_skill = CalculatorSkill() # 收集 Skill 列表，供模型绑定 skills = [search_skill, calculator_skill] # 绑定 Skill 到模型 llm_with_skills = llm.bind_tools(skills)
出处：基于 LangGraph 实现模块化 Skills 型 AI Agent 基于LangGraph实现模块化Skills型AI Agent_langgraph skill-CSDN博客
4.3 ToolNode 与工具执行
ToolNode 是 LangGraph 预构建的工具执行节点，支持同步/异步工具、并发执行、错误处理：
from langgraph.prebuilt import ToolNode
tool_node = ToolNode([get_weather, get_coolest_cities], handle_tool_errors=True)
# 输入：包含 tool_calls 的 AIMessage
# 输出：执行工具后产生的 ToolMessage
result = tool_node.invoke({"messages": [message_with_tool_calls]})
出处：LangGraph 工具调用官方文档 调用工具 - LangChain 教程
4.4 工具错误处理
# 默认错误处理：捕获异常并作为 ToolMessage 返回 agent = create_react_agent(model="...", tools=[multiply]) # 禁用错误处理 ToolNode([multiply], handle_tool_errors=False) # 自定义错误消息 ToolNode([multiply], handle_tool_errors="Custom error message")
出处：LangGraph 工具使用完整技术指南 https://langgraph.com.cn/agents/tools.1.html
6. 结果验证与问题定位
5.1 Validator 节点设计
结果验证是自我修正循环的关键环节。通过专门的 Validator 节点检查 Agent 的输出是否满足预设标准。
def validate_node(state: AgentState):
"""硬核校验节点：检查 AI 的回复是否真的解决了问题"""
last_message = state["messages"][-1]
reflection_prompt = HumanMessage(content=f"""
你正在审核一位初级工程师的回答。
用户问题是：{state["messages"][0].content}
工程师的回答是：{last_message.content}
请判断：该回答是否包含具体的**数据支撑**或**明确的操作步骤**？
回复 'valid' 或 'invalid'。
""")
reflection = model.invoke([reflection_prompt])
if "valid" in reflection.content.lower():
return {"is_valid": True}
else:
correction_msg = HumanMessage(
content="你的回答不够具体，请结合工具重新分析，必须包含数据！"
)
return {"is_valid": False, "messages": [correction_msg]}
出处：基于 LangGraph 构建企业级 Agent — Validator 节点实现 构建具备 Cyclic Loop（循环反思） 与 Self-Correction（自我修正） 能力的企业级 Agent-CSDN博客
5.2 问题定位的状态检查机制
LangGraph 提供多种状态检查机制用于问题定位：
| 机制 | 用途 | API |
|---|---|---|
| 状态字段追踪 | 在 State 中定义 iterations, is_valid, error_message 等字段 |
TypedDict / dataclass |
| 消息历史审计 | 通过 messages 字段追踪完整对话和工具调用历史 |
add_messages reducer |
| Human-in-the-Loop | 在关键节点暂停执行，人工审查 | interrupt() |
| 检查点回放 | 通过 thread_id 回放历史状态 |
checkpointer.get(config) |
5.3 Human-in-the-Loop 审查
from langgraph.types import interrupt
from langgraph.checkpoint.memory import InMemorySaver
def book_hotel(hotel_name: str):
response = interrupt(
f"Trying to call `book_hotel` with args {{'hotel_name': {hotel_name}}}. "
"Please approve or suggest edits."
)
if response["type"] == "accept":
pass
elif response["type"] == "edit":
hotel_name = response["args"]["hotel_name"]
return f"Successfully booked a stay at {hotel_name}."
# 恢复执行
from langgraph.types import Command
for chunk in agent.stream(Command(resume={"type": "accept"}), config):
print(chunk)
出处：LangGraph 人机协作官方文档 人机协作 - LangChain 教程
7. 自我修正机制（Self-Correction）
6.1 核心架构：循环边（Cyclic Loop）
LangGraph 的自我修正通过循环边实现。当 Validator 判定回答不合格时，流程被"打回"到 Agent 节点重新执行，形成"生成 → 验证 → 修正 → 再验证"的闭环。
用户输入 ↓ Agent Node（决策与检索） ↓ 需要工具？ ──是──→ Tool Node（执行） │ ↓ │ Validator Node（校验） │ ↓ │ ┌─校验通过？──是──→ END │ │ │否 │ │ └──→ 回到 Agent Node（重试） │ │ └──直接回答──→ Validator Node
出处：基于 LangGraph 构建企业级 Agent — 系统架构设计 构建具备 Cyclic Loop（循环反思） 与 Self-Correction（自我修正） 能力的企业级 Agent-CSDN博客
6.2 Reflection 模式（自我批评）
Reflection 模式由三个核心节点构成闭环：
-
生成节点：根据输入生成初始内容
-
批评节点：AI 切换为"批评者"身份，评估输出并给出改进建议
-
改进节点：AI 切换回"生成者"身份，根据批评意见改进内容
@dataclass
class ReflectionState:
messages: List[BaseMessage]
original_input: str
iteration_count: int
final_output: Optional[str]
critique_history: List[str]
improvement_history: List[str]
is_converged: bool
# 条件边控制循环
workflow.add_conditional_edges(
"critique",
self._should_continue,
{"continue": "improve", "finish": "finalize"}
)
workflow.add_conditional_edges(
"improve",
self._check_iteration_limit,
{"continue": "critique", "finish": "finalize"}
)
出处：基于 LangGraph 的自我改进智能体 — Reflection 与 Reflexion 技术详解 基于LangGraph的自我改进智能体：Reflection与Reflexion技术详解与实现 - 技术栈
6.3 Reflexion 模式（递进式知识检索）
Reflexion 在 Reflection 基础上增加了"主动寻求外部知识"的能力：
-
初始生成：生成初步响应
-
自省分析：识别知识缺口和需要验证的细节
-
知识检索：根据自省结果生成搜索查询，检索外部信息
-
知识整合：将外部知识与初始响应结合，生成最终答案
@dataclass class ReflexionState: messages: List[BaseMessage] original_input: str current_iteration: int initial_response: Optional[str] self_reflection: Optional[str] search_queries: List[str] external_knowledge: List[Dict] final_response: Optional[str]
出处：基于 LangGraph 的自我改进智能体 — Reflexion 模式 基于LangGraph的自我改进智能体：Reflection与Reflexion技术详解与实现 - 技术栈
6.4 防止死循环的安全机制
| 机制 | 实现方式 | 说明 |
|---|---|---|
| 迭代计数器 | iterations 字段每次 Agent 调用递增 |
在 state 中追踪 |
| 硬上限 | if state["iterations"] > 3: return "end" |
条件边中强制限制 |
| 收敛标志 | is_valid 或 is_converged 字段 |
Validator 设置 |
| 递归限制 | recursion_limit 参数 |
compile 时设置 |
出处：基于 LangGraph 构建企业级 Agent — 重试机制详解 构建具备 Cyclic Loop（循环反思） 与 Self-Correction（自我修正） 能力的企业级 Agent-CSDN博客
6.5 温度参数分层策略
针对自我修正循环的优化策略：
| 阶段 | 温度 | 目的 |
|---|---|---|
| 生成阶段 | 0.7 | 鼓励创造性和多样性 |
| 批评阶段 | 0.3 | 确保评估的逻辑性和严谨性 |
| 改进阶段 | 0.7 | 重新生成时保持灵活性 |
出处：基于 LangGraph 的自我改进智能体 — 性能优化策略 基于LangGraph的自我改进智能体：Reflection与Reflexion技术详解与实现 - 技术栈
8. 完整 Agent 架构设计
8.1 系统架构图
┌─────────────────────────────────────────────────────────────────────────────┐ │ 用户输入层 │ └─────────────────────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────────────────────┐ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ 记忆层 (Memory Layer) │ │ │ │ ┌─────────────────────┐ ┌─────────────────────────────────────┐ │ │ │ │ │ 短期记忆 (STM) │ │ 长期记忆 (LTM) │ │ │ │ │ │ Checkpointer │ │ Store (namespace + key → value) │ │ │ │ │ │ • thread_id 隔离 │ │ • user_id 跨会话共享 │ │ │ │ │ │ • 自动保存状态快照 │ │ • 显式 put / get / search │ │ │ │ │ │ • 多轮对话上下文 │ │ • 语义搜索 / 向量检索 │ │ │ │ │ │ • 支持断点续传 │ │ • 用户画像 / 偏好 / 知识库 │ │ │ │ │ └─────────────────────┘ └─────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ ↓ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 意图识别节点 │───→│ 路径规划节点 │───→│ 任务执行节点 │ │ │ │ (Router Node)│ │(Conditional │ │ (Agent Node) │ │ │ │ │ │ Edges) │ │ │ │ │ └──────────────┘ └──────────────┘ └──────┬───────┘ │ │ │ │ │ ┌───────────────────────────────────────────────┘ │ │ ↓ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 工具层 │ │ MCP 服务器 │ │ Skills 模块 │ │ │ │ (ToolNode) │←──→│ (外部工具) │ │ (内部技能) │ │ │ └──────┬───────┘ └──────────────┘ └──────────────┘ │ │ ↓ │ │ ┌──────────────┐ ┌──────────────┐ │ │ │ 结果验证节点 │───→│ 自我修正循环 │◄─────────────────────────────────────┘ │ │(Validator │ │(Cyclic Loop) │ （校验不通过时回退） │ │ Node) │ │ │ │ └──────┬───────┘ └──────────────┘ │ ↓（校验通过） │ ┌──────────────┐ ┌──────────────────────┐ │ │ 最终输出 │───→│ 记忆更新 (可选) │ │ └──────────────┘ │ store.put() 新事实 │ │ └──────────────────────┘ └─────────────────────────────────────────────────────────────────────────────┘
7.2 完整代码实现框架
from typing import TypedDict, Annotated, List, Union, Optional
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, create_react_agent
from langchain_core.tools import tool
from pydantic import BaseModel, Field
from langchain_mcp_adapters.client import MultiServerMCPClient
# ============================================================
# 1. 状态定义
# ============================================================
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
intent: str
iterations: int
is_valid: bool
tool_results: List[dict]
final_output: Optional[str]
# ============================================================
# 2. 记忆层配置
# ============================================================
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
# 短期记忆：Checkpointer（线程级，自动保存）
short_term_memory = InMemorySaver()
# 长期记忆：Store（跨会话，显式读写）
# 如需语义搜索，可配置 index={"embed": embeddings, "dims": 1536}
long_term_memory = InMemoryStore()
# ============================================================
# 3. Skills 定义（内部技能）
# ============================================================
class CalculatorSkillInput(BaseModel):
expression: str = Field(description="数学表达式")
@tool("calculator", args_schema=CalculatorSkillInput)
def calculator(expression: str) -> str:
"""数学计算器"""
try:
result = eval(expression, {"__builtins__": None}, {})
return f"计算结果：{result}"
except Exception as e:
return f"计算失败：{str(e)}"
# ============================================================
# 3. MCP 工具配置（外部工具）
# ============================================================
# mcp_client = MultiServerMCPClient({
# "math": {"command": "python", "args": ["math_server.py"], "transport": "stdio"}
# })
# mcp_tools = await mcp_client.get_tools()
# ============================================================
# 4. 意图识别节点
# ============================================================
class IntentClassifier(BaseModel):
intent: str = Field(description="意图分类：search/calculate/general")
intent_llm = llm.with_structured_output(IntentClassifier)
def intent_node(state: AgentState) -> dict:
result = intent_llm.invoke([
SystemMessage(content="将用户请求分类为 search/calculate/general"),
HumanMessage(content=state["messages"][-1].content)
])
return {"intent": result.intent}
def route_intent(state: AgentState) -> str:
return state["intent"]
# ============================================================
# 5. 任务执行节点（集成长期记忆）
# ============================================================
def agent_node(
state: AgentState,
config: RunnableConfig,
*,
store: BaseStore
) -> dict:
user_id = config["configurable"].get("user_id", "anonymous")
namespace = ("memories", user_id)
# 检索长期记忆
memories = store.search(
namespace,
query=state["messages"][-1].content if state["messages"] else "",
limit=3
)
memory_context = "\n".join([m.value.get("data", "") for m in memories])
# 构建带记忆的系统提示
system_msg = f"You are a helpful assistant.\n## User Memories\n{memory_context}"
messages = [{"role": "system", "content": system_msg}] + state["messages"]
response = model_with_tools.invoke(messages)
return {
"messages": [response],
"iterations": state.get("iterations", 0) + 1
}
# ============================================================
# 6. 结果验证节点
# ============================================================
def validator_node(state: AgentState) -> dict:
last_message = state["messages"][-1]
if not last_message.content or len(last_message.content) < 10:
return {
"is_valid": False,
"messages": [HumanMessage(content="输出过短，请提供更详细的回答")]
}
return {"is_valid": True}
# ============================================================
# 7. 自我修正条件边
# ============================================================
def should_continue(state: AgentState) -> str:
if state["is_valid"]:
return "end"
if state["iterations"] > 3:
return "end"
return "continue"
# ============================================================
# 8. 构建图
# ============================================================
workflow = StateGraph(AgentState)
workflow.add_node("intent", intent_node)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", ToolNode([calculator]))
workflow.add_node("validator", validator_node)
workflow.add_edge(START, "intent")
workflow.add_conditional_edges("intent", route_intent, {
"search": "agent",
"calculate": "agent",
"general": "agent"
})
workflow.add_conditional_edges("agent",
lambda s: "tools" if s["messages"][-1].tool_calls else "validator",
{"tools": "tools", "validator": "validator"}
)
workflow.add_edge("tools", "validator")
workflow.add_conditional_edges("validator", should_continue, {
"end": END,
"continue": "agent"
})
app = workflow.compile(
checkpointer=short_term_memory, # 启用短期记忆（线程级自动保存）
store=long_term_memory # 启用长期记忆（跨会话显式读写）
)
# 调用时需同时传入 thread_id（短期记忆隔离）和 user_id（长期记忆共享）
config = {
"configurable": {
"thread_id": "session_001",
"user_id": "user_123"
}
}
8.3 架构要点总结
| 特性 | 实现方式 | 出处 |
|---|---|---|
| 意图识别 | Router Node + with_structured_output + Conditional Edges |
https://juejin.cn/post/7600705962248978441 |
| 路径规划 | add_conditional_edges 动态分支选择 |
构建具备 Cyclic Loop（循环反思） 与 Self-Correction（自我修正） 能力的企业级 Agent-CSDN博客 |
| 工具执行 | ToolNode + create_react_agent |
调用工具 - LangChain 教程 |
| MCP 集成 | MultiServerMCPClient + client.get_tools() |
https://langgraph.com.cn/agents/mcp/index.html |
| Skills 支持 | 自定义 BaseSkill 类 + llm.bind_tools(skills) |
基于LangGraph实现模块化Skills型AI Agent_langgraph skill-CSDN博客 |
| 短期记忆 | Checkpointer（InMemorySaver / PostgresSaver）+ thread_id |
添加记忆 - LangChain 教程 |
| 长期记忆 | BaseStore（InMemoryStore / PostgresStore）+ namespace |
添加记忆 - LangChain 教程 |
| 语义搜索 | store.search(namespace, query, limit) + Embedding 配置 |
添加记忆 - LangChain 教程 |
| 结果验证 | Validator Node + LLM 自我反思 | 构建具备 Cyclic Loop（循环反思） 与 Self-Correction（自我修正） 能力的企业级 Agent-CSDN博客 |
| 自我修正 | Cyclic Loop (validator → agent) |
基于LangGraph的自我改进智能体：Reflection与Reflexion技术详解与实现 - 技术栈 |
| 死循环防护 | iterations 计数器 + 硬上限 |
构建具备 Cyclic Loop（循环反思） 与 Self-Correction（自我修正） 能力的企业级 Agent-CSDN博客 |
| 状态持久化 | SqliteSaver / InMemorySaver |
构建具备 Cyclic Loop（循环反思） 与 Self-Correction（自我修正） 能力的企业级 Agent-CSDN博客 |
| 人工审查 | interrupt() + Command(resume=...) |
人机协作 - LangChain 教程 |
9. 参考来源汇总
验证声明：本报告中所有技术细节均来自上述列出的官方文档、GitHub 源码或技术博客。代码示例均基于实际可运行的 API 进行编写，核心概念和 API 签名均与 LangGraph 官方文档一致。未包含任何未经来源验证的臆想内容。
欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！
更多推荐
6
0- 0
已为社区贡献4条内容
所有评论(0)