多智能体系统与记忆机制
一、记忆机制的理论基础
1.1 认知心理学中的记忆模型
AI Agent 的记忆设计大量借鉴了人类记忆的认知模型。最经典的是 Atkinson-Shiffrin 多存储模型（1968），将人类记忆分为三个层次：
感官记忆（数毫秒-数秒）→ 短期记忆（数秒-数分钟）→ 长期记忆（数天-数年）
AI Agent 记忆与人类记忆的映射：
| 人类记忆 | AI Agent 对应 | 容量 | 持续时间 | 检索方式 |
|---|---|---|---|---|
| 感官记忆 | 上下文窗口 | 有限（8K-2M tokens） | 毫秒级（当前请求） | 全注意力直接访问 |
| 短期/工作记忆 | Checkpointer + State | 随存储后端扩展 | 会话/线程级 | 按 thread_id 获取 |
| 长期记忆 | Store + 向量检索 | 理论上无限 | 跨会话持久 | 语义搜索 |
1.2 认知负荷理论对 Agent 设计的启示
认知负荷理论（Sweller, 1988）指出：工作记忆的容量是有限的。LLM 的上下文窗口就是 Agent 的”工作记忆”，同样有限。这指导我们：
- 内在负荷（Intrinsic）：任务的固有复杂度 → 通过分解复杂任务为子任务来降低
- 外在负荷（Extraneous）：不必要的信息（冗余对话、无效的工具返回）→ 通过上下文压缩来降低
- 相关负荷（Germane）：需要的有意义信息 → 通过 Store 中的语义记忆来补充
1.3 为什么 LLM 本身没有”记忆”
理解这一点的关键是区分两种”记忆”：
- 参数记忆：训练过程中内化到模型权重中的知识（如”巴黎是法国首都”）。这是静态的，无法更新，也无法记住与特定用户的交互。
- 上下文记忆：通过 prompt 注入的信息。这是动态的，但受窗口大小限制，且每次新对话都要重新提供。
Agent 的记忆系统本质上是对”上下文记忆”的工程化扩展——通过外部存储和检索机制，让 Agent 能够跨越时间和会话边界维护信息。
二、LangGraph 中的记忆层次
2.1 三层记忆架构详解
┌─────────────────────────────────────────────────────┐
│ 第一层: 上下文窗口 (Context Window) │
│ 容量: 8K-2M tokens | 生命周期: 单次请求 │
│ 内容: System Prompt + 当前消息 + 当前轮的历史 │
├─────────────────────────────────────────────────────┤
│ 第二层: Checkpointer (短期记忆) │
│ 容量: 随存储后端 | 生命周期: 会话线程 │
│ 内容: 完整对话状态、工具调用历史、中间结果 │
│ 技术: LangGraph Checkpointer (Memory/SQLite/PG/Redis) │
├─────────────────────────────────────────────────────┤
│ 第三层: Store (长期记忆) │
│ 容量: 理论上无限 | 生命周期: 跨会话持久 │
│ 内容: 用户偏好、知识碎片、经验总结 │
│ 技术: LangGraph Store + 向量语义检索 │
└─────────────────────────────────────────────────────┘
协同机制：三层记忆不是互相替代而是互补。上下文窗口提供即时感知，Checkpointer 维护会话连续性，Store 提供跨会话的知识累积。每一层解决的问题不同，缺一不可。
2.2 Checkpointer：短期记忆的工程实现
Checkpointer 在每个 Superstep（节点执行后）保存状态快照。它的设计面临一个核心权衡：
保存频率 ↑ → 恢复精度 ↑（少丢失数据）→ 写开销 ↑（性能下降）
保存频率 ↓ → 写开销 ↓ → 恢复精度 ↓（多丢失数据）
LangGraph 通过异步写入（async mode）解决了这个权衡——状态写入后台线程执行，不阻塞 Agent 主流程。
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState
checkpointer = MemorySaver()
builder = StateGraph(MessagesState)
builder.add_node("agent", agent_node)
builder.add_edge(START, "agent")
graph = builder.compile(checkpointer=checkpointer)
# 同一 thread_id = 同一会话，状态自动恢复
config = {"configurable": {"thread_id": "user-001"}}
graph.invoke({"messages": [{"role": "user", "content": "我叫张三"}]}, config)
result = graph.invoke(
{"messages": [{"role": "user", "content": "我叫什么名字？"}]}, config
)
# LLM 能回答"张三"
2.3 Checkpointer 后端对比
| 后端 | 持久化 | 性能 | 并发 | 适用规模 |
|---|---|---|---|---|
MemorySaver |
❌ 进程重启丢失 | 极快 | 单线程 | 开发/测试 |
SqliteSaver |
✅ 本地文件 | 快 | 有限 | 小规模生产 |
PostgresSaver |
✅ 数据库 | 中 | 高 | 中等-大规模 |
RedisSaver |
✅（可选持久化） | 最快 | 极高 | 高性能需求 |
# SQLite
from langgraph.checkpoint.sqlite import SqliteSaver
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
# PostgreSQL
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@localhost:5432/db")
# Redis
from langgraph.checkpoint.redis import RedisSaver
checkpointer = RedisSaver.from_conn_string("redis://localhost:6379")
checkpointer.setup()
2.4 状态设计的 Reducer 模式
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
import operator
class ConversationState(TypedDict):
# add_messages: 新消息追加到列表末尾，去重
messages: Annotated[list, add_messages]
# operator.add: 列表拼接
collected_data: Annotated[list, operator.add]
# 无 Annotated: 覆盖（默认）
user_preferences: dict
turn_count: int
为什么需要 Reducer：不同的字段需要不同的更新策略。消息应该追加（保留历史），计数字段应该覆盖（state[“count”] = 5），收集的片段应该拼接。Reducer 模式让每个字段独立定义更新行为，避免了”一刀切”的覆盖策略。
三、长期记忆：Store 机制
3.1 Store vs Checkpointer：本质区别
两者解决不同层面的问题：
| 维度 | Checkpointer | Store |
|---|---|---|
| 解决什么问题 | “上次对话进行到哪了？” | “关于这个用户，我记住了什么？” |
| 数据组织方式 | 按会话线程（thread_id） | 按命名空间 + 键（namespace, key） |
| 检索方式 | 精确获取（知道 thread_id） | 语义搜索（知道大概想找什么） |
| 典型数据 | 对话历史、工具调用栈 | 用户画像、学到的知识、偏好 |
| 写入模式 | 自动（每个 Superstep） | 手动（代码主动 put） |
3.2 语义记忆的向量检索原理
Store 的语义搜索功能让 Agent 能够通过含义而非精确匹配来检索记忆：
from langgraph.store.memory import InMemoryStore
from langchain.embeddings import init_embeddings
embeddings = init_embeddings("openai:text-embedding-3-small")
store = InMemoryStore(
index={
"embed": embeddings, # 嵌入模型
"dims": 1536, # 向量维度
"fields": ["text"], # 对 value 中的 text 字段建索引
}
)
# 存储语义记忆
store.put(
("user_123", "memories"), "mem_1",
{"text": "用户喜欢 Python 编程，讨厌 Java"}
)
store.put(
("user_123", "memories"), "mem_2",
{"text": "用户是后端工程师，关注性能优化"}
)
# 语义搜索：即使用户说"编程偏好"，也能匹配到"喜欢Python"
results = store.search(
("user_123", "memories"),
query="编程语言偏好",
limit=5
)
for r in results:
print(f"分数: {r.score}, 内容: {r.value['text']}")
命名空间的设计哲学：(user_id, category) 的元组格式提供了层次化的数据组织。例如：
("user_123", "memories")：用户的个人信息("user_123", "preferences")：用户的偏好设置("project_abc", "knowledge")：项目的知识库
3.3 在 Agent 中集成 Store 与 Checkpointer
from langgraph.runtime import Runtime
from dataclasses import dataclass
@dataclass
class Context:
user_id: str
def agent_with_hybrid_memory(state: MessagesState, runtime: Runtime[Context]):
user_id = runtime.context.user_id
namespace = ("memories", user_id)
# 从长期记忆检索
user_query = state["messages"][-1].content
memories = runtime.store.search(namespace, query=user_query, limit=5)
memory_context = "\n".join([f"- {m.value['text']}" for m in memories])
system_prompt = f"用户信息：\n{memory_context}\n根据这些信息个性化回复。"
# 自动存储新信息（记忆编码）
if "记住" in user_query:
extracted_info = extract_memory(user_query)
import uuid
runtime.store.put(
namespace, str(uuid.uuid4()),
{"text": extracted_info}
)
response = model.invoke([
{"role": "system", "content": system_prompt},
*state["messages"] # Checkpointer 恢复的对话历史
])
return {"messages": [response]}
# 编译：同时注入 Checkpointer 和 Store
graph = builder.compile(checkpointer=checkpointer, store=store)
3.4 生产级 Store 实现
# Redis
from langgraph.store.redis import RedisStore
store = RedisStore.from_conn_string("redis://localhost:6379")
# PostgreSQL
from langgraph.store.postgres import PostgresStore
store = PostgresStore.from_conn_string("postgresql://...")
# 异步版本（高并发）
from langgraph.store.postgres import AsyncPostgresStore
store = AsyncPostgresStore.from_conn_string("postgresql://...")
3.5 记忆管理策略深度分析
| 策略 | 原理 | 优势 | 劣势 | 适用场景 |
|---|---|---|---|---|
| 滑动窗口 | 只保留最近 N 条消息 | 简单，时延恒定 | 丢失远期对话上下文 | 短会话、客服机器人 |
| 摘要压缩 | 定期生成对话摘要替代原文 | 压缩比高 | 细节信息丢失 | 长会话、多轮任务 |
| 重要性筛选 | LLM 评分保留高价值消息 | 保留关键信息 | 额外 LLM 调用开销 | 需要长期上下文的场景 |
| 混合策略 | 近 N 条原文 + 远期摘要 | 兼顾细节和容量 | 实现复杂 | 大多数生产场景 |
| 反思记忆 | 定期生成高层抽象和模式 | 层次化理解 | 成本最高 | 需要深度理解用户的系统 |
四、多智能体系统的理论基础
4.1 单 Agent 的局限与多 Agent 的必要性
单 Agent 架构的根本问题不是能力不足，而是认知负荷集中——一个 Agent 需要同时具备所有领域的知识、理解所有类型的任务、掌握所有工具的使用方式。这违反了”分而治之”的基本原则。
多 Agent 系统的核心思想借鉴了：
- 劳动分工理论（Adam Smith）：专业分工带来效率提升
- 分布式问题求解（Distributed Problem Solving）：将问题分解为子问题，由专门求解器处理
- 黑板架构（Blackboard Architecture）：多个专家共享一个公共信息空间，协作解决问题
4.2 多 Agent 协作的五层模型
Layer 5: 集体目标达成 (Collective Goal Achievement)
Layer 4: 任务协调 (Task Coordination)
Layer 3: 知识共享 (Knowledge Sharing)
Layer 2: 通信协议 (Communication Protocol)
Layer 1: 基础设施 (Infrastructure: MCP, Message Bus, State Store)
4.3 四种多 Agent 架构模式
模式一：主从模式（Supervisor-Worker）
核心思想：一个协调 Agent 作为调度中心，将任务分发给下属 Worker Agent。类似操作系统的进程调度器。
优势：中心控制，流程清晰 劣势：协调器成为瓶颈和单点故障
┌────────────────┐
│ Supervisor │ ← 意图识别 + 任务分配
│ (协调者) │
└───┬────┬───┬───┘
│ │ │
┌──────┘ │ └──────┐
▼ ▼ ▼
┌────────┐ ┌────────┐ ┌────────┐
│搜索专家│ │代码专家│ │翻译专家│
└────────┘ └────────┘ └────────┘
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import create_agent
from typing import Literal
def create_worker(name: str, tools: list, prompt: str):
return create_agent(
model="gpt-4o", name=name,
tools=tools, prompt=prompt,
checkpointer=MemorySaver(),
)
search_agent = create_worker("search", [search_tool], "你是搜索专家。")
code_agent = create_worker("code", [exec_tool, lint_tool], "你是编程专家。")
translate_agent = create_worker("translate", [translate_tool], "你是翻译专家。")
def supervisor_router(state: MessagesState) -> Literal["search", "code", "translate", END]:
last_message = state["messages"][-1].content.lower()
if any(kw in last_message for kw in ["搜索", "查找", "search"]):
return "search"
elif any(kw in last_message for kw in ["代码", "code", "编程"]):
return "code"
elif any(kw in last_message for kw in ["翻译", "translate"]):
return "translate"
return END
builder = StateGraph(MessagesState)
builder.add_node("search", search_agent)
builder.add_node("code", code_agent)
builder.add_node("translate", translate_agent)
builder.add_conditional_edges(START, supervisor_router, {
"search": "search", "code": "code",
"translate": "translate", END: END
})
for node in ["search", "code", "translate"]:
builder.add_edge(node, END)
supervisor = builder.compile()
最佳实践 – 让 Supervisor 本身也使用 LLM：上面的硬编码路由在简单场景可行，但真正的 Supervisor 应该使用 LLM 进行动态意图分类和任务分配，这样才能处理边界情况和复合需求。
模式二：层级模式（Hierarchical）
Supervisor 之上再有 Meta-Supervisor，适合大规模系统。
Meta-Supervisor ← 全局战略分配
/ \
Supervisor A Supervisor B ← 局部战术调度
/ \ / \
Worker1 Worker2 Worker3 Worker4 ← 具体执行
理论依据：管理幅度原理（Span of Control）——每个协调者能有效管理的直接下属是有限的。当 Agent 数量超过临界值（通常 5-7 个），引入中间协调层可以保持系统可管理性。
模式三：点对点协作（Peer-to-Peer）
Agent 之间直接通信，无中心节点。适合需要密集专家交互的场景。
Agent A ◄────────► Agent B
│ │
│ 直接对话 │
└────────────────────┘
通信机制：Agent 之间通过消息传递（类似 Actor Model）进行协商。当 Agent A 需要 Agent B 的意见时，A 直接向 B 发送消息，B 处理后返回结果。
模式四：工具委托模式（Tool Delegation）
将 Agent 封装为可供其他 Agent 调用的工具。这是最灵活的模式。
fruit_expert_tool = create_subagent_tool(create_agent(
model="gpt-4o", name="fruit_expert",
tools=[fruit_info], prompt="你是水果专家。"
), description="问水果专家关于水果的问题")
veggie_expert_tool = create_subagent_tool(create_agent(
model="gpt-4o", name="veggie_expert",
tools=[veggie_info], prompt="你是蔬菜专家。"
), description="问蔬菜专家关于蔬菜的问题")
main_agent = create_agent(
model="gpt-4o",
tools=[fruit_expert_tool, veggie_expert_tool],
prompt="将问题委派给合适的专家。",
checkpointer=MemorySaver(),
)
result = main_agent.invoke({
"messages": [{"role": "user", "content": "苹果有什么营养？西兰花呢？"}]
})
# 主 Agent 会自动调用 fruit_expert_tool 和 veggie_expert_tool
4.4 四种模式的对比选择
| 模式 | 控制方式 | 通信开销 | 可扩展性 | 容错性 | 适用场景 |
|---|---|---|---|---|---|
| 主从 | 中心化 | 低 | 中 | 低 | 流程明确的任务分发 |
| 层级 | 树状 | 低 | 高 | 中 | 大规模、多领域 |
| P2P | 去中心 | 高 | 中 | 高 | 需要深度对话协作 |
| 工具委托 | 动态 | 中 | 高 | 高 | 灵活、按需调用 |
4.5 子图命名空间隔离
当多个 Agent 作为子图运行时，LangGraph 为每个子图自动分配独立的检查点命名空间，防止状态污染：
def create_isolated_sub_agent(model, *, name, **kwargs):
agent = create_agent(model=model, name=name, **kwargs)
return (
StateGraph(MessagesState)
.add_node(name, agent)
.add_edge("__start__", name)
.compile()
)
fruit_agent = create_isolated_sub_agent(
"gpt-4o", name="fruit_agent",
tools=[fruit_info], checkpointer=True,
)
veggie_agent = create_isolated_sub_agent(
"gpt-4o", name="veggie_agent",
tools=[veggie_info], checkpointer=True,
)
命名空间隔离实现：每个 StateGraph 的编译实例使用其节点名和 __start__ 边创建独立的检查点前缀，类似于 Docker 容器中的 namespace 隔离——子 Agent A 看不到子 Agent B 的状态，反之亦然。
4.6 多智能体通信机制
| 机制 | 耦合度 | 同步性 | 适用粒度 |
|---|---|---|---|
| 共享 State | 紧耦合 | 同步 | 同一图内的 Agent 节点 |
| 消息传递 | 松耦合 | 异步 | 独立部署的 Agent |
| 工具调用 | 松耦合 | 同步 | 主 Agent 调用子 Agent |
| 事件总线 | 极松散 | 异步 | 大规模分布式系统 |
| 共享 Store | 数据层松耦合 | 异步 | 长期知识积累 |
五、多智能体设计原则
5.1 SOLID 原则在 Agent 中的应用
| 原则 | Agent 中的应用 |
|---|---|
| S – 单一职责 | 每个 Agent 只负责一个明确的领域 |
| O – 开闭原则 | 通过新增 Agent 扩展功能，不修改现有 Agent |
| L – 里氏替换 | 同类型 Agent 可通过统一接口互换（如同一工具的不同实现） |
| I – 接口隔离 | Agent 只依赖它需要的工具/数据，不依赖不需要的 |
| D – 依赖倒置 | Agent 依赖抽象的工具接口，而非具体实现 |
5.2 多 Agent 系统的最佳实践
- 显式定义 Agent 边界：每个 Agent 的能力范围应写在 system prompt 中，让协调 Agent 清楚知道应该找谁
- 容错与降级：设计”如果子 Agent 超时/失败，主 Agent 如何优雅处理”的策略
- 可观测性优先：每个 Agent 的输入、输出、决策理由都应记录到 trace 中
- 避免过度拆分：先验证单 Agent 无法处理，再考虑多 Agent——每增加一个 Agent 就增加一层通信开销
- 使用标准通信协议：MCP 提供了 Agent 间通信的标准——不要发明私有的通信格式
- 上下文预算管理：传递给子 Agent 的上下文应精简到任务必需的信息，避免将所有历史全量传递
5.3 常见反模式
| 反模式 | 表现 | 纠正 |
|---|---|---|
| 过度拆分 | 每个微操作都有一个 Agent | 合并为合理的粒度（一个 Agent 负责一个领域） |
| 循环依赖 | Agent A 调 Agent B，B 又调 A | 引入方向性——下级不应回调上级 |
| 状态泄漏 | 子 Agent 的内部状态污染了父 Agent | 使用子图命名空间隔离 |
| 固定路由 | 硬编码 if-else 决定调哪个 Agent | 使用 LLM 动态路由 |
| 忽略失败 | 子 Agent 失败被静默吞掉 | 实现显式的错误传播和处理 |
六、企业级多 Agent 系统参考架构
┌──────────────────────────────────────────────────────┐
│ 用户入口层 │
│ Web UI / REST API / WebSocket / IDE Plugin │
└──────────────────────┬───────────────────────────────┘
│
┌──────────────────────▼───────────────────────────────┐
│ 编排层（Orchestration Layer） │
│ │
│ ┌─────────────────────────────────────────────┐ │
│ │ 会话协调器 (Session Orchestrator) │ │
│ │ 意图识别 → 任务分解 → Agent路由 → 结果聚合 │ │
│ └──────────────┬──────────────────────────────┘ │
│ │ │
│ ┌──────────────▼──────────────────────────────┐ │
│ │ Agent 池 (Agent Pool) │ │
│ │ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │代码审查 │ │安全分析 │ │性能优化 │ ... │ │
│ │ │ Agent │ │ Agent │ │ Agent │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ │ │
│ └─────────────────────────────────────────────┘ │
└──────────────────────┬───────────────────────────────┘
│
┌──────────────────────▼───────────────────────────────┐
│ 共享服务层 (Shared Services) │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Store │ │Checkpoint│ │MCP Server│ │
│ │(长期记忆)│ │(短期记忆)│ │(工具层) │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │向量数据库 │ │消息队列 │ │日志/追踪 │ │
│ │(Milvus) │ │(Redis) │ │(LangSmith)│ │
│ └──────────┘ └──────────┘ └──────────┘ │
└──────────────────────────────────────────────────────┘
关键架构决策：
- 编排层是系统的”大脑”——负责意图理解和 Agent 调度，不执行具体任务
- Agent 池中的每个 Agent 是独立服务单元，通过 MCP 协议暴露能力，可以独立部署和扩展
- 共享服务层提供跨 Agent 的基础设施——Store 共享知识，Checkpointer 维护会话，MCP Server 暴露工具
- 可观测性是生产系统的必备：每个 Agent 的决定、工具调用、中间结果都应可追踪
三层的职责分离
| 层 | 职责 | 不应做的事情 |
|---|---|---|
| 编排层 | 意图分类、任务路由、结果聚合 | 执行具体的审查/分析/生成 |
| Agent 池 | 执行专业领域任务 | 跨领域的任务协调 |
| 共享服务层 | 存储、检索、消息传递 | 业务逻辑决策 |
相关条目
- Agent搭建
- RAG处理优化
- langgraph4j-study-notes-02-advanced
- 20-memory-ji-yi-xi-tong