记忆与检索 —— Hello-Agent Chapter 8 Learning

记忆与检索

第八章 记忆与检索

代码仓库

框架现有的缺陷？

最显著的一点是：所有的历史上下文，都只在程序运行期间存在的（也就是内存里），程序运行完毕历史就没了，会话之间的上下文也没法共享。

另一个缺陷则是，我们太过于依赖 LLM 本身的能力了，我们必须承认不同供应商的 LLM 的知识库水平是不同的，比如 gemini 知识库很全，claude 在 coding 知识上很强。完全依靠 LLM 的知识库是不稳定的，我们需要构建自己的知识库。

所以我们的框架，当务之急是解决以上问题，我们最容易想到的是将历史上下文持久化，这样就可以解决上下文共享，也可以建立我们自己的知识库。

上手体验：MemoryTool + RAG

官方依然给出了完整的 rag 组件，使用 pip 安装：

pip install "hello-agents[all]==0.2.9"
python -m spacy download zh_core_web_sm
python -m spacy download en_core_web_sm

随后，需要三个组件：Qdrant、Neo4J、Embedding，这里我们使用 docker 进行本地部署前两个，Embedding 模型用云服务。

Qdrant

Qdrant 是向量数据库，擅长存储实体之间的 语义相似度，主要用于 语义搜索 和 Rag增强。也就是说，可以将 Qdrant 作为文本知识库。

在 docker 部署：

docker pull qdrant/qdrant
mkdir /mydata/qrant

docker run -d \
--name qdrant-helloagent \
-p 6333:6333 \
-p 6334:6334 \
-v /mydata/qdrant \
--restart always \
qdrant/qdrant

数据放在哪个文件夹，自定义。

创建完毕后可以尝试进入 http://localhost:6333/dashboard 控制台看看是否成功。此时，服务内一个 Collection 都没有，实际上也不需要我们手动创建。不过这里给出创建一个官方参数 Collection 的方法：

PUT collections/HelloAgent
{
  "vectors": {
    "size": 384,
    "distance": "Cosine",
    "timeout": 30
  }
}

控制台会返回

{
  "result": true,
  "status": "ok",
  "time": 0.139126398
}

最后，在 .env 中配置：

# Qdrant
QDRANT_URL = http://localhost:6333
# Qdrant集合配置
QDRANT_COLLECTION = HelloAgentRag
QDRANT_VECTOR_SIZE=384
QDRANT_DISTANCE=cosine
QDRANT_TIMEOUT=30

docker 部署 qdrant，一般情况下不需要 key。

Neo4j

相比起 Qdrant 专门用来存储相似程度， Neo4j 的重心则放在了如何存储 知识之间的确切关系 ，与 Qdrant 知识库一同为 LLM 匹配更多有关知识的上下文。在 HelloAgent 中，被用于存储 语义记忆。

docker 部署：

docker pull neo4j
mkdir /mydata/neo4j

# 创建neo4j容器
docker run -it -d -p 7474:7474 -p 7687:7687 \
-v /mydata/neo4j \
-e NEO4J_AUTH=neo4j/password \
--name neo4j \
neo4j

控制台的端口是 7474，服务本身的端口是 7687。username 和 password 可以随意配置，这里设置成 neo4j & password。

完成后，进入 http://localhost:7474/browser/ 可以查看是否完成。

随后，配置 .env：

# 或使用本地Neo4j (需要Docker)
NEO4J_URI= bolt://localhost:7687
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=password

# Neo4j连接配置
NEO4J_DATABASE=neo4j
NEO4J_MAX_CONNECTION_LIFETIME=3600
NEO4J_MAX_CONNECTION_POOL_SIZE=50
NEO4J_CONNECTION_TIMEOUT=60

注意 uri 不要填 http，要填 bolt

SQLite

SQLite 是一种十分常见的轻量级数据库，官方将其作为文档数据库使用，用于存储有着格式化数据的存储与查询。

SQLite 不需要我们额外安装，python 中包含 sqlite3 这个库可以直接用，对本地的 .db 文件进行读写操作。

在 HelloAgent 中，被用于存储完整的记忆，包括 content 在内。

Embedding

Embedding 是专门的模型，将文本拆分成 token 再转化为向量。

因为嵌入模型在本地部署压力太大了，官方推荐使用云端 api ，比如阿里云百炼，我这里用了 text-embedding-v4 模型。在 .env 中配置：

# ==========================
# 嵌入（Embedding）配置示例 - 可从阿里云控制台获取：https://dashscope.aliyun.com/
# ==========================
# - 若为空，dashscope 默认 text-embedding-v3；local 默认 sentence-transformers/all-MiniLM-L6-v2
EMBED_MODEL_TYPE=dashscope
EMBED_MODEL_NAME=text-embedding-v4
EMBED_API_KEY= 自己填
EMBED_BASE_URL=https://ws-91vjyamuomyrjpla.cn-beijing.maas.aliyuncs.com/compatible-mode/v1

测试程序

我们使用官方的测试程序，但是 SimpleAgent、LLM、ToolRegistry 可以使用我们自己实现的版本：

from agent.simpleAgent import SimpleAgent
from core.llm import LLM
from tool.toolRegistry import ToolRegistry

from hello_agents.tools import MemoryTool, RAGTool


# 创建LLM实例
llm = LLM()

# 创建Agent
agent = SimpleAgent(
    name="智能助手",
    llm=llm,
    system_prompt="你是一个有记忆和知识检索能力的AI助手"
)

# 创建工具注册表
tool_registry = ToolRegistry()

# 添加记忆工具
memory_tool = MemoryTool(user_id="user123")
tool_registry.register_tool(memory_tool)

# 添加RAG工具
rag_tool = RAGTool(knowledge_base_path="./knowledge_base")
tool_registry.register_tool(rag_tool)

# 为Agent配置工具
agent.tool_registry = tool_registry

# 开始对话
response = agent.run("你好！请记住我叫张三，我是一名Python开发者")
print(response)

结果：

✅ 智能助手 初始化完成，工具调用: 禁用
[OK] SQLite 数据库表和索引创建完成
[OK] SQLite 文档存储初始化完成: ./memory_data\memory.db
INFO:hello_agents.memory.storage.qdrant_store:✅ 成功连接到Qdrant服务: http://localhost:6333
INFO:hello_agents.memory.storage.qdrant_store:✅ 使用现有Qdrant集合: helloAgent
INFO:hello_agents.memory.types.semantic:✅ 嵌入模型就绪，维度: 1024
INFO:hello_agents.memory.types.semantic:✅ Qdrant向量数据库初始化完成
INFO:hello_agents.memory.storage.neo4j_store:✅ 成功连接到Neo4j服务: bolt://localhost:7687
INFO:hello_agents.memory.storage.neo4j_store:✅ Neo4j索引创建完成
INFO:hello_agents.memory.types.semantic:✅ Neo4j图数据库初始化完成
INFO:hello_agents.memory.types.semantic:🏥 数据库健康状态: Qdrant=✅, Neo4j=✅
INFO:hello_agents.memory.types.semantic:✅ 加载中文spaCy模型: zh_core_web_sm
INFO:hello_agents.memory.types.semantic:✅ 加载英文spaCy模型: en_core_web_sm
INFO:hello_agents.memory.types.semantic:🎯 主要使用中文spaCy模型
INFO:hello_agents.memory.types.semantic:📚 可用语言模型: 中文, 英文
INFO:hello_agents.memory.types.semantic:增强语义记忆初始化完成（使用Qdrant+Neo4j专业数据库）
INFO:hello_agents.memory.manager:MemoryManager初始化完成，启用记忆类型: ['working', 'episodic', 'semantic']
✅ 工具 'memory' 已注册。
INFO:hello_agents.memory.storage.qdrant_store:✅ 成功连接到Qdrant服务: http://localhost:6333
INFO:hello_agents.memory.storage.qdrant_store:✅ 创建Qdrant集合: rag_knowledge_base
✅ RAG工具初始化成功: namespace=default, collection=rag_knowledge_base
✅ 工具 'rag' 已注册。
🤖 智能助手 正在处理: 你好！请记住我叫张三，我是一名Python开发者
🧠 正在调用 deepseek-v4-pro 模型...
✅ 大语言模型响应成功:

✅ 智能助手 响应完成
你好，张三！我已经记住了，你是一名 Python 开发者。有什么需要帮忙的吗？

我们现在不清楚到底干了什么，但是可以验证所有服务部署成功了。

memory

我们创建一个 memory 包，专用于处理记忆。这个包与 core、agent、tool 平行，其结构为：

HelloAgent/
   │
   ├── memory/                          # 记忆
   │   ├── memoryItem.py                # 记忆实体类
   │   ├── memoryConfig.py              # 记忆统一配置中心
   │   ├── memoryManager.py             # 记忆管理器
   │   ├── memory.py                    # Memory 基类
   │   └── types                        # 四种不同类型的记忆
   │        ├── workingMemory.py        # 工作记忆（TTL管理，纯内存）
   │        ├── episodicMemory.py       # 情景记忆（事件序列，SQLite+Qdrant）
   │        └── semanticMemory.py       # 语义记忆（知识图谱，Qdrant+Neo4j）
   │
   ├── tool/
   │   └── builtin                      # 四种不同类型的记忆
   │        └── memoryTool.py           # 对外暴露的记忆存取工具接口

这一部分中，我们的核心目标就是构建三种不同的记忆，并完成对不同类型的记忆的统一管理。

官方测试 MemoryTool

官方给出了测试程序，用来测试 MemoryTool：

from agent.simpleAgent import SimpleAgent
from core.llm import LLM
from tool.toolRegistry import ToolRegistry

from hello_agents.tools import MemoryTool


# 创建具有记忆能力的Agent
llm = LLM()
agent = SimpleAgent(name="记忆助手", llm=llm)

# 创建记忆工具
memory_tool = MemoryTool(user_id="user123")
tool_registry = ToolRegistry()
tool_registry.register_tool(memory_tool)
agent.tool_registry = tool_registry

# 体验记忆功能
print("=== 添加多个记忆 ===")

# 添加第一个记忆
result1 = memory_tool.run({"action": "add", "content": "用户张三是一名Python开发者，专注于机器学习和数据分析", "memory_type": "semantic", "importance": 0.8})
print(f"记忆1: {result1}")

# 添加第二个记忆
result2 = memory_tool.run({"action": "add", "content": "李四是前端工程师，擅长React和Vue.js开发", "memory_type": "semantic", "importance": 0.7})
print(f"记忆2: {result2}")

# 添加第三个记忆
result3 = memory_tool.run({"action": "add", "content": "王五是产品经理，负责用户体验设计和需求分析", "memory_type": "semantic", "importance": 0.6})
print(f"记忆3: {result3}")

print("\n=== 搜索特定记忆 ===")
# 搜索前端相关的记忆
print("🔍 搜索 '前端工程师':")
result = memory_tool.run({"action": "search", "query": "前端工程师", "limit": 3})
print(result)

print("\n=== 记忆摘要 ===")
result = memory_tool.run({"action": "summary"})
print(result)

测试结果：

python -m test.__testMemoryTool

✅ 记忆助手 初始化完成，工具调用: 禁用
[OK] SQLite 数据库表和索引创建完成
[OK] SQLite 文档存储初始化完成: ./memory_data\memory.db
INFO:hello_agents.memory.storage.qdrant_store:✅ 成功连接到Qdrant服务: http://localhost:6333
INFO:hello_agents.memory.storage.qdrant_store:✅ 使用现有Qdrant集合: helloAgent
INFO:hello_agents.memory.types.semantic:✅ 嵌入模型就绪，维度: 1024
INFO:hello_agents.memory.types.semantic:✅ Qdrant向量数据库初始化完成
INFO:hello_agents.memory.storage.neo4j_store:✅ 成功连接到Neo4j服务: bolt://localhost:7687
INFO:hello_agents.memory.storage.neo4j_store:✅ Neo4j索引创建完成
INFO:hello_agents.memory.types.semantic:✅ Neo4j图数据库初始化完成
INFO:hello_agents.memory.types.semantic:🏥 数据库健康状态: Qdrant=✅, Neo4j=✅
INFO:hello_agents.memory.types.semantic:✅ 加载中文spaCy模型: zh_core_web_sm
INFO:hello_agents.memory.types.semantic:✅ 加载英文spaCy模型: en_core_web_sm
INFO:hello_agents.memory.types.semantic:🎯 主要使用中文spaCy模型
INFO:hello_agents.memory.types.semantic:📚 可用语言模型: 中文, 英文
INFO:hello_agents.memory.types.semantic:增强语义记忆初始化完成（使用Qdrant+Neo4j专业数据库）
INFO:hello_agents.memory.manager:MemoryManager初始化完成，启用记忆类型: ['working', 'episodic', 'semantic']
✅ 工具 'memory' 已注册。
=== 添加多个记忆 ===
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] add_vectors start: n_vectors=1 n_meta=1 collection=helloAgent
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] upsert begin: points=1
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] upsert done
INFO:hello_agents.memory.storage.qdrant_store:✅ 成功添加 1 个向量到Qdrant
INFO:hello_agents.memory.types.semantic:✅ 添加语义记忆: 1个实体, 0个关系
记忆1: ✅ 记忆已添加 (ID: be8b1a9f...)
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] add_vectors start: n_vectors=1 n_meta=1 collection=helloAgent
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] upsert begin: points=1
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] upsert done
INFO:hello_agents.memory.storage.qdrant_store:✅ 成功添加 1 个向量到Qdrant
INFO:hello_agents.memory.types.semantic:✅ 添加语义记忆: 2个实体, 1个关系
记忆2: ✅ 记忆已添加 (ID: dd982d4f...)
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] add_vectors start: n_vectors=1 n_meta=1 collection=helloAgent
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] upsert begin: points=1
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] upsert done
INFO:hello_agents.memory.storage.qdrant_store:✅ 成功添加 1 个向量到Qdrant
INFO:hello_agents.memory.types.semantic:✅ 添加语义记忆: 0个实体, 0个关系
记忆3: ✅ 记忆已添加 (ID: 89173a66...)

