🧠 构建 Agent 实践指南 — 记忆系统

# smolagents/src/smolagents/agent.py — 极简
class MultiStepAgent:
    def run(self, task):
        self.memory.append({"role": "user", "content": task})
        while not done:
            response = self.model(self.memory)  # 全量传给模型
            self.memory.append(response)
            # ...

全量消息列表直接传模型，没有任何压缩或截断逻辑。对于工具调用链有限的简单任务，这完全够了。

SWE-agent 的 ACI（Agent-Computer Interface）将每一步的观测作为消息传入。当步数超过 max_history 时，丢弃最早的观测：

# swe-agent/sweagent/agent/agents.py
class Agent:
    def _format_trajectory(self):
        # 只保留最近 max_history 步
        trajectory = self.history[-self.config.max_history:]
        return self.templates.format_trajectory(trajectory)

对于 SWE-bench 这种有限步数的任务（通常 20-50 步），滑窗足够。丢掉的早期步骤通常是初始化探索，对后续修复影响不大。

Aider 的压缩策略是 递归分治：

# aider/aider/history.py
class ChatSummary:
    def summarize_real(self, messages, depth=0):
        if total <= self.max_tokens and depth == 0:
            return messages  # 没超限，直接返回

        # 从尾部向前扫描，保留后半部分原文
        tail_tokens = 0
        split_index = len(messages)
        for i in range(len(sized) - 1, -1, -1):
            if tail_tokens + tokens < half_max_tokens:
                tail_tokens += tokens
                split_index = i
            else:
                break

        # 前半部分用 LLM 摘要
        summary = self.summarize_all(keep)  # 调小模型压缩
        # 组合: [摘要] + [近期原文]
        return summary + tail

关键设计决策：

• 保留近期原文 — 最近的消息不压缩，保留完整上下文
• 递归压缩 — 如果摘要+尾部还是太长，递归压缩摘要部分
• 模型降级 — 压缩用便宜的小模型（gpt-3.5-turbo），不浪费主力模型的 token
• 深度限制 — 最多递归 3 层，防止无限循环

OpenHands 把压缩抽象为 Condenser 接口，与主循环解耦：

# openhands SDK
class LLMSummarizingCondenser:
    # 默认: max_size=240, keep_first=2
    # 每轮: 如果事件数 > max_size
    #   → 保留前 keep_first 条 (系统提示词)
    #   → LLM 摘要中间部分
    #   → 保留最近的 N 条

    condenser_kwargs = {
        'llm_type': 'condenser' if is_coding_agent else 'planning_condenser',
    }
    condenser = LLMSummarizingCondenser(**condenser_kwargs)

亮点：两种 Condenser — coding agent 用通用 condenser，planning agent 用专用的 planning_condenser。不同任务类型有不同的压缩策略。

MetaGPT 的三层记忆架构中，BrainMemory 是中间层：

# metagpt/memory/brain_memory.py
class BrainMemory(BaseModel):
    history: List[Message] = []          # 短期：原始消息
    knowledge: List[Message] = []        # 中期：提取的知识
    historical_summary: str = ""         # 长期：历史摘要

    async def set_history_summary(self, summary, redis_key):
        # 压缩后清空 history，保留 summary
        self.historical_summary = summary
        self.history = []  # 原始消息已压缩，释放空间
        await self.dumps(redis_key)  # Redis 持久化

三层分工：history（短期可丢）→ knowledge（中期结构化）→ historical_summary（长期压缩）。

Agent Zero 是向量检索记忆的典范实现——三区域分区 + LLM 辅助查询 + 智能合并：

# agent-zero/plugins/_memory/helpers/memory.py
class Memory:
    class Area(Enum):
        MAIN = "main"           # 手动保存的核心记忆
        FRAGMENTS = "fragments" # 自动提取的信息片段
        SOLUTIONS = "solutions" # 成功解决方案

写入流程（monologue_end 扩展）：

# agent-zero/plugins/_memory/extensions/python/monologue_end/_50_memorize_fragments.py
# LLM 提取 → 合并 → 存储
memories_json = await self.agent.call_utility_model(
    system=system, message=msgs_text, background=True,
)
memories = DirtyJson.parse_string(memories_json)

for memory in memories:
    consolidator = create_memory_consolidator(
        self.agent,
        similarity_threshold=DEFAULT_MEMORY_THRESHOLD,
        max_similar_memories=8,
        max_llm_context_memories=4
    )
    result_obj = await consolidator.process_new_memory(
        new_memory=txt,
        area=Memory.Area.FRAGMENTS.value,
    )

