AI Agent 记忆系统设计：从短期上下文到长期知识的分层架构实践

AI Agent 记忆系统设计：从短期上下文到长期知识的分层架构实践

一、Agent 的"健忘症"：为什么大模型总是"翻脸不认人"

AI Agent 在多轮对话中表现出的"健忘"问题，是当前大模型应用最普遍的痛点。用户告诉 Agent 自己的偏好是"不吃辣"，Agent 在第三轮对话就忘了；用户上传了一份项目文档，Agent 在回答后续问题时完全无视文档内容。根本原因在于，大模型的上下文窗口有限——即使 128K 的窗口，在长对话中也会被历史消息填满，更早的信息被截断丢弃。

更深层的问题是，Agent 的"记忆"不只是"记住对话历史"这么简单。一个实用的 Agent 需要三种记忆：短期记忆（当前对话上下文）、工作记忆（当前任务的中间状态）、长期记忆（用户偏好、领域知识、历史经验）。这三种记忆的存储方式、检索策略和生命周期完全不同，不能简单地把所有信息塞进 Prompt 里。

二、Agent 记忆系统的分层架构

2.1 三层记忆模型

graph TB
subgraph "短期记忆（对话上下文）"
Chat[对话历史窗口<br/>最近N轮消息]
Chat –>|滑动窗口| LLM[大语言模型]
end

subgraph "工作记忆（任务状态）"
Scratch[Scratchpad<br/>当前任务的中间结果]
Plan[执行计划<br/>待完成的子任务列表]
Tool[工具调用记录<br/>已调用的工具及结果]
Scratch –>|注入Prompt| LLM
Plan –>|注入Prompt| LLM
Tool –>|注入Prompt| LLM
end

subgraph "长期记忆（持久知识）"
UserProfile[用户画像<br/>偏好/习惯/历史]
DomainKB[领域知识库<br/>向量检索]
Episodic[情节记忆<br/>历史对话摘要]
UserProfile –>|检索| Retriever[检索器]
DomainKB –>|检索| Retriever
Episodic –>|检索| Retriever
Retriever –>|Top-K结果| LLM
end

短期记忆：直接使用大模型的上下文窗口，保留最近 N 轮对话。超出窗口的历史消息通过摘要压缩保留关键信息。滑动窗口大小根据模型上下文长度动态调整——如果当前任务需要更多工作记忆空间，短期记忆的窗口可以缩小。

工作记忆：Agent 在执行复杂任务时产生的中间状态。例如，一个数据分析 Agent 可能需要记录"已查询了哪些表"、"当前的分析假设是什么"、"下一步要做什么"。工作记忆的生命周期与任务绑定，任务完成后清空。

长期记忆：跨对话持久化的知识，包括用户画像（偏好、习惯）、领域知识库（文档、FAQ）和情节记忆（历史对话的摘要）。长期记忆通过向量检索按需召回，不占用上下文窗口的固定空间。

2.2 记忆检索与注入策略

sequenceDiagram
participant User as 用户
participant Agent as Agent
participant SM as 短期记忆
participant WM as 工作记忆
participant LM as 长期记忆

User->>Agent: "帮我分析这个月的销售数据"
Agent->>SM: 获取最近对话历史
Agent->>WM: 初始化任务状态
Agent->>LM: 检索用户偏好和领域知识

LM–>>Agent: 用户偏好: 喜欢图表展示<br/>领域知识: 销售数据模型

Agent->>Agent: 组装Prompt<br/>[短期+工作+长期记忆]

Agent->>User: "好的，我来分析。先看哪个维度？"

Note over WM: 更新任务状态<br/>待完成: [维度选择, 数据查询, 图表生成]

User->>Agent: "按渠道看"
Agent->>SM: 更新对话历史
Agent->>WM: 更新任务进度
Agent->>LM: 检索渠道分析相关经验

LM–>>Agent: 历史经验: 渠道分析通常关注<br/>GMV占比和环比变化

Agent->>Agent: 生成分析结果
Agent->>User: [渠道分析图表+结论]

三、生产级 Agent 记忆系统实现

3.1 记忆管理器核心实现

"""
Agent 记忆管理器：统一管理三层记忆的读写和检索
核心设计：记忆的生命周期管理 + 按需检索 + Prompt 组装
"""
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime
import json

@dataclass
class Message:
"""对话消息"""
role: str # user / assistant / system
content: str
timestamp: datetime = field(default_factory=datetime.now)
metadata: dict = field(default_factory=dict)

@dataclass
class TaskState:
"""工作记忆：当前任务的执行状态"""
task_id: str
goal: str # 任务目标
plan: list[dict] # 执行计划（子任务列表）
completed_steps: list[str] # 已完成的步骤
intermediate_results: dict # 中间结果
current_step: Optional[str] = None # 当前执行的步骤

