尧图网站建设 
在实际 AI 应用开发中无论是构建聊天助手、智能客服还是自动化工作流一个核心的挑战是如何让 AI 模型记住并有效利用超出其单次处理能力的长对话或复杂任务历史。传统的做法是将所有历史记录一股脑地塞进提示词Prompt但这会迅速耗尽模型的上下文窗口Context Window导致成本飙升、响应变慢甚至因为信息过载而影响回答质量。更棘手的是在多轮对话或长期运行的 AI Agent 中如何让模型记住几天前甚至几周前的关键信息并能在需要时精准调用这构成了所谓的“工作上下文瓶颈”。Lindy 提出的“共享记忆”概念正是为了解决这一瓶颈。它不是简单地将上下文窗口从 8K 扩展到 32K 或 128K而是引入了一种更智能、更结构化的信息管理机制。你可以把它想象成一个为 AI 配备的“外部大脑”或“知识库”这个大脑专门负责存储、索引和检索那些对当前任务至关重要的长期或跨会话信息。当 AI 需要处理新请求时它不再需要读取全部历史而是从这个“共享记忆”中快速查询出最相关的片段与当前问题一起构成一个精简而高效的上下文从而在有限的窗口内做出更精准的决策。本文将深入探讨如何在实际项目中特别是面向 Slack 等协作平台的 AI Agent 开发中实现“共享记忆”机制。我们将从核心概念入手逐步构建一个最小可运行的共享记忆系统涵盖环境准备、数据结构设计、核心代码实现、与 Slack 的集成以及最关键的生产环境部署和问题排查。无论你是正在开发 AI 聊天机器人、自动化流程 Agent还是希望优化现有大模型应用的上下文管理这篇文章都将提供一套可落地的工程方案。1. 理解共享记忆从上下文瓶颈到智能检索在深入代码之前我们必须先厘清几个核心概念理解为什么简单的上下文扩展不足以解决问题以及共享记忆机制是如何工作的。1.1 上下文窗口的限制与成本大语言模型LLM的上下文窗口决定了它一次性能处理多少文本包括你的提示词和它的回复。虽然 Claude 3、GPT-4 等模型支持长达 128K 甚至更多的上下文但这带来了两个现实问题成本问题绝大多数 API 的计费是基于输入和输出的总 Token 数量。将大量无关的历史对话持续放入上下文意味着每一轮交互你都在为这些“可能用不上”的信息付费。性能与质量衰减即使模型理论上能处理长上下文但有研究表明当关键信息被淹没在大量文本中时模型的检索和推理能力会下降。它可能“记得”信息存在但无法精准定位和运用。因此无节制地扩大单次请求的上下文并非最佳实践。1.2 共享记忆的核心思想共享记忆的核心思想是“按需取用长期存储”。它将 AI 工作所需的知识分为两类工作记忆Working Memory即当前对话轮次的上下文是直接送给模型处理的 Prompt。它应该尽可能精简、相关。长期记忆Long-term Memory即共享记忆存储库保存了跨会话、跨任务的历史信息、用户偏好、项目详情、操作结果等。当一个新的用户请求到来时系统不会将整个长期记忆塞进工作记忆而是先执行一个“检索Retrieval”步骤根据当前请求的语义从长期记忆中找出最相关的若干条记录。然后只将这些检索到的相关记录连同当前的用户问题一起构成最终的工作记忆送给模型。这个过程极大地缓解了上下文窗口的压力。1.3 关键技术组件嵌入模型与向量数据库实现智能检索依赖两项关键技术文本嵌入模型Embedding Model如 OpenAI 的text-embedding-3-small、text-embedding-ada-002或开源的BGE、SentenceTransformers模型。它的作用是将一段文本如一条历史消息转换为一个高维度的向量一组数字。语义相似的文本其向量在空间中的距离也更近。向量数据库Vector Database如 Pinecone、Weaviate、Qdrant或集成了向量搜索的 PostgreSQLpgvector、RedisRedisVL。它专门用于高效存储这些向量并执行“近似最近邻ANN”搜索。当新查询到来时先将其转换为向量然后在数据库中快速找到与之最相似的向量所对应的原始文本。通过这两者结合我们就能实现基于语义的、而不仅仅是关键词匹配的智能信息检索这正是共享记忆的“智能”所在。2. 环境准备与项目结构我们将构建一个面向 Slack 的 AI 助手示例它能够记住与用户的历史对话并在后续交互中引用这些信息。技术栈选择 Python因为它拥有最丰富的 AI 开发生态。2.1 基础环境与依赖首先确保你的开发环境已就绪。我们使用 Python 3.9 和pip进行包管理。# 创建并进入项目目录 mkdir ai-slack-agent-with-memory cd ai-slack-agent-with-memory # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate # 创建 requirements.txt 文件并安装核心依赖以下是requirements.txt文件的内容包含了从网络通信到 AI 模型调用的全套依赖# 核心框架与HTTP fastapi0.104.1 uvicorn[standard]0.24.0 slack-sdk3.27.1 httpx0.25.1 # AI/ML 相关 openai1.3.0 langchain0.0.340 langchain-openai0.0.2 tiktoken0.5.1 # 向量数据库与存储 chromadb0.4.18 # 或者使用 pgvector 需要 psycopg2-binary # psycopg2-binary2.9.9 # 工具与工具 python-dotenv1.0.0 pydantic2.5.0 pydantic-settings2.1.0使用 pip 安装pip install -r requirements.txt2.2 关键服务配置与密钥管理本项目需要与多个外部服务交互务必妥善管理密钥。我们使用.env文件来存储敏感信息并通过python-dotenv加载。创建一个名为.env的文件在项目根目录内容如下请替换为你的实际密钥# OpenAI API 配置用于对话和生成嵌入向量 OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_API_BASEhttps://api.openai.com/v1 # 如果使用代理或特定端点 OPENAI_EMBEDDING_MODELtext-embedding-3-small OPENAI_CHAT_MODELgpt-3.5-turbo # 或 gpt-4, gpt-4-turbo-preview # Slack 应用配置 SLACK_BOT_TOKENxoxb-your-slack-bot-token SLACK_SIGNING_SECRETyour-slack-signing-secret SLACK_APP_TOKENxapp-your-slack-app-token # 向量数据库配置以ChromaDB本地运行为例生产环境需配置持久化路径或远程地址 CHROMA_PERSIST_DIRECTORY./chroma_db # 如果使用 Pinecone # PINECONE_API_KEYyour-pinecone-key # PINECONE_ENVIRONMENTyour-environment # PINECONE_INDEX_NAMEyour-index-name注意.env文件必须被添加到.gitignore中绝对不要提交到版本控制系统。生产环境中应使用环境变量、密钥管理服务如 AWS Secrets Manager, HashiCorp Vault或平台提供的配置管理功能。2.3 项目目录结构设计一个清晰的项目结构有助于代码管理和维护。建议按以下方式组织ai-slack-agent-with-memory/ ├── .env # 环境变量本地开发用不上传 ├── .gitignore ├── requirements.txt ├── README.md ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口Slack 事件接收 │ ├── config.py # 配置加载Pydantic Settings │ ├── memory/ # 共享记忆核心模块 │ │ ├── __init__.py │ │ ├── manager.py # 记忆管理器存储、检索、更新 │ │ ├── models.py # 记忆条目的数据模型Pydantic │ │ └── vector_store.py # 向量数据库封装层 │ ├── agents/ # AI Agent 逻辑 │ │ ├── __init__.py │ │ └── slack_agent.py # 处理 Slack 消息协调记忆与LLM │ ├── services/ # 外部服务客户端 │ │ ├── __init__.py │ │ ├── openai_client.py │ │ └── slack_client.py │ └── utils/ # 工具函数 │ ├── __init__.py │ └── helpers.py └── tests/ # 测试目录 ├── __init__.py └── test_memory.py这个结构将配置、记忆管理、Agent 逻辑、服务客户端和工具函数进行了分离符合单一职责原则便于测试和扩展。3. 构建共享记忆系统的核心模块共享记忆系统的核心是记忆管理器Memory Manager和底层的向量存储。我们首先实现这两个部分。3.1 定义记忆数据模型在app/memory/models.py中我们使用 Pydantic 定义记忆条目的结构。这确保了数据的类型安全并便于序列化。from datetime import datetime from typing import Optional, Dict, Any from pydantic import BaseModel, Field from uuid import uuid4 class MemoryItem(BaseModel): 表示一条存储在共享记忆中的信息条目。 id: str Field(default_factorylambda: str(uuid4())) content: str Field(..., description记忆的文本内容) embedding: Optional[list[float]] Field(None, description文本内容的向量表示) metadata: Dict[str, Any] Field(default_factorydict, description关联的元数据) created_at: datetime Field(default_factorydatetime.utcnow) last_accessed_at: Optional[datetime] Field(None) access_count: int Field(default0) class Config: # 允许使用非Pydantic类型如datetime进行ORM操作 from_attributes True def to_dict_for_store(self) - Dict[str, Any]: 转换为适合向量数据库存储的字典格式。 return { id: self.id, content: self.content, metadata: { **self.metadata, created_at: self.created_at.isoformat(), last_accessed_at: self.last_accessed_at.isoformat() if self.last_accessed_at else None, access_count: self.access_count, } } classmethod def from_store_result(cls, id: str, content: str, metadata: Dict[str, Any], distance: float None) - MemoryItem: 从向量数据库查询结果中重建 MemoryItem 对象。 # 处理从数据库取出的元数据 created_at_str metadata.pop(created_at, None) last_accessed_at_str metadata.pop(last_accessed_at, None) access_count metadata.pop(access_count, 0) return cls( idid, contentcontent, metadatametadata, # 剩余的自定义元数据 created_atdatetime.fromisoformat(created_at_str) if created_at_str else datetime.utcnow(), last_accessed_atdatetime.fromisoformat(last_accessed_at_str) if last_accessed_at_str else None, access_countaccess_count, )metadata字段至关重要它可以存储来源如source: “slack#C123456”、用户IDuser_id: “U123ABC”、会话ID、重要性评分等信息为后续的检索过滤提供维度。3.2 封装向量数据库操作接下来在app/memory/vector_store.py中我们抽象一个向量存储层。这里以本地运行的 ChromaDB 为例它轻量且易于上手。import chromadb from chromadb.config import Settings from typing import List, Optional, Dict, Any from app.memory.models import MemoryItem from app.config import settings # 假设有一个全局配置对象 import logging logger logging.getLogger(__name__) class VectorMemoryStore: 向量记忆存储的封装类。 def __init__(self, persist_directory: str ./chroma_db): # 初始化 Chroma 客户端设置持久化路径 self.client chromadb.PersistentClient( pathpersist_directory, settingsSettings(anonymized_telemetryFalse) # 禁用匿名遥测 ) # 获取或创建集合Collection。集合名可以固定也可以按用户/会话划分。 self.collection self.client.get_or_create_collection( nameshared_memory, metadata{hnsw:space: cosine} # 使用余弦相似度进行搜索 ) logger.info(f向量存储初始化完成持久化目录: {persist_directory}) def add_memory(self, memory_item: MemoryItem, embedding: List[float]) - str: 向向量存储添加一条记忆。 try: # ChromaDB 需要将文档、ID、元数据和向量分开传入 self.collection.add( documents[memory_item.content], ids[memory_item.id], metadatas[memory_item.to_dict_for_store()[metadata]], embeddings[embedding] ) logger.debug(f已添加记忆 ID: {memory_item.id}) return memory_item.id except Exception as e: logger.error(f添加记忆失败: {e}) raise def search_similar(self, query_embedding: List[float], filter_metadata: Optional[Dict] None, limit: int 5) - List[MemoryItem]: 根据查询向量搜索相似的记忆。 try: # 执行搜索 results self.collection.query( query_embeddings[query_embedding], n_resultslimit, wherefilter_metadata, # 可选的元数据过滤条件 include[documents, metadatas, distances] ) memories [] # results 的结构是 {ids: [[...]], documents: [[...]], ...} if results[ids] and results[ids][0]: for i in range(len(results[ids][0])): mem_id results[ids][0][i] content results[documents][0][i] metadata results[metadatas][0][i] distance results[distances][0][i] memory_item MemoryItem.from_store_result(mem_id, content, metadata, distance) memories.append(memory_item) return memories except Exception as e: logger.error(f搜索记忆失败: {e}) return [] def update_memory_access(self, memory_id: str): 更新记忆的最后访问时间和次数需要先查询再更新ChromaDB 更新元数据较麻烦此处简化。 # 注意ChromaDB 的 update 方法不能部分更新元数据通常需要先 get修改后再 update。 # 对于频繁更新的 access_count可以考虑使用外部缓存如Redis或定期批量更新。 # 此处为简化示例仅记录日志。生产环境需根据选型实现。 logger.debug(f记忆 {memory_id} 被访问应在元数据中更新访问记录。) def delete_memory(self, memory_id: str): 删除指定记忆。 try: self.collection.delete(ids[memory_id]) logger.info(f已删除记忆 ID: {memory_id}) except Exception as e: logger.error(f删除记忆失败: {e})这个类封装了 ChromaDB 的基本操作。请注意ChromaDB 的update操作在处理嵌套元数据时有限制因此对于需要频繁更新的字段如access_count可能需要更复杂的策略例如结合外部缓存。3.3 实现记忆管理器记忆管理器app/memory/manager.py是业务逻辑层它协调嵌入模型和向量存储提供高级的“存储”和“检索”接口。from typing import List, Optional, Dict, Any from app.memory.models import MemoryItem from app.memory.vector_store import VectorMemoryStore from app.services.openai_client import get_embedding # 假设有一个获取嵌入向量的服务 import logging logger logging.getLogger(__name__) class MemoryManager: 共享记忆管理器负责记忆的存储、检索和生命周期管理。 def __init__(self, vector_store: VectorMemoryStore): self.vector_store vector_store async def store_memory(self, content: str, metadata: Optional[Dict[str, Any]] None) - str: 存储一段文本到共享记忆中。 if metadata is None: metadata {} # 1. 创建记忆条目对象 memory_item MemoryItem(contentcontent, metadatametadata) # 2. 为内容生成嵌入向量 try: embedding await get_embedding(content) memory_item.embedding embedding # 可选存储 except Exception as e: logger.error(f为内容生成嵌入向量失败记忆将被存储但无法被语义检索: {e}) # 即使生成失败也可以存储只是后续无法通过语义搜索找到它。 # 可以根据业务需求决定是抛出异常还是继续。 embedding [] # 使用空向量或随机向量但检索效果会差 # 3. 存入向量数据库 memory_id self.vector_store.add_memory(memory_item, embedding) logger.info(f成功存储记忆ID: {memory_id}, 内容长度: {len(content)}) return memory_id async def retrieve_relevant_memories(self, query: str, filter_by: Optional[Dict] None, top_k: int 3) - List[MemoryItem]: 根据查询文本检索最相关的记忆。 # 1. 为查询文本生成嵌入向量 try: query_embedding await get_embedding(query) except Exception as e: logger.error(f为查询生成嵌入向量失败: {e}) return [] # 检索失败返回空列表 # 2. 在向量数据库中执行相似性搜索 relevant_memories self.vector_store.search_similar( query_embeddingquery_embedding, filter_metadatafilter_by, limittop_k ) # 3. 更新检索到的记忆的访问记录异步或延迟更新 for memory in relevant_memories: self.vector_store.update_memory_access(memory.id) logger.debug(f为查询 {query[:50]}... 检索到 {len(relevant_memories)} 条相关记忆。) return relevant_memories def clear_memories(self, filter_by: Optional[Dict] None): 清除记忆。注意ChromaDB 的 delete 需要 ids 或 where 条件此方法为高级封装需谨慎实现。 # 实现批量删除逻辑例如先根据 filter_by 查询出 ids再删除。 # 生产环境需要非常小心此处省略具体实现。 logger.warning(clear_memories 方法未完全实现防止误操作。)记忆管理器是业务逻辑与底层存储的桥梁。store_memory方法封装了从文本到向量再到存储的完整流程。retrieve_relevant_memories则是共享记忆系统的灵魂它实现了“按需取用”的关键步骤。4. 集成 Slack 与 AI Agent有了共享记忆系统我们需要一个 AI Agent 来协调 Slack 事件、记忆检索和 LLM 调用。4.1 配置 Slack 事件接收与响应首先在app/main.py中设置一个 FastAPI 应用来接收 Slack 的事件推送。from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse import logging from slack_sdk import WebClient from slack_sdk.errors import SlackApiError from app.config import settings from app.agents.slack_agent import SlackAgent # 我们将创建这个Agent import hmac import hashlib import time app FastAPI(titleAI Slack Agent with Shared Memory) slack_client WebClient(tokensettings.SLACK_BOT_TOKEN) slack_agent SlackAgent(slack_client) # 初始化Agent logger logging.getLogger(__name__) def verify_slack_signature(request: Request, signing_secret: str) - bool: 验证 Slack 请求签名防止伪造请求。 timestamp request.headers.get(X-Slack-Request-Timestamp, ) slack_signature request.headers.get(X-Slack-Signature, ) if abs(time.time() - int(timestamp)) 60 * 5: # 请求时间戳与当前时间相差超过5分钟可能是重放攻击 return False sig_basestring fv0:{timestamp}:{await request.body()}.encode() my_signature v0 hmac.new( signing_secret.encode(), sig_basestring, hashlib.sha256 ).hexdigest() return hmac.compare_digest(my_signature, slack_signature) app.post(/slack/events) async def slack_events(request: Request): 接收 Slack 事件 API 的请求。 # 1. 验证签名生产环境必须启用 if not verify_slack_signature(request, settings.SLACK_SIGNING_SECRET): raise HTTPException(status_code403, detailInvalid Slack signature) payload await request.json() logger.debug(fReceived Slack event: {payload.get(type)}) # 2. 处理 URL 验证挑战Slack 事件订阅需要 if payload.get(type) url_verification: return JSONResponse(content{challenge: payload.get(challenge)}) # 3. 处理事件回调 event payload.get(event, {}) event_type event.get(type) # 只处理消息事件且过滤掉机器人自己发的消息和消息修改等子类型 if event_type message and not event.get(subtype): channel event.get(channel) user event.get(user) text event.get(text) ts event.get(ts) if not text or text.strip() : return JSONResponse(content{status: ok}) # 异步处理消息避免 Slack 3秒超时 # 在实际项目中应该使用任务队列如 Celery, RQ来处理 # 这里为了简化直接异步调用 import asyncio asyncio.create_task(slack_agent.handle_message(channel, user, text, ts)) # 4. 立即返回 200 OK 给 Slack return JSONResponse(content{status: ok}) app.get(/health) async def health_check(): return {status: healthy}这个端点负责接收 Slack 的所有事件。我们使用签名验证来确保请求来自 Slack。对于消息事件我们将其异步转发给SlackAgent处理以避免因 LLM 处理耗时导致 Slack 端请求超时。4.2 实现 Slack AI AgentSlackAgent(app/agents/slack_agent.py) 是核心协调者它连接了用户输入、共享记忆和 LLM。import logging from typing import Optional from slack_sdk import WebClient from app.memory.manager import MemoryManager from app.memory.vector_store import VectorMemoryStore from app.services.openai_client import get_chat_completion from app.config import settings logger logging.getLogger(__name__) class SlackAgent: def __init__(self, slack_client: WebClient): self.slack_client slack_client # 初始化共享记忆系统 vector_store VectorMemoryStore(persist_directorysettings.CHROMA_PERSIST_DIRECTORY) self.memory_manager MemoryManager(vector_store) async def handle_message(self, channel: str, user: str, text: str, ts: str): 处理一条 Slack 消息。 logger.info(f处理消息: 用户{user}, 频道{channel}, 内容{text[:100]}...) try: # 1. 将用户消息本身作为一条记忆存储可选取决于业务逻辑 # 可以存储原始消息也可以存储经过LLM摘要后的信息。 memory_metadata { source: slack, channel: channel, user_id: user, timestamp: ts, type: user_message } # 异步存储不阻塞后续流程 # asyncio.create_task(self.memory_manager.store_memory(text, memory_metadata)) # 2. 从共享记忆中检索与当前消息相关的历史信息 relevant_memories await self.memory_manager.retrieve_relevant_memories( querytext, filter_by{source: slack, channel: channel}, # 可选只检索本频道的历史 top_k3 ) # 3. 构建包含“记忆”的提示词Prompt system_prompt 你是一个有帮助的 Slack AI 助手。你拥有一个共享记忆系统可以记住之前对话的重要内容。 以下是一些可能与当前对话相关的历史记忆片段仅供参考 {memories_context} 请基于当前对话和以上相关记忆如果存在友好、专业地回应用户。如果记忆中的信息与当前问题相关可以自然地引用。 如果用户的问题需要长期记住某些信息如偏好、任务细节、项目信息请在你的回复中明确说明你会记住它系统会自动处理存储。 memories_context if relevant_memories: memories_context \n.join([f- {mem.content} for mem in relevant_memories]) else: memories_context 暂无相关记忆 final_system_prompt system_prompt.format(memories_contextmemories_context) # 4. 调用 LLM 生成回复 llm_response await get_chat_completion( system_promptfinal_system_prompt, user_prompttext, modelsettings.OPENAI_CHAT_MODEL ) # 5. 将 AI 的回复发送回 Slack await self._post_message(channel, llm_response, ts) # 6. 可选将 AI 回复中的重要信息或整个对话摘要存储为记忆 # 这里简单地将 AI 回复也存储起来 ai_memory_metadata { source: slack, channel: channel, user_id: assistant, timestamp: ts, type: assistant_response, in_response_to: text[:200] # 关联原始问题 } # 可以只存储回复或者存储“用户问... AI答...”的摘要对 await self.memory_manager.store_memory(llm_response, ai_memory_metadata) except Exception as e: logger.exception(f处理消息时发生错误: {e}) error_msg 抱歉我处理你的消息时遇到了点问题。 await self._post_message(channel, error_msg, ts) async def _post_message(self, channel: str, text: str, thread_ts: Optional[str] None): 向 Slack 频道发送消息。 try: response self.slack_client.chat_postMessage( channelchannel, texttext, thread_tsthread_ts # 如果提供则作为回复发送到线程中 ) logger.debug(f消息发送成功: {response.get(ts)}) except SlackApiError as e: logger.error(f发送 Slack 消息失败: {e.response[error]})这个 Agent 的工作流清晰体现了共享记忆的价值收到消息 - 检索相关记忆 - 构建增强的 Prompt - 调用 LLM - 回复并可能存储新记忆。关键在于第 2 步和第 3 步它让模型在回答时拥有了“上下文感知”能力。4.3 实现 OpenAI 服务客户端我们需要一个服务来调用 OpenAI 的 API 生成嵌入向量和聊天补全。在app/services/openai_client.py中import openai from openai import AsyncOpenAI from app.config import settings import logging import tiktoken # 用于计算 Token控制长度 logger logging.getLogger(__name__) # 初始化异步客户端 client AsyncOpenAI(api_keysettings.OPENAI_API_KEY, base_urlsettings.OPENAI_API_BASE) async def get_embedding(text: str, model: str None) - list[float]: 获取文本的嵌入向量。 if model is None: model settings.OPENAI_EMBEDDING_MODEL try: # 注意需要将文本转换为列表 response await client.embeddings.create( modelmodel, input[text.replace(\n, )] # 替换换行符 ) return response.data[0].embedding except Exception as e: logger.error(f获取嵌入向量失败: {e}, 文本: {text[:100]}) raise async def get_chat_completion(system_prompt: str, user_prompt: str, model: str None, max_tokens: int 1000) - str: 调用 OpenAI Chat Completion API 获取回复。 if model is None: model settings.OPENAI_CHAT_MODEL # 可选计算 Token 以确保不超过模型限制 encoding tiktoken.encoding_for_model(model) system_tokens len(encoding.encode(system_prompt)) user_tokens len(encoding.encode(user_prompt)) logger.debug(fToken 计数 - 系统: {system_tokens}, 用户: {user_tokens}) try: response await client.chat.completions.create( modelmodel, messages[ {role: system, content: system_prompt}, {role: user, content: user_prompt} ], max_tokensmax_tokens, temperature0.7, # 控制创造性 streamFalse ) return response.choices[0].message.content.strip() except Exception as e: logger.error(f调用 Chat Completion 失败: {e}) raise5. 运行、验证与常见问题排查5.1 本地运行与测试启动应用在项目根目录运行以下命令启动 FastAPI 服务。uvicorn app.main:app --reload --host 0.0.0.0 --port 8000服务将在http://localhost:8000启动。配置 Slack 事件订阅在 Slack API 控制台创建应用启用Event Subscriptions。将Request URL设置为你的公网可访问地址本地开发可使用 ngrok 等工具暴露http://localhost:8000/slack/events。订阅message.channels和message.groups事件根据你的需要。安装应用到你的工作空间获取Bot Token、Signing Secret和App Token填入.env文件。验证流程在 Slack 频道中 你的机器人并提问例如“我之前提到过我喜欢蓝色还记得吗”观察应用日志。你应该能看到类似以下的输出INFO:app.agents.slack_agent:处理消息: 用户U123ABC, 频道C123456, 内容我之前提到过我喜欢蓝色还记得吗... DEBUG:app.memory.manager:为查询 我之前提到过我喜欢蓝色还记得吗 检索到 X 条相关记忆。 DEBUG:app.services.openai_client:Token 计数 - 系统: XXX, 用户: XXX INFO:app.agents.slack_agent:消息发送成功: 1234567890.123456如果之前存储过关于“喜欢蓝色”的记忆机器人应该能回答“是的我记得你提到过你喜欢蓝色。”否则它会表示不记得。5.2 核心功能验证清单部署后请按以下清单验证核心功能是否正常验证项操作预期结果检查点1. 服务健康访问http://localhost:8000/health返回{status: healthy}HTTP 200 OK2. Slack 事件接收在 Slack 中发送一条消息应用日志显示“处理消息”日志无报错3. 记忆存储发送一条包含新信息如“我的项目代号是凤凰”的消息检查chroma_db目录下是否生成文件文件大小增加4. 记忆检索稍后发送相关查询如“我的项目代号是什么”机器人能正确回答“凤凰”回复内容正确5. 向量搜索发送语义相似但措辞不同的查询机器人仍能检索到正确记忆回复体现语义理解6. 上下文压缩进行多轮长对话后查看发送给 OpenAI 的 PromptPrompt 长度应受控不会无限增长Token 数稳定5.3 常见问题与排查路径在实际运行中你可能会遇到以下问题。请按此路径排查问题现象可能原因检查方式处理建议Slack 发送消息后机器人无响应1. 事件订阅 URL 未验证或配置错误。2. 签名验证失败。3. 应用未安装到频道。1. 查看 Slack API 控制台 Events Subscriptions 页面URL 是否显示 “Verified”。2. 查看应用日志是否有 “Invalid Slack signature” 错误。3. 在 Slack 频道中输入/invite 你的机器人。1. 确保 ngrok URL 正确且/slack/events端点返回正确的 challenge。2. 核对.env中的SLACK_SIGNING_SECRET。3. 将机器人邀请到频道。机器人回复“抱歉我遇到了问题”1. OpenAI API 密钥错误或额度不足。2. 网络问题导致 API 调用超时。3. 向量数据库操作异常。1. 检查应用日志中 OpenAI 相关的错误信息。2. 尝试直接调用openai_client.py中的函数进行测试。3. 检查chroma_db目录权限。1. 验证 API 密钥检查余额。2. 调整超时设置或检查网络连接。3. 确保运行服务的用户有目录读写权限。记忆检索不准确或检索不到1. 嵌入模型不匹配或生成失败。2. 向量数据库未持久化或数据丢失。3. 检索时过滤条件filter_by太严格。1. 在store_memory和retrieve函数中添加日志打印 embedding 是否成功生成。2. 检查chroma_db目录重启服务看记忆是否还在。3. 暂时移除filter_by参数测试。1. 确认OPENAI_EMBEDDING_MODEL设置正确API 调用正常。2. 确保 ChromaDB 客户端使用PersistentClient且路径正确。3. 调整元数据过滤逻辑或为记忆添加更丰富的元数据标签。Token 超限或 API 调用成本过高1. 存储的记忆内容过长。2. 检索到的记忆片段过多导致 Prompt 过大。3. 未对记忆进行摘要处理。1. 使用tiktoken计算存储和检索时文本的 Token 数。2. 监控 OpenAI API 使用量。1. 在存储前对长文本进行智能摘要可用另一个 LLM 调用。2. 限制top_k参数如从 5 降到 3。3. 设置 Prompt 的 Token 上限并优先截断最不相关的记忆。ChromaDB 在 Docker 或生产环境中数据丢失1. 持久化目录挂载不正确。2. 多实例运行导致数据不一致。1. 检查 Docker 卷映射或 Kubernetes PersistentVolume 配置。2. 检查日志中 ChromaDB 的初始化路径。1. 确保CHROMA_PERSIST_DIRECTORY指向一个持久化存储卷。2. 对于多副本部署考虑使用支持分布式的向量数据库如 Pinecone, Qdrant 集群。6. 生产环境最佳实践与扩展方向将共享记忆系统用于生产环境需要考虑更多关于性能、可靠性、成本和维护性的问题。6.1 生产环境部署建议向量数据库选型本地/小规模ChromaDB 持久化模式足够但需确保数据备份。云服务/大规模优先考虑托管服务如Pinecone、Weaviate Cloud、Qdrant Cloud或Azure AI Search。它们提供高可用、自动扩缩容和更好的性能。自托管考虑使用pgvectorPostgreSQL 扩展或RedisVL可以与你现有的数据库设施集成简化运维。记忆的存储与更新策略摘要化存储不要存储原始的长篇对话。使用 LLM 将一段对话或重要信息总结成简洁的要点再存储。这能显著降低存储和检索成本并提高相关性。记忆衰减与清理为记忆条目设置“有效期”或“重要性评分”。定期清理过于陈旧或低访问频率的记忆避免向量数据库膨胀。结构化元数据充分利用metadata字段。除了user_id,channel,timestamp还可以添加category如“用户偏好”、“项目信息”、“待办事项”、importance0-10分便于更精细的检索过滤。API 与异步处理使用消息队列在slack/events端点中收到事件后应立即放入消息队列如 Redis RQ, RabbitMQ, AWS SQS然后返回 200。由独立的 Worker 进程从队列中取出任务执行耗时的记忆检索和 LLM 调用。这能彻底避免 Slack 的 3 秒超时限制。实现重试与死信队列对于失败的 LLM 调用或存储操作应有重试机制最终仍失败的任务应进入死信队列供人工排查。监控与可观测性关键指标监控 LLM API 调用延迟、错误率、Token 消耗监控向量数据库的查询延迟、内存使用监控应用各端点的请求量和延迟。日志聚合将应用日志、向量数据库日志集中到 ELKElasticsearch, Logstash, Kibana或类似平台便于追踪问题。业务日志记录每次记忆存储和检索的上下文如查询文本、返回的记忆ID、相关性分数用于分析系统效果和优化检索策略。6.2 高级扩展方向混合检索Hybrid Search结合语义搜索向量和关键词搜索如 BM25。例如用户查询“上个月张三说的关于预算的报告”其中“上个月”、“张三”、“预算”是明确的关键词而“报告”是语义概念。混合检索能同时利用两者的优势提高召回率。LangChain 等框架对此有良好支持。记忆分层与压缩实现多级记忆系统。短期记忆最近对话保持高精度长期记忆则进行高度摘要和压缩。对于超长上下文可以采用“递归摘要”或“滑动窗口摘要”技术。记忆主动触发除了被动检索系统可以主动在特定时机向用户确认或提醒记忆。例如当检测到用户开始讨论一个之前提过的项目时Agent 可以主动说“我记得我们之前讨论过‘凤凰’项目需要我回顾一下当时的要点吗”与外部知识库集成共享记忆不仅可以存储对话历史还可以作为连接内部文档、Wiki、CRM 数据的接口。通过将外部文档切片并向量化存入记忆库Agent 就能在回答问题时引用公司内部知识成为一个真正的“企业知识助手”。多模态记忆未来的记忆不应局限于文本。可以结合多模态模型将图像、音频中的关键信息也转化为结构化描述或向量存入记忆库实现更丰富的上下文感知。共享记忆系统是构建实用、智能且经济的 AI Agent 的基石。它通过将庞大的上下文负担从昂贵的大模型调用中剥离出来交由专门的、可优化的存储检索系统处理实现了成本、性能和智能之间的平衡。从本文构建的最小可行系统出发你可以根据实际业务需求在检索精度、记忆结构、系统架构等方面进行深度定制打造出真正理解上下文、拥有长期记忆的 AI 助手。 AI Agent共享记忆系统构建：基于向量数据库与语义检索的上下文管理实践
在实际 AI 应用开发中无论是构建聊天助手、智能客服还是自动化工作流一个核心的挑战是如何让 AI 模型记住并有效利用超出其单次处理能力的长对话或复杂任务历史。传统的做法是将所有历史记录一股脑地塞进提示词Prompt但这会迅速耗尽模型的上下文窗口Context Window导致成本飙升、响应变慢甚至因为信息过载而影响回答质量。更棘手的是在多轮对话或长期运行的 AI Agent 中如何让模型记住几天前甚至几周前的关键信息并能在需要时精准调用这构成了所谓的“工作上下文瓶颈”。Lindy 提出的“共享记忆”概念正是为了解决这一瓶颈。它不是简单地将上下文窗口从 8K 扩展到 32K 或 128K而是引入了一种更智能、更结构化的信息管理机制。你可以把它想象成一个为 AI 配备的“外部大脑”或“知识库”这个大脑专门负责存储、索引和检索那些对当前任务至关重要的长期或跨会话信息。当 AI 需要处理新请求时它不再需要读取全部历史而是从这个“共享记忆”中快速查询出最相关的片段与当前问题一起构成一个精简而高效的上下文从而在有限的窗口内做出更精准的决策。本文将深入探讨如何在实际项目中特别是面向 Slack 等协作平台的 AI Agent 开发中实现“共享记忆”机制。我们将从核心概念入手逐步构建一个最小可运行的共享记忆系统涵盖环境准备、数据结构设计、核心代码实现、与 Slack 的集成以及最关键的生产环境部署和问题排查。无论你是正在开发 AI 聊天机器人、自动化流程 Agent还是希望优化现有大模型应用的上下文管理这篇文章都将提供一套可落地的工程方案。1. 理解共享记忆从上下文瓶颈到智能检索在深入代码之前我们必须先厘清几个核心概念理解为什么简单的上下文扩展不足以解决问题以及共享记忆机制是如何工作的。1.1 上下文窗口的限制与成本大语言模型LLM的上下文窗口决定了它一次性能处理多少文本包括你的提示词和它的回复。虽然 Claude 3、GPT-4 等模型支持长达 128K 甚至更多的上下文但这带来了两个现实问题成本问题绝大多数 API 的计费是基于输入和输出的总 Token 数量。将大量无关的历史对话持续放入上下文意味着每一轮交互你都在为这些“可能用不上”的信息付费。性能与质量衰减即使模型理论上能处理长上下文但有研究表明当关键信息被淹没在大量文本中时模型的检索和推理能力会下降。它可能“记得”信息存在但无法精准定位和运用。因此无节制地扩大单次请求的上下文并非最佳实践。1.2 共享记忆的核心思想共享记忆的核心思想是“按需取用长期存储”。它将 AI 工作所需的知识分为两类工作记忆Working Memory即当前对话轮次的上下文是直接送给模型处理的 Prompt。它应该尽可能精简、相关。长期记忆Long-term Memory即共享记忆存储库保存了跨会话、跨任务的历史信息、用户偏好、项目详情、操作结果等。当一个新的用户请求到来时系统不会将整个长期记忆塞进工作记忆而是先执行一个“检索Retrieval”步骤根据当前请求的语义从长期记忆中找出最相关的若干条记录。然后只将这些检索到的相关记录连同当前的用户问题一起构成最终的工作记忆送给模型。这个过程极大地缓解了上下文窗口的压力。1.3 关键技术组件嵌入模型与向量数据库实现智能检索依赖两项关键技术文本嵌入模型Embedding Model如 OpenAI 的text-embedding-3-small、text-embedding-ada-002或开源的BGE、SentenceTransformers模型。它的作用是将一段文本如一条历史消息转换为一个高维度的向量一组数字。语义相似的文本其向量在空间中的距离也更近。向量数据库Vector Database如 Pinecone、Weaviate、Qdrant或集成了向量搜索的 PostgreSQLpgvector、RedisRedisVL。它专门用于高效存储这些向量并执行“近似最近邻ANN”搜索。当新查询到来时先将其转换为向量然后在数据库中快速找到与之最相似的向量所对应的原始文本。通过这两者结合我们就能实现基于语义的、而不仅仅是关键词匹配的智能信息检索这正是共享记忆的“智能”所在。2. 环境准备与项目结构我们将构建一个面向 Slack 的 AI 助手示例它能够记住与用户的历史对话并在后续交互中引用这些信息。技术栈选择 Python因为它拥有最丰富的 AI 开发生态。2.1 基础环境与依赖首先确保你的开发环境已就绪。我们使用 Python 3.9 和pip进行包管理。# 创建并进入项目目录 mkdir ai-slack-agent-with-memory cd ai-slack-agent-with-memory # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate # 创建 requirements.txt 文件并安装核心依赖以下是requirements.txt文件的内容包含了从网络通信到 AI 模型调用的全套依赖# 核心框架与HTTP fastapi0.104.1 uvicorn[standard]0.24.0 slack-sdk3.27.1 httpx0.25.1 # AI/ML 相关 openai1.3.0 langchain0.0.340 langchain-openai0.0.2 tiktoken0.5.1 # 向量数据库与存储 chromadb0.4.18 # 或者使用 pgvector 需要 psycopg2-binary # psycopg2-binary2.9.9 # 工具与工具 python-dotenv1.0.0 pydantic2.5.0 pydantic-settings2.1.0使用 pip 安装pip install -r requirements.txt2.2 关键服务配置与密钥管理本项目需要与多个外部服务交互务必妥善管理密钥。我们使用.env文件来存储敏感信息并通过python-dotenv加载。创建一个名为.env的文件在项目根目录内容如下请替换为你的实际密钥# OpenAI API 配置用于对话和生成嵌入向量 OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_API_BASEhttps://api.openai.com/v1 # 如果使用代理或特定端点 OPENAI_EMBEDDING_MODELtext-embedding-3-small OPENAI_CHAT_MODELgpt-3.5-turbo # 或 gpt-4, gpt-4-turbo-preview # Slack 应用配置 SLACK_BOT_TOKENxoxb-your-slack-bot-token SLACK_SIGNING_SECRETyour-slack-signing-secret SLACK_APP_TOKENxapp-your-slack-app-token # 向量数据库配置以ChromaDB本地运行为例生产环境需配置持久化路径或远程地址 CHROMA_PERSIST_DIRECTORY./chroma_db # 如果使用 Pinecone # PINECONE_API_KEYyour-pinecone-key # PINECONE_ENVIRONMENTyour-environment # PINECONE_INDEX_NAMEyour-index-name注意.env文件必须被添加到.gitignore中绝对不要提交到版本控制系统。生产环境中应使用环境变量、密钥管理服务如 AWS Secrets Manager, HashiCorp Vault或平台提供的配置管理功能。2.3 项目目录结构设计一个清晰的项目结构有助于代码管理和维护。建议按以下方式组织ai-slack-agent-with-memory/ ├── .env # 环境变量本地开发用不上传 ├── .gitignore ├── requirements.txt ├── README.md ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口Slack 事件接收 │ ├── config.py # 配置加载Pydantic Settings │ ├── memory/ # 共享记忆核心模块 │ │ ├── __init__.py │ │ ├── manager.py # 记忆管理器存储、检索、更新 │ │ ├── models.py # 记忆条目的数据模型Pydantic │ │ └── vector_store.py # 向量数据库封装层 │ ├── agents/ # AI Agent 逻辑 │ │ ├── __init__.py │ │ └── slack_agent.py # 处理 Slack 消息协调记忆与LLM │ ├── services/ # 外部服务客户端 │ │ ├── __init__.py │ │ ├── openai_client.py │ │ └── slack_client.py │ └── utils/ # 工具函数 │ ├── __init__.py │ └── helpers.py └── tests/ # 测试目录 ├── __init__.py └── test_memory.py这个结构将配置、记忆管理、Agent 逻辑、服务客户端和工具函数进行了分离符合单一职责原则便于测试和扩展。3. 构建共享记忆系统的核心模块共享记忆系统的核心是记忆管理器Memory Manager和底层的向量存储。我们首先实现这两个部分。3.1 定义记忆数据模型在app/memory/models.py中我们使用 Pydantic 定义记忆条目的结构。这确保了数据的类型安全并便于序列化。from datetime import datetime from typing import Optional, Dict, Any from pydantic import BaseModel, Field from uuid import uuid4 class MemoryItem(BaseModel): 表示一条存储在共享记忆中的信息条目。 id: str Field(default_factorylambda: str(uuid4())) content: str Field(..., description记忆的文本内容) embedding: Optional[list[float]] Field(None, description文本内容的向量表示) metadata: Dict[str, Any] Field(default_factorydict, description关联的元数据) created_at: datetime Field(default_factorydatetime.utcnow) last_accessed_at: Optional[datetime] Field(None) access_count: int Field(default0) class Config: # 允许使用非Pydantic类型如datetime进行ORM操作 from_attributes True def to_dict_for_store(self) - Dict[str, Any]: 转换为适合向量数据库存储的字典格式。 return { id: self.id, content: self.content, metadata: { **self.metadata, created_at: self.created_at.isoformat(), last_accessed_at: self.last_accessed_at.isoformat() if self.last_accessed_at else None, access_count: self.access_count, } } classmethod def from_store_result(cls, id: str, content: str, metadata: Dict[str, Any], distance: float None) - MemoryItem: 从向量数据库查询结果中重建 MemoryItem 对象。 # 处理从数据库取出的元数据 created_at_str metadata.pop(created_at, None) last_accessed_at_str metadata.pop(last_accessed_at, None) access_count metadata.pop(access_count, 0) return cls( idid, contentcontent, metadatametadata, # 剩余的自定义元数据 created_atdatetime.fromisoformat(created_at_str) if created_at_str else datetime.utcnow(), last_accessed_atdatetime.fromisoformat(last_accessed_at_str) if last_accessed_at_str else None, access_countaccess_count, )metadata字段至关重要它可以存储来源如source: “slack#C123456”、用户IDuser_id: “U123ABC”、会话ID、重要性评分等信息为后续的检索过滤提供维度。3.2 封装向量数据库操作接下来在app/memory/vector_store.py中我们抽象一个向量存储层。这里以本地运行的 ChromaDB 为例它轻量且易于上手。import chromadb from chromadb.config import Settings from typing import List, Optional, Dict, Any from app.memory.models import MemoryItem from app.config import settings # 假设有一个全局配置对象 import logging logger logging.getLogger(__name__) class VectorMemoryStore: 向量记忆存储的封装类。 def __init__(self, persist_directory: str ./chroma_db): # 初始化 Chroma 客户端设置持久化路径 self.client chromadb.PersistentClient( pathpersist_directory, settingsSettings(anonymized_telemetryFalse) # 禁用匿名遥测 ) # 获取或创建集合Collection。集合名可以固定也可以按用户/会话划分。 self.collection self.client.get_or_create_collection( nameshared_memory, metadata{hnsw:space: cosine} # 使用余弦相似度进行搜索 ) logger.info(f向量存储初始化完成持久化目录: {persist_directory}) def add_memory(self, memory_item: MemoryItem, embedding: List[float]) - str: 向向量存储添加一条记忆。 try: # ChromaDB 需要将文档、ID、元数据和向量分开传入 self.collection.add( documents[memory_item.content], ids[memory_item.id], metadatas[memory_item.to_dict_for_store()[metadata]], embeddings[embedding] ) logger.debug(f已添加记忆 ID: {memory_item.id}) return memory_item.id except Exception as e: logger.error(f添加记忆失败: {e}) raise def search_similar(self, query_embedding: List[float], filter_metadata: Optional[Dict] None, limit: int 5) - List[MemoryItem]: 根据查询向量搜索相似的记忆。 try: # 执行搜索 results self.collection.query( query_embeddings[query_embedding], n_resultslimit, wherefilter_metadata, # 可选的元数据过滤条件 include[documents, metadatas, distances] ) memories [] # results 的结构是 {ids: [[...]], documents: [[...]], ...} if results[ids] and results[ids][0]: for i in range(len(results[ids][0])): mem_id results[ids][0][i] content results[documents][0][i] metadata results[metadatas][0][i] distance results[distances][0][i] memory_item MemoryItem.from_store_result(mem_id, content, metadata, distance) memories.append(memory_item) return memories except Exception as e: logger.error(f搜索记忆失败: {e}) return [] def update_memory_access(self, memory_id: str): 更新记忆的最后访问时间和次数需要先查询再更新ChromaDB 更新元数据较麻烦此处简化。 # 注意ChromaDB 的 update 方法不能部分更新元数据通常需要先 get修改后再 update。 # 对于频繁更新的 access_count可以考虑使用外部缓存如Redis或定期批量更新。 # 此处为简化示例仅记录日志。生产环境需根据选型实现。 logger.debug(f记忆 {memory_id} 被访问应在元数据中更新访问记录。) def delete_memory(self, memory_id: str): 删除指定记忆。 try: self.collection.delete(ids[memory_id]) logger.info(f已删除记忆 ID: {memory_id}) except Exception as e: logger.error(f删除记忆失败: {e})这个类封装了 ChromaDB 的基本操作。请注意ChromaDB 的update操作在处理嵌套元数据时有限制因此对于需要频繁更新的字段如access_count可能需要更复杂的策略例如结合外部缓存。3.3 实现记忆管理器记忆管理器app/memory/manager.py是业务逻辑层它协调嵌入模型和向量存储提供高级的“存储”和“检索”接口。from typing import List, Optional, Dict, Any from app.memory.models import MemoryItem from app.memory.vector_store import VectorMemoryStore from app.services.openai_client import get_embedding # 假设有一个获取嵌入向量的服务 import logging logger logging.getLogger(__name__) class MemoryManager: 共享记忆管理器负责记忆的存储、检索和生命周期管理。 def __init__(self, vector_store: VectorMemoryStore): self.vector_store vector_store async def store_memory(self, content: str, metadata: Optional[Dict[str, Any]] None) - str: 存储一段文本到共享记忆中。 if metadata is None: metadata {} # 1. 创建记忆条目对象 memory_item MemoryItem(contentcontent, metadatametadata) # 2. 为内容生成嵌入向量 try: embedding await get_embedding(content) memory_item.embedding embedding # 可选存储 except Exception as e: logger.error(f为内容生成嵌入向量失败记忆将被存储但无法被语义检索: {e}) # 即使生成失败也可以存储只是后续无法通过语义搜索找到它。 # 可以根据业务需求决定是抛出异常还是继续。 embedding [] # 使用空向量或随机向量但检索效果会差 # 3. 存入向量数据库 memory_id self.vector_store.add_memory(memory_item, embedding) logger.info(f成功存储记忆ID: {memory_id}, 内容长度: {len(content)}) return memory_id async def retrieve_relevant_memories(self, query: str, filter_by: Optional[Dict] None, top_k: int 3) - List[MemoryItem]: 根据查询文本检索最相关的记忆。 # 1. 为查询文本生成嵌入向量 try: query_embedding await get_embedding(query) except Exception as e: logger.error(f为查询生成嵌入向量失败: {e}) return [] # 检索失败返回空列表 # 2. 在向量数据库中执行相似性搜索 relevant_memories self.vector_store.search_similar( query_embeddingquery_embedding, filter_metadatafilter_by, limittop_k ) # 3. 更新检索到的记忆的访问记录异步或延迟更新 for memory in relevant_memories: self.vector_store.update_memory_access(memory.id) logger.debug(f为查询 {query[:50]}... 检索到 {len(relevant_memories)} 条相关记忆。) return relevant_memories def clear_memories(self, filter_by: Optional[Dict] None): 清除记忆。注意ChromaDB 的 delete 需要 ids 或 where 条件此方法为高级封装需谨慎实现。 # 实现批量删除逻辑例如先根据 filter_by 查询出 ids再删除。 # 生产环境需要非常小心此处省略具体实现。 logger.warning(clear_memories 方法未完全实现防止误操作。)记忆管理器是业务逻辑与底层存储的桥梁。store_memory方法封装了从文本到向量再到存储的完整流程。retrieve_relevant_memories则是共享记忆系统的灵魂它实现了“按需取用”的关键步骤。4. 集成 Slack 与 AI Agent有了共享记忆系统我们需要一个 AI Agent 来协调 Slack 事件、记忆检索和 LLM 调用。4.1 配置 Slack 事件接收与响应首先在app/main.py中设置一个 FastAPI 应用来接收 Slack 的事件推送。from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse import logging from slack_sdk import WebClient from slack_sdk.errors import SlackApiError from app.config import settings from app.agents.slack_agent import SlackAgent # 我们将创建这个Agent import hmac import hashlib import time app FastAPI(titleAI Slack Agent with Shared Memory) slack_client WebClient(tokensettings.SLACK_BOT_TOKEN) slack_agent SlackAgent(slack_client) # 初始化Agent logger logging.getLogger(__name__) def verify_slack_signature(request: Request, signing_secret: str) - bool: 验证 Slack 请求签名防止伪造请求。 timestamp request.headers.get(X-Slack-Request-Timestamp, ) slack_signature request.headers.get(X-Slack-Signature, ) if abs(time.time() - int(timestamp)) 60 * 5: # 请求时间戳与当前时间相差超过5分钟可能是重放攻击 return False sig_basestring fv0:{timestamp}:{await request.body()}.encode() my_signature v0 hmac.new( signing_secret.encode(), sig_basestring, hashlib.sha256 ).hexdigest() return hmac.compare_digest(my_signature, slack_signature) app.post(/slack/events) async def slack_events(request: Request): 接收 Slack 事件 API 的请求。 # 1. 验证签名生产环境必须启用 if not verify_slack_signature(request, settings.SLACK_SIGNING_SECRET): raise HTTPException(status_code403, detailInvalid Slack signature) payload await request.json() logger.debug(fReceived Slack event: {payload.get(type)}) # 2. 处理 URL 验证挑战Slack 事件订阅需要 if payload.get(type) url_verification: return JSONResponse(content{challenge: payload.get(challenge)}) # 3. 处理事件回调 event payload.get(event, {}) event_type event.get(type) # 只处理消息事件且过滤掉机器人自己发的消息和消息修改等子类型 if event_type message and not event.get(subtype): channel event.get(channel) user event.get(user) text event.get(text) ts event.get(ts) if not text or text.strip() : return JSONResponse(content{status: ok}) # 异步处理消息避免 Slack 3秒超时 # 在实际项目中应该使用任务队列如 Celery, RQ来处理 # 这里为了简化直接异步调用 import asyncio asyncio.create_task(slack_agent.handle_message(channel, user, text, ts)) # 4. 立即返回 200 OK 给 Slack return JSONResponse(content{status: ok}) app.get(/health) async def health_check(): return {status: healthy}这个端点负责接收 Slack 的所有事件。我们使用签名验证来确保请求来自 Slack。对于消息事件我们将其异步转发给SlackAgent处理以避免因 LLM 处理耗时导致 Slack 端请求超时。4.2 实现 Slack AI AgentSlackAgent(app/agents/slack_agent.py) 是核心协调者它连接了用户输入、共享记忆和 LLM。import logging from typing import Optional from slack_sdk import WebClient from app.memory.manager import MemoryManager from app.memory.vector_store import VectorMemoryStore from app.services.openai_client import get_chat_completion from app.config import settings logger logging.getLogger(__name__) class SlackAgent: def __init__(self, slack_client: WebClient): self.slack_client slack_client # 初始化共享记忆系统 vector_store VectorMemoryStore(persist_directorysettings.CHROMA_PERSIST_DIRECTORY) self.memory_manager MemoryManager(vector_store) async def handle_message(self, channel: str, user: str, text: str, ts: str): 处理一条 Slack 消息。 logger.info(f处理消息: 用户{user}, 频道{channel}, 内容{text[:100]}...) try: # 1. 将用户消息本身作为一条记忆存储可选取决于业务逻辑 # 可以存储原始消息也可以存储经过LLM摘要后的信息。 memory_metadata { source: slack, channel: channel, user_id: user, timestamp: ts, type: user_message } # 异步存储不阻塞后续流程 # asyncio.create_task(self.memory_manager.store_memory(text, memory_metadata)) # 2. 从共享记忆中检索与当前消息相关的历史信息 relevant_memories await self.memory_manager.retrieve_relevant_memories( querytext, filter_by{source: slack, channel: channel}, # 可选只检索本频道的历史 top_k3 ) # 3. 构建包含“记忆”的提示词Prompt system_prompt 你是一个有帮助的 Slack AI 助手。你拥有一个共享记忆系统可以记住之前对话的重要内容。 以下是一些可能与当前对话相关的历史记忆片段仅供参考 {memories_context} 请基于当前对话和以上相关记忆如果存在友好、专业地回应用户。如果记忆中的信息与当前问题相关可以自然地引用。 如果用户的问题需要长期记住某些信息如偏好、任务细节、项目信息请在你的回复中明确说明你会记住它系统会自动处理存储。 memories_context if relevant_memories: memories_context \n.join([f- {mem.content} for mem in relevant_memories]) else: memories_context 暂无相关记忆 final_system_prompt system_prompt.format(memories_contextmemories_context) # 4. 调用 LLM 生成回复 llm_response await get_chat_completion( system_promptfinal_system_prompt, user_prompttext, modelsettings.OPENAI_CHAT_MODEL ) # 5. 将 AI 的回复发送回 Slack await self._post_message(channel, llm_response, ts) # 6. 可选将 AI 回复中的重要信息或整个对话摘要存储为记忆 # 这里简单地将 AI 回复也存储起来 ai_memory_metadata { source: slack, channel: channel, user_id: assistant, timestamp: ts, type: assistant_response, in_response_to: text[:200] # 关联原始问题 } # 可以只存储回复或者存储“用户问... AI答...”的摘要对 await self.memory_manager.store_memory(llm_response, ai_memory_metadata) except Exception as e: logger.exception(f处理消息时发生错误: {e}) error_msg 抱歉我处理你的消息时遇到了点问题。 await self._post_message(channel, error_msg, ts) async def _post_message(self, channel: str, text: str, thread_ts: Optional[str] None): 向 Slack 频道发送消息。 try: response self.slack_client.chat_postMessage( channelchannel, texttext, thread_tsthread_ts # 如果提供则作为回复发送到线程中 ) logger.debug(f消息发送成功: {response.get(ts)}) except SlackApiError as e: logger.error(f发送 Slack 消息失败: {e.response[error]})这个 Agent 的工作流清晰体现了共享记忆的价值收到消息 - 检索相关记忆 - 构建增强的 Prompt - 调用 LLM - 回复并可能存储新记忆。关键在于第 2 步和第 3 步它让模型在回答时拥有了“上下文感知”能力。4.3 实现 OpenAI 服务客户端我们需要一个服务来调用 OpenAI 的 API 生成嵌入向量和聊天补全。在app/services/openai_client.py中import openai from openai import AsyncOpenAI from app.config import settings import logging import tiktoken # 用于计算 Token控制长度 logger logging.getLogger(__name__) # 初始化异步客户端 client AsyncOpenAI(api_keysettings.OPENAI_API_KEY, base_urlsettings.OPENAI_API_BASE) async def get_embedding(text: str, model: str None) - list[float]: 获取文本的嵌入向量。 if model is None: model settings.OPENAI_EMBEDDING_MODEL try: # 注意需要将文本转换为列表 response await client.embeddings.create( modelmodel, input[text.replace(\n, )] # 替换换行符 ) return response.data[0].embedding except Exception as e: logger.error(f获取嵌入向量失败: {e}, 文本: {text[:100]}) raise async def get_chat_completion(system_prompt: str, user_prompt: str, model: str None, max_tokens: int 1000) - str: 调用 OpenAI Chat Completion API 获取回复。 if model is None: model settings.OPENAI_CHAT_MODEL # 可选计算 Token 以确保不超过模型限制 encoding tiktoken.encoding_for_model(model) system_tokens len(encoding.encode(system_prompt)) user_tokens len(encoding.encode(user_prompt)) logger.debug(fToken 计数 - 系统: {system_tokens}, 用户: {user_tokens}) try: response await client.chat.completions.create( modelmodel, messages[ {role: system, content: system_prompt}, {role: user, content: user_prompt} ], max_tokensmax_tokens, temperature0.7, # 控制创造性 streamFalse ) return response.choices[0].message.content.strip() except Exception as e: logger.error(f调用 Chat Completion 失败: {e}) raise5. 运行、验证与常见问题排查5.1 本地运行与测试启动应用在项目根目录运行以下命令启动 FastAPI 服务。uvicorn app.main:app --reload --host 0.0.0.0 --port 8000服务将在http://localhost:8000启动。配置 Slack 事件订阅在 Slack API 控制台创建应用启用Event Subscriptions。将Request URL设置为你的公网可访问地址本地开发可使用 ngrok 等工具暴露http://localhost:8000/slack/events。订阅message.channels和message.groups事件根据你的需要。安装应用到你的工作空间获取Bot Token、Signing Secret和App Token填入.env文件。验证流程在 Slack 频道中 你的机器人并提问例如“我之前提到过我喜欢蓝色还记得吗”观察应用日志。你应该能看到类似以下的输出INFO:app.agents.slack_agent:处理消息: 用户U123ABC, 频道C123456, 内容我之前提到过我喜欢蓝色还记得吗... DEBUG:app.memory.manager:为查询 我之前提到过我喜欢蓝色还记得吗 检索到 X 条相关记忆。 DEBUG:app.services.openai_client:Token 计数 - 系统: XXX, 用户: XXX INFO:app.agents.slack_agent:消息发送成功: 1234567890.123456如果之前存储过关于“喜欢蓝色”的记忆机器人应该能回答“是的我记得你提到过你喜欢蓝色。”否则它会表示不记得。5.2 核心功能验证清单部署后请按以下清单验证核心功能是否正常验证项操作预期结果检查点1. 服务健康访问http://localhost:8000/health返回{status: healthy}HTTP 200 OK2. Slack 事件接收在 Slack 中发送一条消息应用日志显示“处理消息”日志无报错3. 记忆存储发送一条包含新信息如“我的项目代号是凤凰”的消息检查chroma_db目录下是否生成文件文件大小增加4. 记忆检索稍后发送相关查询如“我的项目代号是什么”机器人能正确回答“凤凰”回复内容正确5. 向量搜索发送语义相似但措辞不同的查询机器人仍能检索到正确记忆回复体现语义理解6. 上下文压缩进行多轮长对话后查看发送给 OpenAI 的 PromptPrompt 长度应受控不会无限增长Token 数稳定5.3 常见问题与排查路径在实际运行中你可能会遇到以下问题。请按此路径排查问题现象可能原因检查方式处理建议Slack 发送消息后机器人无响应1. 事件订阅 URL 未验证或配置错误。2. 签名验证失败。3. 应用未安装到频道。1. 查看 Slack API 控制台 Events Subscriptions 页面URL 是否显示 “Verified”。2. 查看应用日志是否有 “Invalid Slack signature” 错误。3. 在 Slack 频道中输入/invite 你的机器人。1. 确保 ngrok URL 正确且/slack/events端点返回正确的 challenge。2. 核对.env中的SLACK_SIGNING_SECRET。3. 将机器人邀请到频道。机器人回复“抱歉我遇到了问题”1. OpenAI API 密钥错误或额度不足。2. 网络问题导致 API 调用超时。3. 向量数据库操作异常。1. 检查应用日志中 OpenAI 相关的错误信息。2. 尝试直接调用openai_client.py中的函数进行测试。3. 检查chroma_db目录权限。1. 验证 API 密钥检查余额。2. 调整超时设置或检查网络连接。3. 确保运行服务的用户有目录读写权限。记忆检索不准确或检索不到1. 嵌入模型不匹配或生成失败。2. 向量数据库未持久化或数据丢失。3. 检索时过滤条件filter_by太严格。1. 在store_memory和retrieve函数中添加日志打印 embedding 是否成功生成。2. 检查chroma_db目录重启服务看记忆是否还在。3. 暂时移除filter_by参数测试。1. 确认OPENAI_EMBEDDING_MODEL设置正确API 调用正常。2. 确保 ChromaDB 客户端使用PersistentClient且路径正确。3. 调整元数据过滤逻辑或为记忆添加更丰富的元数据标签。Token 超限或 API 调用成本过高1. 存储的记忆内容过长。2. 检索到的记忆片段过多导致 Prompt 过大。3. 未对记忆进行摘要处理。1. 使用tiktoken计算存储和检索时文本的 Token 数。2. 监控 OpenAI API 使用量。1. 在存储前对长文本进行智能摘要可用另一个 LLM 调用。2. 限制top_k参数如从 5 降到 3。3. 设置 Prompt 的 Token 上限并优先截断最不相关的记忆。ChromaDB 在 Docker 或生产环境中数据丢失1. 持久化目录挂载不正确。2. 多实例运行导致数据不一致。1. 检查 Docker 卷映射或 Kubernetes PersistentVolume 配置。2. 检查日志中 ChromaDB 的初始化路径。1. 确保CHROMA_PERSIST_DIRECTORY指向一个持久化存储卷。2. 对于多副本部署考虑使用支持分布式的向量数据库如 Pinecone, Qdrant 集群。6. 生产环境最佳实践与扩展方向将共享记忆系统用于生产环境需要考虑更多关于性能、可靠性、成本和维护性的问题。6.1 生产环境部署建议向量数据库选型本地/小规模ChromaDB 持久化模式足够但需确保数据备份。云服务/大规模优先考虑托管服务如Pinecone、Weaviate Cloud、Qdrant Cloud或Azure AI Search。它们提供高可用、自动扩缩容和更好的性能。自托管考虑使用pgvectorPostgreSQL 扩展或RedisVL可以与你现有的数据库设施集成简化运维。记忆的存储与更新策略摘要化存储不要存储原始的长篇对话。使用 LLM 将一段对话或重要信息总结成简洁的要点再存储。这能显著降低存储和检索成本并提高相关性。记忆衰减与清理为记忆条目设置“有效期”或“重要性评分”。定期清理过于陈旧或低访问频率的记忆避免向量数据库膨胀。结构化元数据充分利用metadata字段。除了user_id,channel,timestamp还可以添加category如“用户偏好”、“项目信息”、“待办事项”、importance0-10分便于更精细的检索过滤。API 与异步处理使用消息队列在slack/events端点中收到事件后应立即放入消息队列如 Redis RQ, RabbitMQ, AWS SQS然后返回 200。由独立的 Worker 进程从队列中取出任务执行耗时的记忆检索和 LLM 调用。这能彻底避免 Slack 的 3 秒超时限制。实现重试与死信队列对于失败的 LLM 调用或存储操作应有重试机制最终仍失败的任务应进入死信队列供人工排查。监控与可观测性关键指标监控 LLM API 调用延迟、错误率、Token 消耗监控向量数据库的查询延迟、内存使用监控应用各端点的请求量和延迟。日志聚合将应用日志、向量数据库日志集中到 ELKElasticsearch, Logstash, Kibana或类似平台便于追踪问题。业务日志记录每次记忆存储和检索的上下文如查询文本、返回的记忆ID、相关性分数用于分析系统效果和优化检索策略。6.2 高级扩展方向混合检索Hybrid Search结合语义搜索向量和关键词搜索如 BM25。例如用户查询“上个月张三说的关于预算的报告”其中“上个月”、“张三”、“预算”是明确的关键词而“报告”是语义概念。混合检索能同时利用两者的优势提高召回率。LangChain 等框架对此有良好支持。记忆分层与压缩实现多级记忆系统。短期记忆最近对话保持高精度长期记忆则进行高度摘要和压缩。对于超长上下文可以采用“递归摘要”或“滑动窗口摘要”技术。记忆主动触发除了被动检索系统可以主动在特定时机向用户确认或提醒记忆。例如当检测到用户开始讨论一个之前提过的项目时Agent 可以主动说“我记得我们之前讨论过‘凤凰’项目需要我回顾一下当时的要点吗”与外部知识库集成共享记忆不仅可以存储对话历史还可以作为连接内部文档、Wiki、CRM 数据的接口。通过将外部文档切片并向量化存入记忆库Agent 就能在回答问题时引用公司内部知识成为一个真正的“企业知识助手”。多模态记忆未来的记忆不应局限于文本。可以结合多模态模型将图像、音频中的关键信息也转化为结构化描述或向量存入记忆库实现更丰富的上下文感知。共享记忆系统是构建实用、智能且经济的 AI Agent 的基石。它通过将庞大的上下文负担从昂贵的大模型调用中剥离出来交由专门的、可优化的存储检索系统处理实现了成本、性能和智能之间的平衡。从本文构建的最小可行系统出发你可以根据实际业务需求在检索精度、记忆结构、系统架构等方面进行深度定制打造出真正理解上下文、拥有长期记忆的 AI 助手。