2026年AI Agent长任务总是失忆？三层记忆架构与400行Python搭建无限上下文系统

┌─────────────────────────────────────────────────┐ │ Agent 执行循环 │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ 感知输入 │───>│ 决策引擎 │───>│ 工具执行 │ │ │ └──────────┘ └────┬─────┘ └────┬─────┘ │ │ │ │ │ │ v v │ │ ┌─────────────────────────────┐ │ │ │ 记忆管理器(Memory) │ │ │ │ │ │ │ │ ┌───────┐ ┌───────┐ ┌───┐│ │ │ │ │工作记忆│ │情景记忆│ │语义││ │ │ │ │Working│ │Episodic│ │长期││ │ │ │ │Memory │ │Memory │ │记忆││ │ │ │ └───┬───┘ └───┬───┘ └─┬─┘│ │ │ │ │ │ │ │ │ │ └──────┼─────────┼───────┼──┘ │ │ │ │ │ │ │ v v v │ │ ┌─────────────────────────────┐ │ │ │ 持久化存储层 │ │ │ │ (SQLite / PostgreSQL) │ │ │ └─────────────────────────────┘ │ └─────────────────────────────────────────────────┘

pip install openai chromadb sqlite3-api numpy

import sqlite3 import json import time import uuid from dataclasses import dataclass, field, asdict from typing import Optional, List, Dict, Any from enum import Enum

class MemoryType(Enum):
WORKING = "working" # 工作记忆
EPISODIC = "episodic" # 情景记忆
SEMANTIC = "semantic" # 语义记忆

@dataclass
class MemoryItem:
"""记忆条目"""
id: str
memory_type: MemoryType
content: str # 记忆内容
task_id: str # 所属任务
timestamp: float = field(default_factory=time.time)
metadata: Dict[str, Any] = field(default_factory=dict)
embedding: Optional[List[float]] = None # 向量表示
access_count: int = 0 # 访问次数（用于淘汰策略）
importance: float = 0.5 # 重要性评分 0-1

def to_dict(self) -> dict: d = asdict(self) d['memory_type'] = self.memory_type.value return d

第二步：实现持久化存储层