=== 搜索特定记忆 ===
🔍 搜索 '前端工程师':
INFO:hello_agents.memory.types.semantic:✅ 检索到 1 条相关记忆
🔍 找到 1 条相关记忆:
1. [语义记忆] 李四是前端工程师，擅长React和Vue.js开发 (重要性: 0.70)

=== 记忆摘要 ===
ERROR:hello_agents.memory.storage.qdrant_store:❌ 获取集合信息失败: 'CollectionInfo' object has no attribute 'vectors_count'
INFO:hello_agents.memory.types.semantic:✅ 检索到 3 条相关记忆
📊 记忆系统摘要
总记忆数: 3
当前会话: session_20260723_132000
对话轮次: 0

📋 记忆类型分布:
  • 工作记忆: 0 条 (平均重要性: 0.00)
  • 情景记忆: 0 条 (平均重要性: 0.00)
  • 语义记忆: 3 条 (平均重要性: 0.70)

⭐ 重要记忆 (前3条):
  1. 用户张三是一名Python开发者，专注于机器学习和数据分析 (重要性: 0.80)
  2. 李四是前端工程师，擅长React和Vue.js开发 (重要性: 0.70)
  3. 王五是产品经理，负责用户体验设计和需求分析 (重要性: 0.60)

记忆实体类 MemoryItem

首先，我们要定义一个严密的 Memory 最小单位，且要继承数据类 BaseModel ：

from datetime import datetime
from pydantic import BaseModel
from typing import Any, Dict

class MemoryItem(BaseModel):
    """记忆项数据结构"""
    id: str
    content: str
    memory_type: str
    user_id: str
    timestamp: datetime
    importance: float = 0.5
    metadata: Dict[str, Any] = {}
    arbitrary_types_allowed: bool = True

• id：全局唯一标识符，采用 uuid；

• content：核心数据，是一串自然语言；

• userId：标识该记忆的用户；

• timestamp：时间戳；

• importance：重要程度，后续用于计算是否遗忘；

• metadata：附加数据，根据记忆类型的不同，会携带不同的 kv 字段；例如所有记忆都要自带 type，以表明自身的类型，例如 [type: “working”]。

• arbitraryTypesAllowed：允许 datetime 和 Dict 等非标准 JSON 类型。

记忆配置类 MemoryConfig

在定义了最小单位的记忆数据类后，我们再对记忆存储的参数进行统一配置，包括存在哪、怎么存、存什么：

from pydantic import BaseModel
from typing import List

class MemoryConfig(BaseModel):
    """记忆系统配置"""
    
    # 存储路径
    storage_path: str = "./memory_data"
    
    # 统计显示用的基础配置（仅用于展示）
    max_capacity: int = 100
    importance_threshold: float = 0.1
    decay_factor: float = 0.95

    # 工作记忆特定配置
    working_memory_capacity: int = 10
    working_memory_tokens: int = 2000
    working_memory_ttl_minutes: int = 120

    # 感知记忆特定配置
    perceptual_memory_modalities: List[str] = ["text", "image", "audio"]

perceptual_memory_modalities 表示我们的记忆系统支持哪些模态：目前硬编码为文本、图像、音频。

记忆行为抽象类 Memory

我们定义了记忆数据类，和记忆统一配置，接下来我们实现一个抽象类 Memory，对所有三种记忆实现统一规范。

初始化

from abc import ABC, abstractmethod
from typing import List, Dict, Any
from .memoryConfg import MemoryConfig   
from .memoryItem import MemoryItem

class Memory(ABC):
    """记忆基类
    定义所有记忆类型的通用接口和行为
    """
    def __init__(self, config: MemoryConfig, storage_backend=None):
        self.config = config
        self.storage = storage_backend
        self.memory_type = self.__class__.__name__.lower().replace("memory", "")

接受一个 Config 来配置这段记忆的参数；

而 memory_type 表示该记忆的类型，这里没有给输入传参，而是通过继承 Memory 的子类的类名来推断，具体而言：

WorkingMemory  → "working"
EpisodicMemory → "episodic"
SemanticMemory → "semantic"

这是为了方便未来的记忆类型拓展，当然，我承认这种方式很诡异，直接传入参数不就行了。

抽象方法

这里规定所有 Memory 必须实现以下方法：

@abstractmethod
def add(self, memory_item: MemoryItem) -> str:
    """添加记忆项

    Args:
        memory_item: 记忆项对象

    Returns:
        记忆ID
    """
    pass

@abstractmethod
def retrieve(self, query: str, limit: int = 5, **kwargs) -> List[MemoryItem]:
    """检索相关记忆

    Args:
        query: 查询内容
        limit: 返回数量限制
        **kwargs: 其他检索参数

    Returns:
        相关记忆列表
    """
    pass

@abstractmethod
def update(self, memory_id: str, content: str = None,
           importance: float = None, metadata: Dict[str, Any] = None) -> bool:
    """更新记忆

    Args:
        memory_id: 记忆ID
        content: 新内容
        importance: 新重要性
        metadata: 新元数据

    Returns:
        是否更新成功
    """
    pass

@abstractmethod
def remove(self, memory_id: str) -> bool:
    """删除记忆

    Args:
        memory_id: 记忆ID

    Returns:
        是否删除成功
    """
    pass

@abstractmethod
def has_memory(self, memory_id: str) -> bool:
    """检查记忆是否存在

    Args:
        memory_id: 记忆ID

    Returns:
        是否存在
    """
    pass

@abstractmethod
def clear(self):
    """清空所有记忆"""
    pass

@abstractmethod
def get_stats(self) -> Dict[str, Any]:
    """获取记忆统计信息

    Returns:
        统计信息字典
    """
    pass

基本方法就是增删改查。但值得注意的是，Memory 抽象类中并没有强制要求实现类实现 forget 方法，因为可能存在过于重要以至于必须永远记住的记忆存在。

同时，我们再明确一下这里面没有实现的设计：有关 MemoryItem 的存储。记忆存储的最小单元就是 MemoryItem，每个实现了 Memory 的实现类，内部都会维护一个用于存储 MemoryItem 的数据结构。

3+1 种记忆

有了 Memory 基类后，我们就可以通过实现，构建四种 Agent 工作必不可少的记忆。

• 工作记忆 WorkingMemory：主要用于原封不动地存储当前对话的上下文信息，生命周期较短，其容量也不需要太大，更重要的是不需要持久化，存在内存中就行。比方说 时间：21:04:39；用户："使用 openAI 规范"，Agent："好的，我会严格遵守 openAI 规范。"；当前会话轮数：12、时间：21:07:39；用户："/grill-with-docs 我要实现一个 Memory 基类，帮我拟定计划。"，Agent："好的，我先调用。"；当前会话轮数：13 等，都适合存入工作记忆。

• 情景记忆 EpisodicMemory：适合存用户与 Agent 在特定场合下的重要交互，需要长期存储以支持 /resume 或者 --continue，例如 用户在21:04:39告诉我需要使用 openAI 规范等。相当于抛弃不重要的，保留重要的，然后进行上下文压缩，这就是情景记忆。

• 语义记忆 SemanticMemory：用来存储必须遵守的抽象规范，这些记忆一般是由 LLM 自行总结并判断存入的。比如 规范：openAI 或者 用户更加偏好于......，当然这些也需要长期存储。

有了以上三种记忆后，我们的 Agent 的记忆功能就基本成型了：用户使用中，工作记忆保留完整上下文，尽可能保证当前会话信息不失误；同时，随时将重要信息压缩本地化，实现长期记忆和过滤无关信息；最后将必须遵守的抽象规范单独存放，在读取时尽可能保证完全遵守。

此外，官方也提出了一个感知记忆 PerceptualMemory，主要用于处理图片、视频、音频等多模态内容，我们这里就不做实现了，有上面三种记忆支持就可以初步实现用户与 Agent 的完整交互了。

工作记忆 WorkingMemory

前面提到，工作记忆无需存放进 Qdrant 或者 Neo4j，而是存在内存中，也就是说在会话期间直接以一个 WorkingMemory 实例存在就足以满足了。有关 Rag 的功能还没有实现，其他两种记忆依赖于 Rag，所以我们先尝试实现无需 Rag 的 WorkingMemory，存放到 memory/types。

记忆储存机制

继承 Memory，输入 MemoryConfig 作为配置中心。

class WorkingMemory(Memory):
    def __init__(self, config: MemoryConfig, storage_backend=None):
        super().__init__(config, storage_backend)
        
        # 工作记忆特定配置
        self.max_capacity = self.config.working_memory_capacity
        self.max_tokens = self.config.working_memory_tokens
        # 纯内存TTL（分钟），可通过在 MemoryConfig 上挂载 working_memory_ttl_minutes 覆盖
        self.max_age_minutes = getattr(self.config, 'working_memory_ttl_minutes', 120)
        self.current_tokens = 0
        self.session_start = datetime.now()
        
        # 内存存储（工作记忆不需要持久化）
        self.memories: List[MemoryItem] = []

可以看到，我们采用的是 List 对 MemoryItem 进行存储。

不过在官方提供的 hello-agents 包内，官方使用 List 和 heapq 来存储 MemoryItem，官方的意图很明显：分别用作存储和查询，但问题是 heapq 除了在添加记忆和淘汰记忆时，会对 heapq 进行入堆和重建，在关键的查询上，反而没有使用堆，所以我们可以直接看成不存在，只使用 List 其实更便于理解，在后续的代码构成中，我们也会直接忽略 heapq。

新增

添加记忆的方法如下：

def add(self, memory_item: MemoryItem) -> str:
    """添加工作记忆"""
    # 过期清理
    self._expire_old_memories()
    # 计算优先级（重要性 + 时间衰减）
    priority = self._calculate_priority(memory_item)
    
    # 添加到列表中
    self.memories.append(memory_item)
    
    # 更新token计数
    self.current_tokens += len(memory_item.content.split())
    
    # 检查容量限制
    self._enforce_capacity_limits()
    
    return memory_item.id

在真正添加到 List 前，我们还有两件事要做：过期清理 _expire_old_memories 和计算优先级 _calculate_priority。

先来看 _expire_old_memories，大体基本上就是对 List 进行遍历，删除超时的数据。代码：

def _expire_old_memories(self):
    """按TTL清理过期记忆，并同步更新堆与token计数"""
    if not self.memories:
        return
    cutoff_time = datetime.now() - timedelta(minutes=self.max_age_minutes)
    # 过滤保留的记忆
    kept: List[MemoryItem] = []
    removed_token_sum = 0
    for m in self.memories:
        if m.timestamp >= cutoff_time:
            kept.append(m)
        else:
            removed_token_sum += len(m.content.split())
    if len(kept) == len(self.memories):
        return
    # 覆盖列表与token
    self.memories = kept
    self.current_tokens = max(0, self.current_tokens - removed_token_sum)

看 _calculate_priority 函数：

def _calculate_priority(self, memory: MemoryItem) -> float:
    """计算记忆优先级"""
    # 基础优先级 = 重要性
    priority = memory.importance
    
    # 时间衰减
    time_decay = self._calculate_time_decay(memory.timestamp)
    priority *= time_decay
    
    return priority

优先级等于重要程度乘时间衰减 _calculate_time_decay，记忆存在时间越长，衰减越大，这意味着越晚加入 List 的 MemoryItem，优先级也越小，越容易被清除：

def _calculate_time_decay(self, timestamp: datetime) -> float:
    """计算时间衰减因子"""
    time_diff = datetime.now() - timestamp
    hours_passed = time_diff.total_seconds() / 3600
    
    # 指数衰减（工作记忆衰减更快）
    decay_factor = self.config.decay_factor ** (hours_passed / 6)  # 每6小时衰减
    return max(0.1, decay_factor)  # 最小保持10%的权重

而在添加到 List 后，还要干两件事：更新token计数和检查记忆容量 _enforce_capacity_limits，这个函数会一直清除优先级最低的 MemoryItem，直到在容量与 token 数量上均不超过最大限制。

def _enforce_capacity_limits(self):
    """强制执行容量限制"""
    # 检查记忆数量限制
    while len(self.memories) > self.max_capacity:
        self._remove_lowest_priority_memory()
    
    # 检查token限制
    while self.current_tokens > self.max_tokens:
        self._remove_lowest_priority_memory()

检索

检索是最复杂的功能，主要分为四个步骤：1.过滤用户 id；2.TF-IDF 向量计算；3.关键词检索；4.最终合成。

TF-IDF 算法就不做解释了，因为我也不懂。

def retrieve(self, query: str, limit: int = 5, user_id: str = None, **kwargs) -> List[MemoryItem]:
    """检索工作记忆 - 混合语义向量检索和关键词匹配"""
    # 过期清理
    self._expire_old_memories()
    if not self.memories:
        return []

    # 按用户ID过滤（如果提供）
    active_memories = self.memories
    filtered_memories = active_memories
    if user_id:
        filtered_memories = [m for m in active_memories if m.user_id == user_id]

    if not filtered_memories:
        return []

    # 尝试语义向量检索（如果有嵌入模型）
    vector_scores = {}
    try:
        # 简单的语义相似度计算（使用TF-IDF或其他轻量级方法）
        from sklearn.feature_extraction.text import TfidfVectorizer
        from sklearn.metrics.pairwise import cosine_similarity
        import numpy as np
        
        # 准备文档
        documents = [query] + [m.content for m in filtered_memories]
        
        # TF-IDF向量化
        vectorizer = TfidfVectorizer(stop_words=None, lowercase=True)
        tfidf_matrix = vectorizer.fit_transform(documents)
        
        # 计算相似度
        query_vector = tfidf_matrix[0:1]
        doc_vectors = tfidf_matrix[1:]
        similarities = cosine_similarity(query_vector, doc_vectors).flatten()
        
        # 存储向量分数
        for i, memory in enumerate(filtered_memories):
            vector_scores[memory.id] = similarities[i]
            
    except Exception as e:
        # 如果向量检索失败，回退到关键词匹配
        vector_scores = {}

    # 计算最终分数
    query_lower = query.lower()
    scored_memories = []
    
    for memory in filtered_memories:
        content_lower = memory.content.lower()
        
        # 获取向量分数（如果有）
        vector_score = vector_scores.get(memory.id, 0.0)
        
        # 关键词匹配分数
        keyword_score = 0.0
        if query_lower in content_lower:
            keyword_score = len(query_lower) / len(content_lower)
        else:
            # 分词匹配
            query_words = set(query_lower.split())
            content_words = set(content_lower.split())
            intersection = query_words.intersection(content_words)
            if intersection:
                keyword_score = len(intersection) / len(query_words.union(content_words)) * 0.8

        # 混合分数：向量检索 + 关键词匹配
        if vector_score > 0:
            base_relevance = vector_score * 0.7 + keyword_score * 0.3
        else:
            base_relevance = keyword_score
        
        # 时间衰减
        time_decay = self._calculate_time_decay(memory.timestamp)
        base_relevance *= time_decay
        
        # 重要性权重
        importance_weight = 0.8 + (memory.importance * 0.4)
        final_score = base_relevance * importance_weight
        
        if final_score > 0:
            scored_memories.append((final_score, memory))

    # 按分数排序并返回
    scored_memories.sort(key=lambda x: x[0], reverse=True)
    return [memory for _, memory in scored_memories[:limit]]