读取流程（monologue_start 扩展）：

# agent-zero/plugins/_memory/extensions/python/message_loop_prompts_after/_50_recall_memories.py
# Step 1: LLM 生成搜索查询
query = await self.agent.call_utility_model(
    system=system, message=message,
)

# Step 2: 分区搜索
memories = await db.search_similarity_threshold(
    query=query, limit=12, threshold=threshold,
    filter=f"area=='main' or area=='fragments'",
)
solutions = await db.search_similarity_threshold(
    query=query, limit=8, threshold=threshold,
    filter=f"area=='solutions'",
)

# Step 3: LLM 后过滤（验证相关性）
filter = await self.agent.call_utility_model(
    system=self.agent.read_prompt("memory.memories_filter.sys.md"),
    message=...,
)
# Step 4: 注入到 prompt
extras["memories"] = self.agent.parse_prompt(
    "agent.system.memories.md", memories=memories_txt
)

关键设计决策：

• 3 区域分区 — 不同类型的记忆有不同的检索策略和阈值
• LLM 辅助查询生成 — 用户消息可能不是好的搜索查询，让 LLM 提炼
• 后过滤 — 向量搜索可能返回不相关的结果，用 LLM 二次验证
• 后台异步 — 记忆操作在后台线程执行，不阻塞主循环
• 项目隔离 — 不同项目使用独立的 FAISS 索引子目录

CrewAI 的记忆系统是向量检索的进阶版——LLM 分析 + RecallFlow + 复合评分：

# crewai/memory/unified_memory.py
class Memory(BaseModel):
    # 三维复合评分
    recency_weight: float = 0.3      # 时间衰减
    semantic_weight: float = 0.5     # 语义相似度
    importance_weight: float = 0.2   # 重要性

    # 深度读取流程
    def recall(self, query, depth="deep"):
        if depth == "shallow":
            # 直接向量搜索
            return self._storage.search(embedding, ...)
        else:
            # RecallFlow: LLM 拆解查询 → 多子查询并行搜索 → 置信度路由
            flow = RecallFlow(storage, llm, embedder, config)
            flow.kickoff(inputs={...})
            return flow.state.final_results

    # 智能编码
    def _encode_batch(self, contents, ...):
        flow = EncodingFlow(storage, llm, embedder, config)
        # LLM 自动推断: scope, categories, importance, metadata
        flow.kickoff(inputs={"items": items_input})

RecallFlow 的独特之处：置信度路由 — 如果浅层搜索的置信度高，直接返回；否则启动深度探索（LLM 拆解子查询、多轮搜索）。

Hermes 的记忆架构最灵活——MemoryProvider ABC + 插件发现：

# hermes-agent/agent/memory_provider.py
class MemoryProvider(ABC):
    # 核心生命周期
    def initialize(session_id, **kwargs)     # 初始化
    def system_prompt_block() -> str          # 系统提示词注入
    def prefetch(query) -> str                # 每轮前预取
    def queue_prefetch(query)                 # 后台预取下一轮
    def sync_turn(user, assistant)            # 每轮后同步
    def get_tool_schemas() -> List[Dict]      # 暴露的工具
    def handle_tool_call(tool_name, args)     # 工具调用

    # 可选钩子
    def on_session_end(messages)              # 会话结束
    def on_pre_compress(messages) -> str      # 压缩前提取
    def on_session_switch(new_id, ...)        # 会话切换
    def on_memory_write(action, target, ...)  # 镜像写入
    def on_delegation(task, result, ...)      # 子代理观察

内置 8 个 MemoryProvider 实现：

• holographic — 全息编码 (HRR) + 实体解析 + 信任评分
• honcho — 外部语义记忆服务
• mem0 — Mem0 集成
• hindsight — Hindsight 记忆
• supermemory — SuperMemory API
• byterover — ByteRover 集成
• retaindb — RetainDB 向量存储
• openviking — OpenViking 提供商

插件发现机制：

# hermes-agent/plugins/memory/__init__.py
# 扫描两个目录:
# 1. 内置: plugins/memory/<name>/
# 2. 用户: $HERMES_HOME/plugins/<name>/
# 名称冲突时内置优先
# 配置 memory.provider = "holographic" 选择激活哪个

这是目前 Agent 记忆系统中最精密的实现：

HRR 全息编码

# hermes-agent/plugins/memory/holographic/holographic.py
# HRR (Holographic Reduced Representation) 核心思想:
# 将事实和实体编码为高维相位向量 (默认 1024 维)
# 支持组合操作: 绑定(bind)和解绑(unbind)
# 绑定: fact ⊗ entity → 组合编码
# 解绑: pattern ⊗ entity⁻¹ → 恢复关联事实

