AI Agent 记忆系统设计指南

开场

我的社群 Q&A Agent 上线第一个月，同一个成员问了三次"社群的线下活动怎么报名"，Agent 每次都像第一次听到一样从头解释。这不是 Agent 能力的问题，是记忆系统的缺失。加了记忆层之后，Agent 第二次能直接说"和上次一样，链接在这里，你上次报名的是 3 月的活动"。回答质量评分从 7.2 提升到 9.6。记忆是 AI Agent 从"工具"变成"助手"的关键分界线。

问题背景

LLM 本身是无状态的——每次请求都是独立的，模型不记得上一次对话说了什么。目前主流的"记忆"实现方式是把历史对话塞进 prompt（即 conversation history），但这种方式有三个硬伤：

1. 上下文窗口有限：即使是 200K token 的窗口，几十轮对话就满了
2. 成本线性增长：历史越长，每次请求的 token 消耗越大
3. 没有长期积累：对话结束，记忆消失

真正的记忆系统需要解决三个问题：记什么、怎么存、怎么取。

核心架构：三层记忆模型

我把 Agent 的记忆分成三层，模拟人类的记忆系统：

┌─────────────────────────────────────────┐
│  短期记忆 (Working Memory)               │
│  当前对话的上下文窗口                      │
│  存储：内存 / Message Array              │
│  保留时间：当前会话                        │
├─────────────────────────────────────────┤
│  长期记忆 (Long-term Memory)             │
│  跨会话的知识和事实                        │
│  存储：Vector Store (Qdrant/Pinecone)    │
│  保留时间：永久                           │
├─────────────────────────────────────────┤
│  情景记忆 (Episodic Memory)              │
│  用户画像、偏好、历史行为                   │
│  存储：结构化数据库 (PostgreSQL)           │
│  保留时间：永久，定期更新                   │
└─────────────────────────────────────────┘

第一层：短期记忆 — 对话窗口管理

最基础的记忆，也是最容易做错的。核心问题：对话太长时怎么压缩？

from dataclasses import dataclass, field
from openai import OpenAI

client = OpenAI()

@dataclass
class WorkingMemory:
    """短期记忆：管理当前对话的上下文窗口"""
    messages: list[dict] = field(default_factory=list)
    max_tokens: int = 8000       # 短期记忆的 token 上限
    summary_threshold: int = 6000 # 超过此值触发压缩

    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        # 检查是否需要压缩
        if self._estimate_tokens() > self.summary_threshold:
            self._compress()

    def _compress(self):
        """压缩策略：保留最近 3 轮 + 历史摘要"""
        # 保留 system prompt（第一条）和最近 3 轮对话（最后 6 条）
        system = self.messages[0] if self.messages[0]["role"] == "system" else None
        recent = self.messages[-6:]
        old = self.messages[1:-6] if system else self.messages[:-6]

        if not old:
            return

        # 用小模型生成历史摘要
        summary_response = client.chat.completions.create(
            model="gpt-4.1-mini",  # 压缩用便宜模型
            messages=[
                {"role": "system", "content": "将以下对话压缩为关键信息摘要，保留重要的事实、决策和用户偏好。不超过 200 字。"},
                {"role": "user", "content": str(old)}
            ],
            max_tokens=300,
        )
        summary = summary_response.choices[0].message.content

        # 重组消息列表
        self.messages = []
        if system:
            self.messages.append(system)
        self.messages.append({
            "role": "system",
            "content": f"[历史对话摘要] {summary}"
        })
        self.messages.extend(recent)

    def _estimate_tokens(self) -> int:
        """粗略估算 token 数（中文约 1.5 字/token）"""
        total_chars = sum(len(m["content"]) for m in self.messages)
        return int(total_chars * 0.7)  # 中英混合的粗略系数

关键设计决策：

• 压缩阈值设在 max_tokens 的 75% 左右，给新消息留空间
• 摘要用 GPT-4.1-mini，成本约 $0.0003/次压缩
• 保留最近 3 轮原文（用户对最近的对话记忆最清晰，压缩会丢失细节）

第二层：长期记忆 — Vector Store

跨会话的知识积累。每次对话中产生的有价值信息，写入 vector store；下次对话开始时，根据当前话题检索相关记忆。

from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct
from openai import OpenAI
from uuid import uuid4
from datetime import datetime

