> 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.1_memory_architecture.md).

# 6.1 Harness 中的记忆系统工程设计

记忆系统是智能体长期学习和适应的基础。关于记忆的认知模型、三层架构理论（工作、短期和长期记忆），以及非参数记忆与参数记忆的根本区别，请参阅《智能体 AI 权威指南》第三章。同时，更多关于记忆系统在上下文工程中的应用，请参阅《上下文工程权威指南》第四章。

> 💡 **理论参考**：
>
> * 《智能体 AI 权威指南》第三章(3.1-3.3)介绍了多层记忆的认知模型和工程映射
> * 《上下文工程权威指南》第四章讨论了上下文管理与记忆的关系

本节重点讨论 **Harness 框架中的记忆实现**，特别是 **autoDream 系统** 和 **记忆整合管道** 的工程细节。

## 6.1.1 Harness 记忆架构概览

Harness 采用了一个经过验证的三层架构，但针对多任务、多工作流的场景做了特殊优化：

```mermaid
graph TD
    A["当前对话消息"] -->|窗口溢出| B["<b>工作记忆</b><br/>LLM 上下文窗口"]

    B -->|自动整合| D["<b>短期记忆</b><br/>session_cache.jsonl"]
    D -->|触发条件 × 3<br/>定期整合| F["长期记忆"]

    F -->|结构化部分| G1["<b>MEMORY.md</b><br/>用户档案、学习、决策"]
    F -->|向量部分| G2["<b>embedding_index</b><br/>语义检索"]
    F -->|日志部分| G3["<b>session_logs/</b><br/>按日期分片"]

    B -.->|需要时检索| D
    D -.->|需要时检索| G1
    D -.->|语义搜索| G2

    style A fill:#bbdefb
    style B fill:#c8e6c9
    style D fill:#fff9c4
    style G1 fill:#ffe0b2
    style G2 fill:#f8bbd0
    style G3 fill:#d4f1d4
```

图 6-1：Harness 三层记忆架构

**各层特性**：

1. **工作记忆**：完全委托给 LLM 上下文管理，Harness 无需干预
2. **短期记忆**：轻量化的会话缓存（JSON 格式），存储过去 10 个会话的摘要
3. **长期记忆**：组合式存储，分为三个子系统

## 6.1.2 autoDream 系统：记忆整合管道

autoDream 是 Harness 的自动记忆整合系统，负责将会话数据定期整合到长期记忆。它的设计借鉴了 Claude Code 的思想，但针对多工作流、多用户的场景进行了扩展。

> 注：Anthropic 于 2026-05-06 在 Managed Agents 中推出了同类官方机制 **Dreaming**（会话间异步整合记忆），与本节 autoDream 的三门触发 + 四阶段管道思路一致，可作为托管侧参照实现。详见附录 C 时效快照。

### 整合触发的三门机制

autoDream 不是持续运行，而是在满足 **至少一个** 触发条件时才激活：

```python
import time

class MemoryConsolidationTrigger:
    """记忆整合触发器"""

    def __init__(self, config):
        self.time_threshold = config.get("time_threshold", 86400)  # 24小时
        self.session_threshold = config.get("session_threshold", 5)  # 5个会话
        self.last_consolidation_time = time.time()
        self.session_count_since_consolidation = 0

    def should_consolidate(self) -> bool:
        """检查是否应该触发整合"""
        # 触发条件1:时间门槛
        if time.time() - self.last_consolidation_time > self.time_threshold:
            return True

        # 触发条件2:会话计数门槛
        if self.session_count_since_consolidation >= self.session_threshold:
            return True

        # 触发条件3:显式锁定(用户或系统主动触发)
        # 这通过外部API调用实现:POST /consolidate

        return False

    def on_session_end(self):
        """会话结束时调用"""
        self.session_count_since_consolidation += 1
```

### 四阶段整合管道

一旦触发，autoDream 执行标准化的四阶段流程：

