AI Agent记忆系统设计进阶教程
从认知科学到工程实现，构建生产级Agent记忆系统的完整指南
目录
- 概述与认知科学基础
- 记忆系统架构：短期/长期/工作记忆
- 向量数据库存储方案
- MemGPT虚拟内存实现
- LangGraph状态管理
- 多Agent记忆共享
- 记忆压缩与摘要
- 持久化存储策略
- 记忆检索优化
- 隐私与安全考量
- 生产级记忆系统架构
1. 概述与认知科学基础
1.1 为什么Agent需要记忆系统
大语言模型（LLM）本身是"无状态"的——每次推理都是独立的前向传播。虽然通过上下文窗口可以注入历史对话，但这种方式存在根本性限制：
- 容量限制：即使是200K token的上下文窗口，也无法承载数月甚至数年的交互历史
- 成本线性增长：上下文越长，推理成本越高
- 信息检索效率：在长上下文中定位特定信息的效率远低于结构化存储
- 跨会话断裂：上下文窗口无法跨越独立的会话
记忆系统的本质是为Agent提供持久化、结构化、可检索的经验存储，使其具备跨会话的连续性和学习能力。
1.2 人类记忆的启发
认知科学对人类记忆的研究为Agent记忆系统设计提供了重要参考：
┌──────────────────────────────────────────────────────┐
│ 人类记忆系统 │
│ │
│ 感觉记忆 ──→ 短期记忆 ──→ 长期记忆 │
│ (<1秒) (秒~分钟) (小时~终生) │
│ │ │ │
│ │ ┌─────────┼─────────┐ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ 工作记忆 情景记忆 语义记忆 程序记忆 │
│ (主动处理) (事件) (知识) (技能) │
└──────────────────────────────────────────────────────┘
| 人类记忆类型 | 功能 | Agent对应 |
|---|---|---|
| 感觉记忆 | 瞬时输入缓存 | 输入缓冲区 |
| 短期记忆 | 当前任务相关信息 | 对话上下文窗口 |
| 工作记忆 | 主动加工和推理 | 活跃状态/思维链 |
| 情景记忆 | 过去经历的事件 | 对话历史记录 |
| 语义记忆 | 世界知识和事实 | 知识库/RAG |
| 程序记忆 | 技能和操作步骤 | 工具使用经验/技能库 |
2. 记忆系统架构
2.1 三层记忆架构
┌─────────────────────────────────────────────────────────────┐
│ Agent记忆系统 │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 工作记忆 (Working Memory) │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │当前目标 │ │活跃上下文 │ │推理状态 │ │ │
│ │ │(Goal) │ │(Context) │ │(State) │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ │ 容量：有限（受上下文窗口限制） │ │
│ │ 生命周期：单次推理/单个任务 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ 写入/读取 │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 短期记忆 (Short-term Memory) │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │对话历史 │ │当前会话 │ │临时笔记 │ │ │
│ │ │(History) │ │(Session) │ │(Scratch) │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ │ 容量：中等（数千条消息） │ │
│ │ 生命周期：单次会话 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ 编码/检索 │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 长期记忆 (Long-term Memory) │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │用户画像 │ │知识库 │ │交互模式 │ │ │
│ │ │(Profile) │ │(KB) │ │(Patterns)│ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ │ 容量：无限制（外部存储） │ │
│ │ 生命周期：持久化 │ │
│ └─────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
2.2 记忆生命周期
每条记忆从产生到消亡经历以下阶段：
from enum import Enum
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, List
class MemoryStage(Enum):
"""记忆生命周期阶段"""
PERCEPTION = "perception" # 感知：从输入中提取原始信息
ENCODING = "encoding" # 编码：将信息结构化并计算嵌入向量
CONSOLIDATION = "consolidation" # 整合：与已有记忆关联、去重、合并
STORAGE = "storage" # 存储：持久化到长期记忆
RETRIEVAL = "retrieval" # 检索：按需从存储中取出
REFINEMENT = "refinement" # 精炼：根据新信息更新旧记忆
DECAY = "decay" # 衰减：降低不常访问记忆的优先级
FORGETTING = "forgetting" # 遗忘：删除不再相关的记忆
@dataclass
class MemoryItem:
"""记忆条目数据结构"""
id: str
content: str # 原始内容
embedding: Optional[List[float]] = None # 嵌入向量
memory_type: str = "episodic" # episodic/semantic/procedural
stage: MemoryStage = MemoryStage.PERCEPTION
importance: float = 0.5 # 重要性评分 [0, 1]
access_count: int = 0 # 访问次数
last_accessed: Optional[datetime] = None
created_at: datetime = field(default_factory=datetime.now)
metadata: dict = field(default_factory=dict)
associations: List[str] = field(default_factory=list) # 关联记忆ID
2.3 记忆类型详解
class EpisodicMemory:
"""情景记忆：存储具体的交互事件"""
def __init__(self, vector_store, llm):
self.vector_store = vector_store
self.llm = llm
async def store_conversation(self, session_id: str, messages: list):
"""将对话存入情景记忆"""
# 提取关键事件
events = await self._extract_events(messages)
for event in events:
embedding = await self._compute_embedding(event.summary)
await self.vector_store.upsert(
collection="episodic",
id=f"{session_id}_{event.id}",
vector=embedding,
metadata={
"session_id": session_id,
"timestamp": event.timestamp.isoformat(),
"summary": event.summary,
"participants": event.participants,
"emotion": event.emotion_tag,
"importance": event.importance,
}
)
async def _extract_events(self, messages: list):
"""用LLM从对话中提取关键事件"""
prompt = f"""从以下对话中提取关键事件，每个事件包含：
- summary: 一句话摘要
- participants: 参与者
- emotion_tag: 情感标签
- importance: 重要性(0-1)
对话内容：
{self._format_messages(messages)}
输出JSON数组。"""
response = await self.llm.generate(prompt)
return self._parse_events(response)
class SemanticMemory:
"""语义记忆：存储抽象的知识和事实"""
def __init__(self, vector_store, knowledge_graph):
self.vector_store = vector_store
self.knowledge_graph = knowledge_graph
async def store_fact(self, fact: str, source: str, confidence: float):
"""存储一条事实性知识"""
# 检查是否与已有知识矛盾
conflicts = await self._check_conflicts(fact)
if conflicts:
# 存储为待验证状态，而非直接覆盖
await self._handle_conflict(fact, conflicts)
return
embedding = await self._compute_embedding(fact)
await self.vector_store.upsert(
collection="semantic",
id=self._generate_id(fact),
vector=embedding,
metadata={
"fact": fact,
"source": source,
"confidence": confidence,
"verified": True,
}
)
# 同步更新知识图谱
entities, relations = await self._extract_knowledge_graph(fact)
await self.knowledge_graph.add_triples(entities, relations)
3. 向量数据库存储方案
3.1 主流向量数据库对比
| 数据库 | 部署方式 | 最大向量数 | 索引类型 | 混合搜索 | 适用场景 |
|---|---|---|---|---|---|
| Pinecone | 全托管SaaS | 10亿+ | 专有 | ✅ | 快速上线，无需运维 |
| Weaviate | 自托管/SaaS | 10亿+ | HNSW | ✅ | 需要混合搜索 |
| Qdrant | 自托管/云 | 10亿+ | HNSW | ✅ | 高性能需求 |
| Milvus | 自托管 | 100亿+ | 多种 | ⚠️ | 超大规模 |
| ChromaDB | 嵌入式 | 百万级 | HNSW | ❌ | 开发原型、小规模 |
| pgvector | PostgreSQL扩展 | 千万级 | IVFFlat/HNSW | ✅ | 已有PG基础设施 |
3.2 向量数据库核心操作
from qdrant_client import QdrantClient
from qdrant_client.models import (
Distance, VectorParams, PointStruct,
Filter, FieldCondition, MatchValue
)
import numpy as np
class VectorMemoryStore:
"""基于Qdrant的向量记忆存储"""
def __init__(self, host: str = "localhost", port: int = 6333):
self.client = QdrantClient(host=host, port=port)
self._ensure_collections()
def _ensure_collections(self):
"""初始化记忆集合"""
collections = {
"episodic": 1536, # 情景记忆向量维度
"semantic": 1536, # 语义记忆向量维度
"procedural": 768, # 程序记忆向量维度（可以用小模型）
}
for name, dim in collections.items():
if not self.client.collection_exists(name):
self.client.create_collection(
collection_name=name,
vectors_config=VectorParams(
size=dim,
distance=Distance.COSINE
)
)
async def store_memory(
self,
collection: str,
memory_id: str,
content: str,
embedding: list[float],
metadata: dict
):
"""存储一条记忆"""
self.client.upsert(
collection_name=collection,
points=[
PointStruct(
id=self._hash_id(memory_id),
vector=embedding,
payload={
"content": content,
"created_at": datetime.now().isoformat(),
**metadata
}
)
]
)
async def search_memory(
self,
collection: str,
query_embedding: list[float],
top_k: int = 5,
filters: dict = None,
score_threshold: float = 0.7
) -> list[dict]:
"""语义检索记忆"""
query_filter = None
if filters:
conditions = [
FieldCondition(key=k, match=MatchValue(value=v))
for k, v in filters.items()
]
query_filter = Filter(must=conditions)
results = self.client.search(
collection_name=collection,
query_vector=query_embedding,
limit=top_k,
query_filter=query_filter,
score_threshold=score_threshold
)
return [
{
"id": hit.id,
"content": hit.payload.get("content"),
"score": hit.score,
"metadata": {k: v for k, v in hit.payload.items() if k != "content"}
}
for hit in results
]
3.3 混合搜索策略
单纯依赖向量相似度搜索存在局限——语义相近但关键词不同的查询可能遗漏。混合搜索结合了向量检索和关键词检索的优势：
from dataclasses import dataclass
@dataclass
class HybridSearchConfig:
"""混合搜索配置"""
vector_weight: float = 0.7 # 向量搜索权重
keyword_weight: float = 0.3 # 关键词搜索权重
fusion_method: str = "rrf" # 融合方法：rrf / weighted
class HybridMemoryRetriever:
"""混合检索器：向量+关键词"""
def __init__(self, vector_store, bm25_index):
self.vector_store = vector_store
self.bm25_index = bm25_index
async def search(
self,
query: str,
query_embedding: list[float],
top_k: int = 10,
config: HybridSearchConfig = None
) -> list[dict]:
config = config or HybridSearchConfig()
# 向量检索
vector_results = await self.vector_store.search_memory(
collection="episodic",
query_embedding=query_embedding,
top_k=top_k * 2 # 取更多候选用于融合
)
# BM25关键词检索
keyword_results = self.bm25_index.search(query, top_k=top_k * 2)
# 融合结果（Reciprocal Rank Fusion）
if config.fusion_method == "rrf":
return self._rrf_fusion(
vector_results, keyword_results,
config.vector_weight, config.keyword_weight,
top_k
)
else:
return self._weighted_fusion(
vector_results, keyword_results,
config.vector_weight, config.keyword_weight,
top_k
)
def _rrf_fusion(self, r1, r2, w1, w2, k=60):
"""RRF（Reciprocal Rank Fusion）融合算法"""
scores = {}
for rank, item in enumerate(r1):
scores[item["id"]] = scores.get(item["id"], 0) + w1 / (k + rank + 1)
for rank, item in enumerate(r2):
scores[item["id"]] = scores.get(item["id"], 0) + w2 / (k + rank + 1)
# 按融合分数排序
sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
id_to_item = {item["id"]: item for item in r1 + r2}
return [
{**id_to_item[rid], "fusion_score": scores[rid]}
for rid in sorted_ids[:10]
]
4. MemGPT虚拟内存实现
4.1 MemGPT核心思想
MemGPT（2023年，UC Berkeley）提出了一个精妙的类比：将LLM的上下文窗口视为"主内存（RAM）"，将外部存储视为"磁盘（Disk）"，通过操作系统的虚拟内存管理思想来管理Agent的记忆。
┌─────────────────────────────────────────────┐
│ MemGPT架构 │
│ │
│ ┌─────────────────────────────────────┐ │
│ │ 上下文窗口（主内存/RAM） │ │
│ │ │ │
│ │ ┌─────────────────────────────┐ │ │
│ │ │ 系统指令 (System Instructions)│ │ │
│ │ ├─────────────────────────────┤ │ │
│ │ │ 工作上下文 (Working Context) │ │ │
│ │ │ - 用户信息摘要 │ │ │
│ │ │ - 当前任务状态 │ │ │
│ │ ├─────────────────────────────┤ │ │
│ │ │ 对话队列 (FIFO Queue) │ │ │
│ │ │ - 最近N条消息 │ │ │
│ │ └─────────────────────────────┘ │ │
│ └──────────────┬──────────────────────┘ │
│ │ 页面换入/换出 │
│ ▼ │
│ ┌─────────────────────────────────────┐ │
│ │ 外部存储（磁盘/Disk） │ │
│ │ │ │
│ │ ┌──────────┐ ┌──────────┐ │ │
│ │ │ 档案存储 │ │ 回忆存储 │ │ │
│ │ │(Archival)│ │(Recall) │ │ │
│ │ └──────────┘ └──────────┘ │ │
│ └─────────────────────────────────────┘ │
└─────────────────────────────────────────────┘
4.2 MemGPT核心函数实现
import json
from typing import Callable
from dataclasses import dataclass, field
@dataclass
class MemGPTAgent:
"""MemGPT风格的Agent，实现虚拟内存管理"""
system_prompt: str
max_context_tokens: int = 8000
archival_storage: dict = field(default_factory=dict) # 长期存储
recall_storage: list = field(default_factory=list) # 对话历史存储
working_context: dict = field(default_factory=dict) # 工作上下文
message_queue: list = field(default_factory=list) # 消息队列
def get_core_memory_str(self) -> str:
"""生成核心记忆的字符串表示"""
return f"""[系统指令]
{self.system_prompt}
[工作上下文]
{json.dumps(self.working_context, ensure_ascii=False, indent=2)}
[当前对话队列 - 最近{len(self.message_queue)}条]
{self._format_messages(self.message_queue[-10:])}"""
def inner_thoughts(self, observation: str) -> dict:
"""Agent的内部思考过程，决定下一步操作"""
prompt = f"""你是一个具有虚拟内存管理能力的AI Agent。
你有三种记忆存储：
1. 核心记忆（当前上下文，已加载）
2. 回忆存储（全部对话历史，需搜索）
3. 档案存储（长期知识，需搜索）
可用操作：
- send_message(message): 向用户发送消息
- core_memory_append(key, value): 向核心记忆追加信息
- core_memory_replace(key, old_value, new_value): 替换核心记忆中的信息
- archival_memory_insert(content): 向档案存储写入长期记忆
- archival_memory_search(query): 在档案存储中搜索
- conversation_search(query): 在对话历史中搜索
当前状态：
{self.get_core_memory_str()}
用户输入：{observation}
请决定下一步操作（输出JSON格式的function_call）。"""
# 调用LLM生成操作决策
response = self.llm.generate(prompt)
return self._parse_function_call(response)
async def execute_step(self, user_input: str):
"""执行一步操作（核心主循环）"""
# 将用户输入加入消息队列
self.message_queue.append({"role": "user", "content": user_input})
# 检查上下文窗口是否接近上限
current_tokens = self._count_tokens(self.get_core_memory_str())
if current_tokens > self.max_context_tokens * 0.9:
await self._evict_messages()
# Agent内部思考并执行操作
while True:
action = self.inner_thoughts(user_input)
if action["function"] == "send_message":
# 最终动作：向用户回复
self.message_queue.append({
"role": "assistant",
"content": action["args"]["message"]
})
return action["args"]["message"]
elif action["function"] == "core_memory_append":
key = action["args"]["key"]
value = action["args"]["value"]
if key in self.working_context:
self.working_context[key] += f"\n{value}"
else:
self.working_context[key] = value
elif action["function"] == "archival_memory_insert":
content = action["args"]["content"]
memory_id = f"mem_{len(self.archival_storage)}"
embedding = await self._compute_embedding(content)
self.archival_storage[memory_id] = {
"content": content,
"embedding": embedding,
"timestamp": datetime.now().isoformat()
}
elif action["function"] == "archival_memory_search":
query = action["args"]["query"]
results = await self._search_archival(query)
# 将搜索结果注入上下文
self.message_queue.append({
"role": "system",
"content": f"[档案搜索结果]\n{self._format_search_results(results)}"
})
async def _evict_messages(self):
"""消息换出：将旧消息从上下文迁移到回忆存储"""
# 保留最近的消息在队列中
keep_count = len(self.message_queue) // 2
evicted = self.message_queue[:-keep_count]
self.message_queue = self.message_queue[-keep_count:]
# 将换出的消息存入回忆存储
for msg in evicted:
self.recall_storage.append(msg)
# 在队列中插入摘要
summary = await self._summarize_messages(evicted)
self.message_queue.insert(0, {
"role": "system",
"content": f"[以下是之前对话的摘要]\n{summary}"
})
4.3 MemGPT的优势与局限
| 维度 | 优势 | 局限 |
|---|---|---|
| 上下文管理 | 动态换入换出，有效利用有限窗口 | 多次换入换出可能丢失微妙上下文 |
| 长期记忆 | 支持近乎无限的记忆容量 | 依赖检索质量，可能遗漏重要信息 |
| 用户体验 | 跨会话保持连续性 | 初始设置复杂，需要调参 |
| 成本 | 减少不必要的长上下文调用 | 每次换入换出都需要额外的LLM调用 |
5. LangGraph状态管理
5.1 LangGraph中的记忆概念
LangGraph（LangChain团队开发）是一个基于图的Agent编排框架，其状态管理机制天然适合构建记忆系统：
from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage
import operator
class AgentState(TypedDict):
"""Agent状态定义——记忆系统的载体"""
# 对话消息（自动管理）
messages: Annotated[Sequence[BaseMessage], operator.add]
# 长期记忆（跨节点持久化）
long_term_memory: dict
# 当前任务上下文
current_task: str
# 检索到的相关记忆
retrieved_memories: list[str]
# Agent的内部反思
reflection: str
def memory_retrieval_node(state: AgentState) -> dict:
"""记忆检索节点：从长期记忆中检索相关信息"""
latest_message = state["messages"][-1].content
# 语义检索相关记忆
relevant_memories = vector_store.search(
query=latest_message,
collection="long_term",
top_k=5
)
return {
"retrieved_memories": [m["content"] for m in relevant_memories]
}
def reasoning_node(state: AgentState) -> dict:
"""推理节点：基于记忆和当前输入进行思考"""
context = "\n".join(state["retrieved_memories"])
latest_msg = state["messages"][-1].content
prompt = f"""基于以下信息回答用户问题：
相关记忆：
{context}
长期记忆摘要：
{state['long_term_memory'].get('user_summary', '暂无')}
用户问题：{latest_msg}"""
response = llm.generate(prompt)
return {
"messages": [AIMessage(content=response)],
"reflection": f"回答了关于{latest_msg[:30]}的问题"
}
def memory_consolidation_node(state: AgentState) -> dict:
"""记忆整合节点：将新交互写入长期记忆"""
latest_interaction = f"""
用户：{state['messages'][-2].content}
助手：{state['messages'][-1].content}
"""
# 提取值得记忆的信息
extraction_prompt = f"""从以下交互中提取值得长期记忆的信息：
{latest_interaction}
提取格式：
- 用户偏好：...
- 新知识：...
- 任务状态：...
如果没有值得记忆的内容，返回"无"。"""
extracted = llm.generate(extraction_prompt)
if extracted != "无":
# 更新长期记忆
updated_memory = state["long_term_memory"].copy()
updated_memory["recent_extractions"] = extracted
# 存入向量数据库
vector_store.upsert(
collection="long_term",
content=latest_interaction,
metadata={"extraction": extracted}
)
return {"long_term_memory": updated_memory}
return {}
# 构建带记忆的Agent图
def build_memory_agent():
graph = StateGraph(AgentState)
# 添加节点
graph.add_node("retrieve_memory", memory_retrieval_node)
graph.add_node("reason", reasoning_node)
graph.add_node("consolidate_memory", memory_consolidation_node)
# 定义边
graph.set_entry_point("retrieve_memory")
graph.add_edge("retrieve_memory", "reason")
graph.add_edge("reason", "consolidate_memory")
graph.add_edge("consolidate_memory", END)
return graph.compile()
5.2 状态持久化
LangGraph支持将状态持久化到外部存储，实现跨会话的记忆保持：
from langgraph.checkpoint.sqlite import SqliteSaver
# 使用SQLite作为检查点存储
memory = SqliteSaver.from_conn_string(":memory:")
# 编译时绑定检查点
graph = build_memory_agent()
app = graph.compile(checkpointer=memory)
# 使用时指定thread_id实现会话隔离
config = {"configurable": {"thread_id": "user_123"}}
# 第一次对话
result1 = await app.ainvoke(
{"messages": [HumanMessage(content="我叫张三，喜欢Python")]},
config=config
)
# 第二次对话（状态自动恢复）
result2 = await app.ainvoke(
{"messages": [HumanMessage(content="我之前说我喜欢什么语言？")]},
config=config
)
# Agent能回忆起用户喜欢Python
6. 多Agent记忆共享
6.1 共享记忆架构
当多个Agent协作时，记忆共享成为关键挑战：
┌──────────────────────────────────────────────────────┐
│ 共享记忆层 │
│ ┌──────────────────────────────────────────────┐ │
│ │ 共享记忆总线 (Shared Memory Bus) │ │
│ │ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │任务状态 │ │共享知识 │ │协作历史 │ │ │
│ │ │(Task) │ │(KB) │ │(History)│ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ │ │
│ └──────────────────────────────────────────────┘ │
│ ▲ ▲ ▲ │
│ │ │ │ │
│ ┌─────┴──────┐ ┌─────┴──────┐ ┌─────┴──────┐ │
│ │ Agent A │ │ Agent B │ │ Agent C │ │
│ │ (规划) │ │ (执行) │ │ (验证) │ │
│ │ │ │ │ │ │ │
│ │ 私有记忆 │ │ 私有记忆 │ │ 私有记忆 │ │
│ │ + 共享视图 │ │ + 共享视图 │ │ + 共享视图 │ │
│ └────────────┘ └────────────┘ └────────────┘ │
└──────────────────────────────────────────────────────┘
6.2 实现方案
from enum import Enum
from dataclasses import dataclass, field
from typing import Set
import asyncio
class MemoryScope(Enum):
"""记忆作用域"""
PRIVATE = "private" # 仅创建者可见
SHARED = "shared" # 所有Agent可见
DELEGATED = "delegated" # 指定Agent可见
@dataclass
class SharedMemory:
"""共享记忆条目"""
id: str
content: str
creator: str # 创建者Agent ID
scope: MemoryScope
visible_to: Set[str] = field(default_factory=set) # DELEGATED模式下可见的Agent列表
version: int = 1
lock_holder: str = None # 当前持有写锁的Agent
class MultiAgentMemoryManager:
"""多Agent记忆管理器"""
def __init__(self):
self.shared_store: dict[str, SharedMemory] = {}
self.agent_stores: dict[str, dict] = {} # 每个Agent的私有存储
self._locks: dict[str, asyncio.Lock] = {}
def register_agent(self, agent_id: str):
"""注册一个Agent"""
self.agent_stores[agent_id] = {}
async def write_memory(
self,
agent_id: str,
content: str,
scope: MemoryScope = MemoryScope.PRIVATE,
visible_to: Set[str] = None
) -> str:
"""写入记忆"""
memory_id = f"mem_{agent_id}_{len(self.shared_store)}"
if scope == MemoryScope.PRIVATE:
# 私有记忆：仅存入Agent自己的存储
self.agent_stores[agent_id][memory_id] = {
"content": content,
"scope": scope.value
}
else:
# 共享/委托记忆：存入共享存储
self.shared_store[memory_id] = SharedMemory(
id=memory_id,
content=content,
creator=agent_id,
scope=scope,
visible_to=visible_to or set()
)
return memory_id
async def read_memory(
self,
agent_id: str,
query: str,
include_private: bool = True,
include_shared: bool = True
) -> list[dict]:
"""读取记忆——遵守访问控制"""
results = []
if include_private:
# 搜索自己的私有记忆
for mid, mem in self.agent_stores.get(agent_id, {}).items():
if self._relevance_score(query, mem["content"]) > 0.3:
results.append({"id": mid, **mem, "source": "private"})
if include_shared:
# 搜索共享记忆
for mid, mem in self.shared_store.items():
if self._can_access(agent_id, mem):
if self._relevance_score(query, mem.content) > 0.3:
results.append({
"id": mid,
"content": mem.content,
"creator": mem.creator,
"scope": mem.scope.value,
"source": "shared"
})
# 按相关性排序
results.sort(key=lambda x: self._relevance_score(query, x["content"]), reverse=True)
return results[:10]
def _can_access(self, agent_id: str, memory: SharedMemory) -> bool:
"""检查Agent是否有权访问该记忆"""
if memory.scope == MemoryScope.SHARED:
return True
if memory.scope == MemoryScope.DELEGATED:
return agent_id in memory.visible_to or agent_id == memory.creator
return False
async def sync_shared_context(self, agent_id: str) -> str:
"""同步共享上下文：生成Agent可见的所有共享记忆摘要"""
visible_memories = [
mem for mem in self.shared_store.values()
if self._can_access(agent_id, mem)
]
if not visible_memories:
return "暂无共享记忆。"
# 按时间排序，取最近的
visible_memories.sort(key=lambda m: m.version, reverse=True)
recent = visible_memories[:20]
summary = "共享记忆摘要：\n"
for mem in recent:
summary += f"- [{mem.creator}] {mem.content[:100]}...\n"
return summary
def _relevance_score(self, query: str, content: str) -> float:
"""简单的相关性评分（实际应用中应使用嵌入向量）"""
query_words = set(query.lower().split())
content_words = set(content.lower().split())
overlap = len(query_words & content_words)
return overlap / max(len(query_words), 1)
7. 记忆压缩与摘要
7.1 为什么需要压缩
随着交互的积累，记忆存储会快速增长。压缩和摘要机制确保：
- 存储成本可控
- 检索效率不随时间退化
- 重要信息不被海量细节淹没
7.2 分层压缩策略
class MemoryCompressor:
"""记忆压缩器——实现分层压缩策略"""
def __init__(self, llm, embedding_model):
self.llm = llm
self.embedding_model = embedding_model
async def compress_conversation(
self,
messages: list[dict],
target_length: int = 500
) -> str:
"""将对话压缩为目标长度的摘要"""
# 第一层：提取关键信息点
key_points = await self._extract_key_points(messages)
# 第二层：生成结构化摘要
summary = await self._generate_structured_summary(key_points, target_length)
# 第三层：压缩至目标长度
if len(summary) > target_length:
summary = await self._abstractive_compression(summary, target_length)
return summary
async def _extract_key_points(self, messages: list[dict]) -> list[str]:
"""从对话中提取关键信息点"""
conversation = "\n".join(
f"{m['role']}: {m['content']}" for m in messages
)
prompt = f"""从以下对话中提取所有关键信息点，每个点用一句话概括。
重点关注：
1. 用户表达的偏好或需求
2. 做出的决策或达成的共识
3. 待办事项或承诺
4. 重要的事实或数据
对话内容：
{conversation}
输出格式：每行一个关键点，以"- "开头。"""
response = await self.llm.agenerate(prompt)
return [line.strip("- ") for line in response.strip().split("\n") if line.startswith("- ")]
async def _generate_structured_summary(
self,
key_points: list[str],
target_length: int
) -> str:
"""生成结构化摘要"""
points_text = "\n".join(f"- {p}" for p in key_points)
prompt = f"""将以下关键点整合为一段结构化摘要（目标长度：{target_length}字）：
{points_points}
要求：
1. 保留所有重要信息
2. 合并重复或相似的点
3. 按主题分组组织
4. 使用简洁清晰的语言"""
return await self.llm.agenerate(prompt)
async def compress_memories_batch(
self,
memories: list[dict],
max_memories: int = 100
) -> list[dict]:
"""批量压缩记忆：将相似记忆合并"""
if len(memories) <= max_memories:
return memories
# 按语义聚类
embeddings = [
await self.embedding_model.embed(m["content"])
for m in memories
]
clusters = self._cluster_embeddings(embeddings, n_clusters=max_memories)
# 每个聚类生成一条合并记忆
compressed = []
for cluster_indices in clusters:
cluster_memories = [memories[i] for i in cluster_indices]
merged_content = await self._merge_memories(cluster_memories)
# 保留最高重要性的元数据
max_importance = max(m.get("importance", 0.5) for m in cluster_memories)
compressed.append({
"content": merged_content,
"importance": max_importance,
"merged_from": len(cluster_memories),
"original_ids": [m["id"] for m in cluster_memories]
})
return compressed
async def _merge_memories(self, memories: list[dict]) -> str:
"""将多条相关记忆合并为一条"""
contents = "\n".join(f"- {m['content']}" for m in memories)
prompt = f"""将以下相关记忆合并为一条简洁的记忆：
{contents}
要求：保留所有关键信息，去除重复，生成一条综合记忆。"""
return await self.llm.agenerate(prompt)
7.3 渐进式摘要
class ProgressiveSummarizer:
"""渐进式摘要：按时间层级逐步压缩"""
LEVELS = {
"raw": None, # 原始对话
"level_1": 0.5, # 第一层：保留50%
"level_2": 0.25, # 第二层：保留25%
"level_3": 0.1, # 第三层：保留10%
}
async def summarize_timeline(
self,
memories: list[dict],
llm
) -> dict[str, list[dict]]:
"""按时间层级进行渐进式摘要"""
now = datetime.now()
result = {"recent": [], "day_level": [], "week_level": [], "month_level": []}
for memory in memories:
age = now - memory["created_at"]
if age < timedelta(hours=6):
# 最近6小时：保留原始内容
result["recent"].append(memory)
elif age < timedelta(days=1):
# 6-24小时：第一层压缩
memory["content"] = await self._compress(memory["content"], 0.5, llm)
result["day_level"].append(memory)
elif age < timedelta(days=7):
# 1-7天：第二层压缩
memory["content"] = await self._compress(memory["content"], 0.25, llm)
result["week_level"].append(memory)
else:
# 7天以上：第三层压缩
memory["content"] = await self._compress(memory["content"], 0.1, llm)
result["month_level"].append(memory)
return result
async def _compress(self, content: str, ratio: float, llm) -> str:
"""按比例压缩内容"""
target = int(len(content) * ratio)
if target < 50:
return content[:50] + "..."
prompt = f"""将以下内容压缩至约{target}字，保留核心信息：
{content}"""
return await llm.agenerate(prompt)
8. 持久化存储策略
8.1 存储层选型
| 存储层 | 用途 | 推荐方案 | 读写延迟 |
|---|---|---|---|
| 热存储 | 工作记忆、近期对话 | Redis / 内存 | <1ms |
| 温存储 | 短期记忆、会话历史 | PostgreSQL + pgvector | 1-10ms |
| 冷存储 | 归档记忆、历史数据 | S3 + 向量数据库 | 10-100ms |
| 元数据 | 记忆索引、用户画像 | PostgreSQL / MongoDB | 1-5ms |
8.2 多层存储实现
import json
import hashlib
from datetime import datetime, timedelta
from typing import Optional
class TieredMemoryStorage:
"""分层存储管理器"""
def __init__(self, redis_client, pg_pool, s3_client, vector_store):
self.redis = redis_client
self.pg = pg_pool
self.s3 = s3_client
self.vector_store = vector_store
async def write(self, memory: dict):
"""写入记忆——根据热度分层存储"""
tier = self._determine_tier(memory)
# 1. 写入向量索引（所有层都写）
await self.vector_store.upsert(
collection="memory_index",
id=memory["id"],
vector=memory["embedding"],
metadata={
"tier": tier,
"created_at": memory["created_at"].isoformat(),
"importance": memory.get("importance", 0.5),
"type": memory.get("type", "episodic"),
}
)
# 2. 根据层级写入对应存储
if tier == "hot":
await self._write_hot(memory)
elif tier == "warm":
await self._write_warm(memory)
else:
await self._write_cold(memory)
async def read(self, memory_id: str) -> Optional[dict]:
"""读取记忆——先热后冷，自动提升热度"""
# 尝试从热存储读取
result = await self._read_hot(memory_id)
if result:
return result
# 尝试从温存储读取
result = await self._read_warm(memory_id)
if result:
# 热度提升：写入热存储
await self._write_hot(result)
return result
# 从冷存储读取
result = await self._read_cold(memory_id)
if result:
# 写入温存储
await self._write_warm(result)
return result
return None
async def search(self, query_embedding: list, top_k: int = 10) -> list[dict]:
"""跨层检索——向量索引定位，分层读取"""
# 1. 向量索引搜索
index_results = await self.vector_store.search(
collection="memory_index",
query_vector=query_embedding,
top_k=top_k
)
# 2. 从对应层级读取完整内容
results = []
for hit in index_results:
memory_id = hit["id"]
tier = hit["metadata"]["tier"]
if tier == "hot":
content = await self._read_hot(memory_id)
elif tier == "warm":
content = await self._read_warm(memory_id)
else:
content = await self._read_cold(memory_id)
if content:
results.append({
**content,
"relevance_score": hit["score"],
"storage_tier": tier
})
return results
def _determine_tier(self, memory: dict) -> str:
"""判断记忆应存储在哪个层级"""
age = datetime.now() - memory["created_at"]
importance = memory.get("importance", 0.5)
if age < timedelta(hours=24) or importance > 0.8:
return "hot"
elif age < timedelta(days=30) or importance > 0.5:
return "warm"
else:
return "cold"
async def _write_hot(self, memory: dict):
"""写入Redis"""
key = f"memory:{memory['id']}"
await self.redis.setex(
key,
timedelta(hours=24),
json.dumps(memory, default=str)
)
async def _write_warm(self, memory: dict):
"""写入PostgreSQL"""
async with self.pg.acquire() as conn:
await conn.execute("""
INSERT INTO memories (id, content, metadata, embedding, created_at)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (id) DO UPDATE SET
content = EXCLUDED.content,
metadata = EXCLUDED.metadata
""",
memory["id"],
memory["content"],
json.dumps(memory.get("metadata", {})),
memory.get("embedding"),
memory["created_at"]
)
async def _write_cold(self, memory: dict):
"""写入S3"""
key = f"memories/{memory['id']}.json"
await self.s3.put_object(
Bucket="agent-memories",
Key=key,
Body=json.dumps(memory, default=str),
ContentType="application/json"
)
async def _read_hot(self, memory_id: str) -> Optional[dict]:
data = await self.redis.get(f"memory:{memory_id}")
return json.loads(data) if data else None
async def _read_warm(self, memory_id: str) -> Optional[dict]:
async with self.pg.acquire() as conn:
row = await conn.fetchrow(
"SELECT * FROM memories WHERE id = $1", memory_id
)
return dict(row) if row else None
async def _read_cold(self, memory_id: str) -> Optional[dict]:
try:
resp = await self.s3.get_object(
Bucket="agent-memories",
Key=f"memories/{memory_id}.json"
)
return json.loads(await resp["Body"].read())
except Exception:
return None
9. 记忆检索优化
9.1 检索质量评估指标
| 指标 | 含义 | 目标值 |
|---|---|---|
| Recall@K | 前K个结果中包含相关记忆的比例 | >0.85 |
| MRR | 第一个相关结果的排名倒数均值 | >0.7 |
| Precision@K | 前K个结果中相关结果的比例 | >0.6 |
| NDCG | 考虑排名位置的增益累积 | >0.75 |
| 延迟P95 | 95分位检索延迟 | <100ms |
9.2 查询改写
用户输入往往不够精确，查询改写可以显著提升检索效果：
class QueryRewriter:
"""查询改写器：优化检索查询"""
def __init__(self, llm):
self.llm = llm
async def rewrite(self, query: str, context: list[dict] = None) -> list[str]:
"""生成多个改写查询，提高召回率"""
context_str = ""
if context:
recent = context[-3:] # 最近3轮对话
context_str = "\n".join(f"{m['role']}: {m['content']}" for m in recent)
prompt = f"""将以下用户查询改写为3个不同的检索查询，以提高记忆检索的召回率。
用户查询：{query}
对话上下文：
{context_str}
改写要求：
1. 第一个：扩展同义词和相关概念
2. 第二个：提取核心实体和关键事实
3. 第三个：从用户意图角度改写
输出格式（每行一个）：
1. ...
2. ...
3. ..."""
response = await self.llm.agenerate(prompt)
rewrites = [
line.split(". ", 1)[1] if ". " in line else line
for line in response.strip().split("\n")
if line.strip()
]
return [query] + rewrites # 包含原始查询
async def decompose(self, complex_query: str) -> list[str]:
"""复杂查询分解：将多跳问题拆解为子查询"""
prompt = f"""将以下复杂查询分解为可以独立检索的子查询：
复杂查询：{complex_query}
分解原则：
1. 每个子查询应该能独立从记忆库中检索到答案
2. 保持语义完整性
3. 最多分解为5个子查询
输出格式（每行一个子查询）："""
response = await self.llm.agenerate(prompt)
return [line.strip() for line in response.strip().split("\n") if line.strip()]
9.3 重排序（Reranking）
from sentence_transformers import CrossEncoder
class MemoryReranker:
"""记忆重排序器：使用交叉编码器精排"""
def __init__(self, model_name: str = "BAAI/bge-reranker-large"):
self.model = CrossEncoder(model_name)
async def rerank(
self,
query: str,
candidates: list[dict],
top_k: int = 5
) -> list[dict]:
"""对候选记忆进行精排"""
if not candidates:
return []
# 构造(query, candidate)对
pairs = [(query, c["content"]) for c in candidates]
# 计算相关性分数
scores = self.model.predict(pairs)
# 按分数排序
scored_candidates = [
{**candidate, "rerank_score": float(score)}
for candidate, score in zip(candidates, scores)
]
scored_candidates.sort(key=lambda x: x["rerank_score"], reverse=True)
return scored_candidates[:top_k]
class RetrievalPipeline:
"""完整检索管道：查询改写 → 向量检索 → 重排序"""
def __init__(self, rewriter, vector_store, reranker, compressor):
self.rewriter = rewriter
self.vector_store = vector_store
self.reranker = reranker
self.compressor = compressor
async def retrieve(
self,
query: str,
context: list[dict] = None,
top_k: int = 5
) -> list[dict]:
# Step 1: 查询改写
queries = await self.rewriter.rewrite(query, context)
# Step 2: 多查询向量检索
all_candidates = []
seen_ids = set()
for q in queries:
embedding = await self._compute_embedding(q)
results = await self.vector_store.search(
collection="memory_index",
query_vector=embedding,
top_k=top_k * 2
)
for r in results:
if r["id"] not in seen_ids:
all_candidates.append(r)
seen_ids.add(r["id"])
# Step 3: 重排序
reranked = await self.reranker.rerank(query, all_candidates, top_k)
# Step 4: 上下文压缩（只保留与查询相关的部分）
compressed = []
for mem in reranked:
relevant_part = await self.compressor.extract_relevant(
query, mem["content"]
)
compressed.append({**mem, "content": relevant_part})
return compressed
10. 隐私与安全考量
10.1 威胁模型
| 威胁 | 描述 | 风险等级 | 防御措施 |
|---|---|---|---|
| 记忆泄露 | Agent向非授权方暴露私有记忆 | 高 | 访问控制+加密 |
| 记忆注入 | 恶意输入写入有害记忆 | 高 | 输入验证+内容过滤 |
| 记忆投毒 | 通过多轮交互逐渐污染记忆 | 中 | 记忆审计+版本控制 |
| 推断攻击 | 通过行为模式推断敏感信息 | 中 | 差分隐私 |
| 存储泄露 | 持久化存储被非法访问 | 高 | 加密存储+访问控制 |
10.2 安全实现
from cryptography.fernet import Fernet
import hashlib
import re
class SecureMemoryManager:
"""安全记忆管理器"""
def __init__(self, encryption_key: bytes, content_filter):
self.cipher = Fernet(encryption_key)
self.content_filter = content_filter
self.audit_log = []
async def secure_write(self, user_id: str, content: str, memory: dict) -> dict:
"""安全写入记忆"""
# 1. 内容过滤：检测并屏蔽敏感信息
sanitized = await self._sanitize_content(content)
# 2. 加密存储
encrypted_content = self.cipher.encrypt(sanitized.encode()).decode()
# 3. 审计日志
self.audit_log.append({
"action": "write",
"user_id": user_id,
"memory_id": memory["id"],
"timestamp": datetime.now().isoformat(),
"content_hash": hashlib.sha256(sanitized.encode()).hexdigest(),
})
return {**memory, "content": encrypted_content}
async def secure_read(self, user_id: str, memory: dict) -> str:
"""安全读取记忆"""
# 1. 权限验证
if not self._check_access(user_id, memory):
raise PermissionError("无权访问该记忆")
# 2. 解密
decrypted = self.cipher.decrypt(memory["content"].encode()).decode()
# 3. 审计
self.audit_log.append({
"action": "read",
"user_id": user_id,
"memory_id": memory["id"],
"timestamp": datetime.now().isoformat(),
})
return decrypted
async def _sanitize_content(self, content: str) -> str:
"""敏感信息脱敏"""
# 手机号
content = re.sub(r'1[3-9]\d{9}', '[手机号已脱敏]', content)
# 身份证号
content = re.sub(
r'[1-9]\d{5}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx]',
'[身份证已脱敏]', content
)
# 邮箱
content = re.sub(
r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
'[邮箱已脱敏]', content
)
# 银行卡号
content = re.sub(r'\d{16,19}', '[银行卡号已脱敏]', content)
# LLM辅助检测更复杂的敏感信息
if await self.content_filter.contains_sensitive(content):
content = await self.content_filter.redact(content)
return content
def _check_access(self, user_id: str, memory: dict) -> bool:
"""检查用户是否有权访问该记忆"""
# 记忆所有者可以访问
if memory.get("owner") == user_id:
return True
# 共享记忆的授权用户可以访问
if user_id in memory.get("shared_with", []):
return True
# 管理员可以访问（需要额外验证）
if memory.get("admin_accessible") and self._is_admin(user_id):
return True
return False
async def memory_retention_policy(self, user_id: str):
"""记忆保留策略：定期清理过期记忆"""
retention_days = 90 # 默认保留90天
cutoff = datetime.now() - timedelta(days=retention_days)
expired = await self._find_expired_memories(user_id, cutoff)
for memory in expired:
# 根据策略决定删除还是归档
if memory.get("importance", 0) > 0.8:
await self._archive_memory(memory)
else:
await self._delete_memory(memory)
self.audit_log.append({
"action": "retention_cleanup",
"user_id": user_id,
"expired_count": len(expired),
"timestamp": datetime.now().isoformat(),
})
11. 生产级记忆系统架构
11.1 完整架构图
┌─────────────────────────────────────────────────────────────────────┐
│ 生产级Agent记忆系统 │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 接入层 (API Gateway) │ │
│ │ 认证 → 限流 → 路由 → 请求/响应日志 │ │
│ └──────────────────────────┬──────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────▼──────────────────────────────────┐ │
│ │ 记忆服务层 (Memory Service) │ │
│ │ │ │
│ │ ┌───────────┐ ┌───────────┐ ┌───────────┐ │ │
│ │ │写入管道 │ │检索管道 │ │压缩管道 │ │ │
│ │ │Write │ │Retrieve │ │Compress │ │ │
│ │ │Pipeline │ │Pipeline │ │Pipeline │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ │·内容过滤 │ │·查询改写 │ │·定时触发 │ │ │
│ │ │·向量计算 │ │·向量检索 │ │·聚类合并 │ │ │
│ │ │·存储路由 │ │·重排序 │ │·摘要生成 │ │ │
│ │ │·索引更新 │ │·上下文压缩│ │·过期清理 │ │ │
│ │ └───────────┘ └───────────┘ └───────────┘ │ │
│ └──────────────────────────┬──────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────▼──────────────────────────────────┐ │
│ │ 存储层 (Storage Layer) │ │
│ │ │ │
│ │ ┌────────┐ ┌────────────┐ ┌────────┐ ┌──────────┐ │ │
│ │ │ Redis │ │ PostgreSQL │ │Qdrant │ │ S3 │ │ │
│ │ │ 热存储 │ │ 温存储 │ │向量索引 │ │ 冷存储 │ │ │
│ │ └────────┘ └────────────┘ └────────┘ └──────────┘ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 监控层 (Observability) │ │
│ │ Prometheus指标 → Grafana仪表板 → 告警规则 │ │
│ └─────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
11.2 关键设计决策
决策1：嵌入模型选型
| 模型 | 维度 | 中文效果 | 延迟 | 推荐场景 |
|---|---|---|---|---|
| OpenAI text-embedding-3-large | 3072 | 优秀 | 中 | 通用场景，质量优先 |
| OpenAI text-embedding-3-small | 1536 | 良好 | 低 | 成本敏感场景 |
| BAAI/bge-large-zh-v1.5 | 1024 | 优秀 | 低 | 中文为主，自托管 |
| sentence-transformers/all-MiniLM-L6-v2 | 384 | 一般 | 极低 | 原型开发 |
决策2：向量数据库选型
def choose_vector_db(requirements: dict) -> str:
"""根据需求选择向量数据库"""
if requirements.get("managed"):
return "Pinecone" # 全托管，零运维
if requirements.get("scale") == "billion+":
return "Milvus" # 超大规模
if requirements.get("hybrid_search"):
return "Weaviate" # 原生混合搜索
if requirements.get("low_latency"):
return "Qdrant" # 高性能
if requirements.get("simple"):
return "ChromaDB" # 开发原型
return "Qdrant" # 默认推荐
决策3：压缩时机
COMPRESSION_TRIGGERS = {
"time_based": {
"description": "定时压缩，如每天凌晨",
"schedule": "0 2 * * *",
"suitable": "对话量稳定的系统"
},
"threshold_based": {
"description": "记忆数量超过阈值时触发",
"threshold": 10000,
"suitable": "对话量波动大的系统"
},
"event_based": {
"description": "特定事件触发，如用户登出、会话结束",
"events": ["session_end", "user_logout"],
"suitable": "实时性要求高的系统"
}
}
11.3 性能基准
以下是一个生产级记忆系统的典型性能指标：
PERFORMANCE_BENCHMARKS = {
"写入延迟": {
"P50": "5ms", # 热存储写入
"P95": "20ms", # 含向量计算
"P99": "50ms", # 含压缩和索引
},
"检索延迟": {
"P50": "15ms", # 向量检索
"P95": "50ms", # 含重排序
"P99": "100ms", # 含跨层查询
},
"吞吐量": {
"写入": "1000 QPS", # 单节点
"检索": "5000 QPS", # 单节点
"压缩": "100 条/秒", # 批量压缩
},
"存储效率": {
"向量索引": "~6KB/条 (1536维 float32)",
"元数据": "~1KB/条",
"压缩率": "原始对话的10-20%",
}
}
11.4 部署建议
# docker-compose.yml - 记忆系统部署配置
version: "3.8"
services:
memory-service:
build: ./memory-service
ports:
- "8080:8080"
environment:
- REDIS_URL=redis://redis:6379
- PG_URL=postgresql://postgres:5432/memories
- QDRANT_URL=http://qdrant:6333
- EMBEDDING_MODEL=BAAI/bge-large-zh-v1.5
- ENCRYPTION_KEY_FILE=/run/secrets/encryption_key
secrets:
- encryption_key
deploy:
replicas: 3
resources:
limits:
cpus: "2"
memory: 4G
redis:
image: redis:7-alpine
volumes:
- redis-data:/data
command: redis-server --maxmemory 2gb --maxmemory-policy allkeys-lru
postgres:
image: pgvector/pgvector:pg16
volumes:
- pg-data:/var/lib/postgresql/data
environment:
POSTGRES_DB: memories
POSTGRES_PASSWORD_FILE: /run/secrets/pg_password
qdrant:
image: qdrant/qdrant:latest
volumes:
- qdrant-data:/qdrant/storage
ports:
- "6333:6333"
volumes:
redis-data:
pg-data:
qdrant-data:
secrets:
encryption_key:
external: true
pg_password:
external: true
总结
构建生产级Agent记忆系统需要在多个维度做出平衡：
| 维度 | 权衡点 | 建议 |
|---|---|---|
| 容量 vs 成本 | 记忆越多越好，但存储和计算成本线性增长 | 分层存储+定期压缩 |
| 检索精度 vs 延迟 | 精排提升精度但增加延迟 | 向量粗排+交叉编码器精排 |
| 隐私 vs 功能 | 完全加密影响语义检索能力 | 存储加密+检索时解密 |
| 共享 vs 隔离 | 共享记忆提升协作效率，但增加泄露风险 | 细粒度访问控制 |
| 持久化 vs 实时性 | 持久化保证可靠性，但增加写入延迟 | 异步写入+缓存 |
核心原则：
- 记忆即状态：将记忆系统视为Agent的核心状态管理器，而非附属功能
- 分层设计：不同温度的记忆使用不同的存储和检索策略
- 可观测性：每条记忆的生命周期都应可追踪、可审计
- 渐进演化：从简单的对话历史开始，逐步添加向量检索、压缩、共享等高级功能
- 安全优先：在设计阶段就纳入隐私和安全考量，而非事后补救
记忆系统是Agent从"无状态工具"进化为"有记忆的伙伴"的关键。掌握本教程中的架构思想和实现方案，你将能够构建出真正具有持久记忆、跨会话连续性和持续学习能力的AI Agent系统。
本教程最后更新：2024年。记忆系统技术仍在快速演进，建议持续关注学术论文和开源项目（如MemGPT、LangGraph、Mem0等）的最新进展。
