1. 从“hindsight”说起:为什么我们需要给 Agent 装一个“后视镜”
“hindsight”这个词本身挺有意思,字面意思是“事后的洞察力”,也就是我们常说的“后见之明”。放在 LLM Agent 的语境里,它指向一个非常具体且要命的问题:Agent 的记忆到底该怎么存、怎么取、怎么用。
我接触过不少做 Agent 的团队,大家一开始都特别乐观,觉得只要把对话历史一股脑塞进上下文窗口就完事了。结果跑不了几轮就发现,token 烧得飞快,模型还经常“失忆”——明明上一轮刚说过的事情,下一轮就忘得一干二净。更麻烦的是,当你想让 Agent 跨会话记住用户偏好、记住之前踩过的坑,单纯靠 context window 根本撑不住。
这就是“hindsight”这个项目标题背后真正要解决的核心矛盾:Agent 需要一种能够跨越时间、跨越会话、跨越任务边界的记忆机制,而且这种机制不能只是简单的“存下来”,还得能在需要的时候精准地“想起来”。热搜词里出现的 agent memory、working memory、MCP、Docker 这些关键词,其实已经把技术栈的轮廓勾勒出来了——这是一个围绕 LLM Agent 记忆系统展开的工程实践,涉及记忆的存储结构、检索策略、协议对接和部署方式。
我打算把这篇文章写成一份“从零到一搭建 Agent 记忆系统”的实操记录。不管你是刚接触 LLM 应用开发的新手,还是已经在做 Agent 产品但被记忆问题折磨过的老手,下面这些内容应该都能让你少走一些弯路。我会把记忆系统的设计思路、核心数据结构、MCP 协议的对接方式、Docker 化部署的完整流程,以及我在实际调试中踩过的坑,全部摊开来聊。
2. Agent 记忆系统的整体设计与核心思路拆解
2.1 为什么“把历史对话全塞进去”是最蠢的做法
先算一笔账。假设你的 Agent 每轮对话平均产生 500 个 token 的文本,用户和 Agent 一来一回就是 1000 token。如果用户连续聊了 50 轮,那就是 50000 token 的上下文。现在主流模型的上下文窗口虽然标称 128K 甚至 200K,但你真把 50K 的历史全塞进去,先不说成本问题,模型对中间部分的注意力衰减是非常明显的——这就是著名的“lost in the middle”现象。
更关键的是,这 50 轮对话里,真正有价值的信息可能只占 10%。用户说“我明天要去北京出差”,这是有价值的事实;用户说“嗯嗯好的”,这是噪音。你把噪音和事实一起塞进去,模型提取有效信息的难度反而增加了。
所以记忆系统的第一个设计原则就是:存储要全,但注入要精。原始对话可以完整落盘,但每次调用模型时,只把最相关的记忆片段检索出来注入上下文。这个“检索”的过程,就是 hindsight 的核心价值所在。
2.2 记忆的分层结构:working memory、episodic memory、semantic memory
我在设计记忆系统时,习惯把它分成三层,这个分层方式借鉴了认知科学里的记忆模型,但在工程上非常好用:
第一层是 working memory(工作记忆)。这就是当前对话轮次的上下文,直接放在 prompt 里,容量有限,通常控制在 2K 到 4K token。它的作用是保证当前对话的连贯性,让 Agent 知道“刚才用户说了什么,我正在回答什么”。
第二层是 episodic memory(情景记忆)。这是跨会话的对话摘要和关键事件记录。比如用户昨天问过“Docker 怎么安装 MySQL”,今天又问“MySQL 的端口怎么改”,Agent 应该能想起来“这个用户昨天在搞 Docker 环境”。情景记忆通常以时间线的方式组织,每条记录包含时间戳、摘要、涉及实体和原始对话的引用。
第三层是 semantic memory(语义记忆)。这是从大量对话中抽取出来的结构化知识,比如用户的偏好、常用工具、项目背景等。语义记忆不依赖于具体的时间点,而是以“事实”的形式存在。比如“用户偏好使用 Python 而不是 JavaScript”、“用户的项目部署在 Ubuntu 22.04 上”。
这三层记忆的读写策略完全不同。工作记忆是每轮都重写的,情景记忆是每轮追加的,语义记忆是定期归纳更新的。热搜词里提到的“agent 存储 working memory”和“llm 的 token 三个点 key 我是谁、query 我在找什么、value 我能提供什么”,其实就是在说记忆的键值设计——key 是身份标识,query 是检索意图,value 是实际内容。
2.3 为什么选 MCP 作为记忆系统的对接协议
MCP(Model Context Protocol)是最近一年在 Agent 圈子里讨论度非常高的一个协议。它的核心思路是把“模型能调用的能力”标准化成一个个 server,每个 server 暴露一组工具(tools)和资源(resources),模型通过统一的协议去调用。
我选择用 MCP 来对接记忆系统,主要基于三个考虑:
第一,解耦。记忆系统的实现可以独立于 Agent 框架。今天你用 LangChain 做 Agent,明天换成自己写的调度逻辑,记忆系统不需要改,只要 MCP 接口不变就行。
第二,可组合。一个 Agent 可能同时需要记忆、需要浏览器操作(比如 Playwright MCP)、需要代码执行,这些能力都通过 MCP 挂载,互不干扰。
第三,可观测。MCP 的调用是有明确日志的,每次记忆的读写都能追踪到,调试的时候非常方便。
热搜词里出现的“playwright mcp”、“chrome devtools mcp”、“browser use mcp 跟 playwright mcp 有什么区别”这些,说明 MCP 生态正在快速扩张。记忆系统作为其中一个 server,和这些工具 server 是平级关系,Agent 根据需要调用即可。
2.4 Docker 化部署:为什么不用裸机跑
记忆系统涉及向量数据库、关系数据库、缓存服务,如果直接在裸机上装,环境依赖能把你逼疯。Docker 的价值在于把整个技术栈打包成可复现的镜像,换一台机器,docker compose up就能跑起来。
热搜词里“docker 安装”、“docker desktop 安装教程”、“windows 安装 docker”、“virtualization support not detected docker desktop failed to start”这些,说明很多人在 Docker 环境准备阶段就卡住了。我会在后面的实操部分详细讲 Windows 和 Ubuntu 两种环境下的安装要点。
3. 核心细节解析:记忆的存储、检索与更新机制
3.1 记忆的存储结构:不只是向量数据库
很多人一提到 Agent 记忆,第一反应就是“上向量数据库”。向量数据库确实重要,但它不是全部。我的记忆系统用了三种存储:
| 存储类型 | 用途 | 选型建议 | 数据量级 |
|---|---|---|---|
| 关系数据库 | 存储原始对话、会话元数据、记忆索引 | PostgreSQL / MySQL | 百万级记录 |
| 向量数据库 | 存储记忆的语义向量,支持相似度检索 | Qdrant / Milvus / Chroma | 十万级向量 |
| 键值缓存 | 存储工作记忆和热点记忆 | Redis | 千级键值对 |
关系数据库是“真相来源”,所有记忆的原始文本都存在这里。向量数据库是“检索加速器”,它存的是记忆的 embedding,用于快速找到语义相关的记忆。Redis 是“热数据层”,当前会话的工作记忆和最近访问过的记忆放在这里,避免每次都查数据库。
热搜词里“docker 安装 redis 主从”、“docker 安装 mysql8.0 并使用”这些,正好对应了这套存储架构的部署需求。Redis 做主从是为了保证缓存层的高可用,MySQL 8.0 则是关系数据库的常见选择。
3.2 记忆的写入:什么时候该记,什么时候不该记
不是所有对话都值得写入长期记忆。我的策略是:
必记的内容:用户明确表达的偏好(“我喜欢用 VS Code”)、用户提供的项目背景(“我在做一个电商后台”)、用户纠正过的错误(“不对,我说的是 PostgreSQL 不是 MySQL”)、任务的关键结论(“最终方案选的是方案 B”)。
不记的内容:寒暄和客套(“你好”、“谢谢”)、重复确认(“好的”、“明白了”)、临时性的中间状态(“让我想想”)。
选择性记录的内容:技术讨论中的知识点、用户提到的工具和框架、用户的问题模式。
实现上,我会在每轮对话结束后,用一个轻量级的 LLM 调用做一次“记忆抽取”,判断这轮对话是否包含值得长期记忆的信息。这个判断的 prompt 大概长这样:
MEMORY_EXTRACTION_PROMPT = """ 分析以下对话轮次,判断是否包含值得长期记忆的信息。 对话内容: 用户:{user_message} 助手:{assistant_message} 如果包含以下类型的信息,请提取并返回 JSON: 1. 用户偏好(preference) 2. 项目背景(context) 3. 关键事实(fact) 4. 错误纠正(correction) 如果不包含值得记忆的信息,返回 {{"should_remember": false}}。 返回格式: {{ "should_remember": true, "memory_type": "preference|context|fact|correction", "content": "提取的记忆内容", "entities": ["涉及的实体"], "importance": 1-10 }} """这个抽取步骤会增加一次 LLM 调用,但它的成本远低于把全部历史塞进上下文的成本。而且抽取出来的记忆是结构化的,后续检索和更新都更方便。
3.3 记忆的检索:key、query、value 的三元组设计
热搜词里有一句很精辟的总结:“llm 的 token 三个点 key 我是谁、query 我在找什么、value 我能提供什么”。这其实就是记忆检索的核心逻辑。
Key(我是谁):这是记忆的归属标识。在多用户或多 Agent 场景下,每条记忆都必须绑定到一个明确的身份上。可以是用户 ID、会话 ID、项目 ID,或者它们的组合。没有 key 的记忆系统,检索时会把别人的记忆也捞出来,那就乱套了。
Query(我在找什么):这是检索的意图表达。用户当前的问题会被转换成一个检索 query,通常是一个 embedding 向量,也可能包含结构化的过滤条件(比如“只要最近 7 天的记忆”、“只要 preference 类型的记忆”)。
Value(我能提供什么):这是检索返回的记忆内容。它不应该是原始对话的全文,而应该是经过摘要和结构化的记忆片段。每条 value 还应该附带元数据:时间戳、重要性评分、来源会话 ID、相关实体等。
检索的流程我一般这样设计:
- 把用户当前输入转换成 embedding
- 在向量数据库中做相似度搜索,取 top-K(通常 K=10 到 20)
- 用结构化条件做二次过滤(时间范围、记忆类型、重要性阈值)
- 对候选记忆做重排序(可以用一个小的 cross-encoder 模型,也可以直接用 LLM 打分)
- 取最终 top-N(通常 N=3 到 5)注入上下文
这里有个经验:相似度分数不是唯一标准。一条记忆可能和当前 query 的语义相似度不高,但它是用户的核心偏好,那就应该被检索出来。所以我在重排序阶段会加入重要性权重和时间衰减因子。
3.4 记忆的更新与遗忘:不是所有记忆都值得永远保留
记忆系统如果只增不减,迟早会变成一个垃圾场。我的做法是:
合并:如果新记忆和已有记忆表达的是同一个事实,就合并它们,更新置信度和时间戳。比如用户先说“我用 Python”,后来说“我主要用 Python 做数据分析”,这两条应该合并成一条更完整的偏好记忆。
衰减:每条记忆有一个“新鲜度”分数,随着时间推移逐渐降低。检索时,新鲜度低的记忆会被降权。但核心偏好类记忆的衰减速度要慢得多,甚至可以设置为不衰减。
淘汰:当记忆总量超过阈值时,淘汰那些重要性低、访问频率低、时间久远的记忆。淘汰不是删除,而是归档到冷存储,万一以后需要还能找回来。
热搜词里“a-memguard: a proactive defense framework for llm-based agent memory”这个方向,其实就是在解决记忆系统的安全问题——防止恶意注入的记忆污染 Agent 的行为。虽然我的项目没有直接实现这个框架,但在设计记忆写入接口时,我加了一层校验:所有写入的记忆都要经过一个“一致性检查”,如果新记忆和已有高置信度记忆矛盾,就标记为待审核,不直接生效。
4. 实操过程:从零搭建一个可运行的 Agent 记忆系统
4.1 环境准备:Docker 安装与常见坑
先说 Docker 的安装。Windows 用户和 Ubuntu 用户的路径不太一样,我分别说一下。
Windows 环境:
Windows 上装 Docker Desktop 是最省事的方案。去官网下载安装包,双击运行,一路下一步。但有几个坑要注意:
第一个坑是虚拟化支持。热搜词里“virtualization support not detected docker desktop failed to start”就是这个问题。Docker Desktop 需要 WSL2 或者 Hyper-V 支持。你需要在 BIOS 里开启虚拟化(Intel VT-x 或 AMD-V),然后在 Windows 功能里启用“虚拟机平台”和“适用于 Linux 的 Windows 子系统”。
第二个坑是 WSL2 的内存占用。默认情况下 WSL2 会占用大量内存,你可以在用户目录下创建一个.wslconfig文件来限制:
[wsl2] memory=8GB processors=4 swap=2GB第三个坑是镜像拉取速度。国内网络环境下,建议配置镜像加速器。在 Docker Desktop 的设置里找到 Docker Engine,添加 registry-mirrors 配置。
Ubuntu 环境:
Ubuntu 上装 Docker 用官方脚本最方便:
# 卸载旧版本 sudo apt-get remove docker docker-engine docker.io containerd runc # 安装依赖 sudo apt-get update sudo apt-get install ca-certificates curl gnupg lsb-release # 添加 Docker 官方 GPG key sudo mkdir -p /etc/apt/keyrings curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg --dearmor -o /etc/apt/keyrings/docker.gpg # 添加仓库 echo "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.gpg] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null # 安装 Docker Engine sudo apt-get update sudo apt-get install docker-ce docker-ce-cli containerd.io docker-compose-plugin # 把当前用户加入 docker 组,避免每次都要 sudo sudo usermod -aG docker $USER newgrp docker装完之后用docker run hello-world验证一下。如果拉取镜像很慢,同样需要配置镜像加速。
4.2 用 Docker Compose 编排记忆系统的全套服务
我的记忆系统需要四个服务:PostgreSQL、Redis、Qdrant(向量数据库)、以及记忆系统本身的应用服务。用 Docker Compose 编排是最干净的方案。
version: '3.8' services: postgres: image: postgres:16-alpine container_name: memory-postgres environment: POSTGRES_USER: memory_user POSTGRES_PASSWORD: memory_pass_2024 POSTGRES_DB: agent_memory ports: - "5432:5432" volumes: - postgres_data:/var/lib/postgresql/data healthcheck: test: ["CMD-SHELL", "pg_isready -U memory_user -d agent_memory"] interval: 10s timeout: 5s retries: 5 redis: image: redis:7-alpine container_name: memory-redis ports: - "6379:6379" volumes: - redis_data:/data command: redis-server --appendonly yes --maxmemory 512mb --maxmemory-policy allkeys-lru qdrant: image: qdrant/qdrant:latest container_name: memory-qdrant ports: - "6333:6333" - "6334:6334" volumes: - qdrant_data:/qdrant/storage memory-service: build: ./memory-service container_name: memory-service ports: - "8080:8080" environment: DATABASE_URL: postgresql://memory_user:memory_pass_2024@postgres:5432/agent_memory REDIS_URL: redis://redis:6379/0 QDRANT_URL: http://qdrant:6333 EMBEDDING_MODEL: text-embedding-3-small depends_on: postgres: condition: service_healthy redis: condition: service_started qdrant: condition: service_started volumes: postgres_data: redis_data: qdrant_data:这个 compose 文件有几个设计点值得说明:
PostgreSQL 加了 healthcheck,确保应用服务在数据库真正就绪后才启动。Redis 配置了allkeys-lru淘汰策略和 512MB 内存上限,防止缓存把内存吃满。Qdrant 暴露了 6333(HTTP)和 6334(gRPC)两个端口。应用服务通过depends_on控制启动顺序。
启动命令很简单:
docker compose up -d查看日志:
docker compose logs -f memory-service4.3 记忆系统的核心代码实现
记忆系统的应用服务我用 Python 写,基于 FastAPI 暴露 HTTP 接口,同时通过 MCP 协议对外提供工具调用能力。
先看数据库表结构:
-- 原始对话表 CREATE TABLE conversations ( id BIGSERIAL PRIMARY KEY, session_id VARCHAR(64) NOT NULL, user_id VARCHAR(64) NOT NULL, role VARCHAR(16) NOT NULL, content TEXT NOT NULL, created_at TIMESTAMP DEFAULT NOW() ); CREATE INDEX idx_conversations_session ON conversations(session_id); CREATE INDEX idx_conversations_user ON conversations(user_id); -- 记忆表 CREATE TABLE memories ( id BIGSERIAL PRIMARY KEY, user_id VARCHAR(64) NOT NULL, memory_type VARCHAR(32) NOT NULL, content TEXT NOT NULL, entities JSONB DEFAULT '[]', importance INTEGER DEFAULT 5, confidence FLOAT DEFAULT 1.0, source_session_id VARCHAR(64), access_count INTEGER DEFAULT 0, last_accessed_at TIMESTAMP, created_at TIMESTAMP DEFAULT NOW(), updated_at TIMESTAMP DEFAULT NOW() ); CREATE INDEX idx_memories_user_type ON memories(user_id, memory_type); CREATE INDEX idx_memories_importance ON memories(importance DESC);记忆的写入逻辑:
import json from datetime import datetime from openai import OpenAI client = OpenAI() def extract_memory(user_message: str, assistant_message: str) -> dict: """从对话轮次中抽取值得记忆的信息""" response = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": MEMORY_EXTRACTION_PROMPT}, {"role": "user", "content": f"用户:{user_message}\n助手:{assistant_message}"} ], response_format={"type": "json_object"}, temperature=0.1 ) return json.loads(response.choices[0].message.content) def write_memory(user_id: str, session_id: str, memory_data: dict): """将抽取的记忆写入数据库和向量库""" if not memory_data.get("should_remember"): return None # 写入 PostgreSQL with get_db_connection() as conn: with conn.cursor() as cur: cur.execute(""" INSERT INTO memories (user_id, memory_type, content, entities, importance, source_session_id) VALUES (%s, %s, %s, %s, %s, %s) RETURNING id """, ( user_id, memory_data["memory_type"], memory_data["content"], json.dumps(memory_data.get("entities", [])), memory_data.get("importance", 5), session_id )) memory_id = cur.fetchone()[0] conn.commit() # 生成 embedding 并写入 Qdrant embedding = get_embedding(memory_data["content"]) qdrant_client.upsert( collection_name="agent_memories", points=[{ "id": memory_id, "vector": embedding, "payload": { "user_id": user_id, "memory_type": memory_data["memory_type"], "content": memory_data["content"], "importance": memory_data.get("importance", 5), "created_at": datetime.now().isoformat() } }] ) return memory_id检索逻辑:
def retrieve_memories(user_id: str, query: str, top_k: int = 5) -> list: """检索与 query 相关的记忆""" query_embedding = get_embedding(query) # 向量检索 search_results = qdrant_client.search( collection_name="agent_memories", query_vector=query_embedding, query_filter={ "must": [ {"key": "user_id", "match": {"value": user_id}} ] }, limit=top_k * 3 # 多取一些用于重排序 ) # 重排序:结合相似度、重要性、时间衰减 now = datetime.now() scored_memories = [] for result in search_results: payload = result.payload similarity = result.score importance = payload.get("importance", 5) / 10.0 created_at = datetime.fromisoformat(payload["created_at"]) days_old = (now - created_at).days time_decay = 1.0 / (1.0 + days_old * 0.05) final_score = similarity * 0.6 + importance * 0.25 + time_decay * 0.15 scored_memories.append((final_score, payload)) scored_memories.sort(key=lambda x: x[0], reverse=True) return [m[1] for m in scored_memories[:top_k]]4.4 通过 MCP 协议暴露记忆能力
MCP 的核心是让模型能够以标准化的方式调用外部工具。我用 Python 的mcp库来实现记忆服务的 MCP server:
from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent app = Server("memory-service") @app.list_tools() async def list_tools() -> list[Tool]: return [ Tool( name="remember", description="将重要信息写入长期记忆。当用户表达了偏好、提供了项目背景、或纠正了之前的错误时调用。", inputSchema={ "type": "object", "properties": { "content": {"type": "string", "description": "要记忆的内容"}, "memory_type": { "type": "string", "enum": ["preference", "context", "fact", "correction"], "description": "记忆类型" }, "importance": { "type": "integer", "minimum": 1, "maximum": 10, "description": "重要性评分" } }, "required": ["content", "memory_type"] } ), Tool( name="recall", description="检索与当前对话相关的长期记忆。在回答用户问题前调用,以获取用户的背景信息和偏好。", inputSchema={ "type": "object", "properties": { "query": {"type": "string", "description": "检索意图"}, "top_k": {"type": "integer", "default": 5} }, "required": ["query"] } ) ] @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: if name == "remember": memory_id = write_memory( user_id=current_user_id, session_id=current_session_id, memory_data={ "should_remember": True, "memory_type": arguments["memory_type"], "content": arguments["content"], "importance": arguments.get("importance", 5) } ) return [TextContent(type="text", text=f"已记忆,ID: {memory_id}")] elif name == "recall": memories = retrieve_memories( user_id=current_user_id, query=arguments["query"], top_k=arguments.get("top_k", 5) ) if not memories: return [TextContent(type="text", text="没有找到相关记忆。")] formatted = "\n".join([ f"[{m['memory_type']}] {m['content']}" for m in memories ]) return [TextContent(type="text", text=f"找到以下相关记忆:\n{formatted}")] async def main(): async with stdio_server() as (read_stream, write_stream): await app.run(read_stream, write_stream, app.create_initialization_options()) if __name__ == "__main__": import asyncio asyncio.run(main())这个 MCP server 暴露了两个工具:remember和recall。Agent 在对话过程中,可以主动调用recall来获取相关记忆,也可以在发现重要信息时调用remember来写入记忆。
4.5 与 Agent 框架的集成
如果你用的是支持 MCP 的 Agent 框架(比如 Claude Desktop、Trae IDE 等),只需要在配置文件里加上这个 MCP server 的启动命令即可。以 Claude Desktop 为例,配置文件大概长这样:
{ "mcpServers": { "memory-service": { "command": "docker", "args": ["exec", "-i", "memory-service", "python", "-m", "mcp_server"], "env": {} } } }如果你是自己写 Agent 调度逻辑,那就需要在每轮对话前调用recall,在对话后调用remember。这个调度逻辑可以封装成一个装饰器:
def with_memory(func): async def wrapper(user_message: str, user_id: str, session_id: str): # 检索相关记忆 memories = retrieve_memories(user_id, user_message, top_k=5) memory_context = "\n".join([m["content"] for m in memories]) # 注入到 prompt enhanced_message = f"""相关背景记忆: {memory_context} 用户当前输入:{user_message}""" # 调用原始处理函数 response = await func(enhanced_message) # 抽取并写入新记忆 memory_data = extract_memory(user_message, response) write_memory(user_id, session_id, memory_data) return response return wrapper5. 常见问题与排查技巧实录
5.1 Docker 网络不通怎么办
这是最高频的问题之一。热搜词里“docker 网络不通”排在前列,说明很多人被坑过。常见原因和排查步骤:
症状一:容器之间无法互相访问。先检查它们是否在同一个 Docker network 里。默认情况下,docker compose创建的服务都在同一个 network 下,可以用服务名互相访问。如果你手动docker run的容器,需要显式指定--network。
症状二:容器内无法访问外网。检查宿主机的 DNS 配置。可以在docker-compose.yml里给服务加上 DNS 配置:
services: memory-service: dns: - 8.8.8.8 - 114.114.114.114症状三:端口映射不生效。检查宿主机端口是否被占用。netstat -tlnp | grep 8080看一下。另外注意,Docker Desktop for Windows 有时候会有端口转发的延迟,重启 Docker Desktop 可以解决。
排查网络问题的万能命令是docker exec -it <container> sh进入容器,然后用ping、curl、nslookup逐个排查。
5.2 记忆检索不准确怎么调
检索不准确通常有三个原因:
Embedding 模型不合适。如果你用的是通用的 embedding 模型,它在技术领域的语义区分度可能不够。可以考虑换一个在技术文本上表现更好的模型,或者用领域数据做微调。
分块策略有问题。一条记忆如果太长,embedding 会丢失细节;如果太短,又缺乏上下文。我的经验是每条记忆控制在 50 到 200 字之间,超过 200 字的拆成多条。
重排序权重不合理。相似度、重要性、时间衰减这三个因子的权重需要根据你的场景调。如果用户偏好类记忆经常被漏掉,就提高重要性的权重;如果最近的信息更重要,就提高时间衰减的权重。
我一般会用一个小的评测集来调这些参数:准备 20 到 30 个 query,每个 query 标注哪些记忆应该被检索出来,然后看不同参数组合下的召回率和准确率。
5.3 记忆冲突怎么处理
当新记忆和旧记忆矛盾时,比如用户先说“我用 MySQL”,后来说“我改用 PostgreSQL 了”,系统需要能识别这种冲突并更新。
我的做法是在写入前做一次冲突检测:
def check_conflict(user_id: str, new_memory: dict) -> dict: """检查新记忆是否与已有记忆冲突""" similar = retrieve_memories(user_id, new_memory["content"], top_k=3) for old in similar: if old["memory_type"] != new_memory["memory_type"]: continue # 用 LLM 判断是否冲突 judgment = client.chat.completions.create( model="gpt-4o-mini", messages=[{ "role": "user", "content": f"""判断以下两条记忆是否冲突: 旧记忆:{old['content']} 新记忆:{new_memory['content']} 返回 JSON:{{"conflict": true/false, "resolution": "keep_old|keep_new|merge", "merged_content": "如果合并,合并后的内容"}}""" }], response_format={"type": "json_object"} ) result = json.loads(judgment.choices[0].message.content) if result["conflict"]: return {"has_conflict": True, "old_memory": old, "resolution": result} return {"has_conflict": False}如果检测到冲突,根据 resolution 决定是保留旧记忆、用新记忆替换、还是合并两者。对于“改用 PostgreSQL”这种情况,正确的处理是用新记忆替换旧记忆,同时把旧记忆标记为“已过时”。
5.4 常见问题速查表
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| Docker Desktop 启动失败 | 虚拟化未开启 | 检查 BIOS 虚拟化设置 | 开启 VT-x/AMD-V,启用 WSL2 |
| 容器间网络不通 | 不在同一 network | docker network ls | 使用 docker compose 统一编排 |
| 记忆检索返回空 | 向量库未索引 | 检查 Qdrant collection | 确认写入时 embedding 已生成 |
| 记忆重复写入 | 缺少去重逻辑 | 查看 memories 表 | 写入前做相似度检查 |
| 检索速度慢 | 向量数量过大 | 监控 Qdrant 查询延迟 | 加索引、限制 top_k、启用缓存 |
| 上下文超长 | 注入记忆过多 | 统计 prompt token 数 | 限制注入记忆条数和长度 |
| Redis 内存溢出 | 未设淘汰策略 | redis-cli info memory | 配置 maxmemory 和 LRU 策略 |
| PostgreSQL 连接超时 | 连接池耗尽 | 查看 pg_stat_activity | 增大连接池、优化慢查询 |
5.5 几个我踩过的坑
坑一:embedding 维度不一致。我一开始用 OpenAI 的 text-embedding-ada-002(1536 维),后来换成 text-embedding-3-small(也是 1536 维),以为无缝切换,结果发现两个模型的向量空间不兼容,检索质量暴跌。换 embedding 模型时,必须重新生成所有向量。
坑二:时间戳时区问题。PostgreSQL 存的是 UTC 时间,但我在 Python 里用datetime.now()生成的是本地时间,导致时间衰减计算错误。统一用datetime.utcnow()或者带时区的datetime.now(timezone.utc)。
坑三:MCP server 的 stdio 阻塞。MCP 的 stdio 传输模式下,如果 server 里有阻塞操作(比如同步的数据库查询),会导致整个 MCP 连接卡死。所有数据库操作都要用异步驱动,或者放到线程池里执行。
坑四:Docker volume 权限问题。PostgreSQL 容器默认以 postgres 用户运行,如果你挂载的宿主机目录权限不对,容器会启动失败。解决办法是在宿主机上把目录 owner 改成 UID 999(PostgreSQL 容器内的用户 ID),或者用 named volume 而不是 bind mount。
6. 记忆系统的扩展方向与个人体会
这套记忆系统跑通之后,我陆续加了一些扩展。一个是记忆的可视化面板,用简单的 Web 页面展示某个用户的所有记忆,支持按类型、时间、重要性筛选,调试的时候非常直观。另一个是记忆的导入导出,支持把某个用户的记忆打包成 JSON,迁移到另一个环境。
还有一个方向是多 Agent 共享记忆。当多个 Agent 协作完成一个任务时,它们需要一个共享的记忆空间来同步状态。我的做法是在记忆表里加一个scope字段,区分“私有记忆”和“共享记忆”,检索时根据 Agent 的角色决定能访问哪些 scope。
热搜词里提到的“rag graphrag llm wiki 本体 rag”这些,其实指向了记忆系统的另一个进化方向:从扁平的向量检索,走向结构化的知识图谱。向量检索擅长找“语义相似”的内容,但不擅长做多跳推理。比如用户问“我上周说的那个项目,用的什么数据库”,向量检索可能找不到,因为这需要先定位到“上周说的项目”,再查这个项目的数据库选型。知识图谱可以解决这类问题,但构建和维护成本高得多。我的建议是先用向量检索跑起来,等确实遇到多跳推理的需求时,再考虑引入图谱结构。
最后分享一个我在调试记忆系统时的小技巧:给每条记忆加一个“来源追溯”字段,记录它是从哪轮对话、哪个 session 抽取出来的。当检索结果不对劲时,你可以顺着这个字段回溯到原始对话,看看是抽取环节出了问题,还是检索环节出了问题。这个字段在排查问题时能省你很多时间。
这套系统目前在我自己的几个 Agent 项目里跑得挺稳,单用户千级记忆量下,检索延迟在 50ms 以内,注入上下文的记忆控制在 500 token 以内,对整体响应速度的影响可以忽略不计。如果你也在做 Agent 记忆相关的开发,希望这些经验能帮你少踩几个坑。