动手实践：集成向量数据库记忆

理论提供根本，但实际操作巩固理解。本节将指导您如何构建并为一个代理集成一个使用向量 (vector)数据库的持久性长期记忆模块。我们将不再停留在之前讨论的向量存储器的理解层面，而是实现基于语义相似性存储和检索信息的核心机制。这个动手练习对于创建能够保持上下文 (context)、从过往交互中学习并在较长时间内访问相关知识的代理来说是必要的。

本实践使用开源嵌入 (embedding)式数据库ChromaDB，因为它易于在本地开发环境中设置。然而，这里体现的原理也广泛适用于其他向量数据库，如Pinecone、Weaviate或FAISS，只需进行相应的API调整。

前提条件

开始之前，请确保您已安装所需的库。您将需要一个大型语言模型（LLM）库（如langchain或llama-index，尽管我们在此将侧重于核心逻辑），向量 (vector)数据库客户端，以及一个用于生成嵌入 (embedding)向量的句子转换器库。

pip install chromadb sentence-transformers
# 可选：如果您想集成到langchain或llama-index框架中，请安装它们
# pip install langchain openai # 使用LangChain与OpenAI的例子

您还需要一个嵌入模型。我们将使用sentence-transformers库中的一个模型，它可以自动下载。

1. 设置记忆组件

我们的向量 (vector)数据库记忆系统的核心组件是嵌入 (embedding)函数和向量存储器本身。

初始化嵌入函数

首先，我们需要一种将文本转换为密集向量 (dense vector)表示（嵌入）的方法。sentence-transformers库提供了对各种预训练 (pre-training)模型的便捷访问。

from sentence_transformers import SentenceTransformer

# 加载预训练的嵌入模型
# 像'all-MiniLM-L6-v2'这样的模型效率高；'all-mpnet-base-v2'则提供了良好的平衡。
# 选择一个适合您的性能需求和资源限制的模型。
embedding_model_name = 'all-MiniLM-L6-v2'
embedding_function = SentenceTransformer(embedding_model_name)

# 例子：获取一段文本的嵌入向量
text_example = "This is a sample sentence for embedding."
vector_example = embedding_function.encode(text_example)
print(f"Embedding dimension: {len(vector_example)}")
# 输出：嵌入维度：384（对于all-MiniLM-L6-v2）

每次我们需要向记忆中添加文本或查询记忆时，都会使用这个embedding_function。

初始化向量存储器 (ChromaDB)

现在，让我们设置ChromaDB。我们将它配置为本地持久化，这意味着数据将保存到磁盘。

import chromadb
import uuid # 用于生成唯一ID

# 设置一个持久性ChromaDB客户端
# 数据将存储在'agent_memory_db'目录中
client = chromadb.PersistentClient(path="./agent_memory_db")

# 创建或获取一个集合（类似于关系数据库中的表）
# 我们可以通过模型名称隐式地关联我们的嵌入函数，或者稍后显式关联
# 为了简化，这里我们将在ChromaDB的直接集成之外处理嵌入，
# 但ChromaDB也可以直接管理嵌入。
collection_name = "agent_long_term_memory"
try:
    collection = client.get_collection(name=collection_name)
    print(f"Collection '{collection_name}' loaded.")
except Exception: # 在生产环境中，请替换为更具体的异常处理
    print(f"Creating collection '{collection_name}'...")
    # 注意：ChromaDB在创建时也可以直接接受一个embedding_function
    collection = client.create_collection(name=collection_name)
    print(f"Collection '{collection_name}' created.")

我们现在有了一个collection对象，它代表了代理的长期记忆存储。

2. 实现记忆操作

代理需要对其记忆执行两个主要操作：写入（添加信息）和读取（检索信息）。

添加记忆（写入操作）

当代理遇到新信息、完成任务或产生重要想法时，这些都应被存储。我们需要将文本嵌入 (embedding)并连同唯一ID和可能有用的元数据一起添加到集合中。

