1. 项目概述:当“小龙虾”拥有了“记忆”

最近在折腾本地AI智能体,OpenClaw(圈内戏称“小龙虾”)这个名字出现的频率越来越高。它本质上是一个开源的AI智能体框架,你可以把它理解为一个“大脑”,负责调度和协调各种AI能力去完成复杂的任务。但玩过一阵子后,我发现一个普遍痛点:这“大脑”记性不太好。每次对话都像是初次见面,上下文一长就断片,更别提让它记住我的个人偏好、项目背景这些长期信息了。这严重限制了它的实用性,尤其是在处理需要持续跟踪状态的任务时,比如自动化客服、项目管理或者个人知识库助手。

于是,我开始寻找解决方案,目标很明确:给OpenClaw这个聪明的“大脑”装上一个可靠的“记事本”。这就是“Active Memory”(主动记忆)概念的由来。它不是一个简单的聊天记录存储器,而是一个能够被智能体主动查询、更新、关联的结构化记忆系统。简单说,就是让OpenClaw学会“做笔记”和“翻笔记”。

这个融合项目,就是要把OpenClaw的智能决策能力,与一个持久化、可操作的内存系统结合起来。我最终选择了一个基于向量数据库(比如ChromaDB或Qdrant)和关系型数据库(如SQLite)的混合架构来实现Active Memory。向量库负责语义搜索,快速找到相关记忆;关系库则存储记忆的元数据、时间戳和关联关系。下面,我就把这次从设计思路到踩坑填坑的完整实战过程拆解出来。

2. 核心架构设计与技术选型

2.1 为什么是“混合记忆”架构?

一开始,我考虑过几种简单的方案。比如,只用向量数据库,把所有对话都存成向量。但问题很快暴露:查找是快了,但我想按时间筛选(“昨天我们讨论的那个需求”),或者按类型查找(“所有关于‘部署’的笔记”),向量检索就显得力不从心。反之,如果只用关系数据库,虽然能方便地按字段查询,但做“帮我找找和‘自动化流程’相关的所有内容”这种模糊语义搜索,效率又很低。

所以,混合架构成了必然选择。它的核心思想是“分工协作”:

  1. 关系型数据库(如SQLite) :充当“记忆的目录和索引”。它存储每条记忆的唯一ID、创建时间、记忆类型(是“用户偏好”、“项目上下文”还是“会话历史”)、关键标签、以及关联的实体(如项目名、联系人)。它的优势是结构化查询非常快且精准。
  2. 向量数据库(如ChromaDB) :充当“记忆的内容搜索引擎”。它存储记忆文本内容经过Embedding模型转换后的向量。当OpenClaw需要回忆时,可以将当前问题的语义转换成向量,然后在向量空间里快速找到最相似的几条记忆内容。

两者通过一个共同的“记忆ID”进行关联。当智能体需要回忆时,可以先通过关系数据库的元数据做初步筛选(比如限定时间、类型),再用筛选出的记忆ID去向量数据库做精密的语义相似度匹配,最终返回最相关的几条记忆。

2.2 OpenClaw与记忆系统的交互流程

明确了架构,接下来要设计OpenClaw如何与这个记忆系统“对话”。我设计了一个名为 MemoryManager 的核心模块,作为两者之间的“经纪人”。

整个交互流程是这样的:

  1. 记忆写入(Remember) :当OpenClaw在处理任务过程中,产生了值得记录的信息(例如,用户说“我更喜欢用Markdown格式回复”), MemoryManager 会将其封装成一个记忆对象。这个对象包含原始文本、自动提取的关键标签、记忆类型、时间戳等。然后,它 同时 向关系数据库插入一条元数据记录,并向向量数据库插入这条文本的向量化表示。这是一个原子操作,必须确保两者都成功,否则回滚。
  2. 记忆读取(Recall) :当OpenClaw需要背景信息时(例如,用户问“我之前说的格式偏好是什么?”),它会向 MemoryManager 发起一个查询请求。 MemoryManager 首先解析查询意图,如果查询条件明确(如“类型=用户偏好”),则先查询关系数据库获取候选记忆ID列表。然后,将原始查询语句向量化,在向量数据库中,针对这些候选ID(或全部记忆)进行相似度搜索,返回得分最高的前N条记忆。
  3. 记忆更新与清理 :记忆不是只增不减的。我设计了两种策略。一是基于时间的衰减,很久未触发的记忆会被标记为“不活跃”。二是基于重要性的评估,在写入时可以由智能体或规则赋予一个初始重要性权重,每次被成功召回并助力任务完成,该权重增加;反之,如果记忆内容被用户纠正,则权重降低。权重过低或过时的记忆,会被归档或清理。

