實戰(zhàn)：向量與關系數(shù)據(jù)庫融合架構)
1. 項目概述當“小龍蝦”擁有了“記憶”最近在折騰本地AI智能體OpenClaw圈內(nèi)戲稱“小龍蝦”這個名字出現(xiàn)的頻率越來越高。它本質(zhì)上是一個開源的AI智能體框架你可以把它理解為一個“大腦”負責調(diào)度和協(xié)調(diào)各種AI能力去完成復雜的任務。但玩過一陣子后我發(fā)現(xiàn)一個普遍痛點這“大腦”記性不太好。每次對話都像是初次見面上下文一長就斷片更別提讓它記住我的個人偏好、項目背景這些長期信息了。這嚴重限制了它的實用性尤其是在處理需要持續(xù)跟蹤狀態(tài)的任務時比如自動化客服、項目管理或者個人知識庫助手。于是我開始尋找解決方案目標很明確給OpenClaw這個聰明的“大腦”裝上一個可靠的“記事本”。這就是“Active Memory”主動記憶概念的由來。它不是一個簡單的聊天記錄存儲器而是一個能夠被智能體主動查詢、更新、關聯(lián)的結構化記憶系統(tǒng)。簡單說就是讓OpenClaw學會“做筆記”和“翻筆記”。這個融合項目就是要把OpenClaw的智能決策能力與一個持久化、可操作的內(nèi)存系統(tǒng)結合起來。我最終選擇了一個基于向量數(shù)據(jù)庫比如ChromaDB或Qdrant和關系型數(shù)據(jù)庫如SQLite的混合架構來實現(xiàn)Active Memory。向量庫負責語義搜索快速找到相關記憶關系庫則存儲記憶的元數(shù)據(jù)、時間戳和關聯(lián)關系。下面我就把這次從設計思路到踩坑填坑的完整實戰(zhàn)過程拆解出來。2. 核心架構設計與技術選型2.1 為什么是“混合記憶”架構一開始我考慮過幾種簡單的方案。比如只用向量數(shù)據(jù)庫把所有對話都存成向量。但問題很快暴露查找是快了但我想按時間篩選“昨天我們討論的那個需求”或者按類型查找“所有關于‘部署’的筆記”向量檢索就顯得力不從心。反之如果只用關系數(shù)據(jù)庫雖然能方便地按字段查詢但做“幫我找找和‘自動化流程’相關的所有內(nèi)容”這種模糊語義搜索效率又很低。所以混合架構成了必然選擇。它的核心思想是“分工協(xié)作”關系型數(shù)據(jù)庫如SQLite充當“記憶的目錄和索引”。它存儲每條記憶的唯一ID、創(chuàng)建時間、記憶類型是“用戶偏好”、“項目上下文”還是“會話歷史”、關鍵標簽、以及關聯(lián)的實體如項目名、聯(lián)系人。它的優(yōu)勢是結構化查詢非?？烨揖珳省Ｏ蛄繑?shù)據(jù)庫如ChromaDB充當“記憶的內(nèi)容搜索引擎”。它存儲記憶文本內(nèi)容經(jīng)過Embedding模型轉換后的向量。當OpenClaw需要回憶時可以將當前問題的語義轉換成向量然后在向量空間里快速找到最相似的幾條記憶內(nèi)容。兩者通過一個共同的“記憶ID”進行關聯(lián)。當智能體需要回憶時可以先通過關系數(shù)據(jù)庫的元數(shù)據(jù)做初步篩選比如限定時間、類型再用篩選出的記憶ID去向量數(shù)據(jù)庫做精密的語義相似度匹配最終返回最相關的幾條記憶。2.2 OpenClaw與記憶系統(tǒng)的交互流程明確了架構接下來要設計OpenClaw如何與這個記憶系統(tǒng)“對話”。我設計了一個名為MemoryManager的核心模塊作為兩者之間的“經(jīng)紀人”。整個交互流程是這樣的記憶寫入Remember當OpenClaw在處理任務過程中產(chǎn)生了值得記錄的信息例如用戶說“我更喜歡用Markdown格式回復”MemoryManager會將其封裝成一個記憶對象。這個對象包含原始文本、自動提取的關鍵標簽、記憶類型、時間戳等。然后它同時向關系數(shù)據(jù)庫插入一條元數(shù)據(jù)記錄并向向量數(shù)據(jù)庫插入這條文本的向量化表示。這是一個原子操作必須確保兩者都成功否則回滾。記憶讀取Recall當OpenClaw需要背景信息時例如用戶問“我之前說的格式偏好是什么”它會向MemoryManager發(fā)起一個查詢請求。MemoryManager首先解析查詢意圖如果查詢條件明確如“類型用戶偏好”則先查詢關系數(shù)據(jù)庫獲取候選記憶ID列表。然后將原始查詢語句向量化在向量數(shù)據(jù)庫中針對這些候選ID或全部記憶進行相似度搜索返回得分最高的前N條記憶。記憶更新與清理記憶不是只增不減的。我設計了兩種策略。一是基于時間的衰減很久未觸發(fā)的記憶會被標記為“不活躍”。二是基于重要性的評估在寫入時可以由智能體或規(guī)則賦予一個初始重要性權重每次被成功召回并助力任務完成該權重增加反之如果記憶內(nèi)容被用戶糾正則權重降低。權重過低或過時的記憶會被歸檔或清理。注意這里的一個關鍵設計點是“記憶的粒度”。不要把一整段對話都存成一條記憶。更好的做法是按“信息點”進行拆分。比如一段關于項目需求的討論可以拆分為“項目目標實現(xiàn)X”、“技術棧Python, FastAPI”、“截止日期下周五”等多個獨立的記憶單元。這樣在召回時更精準也便于管理。3. 實戰(zhàn)部署搭建OpenClaw與Active Memory環(huán)境3.1 基礎環(huán)境與OpenClaw部署我的實驗環(huán)境是一臺Ubuntu 22.04的云服務器當然在Mac或Windows的Docker環(huán)境下流程也類似。首先解決OpenClaw的部署。目前最穩(wěn)定、隔離性最好的方式就是Docker。OpenClaw社區(qū)提供了官方鏡像但為了靈活性我更喜歡使用docker-compose來編排。# docker-compose.yml version: 3.8 services: openclaw: image: openclaw/openclaw:latest container_name: openclaw restart: unless-stopped ports: - 3000:3000 # Web UI端口 environment: - OLLAMA_BASE_URLhttp://host.docker.internal:11434 # 假設Ollama在宿主機 - DEFAULT_MODELllama3.2:latest # 默認使用的模型 - LOG_LEVELINFO volumes: - ./openclaw_data:/app/data # 掛載配置和數(shù)據(jù)卷 networks: - ai-net # 我們稍后會添加記憶相關的服務這里有幾個關鍵點OLLAMA_BASE_URL: OpenClaw本身不包含大模型它需要連接一個模型服務。我本地用Ollama運行了Llama 3.2模型所以這里配置為宿主機的Ollama服務。如果你把Ollama也放在Docker里需要改為服務名如http://ollama:11434。volumes: 一定要掛載數(shù)據(jù)卷否則容器重啟后所有配置和會話記錄都會丟失。先不急著運行等我們把記憶系統(tǒng)的組件也編排進來。運行docker-compose up -d訪問http://你的服務器IP:3000就能看到OpenClaw的Web界面了。首次使用需要在設置里配置好模型端點。3.2 Active Memory核心組件部署記憶系統(tǒng)需要兩個數(shù)據(jù)庫。為了簡化我都用Docker來部署。1. 向量數(shù)據(jù)庫ChromaDBChromaDB輕量且易于集成是快速原型的最佳選擇。我們在docker-compose.yml中新增一個服務。chromadb: image: chromadb/chroma:latest container_name: chromadb restart: unless-stopped ports: - 8000:8000 command: uvicorn chromadb.app:app --reload --workers 1 --host 0.0.0.0 --port 8000 environment: - IS_PERSISTENTTRUE - PERSIST_DIRECTORY/chroma/chroma_data volumes: - ./chroma_data:/chroma/chroma_data networks: - ai-net2. 關系數(shù)據(jù)庫PostgreSQL雖然SQLite更輕量但考慮到未來可能的多節(jié)點部署和更復雜的查詢我選擇了PostgreSQL。同樣在docker-compose.yml中添加。postgres: image: postgres:15-alpine container_name: postgres-memory restart: unless-stopped environment: POSTGRES_USER: openclaw POSTGRES_PASSWORD: your_secure_password_here # 務必修改 POSTGRES_DB: activememory ports: - 5432:5432 volumes: - ./postgres_data:/var/lib/postgresql/data networks: - ai-net現(xiàn)在更新后的docker-compose.yml包含了三個服務。運行docker-compose up -d一次性啟動所有服務。實操心得在生產(chǎn)環(huán)境中務必為PostgreSQL設置強密碼并將端口映射5432:5432考慮在內(nèi)網(wǎng)訪問或者通過Docker網(wǎng)絡內(nèi)部通信不要直接暴露在公網(wǎng)。我的做法是只將OpenClaw的3000端口通過Nginx反向代理并配置SSL暴露出去ChromaDB和PostgreSQL僅通過Docker內(nèi)部網(wǎng)絡 (ai-net) 供OpenClaw容器訪問。這樣更安全。3.3 開發(fā)MemoryManager橋梁模塊OpenClaw本身沒有內(nèi)置記憶系統(tǒng)我們需要開發(fā)一個插件或中間件。我選擇用Python編寫一個獨立的MemoryManager服務并通過OpenClaw的Skill技能機制或Webhook與之集成。首先創(chuàng)建memory_manager目錄結構如下memory_manager/ ├── app.py # FastAPI主應用提供記憶的CRUD接口 ├── memory_core.py # 記憶的核心邏輯寫入、查詢、向量化 ├── database.py # 數(shù)據(jù)庫連接與操作PostgreSQL, Chroma ├── requirements.txt └── Dockerfile1. 依賴文件 (requirements.txt)fastapi0.104.1 uvicorn[standard]0.24.0 sqlalchemy2.0.23 psycopg2-binary2.9.9 chromadb0.4.22 sentence-transformers2.2.2 # 用于本地Embedding可選 openai1.3.0 # 如果使用OpenAI的Embedding API pydantic2.5.02. 數(shù)據(jù)庫模型與連接 (database.py)這里定義記憶的元數(shù)據(jù)表結構。from sqlalchemy import create_engine, Column, String, DateTime, Text, Float, JSON from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime import os DATABASE_URL os.getenv(DATABASE_URL, postgresql://openclaw:your_passwordpostgres/activememory) engine create_engine(DATABASE_URL) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base declarative_base() class MemoryMetadata(Base): __tablename__ memory_metadata id Column(String, primary_keyTrue) # 與ChromaDB中的ID對應 content Column(Text, nullableFalse) # 原始文本內(nèi)容 memory_type Column(String, indexTrue) # 如user_preference, project_context, conversation tags Column(JSON) # 標簽列表如 [format, preference] source Column(String) # 來源如 openclaw_session_001 importance Column(Float, default1.0) # 重要性權重 last_accessed Column(DateTime, defaultdatetime.utcnow) created_at Column(DateTime, defaultdatetime.utcnow) # 創(chuàng)建表 Base.metadata.create_all(bindengine)3. 記憶核心邏輯 (memory_core.py)這是最核心的部分負責協(xié)調(diào)兩個數(shù)據(jù)庫。import uuid from datetime import datetime import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer # 示例用本地模型 # 或 from openai import OpenAI from database import SessionLocal, MemoryMetadata import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class MemoryCore: def __init__(self, embedding_model_nameall-MiniLM-L6-v2): # 初始化ChromaDB客戶端連接Docker中的服務 self.chroma_client chromadb.HttpClient( hostchromadb, # Docker服務名 port8000, settingsSettings(allow_resetTrue) ) # 獲取或創(chuàng)建集合類似于表 self.collection self.chroma_client.get_or_create_collection(nameactive_memory) # 初始化Embedding模型本地 # 注意本地模型首次加載慢但無需網(wǎng)絡。也可用OpenAI API。 self.embedder SentenceTransformer(embedding_model_name) logger.info(MemoryCore initialized.) def _generate_embedding(self, text: str): 生成文本的向量表示 # 使用本地模型 embedding self.embedder.encode(text).tolist() # 如果使用OpenAI API: # client OpenAI(api_keyyour_key) # response client.embeddings.create(modeltext-embedding-3-small, inputtext) # embedding response.data[0].embedding return embedding def remember(self, content: str, memory_type: str general, tags: list None, source: str None): 保存一條記憶 memory_id str(uuid.uuid4()) embedding self._generate_embedding(content) # 1. 存入向量數(shù)據(jù)庫 (ChromaDB) self.collection.add( documents[content], embeddings[embedding], ids[memory_id] ) # 2. 存入關系數(shù)據(jù)庫 (PostgreSQL) db SessionLocal() try: db_memory MemoryMetadata( idmemory_id, contentcontent, memory_typememory_type, tagstags or [], sourcesource, created_atdatetime.utcnow(), last_accesseddatetime.utcnow() ) db.add(db_memory) db.commit() logger.info(fMemory saved. ID: {memory_id}, Type: {memory_type}) except Exception as e: logger.error(fFailed to save metadata for {memory_id}: {e}) # 理想情況下這里應有事務回滾也需刪除剛存入Chroma的數(shù)據(jù) # 簡化處理記錄錯誤 db.rollback() finally: db.close() return memory_id def recall(self, query: str, memory_type: str None, limit: int 5): 回憶根據(jù)查詢語句和可選類型查找相關記憶 query_embedding self._generate_embedding(query) # 第一步如果指定了類型先從PostgreSQL獲取該類型的所有記憶ID memory_ids_filter None if memory_type: db SessionLocal() try: results db.query(MemoryMetadata.id).filter(MemoryMetadata.memory_type memory_type).all() memory_ids_filter [r[0] for r in results] logger.debug(fFiltering by type {memory_type}, found {len(memory_ids_filter)} IDs.) finally: db.close() # 第二步在ChromaDB中進行向量相似度查詢 # where_document 可以用于ChromaDB自身的元數(shù)據(jù)過濾但我們用PostgreSQL做了這里用ids過濾 results self.collection.query( query_embeddings[query_embedding], n_resultslimit, where{memory_type: memory_type} if memory_type else None, # Chroma的元數(shù)據(jù)過濾需在add時傳入 # 或者使用從PostgreSQL獲取的ID列表進行過濾如果集合很大先過濾更高效 # 這里演示使用Chroma的where條件前提是存入時傳入了memory_type ) # 第三步根據(jù)返回的ID從PostgreSQL獲取完整的元數(shù)據(jù)信息 recalled_memories [] if results and results[ids][0]: db SessionLocal() try: for mem_id in results[ids][0]: db_memory db.query(MemoryMetadata).filter(MemoryMetadata.id mem_id).first() if db_memory: # 更新最后訪問時間 db_memory.last_accessed datetime.utcnow() recalled_memories.append({ id: db_memory.id, content: db_memory.content, type: db_memory.memory_type, tags: db_memory.tags, source: db_memory.source, importance: db_memory.importance, similarity_score: results[distances][0][results[ids][0].index(mem_id)] if results.get(distances) else None }) db.commit() finally: db.close() logger.info(fRecalled {len(recalled_memories)} memories for query: {query}) return recalled_memories4. 構建API接口 (app.py)用FastAPI包裝核心功能供OpenClaw調(diào)用。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional, List from memory_core import MemoryCore import logging app FastAPI(titleActive Memory Service) memory_core MemoryCore() class MemoryCreate(BaseModel): content: str memory_type: Optional[str] general tags: Optional[List[str]] [] source: Optional[str] None class MemoryQuery(BaseModel): query: str memory_type: Optional[str] None limit: Optional[int] 5 app.post(/remember) async def remember(memory: MemoryCreate): 存儲一條新記憶 try: memory_id memory_core.remember( contentmemory.content, memory_typememory.memory_type, tagsmemory.tags, sourcememory.source ) return {message: Memory saved successfully, memory_id: memory_id} except Exception as e: logging.error(fError in /remember: {e}) raise HTTPException(status_code500, detailstr(e)) app.post(/recall) async def recall(query: MemoryQuery): 根據(jù)查詢回憶相關記憶 try: memories memory_core.recall( queryquery.query, memory_typequery.memory_type, limitquery.limit ) return {query: query.query, memories: memories} except Exception as e: logging.error(fError in /recall: {e}) raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health(): return {status: healthy}5. 編寫Dockerfile并加入編排為這個記憶服務也創(chuàng)建一個Docker鏡像。# memory_manager/Dockerfile FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, app:app, --host, 0.0.0.0, --port, 8001]最后更新總的docker-compose.yml加入我們的記憶服務。memory-service: build: ./memory_manager # 指向MemoryManager目錄 container_name: memory-service restart: unless-stopped ports: - 8001:8001 environment: - DATABASE_URLpostgresql://openclaw:your_passwordpostgres/activememory # 可以在這里設置Embedding模型類型或API密鑰 depends_on: - chromadb - postgres networks: - ai-net現(xiàn)在運行docker-compose up -d --build重新構建并啟動所有服務。訪問http://localhost:8001/docs可以看到自動生成的API文檔可以測試/remember和/recall接口。4. 集成與調(diào)試讓OpenClaw“學會”記憶4.1 通過Skill技能機制集成OpenClaw的強大之處在于其Skill系統(tǒng)。我們可以編寫一個自定義Skill讓OpenClaw在對話中自動調(diào)用記憶服務。在OpenClaw的配置目錄我們之前掛載的./openclaw_data下通常會有skills文件夾。我們創(chuàng)建一個新的Skill文件例如active_memory_skill.py。# ./openclaw_data/skills/active_memory_skill.py import requests import json from typing import Dict, Any MEMORY_SERVICE_URL http://memory-service:8001 # Docker內(nèi)部網(wǎng)絡通信 class ActiveMemorySkill: 一個讓OpenClaw具備主動記憶能力的技能 def __init__(self): self.name active_memory self.description 保存或回憶對話中的關鍵信息到長期記憶庫。 self.triggers [ 記住.*, # 當用戶說“記住我喜歡用藍色主題” 我之前說過.*, # 當用戶問“我之前說過我的偏好是什么” 回憶一下.*, 關于.*(你還記得嗎|你知道多少), ] def execute(self, context: Dict[str, Any]) - str: Skill執(zhí)行入口 user_input context.get(user_input, ).lower() session_id context.get(session_id, default) # 1. 判斷意圖是保存記憶還是回憶 if user_input.startswith(記住): # 提取要記憶的內(nèi)容例如“記住我喜歡用藍色主題” - “我喜歡用藍色主題” content_to_remember user_input[2:].strip() # 簡單處理 if content_to_remember: return self._save_memory(content_to_remember, user_preference, session_id) else: return 請告訴我需要記住什么內(nèi)容。 elif any(trigger in user_input for trigger in [我之前說過, 回憶一下, 還記得嗎, 你知道多少]): # 提取查詢關鍵詞這里做簡單提取實際可用更復雜的NLP # 例如“我之前說過我的偏好是什么” - “偏好” query_key user_input # 更佳實踐調(diào)用一個意圖識別函數(shù)來提取核心查詢詞 # 此處簡化直接用整個句子后半部分或關鍵詞 return self._recall_memory(query_key, session_id) # 2. 被動記憶在每次對話輪次結束后由OpenClaw框架自動調(diào)用保存關鍵信息 # 這通常需要在OpenClaw的鉤子函數(shù)中配置此處不展開。 return def _save_memory(self, content: str, mem_type: str, source: str) - str: 調(diào)用記憶服務保存記憶 try: payload { content: content, memory_type: mem_type, tags: self._extract_tags(content), # 可實現(xiàn)的標簽提取函數(shù) source: source } response requests.post(f{MEMORY_SERVICE_URL}/remember, jsonpayload, timeout5) if response.status_code 200: return f好的我已經(jīng)將‘{content[:30]}...’記到我的備忘錄里了。 else: return f記憶保存失敗{response.text} except requests.exceptions.RequestException as e: return f無法連接到記憶服務{(diào)e} def _recall_memory(self, query: str, source: str) - str: 調(diào)用記憶服務回憶 try: payload {query: query, limit: 3} response requests.post(f{MEMORY_SERVICE_URL}/recall, jsonpayload, timeout5) if response.status_code 200: data response.json() memories data.get(memories, []) if memories: # 格式化回憶結果 memory_texts [f- {mem[content]} (相關度: {mem.get(similarity_score, N/A):.2f}) for mem in memories[:2]] reply f根據(jù)我的記錄相關的內(nèi)容有\(zhòng)n \n.join(memory_texts) if len(memories) 2: reply f\n還有{len(memories)-2}條相關記錄 return reply else: return 我的記憶里暫時沒有找到相關信息。 else: return f記憶回憶失敗{response.text} except requests.exceptions.RequestException as e: return f無法連接到記憶服務{(diào)e} def _extract_tags(self, text: str) - list: 簡單的關鍵詞/標簽提取示例實際可用TF-IDF或小模型 # 這里只是一個示例實際應用應使用更成熟的方法 predefined_tags [偏好, 配置, 項目, 需求, 日期, 聯(lián)系人] found_tags [tag for tag in predefined_tags if tag in text] return found_tags if found_tags else [general] # OpenClaw Skill標準導出 def get_skill(): return ActiveMemorySkill()將這個文件放到OpenClaw的技能目錄后需要在OpenClaw的配置中啟用它。具體配置方式因OpenClaw版本而異通常是在Web UI的技能管理頁面添加或修改配置文件config.yaml添加技能路徑和初始化參數(shù)。4.2 配置OpenClaw調(diào)用記憶服務除了主動技能我們更希望OpenClaw能“潛移默化”地使用記憶。這需要在OpenClaw處理對話的流程中插入鉤子Hooks。OpenClaw的架構通常支持“前置處理器”和“后置處理器”。我們可以創(chuàng)建一個后置處理器在OpenClaw生成回復后自動分析本輪對話如果包含值得長期記憶的信息例如用戶明確了某個設置、陳述了一個事實就自動調(diào)用memory-service的/remember接口保存。同樣在OpenClaw生成回復前前置處理器可以自動根據(jù)當前對話的上下文調(diào)用/recall接口獲取相關記憶并將這些記憶作為附加上下文注入給大模型從而讓模型在“知情”的情況下進行回復。這部分集成深度依賴于OpenClaw的具體版本和擴展機制可能需要修改其核心代碼或利用其插件系統(tǒng)。一個常見的模式是在向大模型發(fā)送的Prompt模板中加入一個“相關記憶”的占位符由前置處理器負責填充。例如修改后的Prompt可能看起來像這樣你是一個有幫助的AI助手。以下是一些可能相關的歷史記錄你的記憶 {formatted_memories} 當前對話 用戶{user_input} 助手這樣大模型在生成回復時就能自然地引用記憶中的信息。5. 效果驗證與性能調(diào)優(yōu)5.1 功能測試與效果評估部署并集成完成后需要進行系統(tǒng)測試。1. 基礎CRUD測試通過記憶服務的API文檔頁面直接測試/remember和/recall確保接口工作正常。插入幾條測試記憶如{content: 用戶喜歡在晚上接收每日報告, memory_type: user_preference, tags: [report, schedule]}。用相關查詢?nèi)纭皥蟾鏁r間”進行回憶看是否能正確返回。2. OpenClaw技能測試在OpenClaw的Web界面中直接對AI說“記住我的項目‘AI助手’的API密鑰是sk-abc123請保密?！庇^察回復確認技能被觸發(fā)并返回成功信息。然后問“我之前告訴過你API密鑰嗎”或“關于AI助手項目你還記得什么”檢查OpenClaw的回復是否包含了之前存儲的記憶內(nèi)容。3. 自動化記憶測試進行一段多輪對話討論一個具體問題比如配置郵箱。在對話中故意說出一些關鍵信息如“我的郵箱服務器是smtp.example.com端口是587”。在后續(xù)對話中詢問“郵箱端口是多少”看OpenClaw是否能憑借記憶正確回答而無需你重新告知。5.2 性能瓶頸分析與優(yōu)化在實際使用中可能會遇到一些性能問題。1. 向量檢索速度慢問題當記憶條數(shù)超過數(shù)萬時ChromaDB的暴力相似度搜索可能會變慢。優(yōu)化索引確保ChromaDB使用了合適的索引如HNSW。在創(chuàng)建集合時可以通過參數(shù)配置。預過濾充分利用memory_type和tags在關系數(shù)據(jù)庫中進行預過濾 drastically減少需要做向量相似度計算的候選集大小。這正是我們混合架構的優(yōu)勢。分頁回憶時不要一次性取太多條limit參數(shù)合理設置如5-10條。2. Embedding生成成為瓶頸問題使用本地Sentence Transformer模型如all-MiniLM-L6-v2雖然免費但CPU推理在寫入大量記憶時可能較慢。優(yōu)化批處理將多個記憶內(nèi)容批量生成Embedding減少模型加載和調(diào)用的開銷。使用GPU如果服務器有GPU確保PyTorch和Transformer庫利用了CUDA。換用API服務對于高并發(fā)生產(chǎn)環(huán)境可以考慮使用OpenAI、Cohere或?qū)ｉT的高性能Embedding API服務它們通常速度更快且有速率限制管理。但會引入網(wǎng)絡延遲和成本。3. 記憶冗余與沖突問題用戶可能多次表達相同或矛盾的信息如“我喜歡藍色”和“主題改成黑色吧”。優(yōu)化去重在remember前可以先進行一次recall檢查是否有高度相似相似度超過0.95的現(xiàn)有記憶。如果有可以選擇更新原有記憶例如合并內(nèi)容、更新時間戳、增加權重而不是新增一條。沖突解決當檢測到新舊記憶矛盾時可以設計規(guī)則。例如默認以最新的信息為準但降低舊記憶的權重而非直接刪除或者在回憶時同時返回新舊記憶并在提示詞中告訴大模型“這里有兩條矛盾的信息請根據(jù)上下文判斷”。5.3 高級功能展望一個基礎的Active Memory系統(tǒng)已經(jīng)能極大提升體驗。在此基礎上還可以考慮更多增強功能記憶關聯(lián)圖不僅存儲孤立的記憶點還存儲記憶之間的關系。例如“項目A”使用了“技術B” “技術B”的專家是“聯(lián)系人C”。這可以通過在關系數(shù)據(jù)庫中增加一個related_memory_ids字段存儲關聯(lián)記憶ID列表來實現(xiàn)讓回憶時能進行“聯(lián)想”。記憶摘要對于長時間的對話或文檔可以定期或當記憶數(shù)量過多時調(diào)用大模型生成一個摘要作為一條新的、更高級別的“概要記憶”存儲起來從而壓縮信息提高長期記憶的效率。記憶失效與歸檔策略實現(xiàn)更復雜的記憶生命周期管理。例如設定不同記憶類型的TTL生存時間將長時間未訪問且重要性低的記憶移動到廉價的冷存儲如從ChromaDB/PostgreSQL轉移到文件定期清理“垃圾記憶”。6. 常見問題與故障排查實錄在部署和調(diào)試過程中我遇到了不少坑這里把典型問題和解決方法記錄下來。6.1 部署連接問題問題1OpenClaw容器內(nèi)無法連接到memory-service:8001?，F(xiàn)象Skill執(zhí)行時報錯“無法連接到記憶服務”Connection refused。排查進入OpenClaw容器docker exec -it openclaw bash。嘗試pingmemory-serviceping memory-service。如果不通說明Docker網(wǎng)絡有問題。檢查docker-compose.yml確保所有服務在同一個自定義網(wǎng)絡下如ai-net并且OpenClaw服務定義了depends_on: - memory-service這主要控制啟動順序不保證網(wǎng)絡可達但通常一起定義。解決確認網(wǎng)絡配置正確。最穩(wěn)妥的方式是在OpenClaw容器內(nèi)使用curl http://memory-service:8001/health測試連通性。如果不通檢查Docker網(wǎng)絡docker network ls和docker network inspect ai-net確保所有容器都連接到了該網(wǎng)絡。問題2ChromaDB連接失敗報錯Failed to connect?，F(xiàn)象MemoryCore初始化時連接ChromaDB超時或失敗。排查首先在宿主機上curl http://localhost:8000/api/v1/heartbeat檢查ChromaDB服務本身是否健康。如果宿主機通但memory-service容器內(nèi)不通檢查memory-service的Docker Compose配置中ChromaDB的host名是否正確。在Docker Compose中服務名chromadb就是主機名。檢查ChromaDB容器的日志docker logs chromadb看是否有啟動錯誤。解決確保ChromaDB的command中指定了--host 0.0.0.0以允許所有網(wǎng)絡接口連接。防火墻或安全組規(guī)則確保8000端口在容器間可訪問。6.2 技能與集成問題問題3OpenClaw不觸發(fā)自定義Skill?，F(xiàn)象在Web界面說話技能毫無反應。排查檢查Skill文件是否放在了正確的目錄通常是openclaw_data/skills/并且OpenClaw的配置指向了這個目錄。查看OpenClaw的日志docker logs openclaw尋找加載技能時的錯誤信息。檢查Skill類中的triggers列表。OpenClaw的觸發(fā)機制可能是正則表達式匹配或關鍵字匹配。確保你的用戶輸入能匹配上。例如“記住.*”是一個正則需要用戶輸入以“記住”開頭。可以先用簡單的[test]作為trigger來測試。確認Skill類被正確導出有get_skill()函數(shù)。解決仔細閱讀OpenClaw官方關于Skill開發(fā)的文檔確認其加載機制和觸發(fā)規(guī)則。一個有效的調(diào)試方法是在Skill的execute方法開頭加入日志打印確認方法是否被調(diào)用。問題4記憶回憶的結果不相關?，F(xiàn)象用戶問“郵箱設置”返回的卻是關于“晚餐吃什么”的記憶。排查Embedding模型問題使用的Embedding模型是否適合中文all-MiniLM-L6-v2對英文優(yōu)化更好?？梢試L試換用多語言模型如paraphrase-multilingual-MiniLM-L12-v2。查詢詞過于寬泛“郵箱設置”可能被Embedding成一個比較泛的向量。嘗試在回憶前對用戶查詢進行輕微的改寫或擴展例如結合對話上下文將查詢擴展為“用戶詢問郵箱服務器和端口的設置信息”。記憶粒度問題存入的記憶文本是否太冗長或包含無關信息確保存入的是干凈、核心的信息點。解決更換或微調(diào)Embedding模型優(yōu)化記憶的寫入內(nèi)容使其更聚焦在回憶時嘗試將當前對話的最近幾條消息一起作為查詢上下文提升相關性。6.3 數(shù)據(jù)庫與性能問題問題5PostgreSQL連接數(shù)過多?，F(xiàn)象運行一段時間后memory-service出現(xiàn)too many connections錯誤。原因SQLAlchemy的Session沒有正確關閉。在memory_core.py的recall和remember方法中雖然用了try...finally來關閉session但在異常處理分支中可能仍有遺漏。解決使用上下文管理器確保Session總是被關閉?；蛘邽镕astAPI應用配置SQLAlchemy的scoped_session并確保在每個請求結束后移除session。更簡單的方法是在database.py中創(chuàng)建一個依賴項。# 在database.py中 def get_db(): db SessionLocal() try: yield db finally: db.close() # 在FastAPI路由中 from fastapi import Depends from sqlalchemy.orm import Session app.post(/remember) async def remember(memory: MemoryCreate, db: Session Depends(get_db)): # ... 使用db session # 無需手動關閉依賴項會自動處理問題6ChromaDB數(shù)據(jù)持久化失敗?，F(xiàn)象重啟Docker Compose后之前存儲的記憶全部消失。排查檢查ChromaDB的容器是否配置了持久化卷并且PERSIST_DIRECTORY環(huán)境變量指向了卷內(nèi)路徑。檢查docker-compose.yml中chromadb服務的volumes映射和environment設置。解決確保volumes: - ./chroma_data:/chroma/chroma_data存在并且目錄./chroma_data在宿主機上有寫入權限。同時ChromaDB容器的command中不能有--reload參數(shù)用于開發(fā)在生產(chǎn)中可能引發(fā)問題可以去掉。這個融合項目從構想到實現(xiàn)花費了不少精力但結果是值得的?？粗鳲penClaw從“金魚腦”變成一個有“長期記憶”的靠譜助手能記住項目細節(jié)、用戶偏好并在后續(xù)對話中自然引用那種體驗的提升是質(zhì)的飛躍。最關鍵的是整個架構基于開源組件搭建完全可控可以根據(jù)自己的需求靈活調(diào)整記憶的邏輯和存儲策略。如果你也在探索AI智能體的長期記憶問題希望這份詳細的實戰(zhàn)記錄能幫你少走些彎路。 OpenClaw智能體混合記憶系統(tǒng)實戰(zhàn)：向量與關系數(shù)據(jù)庫融合架構
1. 項目概述當“小龍蝦”擁有了“記憶”最近在折騰本地AI智能體OpenClaw圈內(nèi)戲稱“小龍蝦”這個名字出現(xiàn)的頻率越來越高。它本質(zhì)上是一個開源的AI智能體框架你可以把它理解為一個“大腦”負責調(diào)度和協(xié)調(diào)各種AI能力去完成復雜的任務。但玩過一陣子后我發(fā)現(xiàn)一個普遍痛點這“大腦”記性不太好。每次對話都像是初次見面上下文一長就斷片更別提讓它記住我的個人偏好、項目背景這些長期信息了。這嚴重限制了它的實用性尤其是在處理需要持續(xù)跟蹤狀態(tài)的任務時比如自動化客服、項目管理或者個人知識庫助手。于是我開始尋找解決方案目標很明確給OpenClaw這個聰明的“大腦”裝上一個可靠的“記事本”。這就是“Active Memory”主動記憶概念的由來。它不是一個簡單的聊天記錄存儲器而是一個能夠被智能體主動查詢、更新、關聯(lián)的結構化記憶系統(tǒng)。簡單說就是讓OpenClaw學會“做筆記”和“翻筆記”。這個融合項目就是要把OpenClaw的智能決策能力與一個持久化、可操作的內(nèi)存系統(tǒng)結合起來。我最終選擇了一個基于向量數(shù)據(jù)庫比如ChromaDB或Qdrant和關系型數(shù)據(jù)庫如SQLite的混合架構來實現(xiàn)Active Memory。向量庫負責語義搜索快速找到相關記憶關系庫則存儲記憶的元數(shù)據(jù)、時間戳和關聯(lián)關系。下面我就把這次從設計思路到踩坑填坑的完整實戰(zhàn)過程拆解出來。2. 核心架構設計與技術選型2.1 為什么是“混合記憶”架構一開始我考慮過幾種簡單的方案。比如只用向量數(shù)據(jù)庫把所有對話都存成向量。但問題很快暴露查找是快了但我想按時間篩選“昨天我們討論的那個需求”或者按類型查找“所有關于‘部署’的筆記”向量檢索就顯得力不從心。反之如果只用關系數(shù)據(jù)庫雖然能方便地按字段查詢但做“幫我找找和‘自動化流程’相關的所有內(nèi)容”這種模糊語義搜索效率又很低。所以混合架構成了必然選擇。它的核心思想是“分工協(xié)作”關系型數(shù)據(jù)庫如SQLite充當“記憶的目錄和索引”。它存儲每條記憶的唯一ID、創(chuàng)建時間、記憶類型是“用戶偏好”、“項目上下文”還是“會話歷史”、關鍵標簽、以及關聯(lián)的實體如項目名、聯(lián)系人。它的優(yōu)勢是結構化查詢非?？烨揖珳省Ｏ蛄繑?shù)據(jù)庫如ChromaDB充當“記憶的內(nèi)容搜索引擎”。它存儲記憶文本內(nèi)容經(jīng)過Embedding模型轉換后的向量。當OpenClaw需要回憶時可以將當前問題的語義轉換成向量然后在向量空間里快速找到最相似的幾條記憶內(nèi)容。兩者通過一個共同的“記憶ID”進行關聯(lián)。當智能體需要回憶時可以先通過關系數(shù)據(jù)庫的元數(shù)據(jù)做初步篩選比如限定時間、類型再用篩選出的記憶ID去向量數(shù)據(jù)庫做精密的語義相似度匹配最終返回最相關的幾條記憶。2.2 OpenClaw與記憶系統(tǒng)的交互流程明確了架構接下來要設計OpenClaw如何與這個記憶系統(tǒng)“對話”。我設計了一個名為MemoryManager的核心模塊作為兩者之間的“經(jīng)紀人”。整個交互流程是這樣的記憶寫入Remember當OpenClaw在處理任務過程中產(chǎn)生了值得記錄的信息例如用戶說“我更喜歡用Markdown格式回復”MemoryManager會將其封裝成一個記憶對象。這個對象包含原始文本、自動提取的關鍵標簽、記憶類型、時間戳等。然后它同時向關系數(shù)據(jù)庫插入一條元數(shù)據(jù)記錄并向向量數(shù)據(jù)庫插入這條文本的向量化表示。這是一個原子操作必須確保兩者都成功否則回滾。記憶讀取Recall當OpenClaw需要背景信息時例如用戶問“我之前說的格式偏好是什么”它會向MemoryManager發(fā)起一個查詢請求。MemoryManager首先解析查詢意圖如果查詢條件明確如“類型用戶偏好”則先查詢關系數(shù)據(jù)庫獲取候選記憶ID列表。然后將原始查詢語句向量化在向量數(shù)據(jù)庫中針對這些候選ID或全部記憶進行相似度搜索返回得分最高的前N條記憶。記憶更新與清理記憶不是只增不減的。我設計了兩種策略。一是基于時間的衰減很久未觸發(fā)的記憶會被標記為“不活躍”。二是基于重要性的評估在寫入時可以由智能體或規(guī)則賦予一個初始重要性權重每次被成功召回并助力任務完成該權重增加反之如果記憶內(nèi)容被用戶糾正則權重降低。權重過低或過時的記憶會被歸檔或清理。注意這里的一個關鍵設計點是“記憶的粒度”。不要把一整段對話都存成一條記憶。更好的做法是按“信息點”進行拆分。比如一段關于項目需求的討論可以拆分為“項目目標實現(xiàn)X”、“技術棧Python, FastAPI”、“截止日期下周五”等多個獨立的記憶單元。這樣在召回時更精準也便于管理。3. 實戰(zhàn)部署搭建OpenClaw與Active Memory環(huán)境3.1 基礎環(huán)境與OpenClaw部署我的實驗環(huán)境是一臺Ubuntu 22.04的云服務器當然在Mac或Windows的Docker環(huán)境下流程也類似。首先解決OpenClaw的部署。目前最穩(wěn)定、隔離性最好的方式就是Docker。OpenClaw社區(qū)提供了官方鏡像但為了靈活性我更喜歡使用docker-compose來編排。# docker-compose.yml version: 3.8 services: openclaw: image: openclaw/openclaw:latest container_name: openclaw restart: unless-stopped ports: - 3000:3000 # Web UI端口 environment: - OLLAMA_BASE_URLhttp://host.docker.internal:11434 # 假設Ollama在宿主機 - DEFAULT_MODELllama3.2:latest # 默認使用的模型 - LOG_LEVELINFO volumes: - ./openclaw_data:/app/data # 掛載配置和數(shù)據(jù)卷 networks: - ai-net # 我們稍后會添加記憶相關的服務這里有幾個關鍵點OLLAMA_BASE_URL: OpenClaw本身不包含大模型它需要連接一個模型服務。我本地用Ollama運行了Llama 3.2模型所以這里配置為宿主機的Ollama服務。如果你把Ollama也放在Docker里需要改為服務名如http://ollama:11434。volumes: 一定要掛載數(shù)據(jù)卷否則容器重啟后所有配置和會話記錄都會丟失。先不急著運行等我們把記憶系統(tǒng)的組件也編排進來。運行docker-compose up -d訪問http://你的服務器IP:3000就能看到OpenClaw的Web界面了。首次使用需要在設置里配置好模型端點。3.2 Active Memory核心組件部署記憶系統(tǒng)需要兩個數(shù)據(jù)庫。為了簡化我都用Docker來部署。1. 向量數(shù)據(jù)庫ChromaDBChromaDB輕量且易于集成是快速原型的最佳選擇。我們在docker-compose.yml中新增一個服務。chromadb: image: chromadb/chroma:latest container_name: chromadb restart: unless-stopped ports: - 8000:8000 command: uvicorn chromadb.app:app --reload --workers 1 --host 0.0.0.0 --port 8000 environment: - IS_PERSISTENTTRUE - PERSIST_DIRECTORY/chroma/chroma_data volumes: - ./chroma_data:/chroma/chroma_data networks: - ai-net2. 關系數(shù)據(jù)庫PostgreSQL雖然SQLite更輕量但考慮到未來可能的多節(jié)點部署和更復雜的查詢我選擇了PostgreSQL。同樣在docker-compose.yml中添加。postgres: image: postgres:15-alpine container_name: postgres-memory restart: unless-stopped environment: POSTGRES_USER: openclaw POSTGRES_PASSWORD: your_secure_password_here # 務必修改 POSTGRES_DB: activememory ports: - 5432:5432 volumes: - ./postgres_data:/var/lib/postgresql/data networks: - ai-net現(xiàn)在更新后的docker-compose.yml包含了三個服務。運行docker-compose up -d一次性啟動所有服務。實操心得在生產(chǎn)環(huán)境中務必為PostgreSQL設置強密碼并將端口映射5432:5432考慮在內(nèi)網(wǎng)訪問或者通過Docker網(wǎng)絡內(nèi)部通信不要直接暴露在公網(wǎng)。我的做法是只將OpenClaw的3000端口通過Nginx反向代理并配置SSL暴露出去ChromaDB和PostgreSQL僅通過Docker內(nèi)部網(wǎng)絡 (ai-net) 供OpenClaw容器訪問。這樣更安全。3.3 開發(fā)MemoryManager橋梁模塊OpenClaw本身沒有內(nèi)置記憶系統(tǒng)我們需要開發(fā)一個插件或中間件。我選擇用Python編寫一個獨立的MemoryManager服務并通過OpenClaw的Skill技能機制或Webhook與之集成。首先創(chuàng)建memory_manager目錄結構如下memory_manager/ ├── app.py # FastAPI主應用提供記憶的CRUD接口 ├── memory_core.py # 記憶的核心邏輯寫入、查詢、向量化 ├── database.py # 數(shù)據(jù)庫連接與操作PostgreSQL, Chroma ├── requirements.txt └── Dockerfile1. 依賴文件 (requirements.txt)fastapi0.104.1 uvicorn[standard]0.24.0 sqlalchemy2.0.23 psycopg2-binary2.9.9 chromadb0.4.22 sentence-transformers2.2.2 # 用于本地Embedding可選 openai1.3.0 # 如果使用OpenAI的Embedding API pydantic2.5.02. 數(shù)據(jù)庫模型與連接 (database.py)這里定義記憶的元數(shù)據(jù)表結構。from sqlalchemy import create_engine, Column, String, DateTime, Text, Float, JSON from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime import os DATABASE_URL os.getenv(DATABASE_URL, postgresql://openclaw:your_passwordpostgres/activememory) engine create_engine(DATABASE_URL) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base declarative_base() class MemoryMetadata(Base): __tablename__ memory_metadata id Column(String, primary_keyTrue) # 與ChromaDB中的ID對應 content Column(Text, nullableFalse) # 原始文本內(nèi)容 memory_type Column(String, indexTrue) # 如user_preference, project_context, conversation tags Column(JSON) # 標簽列表如 [format, preference] source Column(String) # 來源如 openclaw_session_001 importance Column(Float, default1.0) # 重要性權重 last_accessed Column(DateTime, defaultdatetime.utcnow) created_at Column(DateTime, defaultdatetime.utcnow) # 創(chuàng)建表 Base.metadata.create_all(bindengine)3. 記憶核心邏輯 (memory_core.py)這是最核心的部分負責協(xié)調(diào)兩個數(shù)據(jù)庫。import uuid from datetime import datetime import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer # 示例用本地模型 # 或 from openai import OpenAI from database import SessionLocal, MemoryMetadata import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class MemoryCore: def __init__(self, embedding_model_nameall-MiniLM-L6-v2): # 初始化ChromaDB客戶端連接Docker中的服務 self.chroma_client chromadb.HttpClient( hostchromadb, # Docker服務名 port8000, settingsSettings(allow_resetTrue) ) # 獲取或創(chuàng)建集合類似于表 self.collection self.chroma_client.get_or_create_collection(nameactive_memory) # 初始化Embedding模型本地 # 注意本地模型首次加載慢但無需網(wǎng)絡。也可用OpenAI API。 self.embedder SentenceTransformer(embedding_model_name) logger.info(MemoryCore initialized.) def _generate_embedding(self, text: str): 生成文本的向量表示 # 使用本地模型 embedding self.embedder.encode(text).tolist() # 如果使用OpenAI API: # client OpenAI(api_keyyour_key) # response client.embeddings.create(modeltext-embedding-3-small, inputtext) # embedding response.data[0].embedding return embedding def remember(self, content: str, memory_type: str general, tags: list None, source: str None): 保存一條記憶 memory_id str(uuid.uuid4()) embedding self._generate_embedding(content) # 1. 存入向量數(shù)據(jù)庫 (ChromaDB) self.collection.add( documents[content], embeddings[embedding], ids[memory_id] ) # 2. 存入關系數(shù)據(jù)庫 (PostgreSQL) db SessionLocal() try: db_memory MemoryMetadata( idmemory_id, contentcontent, memory_typememory_type, tagstags or [], sourcesource, created_atdatetime.utcnow(), last_accesseddatetime.utcnow() ) db.add(db_memory) db.commit() logger.info(fMemory saved. ID: {memory_id}, Type: {memory_type}) except Exception as e: logger.error(fFailed to save metadata for {memory_id}: {e}) # 理想情況下這里應有事務回滾也需刪除剛存入Chroma的數(shù)據(jù) # 簡化處理記錄錯誤 db.rollback() finally: db.close() return memory_id def recall(self, query: str, memory_type: str None, limit: int 5): 回憶根據(jù)查詢語句和可選類型查找相關記憶 query_embedding self._generate_embedding(query) # 第一步如果指定了類型先從PostgreSQL獲取該類型的所有記憶ID memory_ids_filter None if memory_type: db SessionLocal() try: results db.query(MemoryMetadata.id).filter(MemoryMetadata.memory_type memory_type).all() memory_ids_filter [r[0] for r in results] logger.debug(fFiltering by type {memory_type}, found {len(memory_ids_filter)} IDs.) finally: db.close() # 第二步在ChromaDB中進行向量相似度查詢 # where_document 可以用于ChromaDB自身的元數(shù)據(jù)過濾但我們用PostgreSQL做了這里用ids過濾 results self.collection.query( query_embeddings[query_embedding], n_resultslimit, where{memory_type: memory_type} if memory_type else None, # Chroma的元數(shù)據(jù)過濾需在add時傳入 # 或者使用從PostgreSQL獲取的ID列表進行過濾如果集合很大先過濾更高效 # 這里演示使用Chroma的where條件前提是存入時傳入了memory_type ) # 第三步根據(jù)返回的ID從PostgreSQL獲取完整的元數(shù)據(jù)信息 recalled_memories [] if results and results[ids][0]: db SessionLocal() try: for mem_id in results[ids][0]: db_memory db.query(MemoryMetadata).filter(MemoryMetadata.id mem_id).first() if db_memory: # 更新最后訪問時間 db_memory.last_accessed datetime.utcnow() recalled_memories.append({ id: db_memory.id, content: db_memory.content, type: db_memory.memory_type, tags: db_memory.tags, source: db_memory.source, importance: db_memory.importance, similarity_score: results[distances][0][results[ids][0].index(mem_id)] if results.get(distances) else None }) db.commit() finally: db.close() logger.info(fRecalled {len(recalled_memories)} memories for query: {query}) return recalled_memories4. 構建API接口 (app.py)用FastAPI包裝核心功能供OpenClaw調(diào)用。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional, List from memory_core import MemoryCore import logging app FastAPI(titleActive Memory Service) memory_core MemoryCore() class MemoryCreate(BaseModel): content: str memory_type: Optional[str] general tags: Optional[List[str]] [] source: Optional[str] None class MemoryQuery(BaseModel): query: str memory_type: Optional[str] None limit: Optional[int] 5 app.post(/remember) async def remember(memory: MemoryCreate): 存儲一條新記憶 try: memory_id memory_core.remember( contentmemory.content, memory_typememory.memory_type, tagsmemory.tags, sourcememory.source ) return {message: Memory saved successfully, memory_id: memory_id} except Exception as e: logging.error(fError in /remember: {e}) raise HTTPException(status_code500, detailstr(e)) app.post(/recall) async def recall(query: MemoryQuery): 根據(jù)查詢回憶相關記憶 try: memories memory_core.recall( queryquery.query, memory_typequery.memory_type, limitquery.limit ) return {query: query.query, memories: memories} except Exception as e: logging.error(fError in /recall: {e}) raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health(): return {status: healthy}5. 編寫Dockerfile并加入編排為這個記憶服務也創(chuàng)建一個Docker鏡像。# memory_manager/Dockerfile FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, app:app, --host, 0.0.0.0, --port, 8001]最后更新總的docker-compose.yml加入我們的記憶服務。memory-service: build: ./memory_manager # 指向MemoryManager目錄 container_name: memory-service restart: unless-stopped ports: - 8001:8001 environment: - DATABASE_URLpostgresql://openclaw:your_passwordpostgres/activememory # 可以在這里設置Embedding模型類型或API密鑰 depends_on: - chromadb - postgres networks: - ai-net現(xiàn)在運行docker-compose up -d --build重新構建并啟動所有服務。訪問http://localhost:8001/docs可以看到自動生成的API文檔可以測試/remember和/recall接口。4. 集成與調(diào)試讓OpenClaw“學會”記憶4.1 通過Skill技能機制集成OpenClaw的強大之處在于其Skill系統(tǒng)。我們可以編寫一個自定義Skill讓OpenClaw在對話中自動調(diào)用記憶服務。在OpenClaw的配置目錄我們之前掛載的./openclaw_data下通常會有skills文件夾。我們創(chuàng)建一個新的Skill文件例如active_memory_skill.py。# ./openclaw_data/skills/active_memory_skill.py import requests import json from typing import Dict, Any MEMORY_SERVICE_URL http://memory-service:8001 # Docker內(nèi)部網(wǎng)絡通信 class ActiveMemorySkill: 一個讓OpenClaw具備主動記憶能力的技能 def __init__(self): self.name active_memory self.description 保存或回憶對話中的關鍵信息到長期記憶庫。 self.triggers [ 記住.*, # 當用戶說“記住我喜歡用藍色主題” 我之前說過.*, # 當用戶問“我之前說過我的偏好是什么” 回憶一下.*, 關于.*(你還記得嗎|你知道多少), ] def execute(self, context: Dict[str, Any]) - str: Skill執(zhí)行入口 user_input context.get(user_input, ).lower() session_id context.get(session_id, default) # 1. 判斷意圖是保存記憶還是回憶 if user_input.startswith(記住): # 提取要記憶的內(nèi)容例如“記住我喜歡用藍色主題” - “我喜歡用藍色主題” content_to_remember user_input[2:].strip() # 簡單處理 if content_to_remember: return self._save_memory(content_to_remember, user_preference, session_id) else: return 請告訴我需要記住什么內(nèi)容。 elif any(trigger in user_input for trigger in [我之前說過, 回憶一下, 還記得嗎, 你知道多少]): # 提取查詢關鍵詞這里做簡單提取實際可用更復雜的NLP # 例如“我之前說過我的偏好是什么” - “偏好” query_key user_input # 更佳實踐調(diào)用一個意圖識別函數(shù)來提取核心查詢詞 # 此處簡化直接用整個句子后半部分或關鍵詞 return self._recall_memory(query_key, session_id) # 2. 被動記憶在每次對話輪次結束后由OpenClaw框架自動調(diào)用保存關鍵信息 # 這通常需要在OpenClaw的鉤子函數(shù)中配置此處不展開。 return def _save_memory(self, content: str, mem_type: str, source: str) - str: 調(diào)用記憶服務保存記憶 try: payload { content: content, memory_type: mem_type, tags: self._extract_tags(content), # 可實現(xiàn)的標簽提取函數(shù) source: source } response requests.post(f{MEMORY_SERVICE_URL}/remember, jsonpayload, timeout5) if response.status_code 200: return f好的我已經(jīng)將‘{content[:30]}...’記到我的備忘錄里了。 else: return f記憶保存失敗{response.text} except requests.exceptions.RequestException as e: return f無法連接到記憶服務{(diào)e} def _recall_memory(self, query: str, source: str) - str: 調(diào)用記憶服務回憶 try: payload {query: query, limit: 3} response requests.post(f{MEMORY_SERVICE_URL}/recall, jsonpayload, timeout5) if response.status_code 200: data response.json() memories data.get(memories, []) if memories: # 格式化回憶結果 memory_texts [f- {mem[content]} (相關度: {mem.get(similarity_score, N/A):.2f}) for mem in memories[:2]] reply f根據(jù)我的記錄相關的內(nèi)容有\(zhòng)n \n.join(memory_texts) if len(memories) 2: reply f\n還有{len(memories)-2}條相關記錄 return reply else: return 我的記憶里暫時沒有找到相關信息。 else: return f記憶回憶失敗{response.text} except requests.exceptions.RequestException as e: return f無法連接到記憶服務{(diào)e} def _extract_tags(self, text: str) - list: 簡單的關鍵詞/標簽提取示例實際可用TF-IDF或小模型 # 這里只是一個示例實際應用應使用更成熟的方法 predefined_tags [偏好, 配置, 項目, 需求, 日期, 聯(lián)系人] found_tags [tag for tag in predefined_tags if tag in text] return found_tags if found_tags else [general] # OpenClaw Skill標準導出 def get_skill(): return ActiveMemorySkill()將這個文件放到OpenClaw的技能目錄后需要在OpenClaw的配置中啟用它。具體配置方式因OpenClaw版本而異通常是在Web UI的技能管理頁面添加或修改配置文件config.yaml添加技能路徑和初始化參數(shù)。4.2 配置OpenClaw調(diào)用記憶服務除了主動技能我們更希望OpenClaw能“潛移默化”地使用記憶。這需要在OpenClaw處理對話的流程中插入鉤子Hooks。OpenClaw的架構通常支持“前置處理器”和“后置處理器”。我們可以創(chuàng)建一個后置處理器在OpenClaw生成回復后自動分析本輪對話如果包含值得長期記憶的信息例如用戶明確了某個設置、陳述了一個事實就自動調(diào)用memory-service的/remember接口保存。同樣在OpenClaw生成回復前前置處理器可以自動根據(jù)當前對話的上下文調(diào)用/recall接口獲取相關記憶并將這些記憶作為附加上下文注入給大模型從而讓模型在“知情”的情況下進行回復。這部分集成深度依賴于OpenClaw的具體版本和擴展機制可能需要修改其核心代碼或利用其插件系統(tǒng)。一個常見的模式是在向大模型發(fā)送的Prompt模板中加入一個“相關記憶”的占位符由前置處理器負責填充。例如修改后的Prompt可能看起來像這樣你是一個有幫助的AI助手。以下是一些可能相關的歷史記錄你的記憶 {formatted_memories} 當前對話 用戶{user_input} 助手這樣大模型在生成回復時就能自然地引用記憶中的信息。5. 效果驗證與性能調(diào)優(yōu)5.1 功能測試與效果評估部署并集成完成后需要進行系統(tǒng)測試。1. 基礎CRUD測試通過記憶服務的API文檔頁面直接測試/remember和/recall確保接口工作正常。插入幾條測試記憶如{content: 用戶喜歡在晚上接收每日報告, memory_type: user_preference, tags: [report, schedule]}。用相關查詢?nèi)纭皥蟾鏁r間”進行回憶看是否能正確返回。2. OpenClaw技能測試在OpenClaw的Web界面中直接對AI說“記住我的項目‘AI助手’的API密鑰是sk-abc123請保密?！庇^察回復確認技能被觸發(fā)并返回成功信息。然后問“我之前告訴過你API密鑰嗎”或“關于AI助手項目你還記得什么”檢查OpenClaw的回復是否包含了之前存儲的記憶內(nèi)容。3. 自動化記憶測試進行一段多輪對話討論一個具體問題比如配置郵箱。在對話中故意說出一些關鍵信息如“我的郵箱服務器是smtp.example.com端口是587”。在后續(xù)對話中詢問“郵箱端口是多少”看OpenClaw是否能憑借記憶正確回答而無需你重新告知。5.2 性能瓶頸分析與優(yōu)化在實際使用中可能會遇到一些性能問題。1. 向量檢索速度慢問題當記憶條數(shù)超過數(shù)萬時ChromaDB的暴力相似度搜索可能會變慢。優(yōu)化索引確保ChromaDB使用了合適的索引如HNSW。在創(chuàng)建集合時可以通過參數(shù)配置。預過濾充分利用memory_type和tags在關系數(shù)據(jù)庫中進行預過濾 drastically減少需要做向量相似度計算的候選集大小。這正是我們混合架構的優(yōu)勢。分頁回憶時不要一次性取太多條limit參數(shù)合理設置如5-10條。2. Embedding生成成為瓶頸問題使用本地Sentence Transformer模型如all-MiniLM-L6-v2雖然免費但CPU推理在寫入大量記憶時可能較慢。優(yōu)化批處理將多個記憶內(nèi)容批量生成Embedding減少模型加載和調(diào)用的開銷。使用GPU如果服務器有GPU確保PyTorch和Transformer庫利用了CUDA。換用API服務對于高并發(fā)生產(chǎn)環(huán)境可以考慮使用OpenAI、Cohere或?qū)ｉT的高性能Embedding API服務它們通常速度更快且有速率限制管理。但會引入網(wǎng)絡延遲和成本。3. 記憶冗余與沖突問題用戶可能多次表達相同或矛盾的信息如“我喜歡藍色”和“主題改成黑色吧”。優(yōu)化去重在remember前可以先進行一次recall檢查是否有高度相似相似度超過0.95的現(xiàn)有記憶。如果有可以選擇更新原有記憶例如合并內(nèi)容、更新時間戳、增加權重而不是新增一條。沖突解決當檢測到新舊記憶矛盾時可以設計規(guī)則。例如默認以最新的信息為準但降低舊記憶的權重而非直接刪除或者在回憶時同時返回新舊記憶并在提示詞中告訴大模型“這里有兩條矛盾的信息請根據(jù)上下文判斷”。5.3 高級功能展望一個基礎的Active Memory系統(tǒng)已經(jīng)能極大提升體驗。在此基礎上還可以考慮更多增強功能記憶關聯(lián)圖不僅存儲孤立的記憶點還存儲記憶之間的關系。例如“項目A”使用了“技術B” “技術B”的專家是“聯(lián)系人C”。這可以通過在關系數(shù)據(jù)庫中增加一個related_memory_ids字段存儲關聯(lián)記憶ID列表來實現(xiàn)讓回憶時能進行“聯(lián)想”。記憶摘要對于長時間的對話或文檔可以定期或當記憶數(shù)量過多時調(diào)用大模型生成一個摘要作為一條新的、更高級別的“概要記憶”存儲起來從而壓縮信息提高長期記憶的效率。記憶失效與歸檔策略實現(xiàn)更復雜的記憶生命周期管理。例如設定不同記憶類型的TTL生存時間將長時間未訪問且重要性低的記憶移動到廉價的冷存儲如從ChromaDB/PostgreSQL轉移到文件定期清理“垃圾記憶”。6. 常見問題與故障排查實錄在部署和調(diào)試過程中我遇到了不少坑這里把典型問題和解決方法記錄下來。6.1 部署連接問題問題1OpenClaw容器內(nèi)無法連接到memory-service:8001?，F(xiàn)象Skill執(zhí)行時報錯“無法連接到記憶服務”Connection refused。排查進入OpenClaw容器docker exec -it openclaw bash。嘗試pingmemory-serviceping memory-service。如果不通說明Docker網(wǎng)絡有問題。檢查docker-compose.yml確保所有服務在同一個自定義網(wǎng)絡下如ai-net并且OpenClaw服務定義了depends_on: - memory-service這主要控制啟動順序不保證網(wǎng)絡可達但通常一起定義。解決確認網(wǎng)絡配置正確。最穩(wěn)妥的方式是在OpenClaw容器內(nèi)使用curl http://memory-service:8001/health測試連通性。如果不通檢查Docker網(wǎng)絡docker network ls和docker network inspect ai-net確保所有容器都連接到了該網(wǎng)絡。問題2ChromaDB連接失敗報錯Failed to connect?，F(xiàn)象MemoryCore初始化時連接ChromaDB超時或失敗。排查首先在宿主機上curl http://localhost:8000/api/v1/heartbeat檢查ChromaDB服務本身是否健康。如果宿主機通但memory-service容器內(nèi)不通檢查memory-service的Docker Compose配置中ChromaDB的host名是否正確。在Docker Compose中服務名chromadb就是主機名。檢查ChromaDB容器的日志docker logs chromadb看是否有啟動錯誤。解決確保ChromaDB的command中指定了--host 0.0.0.0以允許所有網(wǎng)絡接口連接。防火墻或安全組規(guī)則確保8000端口在容器間可訪問。6.2 技能與集成問題問題3OpenClaw不觸發(fā)自定義Skill?，F(xiàn)象在Web界面說話技能毫無反應。排查檢查Skill文件是否放在了正確的目錄通常是openclaw_data/skills/并且OpenClaw的配置指向了這個目錄。查看OpenClaw的日志docker logs openclaw尋找加載技能時的錯誤信息。檢查Skill類中的triggers列表。OpenClaw的觸發(fā)機制可能是正則表達式匹配或關鍵字匹配。確保你的用戶輸入能匹配上。例如“記住.*”是一個正則需要用戶輸入以“記住”開頭。可以先用簡單的[test]作為trigger來測試。確認Skill類被正確導出有get_skill()函數(shù)。解決仔細閱讀OpenClaw官方關于Skill開發(fā)的文檔確認其加載機制和觸發(fā)規(guī)則。一個有效的調(diào)試方法是在Skill的execute方法開頭加入日志打印確認方法是否被調(diào)用。問題4記憶回憶的結果不相關?，F(xiàn)象用戶問“郵箱設置”返回的卻是關于“晚餐吃什么”的記憶。排查Embedding模型問題使用的Embedding模型是否適合中文all-MiniLM-L6-v2對英文優(yōu)化更好?？梢試L試換用多語言模型如paraphrase-multilingual-MiniLM-L12-v2。查詢詞過于寬泛“郵箱設置”可能被Embedding成一個比較泛的向量。嘗試在回憶前對用戶查詢進行輕微的改寫或擴展例如結合對話上下文將查詢擴展為“用戶詢問郵箱服務器和端口的設置信息”。記憶粒度問題存入的記憶文本是否太冗長或包含無關信息確保存入的是干凈、核心的信息點。解決更換或微調(diào)Embedding模型優(yōu)化記憶的寫入內(nèi)容使其更聚焦在回憶時嘗試將當前對話的最近幾條消息一起作為查詢上下文提升相關性。6.3 數(shù)據(jù)庫與性能問題問題5PostgreSQL連接數(shù)過多?，F(xiàn)象運行一段時間后memory-service出現(xiàn)too many connections錯誤。原因SQLAlchemy的Session沒有正確關閉。在memory_core.py的recall和remember方法中雖然用了try...finally來關閉session但在異常處理分支中可能仍有遺漏。解決使用上下文管理器確保Session總是被關閉?；蛘邽镕astAPI應用配置SQLAlchemy的scoped_session并確保在每個請求結束后移除session。更簡單的方法是在database.py中創(chuàng)建一個依賴項。# 在database.py中 def get_db(): db SessionLocal() try: yield db finally: db.close() # 在FastAPI路由中 from fastapi import Depends from sqlalchemy.orm import Session app.post(/remember) async def remember(memory: MemoryCreate, db: Session Depends(get_db)): # ... 使用db session # 無需手動關閉依賴項會自動處理問題6ChromaDB數(shù)據(jù)持久化失敗?，F(xiàn)象重啟Docker Compose后之前存儲的記憶全部消失。排查檢查ChromaDB的容器是否配置了持久化卷并且PERSIST_DIRECTORY環(huán)境變量指向了卷內(nèi)路徑。檢查docker-compose.yml中chromadb服務的volumes映射和environment設置。解決確保volumes: - ./chroma_data:/chroma/chroma_data存在并且目錄./chroma_data在宿主機上有寫入權限。同時ChromaDB容器的command中不能有--reload參數(shù)用于開發(fā)在生產(chǎn)中可能引發(fā)問題可以去掉。這個融合項目從構想到實現(xiàn)花費了不少精力但結果是值得的?？粗鳲penClaw從“金魚腦”變成一個有“長期記憶”的靠譜助手能記住項目細節(jié)、用戶偏好并在后續(xù)對話中自然引用那種體驗的提升是質(zhì)的飛躍。最關鍵的是整個架構基于開源組件搭建完全可控可以根據(jù)自己的需求靈活調(diào)整記憶的邏輯和存儲策略。如果你也在探索AI智能體的長期記憶問題希望這份詳細的實戰(zhàn)記錄能幫你少走些彎路。