# 这使得 "reason" 操作成为可能:
# 给定 [entity_A, entity_B]，找到同时关联两者的事实
# 传统向量搜索无法做到这一点

信任评分系统

# hermes-agent/plugins/memory/holographic/store.py
# 新事实: default_trust = 0.5
# 每次被检索: retrieval_count += 1
# 用户标记 helpful:   trust += 0.05, helpful_count += 1
# 用户标记 unhelpful: trust -= 0.10  (不对称惩罚!)
# trust 范围: [0.0, 1.0]
# 检索时过滤: min_trust = 0.3 (默认)

# 时间衰减 (可选):
# temporal_decay_half_life → 老事实自然降权

实体解析

# 从事实文本中自动提取实体:
# 1. 大写多词短语: "John Doe"
# 2. 双引号术语: "Python"
# 3. AKA 模式: "Guido aka BDFL" → 两个实体
# 实体别名解析: 搜索 aliases 字段 (逗号分隔)
# 新实体自动创建并链接到事实

自动提取 (会话结束时)

# hermes-agent/plugins/memory/holographic/__init__.py
# on_session_end → _auto_extract_facts()
# 正则匹配用户偏好和决策:
#   "I prefer/like/love/use/want/need ..."
#   "my favorite/preferred/default ... is ..."
#   "we decided/agreed/chose to ..."
# 匹配成功 → 自动存为 user_pref 或 project 类别事实

Agent Zero 的合并器在写入时自动处理重复和冲突：

# agent-zero/plugins/_memory/helpers/memory_consolidation.py
# process_new_memory(new_memory, area, metadata):
# 1. 向量搜索相似记忆 (similarity_threshold, max_similar_memories=8)
# 2. 将新记忆 + 最相似的 N 条送给 LLM
# 3. LLM 决定: 合并? 替换? 新建?
# 4. 执行: 合并→更新原文; 替换→删除旧的+插入新的; 新建→直接插入

# 没有合并器时 (简单模式):
# 设置 memory_memorize_replace_threshold > 0
# → 删除相似度超过阈值的旧记忆，插入新的

RepoMap 的核心问题：一个仓库可能有上千个文件、几万行代码，但模型窗口只有 128K token。哪些代码应该放进去？

# aider/aider/repomap.py
class RepoMap:
    def get_ranked_tags(self, chat_fnames, other_fnames, ...):
        # 1. tree-sitter 解析每个文件 → 提取 Tag(name, kind="def"/"ref")
        for tag in tags:
            if tag.kind == "def":
                defines[tag.name].add(rel_fname)
            elif tag.kind == "ref":
                references[tag.name].append(rel_fname)

        # 2. 构建 NetworkX 有向图
        #    节点 = 文件, 边 = 定义-引用关系
        #    权重 = 引用次数

        # 3. PageRank + personalization
        #    - chat_files (正在编辑的文件): 高权重
        #    - mentioned_fnames (对话中提到的): 高权重
        #    - mentioned_idents (代码中提到的标识符): 高权重
        personalization[rel_fname] = current_pers

        # 4. 运行 PageRank
        G = nx.MultiDiGraph()
        ranks = nx.pagerank(G, personalization=personalization)

        # 5. 按排名选择文件，生成树形代码摘要

关键洞察：PageRank + personalization — 不是简单的文件重要性排名，而是结合当前对话上下文动态调整。你正在编辑的文件、对话中提到的函数名，都会影响哪些代码被选中。

生成的 RepoMap 是代码的结构化摘要：

# 示例输出:
# aider/coders/base_coder.py:
# class BaseCoder:
#   def __init__(self, ...)
#   def run(self, ...)
#   def format_messages(self, ...)
# aider/repomap.py:
# class RepoMap:
#   def get_repo_map(self, ...)
#   def get_ranked_tags(self, ...)

只保留类名、函数签名、定义行——去掉函数体。这是"记忆"的另一种理解：不是记住对话历史，而是记住代码结构。

# langgraph/libs/checkpoint/langgraph/checkpoint/base/__init__.py

class CheckpointMetadata(TypedDict):
    source: Literal["input", "loop", "update", "fork"]
    step: int
    parents: dict[str, str]  # 父 checkpoint ID 映射

class Checkpoint(TypedDict):
    v: int              # 版本号
    id: str             # UUID
    ts: str             # 时间戳
    channel_values: dict  # 各 channel 的值
    channel_versions: dict  # 各 channel 的版本
    versions_seen: dict   # 每个 node 看到的版本