注意 :这里的一个关键设计点是“记忆的粒度”。不要把一整段对话都存成一条记忆。更好的做法是,按“信息点”进行拆分。比如,一段关于项目需求的讨论,可以拆分为“项目目标:实现X”、“技术栈:Python, FastAPI”、“截止日期:下周五”等多个独立的记忆单元。这样在召回时更精准,也便于管理。

3. 实战部署:搭建OpenClaw与Active Memory环境

3.1 基础环境与OpenClaw部署

我的实验环境是一台Ubuntu 22.04的云服务器,当然,在Mac或Windows的Docker环境下流程也类似。首先解决OpenClaw的部署。

目前最稳定、隔离性最好的方式就是Docker。OpenClaw社区提供了官方镜像,但为了灵活性,我更喜欢使用 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_URL=http://host.docker.internal:11434 # 假设Ollama在宿主机
      - DEFAULT_MODEL=llama3.2:latest # 默认使用的模型
      - LOG_LEVEL=INFO
    volumes:
      - ./openclaw_data:/app/data # 挂载配置和数据卷
    networks:
      - ai-net

  # 我们稍后会添加记忆相关的服务

这里有几个关键点:

  • OLLAMA_BASE_URL : OpenClaw本身不包含大模型,它需要连接一个模型服务。我本地用Ollama运行了Llama 3.2模型,所以这里配置为宿主机的Ollama服务。如果你把Ollama也放在Docker里,需要改为服务名,如 http://ollama:11434
  • volumes : 一定要挂载数据卷,否则容器重启后所有配置和会话记录都会丢失。
  • 先不急着运行,等我们把记忆系统的组件也编排进来。

运行 docker-compose up -d ,访问 http://你的服务器IP:3000 ,就能看到OpenClaw的Web界面了。首次使用需要在设置里配置好模型端点。

3.2 Active Memory核心组件部署

记忆系统需要两个数据库。为了简化,我都用Docker来部署。

1. 向量数据库:ChromaDB ChromaDB轻量且易于集成,是快速原型的最佳选择。我们在 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_PERSISTENT=TRUE
      - PERSIST_DIRECTORY=/chroma/chroma_data
    volumes:
      - ./chroma_data:/chroma/chroma_data
    networks:
      - ai-net

2. 关系数据库:PostgreSQL 虽然SQLite更轻量,但考虑到未来可能的多节点部署和更复杂的查询,我选择了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

现在,更新后的 docker-compose.yml 包含了三个服务。运行 docker-compose up -d 一次性启动所有服务。

实操心得 :在生产环境中,务必为PostgreSQL设置强密码,并将端口映射( 5432:5432 )考虑在内网访问,或者通过Docker网络内部通信,不要直接暴露在公网。我的做法是,只将OpenClaw的3000端口通过Nginx反向代理并配置SSL暴露出去,ChromaDB和PostgreSQL仅通过Docker内部网络 ( ai-net ) 供OpenClaw容器访问。这样更安全。

3.3 开发MemoryManager桥梁模块

OpenClaw本身没有内置记忆系统,我们需要开发一个插件或中间件。我选择用Python编写一个独立的 MemoryManager 服务,并通过OpenClaw的Skill(技能)机制或Webhook与之集成。

首先,创建 memory_manager 目录,结构如下:

