在实际 LLM 应用开发中一个常被忽视但至关重要的组件是“记忆”系统。无论是构建一个能记住对话历史的聊天机器人还是一个能根据用户历史行为提供个性化建议的智能助手都需要一个可靠、高效且可扩展的机制来存储、检索和关联上下文信息。直接依赖 LLM 模型自身的上下文窗口不仅成本高昂而且容量有限无法满足长期、复杂的交互需求。因此一个专门为 LLM 应用设计的记忆 API 成为了连接单次请求与长期智能的关键桥梁。Anansi 正是这样一个开源项目它旨在为 LLM 应用提供一个结构化的记忆 API。你可以将其理解为一个专为 AI 应用设计的“记忆中枢”它负责处理诸如对话历史、用户偏好、会话状态、知识片段等信息的存储与检索。与简单地使用数据库或缓存不同Anansi 的设计考虑了 LLM 应用的特有模式例如基于向量相似度的语义检索、会话的隔离与关联、记忆的时效性管理等。本文将带你从零开始理解 Anansi 的核心概念并将其集成到一个简单的 LLM 应用中实现一个具备记忆能力的对话助手。1. 理解 LLM 应用中的“记忆”问题与 Anansi 的定位在深入代码之前我们需要先厘清几个关键概念为什么 LLM 需要外部记忆Anansi 解决了哪些具体问题1.1 为什么需要外部记忆 APILLM 模型本身具有强大的上下文理解能力但其上下文窗口Context Window是有限的。无论是 4K、16K 还是 128K tokens将所有的历史对话、用户资料、产品知识都塞进每一次的提示词Prompt中既不经济也不现实。这会导致成本激增输入模型的 tokens 越多API 调用费用越高。性能下降过长的上下文可能影响模型处理核心问题的能力甚至导致关键信息被“淹没”。无法实现长期记忆模型无法在多次独立的会话或请求之间保持信息的连续性。因此一个外部的记忆系统负责长期存储信息并在每次与 LLM 交互时智能地选取最相关的片段注入上下文这成为了构建复杂 LLM 应用的标配。1.2 Anansi 的核心功能与组件Anansi 作为一个记忆 API其设计通常围绕以下几个核心功能展开基于开源项目常见模式推断记忆的存储Storage提供后端存储接口支持内存、数据库如 Redis、PostgreSQL或向量数据库如 Pinecone、Weaviate来持久化记忆单元。记忆的检索Retrieval这是核心。不仅支持按 ID、会话等键值查询更关键的是支持语义检索。即根据当前查询的语义从记忆库中找到最相关的历史记忆。这通常依赖嵌入模型Embedding Model将文本转换为向量并通过向量相似度搜索实现。记忆的组织Organization记忆不是杂乱无章的。Anansi 需要提供组织记忆的逻辑单元例如会话Session一次完整的交互周期包含多轮对话。用户User跨会话的长期用户档案和偏好。记忆条目Memory Item存储的具体内容可能包含文本、元数据如时间戳、重要性分数、标签和向量表示。记忆的更新与衰减记忆不是一成不变的。新的信息需要合并旧的无用信息需要被淘汰或衰减其重要性。1.3 Anansi 在应用架构中的位置在一个典型的 LLM 应用架构中Anansi 扮演着“记忆层”的角色。[用户请求] - [应用逻辑层] - [记忆层 (Anansi)] - [检索相关记忆] | [LLM API (如 OpenAI)] - [构建增强后的Prompt] - [合并记忆与当前查询]应用逻辑层收到用户请求后会先调用 Anansi 的记忆检索功能获取与当前请求相关的历史信息。然后将这些记忆作为上下文与用户的当前问题一起构建成最终的提示词发送给 LLM API。LLM 的回复在返回给用户之前也可能被选择性地存储回 Anansi形成新的记忆。2. 环境准备与项目初始化我们将使用 Python 来演示如何集成 Anansi。首先需要搭建基础的开发环境。2.1 创建项目并安装核心依赖假设你已经安装了 Python (建议 3.8)。我们创建一个新的项目目录并初始化虚拟环境。# 创建项目目录 mkdir llm-memory-demo cd llm-memory-demo # 创建虚拟环境 (以 venv 为例) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate接下来安装核心包。由于 Anansi 是一个相对较新的开源项目其具体的 PyPI 包名可能需要根据其官方文档确定。这里我们假设它可以通过pip install anansi-ai安装。同时我们还需要 LLM 客户端和向量数据库/嵌入模型相关的库。# 安装假设的 Anansi 包、OpenAI 客户端和 Sentence Transformers (用于本地嵌入模型) pip install anansi-ai openai sentence-transformers # 为了简单演示我们使用 Chroma 作为轻量级向量数据库 pip install chromadb注意anansi-ai是一个假设的包名。在实际操作中你需要根据 Anansi 项目的官方仓库如 GitHub的说明来安装正确的包。可能是pip install anansi或通过githttps://...直接安装。2.2 项目结构设计一个清晰的项目结构有助于管理代码。创建如下文件和目录llm-memory-demo/ ├── .env # 存储环境变量如 API Keys ├── requirements.txt # 依赖列表 ├── src/ │ ├── __init__.py │ ├── memory_manager.py # 封装 Anansi 记忆操作的核心类 │ └── chat_agent.py # 集成记忆的聊天代理逻辑 └── demo.py # 主程序入口将之前安装的依赖写入requirements.txtopenai sentence-transformers chromadb # anansi-ai # 请替换为实际的包名和版本3. 构建记忆管理器封装 Anansi 核心操作我们将首先创建一个MemoryManager类它负责与 Anansi 交互包括初始化客户端、存储记忆和检索记忆。3.1 初始化 Anansi 客户端与记忆后端在src/memory_manager.py中我们开始编写代码。首先需要根据 Anansi 的 SDK 文档来初始化客户端。这里我们基于常见模式进行假设性实现。# src/memory_manager.py import os from typing import List, Dict, Any, Optional from sentence_transformers import SentenceTransformer import chromadb from chromadb.config import Settings class MemoryManager: 记忆管理器封装 Anansi 或类似记忆系统的核心操作。 本示例使用 Chroma 作为向量存储SentenceTransformer 生成嵌入模拟 Anansi 的语义检索功能。 def __init__(self, persist_directory: str ./chroma_db): 初始化记忆系统。 :param persist_directory: Chroma 向量数据库持久化目录 # 初始化嵌入模型用于将文本转换为向量 # 选择一个轻量且高效的模型 self.embedding_model SentenceTransformer(all-MiniLM-L6-v2) # 初始化 Chroma 客户端作为向量存储后端 self.chroma_client chromadb.PersistentClient( pathpersist_directory, settingsSettings(anonymized_telemetryFalse) ) # 获取或创建一个用于存储记忆的集合Collection # 集合名可以按用户或会话区分这里使用全局集合做演示 self.collection_name conversation_memories try: self.collection self.chroma_client.get_collection(nameself.collection_name) except Exception: # 如果集合不存在则创建它。我们定义向量维度为384all-MiniLM-L6-v2的维度 self.collection self.chroma_client.create_collection( nameself.collection_name, metadata{description: Storage for LLM conversation memories}, embedding_functionself._get_embedding # 自定义嵌入函数 ) print(fMemory Manager initialized. Collection {self.collection_name} is ready.) def _get_embedding(self, texts: List[str]) - List[List[float]]: 内部方法将文本列表转换为向量列表。 # SentenceTransformer 已经返回 numpy array需要转换为 list of lists embeddings self.embedding_model.encode(texts, convert_to_numpyTrue) return embeddings.tolist() def store_memory(self, session_id: str, user_id: str, content: str, metadata: Optional[Dict] None): 存储一段记忆。 :param session_id: 会话ID用于关联同一对话中的记忆 :param user_id: 用户ID用于关联同一用户的记忆 :param content: 记忆的文本内容 :param metadata: 额外的元数据如时间戳、类型等 if metadata is None: metadata {} # 完善元数据 metadata.update({ session_id: session_id, user_id: user_id, timestamp: metadata.get(timestamp, default_time) }) # 生成一个唯一的ID在实际项目中可能需要更复杂的逻辑 memory_id f{session_id}_{user_id}_{len(self.collection.get()[ids])} # 存储到向量数据库 # Chroma 会自动调用我们设置的 _get_embedding 函数来为 content 生成向量 self.collection.add( documents[content], metadatas[metadata], ids[memory_id] ) print(fMemory stored. ID: {memory_id}, Content: {content[:50]}...) def retrieve_related_memories(self, query: str, session_id: str None, user_id: str None, n_results: int 3) - List[Dict]: 检索与查询相关的记忆。 优先检索同一会话或同一用户的记忆并基于语义相似度。 :param query: 查询文本 :param session_id: 可选限定检索特定会话的记忆 :param user_id: 可选限定检索特定用户的记忆 :param n_results: 返回最相关的记忆条数 :return: 包含记忆内容、元数据和相似度得分的字典列表 # 构建查询过滤器 where_filter {} if session_id: where_filter[session_id] session_id if user_id: where_filter[user_id] user_id # 执行查询 # Chroma 的 query 方法会计算查询文本与集合中文档的相似度 results self.collection.query( query_texts[query], n_resultsn_results, wherewhere_filter if where_filter else None, # 应用过滤器 include[documents, metadatas, distances] # 返回文档、元数据和距离 ) # 整理返回结果 memories [] if results[documents]: for i in range(len(results[documents][0])): memory { content: results[documents][0][i], metadata: results[metadatas][0][i], # 距离越小越相似我们将其转换为一个简单的相关性分数非精确 relevance_score: 1 - (results[distances][0][i] / 2.0) if results[distances] else 0.5 } memories.append(memory) # 按相关性分数降序排序 memories.sort(keylambda x: x[relevance_score], reverseTrue) return memories关键点解释嵌入模型我们使用SentenceTransformer生成文本的向量表示。这是语义搜索的基础。选择all-MiniLM-L6-v2是因为它在速度和效果上取得了很好的平衡。向量数据库Chroma 是一个轻量级、嵌入式的向量数据库非常适合本地开发和演示。它负责存储向量、元数据并执行高效的相似度搜索。存储逻辑store_memory方法将记忆内容、关联的会话/用户 ID 以及其他元数据一起存储。documents字段存储原始文本metadatas存储关联信息ids是唯一标识。检索逻辑retrieve_related_memories是核心。它接受一个查询文本并返回最相关的记忆。where参数允许我们过滤特定会话或用户的记忆这是实现记忆隔离的关键。返回的distances是向量空间的距离我们将其粗略地转换为相关性分数。3.2 设计记忆的数据结构一个记忆条目通常包含以下信息id: 唯一标识符。content: 记忆的文本内容例如用户说的一句话或 LLM 的一个关键回复。embedding: 内容的向量表示由向量数据库管理。metadata: 元数据字典至少包含session_id: 所属会话。user_id: 所属用户。timestamp: 创建时间。type: 记忆类型如user_message,assistant_response,user_preference。importance: 主观重要性评分可选。在我们的示例中这些字段通过 Chroma 的documents、metadatas和内部向量存储来管理。4. 集成记忆系统到 LLM 聊天代理接下来我们创建一个ChatAgent类它使用 OpenAI API 作为 LLM并利用上面构建的MemoryManager来实现有记忆的对话。4.1 配置 OpenAI 客户端与构建提示词首先在项目根目录创建.env文件存放你的 OpenAI API Key。# .env OPENAI_API_KEYyour_openai_api_key_here然后编写src/chat_agent.py# src/chat_agent.py import os from openai import OpenAI from typing import List, Dict from dotenv import load_dotenv from src.memory_manager import MemoryManager # 加载环境变量 load_dotenv() class ChatAgent: def __init__(self, memory_manager: MemoryManager, model: str gpt-3.5-turbo): 初始化聊天代理。 :param memory_manager: 记忆管理器实例 :param model: 使用的 OpenAI 模型 self.memory_manager memory_manager self.model model self.client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) if not self.client.api_key: raise ValueError(OPENAI_API_KEY not found in environment variables.) # 当前会话和用户简化处理实际应从请求中获取 self.current_session_id session_001 self.current_user_id user_123 def _build_prompt_with_memory(self, user_input: str, related_memories: List[Dict]) - List[Dict]: 构建包含系统指令、相关记忆和当前对话的提示词列表。 :param user_input: 用户当前输入 :param related_memories: 检索到的相关记忆 :return: 符合 OpenAI ChatCompletion 格式的消息列表 messages [] # 1. 系统指令定义助手角色和记忆使用方式 system_message { role: system, content: 你是一个有帮助的助手并且拥有与当前用户对话的记忆。以下是一些可能与当前对话相关的历史信息记忆。请利用这些信息来更好地理解用户的需求和上下文提供连贯且个性化的回复。如果记忆不相关请忽略它。 } messages.append(system_message) # 2. 注入相关记忆作为上下文 if related_memories: memory_context 相关记忆\n for mem in related_memories: # 这里简单拼接更复杂的实现可以格式化时间、来源等 memory_context f- {mem[content]} (相关性: {mem[relevance_score]:.2f})\n messages.append({role: system, content: memory_context}) # 3. 添加当前用户输入 messages.append({role: user, content: user_input}) return messages def chat(self, user_input: str) - str: 处理一轮对话检索记忆 - 构建提示词 - 调用 LLM - 存储新记忆。 :param user_input: 用户输入 :return: 助手的回复 print(f\n[User]: {user_input}) # 步骤1检索相关记忆 related_mems self.memory_manager.retrieve_related_memories( queryuser_input, session_idself.current_session_id, user_idself.current_user_id, n_results2 # 每次检索2条最相关的记忆 ) print(f[Agent] Retrieved {len(related_mems)} related memories.) # 步骤2构建包含记忆的提示词 messages self._build_prompt_with_memory(user_input, related_mems) # 步骤3调用 OpenAI API try: response self.client.chat.completions.create( modelself.model, messagesmessages, temperature0.7, max_tokens500 ) assistant_reply response.choices[0].message.content except Exception as e: assistant_reply f抱歉我遇到了一些问题{e} print(f[Assistant]: {assistant_reply}) # 步骤4将本轮交互的关键信息存储为记忆 # 存储用户输入可选取决于业务逻辑 self.memory_manager.store_memory( session_idself.current_session_id, user_idself.current_user_id, contentfUser said: {user_input}, metadata{type: user_message} ) # 存储助手回复关键记忆点 self.memory_manager.store_memory( session_idself.current_session_id, user_idself.current_user_id, contentfAssistant replied: {assistant_reply}, metadata{type: assistant_response} ) return assistant_reply关键点解释提示词工程_build_prompt_with_memory函数是核心。它构建了一个结构化的消息列表系统消息定义助手角色并说明记忆的使用方式。记忆上下文将检索到的相关记忆以清晰格式插入为另一条系统消息。这比直接拼接更清晰有助于模型区分指令和上下文。用户消息最后是当前用户输入。记忆检索与注入在每次对话前都会根据当前用户输入检索相关记忆。检索时限定在当前会话和用户确保了记忆的隔离性。检索到的记忆被格式化后注入提示词。记忆的存储在获得 LLM 回复后我们将本轮对话的用户输入和助手回复都存储为新的记忆。这实现了记忆的累积。在实际应用中你可能需要更精细的策略例如只存储重要的、总结性的信息而不是每一句话以避免记忆膨胀。4.2 编写主程序进行演示最后创建demo.py来运行一个简单的对话循环。# demo.py from src.memory_manager import MemoryManager from src.chat_agent import ChatAgent def main(): print(初始化记忆系统和聊天代理...) # 初始化记忆管理器 memory_manager MemoryManager(persist_directory./chroma_db_demo) # 初始化聊天代理并传入记忆管理器 agent ChatAgent(memory_managermemory_manager, modelgpt-3.5-turbo) print(\n 具备记忆的 LLM 聊天演示开始 ) print(输入 quit 或 exit 结束对话。) print(- * 40) # 简单对话循环 while True: try: user_input input(\nYou: ).strip() if user_input.lower() in [quit, exit, q]: print(对话结束。) break if not user_input: continue # 调用代理获取回复 reply agent.chat(user_input) # 回复已在 agent.chat 中打印这里可以不做额外操作 except KeyboardInterrupt: print(\n\n程序被中断。) break except Exception as e: print(f\n发生错误: {e}) break if __name__ __main__: main()5. 运行验证与结果分析现在让我们运行程序并观察记忆系统如何工作。5.1 启动与首次对话在终端运行python demo.py程序会初始化记忆数据库和聊天代理。首次运行时向量数据库集合是空的。进行如下对话You: 你好我叫小明。 [Agent] Retrieved 0 related memories. [Assistant]: 你好小明很高兴认识你。有什么我可以帮助你的吗此时记忆库中存储了两条新记忆“User said: 你好我叫小明。” 和 “Assistant replied: 你好小明...”。5.2 测试记忆检索与利用进行第二轮对话You: 你还记得我的名字吗此时retrieve_related_memories会以“你还记得我的名字吗”为查询文本进行语义搜索。由于上一轮对话中提到了“我叫小明”这条记忆的向量与当前查询在语义上高度相关因此会被检索出来。控制台可能会显示[Agent] Retrieved 1 related memories.LLM 收到的提示词中会包含类似内容相关记忆 - User said: 你好我叫小明。 (相关性: 0.85)因此LLM 能够基于此记忆给出回复[Assistant]: 当然记得小明你刚才告诉我你的名字了。有什么其他事情需要我帮忙吗这证明了记忆系统成功检索并利用了历史信息。5.3 验证记忆的隔离性如果我们模拟另一个用户user_456或另一个会话记忆管理器通过session_id和user_id进行过滤可以确保用户 A 无法看到用户 B 的记忆实现了基本的数据隔离。你可以在ChatAgent初始化时修改current_user_id来测试。6. 常见问题排查与优化在实际集成和使用过程中你可能会遇到以下问题。6.1 记忆检索不相关或效果差现象LLM 的回复似乎没有利用到历史信息或者引用了无关的记忆。可能原因与解决方案问题现象常见原因检查与解决方式检索到的记忆完全不相关1. 嵌入模型不适合你的文本领域。2. 查询文本太短或太模糊。3. 向量数据库的相似度算法或参数问题。1.更换嵌入模型尝试paraphrase-multilingual-MiniLM-L12-v2或专门针对你领域微调的模型。2.优化查询尝试对用户输入进行简单的重写或扩展或使用更长的上下文作为查询。3.检查检索参数调整n_results返回数量和where过滤器确保范围正确。记忆没有被存储store_memory方法未被调用或调用失败。1. 检查代码逻辑确保在需要存储记忆的地方调用了该方法。2. 查看向量数据库的日志或检查集合中是否有新数据。记忆内容质量低存储了过多噪音或无关信息如“你好”、“谢谢”。实施记忆过滤与总结在存储前可以用一个简单的规则或另一个 LLM 调用来判断信息是否值得存储或对长对话进行总结后再存储。6.2 性能问题现象对话响应变慢尤其是随着记忆条目增多。可能原因向量搜索变慢当记忆条目数达到十万、百万级别时线性搜索会非常慢。提示词过长检索到的记忆过多导致提示词 token 数激增增加 LLM API 成本和延迟。解决方案索引优化使用支持高效近似最近邻搜索ANN的向量数据库如 Weaviate、Pinecone、Qdrant。它们为大规模向量检索做了优化。记忆分片不要将所有记忆放在一个集合中。可以按用户、时间如按月进行分片。记忆摘要与淘汰定期对旧记忆进行摘要合并或根据时间、访问频率淘汰不重要的记忆。实现一个MemoryCompactor类来处理。限制检索数量严格控制n_results只注入最相关的几条记忆。6.3 生产环境部署考量上述演示代码适用于学习和原型开发。在生产环境中你需要考虑更多安全性API Key 管理使用专业的密钥管理服务而非.env文件。数据隔离确保session_id和user_id来自可信的身份验证系统防止串号。输入输出过滤对存储和检索的内容进行必要的敏感信息过滤和审查。可靠性错误处理为记忆存储和检索操作添加重试机制、降级策略如检索失败时使用空记忆。监控与日志记录记忆操作的耗时、检索结果数量、LLM 调用情况便于排查问题。数据持久化确保向量数据库有备份和恢复方案。架构扩展服务化将MemoryManager封装成独立的 gRPC 或 RESTful 微服务供多个 LLM 应用调用。缓存对高频或固定的记忆如用户画像加入缓存层如 Redis减少向量搜索压力。异步操作记忆的存储操作可以异步执行不阻塞主对话流程。7. 最佳实践与扩展方向7.1 记忆管理的最佳实践定义清晰的记忆模式在项目开始时就规划好要存储什么、以什么格式存储。例如区分“事实型记忆”、“对话型记忆”、“用户偏好记忆”。实施分层记忆短期记忆存储在会话上下文中用于维持当前对话的连贯性。长期记忆存储在向量数据库中用于跨会话的语义检索。工作记忆在单次推理过程中临时维护的信息。主动总结与压缩不要存储原始的、冗长的对话。定期或在对话结束时使用 LLM 对一段时间的对话进行总结将总结作为新的、更精炼的记忆存储起来。设置记忆的 TTL 和重要性衰减为记忆条目设置生存时间或重要性衰减函数让不活跃、不重要的记忆逐渐“被遗忘”。7.2 扩展 Anansi 的功能本文实现的MemoryManager是一个简化版。一个完整的 Anansi 类系统可能还包括以下功能你可以尝试实现记忆链Memory Chaining将相关的记忆条目链接起来形成事件链或故事线。记忆反射Reflection让 LLM 定期回顾记忆发现模式、矛盾或新的见解并生成更高级别的“元记忆”。多模态记忆支持存储和检索图像、音频等非文本信息的嵌入表示。记忆的版本控制当记忆被更新时保留历史版本。7.3 与其他 LLM 框架集成你可以将这套记忆机制集成到更成熟的 LLM 应用框架中例如LangChain实现一个自定义的Memory类并将其接入ConversationChain。LlamaIndex将记忆系统作为一个额外的“上下文检索器”与已有的索引器结合使用。Semantic Kernel将记忆管理封装为一个插件Plugin供内核调用。通过将记忆能力模块化、服务化你可以使其成为任何 LLM 应用项目中一个稳固的基础设施组件从而轻松构建出真正具有长期记忆和个性化能力的智能体。 LLM应用开发：基于Anansi构建结构化记忆API实现智能对话助手
在实际 LLM 应用开发中一个常被忽视但至关重要的组件是“记忆”系统。无论是构建一个能记住对话历史的聊天机器人还是一个能根据用户历史行为提供个性化建议的智能助手都需要一个可靠、高效且可扩展的机制来存储、检索和关联上下文信息。直接依赖 LLM 模型自身的上下文窗口不仅成本高昂而且容量有限无法满足长期、复杂的交互需求。因此一个专门为 LLM 应用设计的记忆 API 成为了连接单次请求与长期智能的关键桥梁。Anansi 正是这样一个开源项目它旨在为 LLM 应用提供一个结构化的记忆 API。你可以将其理解为一个专为 AI 应用设计的“记忆中枢”它负责处理诸如对话历史、用户偏好、会话状态、知识片段等信息的存储与检索。与简单地使用数据库或缓存不同Anansi 的设计考虑了 LLM 应用的特有模式例如基于向量相似度的语义检索、会话的隔离与关联、记忆的时效性管理等。本文将带你从零开始理解 Anansi 的核心概念并将其集成到一个简单的 LLM 应用中实现一个具备记忆能力的对话助手。1. 理解 LLM 应用中的“记忆”问题与 Anansi 的定位在深入代码之前我们需要先厘清几个关键概念为什么 LLM 需要外部记忆Anansi 解决了哪些具体问题1.1 为什么需要外部记忆 APILLM 模型本身具有强大的上下文理解能力但其上下文窗口Context Window是有限的。无论是 4K、16K 还是 128K tokens将所有的历史对话、用户资料、产品知识都塞进每一次的提示词Prompt中既不经济也不现实。这会导致成本激增输入模型的 tokens 越多API 调用费用越高。性能下降过长的上下文可能影响模型处理核心问题的能力甚至导致关键信息被“淹没”。无法实现长期记忆模型无法在多次独立的会话或请求之间保持信息的连续性。因此一个外部的记忆系统负责长期存储信息并在每次与 LLM 交互时智能地选取最相关的片段注入上下文这成为了构建复杂 LLM 应用的标配。1.2 Anansi 的核心功能与组件Anansi 作为一个记忆 API其设计通常围绕以下几个核心功能展开基于开源项目常见模式推断记忆的存储Storage提供后端存储接口支持内存、数据库如 Redis、PostgreSQL或向量数据库如 Pinecone、Weaviate来持久化记忆单元。记忆的检索Retrieval这是核心。不仅支持按 ID、会话等键值查询更关键的是支持语义检索。即根据当前查询的语义从记忆库中找到最相关的历史记忆。这通常依赖嵌入模型Embedding Model将文本转换为向量并通过向量相似度搜索实现。记忆的组织Organization记忆不是杂乱无章的。Anansi 需要提供组织记忆的逻辑单元例如会话Session一次完整的交互周期包含多轮对话。用户User跨会话的长期用户档案和偏好。记忆条目Memory Item存储的具体内容可能包含文本、元数据如时间戳、重要性分数、标签和向量表示。记忆的更新与衰减记忆不是一成不变的。新的信息需要合并旧的无用信息需要被淘汰或衰减其重要性。1.3 Anansi 在应用架构中的位置在一个典型的 LLM 应用架构中Anansi 扮演着“记忆层”的角色。[用户请求] - [应用逻辑层] - [记忆层 (Anansi)] - [检索相关记忆] | [LLM API (如 OpenAI)] - [构建增强后的Prompt] - [合并记忆与当前查询]应用逻辑层收到用户请求后会先调用 Anansi 的记忆检索功能获取与当前请求相关的历史信息。然后将这些记忆作为上下文与用户的当前问题一起构建成最终的提示词发送给 LLM API。LLM 的回复在返回给用户之前也可能被选择性地存储回 Anansi形成新的记忆。2. 环境准备与项目初始化我们将使用 Python 来演示如何集成 Anansi。首先需要搭建基础的开发环境。2.1 创建项目并安装核心依赖假设你已经安装了 Python (建议 3.8)。我们创建一个新的项目目录并初始化虚拟环境。# 创建项目目录 mkdir llm-memory-demo cd llm-memory-demo # 创建虚拟环境 (以 venv 为例) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate接下来安装核心包。由于 Anansi 是一个相对较新的开源项目其具体的 PyPI 包名可能需要根据其官方文档确定。这里我们假设它可以通过pip install anansi-ai安装。同时我们还需要 LLM 客户端和向量数据库/嵌入模型相关的库。# 安装假设的 Anansi 包、OpenAI 客户端和 Sentence Transformers (用于本地嵌入模型) pip install anansi-ai openai sentence-transformers # 为了简单演示我们使用 Chroma 作为轻量级向量数据库 pip install chromadb注意anansi-ai是一个假设的包名。在实际操作中你需要根据 Anansi 项目的官方仓库如 GitHub的说明来安装正确的包。可能是pip install anansi或通过githttps://...直接安装。2.2 项目结构设计一个清晰的项目结构有助于管理代码。创建如下文件和目录llm-memory-demo/ ├── .env # 存储环境变量如 API Keys ├── requirements.txt # 依赖列表 ├── src/ │ ├── __init__.py │ ├── memory_manager.py # 封装 Anansi 记忆操作的核心类 │ └── chat_agent.py # 集成记忆的聊天代理逻辑 └── demo.py # 主程序入口将之前安装的依赖写入requirements.txtopenai sentence-transformers chromadb # anansi-ai # 请替换为实际的包名和版本3. 构建记忆管理器封装 Anansi 核心操作我们将首先创建一个MemoryManager类它负责与 Anansi 交互包括初始化客户端、存储记忆和检索记忆。3.1 初始化 Anansi 客户端与记忆后端在src/memory_manager.py中我们开始编写代码。首先需要根据 Anansi 的 SDK 文档来初始化客户端。这里我们基于常见模式进行假设性实现。# src/memory_manager.py import os from typing import List, Dict, Any, Optional from sentence_transformers import SentenceTransformer import chromadb from chromadb.config import Settings class MemoryManager: 记忆管理器封装 Anansi 或类似记忆系统的核心操作。 本示例使用 Chroma 作为向量存储SentenceTransformer 生成嵌入模拟 Anansi 的语义检索功能。 def __init__(self, persist_directory: str ./chroma_db): 初始化记忆系统。 :param persist_directory: Chroma 向量数据库持久化目录 # 初始化嵌入模型用于将文本转换为向量 # 选择一个轻量且高效的模型 self.embedding_model SentenceTransformer(all-MiniLM-L6-v2) # 初始化 Chroma 客户端作为向量存储后端 self.chroma_client chromadb.PersistentClient( pathpersist_directory, settingsSettings(anonymized_telemetryFalse) ) # 获取或创建一个用于存储记忆的集合Collection # 集合名可以按用户或会话区分这里使用全局集合做演示 self.collection_name conversation_memories try: self.collection self.chroma_client.get_collection(nameself.collection_name) except Exception: # 如果集合不存在则创建它。我们定义向量维度为384all-MiniLM-L6-v2的维度 self.collection self.chroma_client.create_collection( nameself.collection_name, metadata{description: Storage for LLM conversation memories}, embedding_functionself._get_embedding # 自定义嵌入函数 ) print(fMemory Manager initialized. Collection {self.collection_name} is ready.) def _get_embedding(self, texts: List[str]) - List[List[float]]: 内部方法将文本列表转换为向量列表。 # SentenceTransformer 已经返回 numpy array需要转换为 list of lists embeddings self.embedding_model.encode(texts, convert_to_numpyTrue) return embeddings.tolist() def store_memory(self, session_id: str, user_id: str, content: str, metadata: Optional[Dict] None): 存储一段记忆。 :param session_id: 会话ID用于关联同一对话中的记忆 :param user_id: 用户ID用于关联同一用户的记忆 :param content: 记忆的文本内容 :param metadata: 额外的元数据如时间戳、类型等 if metadata is None: metadata {} # 完善元数据 metadata.update({ session_id: session_id, user_id: user_id, timestamp: metadata.get(timestamp, default_time) }) # 生成一个唯一的ID在实际项目中可能需要更复杂的逻辑 memory_id f{session_id}_{user_id}_{len(self.collection.get()[ids])} # 存储到向量数据库 # Chroma 会自动调用我们设置的 _get_embedding 函数来为 content 生成向量 self.collection.add( documents[content], metadatas[metadata], ids[memory_id] ) print(fMemory stored. ID: {memory_id}, Content: {content[:50]}...) def retrieve_related_memories(self, query: str, session_id: str None, user_id: str None, n_results: int 3) - List[Dict]: 检索与查询相关的记忆。 优先检索同一会话或同一用户的记忆并基于语义相似度。 :param query: 查询文本 :param session_id: 可选限定检索特定会话的记忆 :param user_id: 可选限定检索特定用户的记忆 :param n_results: 返回最相关的记忆条数 :return: 包含记忆内容、元数据和相似度得分的字典列表 # 构建查询过滤器 where_filter {} if session_id: where_filter[session_id] session_id if user_id: where_filter[user_id] user_id # 执行查询 # Chroma 的 query 方法会计算查询文本与集合中文档的相似度 results self.collection.query( query_texts[query], n_resultsn_results, wherewhere_filter if where_filter else None, # 应用过滤器 include[documents, metadatas, distances] # 返回文档、元数据和距离 ) # 整理返回结果 memories [] if results[documents]: for i in range(len(results[documents][0])): memory { content: results[documents][0][i], metadata: results[metadatas][0][i], # 距离越小越相似我们将其转换为一个简单的相关性分数非精确 relevance_score: 1 - (results[distances][0][i] / 2.0) if results[distances] else 0.5 } memories.append(memory) # 按相关性分数降序排序 memories.sort(keylambda x: x[relevance_score], reverseTrue) return memories关键点解释嵌入模型我们使用SentenceTransformer生成文本的向量表示。这是语义搜索的基础。选择all-MiniLM-L6-v2是因为它在速度和效果上取得了很好的平衡。向量数据库Chroma 是一个轻量级、嵌入式的向量数据库非常适合本地开发和演示。它负责存储向量、元数据并执行高效的相似度搜索。存储逻辑store_memory方法将记忆内容、关联的会话/用户 ID 以及其他元数据一起存储。documents字段存储原始文本metadatas存储关联信息ids是唯一标识。检索逻辑retrieve_related_memories是核心。它接受一个查询文本并返回最相关的记忆。where参数允许我们过滤特定会话或用户的记忆这是实现记忆隔离的关键。返回的distances是向量空间的距离我们将其粗略地转换为相关性分数。3.2 设计记忆的数据结构一个记忆条目通常包含以下信息id: 唯一标识符。content: 记忆的文本内容例如用户说的一句话或 LLM 的一个关键回复。embedding: 内容的向量表示由向量数据库管理。metadata: 元数据字典至少包含session_id: 所属会话。user_id: 所属用户。timestamp: 创建时间。type: 记忆类型如user_message,assistant_response,user_preference。importance: 主观重要性评分可选。在我们的示例中这些字段通过 Chroma 的documents、metadatas和内部向量存储来管理。4. 集成记忆系统到 LLM 聊天代理接下来我们创建一个ChatAgent类它使用 OpenAI API 作为 LLM并利用上面构建的MemoryManager来实现有记忆的对话。4.1 配置 OpenAI 客户端与构建提示词首先在项目根目录创建.env文件存放你的 OpenAI API Key。# .env OPENAI_API_KEYyour_openai_api_key_here然后编写src/chat_agent.py# src/chat_agent.py import os from openai import OpenAI from typing import List, Dict from dotenv import load_dotenv from src.memory_manager import MemoryManager # 加载环境变量 load_dotenv() class ChatAgent: def __init__(self, memory_manager: MemoryManager, model: str gpt-3.5-turbo): 初始化聊天代理。 :param memory_manager: 记忆管理器实例 :param model: 使用的 OpenAI 模型 self.memory_manager memory_manager self.model model self.client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) if not self.client.api_key: raise ValueError(OPENAI_API_KEY not found in environment variables.) # 当前会话和用户简化处理实际应从请求中获取 self.current_session_id session_001 self.current_user_id user_123 def _build_prompt_with_memory(self, user_input: str, related_memories: List[Dict]) - List[Dict]: 构建包含系统指令、相关记忆和当前对话的提示词列表。 :param user_input: 用户当前输入 :param related_memories: 检索到的相关记忆 :return: 符合 OpenAI ChatCompletion 格式的消息列表 messages [] # 1. 系统指令定义助手角色和记忆使用方式 system_message { role: system, content: 你是一个有帮助的助手并且拥有与当前用户对话的记忆。以下是一些可能与当前对话相关的历史信息记忆。请利用这些信息来更好地理解用户的需求和上下文提供连贯且个性化的回复。如果记忆不相关请忽略它。 } messages.append(system_message) # 2. 注入相关记忆作为上下文 if related_memories: memory_context 相关记忆\n for mem in related_memories: # 这里简单拼接更复杂的实现可以格式化时间、来源等 memory_context f- {mem[content]} (相关性: {mem[relevance_score]:.2f})\n messages.append({role: system, content: memory_context}) # 3. 添加当前用户输入 messages.append({role: user, content: user_input}) return messages def chat(self, user_input: str) - str: 处理一轮对话检索记忆 - 构建提示词 - 调用 LLM - 存储新记忆。 :param user_input: 用户输入 :return: 助手的回复 print(f\n[User]: {user_input}) # 步骤1检索相关记忆 related_mems self.memory_manager.retrieve_related_memories( queryuser_input, session_idself.current_session_id, user_idself.current_user_id, n_results2 # 每次检索2条最相关的记忆 ) print(f[Agent] Retrieved {len(related_mems)} related memories.) # 步骤2构建包含记忆的提示词 messages self._build_prompt_with_memory(user_input, related_mems) # 步骤3调用 OpenAI API try: response self.client.chat.completions.create( modelself.model, messagesmessages, temperature0.7, max_tokens500 ) assistant_reply response.choices[0].message.content except Exception as e: assistant_reply f抱歉我遇到了一些问题{e} print(f[Assistant]: {assistant_reply}) # 步骤4将本轮交互的关键信息存储为记忆 # 存储用户输入可选取决于业务逻辑 self.memory_manager.store_memory( session_idself.current_session_id, user_idself.current_user_id, contentfUser said: {user_input}, metadata{type: user_message} ) # 存储助手回复关键记忆点 self.memory_manager.store_memory( session_idself.current_session_id, user_idself.current_user_id, contentfAssistant replied: {assistant_reply}, metadata{type: assistant_response} ) return assistant_reply关键点解释提示词工程_build_prompt_with_memory函数是核心。它构建了一个结构化的消息列表系统消息定义助手角色并说明记忆的使用方式。记忆上下文将检索到的相关记忆以清晰格式插入为另一条系统消息。这比直接拼接更清晰有助于模型区分指令和上下文。用户消息最后是当前用户输入。记忆检索与注入在每次对话前都会根据当前用户输入检索相关记忆。检索时限定在当前会话和用户确保了记忆的隔离性。检索到的记忆被格式化后注入提示词。记忆的存储在获得 LLM 回复后我们将本轮对话的用户输入和助手回复都存储为新的记忆。这实现了记忆的累积。在实际应用中你可能需要更精细的策略例如只存储重要的、总结性的信息而不是每一句话以避免记忆膨胀。4.2 编写主程序进行演示最后创建demo.py来运行一个简单的对话循环。# demo.py from src.memory_manager import MemoryManager from src.chat_agent import ChatAgent def main(): print(初始化记忆系统和聊天代理...) # 初始化记忆管理器 memory_manager MemoryManager(persist_directory./chroma_db_demo) # 初始化聊天代理并传入记忆管理器 agent ChatAgent(memory_managermemory_manager, modelgpt-3.5-turbo) print(\n 具备记忆的 LLM 聊天演示开始 ) print(输入 quit 或 exit 结束对话。) print(- * 40) # 简单对话循环 while True: try: user_input input(\nYou: ).strip() if user_input.lower() in [quit, exit, q]: print(对话结束。) break if not user_input: continue # 调用代理获取回复 reply agent.chat(user_input) # 回复已在 agent.chat 中打印这里可以不做额外操作 except KeyboardInterrupt: print(\n\n程序被中断。) break except Exception as e: print(f\n发生错误: {e}) break if __name__ __main__: main()5. 运行验证与结果分析现在让我们运行程序并观察记忆系统如何工作。5.1 启动与首次对话在终端运行python demo.py程序会初始化记忆数据库和聊天代理。首次运行时向量数据库集合是空的。进行如下对话You: 你好我叫小明。 [Agent] Retrieved 0 related memories. [Assistant]: 你好小明很高兴认识你。有什么我可以帮助你的吗此时记忆库中存储了两条新记忆“User said: 你好我叫小明。” 和 “Assistant replied: 你好小明...”。5.2 测试记忆检索与利用进行第二轮对话You: 你还记得我的名字吗此时retrieve_related_memories会以“你还记得我的名字吗”为查询文本进行语义搜索。由于上一轮对话中提到了“我叫小明”这条记忆的向量与当前查询在语义上高度相关因此会被检索出来。控制台可能会显示[Agent] Retrieved 1 related memories.LLM 收到的提示词中会包含类似内容相关记忆 - User said: 你好我叫小明。 (相关性: 0.85)因此LLM 能够基于此记忆给出回复[Assistant]: 当然记得小明你刚才告诉我你的名字了。有什么其他事情需要我帮忙吗这证明了记忆系统成功检索并利用了历史信息。5.3 验证记忆的隔离性如果我们模拟另一个用户user_456或另一个会话记忆管理器通过session_id和user_id进行过滤可以确保用户 A 无法看到用户 B 的记忆实现了基本的数据隔离。你可以在ChatAgent初始化时修改current_user_id来测试。6. 常见问题排查与优化在实际集成和使用过程中你可能会遇到以下问题。6.1 记忆检索不相关或效果差现象LLM 的回复似乎没有利用到历史信息或者引用了无关的记忆。可能原因与解决方案问题现象常见原因检查与解决方式检索到的记忆完全不相关1. 嵌入模型不适合你的文本领域。2. 查询文本太短或太模糊。3. 向量数据库的相似度算法或参数问题。1.更换嵌入模型尝试paraphrase-multilingual-MiniLM-L12-v2或专门针对你领域微调的模型。2.优化查询尝试对用户输入进行简单的重写或扩展或使用更长的上下文作为查询。3.检查检索参数调整n_results返回数量和where过滤器确保范围正确。记忆没有被存储store_memory方法未被调用或调用失败。1. 检查代码逻辑确保在需要存储记忆的地方调用了该方法。2. 查看向量数据库的日志或检查集合中是否有新数据。记忆内容质量低存储了过多噪音或无关信息如“你好”、“谢谢”。实施记忆过滤与总结在存储前可以用一个简单的规则或另一个 LLM 调用来判断信息是否值得存储或对长对话进行总结后再存储。6.2 性能问题现象对话响应变慢尤其是随着记忆条目增多。可能原因向量搜索变慢当记忆条目数达到十万、百万级别时线性搜索会非常慢。提示词过长检索到的记忆过多导致提示词 token 数激增增加 LLM API 成本和延迟。解决方案索引优化使用支持高效近似最近邻搜索ANN的向量数据库如 Weaviate、Pinecone、Qdrant。它们为大规模向量检索做了优化。记忆分片不要将所有记忆放在一个集合中。可以按用户、时间如按月进行分片。记忆摘要与淘汰定期对旧记忆进行摘要合并或根据时间、访问频率淘汰不重要的记忆。实现一个MemoryCompactor类来处理。限制检索数量严格控制n_results只注入最相关的几条记忆。6.3 生产环境部署考量上述演示代码适用于学习和原型开发。在生产环境中你需要考虑更多安全性API Key 管理使用专业的密钥管理服务而非.env文件。数据隔离确保session_id和user_id来自可信的身份验证系统防止串号。输入输出过滤对存储和检索的内容进行必要的敏感信息过滤和审查。可靠性错误处理为记忆存储和检索操作添加重试机制、降级策略如检索失败时使用空记忆。监控与日志记录记忆操作的耗时、检索结果数量、LLM 调用情况便于排查问题。数据持久化确保向量数据库有备份和恢复方案。架构扩展服务化将MemoryManager封装成独立的 gRPC 或 RESTful 微服务供多个 LLM 应用调用。缓存对高频或固定的记忆如用户画像加入缓存层如 Redis减少向量搜索压力。异步操作记忆的存储操作可以异步执行不阻塞主对话流程。7. 最佳实践与扩展方向7.1 记忆管理的最佳实践定义清晰的记忆模式在项目开始时就规划好要存储什么、以什么格式存储。例如区分“事实型记忆”、“对话型记忆”、“用户偏好记忆”。实施分层记忆短期记忆存储在会话上下文中用于维持当前对话的连贯性。长期记忆存储在向量数据库中用于跨会话的语义检索。工作记忆在单次推理过程中临时维护的信息。主动总结与压缩不要存储原始的、冗长的对话。定期或在对话结束时使用 LLM 对一段时间的对话进行总结将总结作为新的、更精炼的记忆存储起来。设置记忆的 TTL 和重要性衰减为记忆条目设置生存时间或重要性衰减函数让不活跃、不重要的记忆逐渐“被遗忘”。7.2 扩展 Anansi 的功能本文实现的MemoryManager是一个简化版。一个完整的 Anansi 类系统可能还包括以下功能你可以尝试实现记忆链Memory Chaining将相关的记忆条目链接起来形成事件链或故事线。记忆反射Reflection让 LLM 定期回顾记忆发现模式、矛盾或新的见解并生成更高级别的“元记忆”。多模态记忆支持存储和检索图像、音频等非文本信息的嵌入表示。记忆的版本控制当记忆被更新时保留历史版本。7.3 与其他 LLM 框架集成你可以将这套记忆机制集成到更成熟的 LLM 应用框架中例如LangChain实现一个自定义的Memory类并将其接入ConversationChain。LlamaIndex将记忆系统作为一个额外的“上下文检索器”与已有的索引器结合使用。Semantic Kernel将记忆管理封装为一个插件Plugin供内核调用。通过将记忆能力模块化、服务化你可以使其成为任何 LLM 应用项目中一个稳固的基础设施组件从而轻松构建出真正具有长期记忆和个性化能力的智能体。