遗忘

Memory 抽象类中并没有要求实现 forget 函数，这是出于某些记忆不需要遗忘机制而做出的决定。但对于工作记忆来说，遗忘是必须的，所以我们有必要对其进行实现：

def forget(self, strategy: str = "importance_based", threshold: float = 0.1, max_age_days: int = 1) -> int:
    """工作记忆遗忘机制"""
    forgotten_count = 0
    current_time = datetime.now()
    
    to_remove = []
    
    # 始终先执行TTL过期（分钟级）
    cutoff_ttl = current_time - timedelta(minutes=self.max_age_minutes)
    for memory in self.memories:
        if memory.timestamp < cutoff_ttl:
            to_remove.append(memory.id)
    
    if strategy == "importance_based":
        # 删除低重要性记忆
        for memory in self.memories:
            if memory.importance < threshold:
                to_remove.append(memory.id)
    
    elif strategy == "time_based":
        # 删除过期记忆（工作记忆通常以小时计算）
        cutoff_time = current_time - timedelta(hours=max_age_days * 24)
        for memory in self.memories:
            if memory.timestamp < cutoff_time:
                to_remove.append(memory.id)
    
    elif strategy == "capacity_based":
        # 删除超出容量的记忆
        if len(self.memories) > self.max_capacity:
            # 按优先级排序，删除最低的
            sorted_memories = sorted(
                self.memories,
                key=lambda m: self._calculate_priority(m)
            )
            excess_count = len(self.memories) - self.max_capacity
            for memory in sorted_memories[:excess_count]:
                to_remove.append(memory.id)
    
    # 执行删除
    for memory_id in to_remove:
        if self.remove(memory_id):
            forgotten_count += 1
    
    return forgotten_count

三种遗忘策略：importance_based, time_based, capacity_based。

MemoryManager

我们已经定义了一个具体的 Memory 实现类，那这个记忆是如何生效的？先让我们回想一下，官方给出的包演示是怎么做的：

# 创建工具注册表
tool_registry = ToolRegistry()

# 添加记忆工具
memory_tool = MemoryTool(user_id="user123")
tool_registry.register_tool(memory_tool)

# 添加RAG工具
rag_tool = RAGTool(knowledge_base_path="./knowledge_base")
tool_registry.register_tool(rag_tool)

# 为Agent配置工具
agent.tool_registry = tool_registry

Memory 被作为工具被 ToolRegistry 统一注册发现，然后被 LLM 主动调用。

不过有个问题是，我们现在只实现了底层的 Memory 类及其实现类，如果我们要让工具意图类调用底层记忆，我们最好再加一层统一管理类，这就是 MemoryManager 类的由来。官方对 MemoryTool 和 MemoryManager 的定位是：

这种分层设计体现了软件工程中的关注点分离原则，MemoryTool专注于用户接口和参数处理，而MemoryManager则负责核心的记忆管理逻辑。

初始化

def __init__(
    self,
    config: Optional[MemoryConfig] = None,
    user_id: str = "default_user",
    enable_working: bool = True,
    enable_episodic: bool = False,
    enable_semantic: bool = False
):
    self.config = config or MemoryConfig()
    self.user_id = user_id

    # 初始化各类型记忆
    self.memory_types = {}
    
    if enable_working:
        self.memory_types['working'] = WorkingMemory(self.config)
    
    logger.info(f"MemoryManager初始化完成，启用记忆类型: {list(self.memory_types.keys())}") 

除了 config，其内部还有一个 kv 对成员 memory_types，用来存四种记忆的对象，对其进行管理。我们先熟悉一下其结构。

memory_types 在大多数情况下可以看作一个有着三个元素的 kv map，而每个 v 内部则有着一个 List 用来存 MemoryItem，整个 memory_types 如下：

memory_types：

    "working"   -> WorkingMemory()  -> List<MemoryItem> ->  [memory1, memory2 ......]

    "episodic"  -> EpisodicMemory() -> List<MemoryItem> ->  [memory1, memory2 ......]

    "semantic"  -> SemanticMemory() -> List<MemoryItem> ->  [memory1, memory2 ......]

记忆分类

对于 MemoryTool 来说，它并不需要对记忆的类型进行区分，我们规定让 LLM 在 metadata 中存记忆的类型，便于 MemoryManager 直接存取：

def _classify_memory_type(self, content: str, metadata: Optional[Dict[str, Any]]) -> str:
    """自动分类记忆类型"""
    if metadata and metadata.get("type"):
        return metadata["type"]
    
    # 简单的分类逻辑，可以扩展为更复杂的分类器
    if self._is_episodic_content(content):
        return "episodic"
    elif self._is_semantic_content(content):
        return "semantic"
    else:
        return "working"

同时，官方又给出了另外两个函数，根据 content 关键词来判断是否为情景记忆和语义记忆：

def _is_episodic_content(self, content: str) -> bool:
    """判断是否为情景记忆内容"""
    episodic_keywords = ["昨天", "今天", "明天", "上次", "记得", "发生", "经历"]
    return any(keyword in content for keyword in episodic_keywords)

def _is_semantic_content(self, content: str) -> bool:
    """判断是否为语义记忆内容"""
    semantic_keywords = ["定义", "概念", "规则", "知识", "原理", "方法"]
    return any(keyword in content for keyword in semantic_keywords)

如果 content 中包含任意关键词，则可以直接判定记忆类型，主要是在 LLM 没有响应记忆类型的时候兜底，平时几乎不参与。

CRUD

增：新增一个记忆，就是将其加入到成员 memory_types 对应 type 的 Memory 类对象，这里调用了 Memory 类对象必须实现的 add 函数。

def add_memory(
    self,
    content: str,
    memory_type: str = "working",
    importance: Optional[float] = None,
    metadata: Optional[Dict[str, Any]] = None,
    auto_classify: bool = True
) -> str:
    """添加记忆
    
    Args:
        content: 记忆内容
        memory_type: 记忆类型
        importance: 重要性分数 (0-1)
        metadata: 元数据
        auto_classify: 是否自动分类到合适的记忆类型
        
    Returns:
        记忆ID
    """
    # 自动分类记忆类型
    if auto_classify:
        memory_type = self._classify_memory_type(content, metadata)
    
    # 计算重要性
    if importance is None:
        importance = self._calculate_importance(content, metadata)
    
    # 创建记忆项
    memory_item = MemoryItem(
        id=str(uuid.uuid4()),
        content=content,
        memory_type=memory_type,
        user_id=self.user_id,
        timestamp=datetime.now(),
        importance=importance,
        metadata=metadata or {}
    )
    
    # 添加到对应的记忆类型
    if memory_type in self.memory_types:
        memory_id = self.memory_types[memory_type].add(memory_item)
        logger.debug(f"添加记忆到 {memory_type}: {memory_id}")
        return memory_id
    else:
        raise ValueError(f"不支持的记忆类型: {memory_type}")

查：因为我们是按照关键词查找内容的，事先不知道预查找记忆的类型，所以这里需要遍历 memory_types 中每个 key-value 中的 value，调用其内部实现的 retrieve 函数，加入到最终返回。查找完后，根据重要度排序。