```python
import time
from typing import List, Dict, Any

class MemoryConsolidationPipeline:
    """记忆整合管道"""

    async def consolidate(self, workflow_id: str, session_history: List[Dict[str, Any]]):
        """执行完整的整合流程"""

        # 阶段1:Orient - 分析会话目标和上下文
        orientation = await self.orient_phase(workflow_id, session_history)
        # 结果:{goals, context, decisions, new_patterns}

        # 阶段2:Gather - 从对话历史提取关键信息
        gathered = await self.gather_phase(session_history, orientation)
        # 结果:{facts, preferences, outcomes, patterns}

        # 阶段3:Consolidate - 合并到长期记忆,避免冗余
        consolidated = await self.consolidate_phase(workflow_id, gathered)
        # 结果:更新了MEMORY.md, embedding_index, session_logs

        # 阶段4:Prune - 删除过期或低价值的信息
        pruned = await self.prune_phase(workflow_id, consolidated)
        # 结果:清理过时的会话摘要

        return {
            "orientation": orientation,
            "gathered": gathered,
            "consolidated": consolidated,
            "pruned": pruned
        }

    async def orient_phase(self, workflow_id: str, session_history: List[Message]):
        """阶段1:Orient - 理解会话的核心目标和决策点"""
        prompt = f"""
分析这个工作流的会话记录,提供:
1. 会话的核心目标(1-3个)
2. 关键的决策点和选择(2-3个)
3. 发现的系统级模式或约束(2-3个)
4. 用户展现的新的偏好或工作风格(如有)

会话记录:
{self._format_session_history(session_history)}
"""
        # 调用LLM进行分析
        return await self.llm.analyze(prompt)

    async def gather_phase(self, session_history: List[Message], orientation: dict):
        """阶段2:Gather - 提取结构化事实"""
        facts = {
            "user_preferences": [],
            "task_outcomes": [],
            "technical_discoveries": [],
            "system_constraints": []
        }

        for message in session_history:
            if message.role == "user":
                # 从用户消息提取偏好
                pref = self._extract_preference(message.content)
                if pref:
                    facts["user_preferences"].append(pref)

            elif message.role == "assistant":
                # 从助手消息提取成果
                outcome = self._extract_outcome(message.content)
                if outcome:
                    facts["task_outcomes"].append(outcome)

        return facts

    async def consolidate_phase(self, workflow_id: str, gathered: dict):
        """阶段3:Consolidate - 合并到长期记忆,避免冗余"""

        # 读取现有MEMORY.md
        memory = await self._load_memory_file(workflow_id)

        # 去重:与现有内容比较,只添加新信息
        new_facts = await self._deduplicate(gathered["user_preferences"], memory.get("user_preferences", []))

        # 更新MEMORY.md
        memory["user_preferences"].extend(new_facts)
        memory["last_updated"] = time.time()

        # 更新向量索引(用于语义搜索)
        embeddings = await self._embed_facts(gathered)
        await self._upsert_to_vector_index(workflow_id, embeddings)

        # 追加到日志
        await self._append_to_session_log(workflow_id, gathered)

        return {"updated_memory": memory, "updated_embeddings": embeddings}

    async def prune_phase(self, workflow_id: str, consolidated: dict):
        """阶段4:Prune - 删除过期或低价值的信息"""

        memory = consolidated["updated_memory"]

        # 策略1:删除超过6个月的会话摘要
        cutoff_time = time.time() - 180 * 24 * 3600
        memory["session_summaries"] = [
            s for s in memory.get("session_summaries", [])
            if s.get("timestamp", 0) > cutoff_time
        ]

        # 策略2:删除重复或相似的用户偏好
        memory["user_preferences"] = await self._deduplicate_with_similarity(
            memory["user_preferences"],
            similarity_threshold=0.95
        )

        # 策略3:删除与当前工作无关的信息
        relevant_prefs = await self._filter_by_relevance(
            memory["user_preferences"],
            workflow_id,
            relevance_threshold=0.6
        )
        memory["user_preferences"] = relevant_prefs

        # 保存回磁盘
        await self._save_memory_file(workflow_id, memory)

        return {"pruned_memory": memory}
```

## 6.1.3 短期记忆与缓存策略

短期记忆的目的是快速恢复最近会话的上下文，无需每次都从长期记忆检索。

```python
class SessionMemoryCache:
    """会话记忆缓存(JSON格式)"""

    def __init__(self, max_sessions: int = 10):
        self.max_sessions = max_sessions
        # open() 不会展开 ~，需要 expanduser
        self.cache_file = os.path.expanduser("~/.harness/session_cache.jsonl")

    async def record_session_summary(self, workflow_id: str, session_data: dict):
        """记录一个会话的摘要"""
        summary = {
            "workflow_id": workflow_id,
            "timestamp": time.time(),
            "user_input": session_data.get("user_input", ""),
            "key_outputs": session_data.get("key_outputs", []),
            "decisions_made": session_data.get("decisions", []),
            "errors_encountered": session_data.get("errors", []),
        }

        # 追加到缓存
        with open(self.cache_file, "a") as f:
            f.write(json.dumps(summary) + "\n")

        # 如果超过max_sessions,删除最旧的
        await self._prune_old_sessions()

    async def _prune_old_sessions(self):
        """只保留最近 max_sessions 条会话摘要"""
        with open(self.cache_file, "r") as f:
            lines = f.readlines()
        if len(lines) > self.max_sessions:
            with open(self.cache_file, "w") as f:
                f.writelines(lines[-self.max_sessions:])

    async def get_recent_sessions(self, workflow_id: str, count: int = 5) -> List[dict]:
        """快速获取最近的几个会话"""
        sessions = []
        with open(self.cache_file, "r") as f:
            for line in f:
                session = json.loads(line)
                if session["workflow_id"] == workflow_id:
                    sessions.append(session)

        return sessions[-count:]  # 返回最近的count个
```

