企业官网建设流程全解析

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 的记忆信息是真正相关的。