第 15 章
三层记忆架构:工作/情节/语义记忆
第十五章:三层记忆架构:工作/情节/语义记忆
记忆是智能的基础。没有记忆的 AI Agent 就像每天早醒来都失忆的人——无法积累经验,无法建立关系,无法成长。Hermes 的三层记忆架构,正是让 AI 拥有"时间维度"的关键设计。
15.1 三层记忆架构的认知科学基础
15.1.1 从人类记忆到 AI 记忆
人类记忆系统经过数百万年的进化,形成了精妙的分层结构:
| 人类记忆层 | 特征 | AI 对应层 |
|---|---|---|
| 工作记忆(Working Memory) | 当前注意力范围内的活跃信息,容量约 7±2 个组块 | 上下文窗口(Context Window) |
| 情节记忆(Episodic Memory) | 个人经历的事件序列,"什么时候,在哪里,发生了什么" | 会话历史(Session History) |
| 语义记忆(Semantic Memory) | 去情境化的知识,"事实、概念、技能" | Skill 知识库(Skill Library) |
这种分层不是偶然的——它们各自解决了记忆系统中不同的根本性问题:
- 工作记忆解决当前处理能力的问题
- 情节记忆解决时序经验积累的问题
- 语义记忆解决可泛化知识的问题
15.1.2 三层记忆的交互关系
┌─────────────────────────────────────────────────────────────┐
│ 三层记忆交互图 │
│ │
│ ┌───────────────────────────────────────────────────────┐ │
│ │ 工作记忆 (Working Memory) │ │
│ │ ← 当前上下文窗口 (最多 32K tokens) → │ │
│ │ │ │
│ │ [系统提示] [MEMORY.md] [技能注入] [对话历史] [工具结果]│ │
│ │ │ │
│ │ ← 神圣区保护 → ← 可压缩区域 → │ │
│ └───────────────────────────────────────────────────────┘ │
│ ↑ 注入 ↑ 归档 │
│ │ │ │
│ ┌───────────┴──────┐ ┌─────────────┴────────┐ │
│ │ 语义记忆 │ │ 情节记忆 │ │
│ │ (Semantic Memory)│ │ (Episodic Memory) │ │
│ │ │ │ │ │
│ │ Skill 知识库 │ │ 历史会话存储 │ │
│ │ - 技能名称 │ │ - Session 完整记录 │ │
│ │ - 代码模板 │ │ - 时间戳 │ │
│ │ - 使用统计 │ │ - 任务描述 │ │
│ │ - 触发条件 │ │ - 关键结果 │ │
│ │ │ │ │ │
│ │ 存储:VectorDB │ │ 存储:SQLite │ │
│ │ 检索:语义相似度 │ │ 检索:BM25 + 时间 │ │
│ └──────────────────┘ └───────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
15.2 工作记忆(Working Memory)
15.2.1 工作记忆的本质:上下文窗口管理
工作记忆就是模型当前能"看到"的所有内容——即上下文窗口的全部内容。Hermes 的工作记忆设计围绕一个核心问题:如何在有限的上下文窗口内,放置最重要的信息?
class WorkingMemory:
"""
工作记忆管理:控制上下文窗口的内容结构
"""
def __init__(self, max_tokens: int = 32768):
self.max_tokens = max_tokens
self.slots = {
"system_prompt": None, # 固定位置(神圣区)
"memory_injection": None, # MEMORY.md + Skill 注入
"conversation_history": [], # 对话历史
"tool_results": [] # 工具执行结果
}
def get_token_budget(self) -> dict:
"""计算各区域的 token 预算"""
total = self.max_tokens
return {
"system_prompt": int(total * 0.08), # ~2.6K tokens
"memory_injection": int(total * 0.12), # ~3.9K tokens
"conversation": int(total * 0.30), # ~9.8K tokens
"tool_results": int(total * 0.35), # ~11.4K tokens
"response_reserve": int(total * 0.15), # ~4.9K tokens(模型输出预留)
}
def get_context_window(self) -> List[Message]:
"""组装当前上下文窗口"""
messages = []
# 1. 系统提示(始终放在最前)
if self.slots["system_prompt"]:
messages.append(self.slots["system_prompt"])
# 2. 记忆注入(第一轮用户消息中)
if self.slots["memory_injection"]:
messages.append({
"role": "system",
"content": f"## 相关记忆与技能\n{self.slots['memory_injection']}"
})
# 3. 对话历史(时序顺序,最旧到最新)
messages.extend(self.slots["conversation_history"])
return messages
15.2.2 工作记忆的容量限制
Hermes 4 的上下文窗口为 32K tokens,实际可用分布:
┌──────────────────────────────────────────────────┐
│ 32K token 上下文窗口使用分布 │
│ │
│ ████ 系统提示 (8%) ≈ 2.6K tokens │
│ ████ 记忆注入 (12%) ≈ 3.9K tokens │
│ ████ 对话历史 (30%) ≈ 9.8K tokens │
│ ████ 工具结果 (35%) ≈ 11.4K tokens │
│ ░░░░ 输出预留 (15%) ≈ 4.9K tokens │
│ │
│ 有效利用率:~85%(4.9K 预留给模型输出) │
└──────────────────────────────────────────────────┘
15.2.3 工作记忆的关键策略
策略一:重要信息前置
心理学和 LLM 研究都表明,模型对上下文头部和尾部的信息记忆效果更好("位置偏见")。Hermes 将关键信息放在上下文的开头(系统提示 + 记忆注入):
# 正确:关键信息放在前面
context = [
{"role": "system", "content": f"{SYSTEM_PROMPT}\n\n{INJECTED_SKILLS}"},
{"role": "user", "content": user_message},
# ... 对话历史 ...
]
# 错误:关键信息被埋在中间(容易被"遗忘")
context = [
{"role": "user", "content": user_message},
# ... 很长的工具结果 ...
{"role": "system", "content": INJECTED_SKILLS}, # 太迟了
# ... 更多历史 ...
]
策略二:工具结果压缩
大型工具输出(如 500 行的 CSV 内容)会迅速耗尽上下文空间。Hermes 对超过阈值的工具结果进行即时压缩:
def compress_tool_result(result: str, max_tokens: int = 500) -> str:
"""将过长的工具结果压缩为摘要"""
if count_tokens(result) <= max_tokens:
return result
# 对于代码执行结果,保留开头和结尾
if is_code_output(result):
lines = result.split('\n')
if len(lines) > 50:
return '\n'.join(lines[:25]) + '\n... [截断] ...\n' + '\n'.join(lines[-10:])
# 对于数据输出,保留统计摘要
if is_tabular_data(result):
return create_data_summary(result)
# 通用压缩:保留最重要的部分
return truncate_to_tokens(result, max_tokens)
15.3 情节记忆(Episodic Memory)
15.3.1 情节记忆的设计
情节记忆存储历史会话的结构化摘要,使 Hermes 能够从过去的对话中学习经验、避免重复错误:
@dataclass
class Episode:
"""一条情节记忆"""
session_id: str
created_at: datetime
# 任务信息
task_description: str # 任务的简要描述
task_category: str # 分类(代码/分析/写作等)
# 执行信息
steps_taken: int # 执行步骤数
tools_used: List[str] # 使用的工具列表
execution_time_seconds: float # 完成时间
# 结果信息
success: bool # 是否成功
key_result: str # 关键结果摘要(<200字)
error_encountered: Optional[str] # 遇到的主要错误
error_resolution: Optional[str] # 错误解决方案
# 学习信息
skills_applied: List[str] # 应用的技能 ID
new_skill_created: Optional[str] # 新创建的技能 ID
# 向量嵌入(用于语义检索)
embedding: Optional[List[float]] = None
15.3.2 情节记忆的持久化实现
class EpisodicMemoryStore:
"""情节记忆的 SQLite 持久化实现"""
def __init__(self, db_path: str):
self.db_path = db_path
self._init_db()
def _init_db(self):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS episodes (
session_id TEXT PRIMARY KEY,
created_at TIMESTAMP,
task_description TEXT,
task_category TEXT,
steps_taken INTEGER,
tools_used TEXT, -- JSON array
execution_time_seconds REAL,
success BOOLEAN,
key_result TEXT,
error_encountered TEXT,
error_resolution TEXT,
skills_applied TEXT, -- JSON array
new_skill_created TEXT,
embedding BLOB -- 序列化的 numpy array
)
""")
# 创建索引以加速查询
conn.execute("CREATE INDEX IF NOT EXISTS idx_created_at ON episodes(created_at)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_category ON episodes(task_category)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_success ON episodes(success)")
def save_episode(self, episode: Episode):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
INSERT OR REPLACE INTO episodes VALUES (
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
)
""", (
episode.session_id,
episode.created_at.isoformat(),
episode.task_description,
episode.task_category,
episode.steps_taken,
json.dumps(episode.tools_used),
episode.execution_time_seconds,
episode.success,
episode.key_result,
episode.error_encountered,
episode.error_resolution,
json.dumps(episode.skills_applied),
episode.new_skill_created,
self._serialize_embedding(episode.embedding)
))
def search_similar(self, query: str, top_k: int = 5) -> List[Episode]:
"""混合检索:BM25 关键词匹配 + 向量相似度"""
# 方法一:BM25 关键词检索(快速,适合精确匹配)
keyword_results = self._bm25_search(query, limit=top_k * 2)
# 方法二:向量相似度检索(适合语义相似)
query_embedding = self.embedding_model.encode(query)
vector_results = self._vector_search(query_embedding, limit=top_k * 2)
# 混合重排(RRF:Reciprocal Rank Fusion)
return self._reciprocal_rank_fusion(keyword_results, vector_results, top_k)
def get_recent_errors(self, days: int = 7, limit: int = 10) -> List[Episode]:
"""获取最近 N 天的错误情节(用于错误模式学习)"""
cutoff = datetime.now() - timedelta(days=days)
with sqlite3.connect(self.db_path) as conn:
rows = conn.execute("""
SELECT * FROM episodes
WHERE success = FALSE
AND created_at > ?
AND error_resolution IS NOT NULL
ORDER BY created_at DESC
LIMIT ?
""", (cutoff.isoformat(), limit)).fetchall()
return [self._row_to_episode(r) for r in rows]
15.3.3 情节记忆的自动归档
会话结束时,情节记忆系统自动将本次会话归档:
class EpisodicMemoryBuilder:
"""在会话结束时自动构建情节记忆"""
async def build_episode(self, session: Session) -> Episode:
# 1. 使用 LLM 生成任务摘要
summary = await self._generate_summary(session)
# 2. 分类任务类型
category = await self._classify_task(session.initial_task)
# 3. 识别关键错误和解决方案
errors = self._extract_errors(session)
# 4. 生成向量嵌入
embedding_text = f"{session.initial_task} {summary.key_result}"
embedding = self.embedding_model.encode(embedding_text)
return Episode(
session_id=session.session_id,
created_at=session.start_time,
task_description=session.initial_task[:500], # 截断到 500 字
task_category=category,
steps_taken=len(session.steps),
tools_used=list(set(session.tools_used)),
execution_time_seconds=session.total_time,
success=session.task_completed,
key_result=summary.key_result[:200],
error_encountered=errors.main_error if errors else None,
error_resolution=errors.resolution if errors else None,
skills_applied=session.skills_applied_ids,
new_skill_created=session.new_skill_id,
embedding=embedding.tolist()
)
async def _generate_summary(self, session: Session) -> Summary:
"""用 LLM 生成会话摘要(仅保留关键信息)"""
prompt = f"""请为以下对话生成简洁摘要,重点是任务结果。
任务:{session.initial_task}
工具使用:{', '.join(session.tools_used)}
是否成功:{'是' if session.task_completed else '否'}
请用 1-2 句话总结关键结果:"""
summary_text = await self.model.generate(prompt, max_tokens=100)
return Summary(key_result=summary_text.strip())
15.3.4 情节记忆的注入时机
class MemoryInjector:
async def inject_episodic_memory(self, task: str) -> str:
"""
将相关情节记忆注入当前系统提示
"""
# 检索最相关的 3 条情节
relevant_episodes = await self.episodic_store.search_similar(
query=task, top_k=3
)
if not relevant_episodes:
return ""
# 筛选:只注入成功的情节(或包含错误解决方案的失败情节)
valuable_episodes = [
ep for ep in relevant_episodes
if ep.success or ep.error_resolution is not None
]
if not valuable_episodes:
return ""
# 格式化注入文本
inject_lines = ["## 相关历史经验"]
for ep in valuable_episodes[:2]: # 最多注入 2 条
status = "成功" if ep.success else "失败但有解决方案"
inject_lines.append(
f"- **类似任务**({status},{ep.created_at.strftime('%m月%d日')}):"
f"{ep.key_result}"
)
if ep.error_resolution:
inject_lines.append(
f" - **踩坑记录**:{ep.error_encountered} → {ep.error_resolution}"
)
return "\n".join(inject_lines)
15.4 语义记忆(Semantic Memory):Skill 知识库
15.4.1 语义记忆的核心价值
语义记忆是三层记忆中最有"经济价值"的一层:它将临时的执行经验转化为可重用、可组合的技能单元。
经验 → 提炼 → 技能
一次成功的"数据清洗"任务
↓ 提炼
"csv_data_cleaning" Skill(含代码模板)
↓ 重用
第 50 次相似任务:直接应用,速度提升 3×
15.4.2 Skill 的完整数据模型
@dataclass
class Skill:
"""可重用技能的完整数据结构"""
# 身份信息
id: str # UUID
name: str # 短名称,下划线分隔
version: int = 1 # 版本号(每次重写递增)
# 内容信息
description: str # 技能描述
trigger_conditions: List[str] # 触发条件
code_template: str # 可执行代码模板(含 {参数} 占位符)
natural_language_steps: str # 步骤描述(供模型理解)
# 参数定义
parameters: Dict[str, str] # 参数名 → 描述
required_parameters: List[str] # 必须参数
optional_parameters: Dict[str, Any] # 可选参数及默认值
# 依赖关系
dependencies: List[str] # Python 包依赖
prerequisite_skills: List[str] # 依赖的其他技能 ID
# 质量信息
usage_count: int = 0 # 被使用次数
success_rate: float = 1.0 # 历史成功率
best_efficiency_score: float = 0.0 # 最佳效率得分
pitfalls: List[str] = field(default_factory=list) # 已知陷阱
# 元数据
created_at: datetime = field(default_factory=datetime.now)
last_used_at: Optional[datetime] = None
source_episode_id: str = "" # 来源会话 ID
tags: List[str] = field(default_factory=list)
health_status: str = "active" # active | deprecated | needs_review
# 向量嵌入(内部使用)
embedding: Optional[List[float]] = None
15.4.3 Skill 检索系统
class SkillRetriever:
"""多策略技能检索系统"""
def __init__(self, vector_store: SkillVectorStore, sqlite_store: SkillSQLiteStore):
self.vector_store = vector_store
self.sqlite_store = sqlite_store
async def retrieve_relevant_skills(
self,
task: str,
top_k: int = 5,
strategy: str = "hybrid"
) -> List[Skill]:
if strategy == "semantic":
# 纯语义相似度检索
return await self.vector_store.search(task, top_k)
elif strategy == "keyword":
# 关键词匹配(适合技术术语精确匹配)
keywords = extract_technical_terms(task)
return await self.sqlite_store.keyword_search(keywords, top_k)
elif strategy == "hybrid":
# 混合策略(推荐)
semantic_results = await self.vector_store.search(task, top_k * 2)
keyword_results = await self.sqlite_store.keyword_search(
extract_technical_terms(task), top_k * 2
)
# 按照质量综合排序
all_skills = self._deduplicate(semantic_results + keyword_results)
scored_skills = [
(skill, self._compute_retrieval_score(skill, task))
for skill in all_skills
]
scored_skills.sort(key=lambda x: x[1], reverse=True)
return [skill for skill, score in scored_skills[:top_k]]
def _compute_retrieval_score(self, skill: Skill, task: str) -> float:
"""
综合检索得分 = 语义相似度 × 权重 + 使用频率奖励 + 健康状态惩罚
"""
semantic_score = self._semantic_similarity(skill, task) # 0-1
# 使用频率奖励:高频使用的技能得分加成
frequency_bonus = min(skill.usage_count / 100, 0.2) # 最多 +0.2
# 成功率权重
quality_weight = skill.success_rate
# 健康状态惩罚
health_penalty = -0.3 if skill.health_status == "deprecated" else 0
return semantic_score * quality_weight + frequency_bonus + health_penalty
15.4.4 Skill 组合与调度
在复杂任务中,多个 Skill 可以被组合使用:
class SkillComposer:
"""技能组合器:将多个技能组合成复合执行计划"""
async def compose_for_task(self, task: str, available_skills: List[Skill]) -> ExecutionPlan:
"""
给定任务和可用技能,生成组合执行计划
"""
# 使用 LLM 进行技能选择和排序
composition_prompt = f"""
任务:{task}
可用技能列表:
{self._format_skills_list(available_skills)}
请选择并排序最适合完成此任务的技能组合,格式如下:
1. skill_name_1 - 原因:[为什么在此步骤使用]
2. skill_name_2 - 原因:[为什么在此步骤使用]
...
注意:
- 只选择真正需要的技能
- 考虑技能的依赖顺序
- 如果没有合适的技能,可以选择不使用任何技能
"""
composition = await self.model.generate(composition_prompt, max_tokens=500)
selected_skills = self._parse_skill_selection(composition, available_skills)
return ExecutionPlan(
task=task,
skill_sequence=selected_skills,
estimated_steps=sum(s.typical_steps for s in selected_skills)
)
15.5 跨 Session 记忆持久化实现
15.5.1 持久化架构
┌──────────────────────────────────────────────────────┐
│ 跨 Session 持久化架构 │
│ │
│ Session A结束 Session B开始 │
│ │ │ │
│ ↓ ↓ │
│ ┌────────────────┐ ┌──────────────────┐ │
│ │ 自动归档 │ │ 自动加载 │ │
│ │ - 情节记忆保存 │ │ - MEMORY.md 读取 │ │
│ │ - 技能提取存储 │ │ - 相关技能检索 │ │
│ │ - 状态快照 │ │ - 情节记忆检索 │ │
│ └────────┬───────┘ └──────┬───────────┘ │
│ │ ↑ │
│ └──────── 持久化存储 ─────┘ │
│ │ │
│ ┌───────────┼───────────┐ │
│ ↓ ↓ ↓ │
│ SQLite VectorDB 文件系统 │
│ (情节记忆) (技能库) (MEMORY.md) │
└──────────────────────────────────────────────────────┘
15.5.2 状态恢复机制
class SessionRestorer:
"""跨 Session 的记忆状态恢复"""
async def restore_context(self, new_session: Session) -> str:
"""
在新 Session 开始时,恢复相关的历史记忆
"""
task = new_session.initial_task
# 1. 加载 MEMORY.md(最高优先级)
memory_md = self._load_memory_md()
# 2. 检索相关技能(按相似度排序)
relevant_skills = await self.skill_retriever.retrieve_relevant_skills(
task=task, top_k=5
)
# 3. 检索相关情节(按时间和相似度排序)
relevant_episodes = await self.episodic_store.search_similar(
query=task, top_k=3
)
# 4. 组装恢复上下文
context_parts = []
if memory_md:
context_parts.append(f"## 用户自定义记忆\n{memory_md}")
if relevant_skills:
skills_text = self._format_skills_for_injection(relevant_skills)
context_parts.append(f"## 可用技能(来自过往经验)\n{skills_text}")
if relevant_episodes:
episodes_text = self._format_episodes_for_injection(relevant_episodes)
context_parts.append(f"## 相关历史经验\n{episodes_text}")
return "\n\n---\n\n".join(context_parts)
def _format_skills_for_injection(self, skills: List[Skill]) -> str:
formatted = []
for skill in skills:
formatted.append(
f"**{skill.name}**(使用 {skill.usage_count} 次,成功率 {skill.success_rate:.0%})\n"
f"描述:{skill.description}\n"
f"触发:{', '.join(skill.trigger_conditions[:2])}\n"
f"代码:\n```python\n{skill.code_template[:300]}...\n```"
)
return "\n\n".join(formatted)
15.6 MEMORY.md 注入机制详解
15.6.1 MEMORY.md 的设计哲学
MEMORY.md 是 Hermes 系统中一个独特的机制——它允许用户手动向模型注入持久化的上下文信息。这个文件在每次对话开始时自动被注入到系统提示中。
# 典型的 MEMORY.md 内容示例
## 用户偏好
- 编程语言偏好:Python(优先)> JavaScript > Go
- 代码风格:PEP 8,类型注解,详细注释
- 输出格式:优先 Markdown,代码块使用语法高亮
## 项目背景
- 当前主要项目:YiteAI 博客平台(Next.js + SQLite)
- 代码仓库路径:/Users/hexin/code/yiteai/
- 生产环境:Ubuntu 22.04 VPS,域名 dev.yiteai.com
## 数据库信息
- 主数据库:SQLite,路径 /data/yiteai.db
- 表结构:参见 /docs/schema.md
## 常用命令速查
- 启动开发服务器:cd ~/code/yiteai && npm run dev
- 部署:./scripts/deploy.sh production
- 数据库备份:./scripts/backup.sh
## 技术偏好
- 不喜欢过度工程:避免为简单问题引入复杂框架
- 重视测试:所有新功能应有对应测试
- 文档优先:重要决策应有 ADR 记录
15.6.2 MEMORY.md 的注入流程
class MemoryMdInjector:
"""MEMORY.md 注入机制实现"""
def __init__(self, memory_md_path: str):
self.memory_md_path = memory_md_path
self._cache = None
self._cache_mtime = 0
def load_with_cache(self) -> str:
"""带缓存的 MEMORY.md 加载(避免每次 I/O)"""
current_mtime = os.path.getmtime(self.memory_md_path)
if self._cache is None or current_mtime > self._cache_mtime:
with open(self.memory_md_path, 'r', encoding='utf-8') as f:
self._cache = f.read()
self._cache_mtime = current_mtime
return self._cache
def inject_into_system_prompt(self, base_system_prompt: str) -> str:
"""将 MEMORY.md 内容注入系统提示"""
memory_content = self.load_with_cache()
if not memory_content.strip():
return base_system_prompt
injected = f"""{base_system_prompt}
---
## 用户持久化记忆(MEMORY.md)
{memory_content}
---
以上是用户的持久化上下文信息。请在处理任务时参考这些信息。"""
return injected
async def auto_update(self, session: Session, llm_client) -> bool:
"""
对话结束后,让 LLM 决定是否需要更新 MEMORY.md
返回 True 表示已更新
"""
update_check_prompt = f"""
当前 MEMORY.md 内容:
{self.load_with_cache()}
本次对话发现了以下新信息:
{session.get_notable_discoveries()}
请判断是否需要更新 MEMORY.md 以记录重要的新信息。
如果需要更新,请输出更新后的完整 MEMORY.md 内容(仅输出内容,不要其他说明)。
如果不需要更新,请输出 "NO_UPDATE"。
"""
response = await llm_client.generate(update_check_prompt, max_tokens=2000)
if response.strip() != "NO_UPDATE":
with open(self.memory_md_path, 'w', encoding='utf-8') as f:
f.write(response.strip())
self._cache = None # 清除缓存
return True
return False
15.6.3 MEMORY.md 的注入位置策略
MEMORY.md 内容被放置在系统提示的神圣区中(详见第十六章),确保:
- 永远不会被压缩机制删除
- 在每次对话中都处于上下文窗口的前部
- 对模型的行为具有持久的引导作用
15.7 三层记忆的协同工作示例
# 完整的记忆协同工作流程示例
async def complete_memory_workflow(agent: HermesAgent, task: str):
"""展示三层记忆如何协同工作"""
session = agent.create_session()
# ─── 阶段一:Session 启动时 ─────────────────────────────
# 1. 加载工作记忆(初始化上下文窗口)
working_memory = agent.working_memory.initialize()
# 2. 注入 MEMORY.md(用户持久化记忆)
memory_md = agent.memory_md_injector.load_with_cache()
working_memory.inject_section("user_memory", memory_md)
# 3. 检索并注入语义记忆(相关技能)
relevant_skills = await agent.skill_retriever.retrieve_relevant_skills(task, top_k=5)
working_memory.inject_section("skills", format_skills(relevant_skills))
# 4. 检索并注入情节记忆(相关历史)
relevant_episodes = await agent.episodic_store.search_similar(task, top_k=3)
working_memory.inject_section("episodes", format_episodes(relevant_episodes))
print(f"记忆注入完成:{len(relevant_skills)} 个技能,{len(relevant_episodes)} 条历史")
# ─── 阶段二:任务执行中 ────────────────────────────────
result = await agent.execute_task(task, session, working_memory)
# ─── 阶段三:Session 结束后 ─────────────────────────────
# 5. 归档情节记忆
episode = await agent.episodic_builder.build_episode(session)
agent.episodic_store.save_episode(episode)
print(f"情节记忆已归档:{episode.session_id}")
# 6. 提取并保存语义记忆(新技能)
new_skill = await agent.skill_extractor.extract(session)
if new_skill:
await agent.skill_store.save(new_skill)
print(f"新技能已保存:{new_skill.name}")
# 7. 自动更新 MEMORY.md(如需要)
updated = await agent.memory_md_injector.auto_update(session, agent.llm)
if updated:
print("MEMORY.md 已自动更新")
return result
本章小结
- 三层记忆分别解决不同问题:工作记忆(当前处理)、情节记忆(时序经验)、语义记忆(可泛化知识)
- 工作记忆通过精确的 token 预算分配,确保重要信息优先占用有限的上下文空间
- 情节记忆通过 BM25 + 向量混合检索,找到最相关的历史经验
- 语义记忆(Skill 库)通过多策略检索和综合评分,选出最适合当前任务的技能
- MEMORY.md 是用户可手动控制的持久化记忆注入机制,位于神圣区,永不被压缩
- 三层记忆在 Session 开始时协同注入,结束时协同归档和更新
思考题
- 工作记忆的 token 预算分配(系统提示 8%、技能 12%、对话 30%、工具结果 35%)是如何确定的?对于代码密集型任务,这个比例是否需要调整?
- 情节记忆使用 BM25 + 向量混合检索(RRF 融合)。在什么场景下纯向量检索更优?在什么场景下纯关键词检索更优?
- MEMORY.md 允许用户手动写入,但 LLM 也可以自动更新它。如何防止 LLM 自动更新时错误地删除重要信息?
- 三层记忆各层的"遗忘策略"应该如何设计?情节记忆超过 10000 条后应该优先删除哪些?