> For the complete documentation index, see [llms.txt](https://yeasy.gitbook.io/harness_engineering_guide/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://yeasy.gitbook.io/harness_engineering_guide/di-er-bu-fen-harness-he-xin-zi-xi-tong/06_memory/6.5_miniharness_memory.md).

# 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 {'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 天未更新的项
    low_confidence_count = await self.memory_store.cleanup_low_confidence(
        threshold=0.3, stale_days=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/` 目录的三个模块。