AI Agent记忆系统：从短期上下文到长期知识库的工程实践

1. 为什么 Agent 需要记忆系统

当前大语言模型（LLM）虽然展现出惊人的推理能力，但其上下文窗口本质上是”短期记忆”——一旦对话超出窗口限制，模型就会”遗忘”之前的所有交互。对于需要长时间运行的 AI Agent（如代码助手、客服机器人、个人助理），这种遗忘是致命的。

一个典型的 Agent 记忆挑战场景：

• 上下文窗口溢出：GPT-4 Turbo 有 128K token，Claude 有 200K，但一个持续运行的 Agent 每天可能产生数百万 token 的交互
• 跨会话连续性：用户昨天说”我的项目 deadline 是下周五”，今天 Agent 应该还记得
• 个性化服务：Agent 需要记住用户的偏好、习惯、专业背景，而不是每次都从零开始
• 知识积累：Agent 应该能从过去的错误和成功中学习，而不是重复犯错

受认知科学启发，人类记忆系统分为工作记忆、情景记忆、语义记忆和程序性记忆四个层次。这种分层模型同样适用于 AI Agent，每一层对应不同的技术实现和工程挑战。

2. 记忆系统的四层架构模型

我们提出一个四层记忆架构，每一层有明确的职责、存储介质和生命周期：

这四层并非孤立运行，而是由一个记忆调度引擎统一协调，根据当前任务动态决定从哪一层读取记忆、向哪一层写入记忆。

3. 工作记忆：上下文窗口的智能管理

工作记忆是 Agent 的”当前思维空间”，直接对应 LLM 的上下文窗口。核心挑战在于：如何在有限的窗口内放入最有价值的信息。

3.1 智能上下文压缩

当对话历史超过阈值时，简单的截断会丢失关键信息。我们实现一个基于重要性的压缩策略：

import tiktoken
from dataclasses import dataclass, field
from typing import List, Optional
import time

@dataclass
class MemoryItem:
    role: str                    # user / assistant / system / tool
    content: str
    timestamp: float = field(default_factory=time.time)
    importance: float = 0.5     # 0.0 - 1.0
    token_count: int = 0
    metadata: dict = field(default_factory=dict)

class WorkingMemory:
    def __init__(self, max_tokens: int = 8000, model: str = "gpt-4"):
        self.max_tokens = max_tokens
        self.model = model
        self.items: List[MemoryItem] = []
        self.tokenizer = tiktoken.encoding_for_model(model)
        self._system_prompt_tokens = 0

    def add(self, item: MemoryItem) -> Optional[str]:
        """添加记忆项，如果超出限制则触发压缩"""
        item.token_count = len(self.tokenizer.encode(item.content))
        self.items.append(item)
        
        current_tokens = self._total_tokens()
        if current_tokens > self.max_tokens:
            return self._compress()
        return None

    def _total_tokens(self) -> int:
        return self._system_prompt_tokens + sum(
            item.token_count for item in self.items
        )

    def _compress(self) -> str:
        """基于重要性和时间的智能压缩"""
        # 计算每个记忆项的保留分数
        now = time.time()
        scored_items = []
        for item in self.items:
            age_hours = (now - item.timestamp) / 3600
            recency = 1.0 / (1.0 + age_hours)  # 时间衰减
            score = item.importance * 0.6 + recency * 0.4
            scored_items.append((score, item))

        # 按分数排序，保留高分项
        scored_items.sort(key=lambda x: x[0], reverse=True)
        
        kept_items = []
        total_tokens = self._system_prompt_tokens
        for score, item in scored_items:
            if total_tokens + item.token_count < self.max_tokens * 0.85:
                kept_items.append(item)
                total_tokens += item.token_count
            elif item.importance > 0.8:
                # 高重要性项：摘要压缩而非丢弃
                summary = self._summarize(item)
                summary_item = MemoryItem(
                    role=item.role,
                    content=summary,
                    timestamp=item.timestamp,
                    importance=item.importance * 0.9,
                    metadata={****item.metadata, "compressed": True}
                )
                summary_item.token_count = len(self.tokenizer.encode(summary))
                kept_items.append(summary_item)
                total_tokens += summary_item.token_count

        compressed_count = len(self.items) - len(kept_items)
        self.items = sorted(kept_items, key=lambda x: x.timestamp)
        return f"Compressed {compressed_count} items, kept {len(kept_items)}"

    def _summarize(self, item: MemoryItem) -> str:
        """对单条记忆生成摘要（实际项目中调用 LLM）"""
        return f"[Summary] {item.content[:100]}... (key: {item.metadata.get('topic', 'general')})"

    def get_context(self) -> List[dict]:
        """生成当前上下文窗口的内容"""
        return [
            {"role": item.role, "content": item.content}
            for item in self.items
        ]

3.2 动态重要性评分

