痛点
构建 AI Agent 时，最容易被低估的组件就是 记忆系统（Memory）。一个没有记忆的 Agent 每次对话都是"失忆"状态——无法记住用户偏好、无法从历史任务中学习、无法在多轮交互中保持上下文连贯。
生产环境中常见的记忆痛点：
- 上下文窗口溢出：对话过长导致 token 超限，早期关键信息被截断
- 信息检索低效：所有历史一股脑塞进 prompt，成本高且噪声大
- 跨会话遗忘：用户第二天回来，Agent 对之前的讨论一无所知
- 多 Agent 协作断裂：子 Agent 完成任务后，上下文无法有效传递给主 Agent
方案：三层记忆架构
借鉴认知科学中人类记忆的分层模型，生产级 AI Agent 的记忆系统可以设计为三层：
| 层级 | 类型 | 存储周期 | 实现方式 | 典型场景 |
|---|---|---|---|---|
| L1 | 工作记忆（Working Memory） | 单次会话 | 上下文窗口 + 滑动摘要 | 多轮对话上下文 |
| L2 | 情景记忆（Episodic Memory） | 跨会话 | 向量数据库 + 时间索引 | 历史对话检索、任务回溯 |
| L3 | 语义记忆（Semantic Memory） | 永久 | 结构化知识库 + KV Store | 用户画像、规则、学到的知识 |
实操步骤
Step 1：工作记忆 — 滑动窗口 + 递归摘要
工作记忆的核心挑战是在有限的上下文窗口内保留最有价值的信息。
from dataclasses import dataclass, field
from typing import List
import tiktoken
@dataclass
class WorkingMemory:
"""滑动窗口 + 递归摘要的工作记忆实现"""
max_tokens: int = 8000
summary_threshold: int = 6000 # 触发摘要压缩的阈值
messages: List[dict] = field(default_factory=list)
running_summary: str = ""
def add_message(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
if self._count_tokens() > self.summary_threshold:
self._compress()
def _compress(self):
"""将前半部分消息压缩为摘要，保留最近消息"""
midpoint = len(self.messages) // 2
old_messages = self.messages[:midpoint]
self.messages = self.messages[midpoint:]
# 调用 LLM 生成增量摘要
self.running_summary = self._summarize(
self.running_summary, old_messages
)
def get_context(self) -> List[dict]:
"""构建发送给 LLM 的上下文"""
context = []
if self.running_summary:
context.append({
"role": "system",
"content": f"[对话历史摘要]\n{self.running_summary}"
})
context.extend(self.messages)
return context
def _count_tokens(self) -> int:
enc = tiktoken.get_encoding("cl100k_base")
total = len(enc.encode(self.running_summary))
for msg in self.messages:
total += len(enc.encode(msg["content"]))
return total
def _summarize(self, existing_summary: str, messages: List[dict]) -> str:
"""调用 LLM 做增量摘要（实际实现替换为 API 调用）"""
# 生产中调用轻量模型（如 GPT-4o-mini）做摘要
# 要点：保留关键决策、用户偏好、未完成任务
...
关键设计决策： - 摘要压缩比建议 3:1 到 5:1，过度压缩会丢失关键细节 - 摘要使用轻量模型（如 Claude Haiku、GPT-4o-mini），主对话用强模型 - 保留最近 3-5 轮原始消息不压缩，避免上下文断裂
Step 2：情景记忆 — 向量检索 + 时间衰减
情景记忆让 Agent 能够"回忆"过去的对话和任务执行经历。
import time
import numpy as np
from typing import Optional
class EpisodicMemory:
"""基于向量检索 + 时间衰减的情景记忆"""
def __init__(self, vector_store, embedding_model, decay_factor=0.995):
self.store = vector_store # pgvector / Qdrant / Chroma
self.embedder = embedding_model
self.decay_factor = decay_factor
def store_episode(self, content: str, metadata: dict):
"""存储一条情景记忆"""
embedding = self.embedder.encode(content)
self.store.upsert(
id=metadata.get("episode_id"),
vector=embedding,
payload={
"content": content,
"timestamp": time.time(),
"session_id": metadata.get("session_id"),
"importance": metadata.get("importance", 0.5),
**metadata,
}
)
def recall(self, query: str, top_k: int = 5,
time_weight: float = 0.3) -> list:
"""带时间衰减的相关记忆检索"""
query_embedding = self.embedder.encode(query)
# 向量相似度检索
candidates = self.store.search(
vector=query_embedding, limit=top_k * 3
)
# 综合评分 = 相似度 × (1 - time_weight) + 时效性 × time_weight
now = time.time()
scored = []
for item in candidates:
age_hours = (now - item.payload["timestamp"]) / 3600
recency = self.decay_factor ** age_hours
importance = item.payload.get("importance", 0.5)
final_score = (
item.score * (1 - time_weight) +
recency * time_weight * 0.7 +
importance * 0.3
)
scored.append((final_score, item))
scored.sort(key=lambda x: x[0], reverse=True)
return [item for _, item in scored[:top_k]]
def summarize_session(self, session_id: str) -> str:
"""会话结束时生成情景摘要并存储"""
episodes = self.store.filter(
conditions={"session_id": session_id}
)
# 生成该次会话的结构化摘要
# 包含：任务目标、关键决策、最终结果、待办事项
...
生产要点：
- 不是每句话都存——设置 importance 评分阈值，过滤掉闲聊
- 会话结束时触发 summarize_session，将整次会话浓缩为一条高质量情景记忆
- 时间衰减因子 0.995 意味着一周前的记忆权重约为原来的 70%
Step 3：语义记忆 — 结构化知识持久存储
语义记忆存储 Agent 从交互中"学到"的知识——用户偏好、业务规则、常见问题解法。
import json
from datetime import datetime
class SemanticMemory:
"""结构化语义记忆：用户画像 + 学到的规则"""
def __init__(self, kv_store):
self.store = kv_store # Redis / DynamoDB / SQLite
def update_user_profile(self, user_id: str, facts: dict):
"""增量更新用户画像"""
key = f"user_profile:{user_id}"
profile = json.loads(self.store.get(key) or "{}")
for k, v in facts.items():
profile[k] = {
"value": v,
"updated_at": datetime.utcnow().isoformat(),
"confidence": profile.get(k, {}).get("confidence", 0) + 0.1
}
self.store.set(key, json.dumps(profile, ensure_ascii=False))
def learn_rule(self, rule_type: str, rule: dict):
"""存储 Agent 学到的规则/模式"""
key = f"rules:{rule_type}"
rules = json.loads(self.store.get(key) or "[]")
# 去重：如果已有相似规则，增加置信度
for existing in rules:
if existing["pattern"] == rule["pattern"]:
existing["confidence"] = min(
existing["confidence"] + 0.2, 1.0
)
existing["last_seen"] = datetime.utcnow().isoformat()
self.store.set(key, json.dumps(rules, ensure_ascii=False))
return
rule["confidence"] = 0.5
rule["created_at"] = datetime.utcnow().isoformat()
rules.append(rule)
self.store.set(key, json.dumps(rules, ensure_ascii=False))
def get_relevant_knowledge(self, user_id: str,
context: str) -> dict:
"""获取与当前上下文相关的语义知识"""
return {
"user_profile": self._get_profile(user_id),
"applicable_rules": self._match_rules(context),
"learned_preferences": self._get_preferences(user_id),
}
语义记忆的数据模型示例：
{
"user_profile:u_12345": {
"timezone": {"value": "Asia/Shanghai", "confidence": 0.9},
"preferred_language": {"value": "zh-CN", "confidence": 1.0},
"tech_stack": {"value": ["Python", "K8s", "AWS"], "confidence": 0.8},
"communication_style": {"value": "concise_technical", "confidence": 0.7}
}
}
Step 4：三层记忆的协同调度
class AgentMemoryOrchestrator:
"""记忆协调器：统一管理三层记忆的读写"""
def __init__(self, working, episodic, semantic):
self.working = working # L1
self.episodic = episodic # L2
self.semantic = semantic # L3
def prepare_context(self, user_id: str,
user_message: str) -> List[dict]:
"""为 LLM 调用准备完整上下文"""
context = []
# 1. 语义记忆：注入用户画像和规则（最稳定）
knowledge = self.semantic.get_relevant_knowledge(
user_id, user_message
)
if knowledge["user_profile"]:
context.append({
"role": "system",
"content": f"[用户画像] {json.dumps(knowledge['user_profile'], ensure_ascii=False)}"
})
# 2. 情景记忆：检索相关历史（跨会话）
relevant_episodes = self.episodic.recall(user_message, top_k=3)
if relevant_episodes:
episodes_text = "\n".join(
[ep.payload["content"] for ep in relevant_episodes]
)
context.append({
"role": "system",
"content": f"[相关历史记忆]\n{episodes_text}"
})
# 3. 工作记忆：当前会话上下文（最新鲜）
context.extend(self.working.get_context())
# 加入当前用户消息
context.append({"role": "user", "content": user_message})
return context
def post_response(self, user_id: str,
user_msg: str, agent_response: str):
"""响应后更新各层记忆"""
# 更新工作记忆
self.working.add_message("user", user_msg)
self.working.add_message("assistant", agent_response)
# 异步评估是否值得存入情景记忆
# （importance > 0.6 的交互才存储）
# 异步提取用户偏好更新语义记忆
# （检测到新偏好/规则时触发）
避坑指南
1. 记忆污染问题
坑：Agent 把错误信息或幻觉写入长期记忆，后续对话被持续误导。
解法： - 语义记忆写入设置 confidence 阈值，单次交互提取的事实 confidence 初始值低（0.3-0.5） - 多次确认后才提升 confidence，低于 0.3 的定期清理 - 关键事实写入前做一次 LLM 验证（"这个信息是否可靠？"）
2. 检索噪声与上下文污染
坑：情景记忆检索回来的内容与当前问题相关度低，反而干扰 LLM 推理。
解法： - 设置相似度下限（cosine similarity > 0.75 才纳入） - 检索结果做 reranking，用 Cross-Encoder 二次排序 - 控制注入的记忆总 token 数不超过上下文窗口的 20%
3. 存储成本失控
坑：所有对话都无差别存储到向量数据库，数据量指数增长，检索延迟上升。
解法： - 分级存储：热数据（7天）在内存向量库，温数据（30天）在磁盘向量库，冷数据压缩归档 - 合并策略：同一主题的多条情景记忆定期合并为一条摘要 - 容量预算：每个用户的情景记忆上限设为 1000 条，超限时按重要性淘汰
总结
生产级 AI Agent 的记忆系统不是简单的"把聊天记录存起来"，而是需要：
- 分层设计：工作记忆（快/小/易失）→ 情景记忆（中/向量检索）→ 语义记忆（慢/结构化/持久）
- 主动遗忘：不是所有信息都值得记住，importance 评分 + 时间衰减是关键
- 异步更新：记忆写入不能阻塞主对话流程，后台异步处理
- 防污染机制：confidence 阈值 + 定期清理，避免错误信息持久化
技术选型建议：向量库选 pgvector（如果已有 PostgreSQL）或 Qdrant（独立部署），KV 存储用 Redis + 持久化，摘要压缩用轻量模型降低成本。整套方案在 4C8G 的单机上即可支撑千级并发用户的记忆系统。