AgentScope 集成 mem0 长期记忆中间件：跨会话记忆的三种控制模式与底层适配原理

AgentScope 集成 mem0 长期记忆中间件：跨会话记忆的三种控制模式与底层适配原理

【免费下载链接】agentscopeBuild and run agents you can see, understand and trust.项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope

Mem0Middleware 是 AgentScope 提供的长期记忆中间件，它把开源 mem0 与对应源码，完整讲解其安装、三种构造路径、static_control / agent_control / both 三种控制模式、跨 Agent 共享与记忆隔离（scoping），并深入_agentscope_adapter.py揭示"用 AgentScope 自己的大模型驱动 mem0 抽取与向量化"的底层实现，最终带读者读懂可运行的 oss_demo.py。

为什么需要 mem0 中间件：Agent 会话级记忆的缺口

普通 Agent 每次会话的上下文是临时的：会话结束后，模型既没有"记住"用户说过的偏好，也无法在下一次会话中主动回忆。AgentScope 的中间件机制（MiddlewareBase）恰好提供了在on_reply前后、系统提示词构建时、工具列表中挂钩的能力，mem0 则负责记忆的持久化存储、抽取与向量检索。二者结合后：

• 用户在第 1 轮告诉 Agent"做图表默认用暗色模式、matplotlib，我住在杭州"；
• 第 2 轮开启一个全新的 Agent 实例（空上下文）直接问"给我画月度销售额柱状图"；
• Agent 无需任何提示就能自动检索到上轮记忆，自主选择暗色主题与 matplotlib。

这正是 oss_demo.py 要演示的核心效果。整个链路不引入独立于 AgentScope 之外的第二套模型客户端：记忆抽取（LLM）与向量化（Embedding）全部复用 AgentScope 已有的模型（示例中使用 DashScope 的通义千问与 text-embedding），因此不需要为 mem0 单独配置 OpenAI key。

安装与依赖

mem0 是 AgentScope 的可选依赖，通过 extra 安装即可：

# 等价于 pip install agentscope mem0ai>=2.0.0,<3.0.0 pip install "agentscope[memory-mem0]"

依赖约束可以在 pyproject.toml 中核实：memory-mem0 = ["mem0ai>=2.0.0"]，且该 extra 同时被聚合进mem0组合 extra。

运行 OSS 路径的 demo 需要配置 DashScope 密钥；若切换到 hosted mem0 Platform 则需平台密钥：

export DASHSCOPE_API_KEY=sk-... # OSS 路径 # Platform 路径（仅在切换时）： # export MEM0_API_KEY=m0-... # export OPENAI_API_KEY=sk-... # 仅当你的 Agent 聊天模型是 OpenAI 时才需要

导入路径

Mem0Middleware从 middleware 包顶层导出（见 middleware/init.py），同时需要Toolkit来挂载记忆工具：

from agentscope.middleware import Mem0Middleware from agentscope.tool import Toolkit

三种构造路径与参数优先级

Mem0Middleware的构造签名（见 _middleware.py）为：