def retrieve_memories(
    self,
    query: str,
    memory_types: Optional[List[str]] = None,
    limit: int = 10,
    min_importance: float = 0.0,
    time_range: Optional[tuple] = None
) -> List[MemoryItem]:
    """检索记忆
    
    Args:
        query: 查询内容
        memory_types: 要检索的记忆类型列表
        limit: 返回数量限制
        min_importance: 最小重要性阈值
        time_range: 时间范围 (start_time, end_time)
        
    Returns:
        检索到的记忆列表
    """
    if memory_types is None:
        memory_types = list(self.memory_types.keys())
    
    # 从各个记忆类型中检索
    all_results = []
    per_type_limit = max(1, limit // len(memory_types))

    for memory_type in memory_types:
        if memory_type in self.memory_types:
            memory_instance = self.memory_types[memory_type]
            try:
                # 使用各个记忆类型自己的检索方法
                type_results = memory_instance.retrieve(
                    query=query,
                    limit=per_type_limit,
                    min_importance=min_importance,
                    user_id=self.user_id
                )
                all_results.extend(type_results)
            except Exception as e:
                logger.warning(f"检索 {memory_type} 记忆时出错: {e}")
                continue

    # 按重要性和相关性排序
    all_results.sort(key=lambda x: x.importance, reverse=True)
    return all_results[:limit]

改：指定了 memory_id，这个和 memoryItem 中的成员变量 id 一致。

def update_memory(
    self,
    memory_id: str,
    content: Optional[str] = None,
    importance: Optional[float] = None,
    metadata: Optional[Dict[str, Any]] = None
) -> bool:
    """更新记忆
    
    Args:
        memory_id: 记忆ID
        content: 新内容
        importance: 新重要性
        metadata: 新元数据
        
    Returns:
        是否更新成功
    """
    # 查找记忆所在的类型
    for memory_type, memory_instance in self.memory_types.items():
        if memory_instance.has_memory(memory_id):
            return memory_instance.update(memory_id, content, importance, metadata)
    
    logger.warning(f"未找到记忆: {memory_id}")
    return False

删：遍历 memory_types 里的 key-value 中的 value，将其 List 中对应 id 的 memoryItem 移除就行。

def remove_memory(self, memory_id: str) -> bool:
    """删除记忆
    
    Args:
        memory_id: 记忆ID
        
    Returns:
        是否删除成功
    """
    for memory_type, memory_instance in self.memory_types.items():
        if memory_instance.has_memory(memory_id):
            return memory_instance.remove(memory_id)
    
    logger.warning(f"未找到记忆: {memory_id}")
    return False

遗忘机制

def forget_memories(
    self,
    strategy: str = "importance_based",
    threshold: float = 0.1,
    max_age_days: int = 30
) -> int:
    """记忆遗忘机制
    
    Args:
        strategy: 遗忘策略 ("importance_based", "time_based", "capacity_based")
        threshold: 遗忘阈值
        max_age_days: 最大保存天数
        
    Returns:
        遗忘的记忆数量
    """
    total_forgotten = 0
    
    for memory_type, memory_instance in self.memory_types.items():
        if hasattr(memory_instance, 'forget'):
            forgotten = memory_instance.forget(strategy, threshold, max_age_days)
            total_forgotten += forgotten

    logger.info(f"记忆遗忘完成: {total_forgotten} 条记忆")
    return total_forgotten

官方包内也给出了一些其他的功能，遗忘功能是其中之一，就是简单地调用 memory_types 中的 Memory 实现类对象的 froget 函数就行。

记忆整合

def consolidate_memories(
    self,
    from_type: str = "working",
    to_type: str = "episodic",
    importance_threshold: float = 0.7
) -> int:
    """记忆整合 - 将重要的短期记忆转换为长期记忆
    Args:
        from_type: 源记忆类型
        to_type: 目标记忆类型
        importance_threshold: 重要性阈值
    Returns:
        整合的记忆数量
    """
    if from_type not in self.memory_types or to_type not in self.memory_types:
        logger.warning(f"记忆类型不存在: {from_type} -> {to_type}")
        return 0

    # 获取高重要性的源记忆
    source_memory = self.memory_types[from_type]
    target_memory = self.memory_types[to_type]

    # 获取需要整合的记忆
    all_memories = source_memory.get_all()
    candidates = [
        m for m in all_memories
        if m.importance >= importance_threshold
    ]

    consolidated_count = 0
    for memory in candidates:
        # 移动到目标记忆类型
        if source_memory.remove(memory.id):
            memory.memory_type = to_type
            memory.importance *= 1.1  # 提升重要性
            target_memory.add(memory)
            consolidated_count += 1

    logger.info(f"记忆整合完成: {consolidated_count} 条记忆从 {from_type} 转移到 {to_type}")
    return consolidated_count

记忆整合，实际上就是记忆类型转换，默认是将工作记忆转化为情景记忆。

MemoryTool

我们已经构建了一个基本能用的 MemoryManager，接下来就是实现 MemoryTool。官方的设计理念是：

MemoryTool作为记忆系统的统一接口，其设计遵循了”统一入口，分发处理”的架构模式。

我们在 /tool/builtin 下面构建 memoryTool.py

初始化

class MemoryTool(Tool):
    """记忆工具 —— 让 LLM 能够存取和管理记忆

    作为记忆系统的统一接口，遵循"统一入口，分发处理"的架构模式。
    MemoryTool 专注于参数解析和结果格式化，核心逻辑委托给 MemoryManager。
    """

    def __init__(
        self,
        user_id: str = "default_user",
        config: Optional[MemoryConfig] = None
    ):
        super().__init__(
            name="memory",
            description=(
                "记忆管理工具，用于存储、检索、更新和删除记忆。"
                "支持以下操作：\n"
                "- add: 添加一条新记忆。需要 content（记忆内容），可选 memory_type "
                "（working/episodic/semantic，默认自动推断）、importance（0-1，默认自动计算）\n"
                "- search: 按关键词检索记忆。需要 query（查询内容），可选 limit（返回数量，默认5）\n"
                "- update: 更新已有记忆。需要 memory_id，可选 content/importance\n"
                "- remove: 删除指定记忆。需要 memory_id\n"
                "- summary: 查看记忆系统的统计摘要（总数、各类型容量等）\n"
                "- forget: 触发记忆遗忘机制。可选 strategy（importance_based/time_based/capacity_based）、"
                "threshold（遗忘阈值，默认0.1）\n"
                "- consolidate: 将重要的短期记忆整合为长期记忆。"
                "可选 from_type（默认working）、to_type（默认episodic）、importance_threshold（默认0.7）\n\n"
                "记忆类型说明：\n"
                "- working: 工作记忆（当前会话上下文，容量有限，过期自动清理）\n"
                "- episodic: 情景记忆（重要交互记录，长期保留）\n"
                "- semantic: 语义记忆（抽象规则和用户偏好，永久保留）\n\n"
                "使用建议：\n"
                "- 用户说'记住XXX'时，根据内容类型选择合适的 memory_type\n"
                "- 用户偏好/规范类信息 → semantic，重要性 0.8+\n"
                "- 临时上下文/当前话题 → working，重要性 0.5\n"
                "- 重要事件/决定 → episodic，重要性 0.7+\n"
                "- 在回答用户问题前，先用 search 检索相关记忆"
            )
        )
        self.user_id = user_id
        self._manager = MemoryManager(
            config=config,
            user_id=user_id,
            enable_working=True,
            enable_episodic=False,   # 暂未实现
            enable_semantic=False    # 暂未实现
        )

显然，MemoryTool 需要继承 Tool，内部也定义了 description 以让 LLM 调用，且里面提供了详细的 action 字段的作用。

此外，其包含了一个 MemoryManager 成员，用来充当调用 Memory 实现类的中间层。

统一入口 run 函数

我们希望外部只需要通过一个函数就可以调用所有记忆相关的功能，如同官方做法一样，我们定义一个网关函数 run，通过输入 Dict 字典的参数来路由到不同的函数。

def run(self, parameters: Dict[str, Any]) -> str:
    """统一入口：根据 action 分发给具体处理方法

    Args:
        parameters: 包含 action 及其他所需参数的字典

    Returns:
        格式化后的结果字符串
    """
    action = parameters.get("action", "search")

    dispatch = {
        "add":         self._handle_add,
        "search":      self._handle_search,
        "update":      self._handle_update,
        "remove":      self._handle_remove,
        "summary":     self._handle_summary,
        "forget":      self._handle_forget,
        "consolidate": self._handle_consolidate,
    }

    handler = dispatch.get(action)
    if handler is None:
        return (
            f"❌ 不支持的操作: '{action}'。"
            f"可用操作: {', '.join(dispatch.keys())}"
        )

    try:
        return handler(parameters)
    except ValueError as e:
        return f"❌ 参数错误: {e}"
    except Exception as e:
        return f"❌ 操作失败: {e}"

此外，run 函数返回字符串类型。

测试

我们只测试一下 Tool 的基本功能正不正常，所以我们把所有记忆类型设置为 working：

from tool.builtin.memoryTool import MemoryTool

# 创建记忆工具
memory_tool = MemoryTool(user_id="user123")
 
# 体验记忆功能
print("=== 添加多个记忆 ===")

# 添加第一个记忆
result1 = memory_tool.run({"action": "add", "content": "用户张三是一名Python开发者，专注于机器学习和数据分析", "memory_type": "working", "importance": 0.8})
print(f"记忆1: {result1}")

# 添加第二个记忆
result2 = memory_tool.run({"action": "add", "content": "李四是前端工程师，擅长React和Vue.js开发", "memory_type": "working", "importance": 0.7})
print(f"记忆2: {result2}")

# 添加第三个记忆
result3 = memory_tool.run({"action": "add", "content": "王五是产品经理，负责用户体验设计和需求分析", "memory_type": "working", "importance": 0.6})
print(f"记忆3: {result3}")

print("\n=== 搜索特定记忆 ===")
# 搜索前端相关的记忆
print("🔍 搜索 '前端工程师':")
result = memory_tool.run({"action": "search", "query": "前端工程师", "limit": 3})
print(result)

print("\n=== 记忆摘要 ===")
result = memory_tool.run({"action": "summary"})
print(result)

结果：

> python -m test.__testMemoryTool

✅ 记忆助手 初始化完成，工具调用: 禁用
INFO:memory.memoryManager:MemoryManager初始化完成，启用记忆类型: ['working']
✅ 工具 'memory' 已注册。
=== 添加多个记忆 ===
记忆1: ✅ 记忆已添加
   ID: 5c0ba6bf-bf42-47a9-83cc-32880e7297fc
   类型: working
   内容: 用户张三是一名Python开发者，专注于机器学习和数据分析
记忆2: ✅ 记忆已添加
   ID: e8c899f9-fa4c-4511-a458-166f470658e3
   类型: working
   内容: 李四是前端工程师，擅长React和Vue.js开发
记忆3: ✅ 记忆已添加
   ID: 3ebdb60e-c9df-46f7-9b06-b50d80910080
   类型: working
   内容: 王五是产品经理，负责用户体验设计和需求分析

=== 搜索特定记忆 ===
🔍 搜索 '前端工程师':
🔍 搜索 '前端工程师' 的结果（共 1 条）:
  [1] [working] 李四是前端工程师，擅长React和Vue.js开发  (importance=0.70, id=e8c899f9...)

=== 记忆摘要 ===
📊 记忆系统摘要:
   用户: user123
   总记忆数: 3
   启用的记忆类型: working
   配置: 最大容量=100, 重要性阈值=0.1, 衰减因子=0.95
   [working] 活跃=3, 容量使用率=30.0%

记忆持久化

数据库操作

我们直接大大方方使用官方包提供的四个包，新建一个 storage 包放到下面：

HelloAgent/
   │
   ├── storage/                         # 记忆
   │   ├── __init__.py                  # 包含 load_env() 函数
   │   ├── document_store.py            # 文档数据库
   │   ├── embedding.py                 # 向量模型
   │   ├── neo4j_store.py               # neo4j
   │   └── qdrant_store.py              # qdrant

不过 qdrant_store.py 里有点问题，版本不兼容，会出现错误 'CollectionInfo' object has no attribute 'vectors_count'。解决就是把里面的 get_collection_info 改成：

"""
获取集合信息

Returns:
    Dict: 集合信息
"""
try:
    collection_info = self.client.get_collection(self.collection_name)

    count=0
    if hasattr(collection_info, "points_count"):
        count=collection_info.points_count
    elif hasattr(collection_info, "vectors_count"):
        count=collection_info.vectors_count
    
    info = {
        "name": self.collection_name,
        "points_count": collection_info.points_count,
        "segments_count": collection_info.segments_count,
        "indexed_vectors_count": collection_info.indexed_vectors_count,
        "config": {
            "vector_size": self.vector_size,
            "distance": self.distance.value,
        }
    }
    
    return info
    
except Exception as e:
    logger.error(f"❌ 获取集合信息失败: {e}")
    return {}

这样就不会报错了。

情景记忆 EpisodicMemory

官方对情景记忆的定义是：

情景记忆负责存储具体的事件和经历，它的设计重点在于保持事件的完整性和时间序列关系。

采用了两种数据库：

采用了SQLite+Qdrant的混合存储方案，SQLite负责结构化数据的存储和复杂查询，Qdrant负责高效的向量检索。

所以我们创建一个 EpisodeMemory。

情景实体类 Episodic

class Episode:
    """情景记忆中的单个情景"""
    
    def __init__(
        self,
        episode_id: str,
        user_id: str,
        session_id: str,
        timestamp: datetime,
        content: str,
        context: Dict[str, Any],
        outcome: Optional[str] = None,
        importance: float = 0.5
    ):
        self.episode_id = episode_id
        self.user_id = user_id
        self.session_id = session_id
        self.timestamp = timestamp
        self.content = content
        self.context = context
        self.outcome = outcome
        self.importance = importance

初始化

class EpisodicMemory(Memory):
    """情景记忆实现
    
    特点：
    - 存储具体的交互事件
    - 包含丰富的上下文信息
    - 按时间序列组织
    - 支持模式识别和回溯
    """
    
    def __init__(self, config: MemoryConfig, storage_backend=None):
        super().__init__(config, storage_backend)
        
        # 本地缓存（内存）
        self.episodes: List[Episode] = []
        self.sessions: Dict[str, List[str]] = {}  # session_id -> episode_ids
        
        # 模式识别缓存
        self.patterns_cache = {}
        self.last_pattern_analysis = None

        # 权威文档存储（SQLite）
        db_dir = self.config.storage_path if hasattr(self.config, 'storage_path') else "./memory_data"
        os.makedirs(db_dir, exist_ok=True)
        db_path = os.path.join(db_dir, "memory.db")
        self.doc_store = SQLiteDocumentStore(db_path=db_path)

        # 统一嵌入模型（多语言，默认384维）
        self.embedder = get_text_embedder()

        # 向量存储（Qdrant - 使用连接管理器避免重复连接）
        qdrant_url = os.getenv("QDRANT_URL")
        qdrant_api_key = os.getenv("QDRANT_API_KEY")
        self.vector_store = QdrantConnectionManager.get_instance(
            url=qdrant_url,
            api_key=qdrant_api_key,
            collection_name=os.getenv("QDRANT_COLLECTION", "hello_agents_vectors"),
            vector_size=get_dimension(getattr(self.embedder, 'dimension', 384)),
            distance=os.getenv("QDRANT_DISTANCE", "cosine")
        )

重要的成员变量有以下：

• episodes：存储当前 session 下的情景知识；

• sessions：一个字典，将不同的 session 下的情景知识按照 session_id 分开来，便于用户 /resume；

• doc_store：sqlite 连接器，可以通过这个对数据库进行读写。比如 EpisodicMemory 几乎所有的数据持久化函数都是通过外包给 doc_store 实现的；

• vector_store：qdrant 连接器，可对 qdrant 进行读写。

• embedder：嵌入模型，将文字转成向量，用于辅助 vector_store 的 qdrant 读写。

不过官方在教程文档中提到的 storage_backend 在官方库代码中已经没用了，我们默认采用 qdrant 而非自定义。

add

def add(self, memory_item: MemoryItem) -> str:
    """添加情景记忆"""
    # 从元数据中提取情景信息
    session_id = memory_item.metadata.get("session_id", "default_session")
    context = memory_item.metadata.get("context", {})
    outcome = memory_item.metadata.get("outcome")
    participants = memory_item.metadata.get("participants", [])
    tags = memory_item.metadata.get("tags", [])
    
    # 创建情景（内存缓存）
    episode = Episode(
        episode_id=memory_item.id,
        user_id=memory_item.user_id,
        session_id=session_id,
        timestamp=memory_item.timestamp,
        content=memory_item.content,
        context=context,
        outcome=outcome,
        importance=memory_item.importance
    )
    self.episodes.append(episode)
    if session_id not in self.sessions:
        self.sessions[session_id] = []
    self.sessions[session_id].append(episode.episode_id)

    # 1) 权威存储（SQLite）
    ts_int = int(memory_item.timestamp.timestamp())
    self.doc_store.add_memory(
        memory_id=memory_item.id,
        user_id=memory_item.user_id,
        content=memory_item.content,
        memory_type="episodic",
        timestamp=ts_int,
        importance=memory_item.importance,
        properties={
            "session_id": session_id,
            "context": context,
            "outcome": outcome,
            "participants": participants,
            "tags": tags
        }
    )

    # 2) 向量索引（Qdrant）
    try:
        embedding = self.embedder.encode(memory_item.content)
        if hasattr(embedding, "tolist"):
            embedding = embedding.tolist()
        self.vector_store.add_vectors(
            vectors=[embedding],
            metadata=[{
                "memory_id": memory_item.id,
                "user_id": memory_item.user_id,
                "memory_type": "episodic",
                "importance": memory_item.importance,
                "session_id": session_id,
                "content": memory_item.content
            }],
            ids=[memory_item.id]
        )
    except Exception:
        # 向量入库失败不影响权威存储
        pass

    return memory_item.id

分为两大步骤：1.sqlite 存储完整数据（官方称之为“权威存储”）；2.qdrant 存储向量索引数据。

retrieve

def retrieve(self, query: str, limit: int = 5, **kwargs) -> List[MemoryItem]:
    """检索情景记忆（结构化过滤 + 语义向量检索）"""
    user_id = kwargs.get("user_id")
    session_id = kwargs.get("session_id")
    time_range: Optional[Tuple[datetime, datetime]] = kwargs.get("time_range")
    importance_threshold: Optional[float] = kwargs.get("importance_threshold")

    # 结构化过滤候选（来自权威库）
    candidate_ids: Optional[set] = None
    if time_range is not None or importance_threshold is not None:
        start_ts = int(time_range[0].timestamp()) if time_range else None
        end_ts = int(time_range[1].timestamp()) if time_range else None
        docs = self.doc_store.search_memories(
            user_id=user_id,
            memory_type="episodic",
            start_time=start_ts,
            end_time=end_ts,
            importance_threshold=importance_threshold,
            limit=1000
        )
        candidate_ids = {d["memory_id"] for d in docs}

    # 向量检索（Qdrant）
    try:
        query_vec = self.embedder.encode(query)
        if hasattr(query_vec, "tolist"):
            query_vec = query_vec.tolist()
        where = {"memory_type": "episodic"}
        if user_id:
            where["user_id"] = user_id
        hits = self.vector_store.search_similar(
            query_vector=query_vec,
            limit=max(limit * 5, 20),
            where=where
        )
    except Exception:
        hits = []

    # 过滤与重排
    now_ts = int(datetime.now().timestamp())
    results: List[Tuple[float, MemoryItem]] = []
    seen = set()
    for hit in hits:
        meta = hit.get("metadata", {})
        mem_id = meta.get("memory_id")
        if not mem_id or mem_id in seen:
            continue
        
        # 检查是否已遗忘
        episode = next((e for e in self.episodes if e.episode_id == mem_id), None)
        if episode and episode.context.get("forgotten", False):
            continue  # 跳过已遗忘的记忆
            
        if candidate_ids is not None and mem_id not in candidate_ids:
            continue
        if session_id and meta.get("session_id") != session_id:
            continue

        # 从权威库读取完整记录
        doc = self.doc_store.get_memory(mem_id)
        if not doc:
            continue

        # 计算综合分数：向量0.6 + 近因0.2 + 重要性0.2
        vec_score = float(hit.get("score", 0.0))
        age_days = max(0.0, (now_ts - int(doc["timestamp"])) / 86400.0)
        recency_score = 1.0 / (1.0 + age_days)
        imp = float(doc.get("importance", 0.5))
        
        # 新评分算法：向量检索纯基于相似度，重要性作为加权因子
        # 基础相似度得分（不受重要性影响）
        base_relevance = vec_score * 0.8 + recency_score * 0.2
        
        # 重要性作为乘法加权因子，范围 [0.8, 1.2]
        importance_weight = 0.8 + (imp * 0.4)
        
        # 最终得分：相似度 * 重要性权重
        combined = base_relevance * importance_weight

        item = MemoryItem(
            id=doc["memory_id"],
            content=doc["content"],
            memory_type=doc["memory_type"],
            user_id=doc["user_id"],
            timestamp=datetime.fromtimestamp(doc["timestamp"]),
            importance=doc.get("importance", 0.5),
            metadata={
                **doc.get("properties", {}),
                "relevance_score": combined,
                "vector_score": vec_score,
                "recency_score": recency_score
            }
        )
        results.append((combined, item))
        seen.add(mem_id)

    # 若向量检索无结果，回退到简单关键词匹配（内存缓存）
    if not results:
        fallback = super()._generate_id  # 占位以避免未使用警告
        query_lower = query.lower()
        for ep in self._filter_episodes(user_id, session_id, time_range):
            if query_lower in ep.content.lower():
                recency_score = 1.0 / (1.0 + max(0.0, (now_ts - int(ep.timestamp.timestamp())) / 86400.0))
                # 回退匹配：新评分算法
                keyword_score = 0.5  # 简单关键词匹配的基础分数
                base_relevance = keyword_score * 0.8 + recency_score * 0.2
                importance_weight = 0.8 + (ep.importance * 0.4)
                combined = base_relevance * importance_weight
                item = MemoryItem(
                    id=ep.episode_id,
                    content=ep.content,
                    memory_type="episodic",
                    user_id=ep.user_id,
                    timestamp=ep.timestamp,
                    importance=ep.importance,
                    metadata={
                        "session_id": ep.session_id,
                        "context": ep.context,
                        "outcome": ep.outcome,
                        "relevance_score": combined
                    }
                )
                results.append((combined, item))

    results.sort(key=lambda x: x[0], reverse=True)
    return [it for _, it in results[:limit]]

查询依旧大工程这块。

我们从 SQLite 中提取出满足条件的初步候选，将其 id 存入一个 set，然后经过查询 qdrant 排除相关度低的候选，set 中剩余的 id 再经由 SQLite 查询完整的 MemoryItem，最排序输出。

不过，如果向量数据库的结果为空，可能是由于数据库连接失效，又或者真的相关度不高，那就退化为关键词匹配。

测试

我们把上面测试 MemoryTool 的程序的类型改成 semantic：

# 添加第二个记忆
result2 = memory_tool.run({"action": "add", "content": "李四是前端工程师，擅长React和Vue.js开发", "memory_type": "episodic", "importance": 0.7})
print(f"记忆2: {result2}")

结果：

> python -m test.__testMemoryTool

✅ 记忆助手 初始化完成，工具调用: 禁用
[OK] SQLite 数据库表和索引创建完成
[OK] SQLite 文档存储初始化完成: ./memory_data\memory.db
INFO:storage.qdrant_store:✅ 成功连接到Qdrant服务: http://localhost:6333
INFO:httpx:HTTP Request: GET http://localhost:6333/collections "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: GET http://localhost:6333/collections "HTTP/1.1 200 OK"
INFO:storage.qdrant_store:✅ 使用现有Qdrant集合: helloAgent
INFO:httpx:HTTP Request: PATCH http://localhost:6333/collections/helloAgent "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: GET http://localhost:6333 "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:memory.types.semanticMemory:✅ 嵌入模型就绪，维度: 1024
INFO:storage.qdrant_store:✅ 成功连接到本地Qdrant服务: localhost:6333
INFO:httpx:HTTP Request: GET http://localhost:6333/collections "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: GET http://localhost:6333/collections "HTTP/1.1 200 OK"
INFO:storage.qdrant_store:✅ 使用现有Qdrant集合: helloAgent
INFO:httpx:HTTP Request: PATCH http://localhost:6333/collections/helloAgent "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: GET http://localhost:6333 "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:storage.neo4j_store:✅ 成功连接到Neo4j服务: bolt://localhost:7687
INFO:neo4j.notifications:Received notification from DBMS server: <GqlStatusObject gql_status='00NA0', status_description="note: successful completion - index or constraint already exists. The command 'CREATE RANGE INDEX entity_id_index IF NOT EXISTS FOR (e:Entity) ON (e.id)' has no effect. The index or constraint specified by 'RANGE INDEX entity_id_index FOR (e:Entity) ON (e.id)' already exists.", position=None, raw_classification='SCHEMA', classification=<NotificationClassification.SCHEMA: 'SCHEMA'>, raw_severity='INFORMATION', severity=<NotificationSeverity.INFORMATION: 'INFORMATION'>, diagnostic_record={'_classification': 'SCHEMA', '_severity': 'INFORMATION', 'OPERATION': '', 'OPERATION_CODE': '0', 'CURRENT_SCHEMA': '/'}> for query: 'CREATE INDEX entity_id_index IF NOT EXISTS FOR (e:Entity) ON (e.id)'
INFO:neo4j.notifications:Received notification from DBMS server: <GqlStatusObject gql_status='00NA0', status_description="note: successful completion - index or constraint already exists. The command 'CREATE RANGE INDEX entity_name_index IF NOT EXISTS FOR (e:Entity) ON (e.name)' has no effect. The index or constraint specified by 'RANGE INDEX entity_name_index FOR (e:Entity) ON (e.name)' already exists.", position=None, raw_classification='SCHEMA', classification=<NotificationClassification.SCHEMA: 'SCHEMA'>, raw_severity='INFORMATION', severity=<NotificationSeverity.INFORMATION: 'INFORMATION'>, diagnostic_record={'_classification': 'SCHEMA', '_severity': 'INFORMATION', 'OPERATION': '', 'OPERATION_CODE': '0', 'CURRENT_SCHEMA': '/'}> for query: 'CREATE INDEX entity_name_index IF NOT EXISTS FOR (e:Entity) ON (e.name)'
INFO:neo4j.notifications:Received notification from DBMS server: <GqlStatusObject gql_status='00NA0', status_description="note: successful completion - index or constraint already exists. The command 'CREATE RANGE INDEX entity_type_index IF NOT EXISTS FOR (e:Entity) ON (e.type)' has no effect. The index or constraint specified by 'RANGE INDEX entity_type_index FOR (e:Entity) ON (e.type)' already exists.", position=None, raw_classification='SCHEMA', classification=<NotificationClassification.SCHEMA: 'SCHEMA'>, raw_severity='INFORMATION', severity=<NotificationSeverity.INFORMATION: 'INFORMATION'>, diagnostic_record={'_classification': 'SCHEMA', '_severity': 'INFORMATION', 'OPERATION': '', 'OPERATION_CODE': '0', 'CURRENT_SCHEMA': '/'}> for query: 'CREATE INDEX entity_type_index IF NOT EXISTS FOR (e:Entity) ON (e.type)'
INFO:neo4j.notifications:Received notification from DBMS server: <GqlStatusObject gql_status='00NA0', status_description="note: successful completion - index or constraint already exists. The command 'CREATE RANGE INDEX memory_id_index IF NOT EXISTS FOR (e:Memory) ON (e.id)' has no effect. The index or constraint specified by 'RANGE INDEX memory_id_index FOR (e:Memory) ON (e.id)' already exists.", position=None, raw_classification='SCHEMA', classification=<NotificationClassification.SCHEMA: 'SCHEMA'>, raw_severity='INFORMATION', severity=<NotificationSeverity.INFORMATION: 'INFORMATION'>, diagnostic_record={'_classification': 'SCHEMA', '_severity': 'INFORMATION', 'OPERATION': '', 'OPERATION_CODE': '0', 'CURRENT_SCHEMA': '/'}> for query: 'CREATE INDEX memory_id_index IF NOT EXISTS FOR (m:Memory) ON (m.id)'
INFO:neo4j.notifications:Received notification from DBMS server: <GqlStatusObject gql_status='00NA0', status_description="note: successful completion - index or constraint already exists. The command 'CREATE RANGE INDEX memory_type_index IF NOT EXISTS FOR (e:Memory) ON (e.memory_type)' has no effect. The index or constraint specified by 'RANGE INDEX memory_type_index FOR (e:Memory) ON (e.memory_type)' already exists.", position=None, raw_classification='SCHEMA', classification=<NotificationClassification.SCHEMA: 'SCHEMA'>, raw_severity='INFORMATION', severity=<NotificationSeverity.INFORMATION: 'INFORMATION'>, diagnostic_record={'_classification': 'SCHEMA', '_severity': 'INFORMATION', 'OPERATION': '', 'OPERATION_CODE': '0', 'CURRENT_SCHEMA': '/'}> for query: 'CREATE INDEX memory_type_index IF NOT EXISTS FOR (m:Memory) ON (m.memory_type)'
INFO:neo4j.notifications:Received notification from DBMS server: <GqlStatusObject gql_status='00NA0', status_description="note: successful completion - index or constraint already exists. The command 'CREATE RANGE INDEX memory_timestamp_index IF NOT EXISTS FOR (e:Memory) ON (e.timestamp)' has no effect. The index or constraint specified by 'RANGE INDEX memory_timestamp_index FOR (e:Memory) ON (e.timestamp)' already exists.", position=None, raw_classification='SCHEMA', classification=<NotificationClassification.SCHEMA: 'SCHEMA'>, raw_severity='INFORMATION', severity=<NotificationSeverity.INFORMATION: 'INFORMATION'>, diagnostic_record={'_classification': 'SCHEMA', '_severity': 'INFORMATION', 'OPERATION': '', 'OPERATION_CODE': '0', 'CURRENT_SCHEMA': '/'}> for query: 'CREATE INDEX memory_timestamp_index IF NOT EXISTS FOR (m:Memory) ON (m.timestamp)'
INFO:storage.neo4j_store:✅ Neo4j索引创建完成
INFO:memory.types.semanticMemory:✅ 加载中文spaCy模型: zh_core_web_sm
INFO:memory.types.semanticMemory:✅ 加载英文spaCy模型: en_core_web_sm
INFO:memory.types.semanticMemory:🎯 主要使用中文spaCy模型
INFO:memory.types.semanticMemory:📚 可用语言模型: 中文, 英文
INFO:memory.types.semanticMemory:增强语义记忆初始化完成（使用Qdrant+Neo4j专业数据库）
INFO:memory.memoryManager:MemoryManager初始化完成，启用记忆类型: ['working', 'episodic', 'semantic']
✅ 工具 'memory' 已注册。
=== 添加多个记忆 ===
记忆1: ✅ 记忆已添加
   ID: 872d6d9f-7b37-4b0b-a409-5010349b01af
   类型: working
   内容: 用户张三是一名Python开发者，专注于机器学习和数据分析
INFO:storage.qdrant_store:[Qdrant] add_vectors start: n_vectors=1 n_meta=1 collection=helloAgent
INFO:storage.qdrant_store:[Qdrant] upsert begin: points=1
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/points?wait=true "HTTP/1.1 200 OK"
INFO:storage.qdrant_store:[Qdrant] upsert done
INFO:storage.qdrant_store:✅ 成功添加 1 个向量到Qdrant
记忆2: ✅ 记忆已添加
   ID: 8bb07638-85b3-4355-a41f-4458dd31b5b1
   类型: episodic
   内容: 李四是前端工程师，擅长React和Vue.js开发
记忆3: ✅ 记忆已添加
   ID: 9d3bd0f3-c214-4419-9831-53bffa07a651
   类型: working
   内容: 王五是产品经理，负责用户体验设计和需求分析

=== 搜索特定记忆 ===
🔍 搜索 '前端工程师':
INFO:httpx:HTTP Request: POST http://localhost:6333/collections/helloAgent/points/query "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: POST http://localhost:6333/collections/helloAgent/points/query "HTTP/1.1 200 OK"
INFO:memory.types.semanticMemory:✅ 检索到 1 条相关记忆
🔍 搜索 '前端工程师' 的结果（共 2 条）:
  [1] [episodic] 李四是前端工程师，擅长React和Vue.js开发  (importance=0.70, id=8bb07638...)
  [2] [semantic] 李四是前端工程师，擅长React和Vue.js开发  (importance=0.70, id=5d3dbf35...)

=== 记忆摘要 ===
INFO:httpx:HTTP Request: GET http://localhost:6333/collections/helloAgent "HTTP/1.1 200 OK"
📊 记忆系统摘要:
   用户: user123
   总记忆数: 3
   启用的记忆类型: working, episodic, semantic
   配置: 最大容量=100, 重要性阈值=0.1, 衰减因子=0.95
   [working] 活跃=2, 容量使用率=20.0%
   [episodic] 活跃=1, 容量使用率=0.0%
   [semantic] 活跃=0, 容量使用率=0.0%

语义记忆 SemanticMemory

官方定义；

语义记忆是记忆系统中最复杂的部分，它负责存储抽象的概念、规则和知识。

官方的设计思路是：

设计重点在于知识的结构化表示和智能推理能力。语义记忆采用了Neo4j图数据库和Qdrant向量数据库的混合架构，这种设计让系统既能进行快速的语义检索，又能利用知识图谱进行复杂的关系推理。

和情景记忆不同的是，语义记忆除了用到嵌入模型在 qdrant 中向量化存储，还要在 neo4j 中存储对象间关系等 绝对真理 。

这种设计让系统既能进行快速的语义检索，又能利用知识图谱进行复杂的关系推理。

实体与关系类 Entity & Relation

neo4j 中需要存储 entity 之间的 relation，分别创建类：

class Entity:
    """实体类"""
    
    def __init__(
        self,
        entity_id: str,
        name: str,
        entity_type: str = "MISC",
        description: str = "",
        properties: Dict[str, Any] = None
    ):
        self.entity_id = entity_id
        self.name = name
        self.entity_type = entity_type  # PERSON, ORG, PRODUCT, SKILL, CONCEPT等
        self.description = description
        self.properties = properties or {}
        self.created_at = datetime.now()
        self.updated_at = datetime.now()
        self.frequency = 1  # 出现频率
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "entity_id": self.entity_id,
            "name": self.name,
            "entity_type": self.entity_type,
            "description": self.description,
            "properties": self.properties,
            "frequency": self.frequency
        }