class BaseCheckpointSaver(Generic[V]):
    def put(config, checkpoint, metadata, new_versions) -> RunnableConfig
    def get_tuple(config) -> CheckpointTuple | None
    def list(config, *, before, limit) -> Iterator[CheckpointTuple]
    def put_writes(config, writes, task_id) -> None

关键设计：

• Channel 版本化 — 每个 state channel 独立版本号，支持增量写入
• 父链 — parents 字段记录分叉来源，支持分支追踪
• 多种后端 — MemorySaver (开发), SqliteSaver (单机), PostgresSaver (生产)
• Delta Snapshot — 只写入变化的 channel，节省存储

Agent Zero 也实现了类似的时间旅行机制：

# agent-zero/plugins/_time_travel/helpers/time_travel.py
# 记录每一步的完整状态快照
# 用户可以回溯到任意步骤，从那里重新开始

Agent Zero 区分两种记忆写入：

# Fragments: 自动提取的信息片段
# → 每轮 monologue_end 后，LLM 从历史中提取
# → 用 Consolidator 智能合并
memories_json = await self.agent.call_utility_model(
    system=self.agent.read_prompt("memory.memories_sum.sys.md"),
    message=msgs_text,
    background=True,  # 后台执行，不阻塞
)

# Solutions: 成功的解决方案
# → 只记录有明确 problem-solution 结构的
# → 合并时 max_similar_memories=6 (更谨慎)
if isinstance(solution, dict):
    problem = solution.get('problem', 'Unknown problem')
    solution_text = solution.get('solution', 'Unknown solution')
    txt = f"# Problem\n {problem}\n# Solution\n {solution_text}"

# Hermes 记忆读取的时序:
# 
# Turn N 开始:
#   1. prefetch(query=user_message)
#      → 返回缓存的上一轮预取结果 (零延迟!)
#   2. [模型处理...]
#   3. Turn N 结束:
#      queue_prefetch(query=assistant_response)
#      → 后台启动下一轮的预取
# 
# Turn N+1 开始:
#   1. prefetch(query=user_message)
#      → 直接返回上一轮后台预取的结果

# 这样实现了零延迟的记忆检索:
# 用户感受不到记忆检索的延迟

# CrewAI RecallFlow 的自适应深度:
# 
# 浅层搜索 → 计算置信度
# ├── confidence > 0.8 → 直接返回 (高置信)
# ├── 0.5 < confidence < 0.8 → 
# │   如果是复杂查询 (score < 0.7) → LLM 拆解子查询，深度探索
# │   否则 → 返回浅层结果
# └── confidence < 0.5 → 深度探索

# 深度探索:
# 1. LLM 生成多个子查询
# 2. 并行搜索
# 3. 合并去重
# 4. 迭代: 如果还不满意，再探索一轮 (exploration_budget=1)

向量搜索返回的是"语义相似"的结果，但相似 ≠ 相关。Agent Zero 的后过滤机制：

# agent-zero: LLM 后过滤
# 1. 将搜索结果编号: {0: "事实A", 1: "事实B", 2: "方案C"}
# 2. 让 LLM 判断哪些与当前对话真正相关
# 3. 只保留 LLM 认可的结果

filter = await self.agent.call_utility_model(
    system=self.agent.read_prompt("memory.memories_filter.sys.md"),
    message=self.agent.read_prompt(
        "memory.memories_filter.msg.md",
        memories=mems_list,  # 编号的候选记忆
        history=history,     # 当前对话
        message=user_instruction,  # 用户消息
    ),
)
filter_inds = dirty_json.try_parse(filter)  # [0, 2] → 保留第 0 和第 2 个

目标：让 Agent 能运行，不爆上下文窗口。

# Phase 1 记忆实现 (~50 行)
class SimpleMemory:
    def __init__(self, max_messages=50):
        self.messages = []
        self.max_messages = max_messages
    
    def add(self, msg):
        self.messages.append(msg)
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]
    
    def get_context(self):
        return self.messages  # 直接传给模型

适用：一次性任务、短对话、原型验证。

目标：支持长对话，保持连贯性。