Mem0Middleware( *, user_id: str, # 必填，记忆命名空间 client=None, # 预构建的 mem0 异步客户端 chat_model=None, # AgentScope ChatModelBase embedding_model=None, # AgentScope EmbeddingModelBase mem0_config=None, # mem0 MemoryConfig（自定义向量库/历史库/reranker） mode="both", # static_control / agent_control / both agent_id=None, # 默认取 agent.name top_k=5, # 每次检索的最大条数 threshold=None, # 最低相似度阈值，None 交给 mem0 scope_search_by_agent=True, # 检索时是否按 agent_id 过滤 await_write=True, # 写回是否同步等待 memory_section_header=..., # 注入记忆的标题文案 memory_section_intro=..., # 注入记忆的引导文案 tool_instructions=..., # agent_control/both 模式的系统提示增强 )

README 归纳的三种合法构造方式如下：

# 1. Models —— 构建本地 OSS AsyncMemory，LLM/Embedding 走 AgentScope 模型 Mem0Middleware( user_id="alice", chat_model=my_chat_model, embedding_model=my_embedding_model, mode="both", ) # 2. Models + 自定义 mem0_config —— 保留自定义向量库/历史库/reranker， # 仅覆盖 .llm 与 .embedder 两个槽位为 AgentScope 适配器 Mem0Middleware( user_id="alice", chat_model=my_chat_model, embedding_model=my_embedding_model, mem0_config=MemoryConfig( vector_store=VectorStoreConfig( provider="qdrant", config={"host": "my-qdrant", "port": 6333}, ), history_db_path="/data/mem0_history.db", ), mode="both", ) # 3. Client —— 自带预构建客户端，完全掌控 mem0 装配 # OSS 后端： Mem0Middleware(user_id="alice", client=AsyncMemory(), mode="both") # 托管 Platform 后端： Mem0Middleware(user_id="alice", client=AsyncMemoryClient(api_key="m0-..."), mode="both")

优先级与校验矩阵

_resolve_client（_middleware.py）实现了严格的后端解析逻辑，README 给出了完整矩阵：

两个设计动机值得注意：

• client绝对优先：同一个Mem0Middleware(...)调用既能服务库内用户（传 AgentScope 模型），也能服务生产部署（传预构建client）。被忽略的 kwargs 通过WARNING日志显式暴露，不会静默丢失。
• config 覆盖机制：可以维护一份规范的MemoryConfig模板（自定义向量库、历史库、reranker 等），在每个调用点只通过chat_model/embedding_model替换 LLM 与 Embedder，实现"模板复用、按点切换"。

此外_resolve_client还会校验客户端必须是异步的：mem0.AsyncMemory（OSS）或mem0.AsyncMemoryClient（Platform）均可，同步版Memory/MemoryClient会在构造期直接抛出TypeError。检测逻辑_looks_async使用inspect.unwrap剥开 mem0 Platform 客户端上@api_error_handler这类同步functools.wraps包装，避免把"被同步装饰器包住的 async 方法"误判为同步（该行为有专门的单测覆盖，见 mem0_middleware_test.py）。

三种控制模式：由谁来决定"何时读写记忆"

mode参数决定 Agent 与 mem0 的交互方式，核心区别在于模型能看到什么与什么会自动触发：

static_control：中间件全权代劳，Agent 无感知

该模式复刻了 AgentScope 1.x 中ReActAgent._retrieve_from_long_term_memory的行为，流程（见on_reply，_middleware.py）为：

1. on_reply（前）：用最新用户消息查询 mem0，预取检索结果；
2. ReplyStartEvent时机注入：该事件在 Agent 将新用户输入写入state.context之后、推理循环开始之前触发。中间件此时把一个AssistantMsg(name="memory", ...)追加到state.context，使记忆注记紧跟在用户新消息之后（与 v1 在self.memory.add(msg)之后的插入位置一致，测试 test_memory_message_lands_after_user_message 专门校验了这一顺序）；
3. on_reply（后）：把本轮(user, assistant)对话写回 mem0。

注入的记忆消息会持久保留在上下文中：长会话每轮检索到内容就会累积一条。如果担心 token 膨胀，可用compress_context后处理，或自写中间件将其弹出。

注：_extract_query_text（_utils.py）会跳过ExternalExecutionResultEvent、UserConfirmResultEvent这类 HITL 恢复事件——恢复轮不触发检索也不触发写回。

agent_control：Agent 自主决定，工具驱动

中间件只暴露两个工具——search_memory(keywords, limit)与add_memory(thinking, content)，其余完全旁观：无自动检索、无自动写回。构造 Agent 时需要显式把工具挂进 toolkit（中间件自身不会修改 toolkit，测试 test_middleware_does_not_mutate_toolkit 印证了这一点）：

mw = Mem0Middleware(..., mode="agent_control") agent = Agent( ..., toolkit=Toolkit(tools=await mw.list_tools()), middlewares=[mw], )

系统提示词会追加一小段引导（DEFAULT_TOOL_INSTRUCTIONS，通过on_system_prompt注入，_middleware.py），提示记忆工具存在；每个工具的具体用法由标准 tool schema 承载。两个工具的 schema 与实现见 _tools.py：

• search_memory(keywords: list[str], limit: int = 5)：多关键词并行检索，结果合并去重后返回；失败时返回state=ERROR的ToolChunk，由 toolkit 聚合为失败的工具调用。
• add_memory(thinking: str, content: list[str])：content逐条作为独立完整句写入 mem0；thinking（Agent 的记忆理由）不会进入 mem0——它只出现在工具返回文本中用于审计，避免 mem0 的记忆被 Agent 的自我叙述污染（测试 test_add_memory_does_not_persist_thinking）。

写路径采用两层兜底策略_async_add_with_fallback（_middleware.py）：先让 mem0 的抽取 LLM 正常抽取；若抽取结果为空，则改用infer=False把原文直接入库，保证add_memory调用永远"至少存下点什么"。注释详细解释了为何从 v1 的三层降为两层——mem0 v2.x 按 filters 而非消息角色选择抽取提示词，v1 的"切换 assistant 角色重试"已退化为一次无意义的 LLM 调用（测试 test_add_memory_two_tier_fallback 验证了恰好两次add调用且均使用 user 角色）。

这两个工具继承自_Mem0MemoryToolBase，check_permissions返回PermissionBehavior.ALLOW自动放行——记忆工具属于 Agent 的标准能力，每次调用都弹权限确认反而失去意义（测试 test_tools_auto_allow_permission）。

both（默认）：双轨并行

静态检索注入与按需工具同时生效：记忆被自动检索为上下文中的 assistant 注记，同时search_memory/add_memory工具（含系统提示引导）对 Agent 开放。这与 AgentScope 1.x 中ReActAgent.long_term_memory_mode的默认值一致，也是 oss_demo.py 采用的模式（MODE = "both"）。

跨 Agent 共享中间件与 Qdrant 独占锁

本地 OSS mem0 后端默认使用磁盘上的 Qdrant，而 Qdrant 对存储目录（默认/tmp/qdrant）持有独占锁。若两个Mem0Middleware各自从chat_model+embedding_model构建，会各自创建自己的AsyncMemory，第二个实例将崩溃：

RuntimeError: Storage folder /tmp/qdrant is already accessed by another instance of Qdrant client.

解决方案是构建一个Mem0Middleware实例并传给所有需要共享同一记忆命名空间的 Agent：

mw = Mem0Middleware( user_id="alice", chat_model=chat_model, embedding_model=embedding_model, mode="both", ) agent_a = Agent(..., toolkit=Toolkit(tools=await mw.list_tools()), middlewares=[mw]) agent_b = Agent(..., toolkit=Toolkit(tools=await mw.list_tools()), middlewares=[mw])

这正是 demo 的做法。安全性的依据在于：记忆工具在调用时拿到的是实时的AgentState，中间件通过state.session_id解析当前活跃 Agent，因此单实例跨多 Agent 共享是安全的。若确实需要每个 Agent 独立的 Qdrant 存储，则为每个实例传入带不同vector_store.config.path或collection_name的mem0_config。

推荐：用 Docker 运行 Qdrant（Windows 上尤其必要）

本地磁盘 Qdrant 对单进程 demo 够用，但真实部署中很脆弱——Windows 上尤其痛苦：文件系统锁语义与 Unix 不同，独占锁故障更难恢复。任何超出单进程 Linux/macOS 沙箱的场景，都建议把 Qdrant 作为服务运行：

docker run -p 6333:6333 -p 6334:6334 \ -v $(pwd)/qdrant_storage:/qdrant/storage \ qdrant/qdrant

然后把 mem0 指向网络地址而非本地路径：

from mem0.configs.base import MemoryConfig from mem0.vector_stores.configs import VectorStoreConfig mem0_cfg = MemoryConfig( vector_store=VectorStoreConfig( provider="qdrant", config={ "collection_name": "mem0", "host": "localhost", # Docker 容器 "port": 6333, "embedding_model_dims": 1536, }, ), ) Mem0Middleware( user_id="alice", chat_model=chat_model, embedding_model=embedding_model, mem0_config=mem0_cfg, )

相对磁盘模式的优势：

• 无文件锁争用——多个 Python 进程可同时连接；
• 状态跨运行存活，无需手动清理文件；
• 同一套配置可直接迁移到远程 Qdrant（Qdrant Cloud、自建 Kubernetes 部署），只需更换host/port/api_key。

记忆作用域：user_id×agent_id

mem0 在add时给每条记忆打上user_id与agent_id标签，search时按这些标签做 AND 匹配。中间件通过scope_search_by_agent标志（默认True）控制 Agent 维度是否参与检索过滤：

agent_id默认取agent.name，可通过构造参数agent_id="..."或agent_id=lambda agent: ...覆盖（_resolve_client与_async_search中search_agent_id的取值逻辑见 _middleware.py）。

适合放宽scope_search_by_agent的场景：

• 一个用户拥有多个专业化 Agent（研究 / 编码 / 日程），希望彼此受益于对用户的新发现；
• Agent 的name可能随部署变更，但希望记忆跨名称变更持久存在。

关于 agent-centric 抽取（当前不可达）

mem0 v2 的抽取提示词ADDITIVE_EXTRACTION_PROMPT含一个条件后缀，可将框架从以用户为中心（"User stated X"）切换为以 Agent 为中心（"Agent was informed of X" / "Agent recommended Y"）。该后缀由is_agent_scoped = bool(filters.agent_id) and not filters.user_id门控——即仅当提供agent_id而不提供user_id时生效。由于Mem0Middleware的user_id是必填参数、总是传入，因此该 agent-centric 后缀在当前中间件中永远不可达。实践中这通常没有影响——Agent 的人格与配置通常由系统提示词表达，而非长期记忆。

服务模式集成：接入agentscope.app

上述 demo 是库模式——自行构造Agent并把Mem0Middleware放进middlewares=[...]。对于通过agentscope.app（FastAPI 服务层）的生产部署，user_id已由框架从X-User-IDHTTP 头流入，只需通过extra_agent_middlewares工厂挂钩（该参数类型定义于 app/_types.py）：

from agentscope.app import create_app from agentscope.middleware import Mem0Middleware from agentscope.middleware._longterm_memory._mem0._agentscope_adapter \ import build_mem0_config from mem0 import AsyncMemory # 在模块级只构建一次 mem0 client —— 本地 OSS Qdrant 对存储目录 # 持独占锁，若按请求构建会在并发流量下死锁。 chat_model = ... # 共享的 AgentScope ChatModelBase emb_model = ... # 共享的 AgentScope EmbeddingModelBase mem0_client = AsyncMemory( config=build_mem0_config( chat_model=chat_model, embedding_model=emb_model, ), ) async def long_term_memory_factory( user_id: str, # ← 来自认证的 X-User-ID 头 agent_id: str, session_id: str, ) -> list: return [ Mem0Middleware( user_id=user_id, client=mem0_client, # 跨请求共享 mode="both", ), ] app = create_app( ..., extra_agent_middlewares=long_term_memory_factory, )

要点：

• 工厂签名是async (user_id, agent_id, session_id) -> list[MiddlewareBase]，每次组装 Agent 时调用一次（即每轮聊天 / 每次定时触发）。每次返回全新的Mem0Middleware实例，但底层共享同一个 mem0 client。
• user_id是已认证的调用方，由agentscope.app通过get_current_user_id注入（当前取自X-User-ID头；未来鉴权落地后将改为 JWT）。直接转发给Mem0Middleware(user_id=user_id, ...)即可，无需 resolver 回调。
• 若使用托管 mem0 Platform，把AsyncMemory(config=...)换成AsyncMemoryClient(api_key=...)即可——工厂形态完全一致，也没有 Qdrant 锁问题。

底层原理：AgentScope 作为 mem0 的后端

当传入chat_model+embedding_model时，中间件内部通过build_mem0_config（_agentscope_adapter.py）完成四步装配：

1. 注册 provider：在 mem0 的工厂字典LlmFactory.provider_to_class/EmbedderFactory.provider_to_class中以 provider 名"agentscope"注册AgentScopeLLM/AgentScopeEmbedding；
2. 绕过白名单校验：mem0 的LlmConfig.validate_config/EmbedderConfig.validate_config硬编码了 provider 白名单（不含agentscope）。适配器动态构造只放行"agentscope"的LlmConfig/EmbedderConfig子类完成替换——其他 provider 依旧被拒绝（单测 test_naive_from_config_path_still_rejected 验证了不经过此助手直接构造MemoryConfig(provider="agentscope")会触发 pydanticValidationError）；
3. 构建 AsyncMemory：其.llm与.embedding_model全部路由到 AgentScope 适配器；
4. 同步/异步桥接：mem0 的LLMBase.generate_response/EmbeddingBase.embed是同步接口，AgentScope 模型是异步的。_AsyncBridge（_agentscope_adapter.py）在独立守护线程上长期运行一个事件循环，通过run_coroutine_threadsafe提交协程并阻塞等待结果——事件循环存活于 bridge 生命周期内，因此模型内部的异步客户端（如 Ollama 的AsyncClient）的连接池可以跨调用复用，而不是每调一次就被关闭。

AgentScopeLLM.generate_response会把 mem0 的 OpenAI 风格消息字典转换为 AgentScope 的Msg对象（system/user/assistant 三种角色，未知角色静默丢弃），支持流式与非流式模型（流式响应被排空、取末块，符合 AgentScope 流式契约），并把ChatResponse拍平回 mem0 期望的str或含tool_calls的dict（ToolCallBlock.input是 JSON 字符串，会解析回 dict；解析失败则保留原文）。AgentScopeEmbedding.embed则把文本列表送入 AgentScope embedding 模型并返回第一个向量。上述转换均有对应单测（mem0_agentscope_adapter_test.py）。

维度一致性要求：embedding 模型的dimensions必须与向量库期望的维度一致——mem0 默认 Qdrant 期望 1536，正好匹配 DashScopetext-embedding-v2的dimensions=1536（也是 oss_demo.py 中使用的值，demo 用的是text-embedding-v4）。

端到端 demo 解读：oss_demo.py

oss_demo.py 是可直接运行的单文件演示，要点如下：

• 模式切换：顶部MODE = "both"可改为"static_control"或"agent_control"观察差异；
• 干净起点：每次运行先删除/tmp/qdrant与~/.mem0/history.db保证可复现（刻意不用mem0_client.delete_all()，因为 qdrant-client 的本地 SQLite 层在 mem0 并行asyncio.gather删除下存在竞态）；
• 两个独立的聊天模型实例：Agent 与 mem0 各用一个DashScopeChatModel（qwen3.7-max）。原因是 Agent 在应用事件循环上调用模型，而 mem0 适配器在其专属 bridge 事件循环上调用模型；异步 HTTP 客户端及其连接池绑定事件循环，共享同一实例可能触发 "bound to a different event loop"；
• 显式向量库配置：MemoryConfig(vector_store=VectorStoreConfig(provider="qdrant", config={collection_name, path="/tmp/qdrant", embedding_model_dims=1536, on_disk=False}))——这里只是把 mem0 默认的本地 Qdrant 显式写出，便于按注释替换为 Docker/远程 Qdrant；
• 两个会话：SESSION 1 告诉 Agent 默认暗色 + matplotlib + 家在杭州；SESSION 2 新建空上下文的 Agent 索要柱状图。每轮通过agent.reply_stream事件流打印各中间件贡献，行首的static/agent标签标明控制路径：
  • [mem0 → context (static)]——ReplyStartEvent时从state.context提取的记忆注记子弹列表；
  • [tool call (agent)]—— Agent 自主发起的search_memory/add_memory调用；
  • [assistant]—— 由TextBlockDeltaEvent增量拼接的最终回复；
  • [context → mem0 (static)]—— 回合结束后静态写回抽取出的新事实（用mw._client.get_all(filters={"user_id": ...})前后对比得出）。

托管 Platform 切换只需把Mem0Middleware(...)构造替换为client=AsyncMemoryClient(api_key=os.environ["MEM0_API_KEY"])（oss_demo.py内# For the hosted mem0 Platform, swap …注释处给出了完整示例），其余逻辑完全一致，且无需本地 Qdrant / 向量库配置。

小结

• 三种构造路径（models / models + mem0_config / client）配合优先级矩阵，兼顾库内易用与生产可控；
• 三种控制模式覆盖"无感知自动记忆（static_control）"、"Agent 自主决策（agent_control）"与"双轨并行（both）"；
• 单个中间件实例跨 Agent 共享安全，Qdrant 独占锁问题用 Docker 服务化解决；
• scope_search_by_agent精确控制"按 Agent 隔离"还是"按用户共享"；
• 服务模式通过extra_agent_middlewares工厂接入，user_id由X-User-ID头自动注入；
• 底层用AgentScopeLLM/AgentScopeEmbedding+ 常驻 bridge 事件循环，让 mem0 复用 AgentScope 既有模型，无需为记忆功能另配模型密钥。

需要深入验证行为时，可参考 mem0_middleware_test.py（构造校验、三模式行为、工具语义、兜底写路径）与 mem0_agentscope_adapter_test.py（消息转换、响应拍平、provider 注册、config 覆盖）两个测试文件中的完整断言。

【免费下载链接】agentscopeBuild and run agents you can see, understand and trust.项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope