LLM应用开发：基于Anansi构建结构化记忆API实现智能对话助手

在实际 LLM 应用开发中，一个常被忽视但至关重要的组件是“记忆”系统。无论是构建一个能记住对话历史的聊天机器人，还是一个能根据用户历史行为提供个性化建议的智能助手，都需要一个可靠、高效且可扩展的机制来存储、检索和关联上下文信息。直接依赖 LLM 模型自身的上下文窗口不仅成本高昂，而且容量有限，无法满足长期、复杂的交互需求。因此，一个专门为 LLM 应用设计的记忆 API 成为了连接单次请求与长期智能的关键桥梁。

Anansi 正是这样一个开源项目，它旨在为 LLM 应用提供一个结构化的记忆 API。你可以将其理解为一个专为 AI 应用设计的“记忆中枢”，它负责处理诸如对话历史、用户偏好、会话状态、知识片段等信息的存储与检索。与简单地使用数据库或缓存不同，Anansi 的设计考虑了 LLM 应用的特有模式，例如基于向量相似度的语义检索、会话的隔离与关联、记忆的时效性管理等。本文将带你从零开始，理解 Anansi 的核心概念，并将其集成到一个简单的 LLM 应用中，实现一个具备记忆能力的对话助手。

1. 理解 LLM 应用中的“记忆”问题与 Anansi 的定位

在深入代码之前，我们需要先厘清几个关键概念：为什么 LLM 需要外部记忆？Anansi 解决了哪些具体问题？

1.1 为什么需要外部记忆 API？

LLM 模型本身具有强大的上下文理解能力，但其上下文窗口（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或通过git+https://...直接安装。

2.2 项目结构设计

一个清晰的项目结构有助于管理代码。创建如下文件和目录：

llm-memory-demo/ ├── .env # 存储环境变量，如 API Keys ├── requirements.txt # 依赖列表 ├── src/ │ ├── __init__.py │ ├── memory_manager.py # 封装 Anansi 记忆操作的核心类 │ └── chat_agent.py # 集成记忆的聊天代理逻辑 └── demo.py # 主程序入口

将之前安装的依赖写入requirements.txt：

openai 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( path=persist_directory, settings=Settings(anonymized_telemetry=False) ) # 获取或创建一个用于存储记忆的集合（Collection） # 集合名可以按用户或会话区分，这里使用全局集合做演示 self.collection_name = "conversation_memories" try: self.collection = self.chroma_client.get_collection(name=self.collection_name) except Exception: # 如果集合不存在，则创建它。我们定义向量维度为384（all-MiniLM-L6-v2的维度） self.collection = self.chroma_client.create_collection( name=self.collection_name, metadata={"description": "Storage for LLM conversation memories"}, embedding_function=self._get_embedding # 自定义嵌入函数 ) print(f"Memory 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_numpy=True) 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(f"Memory 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_results=n_results, where=where_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(key=lambda x: x['relevance_score'], reverse=True) return memories

关键点解释：

