建基于共享記憶的智能協(xié)作系統(tǒng))
你是否遇到過這樣的場(chǎng)景當(dāng)你與一個(gè)AI助手在Slack或Teams上討論一個(gè)復(fù)雜項(xiàng)目時(shí)每次對(duì)話重啟它都像得了“健忘癥”需要你重新解釋一遍背景、目標(biāo)和之前的決策或者當(dāng)你試圖讓AI Agent處理一個(gè)跨越數(shù)天、涉及多個(gè)文檔和對(duì)話的任務(wù)時(shí)它很快就因?yàn)椤吧舷挛拇翱凇焙谋M而無法繼續(xù)這正是當(dāng)前AI應(yīng)用特別是AI Agent領(lǐng)域最核心的瓶頸之一上下文限制。無論是Claude的200K還是GPT-4的128K本質(zhì)上都是“一次性”的短期記憶。一旦對(duì)話結(jié)束或窗口填滿所有精心構(gòu)建的上下文就消失了。這導(dǎo)致AI無法真正像人類一樣在長(zhǎng)期、復(fù)雜的協(xié)作中積累知識(shí)和經(jīng)驗(yàn)。今天要介紹的項(xiàng)目Lindy正是為了解決這個(gè)問題而生。它不是一個(gè)新的大模型而是一個(gè)基于共享記憶的AI協(xié)作平臺(tái)。它的核心判斷非常清晰AI的長(zhǎng)期價(jià)值不在于單次問答的聰明而在于持續(xù)協(xié)作中的“成長(zhǎng)”和“傳承”。Lindy試圖通過為AI Agent構(gòu)建一個(gè)可持久化、可共享的“記憶系統(tǒng)”來突破上下文窗口的物理限制讓AI真正融入工作流。如果你正在開發(fā)或集成AI Agent為上下文管理而頭疼或者你是一個(gè)團(tuán)隊(duì)管理者希望AI能成為穩(wěn)定的“數(shù)字同事”而非臨時(shí)的“問答機(jī)器”那么這篇文章將為你提供一個(gè)全新的技術(shù)視角和一套可落地的實(shí)踐思路。我們將深入拆解Lindy的設(shè)計(jì)理念、核心組件并通過一個(gè)模擬的Slack集成示例展示如何讓AI擁有“共享記憶”。1. 這篇文章真正要解決的問題為什么上下文瓶頸是AI Agent的“阿喀琉斯之踵”在深入Lindy之前我們必須先理解“上下文瓶頸”到底卡住了什么。這不僅僅是技術(shù)參數(shù)問題更是AI應(yīng)用從“玩具”走向“工具”的關(guān)鍵障礙。首先上下文窗口的本質(zhì)是什么你可以把它想象成AI工作時(shí)的“桌面”。桌面越大上下文窗口越長(zhǎng)它能同時(shí)攤開的參考資料歷史對(duì)話、文檔、代碼就越多做出的判斷就越連貫、準(zhǔn)確。但無論桌面多大會(huì)議結(jié)束對(duì)話結(jié)束、下班關(guān)機(jī)會(huì)話終止桌面就會(huì)被清空。第二天上班一切又要從頭開始。其次這個(gè)瓶頸導(dǎo)致了三大現(xiàn)實(shí)困境任務(wù)無法延續(xù)一個(gè)需要多步驟、跨時(shí)區(qū)的數(shù)據(jù)分析或代碼審查任務(wù)AI無法記住中間的臨時(shí)結(jié)論和修改記錄。知識(shí)無法沉淀團(tuán)隊(duì)在與AI的協(xié)作中產(chǎn)生的寶貴經(jīng)驗(yàn)、決策邏輯、定制化知識(shí)隨著對(duì)話結(jié)束而煙消云散無法形成組織的“數(shù)字資產(chǎn)”。協(xié)作無法展開多個(gè)AI Agent之間或者AI與多個(gè)人類成員之間缺乏一個(gè)公共的、可同步的“記憶黑板”導(dǎo)致信息孤島和重復(fù)勞動(dòng)。Lindy的提出正是基于一個(gè)深刻的洞察未來的AI不是單次服務(wù)的調(diào)用者而是持續(xù)參與流程的協(xié)作者。因此它需要的不是更大的“一次性桌面”而是一個(gè)專屬的、可隨時(shí)存取的“文件柜”和“會(huì)議紀(jì)要庫”——這就是“共享記憶”Shared Memory的概念。本文將帶你從原理到實(shí)踐完整走通“共享記憶”的構(gòu)建之路。你會(huì)看到Lindy如何將記憶抽象為可存儲(chǔ)、可檢索的結(jié)構(gòu)化數(shù)據(jù)如何通過Slack等日常工具無縫集成以及在實(shí)際開發(fā)中你會(huì)遇到哪些“坑”和最佳實(shí)踐。2. 基礎(chǔ)概念與核心原理從“上下文”到“共享記憶”要理解Lindy需要先厘清幾個(gè)容易混淆的關(guān)鍵概念。2.1 上下文Context vs. 記憶Memory上下文Context Window這是大模型的技術(shù)參數(shù)。指單次請(qǐng)求中模型能“看到”的文本包括你的問題、歷史對(duì)話、系統(tǒng)指令等的最大長(zhǎng)度。它是臨時(shí)的、易失的與本次請(qǐng)求的生命周期綁定。記憶Memory這是一個(gè)更高層的應(yīng)用概念。指AI為了完成長(zhǎng)期目標(biāo)需要持久化存儲(chǔ)和回憶的信息。它超越了單次請(qǐng)求是跨會(huì)話、跨任務(wù)存在的。類比上下文就像你電腦的內(nèi)存RAM處理當(dāng)前任務(wù)時(shí)高速存取關(guān)機(jī)即丟失。記憶則像硬盤HDD/SSD或云盤用于長(zhǎng)期存儲(chǔ)隨時(shí)可以加載到內(nèi)存中使用。2.2 共享記憶Shared Memory的核心思想Lindy的“共享記憶”包含兩層含義持久化Persistence將重要的交互信息如用戶偏好、任務(wù)結(jié)論、學(xué)到的規(guī)則從易失的上下文窗口中提取出來存儲(chǔ)到外部數(shù)據(jù)庫或向量庫中。共享化Sharing這份存儲(chǔ)的記憶可以被同一個(gè)AI Agent在不同時(shí)間點(diǎn)訪問也可以被同一個(gè)團(tuán)隊(duì)內(nèi)的不同AI Agent訪問實(shí)現(xiàn)知識(shí)和狀態(tài)的同步。它的技術(shù)原理可以簡(jiǎn)化為一個(gè)循環(huán)[AI交互] - [記憶提取與編碼] - [外部存儲(chǔ)記憶庫] - [記憶檢索與解碼] - [注入新上下文] - [更智能的AI交互]這個(gè)循環(huán)打破了“一次一清空”的模式讓AI的能力隨時(shí)間推移而增強(qiáng)。2.3 Lindy的架構(gòu)角色根據(jù)有限的資料推斷Lindy很可能扮演以下角色記憶管理層提供API和SDK讓開發(fā)者能方便地定義“什么信息需要被記住”如決策、事實(shí)、用戶反饋以及如何存儲(chǔ)數(shù)據(jù)庫、向量索引。上下文裝配層在AI執(zhí)行任務(wù)前根據(jù)當(dāng)前任務(wù)目標(biāo)自動(dòng)從記憶庫中檢索最相關(guān)的記憶片段并將其作為系統(tǒng)提示詞或上下文的一部分注入給大模型。集成中間件與Slack、Discord等通訊工具深度集成監(jiān)聽對(duì)話自動(dòng)觸發(fā)記憶的存儲(chǔ)和檢索流程讓用戶無感知地使用。理解了這些概念我們就能明白Lindy的目標(biāo)是將AI的“智力”從短暫的上下文約束中解放出來使其建立在可積累、可共享的記憶基石之上。3. 環(huán)境準(zhǔn)備與前置條件由于Lindy是一個(gè)較新的概念和項(xiàng)目從my_ai_town等關(guān)聯(lián)信息看可能處于早期開源階段我們無法獲得官方的、穩(wěn)定的安裝包。因此本節(jié)將基于共享記憶的通用實(shí)現(xiàn)思路規(guī)劃一個(gè)可行的技術(shù)棧和環(huán)境你可以將其視為構(gòu)建“類Lindy”系統(tǒng)或理解其原理的藍(lán)圖。核心技術(shù)棧選擇編程語言Python 3.9。因其在AI和快速開發(fā)領(lǐng)域的生態(tài)優(yōu)勢(shì)。大模型接口OpenAI GPT API 或 Anthropic Claude API需自備API Key。也可用開源的Llama 3等本地模型但需部署相應(yīng)的推理服務(wù)。記憶存儲(chǔ)向量數(shù)據(jù)庫用于存儲(chǔ)非結(jié)構(gòu)化記憶對(duì)話片段、文檔摘要并支持語義檢索。推薦ChromaDB輕量、簡(jiǎn)單或Weaviate功能更全。傳統(tǒng)數(shù)據(jù)庫用于存儲(chǔ)結(jié)構(gòu)化記憶用戶ID、任務(wù)狀態(tài)、固定鍵值對(duì)。推薦SQLite開發(fā)測(cè)試或PostgreSQL生產(chǎn)環(huán)境。應(yīng)用框架LangChain或LlamaIndex。它們提供了構(gòu)建AI應(yīng)用包括記憶系統(tǒng)的高級(jí)抽象和工具鏈能極大簡(jiǎn)化開發(fā)。本文將使用LangChain進(jìn)行演示。通訊平臺(tái)集成Slack API。我們將模擬一個(gè)Slack機(jī)器人作為記憶交互的前端。環(huán)境管理使用Conda或venv創(chuàng)建獨(dú)立的Python環(huán)境。環(huán)境搭建步驟創(chuàng)建并激活Python虛擬環(huán)境。# 使用 venv python -m venv lindy_env source lindy_env/bin/activate # Linux/Mac # lindy_env\Scripts\activate # Windows安裝核心依賴。pip install langchain langchain-openai langchain-community chromadb slack-sdk python-dotenvlangchain: 核心框架。langchain-openai: OpenAI模型集成。langchain-community: 包含社區(qū)貢獻(xiàn)的組件如一些記憶存儲(chǔ)后端。chromadb: 向量數(shù)據(jù)庫。slack-sdk: Slack官方SDK。python-dotenv: 管理環(huán)境變量。準(zhǔn)備配置文件。在項(xiàng)目根目錄創(chuàng)建.env文件用于存放敏感信息。# .env 文件示例 OPENAI_API_KEYsk-your-openai-api-key-here SLACK_BOT_TOKENxoxb-your-slack-bot-token-here SLACK_APP_TOKENxapp-your-slack-app-token-here # 數(shù)據(jù)庫連接信息如果使用PostgreSQL # DATABASE_URLpostgresql://user:passwordlocalhost:5432/lindy_db重要?jiǎng)?wù)必確保.env文件被添加到.gitignore中避免密鑰泄露。4. 核心流程拆解構(gòu)建一個(gè)簡(jiǎn)易的共享記憶系統(tǒng)我們將把一個(gè)復(fù)雜的“共享記憶”系統(tǒng)拆解為幾個(gè)可逐步實(shí)現(xiàn)的核心模塊。下圖描繪了其核心工作流與組件關(guān)系flowchart TD A[用戶向Slack Bot發(fā)送消息] -- B[Slack適配器br接收并格式化消息] B -- C{記憶路由器br判斷是否需要br存儲(chǔ)或檢索記憶} C -- “需要存儲(chǔ)新記憶” -- D[記憶編碼器br提取關(guān)鍵信息并向量化] D -- E[向量記憶庫brChromaDB存儲(chǔ)] E -- F[返回存儲(chǔ)成功] F -- G[結(jié)束流程] C -- “需要檢索相關(guān)記憶” -- H[記憶檢索器br從向量庫查詢相似記憶] H -- I[上下文裝配器br合并歷史記憶與當(dāng)前問題] I -- J[大語言模型brOpenAI/Claude等] J -- K[生成最終回復(fù)] K -- L[Slack適配器br發(fā)送回復(fù)給用戶] L -- G接下來我們深入每個(gè)環(huán)節(jié)的具體實(shí)現(xiàn)。4.1 模塊一記憶的抽象與存儲(chǔ)設(shè)計(jì)記憶不是簡(jiǎn)單的聊天記錄轉(zhuǎn)儲(chǔ)。我們需要設(shè)計(jì)結(jié)構(gòu)。# memory_models.py from pydantic import BaseModel, Field from datetime import datetime from typing import Optional, List from enum import Enum class MemoryType(str, Enum): FACT fact # 客觀事實(shí)如“項(xiàng)目截止日期是2024-06-30” PREFERENCE preference # 用戶偏好如“喜歡用Markdown格式匯報(bào)” DECISION decision # 決策記錄如“決定采用方案A因?yàn)樾阅芴嵘?0%” TASK_CONTEXT task_context # 任務(wù)上下文如“正在調(diào)試用戶登錄模塊” class MemoryEntity(BaseModel): 記憶實(shí)體的數(shù)據(jù)模型 id: Optional[str] None # 由數(shù)據(jù)庫生成 content: str Field(..., description記憶的文本內(nèi)容) memory_type: MemoryType Field(..., description記憶類型) source: str Field(..., description記憶來源如slack#channel-id) embedding: Optional[List[float]] None # 文本的向量表示 metadata: dict Field(default_factorydict) # 附加信息如用戶ID、時(shí)間戳、關(guān)聯(lián)實(shí)體 created_at: datetime Field(default_factorydatetime.utcnow) last_accessed_at: Optional[datetime] None class Config: use_enum_values True這個(gè)模型定義了記憶的“模樣”。embedding字段是為向量檢索準(zhǔn)備的metadata可以靈活擴(kuò)展。4.2 模塊二記憶的存儲(chǔ)與檢索后端我們需要實(shí)現(xiàn)記憶的“存”和“取”。這里使用ChromaDB作為向量存儲(chǔ)后端。# memory_store.py import chromadb from chromadb.config import Settings from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings from typing import List, Dict, Any import uuid class VectorMemoryStore: 基于ChromaDB的向量記憶存儲(chǔ) def __init__(self, persist_directory: str ./chroma_memory): # 初始化嵌入模型 self.embedding_function OpenAIEmbeddings(modeltext-embedding-3-small) # 初始化Chroma客戶端持久化到磁盤 self.client chromadb.PersistentClient(pathpersist_directory) # 獲取或創(chuàng)建集合類似于數(shù)據(jù)庫的表 self.collection self.client.get_or_create_collection( nameshared_memories, metadata{hnsw:space: cosine} # 使用余弦相似度進(jìn)行檢索 ) # LangChain的VectorStore封裝便于與LangChain生態(tài)集成 self.vectorstore Chroma( clientself.client, collection_nameshared_memories, embedding_functionself.embedding_function, ) def store_memory(self, memory_entity: MemoryEntity) - str: 存儲(chǔ)一條記憶 # 生成唯一ID memory_id str(uuid.uuid4()) # 為記憶內(nèi)容生成向量嵌入 embedding self.embedding_function.embed_query(memory_entity.content) # 準(zhǔn)備元數(shù)據(jù) metadata { type: memory_entity.memory_type, source: memory_entity.source, **memory_entity.metadata # 合并自定義元數(shù)據(jù) } # 存入ChromaDB self.collection.add( documents[memory_entity.content], metadatas[metadata], embeddings[embedding], ids[memory_id] ) return memory_id def search_similar_memories(self, query: str, filter_dict: Dict[str, Any] None, k: int 5) - List[Dict]: 檢索與查詢最相似的k條記憶 results self.vectorstore.similarity_search_with_score(query, kk, filterfilter_dict) memories [] for doc, score in results: memories.append({ content: doc.page_content, metadata: doc.metadata, relevance_score: score # 相似度分?jǐn)?shù)越低越相似取決于距離度量 }) return memories def delete_memory(self, memory_id: str): 刪除指定記憶 self.collection.delete(ids[memory_id])這個(gè)類封裝了向量數(shù)據(jù)庫的基本操作。search_similar_memories方法是核心它允許我們根據(jù)當(dāng)前對(duì)話的語義找到歷史上最相關(guān)的記憶。4.3 模塊三記憶的智能路由與管理不是所有對(duì)話都需要存儲(chǔ)也不是所有任務(wù)都需要加載全部記憶。我們需要一個(gè)“記憶路由器”。# memory_manager.py from langchain.chat_models import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from memory_models import MemoryEntity, MemoryType import json class MemoryManager: 管理記憶的存儲(chǔ)、檢索與路由邏輯 def __init__(self, vector_store: VectorMemoryStore, llm: ChatOpenAI): self.vector_store vector_store self.llm llm # 定義判斷是否需要存儲(chǔ)記憶的提示詞 self.store_decision_prompt ChatPromptTemplate.from_messages([ (system, 你是一個(gè)記憶管理助手。判斷以下對(duì)話內(nèi)容是否包含值得長(zhǎng)期記住的信息如關(guān)鍵決策、用戶明確偏好、重要事實(shí)。只回答YES或NO并簡(jiǎn)要說明原因用|分隔。), (human, 對(duì)話內(nèi)容{message}\n對(duì)話發(fā)生的場(chǎng)景{context}) ]) self.decision_chain self.store_decision_prompt | self.llm | StrOutputParser() def should_store_memory(self, message: str, context: str) - (bool, str): 利用LLM判斷當(dāng)前消息是否值得存儲(chǔ)為長(zhǎng)期記憶 response self.decision_chain.invoke({message: message, context: context}) try: decision, reason response.split(|, 1) return decision.strip().upper() YES, reason.strip() except: # 如果LLM輸出不符合格式默認(rèn)不存儲(chǔ) return False, 格式解析失敗 def store_if_necessary(self, message: str, context: str, source: str) - (bool, str): 條件性存儲(chǔ)記憶 should_store, reason self.should_store_memory(message, context) if should_store: # 簡(jiǎn)單起見這里存儲(chǔ)為FACT類型。實(shí)際可根據(jù)LLM進(jìn)一步分類。 memory MemoryEntity( contentf[來自{context}] {message}, memory_typeMemoryType.FACT, sourcesource, metadata{reason_to_store: reason} ) memory_id self.vector_store.store_memory(memory) return True, f已存儲(chǔ)記憶(ID:{memory_id})原因{reason} return False, f未存儲(chǔ)原因{reason} def retrieve_relevant_memories(self, current_query: str, user_id: str None, memory_type: str None) - List[Dict]: 檢索與當(dāng)前查詢相關(guān)的記憶可附加過濾器 filter_dict {} if user_id: filter_dict[user_id] user_id # 假設(shè)metadata里有user_id if memory_type: filter_dict[type] memory_type return self.vector_store.search_similar_memories(current_query, filter_dict, k3)這個(gè)管理器引入了LLM來做智能判斷這是實(shí)現(xiàn)“高質(zhì)量記憶”而非“垃圾信息堆積”的關(guān)鍵。should_store_memory方法防止了無價(jià)值信息的泛濫。4.4 模塊四與Slack的集成適配器這是讓系統(tǒng)在真實(shí)場(chǎng)景中跑起來的前端。我們創(chuàng)建一個(gè)簡(jiǎn)單的Slack機(jī)器人。# slack_bot.py import os from slack_bolt import App from slack_bolt.adapter.socket_mode import SocketModeHandler from dotenv import load_dotenv from memory_manager import MemoryManager from memory_store import VectorMemoryStore from langchain.chat_models import ChatOpenAI # 加載環(huán)境變量 load_dotenv() # 初始化核心組件 llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0.2) vector_store VectorMemoryStore() memory_manager MemoryManager(vector_store, llm) # 初始化Slack Bolt應(yīng)用 app App(tokenos.environ.get(SLACK_BOT_TOKEN)) app.event(message) def handle_message_events(body, say, logger): 處理Slack消息事件 event body.get(event, {}) # 忽略機(jī)器人自己的消息和消息變更事件 if event.get(subtype) or event.get(bot_id): return user event.get(user) channel event.get(channel) text event.get(text) thread_ts event.get(thread_ts) or event.get(ts) # 支持線程內(nèi)回復(fù) if not text: return logger.info(f收到來自用戶 {user} 的消息: {text}) # 1. 判斷并存儲(chǔ)記憶 context fSlack頻道: {channel}, 用戶: {user} stored, store_msg memory_manager.store_if_necessary(text, context, sourcefslack#{channel}) if stored: logger.info(store_msg) # 2. 檢索相關(guān)記憶 relevant_memories memory_manager.retrieve_relevant_memories(text, user_iduser) memory_context if relevant_memories: memory_context \n--- 相關(guān)歷史記憶 ---\n for mem in relevant_memories: memory_context f- {mem[content]} (相關(guān)性: {1 - mem[relevance_score]:.2f})\n memory_context -------------------\n logger.info(f檢索到 {len(relevant_memories)} 條相關(guān)記憶) # 3. 結(jié)合記憶生成回復(fù) prompt_with_memory f {memory_context} 用戶的最新消息是{text} 請(qǐng)根據(jù)上述歷史記憶如果有和當(dāng)前消息給出友好、專業(yè)的回復(fù)。 try: # 這里簡(jiǎn)化了實(shí)際應(yīng)使用更復(fù)雜的鏈或LLM調(diào)用 response llm.invoke(prompt_with_memory).content except Exception as e: logger.error(f調(diào)用LLM失敗: {e}) response 抱歉我暫時(shí)無法處理你的請(qǐng)求。 # 4. 在Slack中回復(fù) say(textresponse, thread_tsthread_ts) if __name__ __main__: # 使用Socket Mode連接適合開發(fā) handler SocketModeHandler(app, os.environ.get(SLACK_APP_TOKEN)) handler.start()這個(gè)機(jī)器人做了四件事監(jiān)聽消息、智能判斷是否存儲(chǔ)、檢索相關(guān)記憶、結(jié)合記憶生成回復(fù)。它構(gòu)成了一個(gè)最小可運(yùn)行的“共享記憶”AI助手。5. 運(yùn)行結(jié)果與效果驗(yàn)證讓我們啟動(dòng)這個(gè)系統(tǒng)看看它如何工作。啟動(dòng)Slack機(jī)器人python slack_bot.py如果一切正?？刂婆_(tái)會(huì)輸出類似[INFO] SocketModeClient started的日志表示機(jī)器人已成功連接到Slack。在Slack中與機(jī)器人互動(dòng)場(chǎng)景一告知偏好你lindy-bot 我更喜歡在每周五下午收到項(xiàng)目周報(bào)。機(jī)器人經(jīng)過LLM判斷這是一條“用戶偏好”類信息值得存儲(chǔ)存儲(chǔ)記憶成功。機(jī)器人回復(fù)“好的我已記下您偏好每周五下午接收項(xiàng)目周報(bào)?！眻?chǎng)景二基于記憶的對(duì)話幾天后你lindy-bot 這周的周報(bào)準(zhǔn)備好了嗎機(jī)器人內(nèi)部流程檢索記憶查詢“周報(bào)”找到之前存儲(chǔ)的“偏好周五下午”的記憶。組裝上下文“歷史記憶用戶偏好每周五下午接收項(xiàng)目周報(bào)。當(dāng)前問題這周的周報(bào)準(zhǔn)備好了嗎”LLM生成回復(fù)基于歷史記憶它知道用戶關(guān)心周報(bào)時(shí)間。機(jī)器人回復(fù)“根據(jù)之前的記錄您希望每周五下午查看周報(bào)。目前是周四周報(bào)正在整理中預(yù)計(jì)明天下午可以為您準(zhǔn)備好?！彬?yàn)證記憶存儲(chǔ)與檢索 你可以編寫一個(gè)簡(jiǎn)單的測(cè)試腳本直接查詢記憶庫。# test_memory.py from memory_store import VectorMemoryStore store VectorMemoryStore() memories store.search_similar_memories(周報(bào)什么時(shí)候發(fā), k2) for mem in memories: print(f內(nèi)容: {mem[content]}) print(f元數(shù)據(jù): {mem[metadata]}) print(f相似度分?jǐn)?shù): {mem[relevance_score]}) print(- * 30)運(yùn)行后你應(yīng)該能看到之前存儲(chǔ)的關(guān)于周報(bào)偏好的記憶被檢索出來并且有一個(gè)相似度分?jǐn)?shù)。效果驗(yàn)證的關(guān)鍵點(diǎn)持久性關(guān)閉機(jī)器人再重啟之前存儲(chǔ)的記憶依然存在因?yàn)镃hromaDB持久化到磁盤。相關(guān)性機(jī)器人能根據(jù)當(dāng)前問題的語義找到歷史上相關(guān)的對(duì)話片段。上下文增強(qiáng)機(jī)器人的回復(fù)體現(xiàn)了對(duì)歷史信息的“記憶”而非僅基于當(dāng)前單句的應(yīng)答。6. 常見問題與排查思路在構(gòu)建和運(yùn)行此類系統(tǒng)時(shí)你可能會(huì)遇到以下典型問題問題現(xiàn)象可能原因排查方式解決方案Slack機(jī)器人無響應(yīng)1. Socket Mode連接失敗2. Bot Token或App Token錯(cuò)誤3. 機(jī)器人未添加到頻道1. 檢查網(wǎng)絡(luò)查看slack_bot.py日志2. 在Slack API控制臺(tái)核對(duì)Token3. 在Slack頻道中/invite 你的機(jī)器人1. 確保網(wǎng)絡(luò)可訪問wss-primary.slack.com2. 重新安裝Slack App獲取新Token3. 邀請(qǐng)機(jī)器人到測(cè)試頻道記憶存儲(chǔ)失敗1. ChromaDB集合未正確創(chuàng)建2. 嵌入模型API調(diào)用失敗3. 磁盤權(quán)限不足1. 檢查chroma_memory目錄是否生成2. 檢查OpenAI API Key余額與網(wǎng)絡(luò)3. 檢查項(xiàng)目目錄寫入權(quán)限1. 刪除chroma_memory目錄重啟2. 更換API Key或使用本地嵌入模型3. 更改項(xiàng)目路徑或目錄權(quán)限記憶檢索結(jié)果不相關(guān)1. 嵌入模型不適合領(lǐng)域2. 檢索參數(shù)k太大或太小3. 記憶內(nèi)容質(zhì)量差存儲(chǔ)了噪音1. 用不同查詢測(cè)試相似度2. 調(diào)整k值觀察結(jié)果變化3. 查看存儲(chǔ)的記憶內(nèi)容1. 嘗試不同的嵌入模型如text-embedding-ada-0022. 根據(jù)場(chǎng)景調(diào)整k通常3-103. 優(yōu)化should_store_memory的判斷邏輯LLM回復(fù)未使用記憶1. 記憶檢索為空2. 提示詞prompt組裝有誤3. 記憶上下文過長(zhǎng)被截?cái)?. 打印relevant_memories變量2. 檢查prompt_with_memory的最終字符串3. 檢查L(zhǎng)LM的上下文窗口限制1. 確保查詢文本與記憶內(nèi)容語義相關(guān)2. 調(diào)試提示詞格式確保記憶部分被清晰標(biāo)注3. 對(duì)檢索到的記憶進(jìn)行摘要或選擇性注入性能緩慢1. 向量檢索未建索引或數(shù)據(jù)量大2. LLM調(diào)用延遲高3. 每次請(qǐng)求都重新初始化組件1. 觀察ChromaDB查詢耗時(shí)2. 檢查OpenAI API響應(yīng)時(shí)間3. 檢查代碼結(jié)構(gòu)1. 確保ChromaDB使用HNSW等索引2. 考慮緩存頻繁查詢的記憶或使用更快的LLM3. 將向量存儲(chǔ)、LLM客戶端等設(shè)為全局單例7. 最佳實(shí)踐與工程建議將共享記憶系統(tǒng)投入實(shí)際生產(chǎn)環(huán)境需要考慮遠(yuǎn)比Demo更多的問題。以下是一些關(guān)鍵建議1. 記憶的粒度與分類不要存儲(chǔ)所有對(duì)話這會(huì)導(dǎo)致記憶庫迅速膨脹檢索效率和質(zhì)量下降。必須像MemoryManager那樣設(shè)計(jì)嚴(yán)格的過濾和摘要邏輯。設(shè)計(jì)更精細(xì)的記憶類型除了示例中的幾種還可以考慮PROCEDURE工作流程、RELATIONSHIP實(shí)體關(guān)系、INSIGHT分析洞察等。不同類型的記憶可以采用不同的存儲(chǔ)和檢索策略。實(shí)施記憶摘要對(duì)于長(zhǎng)對(duì)話存儲(chǔ)前先用LLM生成一個(gè)簡(jiǎn)潔的摘要而不是原始文本。這能極大節(jié)省存儲(chǔ)空間并提升檢索質(zhì)量。2. 檢索策略的優(yōu)化混合檢索不要只依賴向量相似度。結(jié)合關(guān)鍵詞過濾如時(shí)間范圍、用戶ID、記憶類型進(jìn)行混合檢索結(jié)果更精準(zhǔn)。重排序初步檢索出Top-K個(gè)結(jié)果后可以用一個(gè)更小的、更快的“重排序模型”對(duì)它們進(jìn)行精排將最相關(guān)的放在前面。記憶衰減與更新為記憶設(shè)計(jì)“新鮮度”或“訪問頻率”權(quán)重。長(zhǎng)期未被訪問或已被證偽的記憶應(yīng)被降權(quán)或歸檔而非刪除。3. 系統(tǒng)架構(gòu)與擴(kuò)展服務(wù)化將記憶存儲(chǔ)、檢索、管理模塊拆分為獨(dú)立的微服務(wù)如Memory Service通過REST或gRPC對(duì)外提供API。這樣前端Slack Bot、Web UI可以輕量化。多租戶與隔離為不同的團(tuán)隊(duì)、項(xiàng)目或用戶設(shè)計(jì)嚴(yán)格的數(shù)據(jù)隔離。在元數(shù)據(jù)中增加tenant_id、project_id字段并在檢索時(shí)強(qiáng)制過濾。監(jiān)控與可觀測(cè)性記錄關(guān)鍵指標(biāo)記憶存儲(chǔ)量、檢索耗時(shí)、檢索命中率、LLM調(diào)用延遲、用戶反饋如“這條記憶有用嗎”按鈕。這有助于持續(xù)優(yōu)化系統(tǒng)。4. 安全與隱私敏感信息過濾在記憶存儲(chǔ)前必須經(jīng)過一層敏感信息檢測(cè)和脫敏處理如自動(dòng)識(shí)別并遮蓋手機(jī)號(hào)、郵箱、密鑰。用戶控制權(quán)提供用戶界面讓用戶查看、編輯、刪除AI存儲(chǔ)的關(guān)于自己的記憶。這是合規(guī)如GDPR和建立信任的關(guān)鍵。訪問審計(jì)記錄誰哪個(gè)用戶或Agent在什么時(shí)候訪問了哪些記憶用于安全審計(jì)。5. 與現(xiàn)有工作流集成超越Slack將記憶系統(tǒng)與GitHub、Jira、Notion、Confluence等工具連接。例如當(dāng)AI在代碼評(píng)審中學(xué)習(xí)到一個(gè)團(tuán)隊(duì)的編碼規(guī)范時(shí)這條記憶應(yīng)該能被后續(xù)的代碼生成任務(wù)所用。主動(dòng)記憶觸發(fā)除了被動(dòng)響應(yīng)用戶消息系統(tǒng)可以設(shè)置定時(shí)任務(wù)主動(dòng)回顧和整理記憶或基于記憶向用戶推送提醒如“根據(jù)上周的討論您今天需要決定方案選型”。8. 總結(jié)與后續(xù)學(xué)習(xí)方向通過本文的拆解我們實(shí)現(xiàn)了一個(gè)具備“共享記憶”核心能力的AI助手原型。它不再是健忘的對(duì)話者而是能夠積累知識(shí)、并在需要時(shí)準(zhǔn)確回憶的協(xié)作者。這不僅僅是技術(shù)的疊加更是對(duì)AI應(yīng)用范式的重新思考從追求單次交互的驚艷轉(zhuǎn)向構(gòu)建長(zhǎng)期協(xié)作的信任和效率?；仡橪indy項(xiàng)目提出的愿景其核心價(jià)值在于將“記憶”這個(gè)抽象概念工程化為可落地、可擴(kuò)展的系統(tǒng)組件。我們通過向量數(shù)據(jù)庫、LLM智能路由和Slack集成驗(yàn)證了這一路徑的可行性。對(duì)于想要繼續(xù)深入的開發(fā)者下一步可以探索的方向深入研究開源項(xiàng)目關(guān)注類似my_ai_town、LangChain、LlamaIndex中關(guān)于“長(zhǎng)期記憶”、“代理記憶”的社區(qū)討論和實(shí)現(xiàn)方案汲取更成熟的設(shè)計(jì)模式。探索更復(fù)雜的記憶結(jié)構(gòu)嘗試用知識(shí)圖譜來存儲(chǔ)記憶處理實(shí)體、關(guān)系、事件等更結(jié)構(gòu)化的信息實(shí)現(xiàn)邏輯推理而不僅僅是語義相似。實(shí)現(xiàn)多Agent記憶共享構(gòu)建一個(gè)記憶中心讓多個(gè)具有不同技能的AI Agent如一個(gè)負(fù)責(zé)數(shù)據(jù)分析一個(gè)負(fù)責(zé)編寫文檔可以共同讀寫實(shí)現(xiàn)真正的協(xié)同工作。關(guān)注模型本身的進(jìn)步OpenAI的“記憶”功能、Claude的“項(xiàng)目”功能都表明大模型廠商正在原生層面解決此問題。理解其API和設(shè)計(jì)哲學(xué)與外部記憶系統(tǒng)結(jié)合可能是更優(yōu)解。構(gòu)建一個(gè)實(shí)用的共享記憶系統(tǒng)挑戰(zhàn)不在于單點(diǎn)技術(shù)而在于對(duì)業(yè)務(wù)場(chǎng)景的深度理解、對(duì)記憶價(jià)值的精準(zhǔn)判斷以及在性能、成本、隱私之間的復(fù)雜權(quán)衡。希望本文提供的思路和代碼能成為你探索這個(gè)迷人領(lǐng)域的起點(diǎn)。建議收藏本文在動(dòng)手實(shí)踐時(shí)對(duì)照“常見問題”和“最佳實(shí)踐”部分相信能幫你避開不少彎路。 突破AI上下文限制：構(gòu)建基于共享記憶的智能協(xié)作系統(tǒng)
你是否遇到過這樣的場(chǎng)景當(dāng)你與一個(gè)AI助手在Slack或Teams上討論一個(gè)復(fù)雜項(xiàng)目時(shí)每次對(duì)話重啟它都像得了“健忘癥”需要你重新解釋一遍背景、目標(biāo)和之前的決策或者當(dāng)你試圖讓AI Agent處理一個(gè)跨越數(shù)天、涉及多個(gè)文檔和對(duì)話的任務(wù)時(shí)它很快就因?yàn)椤吧舷挛拇翱凇焙谋M而無法繼續(xù)這正是當(dāng)前AI應(yīng)用特別是AI Agent領(lǐng)域最核心的瓶頸之一上下文限制。無論是Claude的200K還是GPT-4的128K本質(zhì)上都是“一次性”的短期記憶。一旦對(duì)話結(jié)束或窗口填滿所有精心構(gòu)建的上下文就消失了。這導(dǎo)致AI無法真正像人類一樣在長(zhǎng)期、復(fù)雜的協(xié)作中積累知識(shí)和經(jīng)驗(yàn)。今天要介紹的項(xiàng)目Lindy正是為了解決這個(gè)問題而生。它不是一個(gè)新的大模型而是一個(gè)基于共享記憶的AI協(xié)作平臺(tái)。它的核心判斷非常清晰AI的長(zhǎng)期價(jià)值不在于單次問答的聰明而在于持續(xù)協(xié)作中的“成長(zhǎng)”和“傳承”。Lindy試圖通過為AI Agent構(gòu)建一個(gè)可持久化、可共享的“記憶系統(tǒng)”來突破上下文窗口的物理限制讓AI真正融入工作流。如果你正在開發(fā)或集成AI Agent為上下文管理而頭疼或者你是一個(gè)團(tuán)隊(duì)管理者希望AI能成為穩(wěn)定的“數(shù)字同事”而非臨時(shí)的“問答機(jī)器”那么這篇文章將為你提供一個(gè)全新的技術(shù)視角和一套可落地的實(shí)踐思路。我們將深入拆解Lindy的設(shè)計(jì)理念、核心組件并通過一個(gè)模擬的Slack集成示例展示如何讓AI擁有“共享記憶”。1. 這篇文章真正要解決的問題為什么上下文瓶頸是AI Agent的“阿喀琉斯之踵”在深入Lindy之前我們必須先理解“上下文瓶頸”到底卡住了什么。這不僅僅是技術(shù)參數(shù)問題更是AI應(yīng)用從“玩具”走向“工具”的關(guān)鍵障礙。首先上下文窗口的本質(zhì)是什么你可以把它想象成AI工作時(shí)的“桌面”。桌面越大上下文窗口越長(zhǎng)它能同時(shí)攤開的參考資料歷史對(duì)話、文檔、代碼就越多做出的判斷就越連貫、準(zhǔn)確。但無論桌面多大會(huì)議結(jié)束對(duì)話結(jié)束、下班關(guān)機(jī)會(huì)話終止桌面就會(huì)被清空。第二天上班一切又要從頭開始。其次這個(gè)瓶頸導(dǎo)致了三大現(xiàn)實(shí)困境任務(wù)無法延續(xù)一個(gè)需要多步驟、跨時(shí)區(qū)的數(shù)據(jù)分析或代碼審查任務(wù)AI無法記住中間的臨時(shí)結(jié)論和修改記錄。知識(shí)無法沉淀團(tuán)隊(duì)在與AI的協(xié)作中產(chǎn)生的寶貴經(jīng)驗(yàn)、決策邏輯、定制化知識(shí)隨著對(duì)話結(jié)束而煙消云散無法形成組織的“數(shù)字資產(chǎn)”。協(xié)作無法展開多個(gè)AI Agent之間或者AI與多個(gè)人類成員之間缺乏一個(gè)公共的、可同步的“記憶黑板”導(dǎo)致信息孤島和重復(fù)勞動(dòng)。Lindy的提出正是基于一個(gè)深刻的洞察未來的AI不是單次服務(wù)的調(diào)用者而是持續(xù)參與流程的協(xié)作者。因此它需要的不是更大的“一次性桌面”而是一個(gè)專屬的、可隨時(shí)存取的“文件柜”和“會(huì)議紀(jì)要庫”——這就是“共享記憶”Shared Memory的概念。本文將帶你從原理到實(shí)踐完整走通“共享記憶”的構(gòu)建之路。你會(huì)看到Lindy如何將記憶抽象為可存儲(chǔ)、可檢索的結(jié)構(gòu)化數(shù)據(jù)如何通過Slack等日常工具無縫集成以及在實(shí)際開發(fā)中你會(huì)遇到哪些“坑”和最佳實(shí)踐。2. 基礎(chǔ)概念與核心原理從“上下文”到“共享記憶”要理解Lindy需要先厘清幾個(gè)容易混淆的關(guān)鍵概念。2.1 上下文Context vs. 記憶Memory上下文Context Window這是大模型的技術(shù)參數(shù)。指單次請(qǐng)求中模型能“看到”的文本包括你的問題、歷史對(duì)話、系統(tǒng)指令等的最大長(zhǎng)度。它是臨時(shí)的、易失的與本次請(qǐng)求的生命周期綁定。記憶Memory這是一個(gè)更高層的應(yīng)用概念。指AI為了完成長(zhǎng)期目標(biāo)需要持久化存儲(chǔ)和回憶的信息。它超越了單次請(qǐng)求是跨會(huì)話、跨任務(wù)存在的。類比上下文就像你電腦的內(nèi)存RAM處理當(dāng)前任務(wù)時(shí)高速存取關(guān)機(jī)即丟失。記憶則像硬盤HDD/SSD或云盤用于長(zhǎng)期存儲(chǔ)隨時(shí)可以加載到內(nèi)存中使用。2.2 共享記憶Shared Memory的核心思想Lindy的“共享記憶”包含兩層含義持久化Persistence將重要的交互信息如用戶偏好、任務(wù)結(jié)論、學(xué)到的規(guī)則從易失的上下文窗口中提取出來存儲(chǔ)到外部數(shù)據(jù)庫或向量庫中。共享化Sharing這份存儲(chǔ)的記憶可以被同一個(gè)AI Agent在不同時(shí)間點(diǎn)訪問也可以被同一個(gè)團(tuán)隊(duì)內(nèi)的不同AI Agent訪問實(shí)現(xiàn)知識(shí)和狀態(tài)的同步。它的技術(shù)原理可以簡(jiǎn)化為一個(gè)循環(huán)[AI交互] - [記憶提取與編碼] - [外部存儲(chǔ)記憶庫] - [記憶檢索與解碼] - [注入新上下文] - [更智能的AI交互]這個(gè)循環(huán)打破了“一次一清空”的模式讓AI的能力隨時(shí)間推移而增強(qiáng)。2.3 Lindy的架構(gòu)角色根據(jù)有限的資料推斷Lindy很可能扮演以下角色記憶管理層提供API和SDK讓開發(fā)者能方便地定義“什么信息需要被記住”如決策、事實(shí)、用戶反饋以及如何存儲(chǔ)數(shù)據(jù)庫、向量索引。上下文裝配層在AI執(zhí)行任務(wù)前根據(jù)當(dāng)前任務(wù)目標(biāo)自動(dòng)從記憶庫中檢索最相關(guān)的記憶片段并將其作為系統(tǒng)提示詞或上下文的一部分注入給大模型。集成中間件與Slack、Discord等通訊工具深度集成監(jiān)聽對(duì)話自動(dòng)觸發(fā)記憶的存儲(chǔ)和檢索流程讓用戶無感知地使用。理解了這些概念我們就能明白Lindy的目標(biāo)是將AI的“智力”從短暫的上下文約束中解放出來使其建立在可積累、可共享的記憶基石之上。3. 環(huán)境準(zhǔn)備與前置條件由于Lindy是一個(gè)較新的概念和項(xiàng)目從my_ai_town等關(guān)聯(lián)信息看可能處于早期開源階段我們無法獲得官方的、穩(wěn)定的安裝包。因此本節(jié)將基于共享記憶的通用實(shí)現(xiàn)思路規(guī)劃一個(gè)可行的技術(shù)棧和環(huán)境你可以將其視為構(gòu)建“類Lindy”系統(tǒng)或理解其原理的藍(lán)圖。核心技術(shù)棧選擇編程語言Python 3.9。因其在AI和快速開發(fā)領(lǐng)域的生態(tài)優(yōu)勢(shì)。大模型接口OpenAI GPT API 或 Anthropic Claude API需自備API Key。也可用開源的Llama 3等本地模型但需部署相應(yīng)的推理服務(wù)。記憶存儲(chǔ)向量數(shù)據(jù)庫用于存儲(chǔ)非結(jié)構(gòu)化記憶對(duì)話片段、文檔摘要并支持語義檢索。推薦ChromaDB輕量、簡(jiǎn)單或Weaviate功能更全。傳統(tǒng)數(shù)據(jù)庫用于存儲(chǔ)結(jié)構(gòu)化記憶用戶ID、任務(wù)狀態(tài)、固定鍵值對(duì)。推薦SQLite開發(fā)測(cè)試或PostgreSQL生產(chǎn)環(huán)境。應(yīng)用框架LangChain或LlamaIndex。它們提供了構(gòu)建AI應(yīng)用包括記憶系統(tǒng)的高級(jí)抽象和工具鏈能極大簡(jiǎn)化開發(fā)。本文將使用LangChain進(jìn)行演示。通訊平臺(tái)集成Slack API。我們將模擬一個(gè)Slack機(jī)器人作為記憶交互的前端。環(huán)境管理使用Conda或venv創(chuàng)建獨(dú)立的Python環(huán)境。環(huán)境搭建步驟創(chuàng)建并激活Python虛擬環(huán)境。# 使用 venv python -m venv lindy_env source lindy_env/bin/activate # Linux/Mac # lindy_env\Scripts\activate # Windows安裝核心依賴。pip install langchain langchain-openai langchain-community chromadb slack-sdk python-dotenvlangchain: 核心框架。langchain-openai: OpenAI模型集成。langchain-community: 包含社區(qū)貢獻(xiàn)的組件如一些記憶存儲(chǔ)后端。chromadb: 向量數(shù)據(jù)庫。slack-sdk: Slack官方SDK。python-dotenv: 管理環(huán)境變量。準(zhǔn)備配置文件。在項(xiàng)目根目錄創(chuàng)建.env文件用于存放敏感信息。# .env 文件示例 OPENAI_API_KEYsk-your-openai-api-key-here SLACK_BOT_TOKENxoxb-your-slack-bot-token-here SLACK_APP_TOKENxapp-your-slack-app-token-here # 數(shù)據(jù)庫連接信息如果使用PostgreSQL # DATABASE_URLpostgresql://user:passwordlocalhost:5432/lindy_db重要?jiǎng)?wù)必確保.env文件被添加到.gitignore中避免密鑰泄露。4. 核心流程拆解構(gòu)建一個(gè)簡(jiǎn)易的共享記憶系統(tǒng)我們將把一個(gè)復(fù)雜的“共享記憶”系統(tǒng)拆解為幾個(gè)可逐步實(shí)現(xiàn)的核心模塊。下圖描繪了其核心工作流與組件關(guān)系flowchart TD A[用戶向Slack Bot發(fā)送消息] -- B[Slack適配器br接收并格式化消息] B -- C{記憶路由器br判斷是否需要br存儲(chǔ)或檢索記憶} C -- “需要存儲(chǔ)新記憶” -- D[記憶編碼器br提取關(guān)鍵信息并向量化] D -- E[向量記憶庫brChromaDB存儲(chǔ)] E -- F[返回存儲(chǔ)成功] F -- G[結(jié)束流程] C -- “需要檢索相關(guān)記憶” -- H[記憶檢索器br從向量庫查詢相似記憶] H -- I[上下文裝配器br合并歷史記憶與當(dāng)前問題] I -- J[大語言模型brOpenAI/Claude等] J -- K[生成最終回復(fù)] K -- L[Slack適配器br發(fā)送回復(fù)給用戶] L -- G接下來我們深入每個(gè)環(huán)節(jié)的具體實(shí)現(xiàn)。4.1 模塊一記憶的抽象與存儲(chǔ)設(shè)計(jì)記憶不是簡(jiǎn)單的聊天記錄轉(zhuǎn)儲(chǔ)。我們需要設(shè)計(jì)結(jié)構(gòu)。# memory_models.py from pydantic import BaseModel, Field from datetime import datetime from typing import Optional, List from enum import Enum class MemoryType(str, Enum): FACT fact # 客觀事實(shí)如“項(xiàng)目截止日期是2024-06-30” PREFERENCE preference # 用戶偏好如“喜歡用Markdown格式匯報(bào)” DECISION decision # 決策記錄如“決定采用方案A因?yàn)樾阅芴嵘?0%” TASK_CONTEXT task_context # 任務(wù)上下文如“正在調(diào)試用戶登錄模塊” class MemoryEntity(BaseModel): 記憶實(shí)體的數(shù)據(jù)模型 id: Optional[str] None # 由數(shù)據(jù)庫生成 content: str Field(..., description記憶的文本內(nèi)容) memory_type: MemoryType Field(..., description記憶類型) source: str Field(..., description記憶來源如slack#channel-id) embedding: Optional[List[float]] None # 文本的向量表示 metadata: dict Field(default_factorydict) # 附加信息如用戶ID、時(shí)間戳、關(guān)聯(lián)實(shí)體 created_at: datetime Field(default_factorydatetime.utcnow) last_accessed_at: Optional[datetime] None class Config: use_enum_values True這個(gè)模型定義了記憶的“模樣”。embedding字段是為向量檢索準(zhǔn)備的metadata可以靈活擴(kuò)展。4.2 模塊二記憶的存儲(chǔ)與檢索后端我們需要實(shí)現(xiàn)記憶的“存”和“取”。這里使用ChromaDB作為向量存儲(chǔ)后端。# memory_store.py import chromadb from chromadb.config import Settings from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings from typing import List, Dict, Any import uuid class VectorMemoryStore: 基于ChromaDB的向量記憶存儲(chǔ) def __init__(self, persist_directory: str ./chroma_memory): # 初始化嵌入模型 self.embedding_function OpenAIEmbeddings(modeltext-embedding-3-small) # 初始化Chroma客戶端持久化到磁盤 self.client chromadb.PersistentClient(pathpersist_directory) # 獲取或創(chuàng)建集合類似于數(shù)據(jù)庫的表 self.collection self.client.get_or_create_collection( nameshared_memories, metadata{hnsw:space: cosine} # 使用余弦相似度進(jìn)行檢索 ) # LangChain的VectorStore封裝便于與LangChain生態(tài)集成 self.vectorstore Chroma( clientself.client, collection_nameshared_memories, embedding_functionself.embedding_function, ) def store_memory(self, memory_entity: MemoryEntity) - str: 存儲(chǔ)一條記憶 # 生成唯一ID memory_id str(uuid.uuid4()) # 為記憶內(nèi)容生成向量嵌入 embedding self.embedding_function.embed_query(memory_entity.content) # 準(zhǔn)備元數(shù)據(jù) metadata { type: memory_entity.memory_type, source: memory_entity.source, **memory_entity.metadata # 合并自定義元數(shù)據(jù) } # 存入ChromaDB self.collection.add( documents[memory_entity.content], metadatas[metadata], embeddings[embedding], ids[memory_id] ) return memory_id def search_similar_memories(self, query: str, filter_dict: Dict[str, Any] None, k: int 5) - List[Dict]: 檢索與查詢最相似的k條記憶 results self.vectorstore.similarity_search_with_score(query, kk, filterfilter_dict) memories [] for doc, score in results: memories.append({ content: doc.page_content, metadata: doc.metadata, relevance_score: score # 相似度分?jǐn)?shù)越低越相似取決于距離度量 }) return memories def delete_memory(self, memory_id: str): 刪除指定記憶 self.collection.delete(ids[memory_id])這個(gè)類封裝了向量數(shù)據(jù)庫的基本操作。search_similar_memories方法是核心它允許我們根據(jù)當(dāng)前對(duì)話的語義找到歷史上最相關(guān)的記憶。4.3 模塊三記憶的智能路由與管理不是所有對(duì)話都需要存儲(chǔ)也不是所有任務(wù)都需要加載全部記憶。我們需要一個(gè)“記憶路由器”。# memory_manager.py from langchain.chat_models import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from memory_models import MemoryEntity, MemoryType import json class MemoryManager: 管理記憶的存儲(chǔ)、檢索與路由邏輯 def __init__(self, vector_store: VectorMemoryStore, llm: ChatOpenAI): self.vector_store vector_store self.llm llm # 定義判斷是否需要存儲(chǔ)記憶的提示詞 self.store_decision_prompt ChatPromptTemplate.from_messages([ (system, 你是一個(gè)記憶管理助手。判斷以下對(duì)話內(nèi)容是否包含值得長(zhǎng)期記住的信息如關(guān)鍵決策、用戶明確偏好、重要事實(shí)。只回答YES或NO并簡(jiǎn)要說明原因用|分隔。), (human, 對(duì)話內(nèi)容{message}\n對(duì)話發(fā)生的場(chǎng)景{context}) ]) self.decision_chain self.store_decision_prompt | self.llm | StrOutputParser() def should_store_memory(self, message: str, context: str) - (bool, str): 利用LLM判斷當(dāng)前消息是否值得存儲(chǔ)為長(zhǎng)期記憶 response self.decision_chain.invoke({message: message, context: context}) try: decision, reason response.split(|, 1) return decision.strip().upper() YES, reason.strip() except: # 如果LLM輸出不符合格式默認(rèn)不存儲(chǔ) return False, 格式解析失敗 def store_if_necessary(self, message: str, context: str, source: str) - (bool, str): 條件性存儲(chǔ)記憶 should_store, reason self.should_store_memory(message, context) if should_store: # 簡(jiǎn)單起見這里存儲(chǔ)為FACT類型。實(shí)際可根據(jù)LLM進(jìn)一步分類。 memory MemoryEntity( contentf[來自{context}] {message}, memory_typeMemoryType.FACT, sourcesource, metadata{reason_to_store: reason} ) memory_id self.vector_store.store_memory(memory) return True, f已存儲(chǔ)記憶(ID:{memory_id})原因{reason} return False, f未存儲(chǔ)原因{reason} def retrieve_relevant_memories(self, current_query: str, user_id: str None, memory_type: str None) - List[Dict]: 檢索與當(dāng)前查詢相關(guān)的記憶可附加過濾器 filter_dict {} if user_id: filter_dict[user_id] user_id # 假設(shè)metadata里有user_id if memory_type: filter_dict[type] memory_type return self.vector_store.search_similar_memories(current_query, filter_dict, k3)這個(gè)管理器引入了LLM來做智能判斷這是實(shí)現(xiàn)“高質(zhì)量記憶”而非“垃圾信息堆積”的關(guān)鍵。should_store_memory方法防止了無價(jià)值信息的泛濫。4.4 模塊四與Slack的集成適配器這是讓系統(tǒng)在真實(shí)場(chǎng)景中跑起來的前端。我們創(chuàng)建一個(gè)簡(jiǎn)單的Slack機(jī)器人。# slack_bot.py import os from slack_bolt import App from slack_bolt.adapter.socket_mode import SocketModeHandler from dotenv import load_dotenv from memory_manager import MemoryManager from memory_store import VectorMemoryStore from langchain.chat_models import ChatOpenAI # 加載環(huán)境變量 load_dotenv() # 初始化核心組件 llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0.2) vector_store VectorMemoryStore() memory_manager MemoryManager(vector_store, llm) # 初始化Slack Bolt應(yīng)用 app App(tokenos.environ.get(SLACK_BOT_TOKEN)) app.event(message) def handle_message_events(body, say, logger): 處理Slack消息事件 event body.get(event, {}) # 忽略機(jī)器人自己的消息和消息變更事件 if event.get(subtype) or event.get(bot_id): return user event.get(user) channel event.get(channel) text event.get(text) thread_ts event.get(thread_ts) or event.get(ts) # 支持線程內(nèi)回復(fù) if not text: return logger.info(f收到來自用戶 {user} 的消息: {text}) # 1. 判斷并存儲(chǔ)記憶 context fSlack頻道: {channel}, 用戶: {user} stored, store_msg memory_manager.store_if_necessary(text, context, sourcefslack#{channel}) if stored: logger.info(store_msg) # 2. 檢索相關(guān)記憶 relevant_memories memory_manager.retrieve_relevant_memories(text, user_iduser) memory_context if relevant_memories: memory_context \n--- 相關(guān)歷史記憶 ---\n for mem in relevant_memories: memory_context f- {mem[content]} (相關(guān)性: {1 - mem[relevance_score]:.2f})\n memory_context -------------------\n logger.info(f檢索到 {len(relevant_memories)} 條相關(guān)記憶) # 3. 結(jié)合記憶生成回復(fù) prompt_with_memory f {memory_context} 用戶的最新消息是{text} 請(qǐng)根據(jù)上述歷史記憶如果有和當(dāng)前消息給出友好、專業(yè)的回復(fù)。 try: # 這里簡(jiǎn)化了實(shí)際應(yīng)使用更復(fù)雜的鏈或LLM調(diào)用 response llm.invoke(prompt_with_memory).content except Exception as e: logger.error(f調(diào)用LLM失敗: {e}) response 抱歉我暫時(shí)無法處理你的請(qǐng)求。 # 4. 在Slack中回復(fù) say(textresponse, thread_tsthread_ts) if __name__ __main__: # 使用Socket Mode連接適合開發(fā) handler SocketModeHandler(app, os.environ.get(SLACK_APP_TOKEN)) handler.start()這個(gè)機(jī)器人做了四件事監(jiān)聽消息、智能判斷是否存儲(chǔ)、檢索相關(guān)記憶、結(jié)合記憶生成回復(fù)。它構(gòu)成了一個(gè)最小可運(yùn)行的“共享記憶”AI助手。5. 運(yùn)行結(jié)果與效果驗(yàn)證讓我們啟動(dòng)這個(gè)系統(tǒng)看看它如何工作。啟動(dòng)Slack機(jī)器人python slack_bot.py如果一切正?？刂婆_(tái)會(huì)輸出類似[INFO] SocketModeClient started的日志表示機(jī)器人已成功連接到Slack。在Slack中與機(jī)器人互動(dòng)場(chǎng)景一告知偏好你lindy-bot 我更喜歡在每周五下午收到項(xiàng)目周報(bào)。機(jī)器人經(jīng)過LLM判斷這是一條“用戶偏好”類信息值得存儲(chǔ)存儲(chǔ)記憶成功。機(jī)器人回復(fù)“好的我已記下您偏好每周五下午接收項(xiàng)目周報(bào)?！眻?chǎng)景二基于記憶的對(duì)話幾天后你lindy-bot 這周的周報(bào)準(zhǔn)備好了嗎機(jī)器人內(nèi)部流程檢索記憶查詢“周報(bào)”找到之前存儲(chǔ)的“偏好周五下午”的記憶。組裝上下文“歷史記憶用戶偏好每周五下午接收項(xiàng)目周報(bào)。當(dāng)前問題這周的周報(bào)準(zhǔn)備好了嗎”LLM生成回復(fù)基于歷史記憶它知道用戶關(guān)心周報(bào)時(shí)間。機(jī)器人回復(fù)“根據(jù)之前的記錄您希望每周五下午查看周報(bào)。目前是周四周報(bào)正在整理中預(yù)計(jì)明天下午可以為您準(zhǔn)備好?！彬?yàn)證記憶存儲(chǔ)與檢索 你可以編寫一個(gè)簡(jiǎn)單的測(cè)試腳本直接查詢記憶庫。# test_memory.py from memory_store import VectorMemoryStore store VectorMemoryStore() memories store.search_similar_memories(周報(bào)什么時(shí)候發(fā), k2) for mem in memories: print(f內(nèi)容: {mem[content]}) print(f元數(shù)據(jù): {mem[metadata]}) print(f相似度分?jǐn)?shù): {mem[relevance_score]}) print(- * 30)運(yùn)行后你應(yīng)該能看到之前存儲(chǔ)的關(guān)于周報(bào)偏好的記憶被檢索出來并且有一個(gè)相似度分?jǐn)?shù)。效果驗(yàn)證的關(guān)鍵點(diǎn)持久性關(guān)閉機(jī)器人再重啟之前存儲(chǔ)的記憶依然存在因?yàn)镃hromaDB持久化到磁盤。相關(guān)性機(jī)器人能根據(jù)當(dāng)前問題的語義找到歷史上相關(guān)的對(duì)話片段。上下文增強(qiáng)機(jī)器人的回復(fù)體現(xiàn)了對(duì)歷史信息的“記憶”而非僅基于當(dāng)前單句的應(yīng)答。6. 常見問題與排查思路在構(gòu)建和運(yùn)行此類系統(tǒng)時(shí)你可能會(huì)遇到以下典型問題問題現(xiàn)象可能原因排查方式解決方案Slack機(jī)器人無響應(yīng)1. Socket Mode連接失敗2. Bot Token或App Token錯(cuò)誤3. 機(jī)器人未添加到頻道1. 檢查網(wǎng)絡(luò)查看slack_bot.py日志2. 在Slack API控制臺(tái)核對(duì)Token3. 在Slack頻道中/invite 你的機(jī)器人1. 確保網(wǎng)絡(luò)可訪問wss-primary.slack.com2. 重新安裝Slack App獲取新Token3. 邀請(qǐng)機(jī)器人到測(cè)試頻道記憶存儲(chǔ)失敗1. ChromaDB集合未正確創(chuàng)建2. 嵌入模型API調(diào)用失敗3. 磁盤權(quán)限不足1. 檢查chroma_memory目錄是否生成2. 檢查OpenAI API Key余額與網(wǎng)絡(luò)3. 檢查項(xiàng)目目錄寫入權(quán)限1. 刪除chroma_memory目錄重啟2. 更換API Key或使用本地嵌入模型3. 更改項(xiàng)目路徑或目錄權(quán)限記憶檢索結(jié)果不相關(guān)1. 嵌入模型不適合領(lǐng)域2. 檢索參數(shù)k太大或太小3. 記憶內(nèi)容質(zhì)量差存儲(chǔ)了噪音1. 用不同查詢測(cè)試相似度2. 調(diào)整k值觀察結(jié)果變化3. 查看存儲(chǔ)的記憶內(nèi)容1. 嘗試不同的嵌入模型如text-embedding-ada-0022. 根據(jù)場(chǎng)景調(diào)整k通常3-103. 優(yōu)化should_store_memory的判斷邏輯LLM回復(fù)未使用記憶1. 記憶檢索為空2. 提示詞prompt組裝有誤3. 記憶上下文過長(zhǎng)被截?cái)?. 打印relevant_memories變量2. 檢查prompt_with_memory的最終字符串3. 檢查L(zhǎng)LM的上下文窗口限制1. 確保查詢文本與記憶內(nèi)容語義相關(guān)2. 調(diào)試提示詞格式確保記憶部分被清晰標(biāo)注3. 對(duì)檢索到的記憶進(jìn)行摘要或選擇性注入性能緩慢1. 向量檢索未建索引或數(shù)據(jù)量大2. LLM調(diào)用延遲高3. 每次請(qǐng)求都重新初始化組件1. 觀察ChromaDB查詢耗時(shí)2. 檢查OpenAI API響應(yīng)時(shí)間3. 檢查代碼結(jié)構(gòu)1. 確保ChromaDB使用HNSW等索引2. 考慮緩存頻繁查詢的記憶或使用更快的LLM3. 將向量存儲(chǔ)、LLM客戶端等設(shè)為全局單例7. 最佳實(shí)踐與工程建議將共享記憶系統(tǒng)投入實(shí)際生產(chǎn)環(huán)境需要考慮遠(yuǎn)比Demo更多的問題。以下是一些關(guān)鍵建議1. 記憶的粒度與分類不要存儲(chǔ)所有對(duì)話這會(huì)導(dǎo)致記憶庫迅速膨脹檢索效率和質(zhì)量下降。必須像MemoryManager那樣設(shè)計(jì)嚴(yán)格的過濾和摘要邏輯。設(shè)計(jì)更精細(xì)的記憶類型除了示例中的幾種還可以考慮PROCEDURE工作流程、RELATIONSHIP實(shí)體關(guān)系、INSIGHT分析洞察等。不同類型的記憶可以采用不同的存儲(chǔ)和檢索策略。實(shí)施記憶摘要對(duì)于長(zhǎng)對(duì)話存儲(chǔ)前先用LLM生成一個(gè)簡(jiǎn)潔的摘要而不是原始文本。這能極大節(jié)省存儲(chǔ)空間并提升檢索質(zhì)量。2. 檢索策略的優(yōu)化混合檢索不要只依賴向量相似度。結(jié)合關(guān)鍵詞過濾如時(shí)間范圍、用戶ID、記憶類型進(jìn)行混合檢索結(jié)果更精準(zhǔn)。重排序初步檢索出Top-K個(gè)結(jié)果后可以用一個(gè)更小的、更快的“重排序模型”對(duì)它們進(jìn)行精排將最相關(guān)的放在前面。記憶衰減與更新為記憶設(shè)計(jì)“新鮮度”或“訪問頻率”權(quán)重。長(zhǎng)期未被訪問或已被證偽的記憶應(yīng)被降權(quán)或歸檔而非刪除。3. 系統(tǒng)架構(gòu)與擴(kuò)展服務(wù)化將記憶存儲(chǔ)、檢索、管理模塊拆分為獨(dú)立的微服務(wù)如Memory Service通過REST或gRPC對(duì)外提供API。這樣前端Slack Bot、Web UI可以輕量化。多租戶與隔離為不同的團(tuán)隊(duì)、項(xiàng)目或用戶設(shè)計(jì)嚴(yán)格的數(shù)據(jù)隔離。在元數(shù)據(jù)中增加tenant_id、project_id字段并在檢索時(shí)強(qiáng)制過濾。監(jiān)控與可觀測(cè)性記錄關(guān)鍵指標(biāo)記憶存儲(chǔ)量、檢索耗時(shí)、檢索命中率、LLM調(diào)用延遲、用戶反饋如“這條記憶有用嗎”按鈕。這有助于持續(xù)優(yōu)化系統(tǒng)。4. 安全與隱私敏感信息過濾在記憶存儲(chǔ)前必須經(jīng)過一層敏感信息檢測(cè)和脫敏處理如自動(dòng)識(shí)別并遮蓋手機(jī)號(hào)、郵箱、密鑰。用戶控制權(quán)提供用戶界面讓用戶查看、編輯、刪除AI存儲(chǔ)的關(guān)于自己的記憶。這是合規(guī)如GDPR和建立信任的關(guān)鍵。訪問審計(jì)記錄誰哪個(gè)用戶或Agent在什么時(shí)候訪問了哪些記憶用于安全審計(jì)。5. 與現(xiàn)有工作流集成超越Slack將記憶系統(tǒng)與GitHub、Jira、Notion、Confluence等工具連接。例如當(dāng)AI在代碼評(píng)審中學(xué)習(xí)到一個(gè)團(tuán)隊(duì)的編碼規(guī)范時(shí)這條記憶應(yīng)該能被后續(xù)的代碼生成任務(wù)所用。主動(dòng)記憶觸發(fā)除了被動(dòng)響應(yīng)用戶消息系統(tǒng)可以設(shè)置定時(shí)任務(wù)主動(dòng)回顧和整理記憶或基于記憶向用戶推送提醒如“根據(jù)上周的討論您今天需要決定方案選型”。8. 總結(jié)與后續(xù)學(xué)習(xí)方向通過本文的拆解我們實(shí)現(xiàn)了一個(gè)具備“共享記憶”核心能力的AI助手原型。它不再是健忘的對(duì)話者而是能夠積累知識(shí)、并在需要時(shí)準(zhǔn)確回憶的協(xié)作者。這不僅僅是技術(shù)的疊加更是對(duì)AI應(yīng)用范式的重新思考從追求單次交互的驚艷轉(zhuǎn)向構(gòu)建長(zhǎng)期協(xié)作的信任和效率?；仡橪indy項(xiàng)目提出的愿景其核心價(jià)值在于將“記憶”這個(gè)抽象概念工程化為可落地、可擴(kuò)展的系統(tǒng)組件。我們通過向量數(shù)據(jù)庫、LLM智能路由和Slack集成驗(yàn)證了這一路徑的可行性。對(duì)于想要繼續(xù)深入的開發(fā)者下一步可以探索的方向深入研究開源項(xiàng)目關(guān)注類似my_ai_town、LangChain、LlamaIndex中關(guān)于“長(zhǎng)期記憶”、“代理記憶”的社區(qū)討論和實(shí)現(xiàn)方案汲取更成熟的設(shè)計(jì)模式。探索更復(fù)雜的記憶結(jié)構(gòu)嘗試用知識(shí)圖譜來存儲(chǔ)記憶處理實(shí)體、關(guān)系、事件等更結(jié)構(gòu)化的信息實(shí)現(xiàn)邏輯推理而不僅僅是語義相似。實(shí)現(xiàn)多Agent記憶共享構(gòu)建一個(gè)記憶中心讓多個(gè)具有不同技能的AI Agent如一個(gè)負(fù)責(zé)數(shù)據(jù)分析一個(gè)負(fù)責(zé)編寫文檔可以共同讀寫實(shí)現(xiàn)真正的協(xié)同工作。關(guān)注模型本身的進(jìn)步OpenAI的“記憶”功能、Claude的“項(xiàng)目”功能都表明大模型廠商正在原生層面解決此問題。理解其API和設(shè)計(jì)哲學(xué)與外部記憶系統(tǒng)結(jié)合可能是更優(yōu)解。構(gòu)建一個(gè)實(shí)用的共享記憶系統(tǒng)挑戰(zhàn)不在于單點(diǎn)技術(shù)而在于對(duì)業(yè)務(wù)場(chǎng)景的深度理解、對(duì)記憶價(jià)值的精準(zhǔn)判斷以及在性能、成本、隱私之間的復(fù)雜權(quán)衡。希望本文提供的思路和代碼能成為你探索這個(gè)迷人領(lǐng)域的起點(diǎn)。建議收藏本文在動(dòng)手實(shí)踐時(shí)對(duì)照“常見問題”和“最佳實(shí)踐”部分相信能幫你避開不少彎路。