面向 AI agent 或人类操作者:在 Windows 本地部署 mem0 开源版记忆系统
通过 stdio MCP 接入 opencode(或其他 MCP 客户端)。
全程无 Docker、无云端存储;仅抽取 LLM 走外部 API。
本文档来自一次完整实战部署(2026-08),所有配置形状均经源码核验与运行验证。
本篇文章由Ox Alpha生成,注意信息核实。
For 人类
你可以将本文章链接
https://www.imjcj.eu.org/posts/2026-08-26
给AI Agent,让Agent安装。
0. 架构总览
MCP 客户端 (opencode / Claude Desktop / Cursor ...)
│ stdio (JSON-RPC)
▼
FastMCP 封装层 (server.py, 4 个工具)
│
▼
mem0ai (OSS) 记忆引擎
├── 抽取 LLM ──── OpenAI 兼容 API(如 uuapi.io),只走 chat 端点
├── 嵌入模型 ──── FastEmbed 本地 ONNX(首次下载权重后完全离线)
├── 向量库 ──── Chroma 嵌入式(本地目录持久化)
└── 历史库 ──── SQLite(本地文件)
产物:一个 Python 进程 + 一个数据目录。备份 = 复制目录。
工具面:add_memory / search_memories / list_memories / delete_memory
1. 前置检查
| 项 | 要求 | 检查命令 |
|---|---|---|
| Python | 3.11+ | python --version |
| uv(推荐) | 任意近期版本 | uv --version |
| MCP 客户端 | opencode 或其他 | — |
| LLM API | 任一 OpenAI 兼容端点 + key | 见检查点 Q1 |
| 磁盘 | ≥2GB(嵌入模型权重 300MB–2GB) | — |
❓ 检查点 Q1:抽取 LLM 选型
记忆抽取是「高频、短输入、要求严格 JSON 输出」的小任务。选型看四点:
JSON 稳定性、事实抽取精度、中文质量、单次成本。不要用旗舰模型干这活。
按实价估算单次抽取(~2K 输入 + 300 输出)填入下表再决定:
| 候选模型 | 输入价/M | 输出价/M | 单次估算 | 判定方法 |
|---|---|---|---|---|
| (填你的候选) |
经验法则:
- 中档模型(flash 级)通常是质量/价格甜点
- 「mini」档便宜但需实测抽取质量,垃圾记忆入库后清理成本更高
- 模型名做成环境变量,随时可切
❓ 检查点 Q2:嵌入模型选型
必须查当前 fastembed 版本实际支持哪些模型(文档经常过时!):
.venv\Scripts\python.exe -c "from fastembed import TextEmbedding; models=[(m['model'], m['dim']) for m in TextEmbedding.list_supported_models()]; [print(f'{m} ({d})') for m,d in models]"
选型原则:
- 中文场景:优先
jinaai/jina-embeddings-v2-base-zh(768 维中英双语) - 轻量备选:
BAAI/bge-small-zh-v1.5(512 维,小而快) - 注意:热门的
BAAI/bge-m3在 fastembed 0.8.0 中不存在,别照抄网上教程
❓ 检查点 Q3:网络与代理
Windows 上 Python 不读注册表系统代理,直连可能超时。先探测:
# 查系统代理设置
Get-ItemProperty "HKCU:\Software\Microsoft\Windows\CurrentVersion\Internet Settings" | Select ProxyEnable, ProxyServer
# 查代理进程监听端口(区分 socks5 / http)
netstat -ano | findstr "<代理端口>"
# 用 curl 验证协议(socks5 示例)
curl.exe -s --connect-timeout 5 --proxy socks5://127.0.0.1:<端口> https://<你的LLM域名>/ -I
- 若无需代理(可直连 LLM API):跳过,后续配置中删掉代理相关行
- 若代理是 socks5:Python 侧需
httpx[socks](提供 socksio) - 若代理是 http:直接用
http://127.0.0.1:<端口>
2. 安装步骤
Step 1:建项目与虚拟环境
New-Item -ItemType Directory -Force -Path "$env:USERPROFILE\mem0-local\data" | Out-Null
cd "$env:USERPROFILE\mem0-local"
uv venv --python 3.11
uv pip install --python .venv mem0ai fastmcp fastembed chromadb "httpx[socks]"
⚠️ 不要假设
mem0ai[chroma]extra 存在——直接显式装chromadb。
版本参考(2026-08 验证):mem0ai 2.0.19 / fastmcp 3.4.7 / fastembed 0.8.0 / chromadb 1.5.9
Step 2:核验 mem0 配置形状(关键!)
不要照抄任何教程,mem0 大版本间 provider 名称会变(如 qdrant_fastembed → fastembed)。
从安装源码直接核验:
# 嵌入 provider 合法名
Select-String -Path ".venv\Lib\site-packages\mem0\embeddings\configs.py" -Pattern "provider"
# 向量库 provider 合法名 + Chroma 配置字段
Select-String -Path ".venv\Lib\site-packages\mem0\configs\vector_stores\chroma.py" -Pattern "path|collection_name"
# OpenAI 兼容 LLM 配置字段
Select-String -Path ".venv\Lib\site-packages\mem0\configs\llms\openai.py" -Pattern "openai_base_url|model|api_key"
# Memory 类方法签名
.venv\Scripts\python.exe -c "import inspect; from mem0.memory.main import Memory; print(inspect.signature(Memory.add)); print(inspect.signature(Memory.search)); print(inspect.signature(Memory.get_all))"
mem0ai 2.0.19 核验结果(供对照):
- LLM:
{"provider": "openai", "config": {"model", "api_key", "openai_base_url", "temperature"}} - 嵌入:
{"provider": "fastembed", "config": {"model"}}(维度自动探测) - 向量库:
{"provider": "chroma", "config": {"path", "collection_name"}} - API:
add(messages, user_id=, metadata=, infer=True)、
search(query, filters={"user_id":...}, top_k=, threshold=)、
get_all(filters=, top_k=)、delete(memory_id) - 返回统一为
{"results": [...]}
Step 3:连通性探测(先于写代码)
用 Python 脚本(别用 PowerShell 拼 JSON,引号地狱)验证:模型名、key、代理三者皆通:
# probe.py —— 逐个模型试,确认真实模型 ID 与所属 key 组
import json, os
from pathlib import Path
import httpx
auth = json.loads((Path.home() / ".local/share/opencode/auth.json").read_text("utf-8"))
transport = httpx.HTTPTransport(proxy="socks5://127.0.0.1:10808") # 无代理则删
with httpx.Client(transport=transport, timeout=15) as c:
for provider, model in [("provider1", "model1"), ("provider2", "model2")]:
r = c.post("https://provider/chat/completions",
headers={"Authorization": f"Bearer {auth[provider]['key']}"},
json={"model": model, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 10})
print(provider, model, "->", r.status_code, r.text[:80])
❓ 检查点 Q4:探测结果确认
实战中踩过的坑,逐一对照:
- 502 Bad Gateway ≠ 模型名错,可能是上游临时挂了;换同组其他模型对照测试
- model_not_found:API 中转商常按模型组分 key
- 请求挂死无输出:九成是代理问题,回检查点 Q3
Step 4:写 MCP 封装 server.py
完整可用版本(按需改环境变量默认值):
"""mem0-local: local memory MCP server."""
import json, os, sys
from pathlib import Path
os.environ.setdefault("MEM0_TELEMETRY", "False") # mem0 遥测关
os.environ.setdefault("ANONYMIZED_TELEMETRY", "False") # chroma 遥测关
os.environ.setdefault("HF_HUB_OFFLINE", "1") # 模型已缓存,跳过 HF 联网检查
os.environ.setdefault("TRANSFORMERS_OFFLINE", "1")
# 代理(无代理环境删除以下三行)
os.environ.setdefault("ALL_PROXY", "socks5://127.0.0.1:10808")
os.environ.setdefault("HTTPS_PROXY", "socks5://127.0.0.1:10808")
os.environ.setdefault("HTTP_PROXY", "socks5://127.0.0.1:10808")
HOME = Path.home()
DATA_DIR = HOME / "mem0-local" / "data"
AUTH_JSON = HOME / ".local" / "share" / "opencode" / "auth.json" # 按实际 key 存放处改
LLM_MODEL = os.environ.get("MEM0_LLM_MODEL", "gemini-3.7-flash")
LLM_BASE_URL = os.environ.get("MEM0_LLM_BASE_URL", "https://uuapi.io/v1")
EMBED_MODEL = os.environ.get("MEM0_EMBED_MODEL", "jinaai/jina-embeddings-v2-base-zh")
USER_ID = os.environ.get("MEM0_USER_ID", "june")
def resolve_api_key(model: str) -> str:
key = os.environ.get("MEM0_API_KEY")
if key:
return key
auth = json.loads(AUTH_JSON.read_text(encoding="utf-8"))
# 按模型前缀选 key 组(按你的中转商实际结构改)
if model.startswith("gemini"):
return auth.get("uugemini", {}).get("key") or auth.get("uugpt", {}).get("key")
return auth.get("uugpt", {}).get("key")
_memory = None
def _pin_fastembed_cache():
"""把嵌入权重固定到稳定目录(Temp 会被清理)。"""
import fastembed
target = DATA_DIR / "models"
target.mkdir(parents=True, exist_ok=True)
orig = fastembed.TextEmbedding.__init__
def patched(self, *a, **kw):
kw.setdefault("cache_dir", str(target))
orig(self, *a, **kw)
fastembed.TextEmbedding.__init__ = patched
def get_memory():
global _memory
if _memory is None:
from mem0 import Memory
_pin_fastembed_cache()
_memory = Memory.from_config({
"llm": {"provider": "openai", "config": {
"model": LLM_MODEL, "api_key": resolve_api_key(LLM_MODEL),
"openai_base_url": LLM_BASE_URL, "temperature": 0.1}},
"embedder": {"provider": "fastembed", "config": {"model": EMBED_MODEL}},
"vector_store": {"provider": "chroma", "config": {
"path": str(DATA_DIR / "chroma"), "collection_name": "mem0"}},
"history_db_path": str(DATA_DIR / "history.db"),
})
return _memory
from fastmcp import FastMCP
mcp = FastMCP("mem0-local")
@mcp.tool
def add_memory(text: str, tags: list[str] | None = None) -> str:
"""Store a durable fact, decision, or preference. The LLM extracts and
deduplicates facts automatically."""
metadata = {"tags": tags} if tags else None
result = get_memory().add(text, user_id=USER_ID, metadata=metadata, infer=True)
events = result.get("results", [])
if not events:
return "No new memory stored (nothing extractable or duplicate)."
return "\n".join(f"{e.get('event')}: {e.get('memory')}" for e in events)
@mcp.tool
def search_memories(query: str, limit: int = 5) -> str:
"""Semantic search over stored memories."""
result = get_memory().search(query, filters={"user_id": USER_ID}, top_k=limit, threshold=0.1)
items = result.get("results", [])
if not items:
return "No matching memories."
return "\n".join(f"- [{m.get('score', 0):.2f}] {m.get('memory')}" for m in items)
@mcp.tool
def list_memories(limit: int = 50) -> str:
"""List recently stored memories."""
result = get_memory().get_all(filters={"user_id": USER_ID}, top_k=limit)
items = result.get("results", [])
if not items:
return "Memory store is empty."
return "\n".join(f"- {m.get('memory')} (id: {m.get('id')})" for m in items)
@mcp.tool
def delete_memory(memory_id: str) -> str:
"""Delete one memory by its id."""
get_memory().delete(memory_id)
return f"Deleted {memory_id}."
if __name__ == "__main__":
import threading
def _warmup():
try:
get_memory()
print("[mem0-local] warmup complete", file=sys.stderr)
except Exception as e:
print(f"[mem0-local] warmup failed: {e}", file=sys.stderr)
threading.Thread(target=_warmup, daemon=True).start()
mcp.run()
设计要点(每条都对应一个真实踩坑):
- 遥测全关 +
HF_HUB_OFFLINE=1:消除首次调用的隐性网络延迟 _pin_fastembed_cache:权重固定到data/models,防 Temp 清理后重新下载 600MB 导致超时- 后台预热线程:服务器启动即加载引擎,工具调用到达时已热
- 所有日志走 stderr:stdio MCP 的 stdout 是协议通道,任何 print 污染都会导致客户端挂死
- key 从 opencode auth.json 读取:不复制秘密,换 key 自动同步
Step 5:冒烟测试(MCP 接入前)
# smoke_test.py
from server import get_memory, list_memories, search_memories
mem = get_memory() # 首次会下载嵌入权重,耐心等
mem.add("用户偏好在 Windows 机器上使用 PowerShell 5.1,且倾向轻量化自托管本地方案",
user_id="june", metadata={"tags": ["preference"]}, infer=True)
print(search_memories("Windows 终端")) # 应命中刚存的记忆
print(list_memories())
chcp 65001; .venv\Scripts\python.exe smoke_test.py
中文乱码是 PowerShell 代码页问题(cp936),
chcp 65001+ UTF-8 读取即可,
库内存储是 UTF-8,无需担心。
Step 6:原始 stdio 协议验证(可选但推荐)
模拟 MCP 客户端直接对话 server.py,确认协议层健康:
# test_raw_stdio.py —— 核心:spawn 子进程,手写 JSON-RPC
import subprocess, json, time
proc = subprocess.Popen([r".venv\Scripts\python.exe", r"server.py"],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
time.sleep(3)
proc.stdin.write((json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize",
"params": {"protocolVersion": "2024-11-05", "capabilities": {},
"clientInfo": {"name": "t", "version": "0"}}}) + "\n").encode())
proc.stdin.flush()
time.sleep(2)
proc.stdin.write((json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/list"}) + "\n").encode())
proc.stdin.flush()
time.sleep(2)
proc.stdin.write((json.dumps({"jsonrpc": "2.0", "id": 3, "method": "tools/call",
"params": {"name": "list_memories", "arguments": {"limit": 5}}}) + "\n").encode())
proc.stdin.flush()
time.sleep(10)
proc.kill()
预期:三条请求各收到一条 {"jsonrpc":"2.0","id":N,"result":...} 响应。
3. 接入 MCP 客户端
opencode(全局配置 ~/.config/opencode/opencode.jsonc)
{
"mcp": {
"mem0": {
"type": "local",
"command": [
"C:\\<你的路径>\\mem0-local\\.venv\\Scripts\\python.exe",
"C:\\<你的路径>\\mem0-local\\server.py"
],
"enabled": true
}
},
"experimental": {
"mcp_timeout": 120000
}
}
command必须是字符串数组;type必填mcp_timeout调到 120s:给冷启动兜底(预热正常后实际调用 <1s)
其他客户端(通用 MCP stdio 形态)
{ "mcpServers": { "mem0": { "command": "<python.exe 绝对路径>", "args": ["<server.py 绝对路径>"] } } }
配套 agent 引导(技能/规则文件)
在客户端的技能系统(如 opencode 的 skills/memory/SKILL.md)中声明:
- 何时调用
search_memories(会话开始、涉及历史决策时) - 何时调用
add_memory(技术决策落定、用户偏好确认、环境约束发现时) - 硬纪律:禁止绕过工具直查数据库文件、禁止 import server.py;
工具未挂载时提示用户完全重启客户端,而不是手搓查询
4. 验证与重启
❓ 检查点 Q5:完全重启
MCP 配置只在客户端进程启动时加载一次。 新建会话/标签页不会重新挂载;
进程内 MCP 一旦被标记死亡(如超时),后续会话都拿不到工具。
正确操作:完全退出客户端进程 → 重新启动 → 新会话中调用 list_memories 验证。
5. 故障排查手册(实战踩坑全集)
| 症状 | 根因 | 修复 |
|---|---|---|
MCP 首次调用 -32001 Request timed out | 冷启动初始化(mem0 导入+模型加载+HF 联网检查)超过客户端超时 | HF_HUB_OFFLINE=1 + 后台预热 + mcp_timeout: 120000 |
后续调用 -32000 Connection closed | 超时后进程被客户端断开/标记死亡 | 修复根因后完全重启客户端进程 |
| LLM 报 502 | 中转商上游抖动,或模型名在该 key 组不可用 | 换同组模型对照探测(见 probe.py) |
model_not_found | 中转商按模型组分 key | 用对应组的 key(如 gemini 组 key 调 gemini) |
| Python 请求挂死但 curl 正常 | Windows 系统代理是注册表级,Python 不读;或代理是 socks5 而按 http 用 | netstat 确认端口协议;装 httpx[socks];显式设 ALL_PROXY |
| 嵌入模型下载慢/失败 | HF 直连受限 | 走代理下载一次,之后 HF_HUB_OFFLINE=1 完全离线 |
| 客户端收到工具响应但乱码 | 终端代码页问题(cp936) | chcp 65001;库内是 UTF-8,数据无损 |
| 工具列表为空/未挂载 | 配置未加载(没重启进程)或 server.py 启动即崩 | 先跑 Step 6 原始协议测试定位 |
| agent 绕过工具手搓查询 | 技能文件缺少硬纪律 | 补”禁止直查”规则(见 Step 3 配套引导) |
| PowerShell 传 JSON 给 curl 报错 | PowerShell 引号转义地狱 | 一律改用 Python 脚本发请求 |
6. 日常维护
| 操作 | 方法 |
|---|---|
| 备份 | 复制 mem0-local\data\ 整目录(含 chroma + history.db + models) |
| 换抽取模型 | 设环境变量 MEM0_LLM_MODEL(重启客户端生效) |
| 换嵌入模型 | 设 MEM0_EMBED_MODEL;⚠️ 换模型后维度可能变化,旧向量不兼容,需清空 data\chroma 重建 |
| 删全部记忆 | 删除 data\chroma 与 data\history.db |
| 升级 mem0ai | 升级后重跑 Step 2 核验配置形状,provider 名可能变 |
7. 部署完成清单
- Python 3.11+ 与 uv 可用
- Q1 抽取模型已选定且 probe 探测 200 OK
- Q2 嵌入模型在当前 fastembed 版本支持列表中
- Q3 代理协议与端口已确认(或确认无需代理)
- server.py 就位,冒烟测试存取闭环通过
- 原始 stdio 协议测试通过(initialize / tools/list / tools/call)
- 客户端配置写入,
mcp_timeout已调大 - 技能/引导文件就位(含禁止绕过工具的硬纪律)
- 完全重启客户端,新会话验证工具可用
- 跨会话验证:本会话存一条 → 新会话搜到