def add_memory(text_content: str, metadata: dict = None):
    """
    将一段文本内容添加到向量存储记忆中。

    参数:
        text_content: 要存储的字符串内容。
        metadata: 可选字典，包含元数据（例如，时间戳、来源）。
    """
    if not text_content:
        print("警告：尝试向记忆中添加空内容。")
        return

    # 为记忆条目生成唯一ID
    memory_id = str(uuid.uuid4())

    # 为文本内容生成嵌入向量
    embedding = embedding_function.encode(text_content).tolist() # 确保它是一个列表

    # 准备元数据 - 确保它可序列化并与ChromaDB兼容
    final_metadata = metadata if metadata else {}
    # 例子：如果未提供时间戳，则自动添加
    if 'timestamp' not in final_metadata:
        import datetime
        final_metadata['timestamp'] = datetime.datetime.utcnow().isoformat()

    # 将记忆添加到集合中
    try:
        collection.add(
            embeddings=[embedding],
            documents=[text_content],
            metadatas=[final_metadata],
            ids=[memory_id]
        )
        print(f"已添加记忆：ID={memory_id}, 内容='{text_content[:50]}...'")
    except Exception as e:
        print(f"添加记忆出错：{e}")

# --- 示例用法 ---
add_memory("用户询问了伦敦的天气。")
add_memory("代理确定天气多云，15°C。", metadata={"source": "weather_tool_api"})
add_memory("计划：1. 检查用户请求。 2. 查询天气API。 3. 格式化响应。")

检索记忆（读取操作）

当代理需要回忆相关信息时（例如，回答问题、辅助规划或维持上下文 (context)），它会查询向量 (vector)存储器。查询本身会被嵌入，数据库会返回语义上最相似的条目。

def retrieve_memories(query_text: str, n_results: int = 3, filter_metadata: dict = None):
    """
    根据语义相似性从向量存储器中检索相关记忆。

    参数:
        query_text: 用于搜索的文本查询。
        n_results: 要返回的相关记忆的最大数量。
        filter_metadata: 可选字典，用于根据元数据过滤记忆。

    返回:
        一个检索到的文档（记忆）列表。
    """
    if not query_text:
        print("警告：尝试使用空查询检索记忆。")
        return []

    # 为查询生成嵌入向量
    query_embedding = embedding_function.encode(query_text).tolist()

    # 如果请求元数据过滤，则准备where过滤器
    where_filter = filter_metadata if filter_metadata else {}

    try:
        results = collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results,
            where=where_filter, # 在此应用元数据过滤器
            include=['documents', 'metadatas', 'distances'] # 请求文档、元数据和相似性距离
        )

        # 提取并返回相关信息
        retrieved_docs = []
        if results and results.get('documents') and results.get('documents')[0]:
            print(f"已为查询'{query_text[:50]}...'检索到 {len(results['documents'][0])} 条记忆。")
            # 将文档与其元数据结合起来以提供上下文
            for doc, meta, dist in zip(results['documents'][0], results['metadatas'][0], results['distances'][0]):
                 retrieved_docs.append({
                     "content": doc,
                     "metadata": meta,
                     "similarity_score": 1 - dist # 将距离转换为相似度分数（越接近1越好）
                 })
            # 如有需要，按相似度分数（降序）排序，尽管Chroma通常会返回已排序的结果
            retrieved_docs.sort(key=lambda x: x['similarity_score'], reverse=True)
            return retrieved_docs
        else:
            print("未找到相关记忆。")
            return []

    except Exception as e:
        print(f"检索记忆出错：{e}")
        return []

# --- 示例用法 ---
print("\n--- 检索相关记忆 ---")
query = "天气查询是关于什么的？"
relevant_memories = retrieve_memories(query, n_results=2)

if relevant_memories:
    print(f"\n与查询: '{query}'最相关的记忆：")
    for i, mem in enumerate(relevant_memories):
        print(f"{i+1}. 分数: {mem['similarity_score']:.4f} | 内容: {mem['content']} | 元数据: {mem['metadata']}")

# 带元数据过滤的示例
print("\n--- 专门从天气工具检索记忆 ---")
tool_memories = retrieve_memories("天气信息", n_results=1, filter_metadata={"source": "weather_tool_api"})
if tool_memories:
    print(f"\n来自'weather_tool_api'的记忆：")
    for mem in tool_memories:
        print(f"- 分数: {mem['similarity_score']:.4f} | 内容: {mem['content']} | 元数据: {mem['metadata']}")

3. 将记忆集成到代理循环中

现在，让我们思考这些add_memory和retrieve_memories函数如何融入简化代理的操作周期。设想一个接收输入、思考（检索记忆、规划）、行动并观察结果的基础代理。