# Phase 2: 加入摘要压缩 (~150 行)
class CompressedMemory(SimpleMemory):
    def __init__(self, max_messages=50, max_tokens=4096, summary_model=None):
        super().__init__(max_messages)
        self.summary = ""
        self.max_tokens = max_tokens
        self.summary_model = summary_model  # 用便宜的小模型!
    
    def get_context(self):
        # 三段式: [摘要] + [近期原文]
        context = []
        if self.summary:
            context.append({"role": "system", "content": f"对话摘要:\n{self.summary}"})
        context.extend(self.messages)
        return context
    
    def compress(self):
        # 当 token 数超限时，压缩旧消息
        if self._estimate_tokens() > self.max_tokens:
            old = self.messages[:len(self.messages)//2]
            self.summary = self._summarize(old)
            self.messages = self.messages[len(self.messages)//2:]

关键决策：用小模型做压缩，保留近期原文。

目标：跨会话知识召回，记住用户偏好。

# Phase 3: 加入向量检索 (~300 行)
class SemanticMemory(CompressedMemory):
    def __init__(self, vector_store, embedder, **kwargs):
        super().__init__(**kwargs)
        self.vector_store = vector_store  # FAISS/Chroma/LanceDB
        self.embedder = embedder

    def remember(self, text, metadata=None):
        # 显式记忆写入
        embedding = self.embedder.embed(text)
        self.vector_store.add(embedding, text, metadata)

    def recall(self, query, top_k=5, threshold=0.7):
        # 语义检索
        embedding = self.embedder.embed(query)
        results = self.vector_store.search(embedding, top_k=top_k)
        return [r for r in results if r.score >= threshold]

    def get_context(self):
        # 四段式: [摘要] + [检索记忆] + [近期原文]
        context = []
        if self.summary:
            context.append({"role": "system", "content": f"摘要:\n{self.summary}"})
        recalled = self.recall(self.messages[-1].content if self.messages else "")
        if recalled:
            context.append({"role": "system", "content": f"相关记忆:\n{recalled}"})
        context.extend(self.messages)
        return context

目标：记忆自我维护，防止碎片化和过时。

# Phase 4: 信任 + 合并 + 矛盾检测
class EvolvingMemory(SemanticMemory):
    # 1. 信任评分
    def record_feedback(self, fact_id, helpful):
        delta = 0.05 if helpful else -0.10  # 不对称惩罚
        self._adjust_trust(fact_id, delta)

    # 2. 合并去重
    def remember(self, text, metadata=None):
        similar = self.recall(text, top_k=5, threshold=0.85)
        if similar:
            # 让 LLM 决定: 合并 or 替换 or 新建
            decision = self._consolidate(text, similar)
            if decision == "merge":
                return self._merge(similar[0], text)
            elif decision == "replace":
                self._delete(similar[0])
        # 新建
        super().remember(text, metadata)

    # 3. 矛盾检测 (定期运行)
    def detect_contradictions(self):
        # 找到语义相似但内容冲突的记忆对
        # 标记为待审查
        pass

📐 架构原则

1. 写入和读取分离 — 写入可以是异步的（Agent Zero 的 DeferredTask），读取必须同步或预取（Hermes 的 queue_prefetch）
2. 分层存储 — 热数据(近期原文) + 温数据(摘要) + 冷数据(向量库) — 类似 CPU 缓存层级
3. 可插拔后端 — Hermes 的 MemoryProvider ABC 是最佳实践，允许用户选择不同的记忆后端
4. 预算控制 — 明确 token 预算：系统提示词占多少、记忆占多少、对话占多少

⚡ 性能原则

1. 预取流水线 — Hermes 的 queue_prefetch 模式：当前轮结束时预取下一轮的记忆，零延迟
2. 后台写入 — Agent Zero 的 DeferredTask：记忆写入不阻塞主循环
3. 缓存嵌入 — Agent Zero 的 CacheBackedEmbeddings：嵌入结果缓存到磁盘，避免重复计算
4. 增量索引 — LangGraph 的 Delta Snapshot：只持久化变化的 channel

🛡️ 安全原则

1. 隔离 — Agent Zero 的项目隔离：不同项目使用独立的 FAISS 索引
2. 隐私 — CrewAI 的 private 标志：私人记忆只对同一 source 可见
3. 验证 — Agent Zero 的 LLM 后过滤：防止不相关记忆污染上下文
4. 审计 — Hermes 的 metadata：每次写入附带来源、会话、平台信息

🔄 运维原则

1. 哈希校验 — Agent Zero 的 FAISS index hash：检测索引损坏并自动重建
2. 嵌入模型迁移 — Agent Zero：检测嵌入模型变化时自动重建索引
3. WAL 模式 — Hermes：SQLite WAL + NFS 降级处理
4. 序列化安全 — FAISS 的 allow_dangerous_deserialization 警告