11_主流 agent 记忆简介

方案概览

四大主流方案对比

方案	代表产品	核心机制	优势	劣势	适用场景
Vector RAG	ChatGPT, Manus	向量检索 + 显式记忆	成熟稳定、成本可控	碎片化、语义漂移、容易”断片”	通用对话场景
Context Caching	Gemini 1.5	长上下文缓存到显卡	连贯性强、检索准确、快	成本高、算力要求极高	高价值场景、深度分析
Agentic Search	Claude Code	实时查询 + 工具调用	时效性强、数据新鲜、真实	延迟高、依赖工具质量	代码开发、动态数据
File-based Memory	ClawdBot, OpenClaw	文件系统 + 混合检索	可审计、可编辑、透明	需要良好的组织结构	个人助手、知识管理

技术深度解析

1. Vector RAG：成熟但有局限

核心机制

对话 → 切分为碎片 → Embedding → 存入向量数据库
查询 → Embedding → 相似度检索 → Top-K → 注入 Prompt

显式 Core Memory 示例

{
  "user_name": "张三",
  "preferences": {
    "languages": ["Python", "Rust"],
    "dislikes": ["JavaScript"]
  },
  "facts": [
    "是一位精通多种编程语言的工程师",
    "偏好简洁准确的技术方案"
  ]
}

典型问题：碎片化与语义漂移

# 问题 1: 时间线混乱
chunks = [
    "2026-01-01: 用户喜欢 Python",
    "2026-01-15: 用户开始学习 Rust",
    "2026-02-01: 用户现在更喜欢 Rust"
]
# Top-K 检索可能返回矛盾信息，缺乏时间权重
 
# 问题 2: 上下文丢失
original = "用户在做机器学习项目时，偏好使用 PyTorch 而不是 TensorFlow"
chunk_1 = "用户偏好使用 PyTorch"  # 丢失了"机器学习项目"的上下文
chunk_2 = "不是 TensorFlow"        # 完全失去语义
 
# 问题 3: "串味"
query = "推荐一个 Web 框架"
retrieved = [
    "用户喜欢 Python",           # 相关但不精确
    "另一个用户使用 Django",     # 串到其他用户
    "FastAPI 性能很好"           # 通用知识，非个性化
]

2. Gemini Context Caching：Google 的黑科技

技术原理

KV Cache 机制:

传统方式: 每次都重新计算整个上下文
Token 1 → Token 2 → Token 3 → ... → Token N

Context Caching: 缓存中间激活值到 GPU
[Cached: Token 1...1M] → 只计算新 Token

核心优势

超长上下文: 支持 100 万+ token（约 70 万汉字）
缓存命中: 命中率高达 90%+，响应速度提升 10-100 倍
成本优化: 缓存内容按 1/10 价格计费

代码示例

import google.generativeai as genai
 
# 1. 创建缓存内容（可以是整本书、完整代码库）
cached_content = genai.caching.CachedContent.create(
    model='gemini-1.5-pro-001',
    system_instruction='你是一个代码助手',
    contents=[
        # 可以缓存百万 token 的代码库
        large_codebase_content
    ],
    ttl=datetime.timedelta(hours=1)  # 缓存时长
)
 
# 2. 使用缓存进行对话
model = genai.GenerativeModel.from_cached_content(cached_content)
response = model.generate_content("这个函数的作用是什么？")

适用场景

长文档分析（法律合同、学术论文）
大型代码库理解
多轮深度对话
需要完整上下文的复杂任务

技术要求

需要 Google Cloud 基础设施
对算力和显存要求极高
成本较传统方案高（但比重复计算便宜）

参考文档:

3. Claude Agentic Search：极客的实时查询

核心思想

# "Bash is All You Need"
# "Everything is File"
# "存储的记忆都是过期的，只有现查的才是真实的"

工具调用示例

# 场景 1: 查找最近修改的 Python 文件
ls -lt | grep ".py" | head -n 10
 
# 场景 2: 搜索函数定义
rg "def process_data" --type py
 
# 场景 3: 查看 Git 历史
git log --oneline --since="1 week ago" | grep "feature"
 