## 6.1.4 向量索引与语义检索

对于大规模的长期记忆，必须使用向量索引来支持高效的语义搜索。

### 向量库选型对比

| 向量库          | 许可证              | 适用规模      | 2026 现状             |
| ------------ | ---------------- | --------- | ------------------- |
| **Hnswlib**  | Apache 2.0       | <1M 向量    | 稳定内存索引、C++ 优化、无运维成本 |
| **Qdrant**   | Apache 2.0/商业云   | 1M-10M 向量 | 高性能 Rust、K8s 部署成熟   |
| **pgvector** | PostgreSQL 许可    | <5M 向量    | HNSW 算法、深度 SQL 集成   |
| **Weaviate** | BSD-3-Clause/商业云 | 10M+ 向量   | 多模态支持、GraphQL API   |

**选型建议**：简单应用使用 Hnswlib，中等规模选 Qdrant，已有 PostgreSQL 基础设施选 pgvector，大规模企业级选 Weaviate。

```python
# EMBEDDING_DIM 需与所用 embedding 模型一致：
# text-embedding-3-small=1536, text-embedding-3-large=3072, voyage-3-large=1024

class MemoryVectorIndex:
    """向量索引用于语义搜索"""

    # 关键：维度必须与 embedding 模型匹配
    # 默认 1536 对应 text-embedding-3-small；如使用其他模型需修改
    EMBEDDING_DIM = 1536  # 默认使用 text-embedding-3-small 维度

    def __init__(self, index_path: str, embedding_dim: int = None):
        # 使用轻量级的向量库(如Hnswlib或Faiss)
        # dim 参数必须与所用 embedding 模型的输出维度一致
        dim = embedding_dim or self.EMBEDDING_DIM
        self.index = hnswlib.Index(space='cosine', dim=dim)
        self.index.load_index(index_path)
        self.metadata = {}  # 存储向量到原始数据的映射

    async def search_memory(self, query: str, top_k: int = 5) -> List[dict]:
        """语义搜索记忆"""
        # 将查询嵌入
        query_embedding = await self._embed(query)

        # 搜索最相似的向量
        labels, distances = self.index.knn_query(query_embedding, k=top_k)

        # 返回元数据
        results = [
            {
                "similarity_score": 1 - distances[0][i],  # 归一化
                "memory_content": self.metadata[label],
            }
            for i, label in enumerate(labels[0])
        ]

        return results
```

## 6.1.5 与 Claude Code 7 层架构的对比

Harness 的三层架构与 Claude Code 的七层递进式架构体现了不同的优化目标：

| 维度        | Harness                           | Claude Code               |
| --------- | --------------------------------- | ------------------------- |
| **触发精度**  | 三门机制（时间、计数、显式）                    | 七层防御，每层阈值独立               |
| **压缩策略**  | 统一的 autoDream 管道                  | 渐进式微压缩 → 完整压缩，差异巨大        |
| **存储媒介**  | 集中式（MEMORY.md + embedding\_index） | 分散式（工具结果冻结 + 会话笔记 + 梦想合并） |
| **成本结构**  | 整合时一次性成本较高                        | 前三层几乎无成本，第四层才产生明显费用       |
| **跨会话恢复** | 向量搜索 + 结构化查询                      | 预构建摘要注入，零查询成本             |

**设计启示**：Claude Code 的“冻结工具结果”策略（第 1-2 层）及“会话摘要预注入”（第 3 层）可直接借鉴至 Harness。特别是对于高频、短周期工作流，避免每次都触发完整 autoDream 管道能显著降低成本。

## 6.1.6 本小节小结

Harness 的记忆系统通过以下机制支持长期学习：

1. **三门触发机制**：通过时间、会话计数或显式触发来激活整合
2. **四阶段管道**：Orient→Gather→Consolidate→Prune 的标准化流程
3. **去重与去噪**：自动清理冗余和过期信息
4. **多模态存储**：结构化(MEMORY.md)、向量(embedding\_index)、日志(session\_logs)
5. **快速检索**：向量索引支持语义搜索，JSON 缓存支持秒级访问

关于记忆的认知基础和理论模型，请参阅《智能体 AI 权威指南》和《上下文工程权威指南》。