# 6.5 MiniHarness 记忆系统架构

本节从“设计决策”角度讲解记忆系统的核心架构。完整实现代码已分离至 `lab/mini_harness/memory/` 目录，本文聚焦核心概念、关键权衡和实现要点。

## 6.5.1 存储层：MemoryEntry 与 MemoryStore

记忆条目是系统的基本单元。关键设计决策：

**为什么使用 Markdown + YAML frontmatter？**

选择文本格式而非二进制 DB，有三个考量：

1. “便于人工审查”：用户可直接读写存储文件，便于调试
2. “版本控制友好”：支持 Git diff，可追踪记忆演变
3. “轻量化”：无需外部数据库依赖

```python
from datetime import datetime
from typing import List, Optional

class MemoryEntry:
    """包含 5 个维度的记忆条目"""
    def __init__(self, memory_id: str, content: str,
                 memory_type: str = "episodic",     # 类型划分
                 tags: List[str] = None,
                 confidence: float = 1.0,           # 置信度
                 expiry: Optional[datetime] = None):  # 过期时间
        self.id = memory_id
        self.content = content
        self.type = memory_type
        self.tags = tags or []
        self.confidence = confidence
        self.expiry = expiry
        # 自动跟踪时间戳和版本
        self.created_at = datetime.now()
        self.last_modified = datetime.now()
        self.version = 1
```

**存储组织策略**：`by_type/` 分类结构。文件组织如下：

```
memory_dir/
  by_type/
    user/          （用户偏好、档案）
    project/       （项目进度、决策）
    episodic/      （每日交互记录）
    reference/     （模式、示例）
    feedback/      （用户反馈、评价）
```

这样设计的好处：

* “热路径优化”：查询特定类型无需全表扫描
* “权限隔离”：便于未来细粒度访问控制
* “过期清理”：批量删除某类型过期项时高效

关键实现位置：`lab/mini_harness/memory/storage.py`

## 6.5.2 上下文组装：需求驱动的选择性加载

ContextAssembler 解决核心问题：“给定用户输入，应该加载哪些记忆？”

**三层决策链路**：

```python
async def analyze_query(self, user_message: str) -> MemoryRequirement:
    """启发式规则:输入 -> 需求"""
    req = MemoryRequirement()
    msg_lower = user_message.lower()

    # 关键词触发
    if any(w in msg_lower for w in ['prefer', 'style', 'like']):
        req.needs_user_profile = True
    if any(w in msg_lower for w in ['project', 'task', 'status']):
        req.needs_project_context = True
    # ... 其他规则

    # 默认回退:当无法识别时加载用户档案
    if not any([...]):
        req.needs_user_profile = True
    return req
```

**为什么选择启发式而非 LLM？**

* “延迟考量”：LLM 调用需另外 1-2 秒，用户感受延迟
* “成本考量”：每次查询都调 LLM 会显著增加开销
* “可调试性”：规则清晰，易于审计和修改

**并行收集 + Token 预算**：

```python
async def assemble(self, user_message: str) -> str:
    requirement = await self.analyze_query(user_message)

    # 根据需求并行加载多个记忆源
    tasks = []
    if requirement.needs_user_profile:
        tasks.append(('user_profile', self._gather_user_profile()))
    if requirement.needs_project_context:
        tasks.append(('project', self._gather_project_context()))
    # ... 其他需求

    # 并行执行,节省 I/O 等待
    results = await asyncio.gather(*[task[1] for task in tasks])

    # 按优先级排序,限制总大小
    # 保证组装结果不超过 token_budget (如 50k)
```

关键权衡：“所有相关记忆”vs“上下文窗口限制”，通过 Token 预算解决。

关键实现位置：`lab/mini_harness/memory/context.py`

## 6.5.3 整合层：四阶段记忆巩固

ConsolidationEngine 实现“睡眠学习”机制——后台自动处理信息。

**三门触发条件** （防止过度整合）：

```python
def should_consolidate(self) -> bool:
    # 时间门:距上次整合已超 24 小时
    time_gate = (datetime.now() - self.state.last_consolidation
                 > timedelta(hours=24))
    # 会话门:至少完成 5 次对话
    session_gate = self.state.sessions_since_consolidation >= 5
    # 显式门:由外部调用者手动触发(如用户主动要求)
    return time_gate or session_gate
```

**四阶段流程**（基于 autoDream）：

1. “确定阶段”(Orient)：分析最近消息的主题
2. “收集阶段”(Gather)：提取偏好、项目更新、教训、决策
3. “融合阶段”(Consolidate)：写入长期记忆
4. “清理阶段”(Prune)：删除过期项和低置信度项

```python
async def orient_phase(self, recent_messages: List[str]) -> Dict:
    """识别关键主题"""
    topics = set()
    for msg in recent_messages[-5:]:
        if any(w in msg.lower() for w in ['bug', 'error']):
            topics.add('issues')
        if any(w in msg.lower() for w in ['feature', 'new']):
            topics.add('features')
    return {'summary': summary, 'topics': list(topics)}

async def gather_phase(self, messages: List[str], orient_result: Dict) -> Dict[str, List[str]]:
    """从消息中抽取不同维度的信息"""
    gathered = {
        'user_preferences': [],      # 用户习惯
        'project_updates': [],       # 进度变化
        'learned_lessons': [],       # 知识积累
        'decisions': []              # 重要决策
    }
    # 按关键词提取相关句子
    return gathered

async def consolidate_phase(self, gathered: Dict) -> bool:
    """写入对应类型的存储"""
    if gathered['user_preferences']:
        entry = MemoryEntry(..., memory_type='user')
        await self.memory_store.save(entry)
    # ...
    return True

async def prune_phase(self) -> int:
    """清理两类项:过期项 + 低置信度项"""
    removed = await self.memory_store.cleanup_expired()
    # 清理 confidence < 0.3 且超过 30 天未更新的项
    return removed + low_confidence_count
```

**设计权衡**：

* “自动整合 vs 隐私”：整合过程完全本地化，不上传服务器
* “频率 vs 成本”：三门限制防止频繁整合浪费资源
* “保留 vs 清理”：置信度和过期时间的组合清理策略

关键实现位置：`lab/mini_harness/memory/consolidation.py`

## 6.5.4 系统集成要点

一个完整的使用流程包括：

```python
# 初始化三个核心组件
memory_store = MemoryStore("/path/to/memory")
assembler = ContextAssembler(memory_store, token_budget=50000)
consolidation = ConsolidationEngine(memory_store)

# 在每次对话前:组装上下文
context = await assembler.assemble(user_message)

# 在后台:定期整合
if consolidation.should_consolidate():
    success = await consolidation.consolidate(recent_messages)
```

**核心设计特点**：

1. “模块化”：存储、组装、整合相互独立，便于扩展
2. “异步友好”：全 async/await，支持并发操作
3. “可观测”：YAML frontmatter 和文本格式便于审查
4. “防御式编程”：大量 try-catch 和类型检查，处理不完整数据

完整代码参考 `lab/mini_harness/memory/` 目录的三个模块。


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://yeasy.gitbook.io/harness_engineering_guide/di-er-bu-fen-harness-he-xin-zi-xi-tong/06_memory/6.5_miniharness_memory.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.