# 场景 4: 分析依赖
cat package.json | jq '.dependencies'
 
# 场景 5: 统计代码行数
tokei --sort lines

Model Context Protocol (MCP)

架构设计:

┌─────────────┐
│   Claude    │
└──────┬──────┘
       │ MCP Protocol
       │
┌──────┴──────────────────────────┐
│        MCP Servers              │
├─────────────┬───────────────────┤
│ Filesystem  │ Git │ Database    │
│ HTTP API    │ Shell │ Custom... │
└─────────────┴───────────────────┘

MCP Server 示例:

from mcp import Server, Tool
 
class FileSystemServer(Server):
    @Tool(name="read_file")
    async def read_file(self, path: str) -> str:
        """读取文件内容"""
        with open(path, 'r') as f:
            return f.read()
    
    @Tool(name="search_files")
    async def search_files(self, pattern: str, path: str = ".") -> list:
        """使用 ripgrep 搜索文件"""
        result = subprocess.run(
            ["rg", pattern, path, "--json"],
            capture_output=True
        )
        return json.loads(result.stdout)

解决的核心问题

时效性问题:

# 传统 RAG: 可能返回过期信息
vector_db.search("项目依赖")  
# → 返回 3 天前的 package.json
 
# Agentic Search: 实时查询
tools.execute("cat package.json")  
# → 返回当前最新的依赖列表

特点总结

✅ 时效性: 数据永远是最新的
✅ 准确性: 直接访问源数据，无语义漂移
✅ 灵活性: 可以组合多个工具调用
❌ 延迟: 每次都需要实际执行工具
❌ 依赖: 需要配置和维护工具集

参考文档:

4. File-based Memory：记忆即文件

核心理念

"记忆既文件"
- 所有记忆以 Markdown 形式存储在文件系统
- 结合本地 Embedding 和 Grep 进行混合查询
- 人类可读、可编辑、可版本控制

文件组织结构

memories/
├── core/
│   ├── user_profile.md          # 核心用户信息
│   ├── preferences.md           # 偏好设置
│   └── relationships.md         # 人际关系图谱
├── conversations/
│   ├── 2026-02/
│   │   ├── 2026-02-09.md       # 按日期组织
│   │   └── 2026-02-08.md
│   └── index.md                 # 对话索引
├── knowledge/
│   ├── projects/
│   │   ├── project_x.md        # 项目相关知识
│   │   └── project_y.md
│   ├── technical/
│   │   ├── python_tips.md
│   │   └── rust_patterns.md
│   └── personal/
│       └── life_events.md
└── .embeddings/                 # 本地向量索引
    └── cache.db

记忆文件示例

user_profile.md:

# 用户档案
 
## 基本信息
- 姓名: [保密]
- 角色: 精通多种编程语言的工程师
- 时区: Asia/Shanghai
 
## 技术栈
- **主力语言**: Python, Rust
- **熟悉领域**: 系统设计, 算法优化, AI/ML
- **偏好**: 简洁准确的技术方案，直接进入核心内容
 
## 沟通偏好
- 语言: 中文
- 风格: 专业、高效、无需过多基础解释
- 代码: 可直接运行，包含必要注释
 
## 更新日志
- 2026-02-09: 初始化档案

2026-02-09.md:

# 2026-02-09 对话记录
 
## 主题: AI 记忆方案研究
 
### 讨论要点
1. 对比了 4 种主流 AI 记忆方案
2. 重点关注 Vector RAG 的碎片化问题
3. 对 Agentic Search 的实时性表示认可
 
### 关键洞察
- 用户认为 "存储的记忆都是过期的，只有现查的才是真实的"
- 提出了 RAG 2.0 的概念：用主动探索代替被动回忆
 
### 后续行动
- [ ] 整理完整的学习文档
- [ ] 探索混合方案的实现

RAG 演进路径

RAG 1.0 → RAG 2.0 的核心转变

RAG 1.0: 被动检索 (Passive Retrieval)
────────────────────────────────────
Query → Embedding → Vector Search → Top-K → LLM
特点: 简单、快速、但缺乏智能