每条记忆的重要性不是静态的，而是由多个因子动态计算：

• 用户显式标记：用户说”记住这个”→ 直接设为 1.0
• 任务相关性：与当前活跃任务相关的记忆提升权重
• 引用频率：被后续对话反复引用的记忆更重要
• 情感强度：用户表达强烈情感（满意/不满）的记忆值得保留

4. 情景记忆：对话历史的压缩与检索

情景记忆存储 Agent 与用户的历史交互序列。与工作记忆不同，情景记忆是持久化的，但直接存储所有历史对话会迅速耗尽存储。核心策略是分层存储 + 语义检索。

4.1 对话摘要的滚动窗口

我们实现一个滚动摘要机制：当对话超过 N 轮后，将早期对话压缩为摘要，保留近期原始对话：

from datetime import datetime
import json

class EpisodicMemory:
    def __init__(self, vector_store, llm_client, 
                 raw_window: int = 20, summary_threshold: int = 5):
        """
        vector_store: 向量数据库客户端（如 Chroma / Qdrant）
        llm_client: LLM 客户端，用于生成摘要
        raw_window: 保留最近 N 轮原始对话
        summary_threshold: 每 N 轮对话生成一个摘要
        """
        self.vector_store = vector_store
        self.llm = llm_client
        self.raw_window = raw_window
        self.summary_threshold = summary_threshold
        self.session_id: Optional[str] = None
        self.raw_turns: List[dict] = []
        self.turn_count = 0

    def start_session(self, session_id: str):
        self.session_id = session_id
        self.raw_turns = []
        self.turn_count = 0

    def record_turn(self, user_msg: str, agent_msg: str, 
                      metadata: dict = None):
        """记录一轮对话"""
        turn = {
            "turn_id": self.turn_count,
            "user": user_msg,
            "agent": agent_msg,
            "timestamp": datetime.utcnow().isoformat(),
            "metadata": metadata or {}
        }
        self.raw_turns.append(turn)
        self.turn_count += 1

        # 检查是否需要生成摘要
        if self.turn_count % self.summary_threshold == 0:
            self._create_summary()

    def _create_summary(self):
        """将早期对话压缩为摘要并存入向量库"""
        turns_to_summarize = self.raw_turns[:-self.raw_window]
        if len(turns_to_summarize) < self.summary_threshold:
            return

        # 构建摘要 prompt
        conversation_text = "\n".join(
            f"[{t['turn_id']}] User: {t['user']}\n[{t['turn_id']}] Agent: {t['agent']}"
            for t in turns_to_summarize
        )
        
        summary_prompt = f"""请将以下对话压缩为结构化摘要，保留关键信息：
- 用户的核心需求和意图
- Agent 的关键决策和行动
- 达成的结论或待办事项
- 用户偏好和重要上下文

对话内容：
{conversation_text}

以 JSON 格式输出：{"summary": "...", "key_topics": [...], "action_items": [...], "user_preferences": {...}}"""

        response = self.llm.complete(summary_prompt)
        summary_data = json.loads(response)

        # 存入向量数据库
        self.vector_store.add(
            ids=[f"summary_{self.session_id}_{self.turn_count}"],
            documents=[summary_data["summary"]],
            metadatas=[{
                "session_id": self.session_id,
                "type": "episodic_summary",
                "turn_range": f"0-{self.turn_count}",
                "key_topics": json.dumps(summary_data["key_topics"])
            }]
        )

    def retrieve_relevant(self, query: str, top_k: int = 5) -> List[str]:
        """根据查询语义检索相关历史记忆"""
        results = self.vector_store.query(
            query_texts=[query],
            n_results=top_k,
            filter={"session_id": self.session_id}
        )
        return results["documents"][0]

4.2 时间感知检索

用户说”上周我们讨论的那个方案”，Agent 需要按时间范围检索。我们在向量数据库的 metadata 中存储时间戳，支持混合过滤：语义相似 + 时间范围。

5. 语义记忆：RAG 知识库的工程实践

语义记忆是 Agent 的”世界知识”，包括领域文档、用户偏好、业务规则等。这是大多数 RAG（检索增强生成）系统的核心，但生产级的语义记忆远不止简单的”切片→嵌入→检索”。

5.1 文档分块策略对比

5.2 父子分块（Parent-Child Chunking）实战

这是目前生产环境中检索精度最高的策略：用小尺寸块做匹配（提高召回），用大尺寸块送入 LLM（提供上下文）：

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
from typing import List, Tuple