@dataclass
class UserProfile:
"""长期记忆：用户画像"""
user_id: str
preferences: dict # 用户偏好
interaction_patterns: dict # 交互模式
last_updated: datetime = field(default_factory=datetime.now)

class MemoryManager:
"""Agent 记忆管理器"""

def __init__(self, llm_client, vector_store, max_short_term_tokens: int = 4000):
self.llm = llm_client
self.vector_store = vector_store
self.max_short_term_tokens = max_short_term_tokens

# 短期记忆：对话历史
self.conversation_history: list[Message] = []

# 工作记忆：当前任务状态
self.current_task: Optional[TaskState] = None

# 长期记忆：用户画像
self.user_profile: Optional[UserProfile] = None

def add_message(self, role: str, content: str, **metadata):
"""添加对话消息到短期记忆"""
msg = Message(role=role, content=content, metadata=metadata)
self.conversation_history.append(msg)

# 短期记忆溢出时，压缩早期对话
if self._estimate_tokens() > self.max_short_term_tokens:
self._compress_history()

def update_task_state(self, step_result: dict):
"""更新工作记忆：任务执行进度"""
if self.current_task is None:
return

step_id = self.current_task.current_step
if step_id:
self.current_task.completed_steps.append(step_id)
self.current_task.intermediate_results[step_id] = step_result

# 推进到下一步
remaining = [
s for s in self.current_task.plan
if s["id"] not in self.current_task.completed_steps
]
if remaining:
self.current_task.current_step = remaining[0]["id"]
else:
self.current_task.current_step = None # 任务完成

def retrieve_long_term_memory(self, query: str, top_k: int = 3) -> list[dict]:
"""从长期记忆中检索相关信息"""
results = []

# 1. 向量检索领域知识
knowledge_hits = self.vector_store.search(
query=query, top_k=top_k
)
results.extend([{"type": "knowledge", "content": hit} for hit in knowledge_hits])

# 2. 用户偏好匹配
if self.user_profile:
relevant_prefs = self._match_preferences(query)
if relevant_prefs:
results.append({"type": "preference", "content": relevant_prefs})

# 3. 情节记忆：历史对话摘要
episodic_hits = self.vector_store.search(
collection="episodic",
query=query, top_k=2
)
results.extend([{"type": "episodic", "content": hit} for hit in episodic_hits])

return results

def build_prompt(self, user_message: str) -> str:
"""
组装最终 Prompt：短期 + 工作 + 长期记忆
按优先级排列，确保关键信息不被截断
"""
parts = []

# 1. 系统指令（最高优先级）
parts.append(self._build_system_instruction())

# 2. 工作记忆（当前任务上下文）
if self.current_task:
parts.append(self._format_task_state())

# 3. 长期记忆（检索结果）
long_term = self.retrieve_long_term_memory(user_message)
if long_term:
parts.append(self._format_long_term_memory(long_term))

# 4. 短期记忆（对话历史）
parts.append(self._format_conversation_history())

# 5. 当前用户消息
parts.append(f"用户: {user_message}")

return "\\n\\n".join(parts)

def _compress_history(self):
"""压缩对话历史：保留最近3轮，早期对话摘要化"""
if len(self.conversation_history) <= 6:
return

# 取早期对话（排除最近3轮）
early_messages = self.conversation_history[:-6]
recent_messages = self.conversation_history[-6:]

# 用 LLM 生成早期对话的摘要
early_text = "\\n".join(
f"{m.role}: {m.content}" for m in early_messages
)

summary_prompt = (
f"请将以下对话历史压缩为一段简洁的摘要，"
f"保留关键事实、用户偏好和未完成的任务：\\n\\n{early_text}"
)
summary = self.llm.chat(summary_prompt)

# 用摘要替换早期对话
summary_msg = Message(
role="system",
content=f"[对话历史摘要] {summary}",
metadata={"is_summary": True}
)
self.conversation_history = [summary_msg] + recent_messages

def _estimate_tokens(self) -> int:
"""估算当前对话历史的 token 数（简化实现）"""
total_chars = sum(len(m.content) for m in self.conversation_history)
return total_chars // 4 # 粗略估计：4字符≈1 token

def _match_preferences(self, query: str) -> Optional[str]:
"""匹配用户偏好与当前查询的相关性"""
if not self.user_profile:
return None

relevant = []
for key, value in self.user_profile.preferences.items():
# 简化的关键词匹配，生产环境可用语义相似度
if any(kw in query for kw in key.split("_")):
relevant.append(f"{key}: {value}")

return "; ".join(relevant) if relevant else None