RAG 2.0: 主动探索 (Agentic Retrieval)
────────────────────────────────────
Query → Planning → Multi-step Search → Validation → Synthesis
         ↓
    Tool Calls (grep/SQL/API/Web Search/...)
特点: 智能、准确、但复杂度高

关键技术演进

1. Self-RAG：自我反思的检索

class SelfRAG:
    async def retrieve(self, query: str):
        # 步骤 1: 判断是否需要检索
        need_retrieval = await self.llm.decide(
            f"回答 '{query}' 是否需要外部知识？"
        )
        
        if not need_retrieval:
            return await self.llm.generate(query)
        
        # 步骤 2: 检索
        docs = await self.vector_db.search(query)
        
        # 步骤 3: 评估检索质量
        relevance_scores = []
        for doc in docs:
            score = await self.llm.evaluate_relevance(query, doc)
            relevance_scores.append(score)
        
        # 步骤 4: 生成答案
        answer = await self.llm.generate(query, docs)
        
        # 步骤 5: 自我验证
        is_supported = await self.llm.verify(answer, docs)
        
        if not is_supported:
            # 重新检索或承认不知道
            return await self.retry_or_abstain(query)
        
        return answer

2. FLARE：主动检索

class FLARE:
    async def generate_with_active_retrieval(self, query: str):
        answer = ""
        remaining_query = query
        
        while not self.is_complete(answer):
            # 生成一部分答案
            next_sentence, confidence = await self.llm.generate_next(
                query, answer
            )
            
            # 如果置信度低，主动检索
            if confidence < 0.5:
                # 识别需要查询的内容
                search_query = self.extract_uncertain_part(next_sentence)
                
                # 主动检索
                docs = await self.search(search_query)
                
                # 基于检索结果重新生成
                next_sentence = await self.llm.generate_with_docs(
                    query, answer, docs
                )
            
            answer += next_sentence
        
        return answer

3. Agentic RAG：工具增强

class AgenticRAG:
    def __init__(self):
        self.tools = {
            "vector_search": self.vector_search,
            "web_search": self.web_search,
            "sql_query": self.sql_query,
            "file_read": self.file_read,
            "code_execution": self.code_execution
        }
    
    async def retrieve(self, query: str):
        # 步骤 1: 制定检索计划
        plan = await self.llm.plan(
            query=query,
            available_tools=list(self.tools.keys())
        )
        # plan = [
        #     {"tool": "file_read", "args": {"path": "config.json"}},
        #     {"tool": "sql_query", "args": {"query": "SELECT * FROM users"}},
        #     {"tool": "vector_search", "args": {"query": "用户偏好"}}
        # ]
        
        # 步骤 2: 执行计划
        results = []
        for step in plan:
            tool = self.tools[step["tool"]]
            result = await tool(**step["args"])
            results.append(result)
            
            # 动态调整计划
            if self.should_adjust_plan(result):
                plan = await self.llm.replan(query, results, plan)
        
        # 步骤 3: 综合结果
        answer = await self.llm.synthesize(query, results)
        
        return answer

对比总结

特性	RAG 1.0	Self-RAG	FLARE	Agentic RAG
检索时机	查询前	按需	生成中	计划驱动
检索次数	1 次	1-2 次	多次	多次
工具支持	仅向量检索	仅向量检索	仅向量检索	多种工具
自我反思	❌	✅	部分	✅
复杂度	低	中	中	高
准确性	中	高	高	最高

关键论文

Self-RAG: Self-Reflective Retrieval-Augmented Generation
FLARE: Active Retrieval Augmented Generation
Toolformer: Language Models Can Teach Themselves to Use Tools
ReAct: Synergizing Reasoning and Acting in Language Models

工程实践

混合方案架构

基于实际工程需求，推荐采用混合方案：

from typing import List, Dict, Any
from dataclasses import dataclass
from datetime import datetime, timedelta
import asyncio
 
@dataclass
class MemorySource:
    """记忆来源"""
    content: str
    source_type: str  # "core", "vector", "tool", "cache"
    confidence: float
    timestamp: datetime
    metadata: Dict[str, Any]
 