class SemanticMemory:
    def __init__(self, embedding_model, vector_store, llm_client):
        self.embedding = embedding_model
        self.vector_store = vector_store
        self.llm = llm_client
        
        # 小块用于语义匹配
        self.child_splitter = RecursiveCharacterTextSplitter(
            chunk_size=256,
            chunk_overlap=30,
            separators=["\n\n", "\n", ". ", " "]
        )
        # 大块用于上下文注入
        self.parent_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1024,
            chunk_overlap=100,
            separators=["\n\n", "\n"]
        )

    def index_document(self, doc_id: str, content: str, 
                        metadata: dict = None):
        """索引文档：父子分块策略"""
        # 1. 生成大块（父块）
        parent_docs = self.parent_splitter.create_documents(
            texts=[content],
            metadatas=[metadata or {}]
        )
        
        all_children = []
        for i, parent in enumerate(parent_docs):
            # 2. 对每个父块生成小块（子块）
            children = self.child_splitter.create_documents(
                texts=[parent.page_content],
                metadatas=[{****parent.metadata, "parent_id": f"{doc_id}_p{i}"}]
            )
            
            # 3. 子块存入向量库，关联父块 ID
            child_texts = [c.page_content for c in children]
            child_ids = [f"{doc_id}_p{i}_c{j}" for j in range(len(children))]
            child_metadatas = [
                {****c.metadata, "parent_id": f"{doc_id}_p{i}"}
                for c in children
            ]
            
            embeddings = self.embedding.embed_documents(child_texts)
            self.vector_store.add(
                ids=child_ids,
                documents=child_texts,
                embeddings=embeddings,
                metadatas=child_metadatas
            )
            
            all_children.extend([(child_ids[j], parent.page_content) 
                                  for j in range(len(children))])

        # 4. 父块存入文档存储（非向量库）
        for i, parent in enumerate(parent_docs):
            self.vector_store.set_document(
                doc_id=f"{doc_id}_p{i}",
                content=parent.page_content
            )

    def query(self, question: str, top_k: int = 3) -> str:
        """检索：先匹配子块，再返回父块"""
        query_embedding = self.embedding.embed_query(question)
        
        # 在子块中搜索
        results = self.vector_store.search(
            embedding=query_embedding,
            n_results=top_k
        )
        
        # 获取对应的父块内容
        parent_ids = [r.metadata["parent_id"] for r in results]
        parent_contents = [
            self.vector_store.get_document(pid) for pid in parent_ids
        ]
        
        # 构建增强上下文
        context = "\n\n".join(
            [f"[Source {i+1}] {content}" 
             for i, content in enumerate(parent_contents)]
        )
        return context

5.3 混合检索：向量 + 关键词 + 知识图谱

纯向量检索在处理精确术语（如 API 名称、错误代码）时表现不佳。生产系统通常采用混合检索：

def hybrid_search(self, query: str, top_k: int = 5) -> List[dict]:
    """混合检索：向量 + BM25 + 知识图谱"""
    
    # 1. 向量检索（语义相似）
    vector_results = self.vector_search(query, top_k=top_k)
    
    # 2. BM25 关键词检索（精确匹配）
    bm25_results = self.bm25_search(query, top_k=top_k)
    
    # 3. 知识图谱检索（实体关系）
    entities = self.extract_entities(query)
    kg_results = self.knowledge_graph.query(entities)
    
    # 4. RRF 融合排序
    fused = self.reciprocal_rank_fusion(
        [vector_results, bm25_results, kg_results],
        weights=[0.5, 0.3, 0.2]
    )
    return fused[:top_k]

def reciprocal_rank_fusion(self, result_lists: List[List], 
                              weights: List[float], k: int = 60) -> List[dict]:
    """RRF 算法：将多路检索结果融合排序"""
    scores = {}
    for weight, results in zip(weights, result_lists):
        for rank, item in enumerate(results):
            doc_id = item["id"]
            if doc_id not in scores:
                scores[doc_id] = {"score": 0, "data": item}
            scores[doc_id]["score"] += weight * (1.0 / (k + rank + 1))
    
    sorted_results = sorted(scores.values(), 
                           key=lambda x: x["score"], 
                           reverse=True)
    return [r["data"] for r in sorted_results]

6. 程序性记忆：技能与工具的自适应学习

程序性记忆是 Agent 的”肌肉记忆”——它知道怎么做，而不仅仅是知道什么。这包括工具使用技能、工作流程模板、错误修复策略等。

6.1 技能库的自动学习

当 Agent 成功完成一个复杂任务后，应该能提取出可复用的”技能”：

