206 lines
7.3 KiB
Python
206 lines
7.3 KiB
Python
"""Memory V2 - 记忆 Provider 抽象基类。
|
|
|
|
设计动机
|
|
========
|
|
Hermes Agent 的可插拔记忆系统的 deerflow 版本。所有具体的记忆后端
|
|
(builtin 本地文件、Hindsight HTTP 等)都实现这个接口,由 ``MemoryManager``
|
|
统一编排。Provider 之间互不感知,可自由组合 (内建 + 至多 1 个外部 plugin)。
|
|
|
|
生命周期
|
|
========
|
|
:func:`MemoryProvider.initialize` 初始化 (创建 bank、加载本地文件、建立连接)
|
|
:func:`MemoryProvider.system_prompt_block` 注入系统提示词的静态说明文本
|
|
:func:`MemoryProvider.prefetch` 每轮对话前召回上下文
|
|
:func:`MemoryProvider.queue_prefetch` 下一轮的后台预热
|
|
:func:`MemoryProvider.sync_turn` 每轮对话后异步写入后端
|
|
:func:`MemoryProvider.get_tool_schemas` 暴露给 LLM 的工具 schema
|
|
:func:`MemoryProvider.handle_tool_call` 路由工具调用
|
|
:func:`MemoryProvider.shutdown` 清理 (flush 队列、关连接)
|
|
|
|
可选 hooks
|
|
==========
|
|
仅按 ``生产推荐档`` 移植两个 ——
|
|
- :func:`MemoryProvider.on_session_switch` thread_id 切换时 (branch/resume/reset/压缩后)
|
|
- :func:`MemoryProvider.on_memory_write` builtin 写入时,通知外部 provider 镜像
|
|
|
|
参见 ``docs/MEMORY_V2_DESIGN_ZH.md`` §4 的完整接口定义和 §16 的实施分期。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MemoryProvider(ABC):
|
|
"""所有记忆 provider 必须实现的接口。"""
|
|
|
|
# ---- 身份 ------------------------------------------------------------
|
|
|
|
@property
|
|
@abstractmethod
|
|
def name(self) -> str:
|
|
"""短标识,如 ``'builtin'`` / ``'hindsight'``。"""
|
|
|
|
# ---- 生命周期 ---------------------------------------------------------
|
|
|
|
@abstractmethod
|
|
def is_available(self) -> bool:
|
|
"""是否已配置并可用。
|
|
|
|
在 agent 启动时被调用,用于判断要不要激活该 provider。
|
|
实现里只检查配置和依赖,不发网络请求。
|
|
"""
|
|
|
|
@abstractmethod
|
|
def initialize(
|
|
self,
|
|
*,
|
|
user_id: str,
|
|
agent_id: str | None = None,
|
|
thread_id: str | None = None,
|
|
base_dir: str,
|
|
**kwargs: Any,
|
|
) -> None:
|
|
"""初始化:打开连接、创建 bank、加载本地文件等。
|
|
|
|
kwargs 至少包含 ``user_id`` / ``base_dir``,其余字段视 provider 而定。
|
|
Provider 可以根据 ``agent_id is None`` 决定走全局还是 per-agent 路径。
|
|
|
|
Parameters
|
|
----------
|
|
user_id:
|
|
当前用户的隔离标识(无认证场景下为 ``"default"``)。
|
|
agent_id:
|
|
当前 agent 的稳定标识;``None`` 表示走全局/默认 agent。
|
|
thread_id:
|
|
当前对话的 thread id;某些 provider 用它做并发隔离或 bank 后缀。
|
|
base_dir:
|
|
``Paths.base_dir`` 字符串形式,避免每个 provider 重复解析路径。
|
|
"""
|
|
|
|
def shutdown(self) -> None:
|
|
"""清理:刷队列、关连接。默认 no-op。"""
|
|
|
|
# ---- 系统提示词块 -----------------------------------------------------
|
|
|
|
def system_prompt_block(self) -> str:
|
|
"""注入系统提示词的静态文本。
|
|
|
|
典型用途:USER.md / MEMORY.md 的 frozen snapshot,provider 状态说明等。
|
|
返回空串表示不贡献内容。
|
|
"""
|
|
return ""
|
|
|
|
# ---- 召回 / 写入 -----------------------------------------------------
|
|
|
|
def prefetch(self, query: str, *, thread_id: str = "") -> str:
|
|
"""对话前召回上下文。
|
|
|
|
在每次 LLM 调用前由 :class:`MemoryMiddleware` 触发。``query`` 通常
|
|
是用户最近一条消息。实现要快;真正的网络召回应该走后台线程,
|
|
这里直接返回缓存结果。
|
|
|
|
返回值:带格式的文本块,会被拼接到系统提示词末尾。空串表示无内容。
|
|
"""
|
|
return ""
|
|
|
|
def queue_prefetch(self, query: str, *, thread_id: str = "") -> None:
|
|
"""后台预热下一轮的召回。默认 no-op。"""
|
|
|
|
def sync_turn(
|
|
self,
|
|
user_content: str,
|
|
assistant_content: str,
|
|
*,
|
|
thread_id: str = "",
|
|
turn_index: int | None = None,
|
|
) -> None:
|
|
"""每轮对话结束后把对话持久化到后端。
|
|
|
|
实现应当非阻塞 —— 如果后端有延迟,排队到后台处理而不是直接调用。
|
|
默认 no-op。
|
|
|
|
``turn_index`` 是当前对话的轮次序号 (从 1 开始,由编排层按对话历史算出)。
|
|
provider 通常被每轮重新构造、无法靠自身计数,需要按轮次节流写入
|
|
(例如 "每 N 轮 retain 一次") 的实现应当用它而非进程内计数器。
|
|
缺省 ``None`` 表示调用方未提供。
|
|
"""
|
|
|
|
# ---- LLM 工具 --------------------------------------------------------
|
|
|
|
@abstractmethod
|
|
def get_tool_schemas(self) -> list[dict[str, Any]]:
|
|
"""暴露给 LLM 的工具 schema 列表。
|
|
|
|
每个 schema 遵循 OpenAI function calling 格式::
|
|
|
|
{"name": "...", "description": "...", "parameters": {...}}
|
|
|
|
空数组表示 context-only(只做自动 recall/retain,不让 LLM 看见工具)。
|
|
"""
|
|
|
|
def handle_tool_call(
|
|
self, tool_name: str, args: dict[str, Any], **kwargs: Any
|
|
) -> str:
|
|
"""处理一个工具调用,必须返回 JSON 字符串。
|
|
|
|
``MemoryManager`` 会确保只把本 provider ``get_tool_schemas`` 里
|
|
声明的工具名路由到这里。
|
|
"""
|
|
raise NotImplementedError(
|
|
f"Provider {self.name} does not handle tool {tool_name}"
|
|
)
|
|
|
|
# ---- 可选 hooks (生产推荐档移植) -------------------------------------
|
|
|
|
def on_session_switch(
|
|
self,
|
|
new_thread_id: str,
|
|
*,
|
|
parent_thread_id: str = "",
|
|
reset: bool = False,
|
|
**kwargs: Any,
|
|
) -> None:
|
|
"""thread_id 切换时触发。
|
|
|
|
deerflow 的 thread_id 在以下场景会变化:
|
|
- 用户新建/切换 thread (``reset=True``)
|
|
- LangGraph thread branch (``reset=False``,``parent_thread_id`` 为原 thread)
|
|
- SummarizationMiddleware 压缩后继续 (``reset=False``)
|
|
|
|
Provider 在 ``initialize`` 时如果缓存了 per-session 的状态
|
|
(``_session_turns`` 累积缓冲、bank 关联、查询缓存等),应当在
|
|
这里刷新。默认 no-op。
|
|
"""
|
|
|
|
def on_memory_write(
|
|
self,
|
|
action: str,
|
|
target: str,
|
|
content: str,
|
|
metadata: dict[str, Any] | None = None,
|
|
) -> None:
|
|
"""builtin 写入时触发,通知外部 provider 镜像同一条记忆。
|
|
|
|
Parameters
|
|
----------
|
|
action:
|
|
``'add'`` / ``'replace'`` / ``'remove'``。
|
|
target:
|
|
``'user'`` / ``'memory'``,决定走全局 user bank 还是 agent 工作 bank。
|
|
content:
|
|
条目内容。``remove`` 时为待删除条目的内容(供 external 做幂等)。
|
|
metadata:
|
|
写入来源元数据 (``user_id`` / ``agent_id`` / ``thread_id`` / ``tool_name``)。
|
|
|
|
默认 no-op —— 只有需要镜像 builtin 写入的外部 provider (例如 Hindsight)
|
|
才需要重写。``MemoryManager`` 调度时会跳过 builtin 自身,避免无意义的回环。
|
|
"""
|
|
|
|
|
|
__all__ = ["MemoryProvider"]
|