class MemoryStore: """持久化存储，SQLite实现"""
def __init__(self, db_path: str = "agent_memory.db"): self.conn = sqlite3.connect(db_path) self._init_tables() def _init_tables(self): self.conn.execute(""" CREATE TABLE IF NOT EXISTS memory_items ( id TEXT PRIMARY KEY, memory_type TEXT NOT NULL, content TEXT NOT NULL, task_id TEXT NOT NULL, timestamp REAL NOT NULL, metadata TEXT DEFAULT '{}', embedding TEXT DEFAULT NULL, access_count INTEGER DEFAULT 0, importance REAL DEFAULT 0.5 ) """) self.conn.execute(""" CREATE INDEX IF NOT EXISTS idx_task_id ON memory_items(task_id) """) self.conn.execute(""" CREATE INDEX IF NOT EXISTS idx_type_task ON memory_items(memory_type, task_id) """) self.conn.commit() def save(self, item: MemoryItem): self.conn.execute( """INSERT OR REPLACE INTO memory_items (id, memory_type, content, task_id, timestamp, metadata, embedding, access_count, importance) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", (item.id, item.memory_type.value, item.content, item.task_id, item.timestamp, json.dumps(item.metadata, ensure_ascii=False), json.dumps(item.embedding) if item.embedding else None, item.access_count, item.importance) ) self.conn.commit() def query_by_task(self, task_id: str, memory_type: Optional[MemoryType] = None, limit: int = 50) -> List[MemoryItem]: if memory_type: rows = self.conn.execute( """SELECT * FROM memory_items WHERE task_id = ? AND memory_type = ? ORDER BY timestamp DESC LIMIT ?""", (task_id, memory_type.value, limit) ).fetchall() else: rows = self.conn.execute( """SELECT * FROM memory_items WHERE task_id = ? ORDER BY timestamp DESC LIMIT ?""", (task_id, limit) ).fetchall() return [self._row_to_item(row) for row in rows] def increment_access(self, item_id: str): self.conn.execute( "UPDATE memory_items SET access_count = access_count + 1 WHERE id = ?", (item_id,) ) self.conn.commit() def _row_to_item(self, row) -> MemoryItem: return MemoryItem( id=row[0], memory_type=MemoryType(row[1]), content=row[2], task_id=row[3], timestamp=row[4], metadata=json.loads(row[5]) if row[5] else {}, embedding=json.loads(row[6]) if row[6] else None, access_count=row[7], importance=row[8] )

第三步：实现工作记忆——带智能淘汰的上下文管理

工作记忆的核心问题：上下文窗口有限，放什么进去、踢什么出来？

朴素做法是FIFO（先进先出），但更好的策略是基于重要性和访问频率的淘汰。

class WorkingMemory: """ 工作记忆：管理当前进入LLM上下文的信息 核心策略：基于重要性 + 时效性的动态淘汰 """
def __init__(self, max_tokens: int = 8000): self.max_tokens = max_tokens self.items: List[MemoryItem] = [] self.system_prompt: str = "" def add(self, item: MemoryItem): """添加信息到工作记忆""" self.items.append(item) self._evict_if_needed() def get_context(self) -> str: """获取当前工作记忆的完整上下文""" parts = [] if self.system_prompt: parts.append(f"[系统指令]

{self.system_prompt}")

# 按时间排序 sorted_items = sorted(self.items, key=lambda x: x.timestamp) for item in sorted_items: role = item.metadata.get("role", "assistant") parts.append(f"[{role}] {item.content}") return "

".join(parts)

def _estimate_tokens(self, text: str) -> int: """粗略估算token数（中文约1.5字/token）""" return int(len(text) / 1.5) def _evict_if_needed(self): """超限时，按重要性+时效性评分淘汰""" while self._total_tokens() > self.max_tokens and len(self.items) > 2: # 计算每条的保留优先级分数 now = time.time() scored = [] for item in self.items: age_hours = (now - item.timestamp) / 3600 # 分数 = 重要性 * 0.6 + 时效性 * 0.3 + 访问频率 * 0.1 # 时效性：越新分越高 recency = max(0, 1.0 - age_hours / 24) # 24小时衰减到0 access_score = min(1.0, item.access_count / 10) score = (item.importance * 0.6 + recency * 0.3 + access_score * 0.1) scored.append((score, item)) # 淘汰分数最低的 scored.sort(key=lambda x: x[0]) evicted = scored[0][1] self.items.remove(evicted) def _total_tokens(self) -> int: return self._estimate_tokens(self.get_context())

这里的设计关键点：不是简单地保留最近N条，而是给每条信息打分。一条被频繁访问的高重要性信息，即使时间较早，也应该保留在工作记忆中。这和人脑的工作机制很像——你今天一直在想的那个bug，即使发生在早上，到下午也不会忘掉。

第四步：实现情景记忆——全量无损记录

class EpisodicMemory: """ 情景记忆：完整记录Agent执行的每一步 关键原则：无损存储，不做任何压缩 """
def __init__(self, store: MemoryStore, embedding_fn=None): self.store = store self.embedding_fn = embedding_fn def record(self, task_id: str, step: str, action: str, observation: str, decision_reason: str = "", importance: float = 0.5): """ 记录一个完整的事件 step: 步骤编号或名称 action: Agent采取的行动 observation: 行动的结果/观察 decision_reason: 为什么这么做 """ content = json.dumps({ "step": step, "action": action, "observation": observation[:2000], # 限制单条长度 "decision_reason": decision_reason, "timestamp_readable": time.strftime( "%Y-%m-%d %H:%M:%S" ) }, ensure_ascii=False) item = MemoryItem( id=str(uuid.uuid4()), memory_type=MemoryType.EPISODIC, content=content, task_id=task_id, importance=importance ) # 生成向量（如果有embedding函数） if self.embedding_fn: item.embedding = self.embedding_fn(content) self.store.save(item) def recall(self, task_id: str, query: str = "", top_k: int = 10) -> List[Dict]: """ 检索情景记忆 如果提供query，做语义检索；否则返回最近记录 """ if query and self.embedding_fn: # 语义检索模式 all_items = self.store.query_by_task( task_id, MemoryType.EPISODIC, limit=200 ) query_vec = self.embedding_fn(query) # 计算余弦相似度 scored = [] for item in all_items: if item.embedding: sim = self._cosine_similarity( query_vec, item.embedding ) scored.append((sim, item)) scored.sort(key=lambda x: x[0], reverse=True) results = [item for _, item in scored[:top_k]] else: # 时间排序模式 results = self.store.query_by_task( task_id, MemoryType.EPISODIC, limit=top_k ) # 更新访问次数 for item in results: self.store.increment_access(item.id) return [json.loads(item.content) for item in results] def _cosine_similarity(self, vec_a: List[float], vec_b: List[float]) -> float: import numpy as np a, b = np.array(vec_a), np.array(vec_b) return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-8))

第五步：实现语义记忆——跨任务的知识沉淀

class SemanticMemory: """ 语义记忆：从历史任务中提炼的稳定知识 关键原则：只添加不修改，保留知识演化轨迹 """
def __init__(self, store: MemoryStore, embedding_fn=None, llm_client=None): self.store = store self.embedding_fn = embedding_fn self.llm_client = llm_client def add_knowledge(self, task_id: str, knowledge: str, category: str = "general", importance: float = 0.7): """添加一条语义知识""" item = MemoryItem( id=str(uuid.uuid4()), memory_type=MemoryType.SEMANTIC, content=knowledge, task_id=task_id, metadata={"category": category}, importance=importance ) if self.embedding_fn: item.embedding = self.embedding_fn(knowledge) self.store.save(item) def search(self, query: str, top_k: int = 5, category: Optional[str] = None) -> List[Dict]: """语义检索相关知识""" # 获取所有语义记忆 all_items = self.store.query_by_task( task_id="*", # 语义记忆跨任务 memory_type=MemoryType.SEMANTIC, limit=500 ) if category: all_items = [ item for item in all_items if item.metadata.get("category") == category ] if not self.embedding_fn or not all_items: return [{"content": item.content, "metadata": item.metadata} for item in all_items[:top_k]] query_vec = self.embedding_fn(query) scored = [] for item in all_items: if item.embedding: sim = self._cosine_similarity(query_vec, item.embedding) scored.append((sim, item)) scored.sort(key=lambda x: x[0], reverse=True) for _, item in scored[:top_k]: self.store.increment_access(item.id) return [{"content": item.content, "metadata": item.metadata, "relevance": round(score, 4)} for score, item in scored[:top_k]] def extract_and_save(self, task_id: str, episodic_records: List[Dict]): """ 从情景记忆中提炼语义知识 这一步在任务结束时调用，用LLM总结可复用的知识 """ if not self.llm_client: return records_text = json.dumps( episodic_records[:20], ensure_ascii=False, indent=2 ) prompt = f"""分析以下Agent执行记录，提炼出可复用的知识。

要求：

1. 提取用户偏好、项目规则、已验证的技术结论
2. 每条知识用一句话表述，要具体可操作
3. 不要提取临时性的中间结论
4. 输出JSON数组格式

执行记录：
{records_text}

输出格式：
[{{"knowledge": "…", "category": "user_pref|tech_rule|project_rule|general", "importance": 0.0-1.0}}]"""

response = self.llm_client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0.3 ) try: knowledge_list = json.loads( response.choices[0].message.content ) for k in knowledge_list: self.add_knowledge( task_id=task_id, knowledge=k["knowledge"], category=k.get("category", "general"), importance=k.get("importance", 0.7) ) except (json.JSONDecodeError, KeyError) as e: print(f"知识提炼失败: {e}") def _cosine_similarity(self, vec_a, vec_b): import numpy as np a, b = np.array(vec_a), np.array(vec_b) return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-8))

第六步：记忆管理器——把三层串起来

class MemoryManager: """ 记忆管理器：统一调度三层记忆 这是Agent执行循环中唯一需要调用的接口 """
def __init__(self, embedding_fn=None, llm_client=None): self.store = MemoryStore() self.working = WorkingMemory(max_tokens=8000) self.episodic = EpisodicMemory( self.store, embedding_fn ) self.semantic = SemanticMemory( self.store, embedding_fn, llm_client ) self.current_task_id = str(uuid.uuid4()) def start_task(self, task_description: str, user_context: Dict = None): """启动新任务""" self.current_task_id = str(uuid.uuid4()) self.working.items.clear() # 工作记忆：放入任务描述 self.working.add(MemoryItem( id=str(uuid.uuid4()), memory_type=MemoryType.WORKING, content=f"任务目标：{task_description}", task_id=self.current_task_id, importance=1.0, metadata={"role": "system"} )) # 从语义记忆中检索相关知识 if user_context: related = self.semantic.search( task_description, top_k=5 ) if related: context_text = "相关历史知识：

" + "
“.join(
[f”- {r[‘content’]}" for r in related]
)
self.working.add(MemoryItem(
id=str(uuid.uuid4()),
memory_type=MemoryType.WORKING,
content=context_text,
task_id=self.current_task_id,
importance=0.8,
metadata={“role”: “system”}
))

def record_step(self, step: str, action: str, observation: str, decision_reason: str = ""): """记录一步执行""" # 情景记忆：全量记录 self.episodic.record( task_id=self.current_task_id, step=step, action=action, observation=observation, decision_reason=decision_reason, importance=0.6 ) # 工作记忆：只放当前步骤的摘要 summary = f"步骤{step}：执行了{action[:100]}，结果：{observation[:200]}" self.working.add(MemoryItem( id=str(uuid.uuid4()), memory_type=MemoryType.WORKING, content=summary, task_id=self.current_task_id, importance=0.6, metadata={"role": "assistant"} )) def recall_relevant(self, query: str) -> str: """ 当Agent需要回忆时调用 同时检索情景记忆和语义记忆，合并注入工作记忆 """ # 从情景记忆中检索 episodes = self.episodic.recall( self.current_task_id, query, top_k=5 ) # 从语义记忆中检索 semantics = self.semantic.search(query, top_k=3) context_parts = [] if episodes: context_parts.append("== 历史执行记录 ==") for ep in episodes: context_parts.append( f"[{ep.get('step', '?')}] " f"{ep.get('action', '')} -> " f"{ep.get('observation', '')[:200]}" ) if semantics: context_parts.append("

== 相关知识 ==“)
for s in semantics:
context_parts.append(f”- {s[‘content’]}")

return "

".join(context_parts)

def finish_task(self): """任务结束：提炼语义知识""" # 获取本次任务的所有情景记录 episodes = self.episodic.recall( self.current_task_id, top_k=30 ) # 提炼为语义知识 self.semantic.extract_and_save( self.current_task_id, episodes ) def get_llm_context(self) -> str: """获取当前应注入LLM的上下文""" return self.working.get_context()

第七步：组装一个完整的Agent执行循环

class MemoryAwareAgent: """带三层记忆的Agent"""
def __init__(self, llm_client, embedding_fn=None): self.llm = llm_client self.memory = MemoryManager(embedding_fn, llm_client) def run_task(self, task: str, max_steps: int = 20): """执行一个长任务""" self.memory.start_task(task) for step in range(1, max_steps + 1): # 获取当前上下文 context = self.memory.get_llm_context() # 如果Agent需要回忆，主动检索 if step > 5 and step % 3 == 0: recalled = self.memory.recall_relevant(task) if recalled: # 把检索到的信息注入工作记忆 self.memory.working.add(MemoryItem( id=str(uuid.uuid4()), memory_type=MemoryType.WORKING, content=f"[记忆检索结果]

{recalled}",
task_id=self.memory.current_task_id,
importance=0.7,
metadata={"role": "system"}
))
context = self.memory.get_llm_context()

# 调用LLM决策 response = self.llm.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": "你是一个任务执行Agent。" "根据当前上下文决定下一步行动。" "如果要结束，输出 DONE。"}, {"role": "user", "content": context} ], temperature=0.3 ) action = response.choices[0].message.content if "DONE" in action: print(f"任务在第{step}步完成") break # 模拟执行（实际场景中这里调用工具） observation = f"执行结果：步骤{step}成功" # 记录这一步 self.memory.record_step( step=str(step), action=action, observation=observation, decision_reason="基于当前上下文判断" ) # 任务结束，提炼知识 self.memory.finish_task() print("任务执行完毕，知识已沉淀")

整套代码大约400行，但已经实现了完整的三层记忆架构。实际项目中你需要补充的部分：

1. Embedding函数：可以用OpenAI的text-embedding-3-small或本地BGE模型
2. 工具调用层：根据你的业务场景对接具体的API和数据源
3. 并发控制：多任务并行时的情景记忆隔离
4. 记忆清理：定期归档过期的语义记忆

这套架构解决了什么，还有什么没解决

先说解决的问题：

工作记忆的智能淘汰，让Agent不再因为上下文窗口溢出而丢失关键信息。它不再简单地"忘掉最早的"，而是根据重要性和时效性做决策。

情景记忆的无损存储，让Agent在需要时可以回溯任何历史步骤的完整细节。不再是模糊的摘要，而是精确的原始记录。

语义记忆的跨任务复用，让Agent能从历史任务中学习。第一次踩过的坑，第二次就不会再踩。

但也有诚实的局限：

1. 检索质量依赖Embedding模型。如果Embedding不够好，语义检索可能找不到真正相关的记忆。这个问题目前没有完美解，建议用领域数据做微调。
2. 知识提炼的质量不稳定。LLM总结出的"语义知识"偶尔会有偏差。建议加一个人工审核环节，或者用置信度阈值过滤。
3. 存储成本。全量无损存储在长时间运行的场景下会消耗大量存储。生产环境建议对超过7天的情景记忆做分级存储。

黑箭科技在为企业客户部署舆情监控Agent时，也遇到了类似的长任务记忆丢失问题。最终采用分层记忆架构后，Agent在72小时以上的持续监控任务中，上下文保持率从不到45%提升到了92%。这说明三层记忆的思路不是理论上的美好设想，而是确实能落地的工程方案。

和业界方案的对比

目前行业里有几个值得关注的记忆方案：

如果你的Agent基于LangGraph或CrewAI开发，LangChain Memory是最省事的选择。但如果你在处理超过10步以上的长任务，建议认真考虑分层记忆的方案。

总结

写这篇文章不是要造一个新轮子，而是把"Agent为什么会失忆"这个问题拆开来看清楚。

核心结论就三条：

1. 上下文窗口不等于记忆。把记忆等同于上下文窗口，是当前大多数Agent框架的根本缺陷。窗口是用完即弃的工作台，记忆是需要持久化和检索的知识体系。

2. 分层是必然选择。工作记忆管"正在用的"，情景记忆管"发生过什么"，语义记忆管"学到了什么"。三层各司其职，互不干扰。

3. 无损比有损强。摘要压缩看起来省空间，但丢掉的细节往往在后续步骤中成为致命盲点。存储成本在2026年已经不是瓶颈了，信息丢失才是。

如果你在开发Agent的过程中也遇到过"失忆"问题，或者对三层记忆架构有更具体的落地疑问，评论区聊聊。

参考资料：

• APUS Reflex Agent 开源项目（2026-10-14）：https://github.com/APUS-AI-Lab/APUS-Reflex-Agent
• Atkinson & Shiffrin, "Human memory: A proposed system and its control processes", 1968
• MCP 2026-07-28 Specification: https://modelcontextprotocol.io/specification