class LongTermMemory:
    """长期记忆：基于 Vector Store 的跨会话知识库"""

    def __init__(self):
        self.qdrant = QdrantClient(url="http://localhost:6333")
        self.openai = OpenAI()
        self.collection = "agent_memory"

    async def remember(self, content: str, metadata: dict):
        """存入新记忆"""
        # 生成 embedding
        embedding = self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=content
        ).data[0].embedding

        # 存入 vector store
        self.qdrant.upsert(
            collection_name=self.collection,
            points=[PointStruct(
                id=str(uuid4()),
                vector=embedding,
                payload={
                    "content": content,
                    "timestamp": datetime.now().isoformat(),
                    "access_count": 0,    # 被检索次数
                    "importance": metadata.get("importance", 0.5),
                    **metadata
                }
            )]
        )

    async def recall(self, query: str, top_k: int = 5,
                     min_relevance: float = 0.75) -> list[dict]:
        """检索相关记忆"""
        query_vector = self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=query
        ).data[0].embedding

        results = self.qdrant.query_points(
            collection_name=self.collection,
            query=query_vector,
            limit=top_k,
            score_threshold=min_relevance,
        )

        memories = []
        for r in results.points:
            memories.append({
                "content": r.payload["content"],
                "relevance": r.score,
                "timestamp": r.payload["timestamp"],
                "importance": r.payload["importance"],
            })
            # 更新访问计数
            self._update_access_count(r.id)

        return memories

    async def forget(self, older_than_days: int = 90,
                     min_access_count: int = 0):
        """遗忘机制：清理低价值记忆"""
        cutoff = datetime.now() - timedelta(days=older_than_days)
        # 删除超过 90 天且从未被检索过的记忆
        self.qdrant.delete(
            collection_name=self.collection,
            points_selector=Filter(
                must=[
                    FieldCondition(key="timestamp", range=Range(lt=cutoff.isoformat())),
                    FieldCondition(key="access_count", range=Range(lte=min_access_count)),
                    FieldCondition(key="importance", range=Range(lt=0.8)),
                ]
            )
        )

遗忘机制很重要。没有遗忘的记忆系统会无限膨胀，检索质量会随着数据量增加而下降。我的策略是：90 天未被检索且 importance < 0.8 的记忆自动清理。

第三层：情景记忆 — 用户画像

结构化的用户信息，用传统数据库存储。

from dataclasses import dataclass
from datetime import datetime
import json

@dataclass
class UserProfile:
    """情景记忆：结构化的用户画像"""
    user_id: str
    name: str
    interests: list[str]           # 感兴趣的话题
    expertise_level: str           # beginner / intermediate / expert
    preferred_language: str        # zh / en
    interaction_count: int         # 总互动次数
    last_topics: list[str]         # 最近讨论的话题
    preferences: dict              # 个性化偏好
    first_seen: datetime
    last_seen: datetime

class EpisodicMemory:
    """情景记忆管理器"""

    def __init__(self, db_connection):
        self.db = db_connection

    async def update_profile(self, user_id: str,
                             conversation: list[dict]):
        """从对话中提取信息更新用户画像"""
        profile = await self.get_profile(user_id)

        # 用 LLM 从对话中提取结构化信息
        extraction = await client.chat.completions.create(
            model="gpt-4.1-mini",
            messages=[
                {"role": "system", "content": """从对话中提取用户信息，输出 JSON：
{
    "new_interests": ["话题1", "话题2"],
    "expertise_indicators": "beginner/intermediate/expert",
    "preferences": {"key": "value"},
    "key_facts": ["重要事实1", "重要事实2"]
}
只提取明确提到的信息，不推测。"""},
                {"role": "user", "content": json.dumps(conversation, ensure_ascii=False)}
            ],
            response_format={"type": "json_object"},
        )
        updates = json.loads(extraction.choices[0].message.content)

        # 增量更新画像
        profile.interests = list(set(profile.interests + updates.get("new_interests", [])))
        profile.last_topics = updates.get("new_interests", profile.last_topics)
        profile.interaction_count += 1
        profile.last_seen = datetime.now()

        await self.save_profile(profile)

    async def get_context_for_agent(self, user_id: str) -> str:
        """生成给 Agent 用的用户上下文"""
        profile = await self.get_profile(user_id)
        return f"""用户信息：
- 名字：{profile.name}
- 关注领域：{', '.join(profile.interests[-5:])}
- 专业水平：{profile.expertise_level}
- 互动次数：{profile.interaction_count}
- 最近话题：{', '.join(profile.last_topics[-3:])}
- 偏好：{json.dumps(profile.preferences, ensure_ascii=False)}"""