实体类规定了一个实体的名称、类型、所含参数等，使用唯一标识符 entity_id 进行校准。

class Relation:
    """关系类"""
    
    def __init__(
        self,
        from_entity: str,
        to_entity: str,
        relation_type: str,
        strength: float = 1.0,
        evidence: str = "",
        properties: Dict[str, Any] = None
    ):
        self.from_entity = from_entity
        self.to_entity = to_entity
        self.relation_type = relation_type
        self.strength = strength
        self.evidence = evidence  # 支持该关系的原文本
        self.properties = properties or {}
        self.created_at = datetime.now()
        self.frequency = 1  # 关系出现频率
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "from_entity": self.from_entity,
            "to_entity": self.to_entity,
            "relation_type": self.relation_type,
            "strength": self.strength,
            "evidence": self.evidence,
            "properties": self.properties,
            "frequency": self.frequency
        }

关系类与实体类不同，不包含一个 id，而是使用 from_entity 和 to_entity 进行唯一标识。同时，一个关系也展示了两个实体间的关系、关系强度、证据、参数等。

初始化

class SemanticMemory(Memory):
    """增强语义记忆实现
    
    特点：
    - 使用HuggingFace中文预训练模型进行文本嵌入
    - 向量检索进行快速相似度匹配
    - 知识图谱存储实体和关系
    - 混合检索策略：向量+图+语义推理
    """
    
    def __init__(self, config: MemoryConfig, storage_backend=None):
        super().__init__(config, storage_backend)
        
        # 统一嵌入模型（多语言，默认384维）
        self.embedder = get_text_embedder()

        # 初始化专业数据库存储


        # 实体和关系缓存 (用于快速访问)
        self.entities: Dict[str, Entity] = {}
        self.relations: List[Relation] = []
        
        # 实体识别器
        self.nlp = None
        self._init_nlp()
        
        # 记忆存储
        self.semantic_memories: List[MemoryItem] = []
        self.memory_embeddings: Dict[str, np.ndarray] = {}
        
        logger.info("增强语义记忆初始化完成（使用Qdrant+Neo4j专业数据库）")

    def _init_embedding_model(self):
        """初始化统一嵌入模型（由 embedding_provider 管理）。"""
        try:
            self.embedding_model = get_text_embedder()
            # 轻量健康检查与日志
            try:
                test_vec = self.embedding_model.encode("health_check")
                dim = getattr(self.embedding_model, "dimension", len(test_vec))
                logger.info(f"✅ 嵌入模型就绪，维度: {dim}")
            except Exception:
                logger.info("✅ 嵌入模型就绪")
        except Exception as e:
            logger.error(f"❌ 嵌入模型初始化失败: {e}")
            raise
    
    def _init_databases(self):
        """初始化专业数据库存储"""
        try:
            # 向量存储
            qdrant_url = os.getenv("QDRANT_URI")
            qdrant_api_key = os.getenv("QDRANT_API_KEY")
            self.vector_store = QdrantConnectionManager.get_instance(
                url=qdrant_url,
                api_key=qdrant_api_key,
                collection_name=os.getenv("QDRANT_COLLECTION", "hello_agents_vectors"),
                vector_size=get_dimension(getattr(self.embedder, 'dimension', 384)),
                distance=os.getenv("QDRANT_DISTANCE", "cosine")
            )
            # 图存储
            self.graph_store = Neo4jGraphStore(
                uri=os.getenv("NEO4J_URI", "bolt://localhost:7687"),
                username=os.getenv("NEO4J_USERNAME", "neo4j"),
                password=os.getenv("NEO4J_PASSWORD", "password"),
                database=os.getenv("NEO4J_DATABASE", "neo4j")
            )
            
        except Exception as e:
            logger.error(f"❌ 数据库初始化失败: {e}")
            logger.info("💡 请检查数据库配置和网络连接")
            logger.info("💡 参考 DATABASE_SETUP_GUIDE.md 进行配置")
            raise
    
    def _init_nlp(self):
        """初始化NLP处理器 - 智能多语言支持"""
        try:
            self.nlp_models = {}
            
            # 尝试加载多语言模型
            models_to_try = [
                ("zh_core_web_sm", "中文"),
                ("en_core_web_sm", "英文")
            ]
            
            loaded_models = []
            for model_name, lang_name in models_to_try:
                try:
                    nlp = spacy.load(model_name)
                    self.nlp_models[model_name] = nlp
                    loaded_models.append(lang_name)
                    logger.info(f"✅ 加载{lang_name}spaCy模型: {model_name}")
                except OSError:
                    logger.warning(f"⚠️ {lang_name}spaCy模型不可用: {model_name}")
            
            # 设置主要NLP处理器
            if "zh_core_web_sm" in self.nlp_models:
                self.nlp = self.nlp_models["zh_core_web_sm"]
                logger.info("🎯 主要使用中文spaCy模型")
            elif "en_core_web_sm" in self.nlp_models:
                self.nlp = self.nlp_models["en_core_web_sm"]
                logger.info("🎯 主要使用英文spaCy模型")
            else:
                self.nlp = None
                logger.warning("⚠️ 无可用spaCy模型，实体提取将受限")
            
            if loaded_models:
                logger.info(f"📚 可用语言模型: {', '.join(loaded_models)}")
                
        except ImportError:
            logger.warning("⚠️ spaCy不可用，实体提取将受限")
            self.nlp = None
            self.nlp_models = {}