# --- 简化代理模拟 ---

def agent_step(user_input: str = None, previous_observation: str = None):
    """模拟代理使用记忆的单一步骤。"""
    print("\n--- 代理步骤 ---")

    # 1. 收集上下文（输入、过往观察）
    context = ""
    if user_input:
        context += f"User Input: {user_input}\n"
        add_memory(f"Received user input: {user_input}", {"type": "user_interaction"}) # 存储交互
    if previous_observation:
        context += f"Previous Observation: {previous_observation}\n"
        # 决定该观察是否值得长期存储
        if len(previous_observation) > 10: # 简单启发式
             add_memory(f"Observation: {previous_observation}", {"type": "agent_observation"})

    # 2. 检索相关记忆
    query_for_memory = f"Current context: {context} What should I recall?"
    # 实际中需要更复杂的查询生成
    retrieved = retrieve_memories(query_for_memory, n_results=3)
    memory_context = "\nRelevant Past Information:\n"
    if retrieved:
        for mem in retrieved:
            memory_context += f"- {mem['content']} (Timestamp: {mem['metadata'].get('timestamp', 'N/A')})\n"
    else:
        memory_context += "- None found.\n"

    # 3. 思考/规划（简化：只打印检索到的上下文）
    # 在真实代理中，此上下文将作为LLM提示的输入，用于推理/规划
    print(f"为LLM提供的上下文（输入 + 检索到的记忆）：\n{context}{memory_context}")
    simulated_llm_prompt = f"{context}{memory_context} \nGiven this, what is the next action?"
    print(f"模拟提示片段：\n{simulated_llm_prompt[:200]}...") # 显示潜在提示的一部分

    # 4. 行动（简化：占位符行动）
    action = "模拟行动：查询数据库获取'伦敦人口'"
    print(f"代理行动：{action}")
    # 执行行动...（省略）

    # 5. 观察（模拟行动结果）
    observation = "数据库结果：伦敦人口大约为900万。"
    print(f"代理观察：{observation}")

    # 返回观察结果以供下一步使用
    return observation

# --- 运行几个步骤 ---
observation = None
observation = agent_step(user_input="你好代理，告诉我关于伦敦的情况。", previous_observation=observation)
observation = agent_step(previous_observation=observation) # 代理根据之前的观察继续
observation = agent_step(user_input="我最初问了什么？", previous_observation=observation)

此模拟展示了基本流程：输入和观察结果可能会被添加到记忆中，并且会检索相关记忆以充实代理下一个决策或响应的上下文 (context)。

4. 数据流可视化

代理、嵌入 (embedding)模型和向量 (vector)数据库之间的交互可以如下所示：

使用向量数据库添加和检索记忆的数据流。代理使用嵌入模型将文本转换为向量以进行存储和查询。

考量与后续步骤

这个实际设置提供了一个功能性的长期记忆，但对于生产系统来说，有几个方面可以改进：

• 可伸缩性： ChromaDB的本地持久化对开发很有利，但对于大规模应用，请考虑托管的云向量 (vector)数据库（Pinecone、Weaviate云服务）或自托管的分布式方案。
• 嵌入 (embedding)模型选择： 嵌入向量的质量和维度明显影响检索相关性和性能。根据您的特定应用范围和任务尝试不同的模型。考虑在您的数据上微调 (fine-tuning)嵌入模型。
• 高级检索： 研究超过基本语义相似性的方法，例如用于多样性的最大边缘相关性（MMR），用于特异性的元数据过滤，或时间衰减相关性。这些内容已在本章前面讨论过。
• 记忆管理： 随着记忆的增长，实施策略来总结不那么相关的信息，整合相关条目，或归档旧数据，以管理存储成本和检索效率。这与“记忆整合和总结技巧”部分相关联。
• 错误处理： 对数据库操作和嵌入生成实施错误处理。
• 异步操作： 为了响应性，特别是在交互式代理中，异步执行数据库写入和可能的读取操作。

这个动手练习为您提供了为代理实现向量数据库记忆的核心技能。通过掌握这些技巧，您可以构建出表现出更强连贯性、从经验中学习并有效运用广泛知识库的代理。尝试不同的数据类型、元数据结构和检索参数 (parameter)，以根据代理的特定需求定制记忆系统。