class HybridMemorySystem:
    """混合记忆系统"""
    
    def __init__(self):
        # 1. 文件系统 (Core Memory)
        self.file_store = MarkdownMemoryStore("./memories")
        
        # 2. 向量数据库 (Long-term Memory)
        self.vector_store = ChromaDB()
        
        # 3. 工具集 (Real-time Query)
        self.tools = ToolRegistry([
            GrepTool(),
            GitTool(),
            SQLTool(),
            WebSearchTool()
        ])
        
        # 4. 上下文缓存 (Short-term Memory)
        self.context_cache = LRUCache(maxsize=10)
    
    async def retrieve(
        self, 
        query: str,
        strategy: str = "adaptive"
    ) -> List[MemorySource]:
        """
        自适应检索策略
        
        Args:
            query: 查询内容
            strategy: "adaptive" | "fast" | "accurate" | "comprehensive"
        """
        results = []
        
        # 阶段 1: 核心记忆 (必查)
        core_memory = await self._retrieve_core_memory()
        results.extend(core_memory)
        
        # 阶段 2: 根据策略选择检索方式
        if strategy == "fast":
            # 只查缓存和文件
            cached = self.context_cache.get(query)
            if cached:
                results.append(cached)
            
            file_results = await self.file_store.search(query, method="grep")
            results.extend(file_results)
        
        elif strategy == "accurate":
            # 优先实时工具
            tool_results = await self._tool_search(query)
            results.extend(tool_results)
            
            # 补充向量检索
            vector_results = await self.vector_store.search(query, k=3)
            results.extend(vector_results)
        
        elif strategy == "comprehensive":
            # 全面检索
            parallel_tasks = [
                self.file_store.search(query),
                self.vector_store.search(query, k=5),
                self._tool_search(query)
            ]
            all_results = await asyncio.gather(*parallel_tasks)
            for result_set in all_results:
                results.extend(result_set)
        
        else:  # adaptive
            # 自适应：根据查询类型选择策略
            query_type = await self._classify_query(query)
            
            if query_type == "factual":
                # 事实性查询 → 实时工具
                results.extend(await self._tool_search(query))
            
            elif query_type == "personal":
                # 个人信息 → 文件 + 向量
                results.extend(await self.file_store.search(query))
                results.extend(await self.vector_store.search(query, k=3))
            
            elif query_type == "recent":
                # 最近事件 → 时间过滤
                recent_results = await self.file_store.search(
                    query, 
                    time_range=timedelta(days=7)
                )
                results.extend(recent_results)
        
        # 阶段 3: 结果融合与排序
        ranked_results = self._rank_and_deduplicate(results, query)
        
        # 阶段 4: 更新缓存
        self.context_cache.put(query, ranked_results[:3])
        
        return ranked_results[:10]  # Top-10
    
    async def _retrieve_core_memory(self) -> List[MemorySource]:
        """检索核心记忆（用户档案、偏好等）"""
        core_files = [
            "core/user_profile.md",
            "core/preferences.md"
        ]
        
        results = []
        for file in core_files:
            content = await self.file_store.read(file)
            results.append(MemorySource(
                content=content,
                source_type="core",
                confidence=1.0,
                timestamp=datetime.now(),
                metadata={"file": file}
            ))
        
        return results
    
    async def _tool_search(self, query: str) -> List[MemorySource]:
        """使用工具进行实时查询"""
        # 让 LLM 决定使用哪些工具
        tool_plan = await self.llm.select_tools(
            query=query,
            available_tools=self.tools.list()
        )
        
        results = []
        for tool_call in tool_plan:
            result = await self.tools.execute(
                tool_call["name"],
                **tool_call["args"]
            )
            results.append(MemorySource(
                content=result,
                source_type="tool",
                confidence=0.9,
                timestamp=datetime.now(),
                metadata={"tool": tool_call["name"]}
            ))
        
        return results
    
    def _rank_and_deduplicate(
        self, 
        results: List[MemorySource], 
        query: str
    ) -> List[MemorySource]:
        """结果排序与去重"""
        # 1. 去重（基于内容相似度）
        unique_results = self._deduplicate(results)
        
        # 2. 计算综合得分
        for result in unique_results:
            # 语义相似度
            semantic_score = self._compute_similarity(query, result.content)
            
            # 时效性得分
            age_hours = (datetime.now() - result.timestamp).total_seconds() / 3600
            recency_score = 0.5 ** (age_hours / 168)  # 每周衰减 50%
            
            # 来源可信度
            source_weight = {
                "core": 1.0,
                "tool": 0.9,
                "vector": 0.7,
                "cache": 0.8
            }[result.source_type]
            
            # 综合得分
            result.score = (
                semantic_score * 0.5 +
                recency_score * 0.2 +
                result.confidence * source_weight * 0.3
            )
        
        # 3. 排序
        unique_results.sort(key=lambda x: x.score, reverse=True)
        
        return unique_results
    
    async def _classify_query(self, query: str) -> str:
        """分类查询类型"""
        # 使用小模型快速分类
        classification = await self.classifier.predict(query)
        return classification  # "factual" | "personal" | "recent" | "general"

