第 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 内容被放置在系统提示的神圣区中(详见第十六章),确保:

  1. 永远不会被压缩机制删除
  2. 在每次对话中都处于上下文窗口的前部
  3. 对模型的行为具有持久的引导作用

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

本章小结

思考题

  1. 工作记忆的 token 预算分配(系统提示 8%、技能 12%、对话 30%、工具结果 35%)是如何确定的?对于代码密集型任务,这个比例是否需要调整?
  2. 情节记忆使用 BM25 + 向量混合检索(RRF 融合)。在什么场景下纯向量检索更优?在什么场景下纯关键词检索更优?
  3. MEMORY.md 允许用户手动写入,但 LLM 也可以自动更新它。如何防止 LLM 自动更新时错误地删除重要信息?
  4. 三层记忆各层的"遗忘策略"应该如何设计?情节记忆超过 10000 条后应该优先删除哪些?
本章评分
4.8  / 5  (25 评分)

💬 留言讨论