add

自己看。

retrieve

自己看。

测试

我们把上面测试 MemoryTool 的程序的类型改成 semantic：

# 添加第二个记忆
result2 = memory_tool.run({"action": "add", "content": "李四是前端工程师，擅长React和Vue.js开发", "memory_type": "semantic", "importance": 0.7})
print(f"记忆2: {result2}")

再次跑：

> python -m test.__testMemoryTool

✅ 记忆助手 初始化完成，工具调用: 禁用
[OK] SQLite 数据库表和索引创建完成
[OK] SQLite 文档存储初始化完成: ./memory_data\memory.db
INFO:storage.qdrant_store:✅ 成功连接到Qdrant服务: http://localhost:6333
INFO:httpx:HTTP Request: GET http://localhost:6333/collections "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: GET http://localhost:6333/collections "HTTP/1.1 200 OK"
INFO:storage.qdrant_store:✅ 使用现有Qdrant集合: helloAgent
INFO:httpx:HTTP Request: PATCH http://localhost:6333/collections/helloAgent "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: GET http://localhost:6333 "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:memory.types.semanticMemory:✅ 嵌入模型就绪，维度: 1024
INFO:storage.qdrant_store:✅ 成功连接到本地Qdrant服务: localhost:6333
INFO:httpx:HTTP Request: GET http://localhost:6333/collections "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: GET http://localhost:6333/collections "HTTP/1.1 200 OK"
INFO:storage.qdrant_store:✅ 使用现有Qdrant集合: helloAgent
INFO:httpx:HTTP Request: PATCH http://localhost:6333/collections/helloAgent "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: GET http://localhost:6333 "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/index?wait=true "HTTP/1.1 200 OK"
INFO:storage.neo4j_store:✅ 成功连接到Neo4j服务: bolt://localhost:7687
INFO:neo4j.notifications:Received notification from DBMS server: <GqlStatusObject gql_status='00NA0', status_description="note: successful completion - index or constraint already exists. The command 'CREATE RANGE INDEX entity_id_index IF NOT EXISTS FOR (e:Entity) ON (e.id)' has no effect. The index or constraint specified by 'RANGE INDEX entity_id_index FOR (e:Entity) ON (e.id)' already exists.", position=None, raw_classification='SCHEMA', classification=<NotificationClassification.SCHEMA: 'SCHEMA'>, raw_severity='INFORMATION', severity=<NotificationSeverity.INFORMATION: 'INFORMATION'>, diagnostic_record={'_classification': 'SCHEMA', '_severity': 'INFORMATION', 'OPERATION': '', 'OPERATION_CODE': '0', 'CURRENT_SCHEMA': '/'}> for query: 'CREATE INDEX entity_id_index IF NOT EXISTS FOR (e:Entity) ON (e.id)'
INFO:neo4j.notifications:Received notification from DBMS server: <GqlStatusObject gql_status='00NA0', status_description="note: successful completion - index or constraint already exists. The command 'CREATE RANGE INDEX entity_name_index IF NOT EXISTS FOR (e:Entity) ON (e.name)' has no effect. The index or constraint specified by 'RANGE INDEX entity_name_index FOR (e:Entity) ON (e.name)' already exists.", position=None, raw_classification='SCHEMA', classification=<NotificationClassification.SCHEMA: 'SCHEMA'>, raw_severity='INFORMATION', severity=<NotificationSeverity.INFORMATION: 'INFORMATION'>, diagnostic_record={'_classification': 'SCHEMA', '_severity': 'INFORMATION', 'OPERATION': '', 'OPERATION_CODE': '0', 'CURRENT_SCHEMA': '/'}> for query: 'CREATE INDEX entity_name_index IF NOT EXISTS FOR (e:Entity) ON (e.name)'
INFO:neo4j.notifications:Received notification from DBMS server: <GqlStatusObject gql_status='00NA0', status_description="note: successful completion - index or constraint already exists. The command 'CREATE RANGE INDEX entity_type_index IF NOT EXISTS FOR (e:Entity) ON (e.type)' has no effect. The index or constraint specified by 'RANGE INDEX entity_type_index FOR (e:Entity) ON (e.type)' already exists.", position=None, raw_classification='SCHEMA', classification=<NotificationClassification.SCHEMA: 'SCHEMA'>, raw_severity='INFORMATION', severity=<NotificationSeverity.INFORMATION: 'INFORMATION'>, diagnostic_record={'_classification': 'SCHEMA', '_severity': 'INFORMATION', 'OPERATION': '', 'OPERATION_CODE': '0', 'CURRENT_SCHEMA': '/'}> for query: 'CREATE INDEX entity_type_index IF NOT EXISTS FOR (e:Entity) ON (e.type)'
INFO:neo4j.notifications:Received notification from DBMS server: <GqlStatusObject gql_status='00NA0', status_description="note: successful completion - index or constraint already exists. The command 'CREATE RANGE INDEX memory_id_index IF NOT EXISTS FOR (e:Memory) ON (e.id)' has no effect. The index or constraint specified by 'RANGE INDEX memory_id_index FOR (e:Memory) ON (e.id)' already exists.", position=None, raw_classification='SCHEMA', classification=<NotificationClassification.SCHEMA: 'SCHEMA'>, raw_severity='INFORMATION', severity=<NotificationSeverity.INFORMATION: 'INFORMATION'>, diagnostic_record={'_classification': 'SCHEMA', '_severity': 'INFORMATION', 'OPERATION': '', 'OPERATION_CODE': '0', 'CURRENT_SCHEMA': '/'}> for query: 'CREATE INDEX memory_id_index IF NOT EXISTS FOR (m:Memory) ON (m.id)'
INFO:neo4j.notifications:Received notification from DBMS server: <GqlStatusObject gql_status='00NA0', status_description="note: successful completion - index or constraint already exists. The command 'CREATE RANGE INDEX memory_type_index IF NOT EXISTS FOR (e:Memory) ON (e.memory_type)' has no effect. The index or constraint specified by 'RANGE INDEX memory_type_index FOR (e:Memory) ON (e.memory_type)' already exists.", position=None, raw_classification='SCHEMA', classification=<NotificationClassification.SCHEMA: 'SCHEMA'>, raw_severity='INFORMATION', severity=<NotificationSeverity.INFORMATION: 'INFORMATION'>, diagnostic_record={'_classification': 'SCHEMA', '_severity': 'INFORMATION', 'OPERATION': '', 'OPERATION_CODE': '0', 'CURRENT_SCHEMA': '/'}> for query: 'CREATE INDEX memory_type_index IF NOT EXISTS FOR (m:Memory) ON (m.memory_type)'
INFO:neo4j.notifications:Received notification from DBMS server: <GqlStatusObject gql_status='00NA0', status_description="note: successful completion - index or constraint already exists. The command 'CREATE RANGE INDEX memory_timestamp_index IF NOT EXISTS FOR (e:Memory) ON (e.timestamp)' has no effect. The index or constraint specified by 'RANGE INDEX memory_timestamp_index FOR (e:Memory) ON (e.timestamp)' already exists.", position=None, raw_classification='SCHEMA', classification=<NotificationClassification.SCHEMA: 'SCHEMA'>, raw_severity='INFORMATION', severity=<NotificationSeverity.INFORMATION: 'INFORMATION'>, diagnostic_record={'_classification': 'SCHEMA', '_severity': 'INFORMATION', 'OPERATION': '', 'OPERATION_CODE': '0', 'CURRENT_SCHEMA': '/'}> for query: 'CREATE INDEX memory_timestamp_index IF NOT EXISTS FOR (m:Memory) ON (m.timestamp)'
INFO:storage.neo4j_store:✅ Neo4j索引创建完成
INFO:memory.types.semanticMemory:✅ 加载中文spaCy模型: zh_core_web_sm
INFO:memory.types.semanticMemory:✅ 加载英文spaCy模型: en_core_web_sm
INFO:memory.types.semanticMemory:🎯 主要使用中文spaCy模型
INFO:memory.types.semanticMemory:📚 可用语言模型: 中文, 英文
INFO:memory.types.semanticMemory:增强语义记忆初始化完成（使用Qdrant+Neo4j专业数据库）
INFO:memory.memoryManager:MemoryManager初始化完成，启用记忆类型: ['working', 'episodic', 'semantic']
✅ 工具 'memory' 已注册。
=== 添加多个记忆 ===
记忆1: ✅ 记忆已添加
   ID: 3ef16625-c077-4002-9725-e069e089f6d2
   类型: working
   内容: 用户张三是一名Python开发者，专注于机器学习和数据分析
