# 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.md"]
    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
        self.cache_file = "~/.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 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**   | AGPL v3/商业    | 1M-10M 向量 | v1.7+、高性能 Rust、K8s 部署成熟 |
| **pgvector** | PostgreSQL 许可 | <5M 向量    | v0.5+、HNSW 算法、深度 SQL 集成 |
| **Weaviate** | BSL 1.1/商业    | 10M+ 向量   | v1.6+、多模态支持、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 权威指南》和《上下文工程权威指南》。


---

# 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.1_memory_architecture.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.