实际应用场景

场景 1: 代码助手

# 用户: "上次那个 API 的认证方式是什么？"
 
# 系统决策:
# 1. 查询类型: "recent" + "factual"
# 2. 检索策略:
#    - Tool: git log + rg "auth" "API"
#    - File: 搜索最近 7 天的对话记录
# 3. 结果:
#    - Git 历史: 3 天前提交了 OAuth2 实现
#    - 对话记录: 讨论过使用 JWT
#    - 代码文件: auth.py 中的实现细节

场景 2: 个人助手

# 用户: "我喜欢什么编程语言？"
 
# 系统决策:
# 1. 查询类型: "personal"
# 2. 检索策略:
#    - Core: user_profile.md (必查)
#    - Vector: 搜索历史对话中的偏好表达
# 3. 结果:
#    - Core Memory: "主力语言: Python, Rust"
#    - 对话历史: 多次表达对 Rust 的兴趣增加
#    - 综合答案: "你主要使用 Python 和 Rust，最近对 Rust 的兴趣在增加"

场景 3: 知识问答

# 用户: "RAG 2.0 和 RAG 1.0 有什么区别？"
 
# 系统决策:
# 1. 查询类型: "general" + "recent"
# 2. 检索策略:
#    - File: 搜索今天的对话记录（刚讨论过）
#    - Vector: 搜索相关技术文档
#    - Tool: 如果需要，可以 Web 搜索最新论文
# 3. 结果:
#    - 今日对话: 完整的对比分析
#    - 知识库: 相关论文和实现

总结

方案选择指南

场景	推荐方案	理由
通用对话机器人	Vector RAG	成本低、成熟稳定
企业知识库	Vector RAG + File-based	可审计、可控制
代码助手	Agentic Search + File-based	时效性强、准确度高
长文档分析	Context Caching	上下文完整、理解深入
个人 AI 助手	Hybrid (混合方案)	灵活、可定制

🪴 Quartz 4.0

Explorer

11_主流 agent 记忆简介

方案概览

四大主流方案对比

技术深度解析

1. Vector RAG：成熟但有局限

核心机制

显式 Core Memory 示例

典型问题：碎片化与语义漂移

相关技术栈

2. Gemini Context Caching：Google 的黑科技

技术原理

核心优势

代码示例

适用场景

技术要求

3. Claude Agentic Search：极客的实时查询

核心思想

工具调用示例

Model Context Protocol (MCP)

解决的核心问题

特点总结

4. File-based Memory：记忆即文件

核心理念

文件组织结构

记忆文件示例

相关项目

RAG 演进路径

RAG 1.0 → RAG 2.0 的核心转变

关键技术演进

1. Self-RAG：自我反思的检索

2. FLARE：主动检索

3. Agentic RAG：工具增强

对比总结

关键论文

工程实践

混合方案架构

实际应用场景

场景 1: 代码助手

场景 2: 个人助手

场景 3: 知识问答

总结

方案选择指南

Graph View

Table of Contents

Backlinks