INFO:storage.qdrant_store:[Qdrant] add_vectors start: n_vectors=1 n_meta=1 collection=helloAgent
INFO:storage.qdrant_store:[Qdrant] upsert begin: points=1
INFO:httpx:HTTP Request: PUT http://localhost:6333/collections/helloAgent/points?wait=true "HTTP/1.1 200 OK"
INFO:storage.qdrant_store:[Qdrant] upsert done
INFO:storage.qdrant_store:✅ 成功添加 1 个向量到Qdrant
INFO:memory.types.semanticMemory:✅ 添加语义记忆: 2个实体, 1个关系
记忆2: ✅ 记忆已添加
   ID: 5d3dbf35-7f8f-46b3-b3f2-970d80640b17
   类型: semantic
   内容: 李四是前端工程师，擅长React和Vue.js开发
记忆3: ✅ 记忆已添加
   ID: b5857a65-7ee6-4e47-b177-6dff006710cc
   类型: working
   内容: 王五是产品经理，负责用户体验设计和需求分析

=== 搜索特定记忆 ===
🔍 搜索 '前端工程师':
INFO:httpx:HTTP Request: POST http://localhost:6333/collections/helloAgent/points/query "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: POST http://localhost:6333/collections/helloAgent/points/query "HTTP/1.1 200 OK"
INFO:memory.types.semanticMemory:✅ 检索到 1 条相关记忆
🔍 搜索 '前端工程师' 的结果（共 1 条）:
  [1] [semantic] 李四是前端工程师，擅长React和Vue.js开发  (importance=0.70, id=5d3dbf35...)

=== 记忆摘要 ===
INFO:httpx:HTTP Request: GET http://localhost:6333/collections/helloAgent "HTTP/1.1 200 OK"
📊 记忆系统摘要:
   用户: user123
   总记忆数: 3
   启用的记忆类型: working, episodic, semantic
   配置: 最大容量=100, 重要性阈值=0.1, 衰减因子=0.95
   [working] 活跃=2, 容量使用率=20.0%
   [episodic] 活跃=0, 容量使用率=0.0%
   [semantic] 活跃=1, 容量使用率=0.0%

成功了。

rag

检索增强生成 Retrieval-Augmented Generation，其核心思想就是：将知识库注入 prompt，补足 LLM 的知识水平或是让其遵守一定事实。

体验官方 RAG

from agent.simpleAgent import SimpleAgent
from core.llm import LLM
from tool.toolRegistry import ToolRegistry
from hello_agents.tools import RAGTool

# 创建具有RAG能力的Agent
llm = LLM()
agent = SimpleAgent(name="知识助手", llm=llm)

# 创建RAG工具
rag_tool = RAGTool(
    knowledge_base_path="./knowledge_base",
    collection_name="test_collection",
    rag_namespace="test"
)

tool_registry = ToolRegistry()
tool_registry.register_tool(rag_tool)
agent.tool_registry = tool_registry

# 体验RAG功能
# 添加第一个知识
result1 = rag_tool.run({"action":"add_text", 
    "text":"Python是一种高级编程语言，由Guido van Rossum于1991年首次发布。Python的设计哲学强调代码的可读性和简洁的语法。",
    "document_id":"python_intro"})
print(f"知识1: {result1}")

# 添加第二个知识  
result2 = rag_tool.run({"action":"add_text",
    "text":"机器学习是人工智能的一个分支，通过算法让计算机从数据中学习模式。主要包括监督学习、无监督学习和强化学习三种类型。",
    "document_id":"ml_basics"})
print(f"知识2: {result2}")

# 添加第三个知识
result3 = rag_tool.run({"action":"add_text",
    "text":"RAG（检索增强生成）是一种结合信息检索和文本生成的AI技术。它通过检索相关知识来增强大语言模型的生成能力。",
    "document_id":"rag_concept"})
print(f"知识3: {result3}")

print("\n=== 搜索知识 ===")
result = rag_tool.run({"action":"search",
    "query":"Python编程语言的历史",
    "limit":3,
    "min_score":0.1
})
print(result)

print("\n=== 知识库统计 ===")
result = rag_tool.run({"action":"stats"})
print(result)

结果：

> python -m test.__testRagTool

✅ 知识助手 初始化完成，工具调用: 禁用
INFO:hello_agents.memory.storage.qdrant_store:✅ 成功连接到Qdrant服务: http://localhost:6333
INFO:hello_agents.memory.storage.qdrant_store:✅ 使用现有Qdrant集合: test_collection
✅ RAG工具初始化成功: namespace=test, collection=test_collection
✅ 工具 'rag' 已注册。
INFO:hello_agents.memory.storage.qdrant_store:✅ 成功连接到Qdrant服务: http://localhost:6333
INFO:hello_agents.memory.storage.qdrant_store:✅ 使用现有Qdrant集合: test_collection
[RAG] Universal loader start: files=1 chunk_size=800 overlap=100 ns=default
[RAG] Processing: ./knowledge_base\python_intro.md
C:\Users\a1829\AppData\Local\Programs\Python\Python312\Lib\site-packages\pydub\utils.py:170: RuntimeWarning: Couldn't find ffmpeg or avconv - defaulting to ffmpeg, but may not work
  warn("Couldn't find ffmpeg or avconv - defaulting to ffmpeg, but may not work", RuntimeWarning)
[RAG] Universal loader done: total_chunks=1
[RAG] Embedding start: total_texts=1 batch_size=64
[RAG] Embedding progress: 1/1
[RAG] Qdrant upsert start: n=1
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] add_vectors start: n_vectors=1 n_meta=1 collection=test_collection
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] upsert begin: points=1
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] upsert done
INFO:hello_agents.memory.storage.qdrant_store:✅ 成功添加 1 个向量到Qdrant
[RAG] Qdrant upsert done: 1 vectors indexed
知识1: ✅ 文本已添加到知识库: python_intro
📊 分块数量: 1
⏱️ 处理时间: 1520ms
📝 命名空间: default
[RAG] Universal loader start: files=1 chunk_size=800 overlap=100 ns=default
[RAG] Processing: ./knowledge_base\ml_basics.md
[RAG] Universal loader done: total_chunks=1
[RAG] Embedding start: total_texts=1 batch_size=64
[RAG] Embedding progress: 1/1
[RAG] Qdrant upsert start: n=1
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] add_vectors start: n_vectors=1 n_meta=1 collection=test_collection
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] upsert begin: points=1
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] upsert done
INFO:hello_agents.memory.storage.qdrant_store:✅ 成功添加 1 个向量到Qdrant
[RAG] Qdrant upsert done: 1 vectors indexed
知识2: ✅ 文本已添加到知识库: ml_basics
📊 分块数量: 1
⏱️ 处理时间: 510ms
📝 命名空间: default
[RAG] Universal loader start: files=1 chunk_size=800 overlap=100 ns=default
[RAG] Processing: ./knowledge_base\rag_concept.md
[RAG] Universal loader done: total_chunks=1
[RAG] Embedding start: total_texts=1 batch_size=64
[RAG] Embedding progress: 1/1
[RAG] Qdrant upsert start: n=1
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] add_vectors start: n_vectors=1 n_meta=1 collection=test_collection
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] upsert begin: points=1
INFO:hello_agents.memory.storage.qdrant_store:[Qdrant] upsert done
INFO:hello_agents.memory.storage.qdrant_store:✅ 成功添加 1 个向量到Qdrant
[RAG] Qdrant upsert done: 1 vectors indexed
知识3: ✅ 文本已添加到知识库: rag_concept
📊 分块数量: 1
⏱️ 处理时间: 470ms
📝 命名空间: default

=== 搜索知识 ===
搜索结果：

1. 文档: **./knowledge_base\python_intro.md** (相似度: 0.702)
   Python是一种高级编程语言，由Guido van Rossum于1991年首次发布。Python的设计哲学强调代码的可读性和简洁的语法。...

2. 文档: **./knowledge_base\rag_concept.md** (相似度: 0.241)
   RAG（检索增强生成）是一种结合信息检索和文本生成的AI技术。它通过检索相关知识来增强大语言模型的生成能力。...

3. 文档: **./knowledge_base\ml_basics.md** (相似度: 0.218)
   机器学习是人工智能的一个分支，通过算法让计算机从数据中学习模式。主要包括监督学习、无监督学习和强化学习三种类型。...

=== 知识库统计 ===
📊 **RAG 知识库统计**
📝 命名空间: default
📋 集合名称: test_collection
📂 存储根路径: ./knowledge_base
📦 存储类型: qdrant
📊 文档分块数: 3
🔢 向量维度: 1024
📎 距离度量: Cosine

🟢 **系统状态**
✅ RAG 管道: 正常
✅ LLM 连接: 正常

本节目标

HelloAgent/
   │
   ├── rag/                         
   │   └── pipeline.py              # 管线
   │
   ├── tool/
   │   ├── toolAction.py            # Memory 基类 
   │   └── builtin
   │        └── ragTool.py          # RAG 工具包

Pipeline 基本流程

RAGTool 是一种工具，那么这就和前面提到的 MemoryTool 存在着一样的问题：RAGTool 是对外暴露的接口，它是否应该直接引用底层存储组件（例如QdrantVectorStore、Neo4j等数据库接口）？不要，我们需要构建中间层。不过，官方的意图不只是为了抽象封装，更主要的原因也是 RAGTool 到底层数据库之间的操作，其本身也便于梳理成一个工作流，也就是 Pipeline 的由来。

在官方的设计中，pipeline 是一个包含了若干数据处理的函数库，包括多模态文件统一转文本、数据分块、向量化等，其在 RAGTool 的调用流程大致是：

任意格式文档 → MarkItDown转换 → Markdown文本 → 智能分块 → 向量化 → 存储检索

该流程只涉及到了一些函数，官方提出来重点进行了介绍。

多模态文档载入

RAG 直接对接知识库，最直接最权威的知识来源就是各种文件，包括 pdf、txt 等文本文件，或者 mp3、avi、img 等多模态文件，我们希望 RAG 能够直接接受各种模态的文件。

Pipelin 使用 MarkItDown 作为统一的文档转换引擎，支持几乎所有常见的文档格式，无论输入是 PDF、Word、Excel、图片还是音频，最终都会转换为标准的Markdown格式，然后进入统一的分块、向量化和存储流程。

def _convert_to_markdown(path: str) -> str:

函数返回一个存储了 md 格式原文的字符串。

智能分块

使用 MarkItDown 将文件转为 md 格式后，需要进行分块，可以利用Markdown的标题结构（#、##、###等）进行精确的语义分割。流程为：

标准Markdown文本 → 标题层次解析 → 段落语义分割 → Token计算分块 → 重叠策略优化 → 向量化准备
       ↓                ↓              ↓            ↓           ↓            ↓
   统一格式          #/##/###        语义边界      大小控制     信息连续性    嵌入向量
   结构清晰          层次识别        完整性保证    检索优化     上下文保持    相似度匹配W

首先，对完整 md 文本使用下列函数：

def _split_paragraphs_with_headings(text: str) -> List[Dict]:

这个函数能够完成 标题层次解析 和 段落语义分割，将完整的 md 文本转化为若干小块，分别代表局部完整的文本内容。

随后，对字典使用下列函数：

def _chunk_paragraphs(paragraphs: List[Dict], chunk_tokens: int, overlap_tokens: int) -> List[Dict]:

该函数进一步完成后三个步骤，将每个字典进一步处理为分块文本，以能够直接向量化。

统一嵌入与向量存储

经过前面两个阶段的处理后，我们终于将各种文件，通过各种处理，将其变得能够存入向量数据库，也即加入知识库了。

嵌入模型负责将文本转换为高维向量，使得计算机能够理解和比较文本的语义相似性。

下列函数负责将分块文本存入知识库：

def index_chunks(
    store = None, 
    chunks: List[Dict] = None, 
    cache_db: Optional[str] = None, 
    batch_size: int = 64,
    rag_namespace: str = "default"
) -> None:

Pipeline 高级扩展

上面的三大流程囊括了文件从输入到处理到存储的全过程，属于基础功能，而在此之外，官方的 pipeline 还提供了众多高级功能，这是为了解决很有可能出现的问题，官方将其描述为：

RAG系统的检索能力是其核心竞争力，在实际应用中，用户的查询表述与文档中的实际内容可能存在用词差异，导致相关文档无法被检索到。

官方提出了三种高级检索策略进行着重讲解：多查询扩展（MQE）、假设文档嵌入（HyDE）和统一的扩展检索框架。

这三种高级策略，有助于进一步提升 RAG 搜索的召回率。

多查询扩展 (Multi-Query Expansion, MQE)

该功能的核心思想是：同一个问题有多种不同的表述方式。我们知道 RAG 系统实际上并非直接进行逻辑判断，而是对 query 本身进行神经网络式的黑箱匹配，这就表示，同一个逻辑事件，采用多种表示方式，同时作为 query 进行查询，能够匹配出更多、更精确的结果。这就是 多查询扩展 。

思想很简单，实现上也很简单，多种方法表达让 LLM 进行就可以：

def _prompt_mqe(query: str, n: int) -> List[str]:
    """使用LLM生成多样化的查询扩展"""
    try:
        llm = LLM()
        prompt = [
            {"role": "system", "content": "你是检索查询扩展助手。生成语义等价或互补的多样化查询。使用中文，简短，避免标点。"},
            {"role": "user", "content": f"原始查询：{query}\n请给出{n}个不同表述的查询，每行一个。"}
        ]
        text = llm.think(prompt)
        lines = [ln.strip("- \t") for ln in (text or "").splitlines()]
        outs = [ln for ln in lines if ln]
        return outs[:n] or [query]
    except Exception:
        return [query]

假设文档嵌入（Hypothetical Document Embeddings ,HyDE）

这个方法的重要前提是，query 和 answer 之间存在一种现象，在语义空间中的分布往往存在差异 —— 问题通常是疑问句，而文档内容是陈述句。

这种现象带来一种问题，即从我们人类的思维上看，疑问对应解答，这是十分合理且理所应当的，但是在 RAG 系统中，疑问句与陈述句之间的语义特征几乎完全不同。

疑问句的向量，很大一部分维度被句法结构占据，在 RAG 看来：