实现细节：三层记忆的整合

class AgentMemorySystem:
    """整合三层记忆的完整系统"""

    def __init__(self):
        self.working = WorkingMemory()
        self.long_term = LongTermMemory()
        self.episodic = EpisodicMemory(db)

    async def prepare_context(self, user_id: str,
                               user_message: str) -> list[dict]:
        """为 Agent 准备完整的上下文（三层记忆融合）"""

        # 1. 情景记忆：获取用户画像
        user_context = await self.episodic.get_context_for_agent(user_id)

        # 2. 长期记忆：检索相关知识
        memories = await self.long_term.recall(user_message, top_k=3)
        memory_text = "\n".join([
            f"- [{m['timestamp'][:10]}] {m['content']}"
            for m in memories
        ]) if memories else "无相关历史记忆"

        # 3. 短期记忆：当前对话历史
        self.working.add("user", user_message)

        # 4. 组装完整上下文
        system_prompt = f"""你是社群助手。

{user_context}

相关历史记忆：
{memory_text}

根据以上信息回答用户问题。如果历史记忆中有相关内容，引用它。"""

        messages = [{"role": "system", "content": system_prompt}]
        messages.extend(self.working.messages[1:])  # 跳过原 system prompt
        return messages

    async def post_response(self, user_id: str,
                            user_message: str, agent_response: str):
        """回复后的记忆更新"""
        # 更新短期记忆
        self.working.add("assistant", agent_response)

        # 判断是否值得存入长期记忆
        if await self._is_worth_remembering(user_message, agent_response):
            await self.long_term.remember(
                content=f"Q: {user_message}\nA: {agent_response}",
                metadata={"user_id": user_id, "importance": 0.6}
            )

        # 更新用户画像
        await self.episodic.update_profile(
            user_id,
            [{"role": "user", "content": user_message},
             {"role": "assistant", "content": agent_response}]
        )

实战经验

生产数据

8-Agent 社群系统加了记忆层后的对比：

指标	无记忆	有记忆	变化
Q&A 满意度	72%	89%	+17%
重复问题的回答质量	6.8/10	9.6/10	+41%
平均响应 token	380	290	-24%
每条消息成本	$0.014	$0.016	+14%
用户回访率	34%	52%	+53%

记忆层增加的成本（+14%）远低于它带来的价值（满意度 +17%，回访率 +53%）。而且回答 token 反而减少了，因为有了记忆后 Agent 不需要每次从头解释。

成本拆解

组件	月成本
Embedding 生成 (text-embedding-3-small)	$1.80
Qdrant Cloud (长期记忆)	$16.00
PostgreSQL (情景记忆)	$5.00
短期记忆压缩 (GPT-4.1-mini)	$2.40
画像提取 (GPT-4.1-mini)	$3.60
合计	$28.80

踩过的坑

坑 1：什么都存。最初把每句对话都写入长期记忆，三周后 vector store 里有 50,000 条记录，检索质量断崖式下降。解决方案：加入 _is_worth_remembering 过滤器，只存有实质性信息的内容，记录量降到 8,000 条，检索准确率回到 90%+。

坑 2：短期记忆的压缩丢信息。有用户说"我刚才说的那个价格"，但那条消息已经被压缩成摘要了，细节丢了。解决方案：压缩时对包含数字、价格、日期等关键信息的消息做特殊处理，保留原文。

坑 3：用户画像的更新延迟。画像更新是异步的，偶尔出现 Agent 在同一轮对话里用了过期的画像。解决方案：把画像更新改成同步（在生成回复之前更新），增加了约 200ms 延迟但保证了一致性。

总结

三条 takeaway：

1. 三层记忆缺一不可——短期记忆保证对话连贯，长期记忆积累知识，情景记忆理解用户。只做短期记忆的 Agent 是金鱼，只做长期记忆的 Agent 不认识用户
2. 遗忘和记忆一样重要——没有遗忘机制的记忆系统会变成垃圾场。定期清理低价值记忆，保持检索质量
3. 记忆层的投入产出比极高——月成本 $28.80 换来满意度 +17% 和回访率 +53%。如果你的 Agent 还没有记忆层，这是优先级最高的优化

如果你在搭 Agent 系统，先加短期记忆（最简单，2 小时搞定），再加长期记忆（半天），最后加情景记忆（1-2 天）。按这个顺序来，每一步都能看到效果。

你的 Agent 有记忆吗？用了什么存储方案？来一人独角兽俱乐部分享。