memory_manager/
├── app.py          # FastAPI主应用,提供记忆的CRUD接口
├── memory_core.py  # 记忆的核心逻辑(写入、查询、向量化)
├── database.py     # 数据库连接与操作(PostgreSQL, Chroma)
├── requirements.txt
└── Dockerfile

1. 依赖文件 ( requirements.txt )

fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
psycopg2-binary==2.9.9
chromadb==0.4.22
sentence-transformers==2.2.2 # 用于本地Embedding,可选
openai==1.3.0 # 如果使用OpenAI的Embedding API
pydantic==2.5.0

2. 数据库模型与连接 ( database.py ) 这里定义记忆的元数据表结构。

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_password@postgres/activememory")
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class MemoryMetadata(Base):
    __tablename__ = "memory_metadata"
    id = Column(String, primary_key=True) # 与ChromaDB中的ID对应
    content = Column(Text, nullable=False) # 原始文本内容
    memory_type = Column(String, index=True) # 如:user_preference, project_context, conversation
    tags = Column(JSON) # 标签列表,如 ["format", "preference"]
    source = Column(String) # 来源,如 "openclaw_session_001"
    importance = Column(Float, default=1.0) # 重要性权重
    last_accessed = Column(DateTime, default=datetime.utcnow)
    created_at = Column(DateTime, default=datetime.utcnow)

# 创建表
Base.metadata.create_all(bind=engine)

3. 记忆核心逻辑 ( memory_core.py ) 这是最核心的部分,负责协调两个数据库。

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(level=logging.INFO)
logger = logging.getLogger(__name__)

class MemoryCore:
    def __init__(self, embedding_model_name='all-MiniLM-L6-v2'):
        # 初始化ChromaDB客户端(连接Docker中的服务)
        self.chroma_client = chromadb.HttpClient(
            host='chromadb', # Docker服务名
            port=8000,
            settings=Settings(allow_reset=True)
        )
        # 获取或创建集合(类似于表)
        self.collection = self.chroma_client.get_or_create_collection(name="active_memory")
        
        # 初始化Embedding模型(本地)
        # 注意:本地模型首次加载慢,但无需网络。也可用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_key=your_key)
        # response = client.embeddings.create(model="text-embedding-3-small", input=text)
        # 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. 存入向量数据库 (ChromaDB)
        self.collection.add(
            documents=[content],
            embeddings=[embedding],
            ids=[memory_id]
        )
        
        # 2. 存入关系数据库 (PostgreSQL)
        db = SessionLocal()
        try:
            db_memory = MemoryMetadata(
                id=memory_id,
                content=content,
                memory_type=memory_type,
                tags=tags or [],
                source=source,
                created_at=datetime.utcnow(),
                last_accessed=datetime.utcnow()
            )
            db.add(db_memory)
            db.commit()
            logger.info(f"Memory saved. ID: {memory_id}, Type: {memory_type}")
        except Exception as e:
            logger.error(f"Failed to save metadata for {memory_id}: {e}")
            # 理想情况下,这里应有事务回滚,也需删除刚存入Chroma的数据
            # 简化处理,记录错误
            db.rollback()
        finally:
            db.close()
        
        return memory_id

    def recall(self, query: str, memory_type: str = None, limit: int = 5):
        """回忆:根据查询语句和可选类型查找相关记忆"""
        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(f"Filtering by type '{memory_type}', found {len(memory_ids_filter)} IDs.")
            finally:
                db.close()
        
        # 第二步:在ChromaDB中进行向量相似度查询
        # where_document 可以用于ChromaDB自身的元数据过滤,但我们用PostgreSQL做了,这里用ids过滤
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=limit,
            where={"memory_type": memory_type} if memory_type else None, # Chroma的元数据过滤,需在add时传入
            # 或者使用从PostgreSQL获取的ID列表进行过滤(如果集合很大,先过滤更高效)
            # 这里演示使用Chroma的where条件,前提是存入时传入了memory_type
        )
        
        # 第三步:根据返回的ID,从PostgreSQL获取完整的元数据信息
        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(f"Recalled {len(recalled_memories)} memories for query: '{query}'")
        return recalled_memories