疑问句 = 疑问词（什么/如何/为什么/what/how/why）+ 倒装语序（英文）+ 极短的长度 + 低占比的术语密度

而陈述句则几乎全部维度被语义内容占据，在 RAG 看来：

陈述句 = 少量语法结构词 + 高占比的术语密度

如果我们要搜索真正的术语，那么疑问句中的噪音未免过多，因为疑问句和陈述句完全是两个物种，即 句法差异贡献了主要方差。

而实际上，在 RAG 看来，疑问句并非直接用来存储知识库，反而是另一个陈述句的指针，比如疑问句 什么是机器学习？ 反而是在引导系统向另一个陈述句 机器学习是人工智能的一个分支，它使计算机系统能够从数据中... 联想。

所以，如果利用 LLM 生成一个假想的陈述句回答（重点是陈述而非解答），例如 LLM 通过问题 什么是机器学习？ 生成了解答 机器学习是人工智能的一个分支，它使计算机系统能够从数据中...，我们再通过 rag 搜索就很有可能搜索到相关度高的解答。

def _prompt_hyde(query: str) -> Optional[str]:
    """生成假设性文档用于改善检索"""
    try:
        llm = LLM()
        prompt = [
            {"role": "system", "content": "根据用户问题，先写一段可能的答案性段落，用于向量检索的查询文档（不要分析过程）。"},
            {"role": "user", "content": f"问题：{query}\n请直接写一段中等长度、客观、包含关键术语的段落。"}
        ]
        return llm.think(prompt)
    except Exception:
        return None

不过一个问题是，让 LLM 生成假想答案，如果问题是 什么是机器学习？ 这种在网络中存在客观解答的问题，LLM 确实可以很简单地生成假想答案（这个答案甚至有可能比 RAG 知识库中还要准确）；但如果是 我家的门牌号是多少？ 这种网络知识库不可能知道的问题， LLM 应该生成什么？

LLM 很有可能生成三种回答：

• 1.胡说八道，例如 “住宅门牌号是用于邮政投递和地址识别的数字标识系统，通常由当地政府统一规划分配，门牌号的编码规则因地区而异，常见的有顺序编号和距离编号两种方式…”

• 2.拒绝回答，例如 “我无法知道您家的具体门牌号，因为这属于个人隐私信息。”

• 3.歧义回复，例如 “用户家的门牌号是 128 号，位于社区的核心区域。”

如果的第一种，看似牛头不对马嘴，实则反而会其效果，因为解答中存在 “住宅门牌号”、”邮政投递”、”码规则” 等和门牌号高度相关的名词，如果 RAG 中确实存在 用户家的门牌号是 xxx 号 这个权威结果，反而会提升搜索到的概率。

如果是第二种安全性拒绝，那就无功无过，至少未产生有害影响。

如果是第三种歧义回复，很有可能造成有害影响。我家的门牌号是多少？ 这种问题还不是最狠的，更狠的是确实存在复数含义的名词，例如 Java 是什么？，LLM 极大概率生成 “Java 是一种广泛使用的编程语言，由 Sun Microsystems 于 1995 年发布…”，然而在 RAG 中，”Java 是隶属于印度尼西亚的一个岛屿。”，那么就会造成极其深刻的有害影响。

扩展检索框架

结合 MQE 和 HyDE 的搜索强化策略，通过 enable_mqe 和 enable_hyde 参数让用户可以根据具体场景选择启用哪些策略：对于需要高召回率的场景可以同时启用两种策略，对于性能敏感的场景可以只使用基础检索。

先使用 MQE 拓展为多个同义疑问句，再对每个疑问句采用 HyDE 改写为陈述句，最后合并排序返回最相关的 top-k 文档。

def search_vectors_expanded(
    store = None,
    query: str = "",
    top_k: int = 8,
    rag_namespace: Optional[str] = None,
    only_rag_data: bool = True,
    score_threshold: Optional[float] = None,
    enable_mqe: bool = False,
    mqe_expansions: int = 2,
    enable_hyde: bool = False,
    candidate_pool_multiplier: int = 4,
) -> List[Dict]:

tool_action

RAGTool 需要注解 tool_action：

from typing import Callable
def tool_action(name: str = None, description: str = None):
    """装饰器：标记一个方法为可展开的工具 action

    用法:
        @tool_action("memory_add", "添加新记忆")
        def _add_memory(self, content: str, importance: float = 0.5) -> str:
            '''添加记忆

            Args:
                content: 记忆内容
                importance: 重要性分数
            '''
            ...

    Args:
        name: 工具名称（如果不提供，从方法名自动生成）
        description: 工具描述（如果不提供，从 docstring 提取）
    """
    def decorator(func: Callable):
        func._is_tool_action = True
        func._tool_name = name
        func._tool_description = description
        return func
    return decorator

RAGTool

RAGTool 是对外工具，其设计与 memoryTool 一样，使用统一接口输入操作符调用，官方演示：

# 添加第一个知识
result1 = rag_tool.run({
    "action":"add_text", 
    "text":"Python是一种高级编程语言，由Guido van Rossum于1991年首次发布。Python的设计哲学强调代码的可读性和简洁的语法。",
    "document_id":"python_intro"
    })

result = rag_tool.run({"action":"search",
    "query":"Python编程语言的历史",
    "limit":3,
    "min_score":0.1
})

我们通过 ragTool 的 run 函数，给定一个字典进行 RAG 操作，自然，run 函数就是统一接口和学习重点。

run 函数调用体系

def run(self, parameters: Dict[str, Any]) -> str:
    """执行工具（非展开模式）
    Args:
        parameters: 工具参数字典，必须包含action参数
    Returns:
        执行结果字符串
    """
    if not self.validate_parameters(parameters):
        return "❌ 参数验证失败：缺少必需的参数"
    if not self.initialized:
        return f"❌ RAG工具未正确初始化，请检查配置: {getattr(self, 'init_error', '未知错误')}
    action = parameters.get("action")

    # 根据action调用对应的方法，传入提取的参数
    try:
        if action == "add_document":
            return self._add_document(
                file_path=parameters.get("file_path"),
                document_id=parameters.get("document_id"),
                namespace=parameters.get("namespace", "default"),
                chunk_size=parameters.get("chunk_size", 800),
                chunk_overlap=parameters.get("chunk_overlap", 100)
            )
        elif action == "add_text":
            return self._add_text(
                text=parameters.get("text"),

        # ······

        else:
            return f"❌ 不支持的操作: {action}"
    except Exception as e:
        return f"❌ 执行操作 '{action}' 时发生错误: {str(e)}"

可以看到，run 函数内解析参数 action 后，就是将其路由到各自的函数，我们以 _add_document 为例，看看里面是什么：

@tool_action("rag_add_document", "添加文档到知识库（支持PDF、Word、Excel、PPT、图片、音频等多种格式）")
def _add_document(
    self,
    file_path: str,
    document_id: str = None,
    namespace: str = "default",
    chunk_size: int = 800,
    chunk_overlap: int = 100
) -> str:
    """添加文档到知识库

    Args:
        file_path: 文档文件路径
        document_id: 文档ID（可选）
        namespace: 知识库命名空间（用于隔离不同项目）
        chunk_size: 分块大小
        chunk_overlap: 分块重叠大小

    Returns:
        执行结果
    """
    try:
        if not file_path or not os.path.exists(file_path):
            return f"❌ 文件不存在: {file_path}"
        
        pipeline = self._get_pipeline(namespace)
        t0 = time.time()

        chunks_added = pipeline["add_documents"](
            file_paths=[file_path],
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap
        )
        
        t1 = time.time()
        process_ms = int((t1 - t0) * 1000)
        
        if chunks_added == 0:
            return f"⚠️ 未能从文件解析内容: {os.path.basename(file_path)}"
        
        return (
            f"✅ 文档已添加到知识库: {os.path.basename(file_path)}\n"
            f"📊 分块数量: {chunks_added}\n"
            f"⏱️ 处理时间: {process_ms}ms\n"
            f"📝 命名空间: {pipeline.get('namespace', self.rag_namespace)}"
        )
        
    except Exception as e:
        return f"❌ 添加文档失败: {str(e)}"

这里终于用到我们之前详细讲述的 pipeline 了。前面提到，pipeline 本质是一个闭包字典，字典所有的值都是对底层数据库的 crud，pipeline["add_documents"] 实际上就是一个函数，因为 pipeline 内部的一个字典是：

"add_documents": add_documents,

add_documents 是 pipeline 内部的一个函数，直接操作底层数据库了。

RAGTool 如何拓展操作

前面提到，pipeline 的字典在大多数时候是固定的，即：

{
    "store": store,
    "namespace": rag_namespace,
    "add_documents": add_documents,
    "search": search,
    "search_advanced": search_advanced,
    "get_stats": get_stats
}

我们上面提到的 RAGTool 函数 _add_document 当然直接由 pipeline:add_documents 代理了。但是，我们同时也注意到 RAGTool 里面同时存在同样为添加数据的函数 _add_text，而 pipeline 内部根本没有直接的代理函数，那么这个时候怎么办？

让我们看一看 _add_text 函数：

@tool_action("rag_add_text", "添加文本到知识库")
def _add_text(
    self,
    text: str,
    document_id: str = None,
    namespace: str = "default",
    chunk_size: int = 800,
    chunk_overlap: int = 100
) -> str:
    """添加文本到知识库

    Args:
        text: 要添加的文本内容
        document_id: 文档ID（可选）
        namespace: 知识库命名空间
        chunk_size: 分块大小
        chunk_overlap: 分块重叠大小

    Returns:
        执行结果
    """
    metadata = None
    try:
        if not text or not text.strip():
            return "❌ 文本内容不能为空"
        
        # 创建临时文件
        document_id = document_id or f"text_{abs(hash(text)) % 100000}"
        tmp_path = os.path.join(self.knowledge_base_path, f"{document_id}.md")
        
        try:
            with open(tmp_path, 'w', encoding='utf-8') as f:
                f.write(text)
            
            pipeline = self._get_pipeline(namespace)
            t0 = time.time()

            chunks_added = pipeline["add_documents"](
                file_paths=[tmp_path],
                chunk_size=chunk_size,
                chunk_overlap=chunk_overlap
            )
            
            # ······

注意到，用的居然还是 add_documents。

不仅如此，让我们看看 RAGTool 里面的两个搜索函数 _search 和 _ask：

@tool_action("rag_search", "搜索知识库中的相关内容")
def _search(
    self,
    query: str,
    limit: int = 5,
    min_score: float = 0.1,
    enable_advanced_search: bool = True,
    max_chars: int = 1200,
    include_citations: bool = True,
    namespace: str = "default"
) -> str:
    try:
        if not query or not query.strip():
            return "❌ 搜索查询不能为空"
        
        # 使用统一 RAG 管道搜索
        pipeline = self._get_pipeline(namespace)

        if enable_advanced_search:
            results = pipeline["search_advanced"](
                query=query,
                top_k=limit,
                enable_mqe=True,
                enable_hyde=True,
                score_threshold=min_score if min_score > 0 else None
            )
        else:
            results = pipeline["search"](
                query=query,
                top_k=limit,
                score_threshold=min_score if min_score > 0 else None
            )

        # ······

@tool_action("rag_ask", "基于知识库进行智能问答")
def _ask(
    self,
    question: str,
    limit: int = 5,
    enable_advanced_search: bool = True,
    include_citations: bool = True,
    max_chars: int = 1200,
    namespace: str = "default"
) -> str:
    try:
        # 验证问题
        if not question or not question.strip():
            return "❌ 请提供要询问的问题"

        user_question = question.strip()
        print(f"🔍 智能问答: {user_question}")
        
        # 1. 检索相关内容
        pipeline = self._get_pipeline(namespace)
        search_start = time.time()
        
        if enable_advanced_search:
            results = pipeline["search_advanced"](
                query=user_question,
                top_k=limit,
                enable_mqe=True,
                enable_hyde=True
            )
        else:
            results = pipeline["search"](
                query=user_question,
                top_k=limit
            )
        # ······

我们会发现，两个函数调用的均是 pipeline 的 search_advanced 和 search ，只是在调用顺序、后续处理的环节上存在较大差异。

所以我们会发现，pipeline 只负责 CRUD，只暴露出少量单一职责接口；而 RAGTool 负责基于单一职责接口进行功能拓展。

总结

本章将重心放在了 Agent 记忆构建，与记忆检索上，先后讲述了 hello agent 的记忆体系、记忆如何持久化、如何从持久化底层检索数据。

针对记忆体系，我们遵照了自下而上的构建原则，首先完成了记忆实体类和配置类；随后在此之上构建了抽象类 Memory，定义了记忆应该有的行为，并在此之上构建了三大记忆之一的工作记忆；而避免外部接口 MemoryTool 和底层记忆类 Memory 的直接接触，我们设计了 MemoryManager，供 MemoryTool 调用从而代理记忆管理。

在记忆体系构建完毕后，我们转向了持久化的设计，我们基于已有的 Qdrant、Neo4j、sqlite 三大数据库和 embedding 嵌入模型的库函数，设计了 storage 包，并完成了情景记忆与语义记忆这两个需要将记忆持久化的记忆类型，初步建立了 RAG 底层支持。

在存储体系趋于完善后，我们设计了 rag 包，其设计直接对标 memory 的工具体系，自下而上地构建了工具和底层直接的代理组件 pipeline 和直接对外暴露的统一 RAG 操作工具 RAGTool。我们首先学习了 pipeline 实现的主要职责与流程，其负责将文件统一转为文本格式，并通过分块、向量化等操作将文件存入向量数据库，初步实现 RAG 功能；随后我们还学习了官方介绍的两种提升 RAG 搜索精度的方法：MQE 和 HyDE；最后，我们详细学习了 RAGTool 的 run 函数设计，通过对同一个函数输入不同的操作符与参数，实现了一个函数统一路由到其他功能，也即仅对外暴露一个接口便实现全部功能的目的。

综上，我们通过构建记忆与检索体系，将 agent 的能力进一步增强。然而，记忆与搜索也仅仅是对 agent 的特定能力进行了增强，我们需要研究如何让 agent 以最正确、最及时、最有性价比地进行记忆与检索，这一点将在下一章进行。