class ProceduralMemory:
    def __init__(self, skill_store, llm_client):
        self.skill_store = skill_store  # 技能数据库
        self.llm = llm_client

    def extract_skill(self, task_description: str, 
                       execution_trace: List[dict]) -> dict:
        """从成功的执行轨迹中提取可复用技能"""
        trace_text = json.dumps(execution_trace, indent=2, ensure_ascii=False)
        
        prompt = f"""从以下任务执行轨迹中提取一个可复用的技能模板：

任务：{task_description}

执行轨迹：{trace_text}

提取以下字段：
1. skill_name: 技能名称
2. description: 适用场景描述
3. trigger_conditions: 触发条件（列表）
4. steps: 执行步骤模板（含变量占位符）
5. common_pitfalls: 常见陷阱
6. success_criteria: 成功标准
输出 JSON 格式。"""
        
        response = self.llm.complete(prompt)
        skill = json.loads(response)
        
        # 存入技能库
        self.skill_store.save(skill)
        return skill

    def find_applicable_skill(self, task: str) -> Optional[dict]:
        """根据任务描述匹配最合适的技能"""
        # 技能库也使用向量检索
        results = self.skill_store.search(
            query=task,
            n_results=3,
            filter={"min_success_rate": 0.7}
        )
        return results[0] if results else None

    def update_skill_success_rate(self, skill_id: str, success: bool):
        """更新技能成功率（用于排序和淘汰）"""
        skill = self.skill_store.get(skill_id)
        total = skill["total_uses"] + 1
        successes = skill["successes"] + (1 if success else 0)
        self.skill_store.update(skill_id, {
            "total_uses": total,
            "successes": successes,
            "success_rate": successes / total,
            "last_used": datetime.utcnow().isoformat()
        })

7. 四层记忆的协同调度引擎

核心问题来了：Agent 在每一步决策时，应该从哪一层读取记忆？答案由一个记忆调度器根据当前任务类型动态决定：

class MemoryOrchestrator:
    def __init__(self, working: WorkingMemory, 
                 episodic: EpisodicMemory,
                 semantic: SemanticMemory, 
                 procedural: ProceduralMemory,
                 llm_client):
        self.working = working
        self.episodic = episodic
        self.semantic = semantic
        self.procedural = procedural
        self.llm = llm_client

    def build_context(self, user_message: str, 
                       task_type: str = "general") -> str:
        """根据任务类型构建最优上下文"""
        context_parts = []

        # 1. 始终包含工作记忆（当前对话上下文）
        context_parts.append(
            f"## Current Conversation\n{self._format_working_memory()}"
        )

        # 2. 根据任务类型决定其他记忆层的加载
        if task_type == "debugging":
            # 调试任务：优先语义记忆（文档）+ 程序性记忆（修复技能）
            context_parts.append(
                self.semantic.query(user_message, top_k=3)
            )
            skill = self.procedural.find_applicable_skill(user_message)
            if skill:
                context_parts.append(
                    f"## Relevant Skill\n{json.dumps(skill, ensure_ascii=False)}"
                )

        elif task_type == "planning":
            # 规划任务：优先情景记忆（历史经验）+ 语义记忆（领域知识）
            context_parts.append(
                self.episodic.retrieve_relevant(user_message, top_k=3)
            )
            context_parts.append(
                self.semantic.query(user_message, top_k=2)
            )

        elif task_type == "personalized":
            # 个性化服务：优先情景记忆（用户偏好）
            context_parts.append(
                self.episodic.retrieve_relevant(user_message, top_k=5)
            )

        # 3. 使用 LLM 进行记忆精炼（可选，用于高价值场景）
        raw_context = "\n\n".join(context_parts)
        if len(raw_context) > 12000:
            raw_context = self._refine_context(raw_context, user_message)

        return raw_context

    def post_interaction(self, user_msg: str, agent_msg: str, 
                            task_type: str, success: bool):
        """交互后：将重要信息写入相应记忆层"""
        # 更新工作记忆
        self.working.add(MemoryItem(role="user", content=user_msg))
        self.working.add(MemoryItem(role="assistant", content=agent_msg))

        # 更新情景记忆
        self.episodic.record_turn(user_msg, agent_msg)

        # 如果任务成功且有学习价值，更新程序性记忆
        if success and task_type in ["debugging", "coding", "deployment"]:
            self.procedural.extract_skill_from_interaction(
                user_msg, agent_msg
            )

8. 生产部署关键考量

8.1 记忆一致性与并发安全

当多个 Agent 实例共享同一个记忆系统时，需要考虑并发写入冲突。推荐使用乐观锁或CRDT（无冲突复制数据类型）来处理并发场景。

8.2 记忆隐私与安全

• 数据隔离：多租户场景下，每个用户的记忆必须严格隔离（使用 user_id 作为向量数据库的 namespace）
• 敏感信息过滤：在写入记忆前，使用 NER 模型检测并脱敏 PII（个人身份信息）
• 记忆过期策略：设置 TTL，自动清理超过一定时间的历史记忆
• 用户控制权：提供”忘记”接口，允许用户删除特定记忆

8.3 成本优化

记忆系统的最大成本来自 LLM 调用（摘要生成、重要性评分、技能提取）。优化策略：

8.4 评估指标

记忆系统的质量需要量化评估：

9. 总结与展望

🐱 虾仔出品 | 2026-06-17 | 技术文章 #20