def _build_system_instruction(self) -> str:
return "你是一个智能助手，请基于提供的记忆信息回答用户问题。"

def _format_task_state(self) -> str:
task = self.current_task
return (
f"[当前任务] 目标: {task.goal}\\n"
f"已完成: {task.completed_steps}\\n"
f"当前步骤: {task.current_step}\\n"
f"中间结果: {json.dumps(task.intermediate_results, ensure_ascii=False)}"
)

def _format_long_term_memory(self, memories: list[dict]) -> str:
lines = ["[相关记忆]"]
for m in memories:
lines.append(f" ({m['type']}) {m['content']}")
return "\\n".join(lines)

def _format_conversation_history(self) -> str:
lines = ["[对话历史]"]
for m in self.conversation_history:
prefix = m.role
if m.metadata.get("is_summary"):
prefix = "摘要"
lines.append(f"{prefix}: {m.content}")
return "\\n".join(lines)

3.2 向量存储与情节记忆

"""
情节记忆：将历史对话摘要存入向量数据库，支持语义检索
核心设计：对话结束时自动摘要 + 向量化 + 持久化
"""
import hashlib
from datetime import datetime

class EpisodicMemoryStore:
"""情节记忆存储：管理历史对话的摘要和检索"""

def __init__(self, vector_store, embedding_client):
self.vector_store = vector_store
self.embedder = embedding_client

def save_conversation(self, conversation_id: str,
messages: list[Message],
user_id: str):
"""对话结束时，生成摘要并存入向量数据库"""
# 生成对话摘要
conversation_text = "\\n".join(
f"{m.role}: {m.content}" for m in messages
)

summary_prompt = (
"请为以下对话生成一段结构化摘要，包含：\\n"
"1. 讨论的主要话题\\n"
"2. 关键结论和决策\\n"
"3. 用户的偏好和需求\\n"
"4. 未解决的问题\\n\\n"
f"对话内容：\\n{conversation_text}"
)
summary = self.embedder.llm.chat(summary_prompt)

# 向量化并存储
embedding = self.embedder.embed(summary)

doc_id = hashlib.md5(
f"{conversation_id}:{datetime.now().isoformat()}".encode()
).hexdigest()

self.vector_store.upsert(
collection="episodic",
id=doc_id,
vector=embedding,
metadata={
"user_id": user_id,
"conversation_id": conversation_id,
"summary": summary,
"timestamp": datetime.now().isoformat(),
"message_count": len(messages),
}
)

def search(self, query: str, user_id: str, top_k: int = 3) -> list[str]:
"""检索与当前查询相关的历史对话摘要"""
query_embedding = self.embedder.embed(query)

results = self.vector_store.search(
collection="episodic",
vector=query_embedding,
filter={"user_id": user_id}, # 只检索当前用户的历史
top_k=top_k,
)

return [r["metadata"]["summary"] for r in results]

四、记忆系统的 Trade-offs 分析

方案一：全量上下文 vs 摘要压缩

方案二：固定检索 vs 动态检索

固定检索在每轮对话都检索长期记忆，简单但浪费——用户问"今天天气"不需要检索用户偏好。动态检索根据查询意图决定是否检索长期记忆，节省 Token 但引入意图分类的额外开销和误判风险。

关键边界条件：

• 摘要压缩是不可逆的。一旦早期对话被压缩为摘要，原始对话的细节信息无法恢复。对于需要精确引用历史对话的场景（如法律咨询），应保留原始对话的完整记录，仅对非关键信息做摘要
• 向量检索的召回率受 Embedding 模型质量影响。当用户的查询与历史对话的表述差异较大时（如同义改写），可能检索不到相关记忆。解决方案是结合关键词检索和向量检索的混合策略
• 工作记忆的大小需要根据任务复杂度动态调整。简单问答任务不需要工作记忆，复杂的多步骤任务可能需要大量中间状态。固定大小的工作记忆要么浪费空间，要么不够用

五、总结

Agent 记忆系统的核心设计原则是分层存储、按需检索、生命周期管理。短期记忆用滑动窗口 + 摘要压缩控制 Token 消耗；工作记忆与任务绑定，任务完成即清空；长期记忆通过向量检索按需召回，不占用固定的上下文空间。

落地建议：先实现短期记忆的滑动窗口和摘要压缩，这是最基础也最通用的需求；再根据业务场景逐步引入工作记忆（多步骤任务）和长期记忆（用户偏好、领域知识）。记忆系统的效果评估不能只看"记住了多少"，更要看"检索的准确率"——检索到无关记忆反而会干扰 Agent 的判断。建议定期评估检索的 Precision@K，确保注入 Prompt 的记忆信息是真正相关的。