4. 构建API接口 ( app.py ) 用FastAPI包装核心功能,供OpenClaw调用。

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List
from memory_core import MemoryCore
import logging

app = FastAPI(title="Active 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(
            content=memory.content,
            memory_type=memory.memory_type,
            tags=memory.tags,
            source=memory.source
        )
        return {"message": "Memory saved successfully", "memory_id": memory_id}
    except Exception as e:
        logging.error(f"Error in /remember: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/recall")
async def recall(query: MemoryQuery):
    """根据查询回忆相关记忆"""
    try:
        memories = memory_core.recall(
            query=query.query,
            memory_type=query.memory_type,
            limit=query.limit
        )
        return {"query": query.query, "memories": memories}
    except Exception as e:
        logging.error(f"Error in /recall: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "healthy"}

5. 编写Dockerfile并加入编排 为这个记忆服务也创建一个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_URL=postgresql://openclaw:your_password@postgres/activememory
      # 可以在这里设置Embedding模型类型或API密钥
    depends_on:
      - chromadb
      - postgres
    networks:
      - ai-net

现在,运行 docker-compose up -d --build 重新构建并启动所有服务。访问 http://localhost:8001/docs 可以看到自动生成的API文档,可以测试 /remember /recall 接口。

4. 集成与调试:让OpenClaw“学会”记忆

4.1 通过Skill技能机制集成

OpenClaw的强大之处在于其Skill系统。我们可以编写一个自定义Skill,让OpenClaw在对话中自动调用记忆服务。

在OpenClaw的配置目录(我们之前挂载的 ./openclaw_data )下,通常会有 skills 文件夹。我们创建一个新的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内部网络通信

class ActiveMemorySkill:
    """一个让OpenClaw具备主动记忆能力的技能"""
    
    def __init__(self):
        self.name = "active_memory"
        self.description = "保存或回忆对话中的关键信息到长期记忆库。"
        self.triggers = [
            "记住.*", # 当用户说“记住,我喜欢用蓝色主题”
            "我之前说过.*", # 当用户问“我之前说过我的偏好是什么?”
            "回忆一下.*",
            "关于.*(你还记得吗|你知道多少)",
        ]
        
    def execute(self, context: Dict[str, Any]) -> str:
        """Skill执行入口"""
        user_input = context.get("user_input", "").lower()
        session_id = context.get("session_id", "default")
        
        # 1. 判断意图:是保存记忆还是回忆?
        if user_input.startswith("记住"):
            # 提取要记忆的内容,例如“记住,我喜欢用蓝色主题” -> “我喜欢用蓝色主题”
            content_to_remember = user_input[2:].strip() # 简单处理
            if content_to_remember:
                return self._save_memory(content_to_remember, "user_preference", session_id)
            else:
                return "请告诉我需要记住什么内容。"
        
        elif any(trigger in user_input for trigger in ["我之前说过", "回忆一下", "还记得吗", "你知道多少"]):
            # 提取查询关键词,这里做简单提取,实际可用更复杂的NLP
            # 例如“我之前说过我的偏好是什么?” -> “偏好”
            query_key = user_input
            # 更佳实践:调用一个意图识别函数来提取核心查询词
            # 此处简化,直接用整个句子后半部分或关键词
            return self._recall_memory(query_key, session_id)
        
        # 2. 被动记忆:在每次对话轮次结束后,由OpenClaw框架自动调用,保存关键信息
        # 这通常需要在OpenClaw的钩子函数中配置,此处不展开。
        return ""

    def _save_memory(self, content: str, mem_type: str, source: str) -> str:
        """调用记忆服务保存记忆"""
        try:
            payload = {
                "content": content,
                "memory_type": mem_type,
                "tags": self._extract_tags(content), # 可实现的标签提取函数
                "source": source
            }
            response = requests.post(f"{MEMORY_SERVICE_URL}/remember", json=payload, timeout=5)
            if response.status_code == 200:
                return f"好的,我已经将‘{content[:30]}...’记到我的备忘录里了。"
            else:
                return f"记忆保存失败:{response.text}"
        except requests.exceptions.RequestException as e:
            return f"无法连接到记忆服务:{e}"

    def _recall_memory(self, query: str, source: str) -> str:
        """调用记忆服务回忆"""
        try:
            payload = {"query": query, "limit": 3}
            response = requests.post(f"{MEMORY_SERVICE_URL}/recall", json=payload, timeout=5)
            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"根据我的记录,相关的内容有:\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"无法连接到记忆服务:{e}"

    def _extract_tags(self, text: str) -> list:
        """简单的关键词/标签提取(示例,实际可用TF-IDF或小模型)"""
        # 这里只是一个示例,实际应用应使用更成熟的方法
        predefined_tags = ["偏好", "配置", "项目", "需求", "日期", "联系人"]
        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 ,添加技能路径和初始化参数。

4.2 配置OpenClaw调用记忆服务

除了主动技能,我们更希望OpenClaw能“潜移默化”地使用记忆。这需要在OpenClaw处理对话的流程中插入钩子(Hooks)。OpenClaw的架构通常支持“前置处理器”和“后置处理器”。

我们可以创建一个后置处理器,在OpenClaw生成回复后,自动分析本轮对话,如果包含值得长期记忆的信息(例如,用户明确了某个设置、陈述了一个事实),就自动调用 memory-service /remember 接口保存。

同样,在OpenClaw生成回复前(前置处理器),可以自动根据当前对话的上下文,调用 /recall 接口获取相关记忆,并将这些记忆作为附加上下文注入给大模型,从而让模型在“知情”的情况下进行回复。

这部分集成深度依赖于OpenClaw的具体版本和扩展机制,可能需要修改其核心代码或利用其插件系统。一个常见的模式是,在向大模型发送的Prompt模板中,加入一个“相关记忆”的占位符,由前置处理器负责填充。

例如,修改后的Prompt可能看起来像这样:

你是一个有帮助的AI助手。以下是一些可能相关的历史记录(你的记忆):
{formatted_memories}

当前对话:
用户:{user_input}
助手:

这样,大模型在生成回复时,就能自然地引用记忆中的信息。

5. 效果验证与性能调优

5.1 功能测试与效果评估

部署并集成完成后,需要进行系统测试。

1. 基础CRUD测试:

  • 通过记忆服务的API文档页面,直接测试 /remember /recall ,确保接口工作正常。
  • 插入几条测试记忆,如 {"content": "用户喜欢在晚上接收每日报告", "memory_type": "user_preference", "tags": ["report", "schedule"]}
  • 用相关查询如“报告时间”进行回忆,看是否能正确返回。

2. OpenClaw技能测试:

  • 在OpenClaw的Web界面中,直接对AI说:“记住,我的项目‘AI助手’的API密钥是sk-abc123,请保密。”
  • 观察回复,确认技能被触发,并返回成功信息。
  • 然后问:“我之前告诉过你API密钥吗?”或“关于AI助手项目,你还记得什么?”
  • 检查OpenClaw的回复是否包含了之前存储的记忆内容。

3. 自动化记忆测试:

  • 进行一段多轮对话,讨论一个具体问题,比如配置邮箱。
  • 在对话中,故意说出一些关键信息,如“我的邮箱服务器是smtp.example.com,端口是587”。
  • 在后续对话中,询问“邮箱端口是多少?”,看OpenClaw是否能凭借记忆正确回答,而无需你重新告知。

5.2 性能瓶颈分析与优化

在实际使用中,可能会遇到一些性能问题。

1. 向量检索速度慢:

  • 问题 :当记忆条数超过数万时,ChromaDB的暴力相似度搜索可能会变慢。
  • 优化
    • 索引 :确保ChromaDB使用了合适的索引(如HNSW)。在创建集合时可以通过参数配置。
    • 预过滤 :充分利用 memory_type tags 在关系数据库中进行预过滤, drastically减少需要做向量相似度计算的候选集大小。这正是我们混合架构的优势。
    • 分页 :回忆时不要一次性取太多条( limit 参数合理设置,如5-10条)。

2. Embedding生成成为瓶颈:

  • 问题 :使用本地Sentence Transformer模型(如 all-MiniLM-L6-v2 )虽然免费,但CPU推理在写入大量记忆时可能较慢。
  • 优化
    • 批处理 :将多个记忆内容批量生成Embedding,减少模型加载和调用的开销。
    • 使用GPU :如果服务器有GPU,确保PyTorch和Transformer库利用了CUDA。
    • 换用API服务 :对于高并发生产环境,可以考虑使用OpenAI、Cohere或专门的高性能Embedding API服务,它们通常速度更快且有速率限制管理。但会引入网络延迟和成本。

3. 记忆冗余与冲突:

  • 问题 :用户可能多次表达相同或矛盾的信息(如“我喜欢蓝色”和“主题改成黑色吧”)。
  • 优化
    • 去重 :在 remember 前,可以先进行一次 recall ,检查是否有高度相似(相似度超过0.95)的现有记忆。如果有,可以选择更新原有记忆(例如,合并内容、更新时间戳、增加权重),而不是新增一条。
    • 冲突解决 :当检测到新旧记忆矛盾时,可以设计规则。例如,默认以最新的信息为准,但降低旧记忆的权重而非直接删除;或者,在回忆时同时返回新旧记忆,并在提示词中告诉大模型“这里有两条矛盾的信息,请根据上下文判断”。

5.3 高级功能展望

一个基础的Active Memory系统已经能极大提升体验。在此基础上,还可以考虑更多增强功能:

  1. 记忆关联图 :不仅存储孤立的记忆点,还存储记忆之间的关系。例如,“项目A”使用了“技术B”, “技术B”的专家是“联系人C”。这可以通过在关系数据库中增加一个 related_memory_ids 字段(存储关联记忆ID列表)来实现,让回忆时能进行“联想”。
  2. 记忆摘要 :对于长时间的对话或文档,可以定期(或当记忆数量过多时)调用大模型生成一个摘要,作为一条新的、更高级别的“概要记忆”存储起来,从而压缩信息,提高长期记忆的效率。
  3. 记忆失效与归档策略 :实现更复杂的记忆生命周期管理。例如,设定不同记忆类型的TTL(生存时间);将长时间未访问且重要性低的记忆移动到廉价的冷存储(如从ChromaDB/PostgreSQL转移到文件);定期清理“垃圾记忆”。

6. 常见问题与故障排查实录

在部署和调试过程中,我遇到了不少坑,这里把典型问题和解决方法记录下来。

6.1 部署连接问题

问题1:OpenClaw容器内无法连接到 memory-service:8001

  • 现象 :Skill执行时报错“无法连接到记忆服务”, Connection refused
  • 排查
    1. 进入OpenClaw容器: docker exec -it openclaw bash
    2. 尝试ping memory-service ping memory-service 。如果不通,说明Docker网络有问题。
    3. 检查 docker-compose.yml ,确保所有服务在同一个自定义网络下(如 ai-net ),并且OpenClaw服务定义了 depends_on: - memory-service (这主要控制启动顺序,不保证网络可达,但通常一起定义)。
  • 解决 :确认网络配置正确。最稳妥的方式是在OpenClaw容器内使用 curl http://memory-service:8001/health 测试连通性。如果不通,检查Docker网络 docker network ls docker network inspect ai-net ,确保所有容器都连接到了该网络。

问题2:ChromaDB连接失败,报错 Failed to connect

  • 现象 MemoryCore 初始化时连接ChromaDB超时或失败。
  • 排查
    1. 首先在宿主机上 curl http://localhost:8000/api/v1/heartbeat 检查ChromaDB服务本身是否健康。
    2. 如果宿主机通,但 memory-service 容器内不通,检查 memory-service 的Docker Compose配置中,ChromaDB的host名是否正确。在Docker Compose中,服务名( chromadb )就是主机名。
    3. 检查ChromaDB容器的日志: docker logs chromadb ,看是否有启动错误。
  • 解决 :确保ChromaDB的 command 中指定了 --host 0.0.0.0 以允许所有网络接口连接。防火墙或安全组规则确保8000端口在容器间可访问。

6.2 技能与集成问题

问题3:OpenClaw不触发自定义Skill。

  • 现象 :在Web界面说话,技能毫无反应。
  • 排查
    1. 检查Skill文件是否放在了正确的目录(通常是 openclaw_data/skills/ ),并且OpenClaw的配置指向了这个目录。
    2. 查看OpenClaw的日志 docker logs openclaw ,寻找加载技能时的错误信息。
    3. 检查Skill类中的 triggers 列表。OpenClaw的触发机制可能是正则表达式匹配或关键字匹配。确保你的用户输入能匹配上。例如, “记住.*” 是一个正则,需要用户输入以“记住”开头。可以先用简单的 ["test"] 作为trigger来测试。
    4. 确认Skill类被正确导出(有 get_skill() 函数)。
  • 解决 :仔细阅读OpenClaw官方关于Skill开发的文档,确认其加载机制和触发规则。一个有效的调试方法是,在Skill的 execute 方法开头加入日志打印,确认方法是否被调用。

问题4:记忆回忆的结果不相关。

  • 现象 :用户问“邮箱设置”,返回的却是关于“晚餐吃什么”的记忆。
  • 排查
    1. Embedding模型问题 :使用的Embedding模型是否适合中文? all-MiniLM-L6-v2 对英文优化更好。可以尝试换用多语言模型,如 paraphrase-multilingual-MiniLM-L12-v2
    2. 查询词过于宽泛 :“邮箱设置”可能被Embedding成一个比较泛的向量。尝试在回忆前,对用户查询进行轻微的改写或扩展,例如,结合对话上下文,将查询扩展为“用户询问邮箱服务器和端口的设置信息”。
    3. 记忆粒度问题 :存入的记忆文本是否太冗长或包含无关信息?确保存入的是干净、核心的信息点。
  • 解决 :更换或微调Embedding模型;优化记忆的写入内容,使其更聚焦;在回忆时,尝试将当前对话的最近几条消息一起作为查询上下文,提升相关性。

6.3 数据库与性能问题

问题5:PostgreSQL连接数过多。

  • 现象 :运行一段时间后, memory-service 出现 too many connections 错误。
  • 原因 :SQLAlchemy的Session没有正确关闭。在 memory_core.py recall remember 方法中,虽然用了 try...finally 来关闭session,但在异常处理分支中可能仍有遗漏。
  • 解决 :使用上下文管理器确保Session总是被关闭。或者,为FastAPI应用配置SQLAlchemy的 scoped_session ,并确保在每个请求结束后移除session。更简单的方法是,在 database.py 中创建一个依赖项。
# 在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
    # 无需手动关闭,依赖项会自动处理

问题6:ChromaDB数据持久化失败。

  • 现象 :重启Docker Compose后,之前存储的记忆全部消失。
  • 排查 :检查ChromaDB的容器是否配置了持久化卷,并且 PERSIST_DIRECTORY 环境变量指向了卷内路径。检查 docker-compose.yml chromadb 服务的 volumes 映射和 environment 设置。
  • 解决 :确保 volumes: - ./chroma_data:/chroma/chroma_data 存在,并且目录 ./chroma_data 在宿主机上有写入权限。同时,ChromaDB容器的 command 中不能有 --reload 参数(用于开发),在生产中可能引发问题,可以去掉。

这个融合项目从构想到实现,花费了不少精力,但结果是值得的。看着OpenClaw从“金鱼脑”变成一个有“长期记忆”的靠谱助手,能记住项目细节、用户偏好,并在后续对话中自然引用,那种体验的提升是质的飞跃。最关键的是,整个架构基于开源组件搭建,完全可控,可以根据自己的需求灵活调整记忆的逻辑和存储策略。如果你也在探索AI智能体的长期记忆问题,希望这份详细的实战记录能帮你少走些弯路。

更多推荐