1. 嵌入模型：我们使用SentenceTransformer生成文本的向量表示。这是语义搜索的基础。选择all-MiniLM-L6-v2是因为它在速度和效果上取得了很好的平衡。
2. 向量数据库：Chroma 是一个轻量级、嵌入式的向量数据库，非常适合本地开发和演示。它负责存储向量、元数据，并执行高效的相似度搜索。
3. 存储逻辑：store_memory方法将记忆内容、关联的会话/用户 ID 以及其他元数据一起存储。documents字段存储原始文本，metadatas存储关联信息，ids是唯一标识。
4. 检索逻辑：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_KEY=your_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_key=os.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( query=user_input, session_id=self.current_session_id, user_id=self.current_user_id, n_results=2 # 每次检索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( model=self.model, messages=messages, temperature=0.7, max_tokens=500 ) 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_id=self.current_session_id, user_id=self.current_user_id, content=f"User said: {user_input}", metadata={"type": "user_message"} ) # 存储助手回复（关键记忆点） self.memory_manager.store_memory( session_id=self.current_session_id, user_id=self.current_user_id, content=f"Assistant replied: {assistant_reply}", metadata={"type": "assistant_response"} ) return assistant_reply

关键点解释：

1. 提示词工程：_build_prompt_with_memory函数是核心。它构建了一个结构化的消息列表：
   • 系统消息：定义助手角色并说明记忆的使用方式。
   • 记忆上下文：将检索到的相关记忆以清晰格式插入为另一条系统消息。这比直接拼接更清晰，有助于模型区分指令和上下文。
   • 用户消息：最后是当前用户输入。
2. 记忆检索与注入：在每次对话前，都会根据当前用户输入检索相关记忆。检索时限定在当前会话和用户，确保了记忆的隔离性。检索到的记忆被格式化后注入提示词。
3. 记忆的存储：在获得 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_manager=memory_manager, model="gpt-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 性能问题

现象：对话响应变慢，尤其是随着记忆条目增多。可能原因：

1. 向量搜索变慢：当记忆条目数达到十万、百万级别时，线性搜索会非常慢。
2. 提示词过长：检索到的记忆过多，导致提示词 token 数激增，增加 LLM API 成本和延迟。

解决方案：

• 索引优化：使用支持高效近似最近邻搜索（ANN）的向量数据库，如 Weaviate、Pinecone、Qdrant。它们为大规模向量检索做了优化。
• 记忆分片：不要将所有记忆放在一个集合中。可以按用户、时间（如按月）进行分片。
• 记忆摘要与淘汰：定期对旧记忆进行摘要合并，或根据时间、访问频率淘汰不重要的记忆。实现一个MemoryCompactor类来处理。
• 限制检索数量：严格控制n_results，只注入最相关的几条记忆。

6.3 生产环境部署考量

上述演示代码适用于学习和原型开发。在生产环境中，你需要考虑更多：

1. 安全性：

   • API Key 管理：使用专业的密钥管理服务，而非.env文件。
   • 数据隔离：确保session_id和user_id来自可信的身份验证系统，防止串号。
   • 输入输出过滤：对存储和检索的内容进行必要的敏感信息过滤和审查。

2. 可靠性：

   • 错误处理：为记忆存储和检索操作添加重试机制、降级策略（如检索失败时使用空记忆）。
   • 监控与日志：记录记忆操作的耗时、检索结果数量、LLM 调用情况，便于排查问题。
   • 数据持久化：确保向量数据库有备份和恢复方案。

3. 架构扩展：

   • 服务化：将MemoryManager封装成独立的 gRPC 或 RESTful 微服务，供多个 LLM 应用调用。
   • 缓存：对高频或固定的记忆（如用户画像）加入缓存层（如 Redis），减少向量搜索压力。
   • 异步操作：记忆的存储操作可以异步执行，不阻塞主对话流程。

7. 最佳实践与扩展方向

7.1 记忆管理的最佳实践

1. 定义清晰的记忆模式：在项目开始时就规划好要存储什么、以什么格式存储。例如，区分“事实型记忆”、“对话型记忆”、“用户偏好记忆”。
2. 实施分层记忆：
   • 短期记忆：存储在会话上下文中，用于维持当前对话的连贯性。
   • 长期记忆：存储在向量数据库中，用于跨会话的语义检索。
   • 工作记忆：在单次推理过程中临时维护的信息。
3. 主动总结与压缩：不要存储原始的、冗长的对话。定期（或在对话结束时）使用 LLM 对一段时间的对话进行总结，将总结作为新的、更精炼的记忆存储起来。
4. 设置记忆的 TTL 和重要性衰减：为记忆条目设置生存时间或重要性衰减函数，让不活跃、不重要的记忆逐渐“被遗忘”。

7.2 扩展 Anansi 的功能

本文实现的MemoryManager是一个简化版。一个完整的 Anansi 类系统可能还包括以下功能，你可以尝试实现：

• 记忆链（Memory Chaining）：将相关的记忆条目链接起来，形成事件链或故事线。
• 记忆反射（Reflection）：让 LLM 定期回顾记忆，发现模式、矛盾或新的见解，并生成更高级别的“元记忆”。
• 多模态记忆：支持存储和检索图像、音频等非文本信息的嵌入表示。
• 记忆的版本控制：当记忆被更新时，保留历史版本。

7.3 与其他 LLM 框架集成

你可以将这套记忆机制集成到更成熟的 LLM 应用框架中，例如：

• LangChain：实现一个自定义的Memory类，并将其接入ConversationChain。
• LlamaIndex：将记忆系统作为一个额外的“上下文检索器”，与已有的索引器结合使用。
• Semantic Kernel：将记忆管理封装为一个插件（Plugin），供内核调用。

通过将记忆能力模块化、服务化，你可以使其成为任何 LLM 应用项目中一个稳固的基础设施组件，从而轻松构建出真正具有长期记忆和个性化能力的智能体。