github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-25 07:28:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

[feat] memory: pluggable MemoryManager interface for backend onboarding (#4326)

Browse Source 
* refactor(memory): pluggable MemoryManager interface for backend onboarding

Optimize the MemoryManager interface layer so new backends (mem0/openviking)
onboard with less code and the contract stays stable as capabilities are
added. A minimal backend now implements only from_config + add + get_context
(verified by test_memory_manager_interface.py::_MinimalBackend onboarding via
the factory); the factory no longer knows a backend's private hooks.

- MemoryManager: ABC -> pydantic BaseModel; three-tier methods (tier-1
  add/get_context abstract; tier-2 management defaults; tier-3 optional hooks
  warm/reload/fact + on_pre_compress/on_turn_start). Dropped 3 self-serving
  hooks. 6 hasattr probe sites -> direct call + try/except NotImplementedError.
- from_config classmethod: factory thins to resolve + inject storage_path +
  collect host hooks + call from_config; DeerMem-specific hook consumption
  moved from factory to DeerMem.from_config.
- Invariants: @model_validator (mode='tool' requires search via supports_search
  ClassVar); DeerMemConfig storage_path-is-file check moved here from factory.
- Async: aadd/aget_context/asearch default to the sync path (speculative).
- Callbacks: MemoryCallbacks + LangfuseMemoryCallbacks; on_memory_llm_call
  subsumes tracing_callback (same signature/timing/mutation); deleted the
  tracing_callback field. DeerMem decoupled from langfuse (portability).
- noop keeps read-op empty overrides (avoids router 500s on the
  disable-memory-via-noop path); only delete/export inherit the base raise.

Behavior preserved: 661 passed / 13 skipped. Docs: backends/README.md rewritten
(three-tier + from_config + callbacks); samples README updated; removed stale
private doc paths.

Co-Authored-By: Claude <noreply@anthropic.com>

* fix(memory): 501 on unsupported read/manage endpoints + accurate warm log

Review follow-up on the three-tier MemoryManager refactor.

- Read/manage endpoints (GET /memory, /memory/export, /memory/status,
  DELETE /memory, POST /memory/import) and the /memory/reload fallback now
  catch NotImplementedError -> 501, matching the fact-CRUD endpoints. The
  hasattr->try/except migration had skipped these: they were @abstractmethod
  before (every backend implemented them, so they never raised), so once they
  became tier-2 default-raise a minimal backend (only add + get_context) hit a
  raw 500 -- there is no global NotImplementedError handler. get_memory is
  shared via _get_memory_or_501 (covers /memory, export, status, reload
  fallback). noop is unchanged: its read-op empty overrides never raise.
- warm() base default returns None (tri-state: True=warmed, False=failed,
  None=nothing to warm) so the Gateway lifespan logs "skipping" for a
  non-DeerMem backend (e.g. noop) instead of the inaccurate "warmed
  successfully" it never earned. DeerMem.warm keeps True/False.
- Tests: 6 router 501 tests (read/manage + reload fallback) + 2 lifespan
  warm-log tests (None->skipping, False->warning); conformance/pluggable
  assert warm() is None.

705 passed / 13 skipped; lint clean.

Co-Authored-By: Claude <noreply@anthropic.com>

* fix(memory): review follow-ups - search-flag consistency, client reload, backend_config purity

Address review feedback on the three-tier MemoryManager refactor:

- [Medium] supports_search/search drift: the invariant now requires the
  supports_search ClassVar flag to MATCH whether search() is actually
  overridden (type(self).search is not MemoryManager.search), so the flag
  can't drift from the impl. Catches both directions at instantiation: a
  backend that overrides search() but forgets supports_search=True (was a
  misleading tool-mode rejection), and one that sets the flag without
  overriding (was a runtime NotImplementedError on the first memory_search).
  noop sets supports_search=True to match its search() override. Conformance
  adds drift + consistent-backend tests.
- [Low] client.reload_memory fallback: wrap the get_memory fallback so a
  minimal backend (only add + get_context) surfaces a clean NotImplementedError
  ("implements neither reload_memory nor get_memory") instead of an uncaught
  propagation -- mirrors the router's 501. Test added.
- [Low] backend_config purity: DeerMem.from_config restores backend_config to
  the pure data the host passed after model_post_init parses the injected hooks
  into DeerMemConfig (self._config, PrivateAttr); the field stays serializable
  (no callables/LLM) and matches the README ("host hooks NOT in backend_config").
  Test asserts purity + hooks wired.
- [Low] CHANGELOG: breaking-change note that mode='tool' + non-search backend
  now fails fast at startup (was silently empty) so operators recognize it on
  upgrade.
- [Nit] .gitignore: drop the env-specific .tmp-pytest/ entry (--basetemp is
  local-only, not make test/CI).

709 passed / 13 skipped; lint clean.

Co-Authored-By: Claude <noreply@anthropic.com>

* docs(changelog): correct memory tool-mode fail-fast note

The CHANGELOG entry said mode='tool' + a non-search backend "(e.g. noop)"
fails fast at startup, but noop overrides search() (returns []) and sets
supports_search=True (required by the consistency invariant), so noop IS
search-capable and noop+tool does NOT fail fast. The fail-fast only affects a
custom backend that onboards without overriding search(). Reworded to drop the
misleading noop example and state both shipping backends implement search().

Co-Authored-By: Claude <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
lllyfff 2026-07-22 14:40:57 +08:00 committed by GitHub
parent 05e4f4f6d8
commit 01a89f2379
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
21 changed files with 1295 additions and 453 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
CHANGELOG.md
View File

@ -48,6 +48,15 @@ This section accumulates work toward the **2.1.0** milestone
with a warning** and the factory **raises** if `storage_path` resolves to an
existing file. Set `memory.backend_config.storage_path` to a directory for a
custom root. ([#4122])
- **memory:** `memory.mode: tool` with a backend that does not implement
`search()` now fails fast at Gateway startup with a `ValueError` from the
`MemoryManager` invariant, instead of starting successfully and silently
returning empty results on every `memory_search` call. Both shipping backends
implement `search()` (DeerMem retrieves; `noop` returns `[]`), so this only
affects a custom backend that onboards without overriding `search()`. It is
intentional -- silent empties are worse than a loud startup error. Fix: switch
to `mode: middleware` or override `search()` (and set `supports_search=True`).
([#4324])
### Added
@ -1099,3 +1108,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
[#4264]: https://github.com/bytedance/deer-flow/pull/4264
[#4287]: https://github.com/bytedance/deer-flow/pull/4287
[#4288]: https://github.com/bytedance/deer-flow/pull/4288
[#4324]: https://github.com/bytedance/deer-flow/issues/4324

31
backend/app/gateway/app.py
View File

@ -206,27 +206,26 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
# Pre-warm tiktoken encoding cache so the first memory-injection request
# never blocks on the BPE data download (which hits an OpenAI/Azure URL
# that may be unreachable in restricted networks — see issue #3402).
# Warm-up runs via the manager's `warm` capability (getattr-probed, so
# non-DeerMem backends skip it). DeerMem.warm re-checks token_counting==
# "char" and returns early, so char-mode backends never touch tiktoken
# (avoids even the 5s probe in
# network-restricted deployments — see issue #3429).
# Warm-up runs via the manager's `warm()` tier-3 hook. DeerMem.warm re-checks
# token_counting=="char" and returns early, so char-mode backends never touch
# tiktoken (avoids even the 5s probe in network-restricted deployments - see
# issue #3429). A backend with nothing to warm (e.g. noop) returns None from
# the base default -- log "skipping" instead of the misleading "warmed
# successfully" so the log reflects what actually happened.
try:
from deerflow.agents.memory import get_memory_manager
manager = get_memory_manager()
warm = getattr(manager, "warm", None)
if not callable(warm):
logger.info("Memory backend %s has no warm-up hook; skipping tiktoken warm-up", type(manager).__name__)
warmed = await asyncio.wait_for(
asyncio.to_thread(manager.warm),
timeout=5,
)
if warmed is None:
logger.info("Memory backend %s has nothing to warm; skipping tiktoken warm-up", type(manager).__name__)
elif warmed:
logger.info("tiktoken encoding cache warmed successfully")
else:
warmed = await asyncio.wait_for(
asyncio.to_thread(warm),
timeout=5,
)
if warmed:
logger.info("tiktoken encoding cache warmed successfully")
else:
logger.warning("tiktoken encoding cache warm-up failed; token counting will use character-based fallback until tiktoken loads successfully")
logger.warning("tiktoken encoding cache warm-up failed; token counting will use character-based fallback until tiktoken loads successfully")
except TimeoutError:
logger.warning("tiktoken encoding cache warm-up timed out; token counting will use character-based fallback until tiktoken loads successfully")
except Exception:

109
backend/app/gateway/routers/memory.py
View File

@ -6,7 +6,7 @@ from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field, field_validator
from app.gateway.internal_auth import get_trusted_internal_owner_user_id
from deerflow.agents.memory import MemoryConflictError, MemoryCorruptionError, get_memory_manager
from deerflow.agents.memory import MemoryConflictError, MemoryCorruptionError, MemoryManager, get_memory_manager
from deerflow.config.memory_config import get_memory_config
from deerflow.config.paths import make_safe_user_id
from deerflow.runtime.user_context import get_effective_user_id
@ -127,22 +127,41 @@ def _map_memory_manager_error(exc: MemoryConflictError | MemoryCorruptionError)
return HTTPException(status_code=500, detail="Stored memory data is corrupted.")
def _require_capability(name: str, *, label: str):
"""Return a DeerMem-internal capability (bound method) or raise 501.
def _unsupported_501(manager: object, label: str) -> HTTPException:
"""501 for an unsupported memory operation.
``reload_memory`` / ``create_fact`` / ``delete_fact`` / ``update_fact`` are
not on the ``MemoryManager`` ABC -- they are DeerMem-internal. Probe with
``hasattr`` rather than importing DeerMem, so this router has no hard
dependency on the default backend: a non-DeerMem (or removed) backend
simply lacks the attribute and the endpoint returns 501.
Tier-3 hooks (``reload_memory`` / ``create_fact`` / ``delete_fact`` /
``update_fact``) and tier-2 management ops (``get_memory`` / ``clear_memory``
/ ``import_memory``) all default to ``raise NotImplementedError``; backends
that support them override, unsupported ones inherit the raise. Before the
contract change these were ``@abstractmethod`` (every backend implemented
them, so the endpoints could never raise); now a minimal backend (only
``add`` + ``get_context``) inherits the raise, so endpoints invoke the
method directly and catch ``NotImplementedError`` -> this 501. There is no
global ``NotImplementedError`` handler, so an uncaught raise is a raw 500.
"""
manager = get_memory_manager()
if not hasattr(manager, name):
raise HTTPException(
status_code=501,
detail=f"Operation '{label}' not supported by memory backend '{type(manager).__name__}'.",
)
return getattr(manager, name)
return HTTPException(
status_code=501,
detail=f"Operation '{label}' not supported by memory backend '{type(manager).__name__}'.",
)
def _get_memory_or_501(manager: MemoryManager, user_id: str, label: str) -> dict[str, Any]:
"""Read the full memory doc; 501 if the backend doesn't expose one.
``get_memory`` is tier-2 (default ``raise NotImplementedError``); a minimal
backend doesn't expose a full doc. The standalone read endpoints (GET
/memory, /memory/export, /memory/status) and the /memory/reload fallback all
route reads through here so an unsupported backend gets a clean 501 instead
of a raw 500. ``label`` is the operation name in the 501 detail (the
endpoint's verb, e.g. "get memory" / "export memory" / "reload memory").
"""
try:
return manager.get_memory(user_id=user_id)
except NotImplementedError:
raise _unsupported_501(manager, label) from None
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
class FactCreateRequest(BaseModel):
@ -220,10 +239,8 @@ async def get_memory(http_request: Request) -> MemoryResponse:
}
```
"""
try:
memory_data = get_memory_manager().get_memory(user_id=_resolve_memory_user_id(http_request))
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
manager = get_memory_manager()
memory_data = _get_memory_or_501(manager, _resolve_memory_user_id(http_request), "get memory")
return MemoryResponse(**memory_data)
@ -246,14 +263,15 @@ async def reload_memory(http_request: Request) -> MemoryResponse:
user_id = _resolve_memory_user_id(http_request)
manager = get_memory_manager()
try:
if hasattr(manager, "reload_memory"):
memory_data = manager.reload_memory(user_id=user_id)
else:
# Non-DeerMem backends have no reload concept; return current memory.
# (Asymmetry vs fact CRUD, which raises 501 when unsupported: reload is a
# read-only refresh, so degrading to get_memory is safe and still useful;
# silently no-op'ing a write would hide data loss, so writes fail loud.)
memory_data = manager.get_memory(user_id=user_id)
memory_data = manager.reload_memory(user_id=user_id)
except NotImplementedError:
# Non-DeerMem backends have no reload concept; fall back to get_memory
# (read-only refresh, so degrading is safe and still useful -- vs fact
# CRUD writes, which fail loud at 501 since silently no-op'ing a write
# would hide data loss). If get_memory is also unsupported (a minimal
# backend with no full doc), surface 501 rather than a raw 500: reads
# degrade only when there is a doc to degrade to.
memory_data = _get_memory_or_501(manager, user_id, "reload memory")
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
return MemoryResponse(**memory_data)
@ -268,8 +286,11 @@ async def reload_memory(http_request: Request) -> MemoryResponse:
)
async def clear_memory(http_request: Request) -> MemoryResponse:
"""Clear all persisted memory data."""
manager = get_memory_manager()
try:
memory_data = get_memory_manager().clear_memory(user_id=_resolve_memory_user_id(http_request))
memory_data = manager.clear_memory(user_id=_resolve_memory_user_id(http_request))
except NotImplementedError:
raise _unsupported_501(manager, "clear memory") from None
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
except OSError as exc:
@ -287,14 +308,16 @@ async def clear_memory(http_request: Request) -> MemoryResponse:
)
async def create_memory_fact_endpoint(request: FactCreateRequest, http_request: Request) -> MemoryResponse:
"""Create a single fact manually."""
manager = get_memory_manager()
try:
create_fact = _require_capability("create_fact", label="create fact")
memory_data, fact_id = create_fact(
memory_data, fact_id = manager.create_fact(
content=request.content,
category=request.category,
confidence=request.confidence,
user_id=_resolve_memory_user_id(http_request),
)
except NotImplementedError:
raise _unsupported_501(manager, "create fact") from None
except ValueError as exc:
raise _map_memory_fact_value_error(exc) from exc
except (MemoryConflictError, MemoryCorruptionError) as exc:
@ -317,9 +340,11 @@ async def create_memory_fact_endpoint(request: FactCreateRequest, http_request:
)
async def delete_memory_fact_endpoint(fact_id: str, http_request: Request) -> MemoryResponse:
"""Delete a single fact from memory by fact id."""
manager = get_memory_manager()
try:
delete_fact = _require_capability("delete_fact", label="delete fact")
memory_data = delete_fact(fact_id, user_id=_resolve_memory_user_id(http_request))
memory_data = manager.delete_fact(fact_id, user_id=_resolve_memory_user_id(http_request))
except NotImplementedError:
raise _unsupported_501(manager, "delete fact") from None
except KeyError as exc:
raise HTTPException(status_code=404, detail=f"Memory fact '{fact_id}' not found.") from exc
except (MemoryConflictError, MemoryCorruptionError) as exc:
@ -339,15 +364,17 @@ async def delete_memory_fact_endpoint(fact_id: str, http_request: Request) -> Me
)
async def update_memory_fact_endpoint(fact_id: str, request: FactPatchRequest, http_request: Request) -> MemoryResponse:
"""Partially update a single fact manually."""
manager = get_memory_manager()
try:
update_fact = _require_capability("update_fact", label="update fact")
memory_data = update_fact(
memory_data = manager.update_fact(
fact_id=fact_id,
content=request.content,
category=request.category,
confidence=request.confidence,
user_id=_resolve_memory_user_id(http_request),
)
except NotImplementedError:
raise _unsupported_501(manager, "update fact") from None
except ValueError as exc:
raise _map_memory_fact_value_error(exc) from exc
except KeyError as exc:
@ -369,10 +396,8 @@ async def update_memory_fact_endpoint(fact_id: str, request: FactPatchRequest, h
)
async def export_memory(http_request: Request) -> MemoryResponse:
"""Export the current memory data."""
try:
memory_data = get_memory_manager().get_memory(user_id=_resolve_memory_user_id(http_request))
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
manager = get_memory_manager()
memory_data = _get_memory_or_501(manager, _resolve_memory_user_id(http_request), "export memory")
return MemoryResponse(**memory_data)
@ -385,8 +410,11 @@ async def export_memory(http_request: Request) -> MemoryResponse:
)
async def import_memory(request: MemoryResponse, http_request: Request) -> MemoryResponse:
"""Import and persist memory data."""
manager = get_memory_manager()
try:
memory_data = get_memory_manager().import_memory(request.model_dump(exclude_none=True), user_id=_resolve_memory_user_id(http_request))
memory_data = manager.import_memory(request.model_dump(exclude_none=True), user_id=_resolve_memory_user_id(http_request))
except NotImplementedError:
raise _unsupported_501(manager, "import memory") from None
except (MemoryConflictError, MemoryCorruptionError) as exc:
raise _map_memory_manager_error(exc) from exc
except OSError as exc:
@ -458,7 +486,8 @@ async def get_memory_status(http_request: Request) -> MemoryStatusResponse:
Combined memory configuration and current data.
"""
config = get_memory_config()
memory_data = get_memory_manager().get_memory(user_id=_resolve_memory_user_id(http_request))
manager = get_memory_manager()
memory_data = _get_memory_or_501(manager, _resolve_memory_user_id(http_request), "get memory status")
return MemoryStatusResponse(
config=MemoryConfigResponse(

45
backend/packages/harness/deerflow/agents/memory/backends/README.md
View File

@ -25,7 +25,7 @@ Copy `noop/` to `backends/<yourname>/` and edit three files in this folder plus
| File | What to change |
|---|---|
| `backends/<yourname>/config.py` | Declare your config fields + `from_backend_config` (parse `backend_config`; read `storage_path` from it - **do not import deer-flow path helpers**) |
| `backends/<yourname>/<yourname>_manager.py` | Rename the class; parse config in `__init__`; implement the 9 ABC methods; optionally implement fact CRUD (see [Backend Contract](#backend-contract)) |
| `backends/<yourname>/<yourname>_manager.py` | Rename the class; parse config in `model_post_init`; implement `from_config` + the tier-1 abstracts (`add`/`get_context`); override tier-2/3 methods as needed (see [Backend Contract](#backend-contract)) |
| `backends/<yourname>/__init__.py` | `MANAGER_CLASS = YourManager` (relative import) |
| `config.yaml` (repo root, parent of `backend/`) | `memory.manager_class: <yourname>` + your knobs under `memory.backend_config` |
| `packages/harness/pyproject.toml` | **Only if the backend needs external libs**: declare the dependency; add `[tool.uv.sources]` for vendored source. Otherwise `uv sync` purges it (see [Common Pitfalls](#common-pitfalls)) |
@ -46,9 +46,15 @@ Then **restart deer-flow** - the memory manager is a process-level singleton; a
## Backend Contract
### 1. The 9 ABC methods
### 1. The three-tier contract
Implement every method on `MemoryManager` in `packages/harness/deerflow/agents/memory/manager.py`. Signatures must match (parameter names, keyword-only args). `noop` is the empty-implementation reference.
`MemoryManager` is a pydantic `BaseModel` (not a bare ABC). Methods are tiered:
- **Tier 1 (abstract)** -- `add` + `get_context`: every backend MUST implement (write + read-inject are the backend's fundamental duties; missing one is caught at instantiation).
- **Tier 2 (management, with defaults)** -- `add_nowait` (delegates to `add`), `search` / `get_memory` / `clear_memory` / `import_memory` / `export_memory` / `delete_memory` (default `raise NotImplementedError`), `shutdown_flush` (default `True`). Override the ones your backend supports.
- **Tier 3 (optional hooks, with defaults)** -- `warm` (default `True`), `reload_memory` / `create_fact` / `delete_fact` / `update_fact` (default raise), `on_pre_compress` / `on_turn_start` (default no-op).
A new backend implements `from_config` + `add` + `get_context` and overrides only what it supports; the rest inherits defaults. Signatures must match (parameter names, keyword-only args). `noop` is the minimal reference.
### 2. Return shape (critical, easy to get wrong)
@ -59,15 +65,15 @@ Implement every method on `MemoryManager` in `packages/harness/deerflow/agents/m
A non-DeerMem backend maps its native records (e.g. `{"results": [...]}`) into this shape via a small adapter helper.
### 3. Optional capabilities (DeerMem-internal, not on the ABC)
### 3. Tier-3 hooks (contracted, no `hasattr` probing)
The gateway probes these with `hasattr(manager, "<name>")` and returns 501 when absent:
`create_fact` / `delete_fact` / `update_fact` / `reload_memory` / `warm` are tier-3 hooks ON the base contract (with defaults). Callers (gateway / client / tools) invoke them directly and catch `NotImplementedError` for unsupported backends -- no more `hasattr` probing.
- `create_fact` / `delete_fact` / `update_fact` - the frontend's add/delete/edit-fact buttons. Signatures are in the commented block at the bottom of `noop/noop_manager.py`.
- `reload_memory` - the frontend's reload button (delegate to `get_memory` if your backend has no cache).
- `warm` - one-time warm-up at gateway startup (skipped if absent).
- `create_fact` / `delete_fact` / `update_fact` - the frontend's add/delete/edit-fact buttons. Default raises (caller returns 501).
- `reload_memory` - the frontend's reload button (caller falls back to `get_memory` on `NotImplementedError`).
- `warm` - one-time warm-up at gateway startup (default `True` = nothing to warm).
Implement the ones you support; leave the rest as 501.
Implement the ones your backend supports; the rest inherit the default raise.
### 4. Portability (the golden rule)
@ -80,15 +86,17 @@ from deerflow.agents.memory.manager import MemoryManager
Change that one line (and only that line) to port the backend to another agent. **Do not import deer-flow path helpers, config singletons, or models** - get `storage_path` and everything else from `backend_config`.
### 5. What the host injects into `backend_config`
### 5. What the host provides
The factory (`manager.py::get_memory_manager`) injects these for every backend:
The factory (`manager.py::get_memory_manager`) resolves the backend class, injects `storage_path` into `backend_config`, then calls `cls.from_config(backend_config, mode=cfg.mode, **host_hooks)`. The host hooks (passed as `from_config` kwargs, NOT in `backend_config`):
- `storage_path` (str) - a writable state dir (the host's `runtime_home` by default, or whatever `config.yaml` sets). **Use this as your storage root.**
- `tracing_callback` (Callable | None) - trace your LLM calls (langfuse). Ignore if you don't trace.
- `should_keep_hidden_message` (Callable | None) - filter `hide_from_ui` messages. Ignore if not relevant.
- `backend_config["storage_path"]` (str) - a writable state dir (the host's `runtime_home` by default, or whatever `config.yaml` sets). **Use this as your storage root.**
- `callbacks` (`MemoryCallbacks` | None) - observability; `on_memory_llm_call` merges trace metadata before your LLM call (langfuse). Pass it to your LLM path; ignore if you don't trace.
- `should_keep_hidden_message` / `trace_context_manager` / `host_llm_factory` - other host hooks; consume in `from_config` if relevant.
- Plus whatever the user puts under `config.yaml::memory.backend_config` (your backend's own knobs).
Each backend's `from_config` consumes the hooks it needs (DeerMem does; noop ignores them).
## Do Not Modify
These are backend-agnostic. Don't touch them when swapping backends (unless you're changing the **shared contract**, which affects every backend):
@ -99,12 +107,12 @@ These are backend-agnostic. Don't touch them when swapping backends (unless you'
| `packages/harness/deerflow/agents/middlewares/memory_middleware.py` | `after_agent` -> `manager.add` |
| `packages/harness/deerflow/agents/memory/summarization_hook.py` | summarization -> `manager.add_nowait` |
| `packages/harness/deerflow/agents/lead_agent/prompt.py` | `_get_memory_context` -> `manager.get_context` |
| `app/gateway/routers/memory.py` | HTTP endpoints -> `manager.*` (hasattr-probed) |
| `app/gateway/routers/memory.py` | HTTP endpoints -> `manager.*` (direct call + try/except `NotImplementedError`) |
| `packages/harness/deerflow/config/memory_config.py` | shared 4 fields (`enabled` / `injection_enabled` / `manager_class` / `backend_config`) |
| `frontend/src/components/workspace/settings/memory-settings-page.tsx` | frontend memory page (assumes DeerMem shape) |
> [!NOTE]
> The gateway and frontend are currently hard-coded to the DeerMem shape - that's why backends must return DeerMem-shape data (contract #2). Making them fully backend-agnostic is a larger refactor; see `E:\deerflow\memory\plugin\00-插件兼容性矩阵.md`.
> The gateway and frontend are currently hard-coded to the DeerMem shape - that's why backends must return DeerMem-shape data (contract #2). Making them fully backend-agnostic is a larger refactor.
## Common Pitfalls
@ -119,6 +127,5 @@ Lessons from integrating external backends:
## Reference
- **Template**: `noop/` - empty implementation with full docstrings; copy and go.
- **Design proposal**: `E:\deerflow\memory\记忆系统方案.md`.
- **Plugin plans + compatibility matrix**: `E:\deerflow\memory\plugin\` (`00-插件兼容性矩阵.md` is the spine; defines the 9 shared contracts S1-S9).
- **Template**: `noop/` - minimal implementation with full docstrings; copy and go.
- **Contract + factory**: `packages/harness/deerflow/agents/memory/manager.py` (`MemoryManager` base, `MemoryCallbacks`, `get_memory_manager` factory).

131
backend/packages/harness/deerflow/agents/memory/backends/deermem/deer_mem.py
View File

@ -3,29 +3,30 @@
DeerMem wraps the DeerFlow memory machinery (the five ``core/`` modules:
storage / queue / updater / prompt / message_processing) behind the
backend-neutral :class:`~deerflow.agents.memory.manager.MemoryManager`
contract. DeerMem owns its storage / queue / updater as injected instance
attributes (no module-level singletons): the factory passes ``backend_config``
to ``__init__``, which parses it into a :class:`DeerMemConfig` and constructs
the dependencies. Behaviour matches the pre-abstraction code: the same filter +
contract. DeerMem owns its storage / queue / updater as ``PrivateAttr`` dependencies
(no module-level singletons): the factory passes ``backend_config`` to the
BaseModel field, and ``model_post_init`` parses it into a :class:`DeerMemConfig`
and constructs the dependencies. Behaviour matches the pre-abstraction code: the same filter +
human/ai validation + correction/reinforcement detection feeds the same
debounced queue; the same ``format_memory_for_injection`` produces injection
text; the same CRUD backs the management endpoints.
DeerMem-private concerns (filter/detect, the ``<memory>`` wrap, ``enabled``
gating, the facts model) deliberately stay OUT of the ABC -- they live here.
Methods not on the ABC (``warm`` / ``reload_memory`` / ``create_fact`` /
``delete_fact`` / ``update_fact``) are DeerMem internals exposed for
``hasattr`` capability probing: the gateway probes ``hasattr(manager, "warm")``
at startup and the gateway/client probe ``hasattr(manager, "create_fact")`` for
fact CRUD, rather than importing DeerMem, so a non-DeerMem (or removed) backend
never breaks those modules at import time (see MemoryManager plan, step 8).
``warm`` / ``reload_memory`` / fact CRUD are tier-3 optional hooks ON the ABC
(with defaults: ``warm``=True, the rest raise ``NotImplementedError``); DeerMem
overrides the ones it supports. Callers (gateway / client / tools) invoke them
directly and catch ``NotImplementedError`` for unsupported backends -- no more
``hasattr`` probing.
"""
from __future__ import annotations
import copy
import logging
from typing import Any
from typing import Any, ClassVar, Literal
from pydantic import PrivateAttr
from deerflow.agents.memory.manager import MemoryConflictError, MemoryCorruptionError, MemoryManager
@ -90,15 +91,32 @@ def _compat_document(memory_data: dict[str, Any]) -> dict[str, Any]:
class DeerMem(MemoryManager):
"""Default memory backend: file-backed facts + debounced LLM extraction."""
def __init__(self, backend_config: dict[str, Any] | None = None) -> None:
"""Construct DeerMem with its dependencies (dependency injection).
# Backend-private dependencies are PrivateAttr (not pydantic fields): they
# are non-pydantic objects (storage / llm / queue) that must NOT participate
# in validation / serialization. Built once in model_post_init from
# self.backend_config -> DeerMemConfig.
_config: Any = PrivateAttr(default=None)
_storage: Any = PrivateAttr(default=None)
_llm: Any = PrivateAttr(default=None)
_updater: Any = PrivateAttr(default=None)
_queue: Any = PrivateAttr(default=None)
_correction_patterns: Any = PrivateAttr(default=None)
_reinforcement_patterns: Any = PrivateAttr(default=None)
Args:
backend_config: DeerMem-private config dict (from
``MemoryConfig.backend_config``). Parsed into a
:class:`DeerMemConfig` (defaults apply when empty/None).
# DeerMem implements search() (case-insensitive substring over stored facts),
# so it is valid for mode="tool" (the base invariant validator requires this
# for tool mode). Backends without real search inherit the False default and
# cannot be used with mode="tool".
supports_search: ClassVar[bool] = True
def model_post_init(self, __context: Any) -> None:
"""Construct DeerMem's dependencies from ``self.backend_config``.
Runs after pydantic's ``__init__`` validates the fields. Parses
``backend_config`` into a :class:`DeerMemConfig` (defaults apply when
empty/None) and wires storage / patterns / llm / updater / queue (DI).
"""
self._config = DeerMemConfig.from_backend_config(backend_config)
self._config = DeerMemConfig.from_backend_config(self.backend_config)
self._storage = create_storage(self._config)
# Signal-detection patterns (externalized YAML; ``patterns_dir`` override
# or bundled defaults = pre-externalization behavior). Loaded once at
@ -109,7 +127,7 @@ class DeerMem(MemoryManager):
# so zero-config DeerMem (empty `model`) still extracts via the app default,
# mirroring pre-abstraction `model_name: null`. Standalone (no factory) -> None.
self._llm = self._config.host_llm if self._config.host_llm is not None else build_llm(self._config.model)
self._updater = MemoryUpdater(self._config, self._storage, self._llm, prompts_dir=self._config.prompts_dir)
self._updater = MemoryUpdater(self._config, self._storage, self._llm, prompts_dir=self._config.prompts_dir, callbacks=self.callbacks)
# Validate the *global* explicit prompt templates at construction so a
# misconfigured prompts_dir surfaces at startup rather than as a silent
# dropped update. Per-agent overrides ({prompts_dir}/{agent}/*.yaml)
@ -129,6 +147,47 @@ class DeerMem(MemoryManager):
load_prompt_messages("memory_update", _dummy_vars, prompts_dir=self._config.prompts_dir)
self._queue = MemoryUpdateQueue(self._config, self._updater)
@classmethod
def from_config(
cls,
backend_config: dict[str, Any] | None = None,
*,
mode: Literal["middleware", "tool"] = "middleware",
**host_hooks: Any,
) -> DeerMem:
"""Build a DeerMem with dependencies wired, consuming host hooks.
The factory passes host hooks (tracing, hidden-message filter,
trace-context manager, a host-llm factory) as kwargs rather than
injecting them into ``backend_config``; DeerMem merges the ones it
consumes (DeerMemConfig fields) here, respecting explicit
``backend_config`` values. ``host_llm`` is built from the host factory
only when no model is configured (host_llm takes precedence over
``build_llm(model)``; building an unused host default when a model
exists would waste startup time). The actual dependency wiring runs in
``model_post_init`` (shared with direct construction).
"""
config_dict = dict(backend_config or {})
for key in ("should_keep_hidden_message", "trace_context_manager"):
if key not in config_dict and key in host_hooks:
config_dict[key] = host_hooks[key]
if "host_llm" not in config_dict:
model_cfg = config_dict.get("model")
if not (isinstance(model_cfg, dict) and model_cfg.get("model")):
host_llm_factory = host_hooks.get("host_llm_factory")
if host_llm_factory is not None:
config_dict["host_llm"] = host_llm_factory()
# callbacks is a base MemoryManager field (not DeerMemConfig); pass through.
# config_dict carries the host hooks merged above so model_post_init can
# parse them into DeerMemConfig (self._config, PrivateAttr). After wiring,
# restore backend_config to the pure data the host passed (no injected
# hooks) so the field stays serializable and matches the README contract
# ("host hooks arrive as from_config kwargs, NOT in backend_config") --
# the hooks live in self._config, not the backend_config field.
instance = cls(backend_config=config_dict, mode=mode, callbacks=host_hooks.get("callbacks"))
instance.backend_config = dict(backend_config or {})
return instance
# ── Write ────────────────────────────────────────────────────────────
def add(
self,
@ -284,14 +343,9 @@ class DeerMem(MemoryManager):
memory_data = _call_backend(lambda: self._updater.get_memory_data(agent_name=_resolve_agent_name(agent_name), user_id=user_id))
return _compat_document(memory_data)
def delete_memory(
self,
*,
user_id: str | None = None,
agent_name: str | None = None,
) -> None:
"""Not implemented this phase (storage/updater deletion is a future ``core/`` addition)."""
raise NotImplementedError("DeerMem.delete_memory is not implemented yet")
# delete_memory / export_memory inherit the base tier-2 default (raise
# NotImplementedError) -- they are dead contract (zero callers; /memory/export
# routes via get_memory), so DeerMem no longer repeats the raise.
def clear_memory(
self,
@ -321,15 +375,6 @@ class DeerMem(MemoryManager):
)
return _compat_document(imported)
def export_memory(
self,
*,
user_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
"""Not implemented this phase (no distinct export yet; /export routes via get_memory)."""
raise NotImplementedError("DeerMem.export_memory is not implemented yet")
# ── Lifecycle ───────────────────────────────────────────────────────
def shutdown_flush(self, timeout: float) -> bool:
"""Drain the debounce queue within ``timeout`` on graceful shutdown.
@ -343,16 +388,16 @@ class DeerMem(MemoryManager):
"""
return self._queue.flush_sync(timeout)
# ── DeerMem-internal (NOT on the ABC; reached via hasattr probing) ───
# ── Tier 3 hooks (override the base defaults; warm/reload/fact CRUD) ─
def warm(self) -> bool:
"""Pre-warm DeerMem-specific resources (the tiktoken encoding cache).
Backend-agnostic startup code probes ``hasattr(manager, "warm")`` and
calls this off the event loop. Non-DeerMem backends lack the attribute,
so their warm-up is skipped entirely (e.g. mem0 does not use tiktoken).
Returns True if the encoding loaded (or was already cached, or warming
was unnecessary); False if tiktoken is unavailable or the download
failed.
Overrides the base tier-3 hook (default None = nothing to warm). The
Gateway lifespan calls ``manager.warm()`` directly off the event loop;
backends without heavy init inherit the None default (the host logs
"skipping"). Returns True if the encoding loaded (or was already cached,
or warming was unnecessary); False if tiktoken is unavailable or the
download failed.
"""
if self._config.token_counting == "char":
logger.info("token_counting='char'; tiktoken not used, skipping warm-up")

49
backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/config.py
View File

@ -10,16 +10,18 @@ Field names mirror the pre-abstraction ``MemoryConfig`` private fields so the
migration is a pure move (config.yaml ``memory.<field>`` ->
``memory.backend_config.<field>``). ``model`` is a nested ``DeerMemModelConfig``
(provider/model/api_key/base_url/temperature) consumed by ``core/llm.py``;
``tracing_callback`` (step 14) and ``should_keep_hidden_message`` (step 15) are
optional host-injected hooks (None = DeerMem defaults).
``should_keep_hidden_message`` is an optional host-injected hook (None =
DeerMem default); tracing is via the base ``MemoryManager.callbacks`` field
(``on_memory_llm_call`` before the LLM call), not a DeerMemConfig slot.
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Any, Literal
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, model_validator
logger = logging.getLogger(__name__)
@ -206,7 +208,7 @@ class DeerMemConfig(BaseModel):
default=None,
description=("Directory with custom memory-extraction prompt templates (memory_update.chat.yaml, staleness_review.yaml, consolidation.yaml, fact_extraction.yaml). None (default) = bundled core/prompts/."),
)
# ── LLM (step 13: structured model sub-config consumed by core/llm.py build_llm) ──
# ── LLM (structured model sub-config consumed by core/llm.py build_llm) ──
model: DeerMemModelConfig = Field(
default_factory=DeerMemModelConfig,
description=(
@ -217,17 +219,9 @@ class DeerMemConfig(BaseModel):
"but non-LLM ops still work."
),
)
# ── Hooks (steps 14-15: optional host-injected callables; None = DeerMem defaults) ──
tracing_callback: Any = Field(
default=None,
description=(
"Optional observability callback (e.g. langfuse) invoked before the "
"memory-update LLM call as "
"``callback(invoke_config, *, thread_id, user_id, trace_id, model_name)``. "
"None = no tracing (langfuse not hard-required). Set programmatically "
"(callables cannot come from YAML)."
),
)
# ── Hooks (optional host-injected callables; None = DeerMem defaults) ──
# Tracing is via the base ``MemoryManager.callbacks`` field
# (``on_memory_llm_call`` before the LLM call), not a DeerMemConfig slot.
should_keep_hidden_message: Any = Field(
default=None,
description=("Optional ``hook(additional_kwargs) -> bool``; when set, ``hide_from_ui`` messages are kept if it returns True. None = skip all ``hide_from_ui`` (host-agnostic safe default). Set programmatically."),
@ -250,11 +244,34 @@ class DeerMemConfig(BaseModel):
"``trace_id`` into the host request-trace ContextVar for the memory-"
"update worker thread (Timer / executor), restoring structured-log "
"trace correlation. None = no binding (DeerMem standalone; trace_id "
"still reaches ``tracing_callback`` and the log message text). Set "
"still reaches ``on_memory_llm_call`` and the log message text). Set "
"programmatically."
),
)
@model_validator(mode="after")
def _check_storage_path_is_directory(self) -> DeerMemConfig:
"""DeerMem treats ``storage_path`` as a root DIRECTORY (per-user memory
under ``{storage_path}/users/{uid}/memory.json``). A file-style value
(e.g. a leftover ``.json`` from the pre-abstraction file-path semantics)
would make ``FileMemoryStorage.save``'s ``mkdir(parents=True)`` raise
``NotADirectoryError`` (caught as OSError -> silent write failure). Fail
loud at construction instead (memory is persistent state -- a wrong root
is a data-integrity footgun). Lives here (not the host factory) so it
fires even when DeerMem is built standalone / bypassing the factory.
Empty storage_path (zero-config -> host injects a dir) is allowed.
"""
if self.storage_path:
resolved = Path(self.storage_path)
if resolved.is_file():
raise ValueError(
f"memory.backend_config.storage_path={self.storage_path!r} "
f"resolves to an existing file {resolved}; DeerMem treats "
f"storage_path as a root DIRECTORY (per-user memory under "
f"{{storage_path}}/users/{{uid}}/memory.json). Point it at a directory."
)
return self
@classmethod
def from_backend_config(cls, backend_config: dict[str, Any] | None) -> DeerMemConfig:
"""Parse a ``backend_config`` dict.

19
backend/packages/harness/deerflow/agents/memory/backends/deermem/deermem/core/updater.py
View File

@ -613,7 +613,7 @@ def _escape_memory_for_prompt(memory: Any) -> Any:
class MemoryUpdater:
"""Updates memory using LLM based on conversation context."""
def __init__(self, config: DeerMemConfig, storage: MemoryStorage, llm: Any = None, *, prompts_dir: str | None = None):
def __init__(self, config: DeerMemConfig, storage: MemoryStorage, llm: Any = None, *, prompts_dir: str | None = None, callbacks: Any = None):
"""Initialize the memory updater with injected config + storage + llm (DI).
Args:
@ -623,11 +623,15 @@ class MemoryUpdater:
here). None when no LLM is configured; an update raises in that case.
prompts_dir: Optional custom prompt-template directory forwarded to
``load_prompt`` / ``load_prompt_messages``. None = bundled defaults.
callbacks: Optional ``MemoryCallbacks`` (owned by DeerMem, injected
here). ``on_memory_llm_call`` is invoked before the LLM call to
merge trace metadata into ``invoke_config``; None = no tracing.
"""
self._config = config
self._storage = storage
self._llm = llm
self._prompts_dir = prompts_dir
self._callbacks = callbacks
# ── Data access + fact CRUD (formerly module-level functions; use self._storage) ──
@ -1093,7 +1097,7 @@ class MemoryUpdater:
The update runs on a Timer / executor thread with no request ContextVar
inheritance, so log records emitted here would otherwise lose the
request trace id (it only reached ``tracing_callback`` before). The
request trace id (it only reached the pre-LLM-call tracing hook before). The
host-injected ``trace_context_manager`` hook (``None`` when DeerMem runs
standalone, outside the deer-flow factory) binds ``trace_id`` for the
duration of the call and restores the prior binding on exit. A ``None``
@ -1156,11 +1160,12 @@ class MemoryUpdater:
if model is None:
raise RuntimeError("DeerMem memory update requested but no LLM is configured (set memory.backend_config.model in config).")
invoke_config: dict[str, Any] = {"run_name": "memory_agent"}
# Optional observability callback (e.g. langfuse), injected via
# backend_config.tracing_callback. None = no tracing (langfuse is not
# hard-required); the host may pass a wrapper around its own tracer.
if self._config.tracing_callback is not None:
self._config.tracing_callback(
# Pre-LLM-call observability hook (e.g. langfuse): merge trace
# metadata into invoke_config before the call so a tracer emits a
# span at the LLM boundary. None = no tracing (langfuse not
# hard-required). Subsumes the former backend_config.tracing_callback.
if self._callbacks is not None:
self._callbacks.on_memory_llm_call(
invoke_config,
thread_id=thread_id,
user_id=user_id,

39
backend/packages/harness/deerflow/agents/memory/backends/noop/config.py
View File

@ -17,17 +17,15 @@ rule** (read before writing a backend):
port the backend to another agent. Everything else -- storage root, model,
hooks -- arrives via ``backend_config``.
What the factory (``manager.py::get_memory_manager``) injects into
``backend_config`` for every backend:
- ``storage_path`` (str): a writable state dir (the host's default, or
whatever the user sets in config.yaml). **Use this as your storage root** --
do NOT call a deer-flow path helper yourself.
- ``tracing_callback`` (Callable | None): host default for tracing the
backend's LLM calls (langfuse). Declare a slot + consume it if your backend
traces; otherwise ignore (unknown-key filtering drops it).
- ``should_keep_hidden_message`` (Callable | None): host default for keeping
``hide_from_ui`` messages (human-clarification). Consume if your backend
filters hidden messages; otherwise ignore.
What the factory (``manager.py::get_memory_manager``) provides to each backend:
- ``backend_config["storage_path"]`` (str): a writable state dir (the host's
default, or whatever the user sets in config.yaml). **Use this as your
storage root** -- do NOT call a deer-flow path helper yourself.
- host hooks (passed as kwargs to ``from_config``, NOT in backend_config):
``callbacks`` (a ``MemoryCallbacks`` for tracing via ``on_memory_llm_call``),
``should_keep_hidden_message``, ``trace_context_manager``, and
``host_llm_factory``. Consume the ones your backend needs in ``from_config``;
ignore the rest.
- Plus the user's ``config.yaml::memory.backend_config`` keys (your backend's
own knobs: ``model``, ``vector_store``, ``embedder``, thresholds, etc.).
@ -58,11 +56,6 @@ class NoopConfig:
#: ``memory.backend_config.example_option``). Replace with your own.
example_option: str = "default"
#: Host-injected hook (optional). A backend that traces its LLM calls calls
#: ``self._config.tracing_callback(invoke_config, *, thread_id, user_id,
#: trace_id, model_name)`` before invoking. ``None`` = no tracing.
tracing_callback: Callable[..., Any] | None = None
#: Host-injected hook (optional). A backend that filters ``hide_from_ui``
#: messages calls ``self._config.should_keep_hidden_message(additional_kwargs)``
#: -> bool (True = keep despite hide_from_ui). ``None`` = skip all hidden.
@ -72,20 +65,18 @@ class NoopConfig:
def from_backend_config(cls, backend_config: dict[str, Any] | None) -> NoopConfig:
"""Build a config from the ``backend_config`` dict.
Usage in your manager's ``__init__``::
Usage in your manager's ``model_post_init``::
super().__init__(backend_config)
self._config = YourConfig.from_backend_config(backend_config)
self._config = YourConfig.from_backend_config(self.backend_config)
Reads ONLY known keys; unknown keys (including host-injected slots this
backend doesn't consume) are ignored -- so the host can safely inject
shared slots like ``tracing_callback`` for every backend without
breaking ones that don't use them.
Reads ONLY known keys; unknown keys are ignored -- so the host can
safely inject ``storage_path`` into ``backend_config`` for every backend
without breaking ones that don't use it. (Host hooks like tracing arrive
as ``from_config`` kwargs, not in ``backend_config``.)
"""
cfg = dict(backend_config or {})
return cls(
storage_path=str(cfg.get("storage_path") or ""),
example_option=str(cfg.get("example_option", "default")),
tracing_callback=cfg.get("tracing_callback"),
should_keep_hidden_message=cfg.get("should_keep_hidden_message"),
)

130
backend/packages/harness/deerflow/agents/memory/backends/noop/noop_manager.py
View File

@ -14,13 +14,14 @@ Writing a new backend:
1. Copy this folder to ``backends/<yourname>/``.
2. ``config.py``: declare your config knobs + ``from_backend_config`` (parse
``backend_config``; read ``storage_path`` from it, NOT from deer-flow).
3. ``<yourname>_manager.py``: rename the class; ``__init__`` parses
``backend_config`` into your config; implement the 9 ABC methods against
your memory system.
4. (Optional) implement the DeerMem-internal capability methods at the bottom
(``create_fact`` / ``delete_fact`` / ``update_fact`` / ``reload_memory`` /
``warm``) so the host gateway's ``hasattr`` probes find them and the
fact-CRUD / reload / warm-up UI works.
3. ``<yourname>_manager.py``: rename the class; declare your deps as
``PrivateAttr``; ``model_post_init`` parses ``self.backend_config`` into
your config; implement the ABC methods against your memory system.
4. (Optional) override the tier-3 hooks at the bottom (``create_fact`` /
``delete_fact`` / ``update_fact`` / ``reload_memory`` / ``warm``) -- they
have base defaults (``warm``=True, the rest raise ``NotImplementedError``)
so your backend only overrides the ones it supports; callers catch
``NotImplementedError`` for the rest.
5. ``__init__.py``: set ``MANAGER_CLASS = YourManager`` (relative import).
6. ``config.yaml``: ``manager_class: <yourname>``.
@ -38,7 +39,9 @@ disabling memory without touching ``enabled``, and as a baseline.
from __future__ import annotations
from typing import Any
from typing import Any, ClassVar, Literal
from pydantic import PrivateAttr
# ABC contract -- the ONE allowed `from deerflow` in this backend folder.
# Change this single line (to the other agent's MemoryManager) to port.
@ -60,17 +63,35 @@ def _empty_memory() -> dict[str, Any]:
class NoopMemoryManager(MemoryManager):
"""Backend that stores and recalls nothing.
``__init__`` parses ``backend_config`` into a :class:`NoopConfig` purely to
demonstrate the pattern -- noop ignores every field. A real backend reads
its knobs (storage root, model, ...) from ``self._config``.
``model_post_init`` parses ``backend_config`` into a :class:`NoopConfig`
purely to demonstrate the pattern -- noop ignores every field. A real
backend reads its knobs (storage root, model, ...) from ``self._config``.
"""
def __init__(self, backend_config: dict[str, Any] | None = None) -> None:
super().__init__(backend_config)
# Parse backend_config into a typed config. Noop ignores it; a real
# backend uses self._config.* for storage root, model, etc. storage_path
# comes from here (host-injected) -- never import a deer-flow path helper.
self._config: NoopConfig = NoopConfig.from_backend_config(backend_config)
# Parsed config (PrivateAttr: not a validated/serialized field). Noop ignores
# every field; a real backend uses self._config.* for storage root, model,
# etc. storage_path comes from here (host-injected) -- never import a
# deer-flow path helper.
_config: Any = PrivateAttr(default=None)
# noop overrides search() to return [] (its "store/recall nothing" design --
# every read returns empty, never raises), so it is search-capable; the flag
# is True to match the override (the invariant requires flag == override).
supports_search: ClassVar[bool] = True
def model_post_init(self, __context: Any) -> None:
self._config = NoopConfig.from_backend_config(self.backend_config)
@classmethod
def from_config(
cls,
backend_config: dict[str, Any] | None = None,
*,
mode: Literal["middleware", "tool"] = "middleware",
**host_hooks: Any,
) -> NoopMemoryManager:
"""Noop has no dependencies to wire; ignore host_hooks."""
return cls(backend_config=backend_config, mode=mode)
# ── Write ────────────────────────────────────────────────────────────
def add(
@ -124,13 +145,9 @@ class NoopMemoryManager(MemoryManager):
) -> dict[str, Any]:
return _empty_memory()
def delete_memory(
self,
*,
user_id: str | None = None,
agent_name: str | None = None,
) -> None:
return None
# delete_memory / export_memory inherit the base tier-2 default (raise
# NotImplementedError) -- dead contract (zero callers); noop does not
# override them.
def clear_memory(
self,
@ -149,59 +166,24 @@ class NoopMemoryManager(MemoryManager):
) -> dict[str, Any]:
return _empty_memory()
def export_memory(
self,
*,
user_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
return _empty_memory()
# ── Lifecycle ───────────────────────────────────────────────────────
def shutdown_flush(self, timeout: float) -> bool:
"""Nothing is ever queued, so shutdown drain is a clean no-op success."""
return True
# ── Optional DeerMem-internal capabilities (NOT on the ABC) ──────────
# The host gateway discovers these via ``hasattr(manager, "<name>")`` and
# returns 501 when absent. Implement the ones your backend supports so the
# frontend's fact-CRUD / reload / warm-up works. Signatures must match what
# the gateway calls. Uncomment & adapt for your backend:
# ── Tier 3 hooks (inherit base defaults; override if your backend supports) ──
# warm / reload_memory / fact CRUD are tier-3 optional hooks ON the base
# MemoryManager with defaults: warm=True (nothing to warm), the rest raise
# NotImplementedError. Noop does not support fact CRUD / reload, so it
# inherits the defaults (callers catch NotImplementedError -> 501 / fallback).
# Override the ones your backend supports; signatures must match the base
# (see DeerMem for full implementations):
#
# def delete_fact(self, fact_id, *, user_id=None, agent_name=None) -> dict:
# """Delete one memory by id (DELETE /memory/facts/{id})."""
# ... # your_store.delete(fact_id)
# return self.get_memory(user_id=user_id, agent_name=agent_name)
#
# def create_fact(self, content: str, category: str = "context",
# confidence: float = 0.5, *,
# user_id: str | None = None,
# agent_name: str | None = None,
# ) -> tuple[dict, str | None]:
# """Manually add one memory (POST /memory/facts).
#
# Returns ``(memory_data, fact_id)`` -- NOT a bare dict. ``content`` is
# positional (the memory_add tool passes it positionally); ``fact_id``
# is None when a storage cap (e.g. max_facts) evicted the just-added
# fact, so the caller reports "not stored" instead of a dangling id.
# Signatures must match what the gateway/client/tools call (see DeerMem).
# """
# ... # your_store.add(content); fact_id = your_store.last_id()
# return self.get_memory(user_id=user_id, agent_name=agent_name), fact_id
#
# def update_fact(self, *, fact_id, content=None, category=None,
# confidence=None, user_id=None, agent_name=None) -> dict:
# """Update one memory's text by id (PATCH /memory/facts/{id})."""
# ... # your_store.update(fact_id, content)
# return self.get_memory(user_id=user_id, agent_name=agent_name)
#
# def reload_memory(self, *, user_id=None, agent_name=None) -> dict:
# """Drop caches & re-read storage (POST /memory/reload).
# If your backend has no cache, just delegate to get_memory(...)."""
# return self.get_memory(user_id=user_id, agent_name=agent_name)
#
# def warm(self) -> None:
# """Heavy one-time init at gateway startup (e.g. load a tokenizer).
# Probed via hasattr; absent = skipped. Keep it fast (host guards it
# with a timeout)."""
# ...
# def create_fact(self, content, category="context", confidence=0.5, *,
# agent_name=None, user_id=None) -> tuple[dict, str | None]:
# ... # return (memory_data, fact_id); fact_id=None if a cap evicted it
# def delete_fact(self, fact_id, *, agent_name=None, user_id=None) -> dict: ...
# def update_fact(self, fact_id, content=None, category=None, confidence=None,
# *, agent_name=None, user_id=None) -> dict: ...
# def reload_memory(self, *, user_id=None, agent_name=None) -> dict: ...
# def warm(self) -> bool | None: ... # default None (nothing to warm); override for heavy one-time init

492
backend/packages/harness/deerflow/agents/memory/manager.py
View File

@ -20,10 +20,12 @@ import importlib
import logging
import os
import threading
from abc import ABC, abstractmethod
from abc import abstractmethod
from pathlib import Path
from types import ModuleType
from typing import Any
from typing import Any, ClassVar, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from deerflow.config.memory_config import get_memory_config
@ -41,6 +43,27 @@ _backends_cache: dict[str, type[MemoryManager]] | None = None
_manager_lock = threading.Lock()
class MemoryCallbacks:
"""Observability hooks for memory backends. Default implementations are
no-ops; override the ones you need. The pre-LLM-call hook
``on_memory_llm_call`` mutates ``invoke_config`` before the LLM call so a
tracer (e.g. langfuse) emits a span at the LLM boundary. (More hooks --
post-extract / search / inject / error -- can be added when callers need
them.)"""
def on_memory_llm_call(
self,
invoke_config: dict[str, Any],
*,
thread_id: str | None,
user_id: str | None,
trace_id: str | None,
model_name: str | None,
) -> None:
"""Pre-LLM-call: mutate ``invoke_config`` (e.g. merge trace metadata)
before the backend invokes the model. Default: no-op."""
class MemoryManagerError(RuntimeError):
"""Backend-neutral base error exposed at the MemoryManager boundary."""
@ -53,8 +76,19 @@ class MemoryCorruptionError(MemoryManagerError):
"""Persisted memory cannot be read safely."""
class MemoryManager(ABC):
"""Backend-neutral memory manager contract (9 methods).
class MemoryManager(BaseModel):
"""Backend-neutral memory manager contract.
A pydantic ``BaseModel`` (not a bare ``ABC``) so the contract gains field
validation + serialization for free and shares the pydantic v2 type system
with backend configs (e.g. ``DeerMemConfig``). Subclasses still MUST
implement the ``@abstractmethod``s -- pydantic's ``ModelMetaclass`` derives
from ``ABCMeta``, so unimplemented abstractmethods raise ``TypeError`` at
instantiation (memory is persistent state; a backend missing ``add`` /
``get_context`` is a severe bug, caught at construction). Backend-private
dependencies (storage / llm / queue / ...) are NOT fields -- they are
``PrivateAttr`` set in ``model_post_init`` (or ``from_config``), kept out of
validation / serialization.
Memories are bucketed per ``(agent_name, user_id)``; ``thread_id`` aligns
with the deer-flow conversation thread. The contract is deliberately
@ -70,21 +104,80 @@ class MemoryManager(ABC):
private concern (not on the contract).
- No facts-model assumption: a backend need not store "facts" at all.
Methods marked *stub* are part of the contract but have no caller yet in
this phase; DeerMem raises ``NotImplementedError`` for them, a future
backend (or a later DeerMem ``core/`` module) may implement them for real.
Methods are tiered: tier-1 (``add`` / ``get_context``) are ``@abstractmethod``;
tier-2 management ops and tier-3 optional hooks carry defaults (``raise
NotImplementedError`` or a no-op) so a backend implements only what it
supports. ``delete_memory`` / ``export_memory`` are dead contract (zero
callers; ``/memory/export`` routes via ``get_memory``) kept available via the
default raise.
"""
def __init__(self, backend_config: dict[str, Any] | None = None) -> None:
"""Receive backend-private config (the factory passes ``backend_config``).
model_config = ConfigDict(arbitrary_types_allowed=True)
Default stores the raw dict; backends that need to parse it (e.g. DeerMem
into a ``DeerMemConfig``) override ``__init__``. Backends that ignore
private config (e.g. noop) inherit this unchanged.
# Backend-private config (factory passes it through). Backends that need to
# parse it (DeerMem -> DeerMemConfig) do so in model_post_init / from_config.
# None is coerced to {} so zero-config ``Backend(backend_config=None)`` stays
# valid (BaseModel would otherwise reject None for a dict field).
backend_config: dict[str, Any] = Field(default_factory=dict)
# Operation mode mirrors host ``MemoryConfig.mode`` ("middleware" | "tool");
# the factory passes ``cfg.mode``. The invariant validator requires
# tool-mode backends to support search.
mode: Literal["middleware", "tool"] = "middleware"
# Observability callbacks (optional; a ``MemoryCallbacks`` instance). The
# factory injects ``LangfuseMemoryCallbacks`` so memory-LLM calls surface in
# langfuse; None = no callbacks (direct construction / standalone). Backends
# pass this to their LLM path and call ``on_memory_llm_call`` before invoke.
callbacks: MemoryCallbacks | None = None
@field_validator("backend_config", mode="before")
@classmethod
def _coerce_backend_config(cls, value: Any) -> dict[str, Any]:
"""Accept None (zero-config) as an empty dict; leave dicts untouched."""
return value or {}
# Search capability flag (ClassVar, not a field): set True iff the backend
# overrides search(). The invariant validator checks the flag MATCHES whether
# search() is actually overridden (type(self).search is not MemoryManager.search),
# so the two can't drift -- required for mode="tool" (the agent calls
# memory_search in tool mode, so a non-search backend is a misconfiguration
# that fails fast at instantiation rather than silently returning empty
# results). Default False: a new backend must explicitly opt in to tool mode.
supports_search: ClassVar[bool] = False
@model_validator(mode="after")
def _check_invariants(self) -> MemoryManager:
"""Cross-field invariants every backend must satisfy at instantiation.
Fires on the factory path AND when a backend is constructed directly
(bypassing the factory), since it lives on the base model. DeerMem-private
invariants (e.g. storage_path is a directory) stay on ``DeerMemConfig``.
``supports_search`` (ClassVar flag) must match whether ``search()`` is
actually overridden, so the declarative flag can't drift from the
implementation -- a backend that overrides ``search()`` but forgets
``supports_search = True`` (or sets the flag without overriding) is a bug
caught at instantiation, not a misleading tool-mode rejection or a runtime
``NotImplementedError`` on the first ``memory_search`` call.
"""
self._backend_config = backend_config
search_overridden = type(self).search is not MemoryManager.search
if type(self).supports_search != search_overridden:
raise ValueError(
f"{type(self).__name__}.supports_search={type(self).supports_search} "
f"is inconsistent with search(): search() is "
f"{'overridden' if search_overridden else 'inherited (not implemented)'}. "
f"Set supports_search={search_overridden} on the backend to match."
)
if self.mode == "tool" and not search_overridden:
raise ValueError(
f"memory mode='tool' requires a backend that implements search(), but {type(self).__name__} does not override search(). Use mode='middleware' or a backend that overrides search() (and sets supports_search=True)."
)
return self
# ── Write ────────────────────────────────────────────────────────────
# ── Tier 1: @abstractmethod ─────────────────────────────────────────
# Every backend MUST implement these (write + read-inject are the backend's
# fundamental duties). Missing one is a severe bug (memory is persistent
# state) -- @abstractmethod catches it at instantiation. noop implements
# them as no-op / "".
@abstractmethod
def add(
self,
@ -106,22 +199,6 @@ class MemoryManager(ABC):
trace_id: Request trace id captured for memory-LLM tracing.
"""
@abstractmethod
def add_nowait(
self,
thread_id: str,
messages: list[Any],
*,
agent_name: str | None = None,
user_id: str | None = None,
) -> None:
"""Queue a conversation for *immediate* memory update (emergency flush).
Used right before summarization removes messages from state, so the
content is captured instead of lost.
"""
# ── Read ─────────────────────────────────────────────────────────────
@abstractmethod
def get_context(
self,
@ -138,7 +215,32 @@ class MemoryManager(ABC):
``backend_config`` at construction), NOT a host config on this method.
"""
@abstractmethod
# ── Tier 2: management ops with defaults ────────────────────────────
# A backend that does not support an operation inherits the default (raise
# ``NotImplementedError``) instead of having to write the raise. ``add_nowait``
# defaults to delegating to ``add`` (a backend without a debounce queue has no
# "immediate" vs "queued" distinction); ``shutdown_flush`` defaults to True (a
# backend without a buffer has nothing to drain) so the host can call it
# unconditionally. ``delete_memory`` / ``export_memory`` are dead contract
# (zero callers; /memory/export routes via get_memory) -- the default raise
# keeps them available without forcing backends to implement.
def add_nowait(
self,
thread_id: str,
messages: list[Any],
*,
agent_name: str | None = None,
user_id: str | None = None,
) -> None:
"""Queue a conversation for *immediate* memory update (emergency flush).
Used right before summarization removes messages from state, so the
content is captured instead of lost. Default: delegate to :meth:`add`
(backends without a debounce queue override to enqueue with nowait
priority).
"""
self.add(thread_id, messages, agent_name=agent_name, user_id=user_id)
def search(
self,
query: str,
@ -151,42 +253,45 @@ class MemoryManager(ABC):
"""Search the bucket's memory for facts matching ``query``; return up to
``top_k`` ranked by relevance. ``category`` (optional) filters BEFORE the
``top_k`` slice so a category-scoped search is not starved by other
categories' higher-ranked facts."""
categories' higher-ranked facts. Default: unsupported (raise); backends
with retrieval override AND set ``supports_search = True`` (required for
``mode='tool'``)."""
raise NotImplementedError(f"search not supported by {type(self).__name__}")
# ── Manage ───────────────────────────────────────────────────────────
@abstractmethod
def get_memory(
self,
*,
user_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
"""Return the full memory document for the bucket."""
"""Return the full memory document for the bucket. Default: unsupported."""
raise NotImplementedError(f"get_memory not supported by {type(self).__name__}")
@abstractmethod
def delete_memory(
self,
*,
user_id: str | None = None,
agent_name: str | None = None,
) -> None:
"""Delete the entire memory document for the bucket. *stub* this phase."""
"""Delete the entire memory document for the bucket. Default: unsupported
(dead contract -- zero callers)."""
raise NotImplementedError(f"delete_memory not supported by {type(self).__name__}")
@abstractmethod
def clear_memory(
self,
*,
user_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
"""Clear memory and return the empty compatibility document.
"""Clear the bucket's memory; return the cleared (now-empty) document.
``agent_name=None`` means all memory owned by the user. An explicit
agent name clears only that agent's memory and must preserve shared
user-level summaries.
user-level summaries. Default: unsupported (raise
``NotImplementedError``); backends that support clearing override.
"""
raise NotImplementedError(f"clear_memory not supported by {type(self).__name__}")
@abstractmethod
def import_memory(
self,
memory_data: dict[str, Any],
@ -194,19 +299,20 @@ class MemoryManager(ABC):
user_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
"""Import a memory document into the bucket; return the merged result."""
"""Import a memory document into the bucket; return the merged result.
Default: unsupported."""
raise NotImplementedError(f"import_memory not supported by {type(self).__name__}")
@abstractmethod
def export_memory(
self,
*,
user_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
"""Export the memory document for the bucket. *stub* this phase (no caller yet)."""
"""Export the memory document for the bucket. Default: unsupported (dead
contract -- zero callers; /memory/export routes via get_memory)."""
raise NotImplementedError(f"export_memory not supported by {type(self).__name__}")
# ── Lifecycle ───────────────────────────────────────────────────────
@abstractmethod
def shutdown_flush(self, timeout: float) -> bool:
"""Best-effort bounded drain of pending updates on graceful shutdown.
@ -227,10 +333,152 @@ class MemoryManager(ABC):
Returns ``True`` if the drain genuinely finished within ``timeout``
(buffer empty, no worker still running, no exception); ``False`` on
timeout or failure (the caller logs a warning and proceeds to exit --
any unfinished tail is dropped, strictly better than no flush). A
backend with no pending work (or no buffer at all) returns ``True``
immediately, so the host may call this unconditionally when memory is
enabled without gating on backend-private queue state.
any unfinished tail is dropped, strictly better than no flush). Default:
``True`` (a backend with no buffer has nothing to drain), so the host may
call this unconditionally when memory is enabled; backends with a debounce
queue override to flush within ``timeout``.
"""
return True
# ── Tier 3: optional hooks ──────────────────────────────────────────
# A-class: agent-side has real callers (startup warm-up, manual reload, fact
# CRUD). Previously reached via ``hasattr`` probing; now contracted with
# defaults so callers invoke directly and catch ``NotImplementedError``.
# ``warm`` defaults to None (nothing to warm); the rest default to raise.
def warm(self) -> bool | None:
"""Pre-warm backend resources at startup (e.g. the tiktoken encoding
cache). Probed off the event loop by the Gateway lifespan.
Return contract (tri-state so the host logs accurately):
* ``True`` -- warmed successfully (or already cached / unnecessary).
* ``False`` -- warming was attempted and failed (host falls back).
* ``None`` -- this backend has nothing to warm (the default). The
host logs a "skipping" message instead of the misleading "warmed
successfully", so a non-DeerMem backend doesn't claim a tiktoken
cache it never touched.
Backends with heavy one-time init override and return ``True``/``False``.
"""
return None
def reload_memory(
self,
*,
user_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
"""Drop the cached memory document and reload from storage. Default:
unsupported (callers fall back to :meth:`get_memory`). Backends with a
cache override."""
raise NotImplementedError(f"reload_memory not supported by {type(self).__name__}")
def create_fact(
self,
content: str,
category: str = "context",
confidence: float = 0.5,
*,
agent_name: str | None = None,
user_id: str | None = None,
) -> tuple[dict[str, Any], str | None]:
"""Manually add one fact. Returns ``(memory_data, fact_id)`` -- ``fact_id``
is None when a storage cap evicted the just-added fact. Default: unsupported."""
raise NotImplementedError(f"create_fact not supported by {type(self).__name__}")
def delete_fact(
self,
fact_id: str,
*,
agent_name: str | None = None,
user_id: str | None = None,
) -> dict[str, Any]:
"""Delete one fact by id. Default: unsupported."""
raise NotImplementedError(f"delete_fact not supported by {type(self).__name__}")
def update_fact(
self,
fact_id: str,
content: str | None = None,
category: str | None = None,
confidence: float | None = None,
*,
agent_name: str | None = None,
user_id: str | None = None,
) -> dict[str, Any]:
"""Update one fact by id (preserving omitted fields). Default: unsupported."""
raise NotImplementedError(f"update_fact not supported by {type(self).__name__}")
# B-class: no agent-side caller yet -- signatures only, for future scenarios.
# Default no-op so callers can invoke unconditionally without gating. (The
# self-serving hooks on_delegation / on_session_end / on_memory_write are
# deliberately NOT contracted: no caller, no event source, or subsumed by
# the callbacks field.)
def on_pre_compress(self, messages: list[Any]) -> str:
"""Memory -> compressor feedback (future memory-driven summary
enrichment). Returns text to inject into the compression prompt
(default: none)."""
return ""
def on_turn_start(self, turn_number: int, message: Any, **kwargs: Any) -> None:
"""Turn-start nudge (future background review). Default: no-op."""
return None
# ── Async (speculative) ──────────────────────────────────────────────
# Interface placeholders so a future async LLM client can override without
# changing the contract. Defaults delegate to the sync methods (no
# concurrency benefit -- the real LLM call stays sync); callers today use
# the sync path.
async def aadd(
self,
thread_id: str,
messages: list[Any],
*,
agent_name: str | None = None,
user_id: str | None = None,
trace_id: str | None = None,
) -> None:
return self.add(thread_id, messages, agent_name=agent_name, user_id=user_id, trace_id=trace_id)
async def aget_context(
self,
user_id: str | None,
*,
agent_name: str | None = None,
thread_id: str | None = None,
) -> str:
return self.get_context(user_id, agent_name=agent_name, thread_id=thread_id)
async def asearch(
self,
query: str,
top_k: int = 5,
*,
user_id: str | None = None,
agent_name: str | None = None,
category: str | None = None,
) -> list[dict[str, Any]]:
return self.search(query, top_k, user_id=user_id, agent_name=agent_name, category=category)
# ── Construction ─────────────────────────────────────────────────────
@classmethod
@abstractmethod
def from_config(
cls,
backend_config: dict[str, Any],
*,
mode: Literal["middleware", "tool"] = "middleware",
**host_hooks: Any,
) -> MemoryManager:
"""Build a fully-wired instance from backend config + host-provided hooks.
The factory calls this instead of constructing the class directly, so
each backend owns its own assembly (parse config, wire dependencies,
consume the host hooks it needs). Adding a backend = implement
``from_config``; the factory stays unchanged. ``host_hooks`` carries
host-provided callables/values (tracing, hidden-message filter,
trace-context manager, a host-llm factory); a backend that needs none
of them (e.g. noop) ignores them and returns
``cls(backend_config=backend_config, mode=mode)``.
"""
@ -328,47 +576,51 @@ def _resolve_manager_class(manager_class: str) -> type[MemoryManager]:
)
# ── Host-default hooks (injected into backend_config by the factory) ──────
# ── Host-default hook providers (passed to from_config by the factory) ────
#
# DeerMemConfig declares ``tracing_callback`` and ``should_keep_hidden_message``
# as optional, host-agnostic slots (default ``None``). The portable package
# never names a deer-flow concept, so the host fills these slots HERE -- in the
# factory, which is host code outside ``backends/deermem/``. Backends whose
# config schema declares these slots (DeerMem) consume them via
# ``from_backend_config``'s known-field filter; others (e.g. noop) ignore
# them. An explicit value in ``backend_config`` (set programmatically) takes
# precedence and is left untouched.
# These callables are the host's defaults for the slots a backend may consume
# (tracing, hidden-message filtering, trace-context binding, a host default
# LLM). The portable backend package never names a deer-flow concept; the host
# supplies them HERE (host code outside ``backends/deermem/``). The factory
# passes them to ``cls.from_config(..., **host_hooks)``; each backend's
# ``from_config`` consumes the ones it needs (DeerMem does; noop ignores them).
# An explicit value in ``backend_config`` (set programmatically) takes
# precedence and is left untouched (from_config's merge skips present keys).
#
# Imports are lazy (matching the ``runtime_home`` precedent) so this module
# stays cheap to import and so another agent vendoring the contract only has
# to edit these two helpers, not the top-level imports.
def _host_default_tracing_callback(
invoke_config: dict[str, Any],
*,
thread_id: str | None,
user_id: str | None,
trace_id: str | None,
model_name: str | None,
) -> None:
"""deer-flow default for DeerMem's ``tracing_callback`` slot.
# to edit these helpers, not the top-level imports.
class LangfuseMemoryCallbacks(MemoryCallbacks):
"""Host default callbacks: emit langfuse spans at the memory-LLM boundary.
Merges Langfuse trace metadata into ``invoke_config`` (no-op when
Langfuse is not an enabled tracing provider). Maps DeerMem's ``trace_id``
onto ``inject_langfuse_metadata``'s ``deerflow_trace_id`` kwarg -- the
name mismatch that previously made memory LLM tracing silently TypeError
is bridged here, at the host seam, so the portable package is untouched.
Implements ``on_memory_llm_call`` (pre-LLM-call) by merging langfuse trace
metadata into ``invoke_config`` -- identical to the former
``_host_default_tracing_callback`` (same signature, same timing, same
mutation), repackaged as a callbacks method so the langfuse binding lives in
host code and the portable backend package never names langfuse. No-op when
langfuse is not an enabled tracing provider.
"""
from deerflow.tracing import inject_langfuse_metadata
inject_langfuse_metadata(
invoke_config,
thread_id=thread_id,
user_id=user_id,
assistant_id="memory_agent",
model_name=model_name,
environment=os.environ.get("DEER_FLOW_ENV") or os.environ.get("ENVIRONMENT"),
deerflow_trace_id=trace_id,
)
def on_memory_llm_call(
self,
invoke_config: dict[str, Any],
*,
thread_id: str | None,
user_id: str | None,
trace_id: str | None,
model_name: str | None,
) -> None:
from deerflow.tracing import inject_langfuse_metadata
inject_langfuse_metadata(
invoke_config,
thread_id=thread_id,
user_id=user_id,
assistant_id="memory_agent",
model_name=model_name,
environment=os.environ.get("DEER_FLOW_ENV") or os.environ.get("ENVIRONMENT"),
deerflow_trace_id=trace_id,
)
def _host_default_should_keep_hidden_message(additional_kwargs: Any) -> bool:
@ -390,7 +642,7 @@ def _host_default_llm() -> Any:
Builds the host's default chat model (``create_chat_model(name=None)`` ->
app default, ``attach_tracing=True`` so memory LLM calls surface in langfuse
via the metadata ``tracing_callback`` merges), mirroring pre-abstraction
via the callbacks' ``on_memory_llm_call`` metadata merge), mirroring pre-abstraction
``model_name: null``. Returns ``None`` if no model is available (no models
configured) so DeerMem no-ops extraction with a clear error rather than
crashing startup.
@ -404,6 +656,27 @@ def _host_default_llm() -> Any:
return None
def _collect_host_hooks() -> dict[str, Any]:
"""Provide host hook callables for backends to consume in ``from_config``.
The factory is a hook *provider*; each backend's ``from_config`` is the
*consumer* that decides which hooks to use (so adding a backend does not
change this factory). ``host_llm`` is provided as a factory callable
(``host_llm_factory``) rather than a built instance so a backend only
builds the host default model when it actually needs one (i.e. has no model
of its own) -- building an unused default on every startup would waste
time. The others are direct values (cheap function refs).
"""
from deerflow.trace_context import request_trace_context
return {
"callbacks": LangfuseMemoryCallbacks(),
"should_keep_hidden_message": _host_default_should_keep_hidden_message,
"trace_context_manager": request_trace_context,
"host_llm_factory": _host_default_llm,
}
# ── Singleton factory ─────────────────────────────────────────────────────
def get_memory_manager() -> MemoryManager:
"""Return the singleton :class:`MemoryManager` for the active config.
@ -447,44 +720,17 @@ def get_memory_manager() -> MemoryManager:
from deerflow.config.runtime_paths import runtime_home
backend_config["storage_path"] = str((Path(runtime_home()) / backend_config["storage_path"]).resolve())
# Guard: DeerMem treats storage_path as a root DIRECTORY (per-user memory
# under {storage_path}/users/{uid}/memory.json). A file-style value (e.g. a
# leftover .json file from the pre-abstraction file-path semantics) would
# make FileMemoryStorage.save's mkdir(parents=True) raise NotADirectoryError,
# caught as OSError -> silent write failure. Fail loud at startup instead
# (memory is persistent state -- a wrong root is a data-integrity footgun).
_resolved_storage_path = Path(backend_config["storage_path"])
if _resolved_storage_path.is_file():
raise ValueError(
f"memory.backend_config.storage_path={backend_config['storage_path']!r} "
f"resolves to an existing file {_resolved_storage_path}; DeerMem treats "
f"storage_path as a root DIRECTORY (per-user memory under "
f"{{storage_path}}/users/{{uid}}/memory.json). Point it at a directory."
)
# Host-default hooks: callables cannot come from YAML, so the host
# injects them here. DeerMem consumes them (known config fields);
# noop ignores them (unknown-field filter in from_backend_config).
# An explicit value (incl. ``null`` in YAML) takes precedence -> the
# host default is only filled when the key is absent.
if "tracing_callback" not in backend_config:
backend_config["tracing_callback"] = _host_default_tracing_callback
if "should_keep_hidden_message" not in backend_config:
backend_config["should_keep_hidden_message"] = _host_default_should_keep_hidden_message
# Zero-config LLM: when no memory model is configured, inject the host's
# default chat model so memory extraction works out of the box (mirrors
# pre-abstraction `model_name: null` -> app default). DeerMem prefers
# host_llm over build_llm(model); other backends ignore the slot.
model_cfg = backend_config.get("model")
if not (isinstance(model_cfg, dict) and model_cfg.get("model")) and "host_llm" not in backend_config:
backend_config["host_llm"] = _host_default_llm()
# Restore structured-log trace correlation on the memory-update worker
# thread (Timer / executor): bind trace_id into the request-trace
# ContextVar. A None trace_id is left unbound by the updater's guard.
if "trace_context_manager" not in backend_config:
from deerflow.trace_context import request_trace_context
backend_config["trace_context_manager"] = request_trace_context
_memory_manager = cls(backend_config=backend_config)
# storage_path-is-a-file guard lives on DeerMemConfig.model_validator
# now (DeerMem-private semantics; fires even when the factory bypassed).
# Host hook providers: the factory supplies these as kwargs; each
# backend's from_config decides which to consume (the factory stays
# backend-agnostic -- it no longer knows which hooks a backend needs).
# An explicit value in backend_config still wins (from_config's merge
# skips keys already present).
host_hooks = _collect_host_hooks()
# ``mode`` mirrors host MemoryConfig.mode so the invariant validator
# (mode=="tool" requires search) can fire on the factory path too.
_memory_manager = cls.from_config(backend_config, mode=cfg.mode, **host_hooks)
logger.info("Memory manager resolved: %s (manager_class=%r)", cls.__name__, manager_class)
return _memory_manager

57
backend/packages/harness/deerflow/agents/memory/tools.py
View File

@ -9,11 +9,11 @@ its own persistent memory: it decides what to remember, when to
search, and when to update or remove stale facts.
Backend-agnostic: every tool goes through the ``MemoryManager`` ABC
(:func:`get_memory_manager`) -- ``search``/``get_memory`` are on the ABC;
``create_fact``/``update_fact``/``delete_fact`` are backend-internal
capabilities reached via attribute access (absent -> the tool returns a
JSON ``error`` instead of crashing). So tool mode works for any backend
that exposes those ops (DeerMem does; noop returns empty/errors).
(:func:`get_memory_manager`) -- ``search``/``get_memory`` are tier-2 methods;
``create_fact``/``update_fact``/``delete_fact`` are tier-3 hooks with a default
``raise NotImplementedError`` (unsupported -> the tool catches it and returns a
JSON ``error`` instead of crashing). So tool mode works for any backend that
overrides those ops (DeerMem does; noop inherits the raises -> errors).
"""
import json
@ -124,19 +124,20 @@ def memory_add_tool(
if any(_memory_content_key(str(fact.get("content", ""))) == content_key for fact in existing_facts):
return json.dumps({"error": "Duplicate fact"})
create = getattr(manager, "create_fact", None)
if not callable(create):
return json.dumps({"error": f"memory backend {type(manager).__name__} does not support create_fact"})
# create_fact returns (memory_data, fact_id) -- use the id directly rather
# than re-deriving it by content matching (which would couple the tool to
# the backend's content normalization and could misreport a storage cap).
_memory_data, fact_id = create(
normalized_content,
category=category,
confidence=confidence,
agent_name=agent_name,
user_id=user_id,
)
# Unsupported backends raise NotImplementedError (tier-3 default) -> JSON error.
try:
_memory_data, fact_id = manager.create_fact(
normalized_content,
category=category,
confidence=confidence,
agent_name=agent_name,
user_id=user_id,
)
except NotImplementedError:
return json.dumps({"error": f"memory backend {type(manager).__name__} does not support create_fact"})
if fact_id is None:
# max_facts cap kept higher-confidence facts and evicted the new one;
# the fact was not stored -- report honestly instead of a dangling id.
@ -182,17 +183,17 @@ def memory_update_tool(
agent_name, user_id = _resolve_scope(runtime)
try:
manager = get_memory_manager()
update = getattr(manager, "update_fact", None)
if not callable(update):
try:
manager.update_fact(
fact_id,
content=content,
category=category,
confidence=confidence,
agent_name=agent_name,
user_id=user_id,
)
except NotImplementedError:
return json.dumps({"error": f"memory backend {type(manager).__name__} does not support update_fact"})
update(
fact_id,
content=content,
category=category,
confidence=confidence,
agent_name=agent_name,
user_id=user_id,
)
return json.dumps({"fact_id": fact_id, "status": "updated"})
except KeyError:
return json.dumps({"error": f"Fact not found: {fact_id}"})
@ -220,10 +221,10 @@ def memory_delete_tool(runtime: Runtime, fact_id: str) -> str:
agent_name, user_id = _resolve_scope(runtime)
try:
manager = get_memory_manager()
delete = getattr(manager, "delete_fact", None)
if not callable(delete):
try:
manager.delete_fact(fact_id, agent_name=agent_name, user_id=user_id)
except NotImplementedError:
return json.dumps({"error": f"memory backend {type(manager).__name__} does not support delete_fact"})
delete(fact_id, agent_name=agent_name, user_id=user_id)
return json.dumps({"fact_id": fact_id, "status": "deleted"})
except KeyError:
return json.dumps({"error": f"Fact not found: {fact_id}"})

24
backend/packages/harness/deerflow/client.py
View File

@ -1302,14 +1302,24 @@ class DeerFlowClient:
Returns:
The reloaded memory data dict.
Backends without a reload concept (e.g. noop) fall back to
``get_memory``; a backend that exposes neither (a minimal ``add`` +
``get_context`` backend) raises ``NotImplementedError`` so the caller
sees a clean unsupported-op error instead of an uncaught propagation.
"""
from deerflow.agents.memory import get_memory_manager
manager = get_memory_manager()
if hasattr(manager, "reload_memory"):
return manager.reload_memory(user_id=get_effective_user_id())
# Non-DeerMem backends have no reload concept; return current memory.
return manager.get_memory(user_id=get_effective_user_id())
user_id = get_effective_user_id()
try:
return manager.reload_memory(user_id=user_id)
except NotImplementedError:
pass # no reload concept; fall back to current memory below
try:
return manager.get_memory(user_id=user_id)
except NotImplementedError:
raise NotImplementedError(f"reload_memory not supported by memory backend {type(manager).__name__}: implements neither reload_memory nor get_memory") from None
def clear_memory(self) -> dict:
"""Clear all persisted memory data."""
@ -1322,8 +1332,6 @@ class DeerFlowClient:
from deerflow.agents.memory import get_memory_manager
manager = get_memory_manager()
if not hasattr(manager, "create_fact"):
raise NotImplementedError(f"create_fact not supported by memory backend '{type(manager).__name__}'")
memory_data, fact_id = manager.create_fact(content=content, category=category, confidence=confidence, user_id=get_effective_user_id())
if fact_id is None:
raise ValueError("Fact was not stored because memory.max_facts kept higher-confidence facts")
@ -1334,8 +1342,6 @@ class DeerFlowClient:
from deerflow.agents.memory import get_memory_manager
manager = get_memory_manager()
if not hasattr(manager, "delete_fact"):
raise NotImplementedError(f"delete_fact not supported by memory backend '{type(manager).__name__}'")
return manager.delete_fact(fact_id, user_id=get_effective_user_id())
def update_memory_fact(
@ -1349,8 +1355,6 @@ class DeerFlowClient:
from deerflow.agents.memory import get_memory_manager
manager = get_memory_manager()
if not hasattr(manager, "update_fact"):
raise NotImplementedError(f"update_fact not supported by memory backend '{type(manager).__name__}'")
return manager.update_fact(
fact_id=fact_id,
content=content,

5
backend/samples/other_agent_demo/README.md
View File

@ -51,6 +51,7 @@ backend_config:
base_url: https://api.openai.com/v1
debounce_seconds: 30
max_facts: 100
# tracing_callback / should_keep_hidden_message: programmatic callables
# (cannot come from YAML); set them on the config object in code if desired.
# callbacks / should_keep_hidden_message: programmatic (cannot come from
# YAML); the host factory injects them via from_config. Set programmatically
# only if overriding the host defaults.
```

13
backend/tests/test_client.py
View File

@ -1771,6 +1771,19 @@ class TestMemoryManagement:
result = client.reload_memory()
assert result == data
def test_reload_memory_raises_clean_error_when_read_also_unsupported(self, client):
"""A minimal backend (only add + get_context) exposes neither reload nor
get_memory; reload_memory surfaces a clean NotImplementedError instead of
an uncaught propagation from the fallback (mirrors the router's 501)."""
mock_mgr = MagicMock()
mock_mgr.reload_memory.side_effect = NotImplementedError("reload not supported")
mock_mgr.get_memory.side_effect = NotImplementedError("get_memory not supported")
with patch("deerflow.agents.memory.get_memory_manager", return_value=mock_mgr):
with pytest.raises(NotImplementedError, match="implements neither"):
client.reload_memory()
mock_mgr.reload_memory.assert_called_once()
mock_mgr.get_memory.assert_called_once()
def test_clear_memory(self, client):
data = {"version": "1.0", "facts": []}
mock_mgr = MagicMock()

135
backend/tests/test_deermem_self_contained.py
View File

@ -1,7 +1,7 @@
"""Phase-2 (self-contained DeerMem) tests.
Covers: DI construction (owns storage/updater/queue/llm), zero-config defaults,
``trace_id`` threading to the optional ``tracing_callback``, langfuse being
``trace_id`` threading to the optional ``callbacks`` hook, langfuse being
optional, ``hide_from_ui`` default-skip + hook-keep, empty ``storage_class``
(portable default), and portability -- ``backends/deermem/`` has exactly one
``from deerflow`` line (the ABC contract) and can be vendored into another agent
@ -25,6 +25,7 @@ from deerflow.agents.memory.backends.deermem.deermem.core.message_processing imp
)
from deerflow.agents.memory.backends.deermem.deermem.core.storage import FileMemoryStorage
from deerflow.agents.memory.backends.deermem.deermem.core.updater import _trim_facts_to_max
from deerflow.agents.memory.manager import MemoryCallbacks
@pytest.fixture
@ -46,8 +47,8 @@ class _FakeLLM:
return type("R", (), {"content": self._payload})()
def _deermem_with_fake_llm(backend_config=None, payload=None) -> DeerMem:
dm = DeerMem(backend_config=backend_config)
def _deermem_with_fake_llm(backend_config=None, payload=None, callbacks=None) -> DeerMem:
dm = DeerMem(backend_config=backend_config, callbacks=callbacks)
fake = _FakeLLM(payload)
dm._llm = fake
dm._updater._llm = fake
@ -63,6 +64,37 @@ def test_di_construction_owns_dependencies():
assert dm._queue._updater is dm._updater
def test_from_config_keeps_backend_config_pure_of_injected_hooks(deermem_data_dir):
"""Host hooks (should_keep_hidden_message / trace_context_manager / host_llm)
arrive as from_config kwargs and are parsed into DeerMemConfig
(self._config, PrivateAttr); the instance's backend_config field stays the
pure data the host passed (no callables / LLM), so it remains serializable
and matches the README contract ("host hooks ... NOT in backend_config")."""
def _keep(ak):
return False
trace_cm = object() # sentinel; trace_context_manager is typed Any
fake_llm = _FakeLLM()
dm = DeerMem.from_config(
backend_config={"storage_path": str(deermem_data_dir), "max_facts": 20},
mode="middleware",
should_keep_hidden_message=_keep,
trace_context_manager=trace_cm,
host_llm_factory=lambda: fake_llm,
callbacks=None,
)
# hooks reached DeerMemConfig (PrivateAttr) -- wired, not lost
assert dm._config.should_keep_hidden_message is _keep
assert dm._config.trace_context_manager is trace_cm
assert dm._config.host_llm is fake_llm
# backend_config field is the pure original data (no injected hooks)
assert dm.backend_config == {"storage_path": str(deermem_data_dir), "max_facts": 20}
assert "should_keep_hidden_message" not in dm.backend_config
assert "trace_context_manager" not in dm.backend_config
assert "host_llm" not in dm.backend_config
def test_zero_config_defaults_run_non_llm_ops(deermem_data_dir):
dm = DeerMem(backend_config=None) # zero config
assert dm._llm is None # no model -> no LLM
@ -120,13 +152,15 @@ def test_import_empty_summary_sections_replace_existing_summaries_with_complete_
}
def test_trace_id_threads_through_to_tracing_callback(deermem_data_dir):
def test_trace_id_threads_through_to_callbacks(deermem_data_dir):
"""trace_id reaches the pre-LLM-call callbacks hook (on_memory_llm_call)."""
calls = []
def tracer(cfg, *, thread_id, user_id, trace_id, model_name):
calls.append((thread_id, trace_id, model_name))
class _RecordingCallbacks(MemoryCallbacks):
def on_memory_llm_call(self, invoke_config, *, thread_id, user_id, trace_id, model_name):
calls.append((thread_id, trace_id, model_name))
dm = _deermem_with_fake_llm({"tracing_callback": tracer, "model": {"provider": "openai", "model": "gpt-x", "api_key": "k", "base_url": "u"}})
dm = _deermem_with_fake_llm({"model": {"provider": "openai", "model": "gpt-x", "api_key": "k", "base_url": "u"}}, callbacks=_RecordingCallbacks())
dm.add(
thread_id="t1",
messages=[HumanMessage(content="hi"), AIMessage(content="hello")],
@ -195,9 +229,9 @@ def test_scoped_clear_preserves_shared_summaries(deermem_data_dir):
assert dm.get_memory(user_id="alice")["user"]["workContext"]["summary"] == "shared profile"
def test_tracing_callback_optional_no_langfuse(deermem_data_dir):
def test_callbacks_optional_no_langfuse(deermem_data_dir):
dm = _deermem_with_fake_llm({"model": {"provider": "openai", "model": "gpt-x", "api_key": "k", "base_url": "u"}})
assert dm._config.tracing_callback is None # langfuse not hard-required
assert dm.callbacks is None # no callbacks = no langfuse (not hard-required)
dm.add(
thread_id="t2",
messages=[HumanMessage(content="hi"), AIMessage(content="hello")],
@ -205,7 +239,7 @@ def test_tracing_callback_optional_no_langfuse(deermem_data_dir):
user_id="u2",
trace_id="t-99",
)
dm._queue.flush() # no callback, no error, update completes
dm._queue.flush() # no callbacks, no error, update completes
def test_hide_from_ui_default_skip_hook_keeps():
@ -244,31 +278,74 @@ def test_portability_only_abc_contract_imports_deerflow():
# Minimal vendored host contract (what another agent would ship). DeerMem only
# needs this ABC -- nothing else from a host.
_VENDORED_MANAGER_PY = '''
"""Vendored host contract (minimal ABC) for the portability demo."""
from abc import ABC, abstractmethod
from typing import Any
"""Vendored host contract (pydantic BaseModel + three-tier ABC) for the portability demo."""
from abc import abstractmethod
from typing import Any, ClassVar, Literal
class MemoryManager(ABC):
def __init__(self, backend_config: dict | None = None) -> None:
self._backend_config = backend_config
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
class MemoryManager(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
backend_config: dict[str, Any] = Field(default_factory=dict)
mode: Literal["middleware", "tool"] = "middleware"
callbacks: Any = None
supports_search: ClassVar[bool] = False
@field_validator("backend_config", mode="before")
@classmethod
def _coerce_backend_config(cls, value):
return value or {}
@model_validator(mode="after")
def _check_invariants(self):
if self.mode == "tool" and not type(self).supports_search:
raise ValueError("tool mode requires search")
return self
# Tier 1: abstract (every backend must implement).
@abstractmethod
def add(self, thread_id, messages, *, agent_name=None, user_id=None, trace_id=None) -> None: ...
@abstractmethod
def add_nowait(self, thread_id, messages, *, agent_name=None, user_id=None) -> None: ...
@abstractmethod
def get_context(self, user_id, *, agent_name=None, thread_id=None) -> str: ...
@classmethod
@abstractmethod
def search(self, query, top_k=5, *, user_id=None, agent_name=None) -> list: ...
@abstractmethod
def get_memory(self, *, user_id=None, agent_name=None) -> dict: ...
@abstractmethod
def delete_memory(self, *, user_id=None, agent_name=None) -> None: ...
@abstractmethod
def clear_memory(self, *, user_id=None, agent_name=None) -> dict: ...
@abstractmethod
def import_memory(self, memory_data, *, user_id=None, agent_name=None) -> dict: ...
@abstractmethod
def export_memory(self, *, user_id=None, agent_name=None) -> dict: ...
def from_config(cls, backend_config, *, mode="middleware", **host_hooks): ...
# Tier 2: management defaults (override if supported).
def add_nowait(self, thread_id, messages, *, agent_name=None, user_id=None) -> None:
self.add(thread_id, messages, agent_name=agent_name, user_id=user_id)
def search(self, query, top_k=5, *, user_id=None, agent_name=None, category=None) -> list:
raise NotImplementedError
def get_memory(self, *, user_id=None, agent_name=None) -> dict:
raise NotImplementedError
def delete_memory(self, *, user_id=None, agent_name=None) -> None:
raise NotImplementedError
def clear_memory(self, *, user_id=None, agent_name=None) -> dict:
raise NotImplementedError
def import_memory(self, memory_data, *, user_id=None, agent_name=None) -> dict:
raise NotImplementedError
def export_memory(self, *, user_id=None, agent_name=None) -> dict:
raise NotImplementedError
def shutdown_flush(self, timeout) -> bool:
return True
# Tier 3: optional hooks (override if supported).
def warm(self) -> bool | None:
return None
def reload_memory(self, *, user_id=None, agent_name=None) -> dict:
raise NotImplementedError
def create_fact(self, content, category="context", confidence=0.5, *, agent_name=None, user_id=None):
raise NotImplementedError
def delete_fact(self, fact_id, *, agent_name=None, user_id=None) -> dict:
raise NotImplementedError
def update_fact(self, fact_id, content=None, category=None, confidence=None, *, agent_name=None, user_id=None) -> dict:
raise NotImplementedError
def on_pre_compress(self, messages) -> str:
return ""
def on_turn_start(self, turn_number, message, **kwargs) -> None:
return None
class MemoryConflictError(RuntimeError): ...
class MemoryCorruptionError(RuntimeError): ...

67
backend/tests/test_gateway_lifespan_shutdown.py
View File

@ -189,3 +189,70 @@ def test_lifespan_skips_memory_flush_when_disabled() -> None:
"""memory.enabled=False skips the drain entirely."""
manager = asyncio.run(_run_lifespan_with_memory_flush(enabled=False, flush_return=True))
manager.shutdown_flush.assert_not_called()
# ── startup warm-up log accuracy ────────────────────────────────────────────
async def _run_lifespan_with_warm_return(warm_return: bool | None) -> MagicMock:
"""Drive lifespan with a spied ``manager.warm`` returning ``warm_return``.
The startup warm block reads the tri-state return: None = nothing to warm
(logs "skipping"), True = warmed, False = failed (logs WARNING). Returns the
manager mock so the caller can assert warm was reached.
"""
from app.gateway.app import lifespan
app = FastAPI()
startup_config = SimpleNamespace(
log_level="INFO",
memory=SimpleNamespace(
token_counting="char",
enabled=False,
shutdown_flush_timeout_seconds=5.0,
),
)
fake_service = MagicMock()
fake_service.get_status = MagicMock(return_value={})
close_oidc_service = AsyncMock()
stop_channel_service = AsyncMock()
async def fake_start(_startup_config):
return fake_service
manager = MagicMock()
manager.warm.return_value = warm_return
with (
patch("app.gateway.app.get_app_config", return_value=startup_config),
patch("app.gateway.app.get_gateway_config", return_value=MagicMock(host="x", port=0)),
patch("app.gateway.app.langgraph_runtime", _noop_langgraph_runtime),
patch("app.gateway.app.auth.close_oidc_service", close_oidc_service),
patch("app.channels.service.start_channel_service", side_effect=fake_start),
patch("app.channels.service.stop_channel_service", stop_channel_service),
patch("deerflow.agents.memory.get_memory_manager", return_value=manager),
):
async with lifespan(app):
pass
return manager
def test_lifespan_logs_skipping_when_backend_has_nothing_to_warm(caplog) -> None:
"""A backend whose warm() returns None (base default -- nothing to warm,
e.g. noop) logs "skipping" at INFO, not the misleading "warmed successfully"
(a non-DeerMem backend never touched the tiktoken cache)."""
caplog.set_level(logging.INFO, logger="app.gateway.app")
manager = asyncio.run(_run_lifespan_with_warm_return(None))
manager.warm.assert_called_once_with()
assert any(r.levelno == logging.INFO and "nothing to warm" in r.message for r in caplog.records)
assert not any("warmed successfully" in r.message for r in caplog.records)
def test_lifespan_warns_when_warm_returns_false(caplog) -> None:
"""warm()=False means warming was attempted and failed; the host logs a
WARNING so the operator sees the character-based-fallback degradation."""
caplog.set_level(logging.WARNING, logger="app.gateway.app")
manager = asyncio.run(_run_lifespan_with_warm_return(False))
manager.warm.assert_called_once_with()
assert any(r.levelno == logging.WARNING and "warm-up failed" in r.message for r in caplog.records)

246
backend/tests/test_memory_manager_interface.py Normal file
View File

@ -0,0 +1,246 @@
"""Conformance tests for the MemoryManager interface contract.
Pins the three-tier ABC + from_config + invariant validator + async a* +
callbacks surface so the contract stays stable as backends are added. The
centerpiece is ``_MinimalBackend`` (implements ONLY from_config + add +
get_context) -- it instantiates via the factory and runs with everything else
inherited, proving a new backend needs nothing else. That is the direct
evidence the optimization lowered onboarding cost (no full method surface, no
factory edit).
Each test resets the singleton + restores config so they are order-independent.
"""
from __future__ import annotations
import asyncio
import pytest
from pydantic import PrivateAttr
from deerflow.agents.memory import MemoryManager, get_memory_manager, reset_memory_manager
from deerflow.agents.memory.manager import MemoryCallbacks
from deerflow.config.memory_config import MemoryConfig, get_memory_config, set_memory_config
class _MinimalBackend(MemoryManager):
"""Implements ONLY tier-1 (add/get_context) + from_config.
Everything else inherits base defaults -- the minimal onboarding surface a
new memory system needs. (No search, no management ops, no fact CRUD, no
cache reload; ``supports_search`` stays False, so it cannot run in tool mode.)
"""
_adds: list = PrivateAttr(default_factory=list)
def add(self, thread_id, messages, *, agent_name=None, user_id=None, trace_id=None) -> None:
self._adds.append((thread_id, user_id))
def get_context(self, user_id, *, agent_name=None, thread_id=None) -> str:
return f"ctx:{user_id}"
@classmethod
def from_config(cls, backend_config, *, mode="middleware", **host_hooks):
# Consumes nothing from host_hooks -- a truly minimal backend.
return cls(backend_config=backend_config or {}, mode=mode)
@pytest.fixture(autouse=True)
def _isolate_memory_manager():
orig = get_memory_config()
reset_memory_manager()
yield
set_memory_config(orig)
reset_memory_manager()
def test_minimal_backend_onboards_via_factory_with_only_add_get_context():
"""Centerpiece: a backend implementing ONLY from_config + add + get_context
instantiates via the factory and runs -- all else inherits defaults. Direct
evidence that onboarding cost dropped (no full method surface, no factory
edit)."""
set_memory_config(MemoryConfig(manager_class=f"{__name__}:_MinimalBackend"))
manager = get_memory_manager()
assert isinstance(manager, _MinimalBackend)
# tier-1 works
manager.add("t1", [], user_id="u1")
assert manager._adds == [("t1", "u1")]
assert manager.get_context("u1") == "ctx:u1"
# tier-2 inherits defaults: add_nowait delegates to add; shutdown_flush=True;
# the rest raise NotImplementedError.
manager.add_nowait("t1", [], user_id="u1")
assert manager._adds[-1] == ("t1", "u1")
assert manager.shutdown_flush(1.0) is True
with pytest.raises(NotImplementedError):
manager.search("q")
with pytest.raises(NotImplementedError):
manager.get_memory(user_id="u")
with pytest.raises(NotImplementedError):
manager.clear_memory(user_id="u")
with pytest.raises(NotImplementedError):
manager.import_memory({}, user_id="u")
with pytest.raises(NotImplementedError):
manager.export_memory(user_id="u")
with pytest.raises(NotImplementedError):
manager.delete_memory(user_id="u")
# tier-3 inherits defaults: warm=None (nothing to warm); B-class no-op;
# A-class (excl. warm) raise.
assert manager.warm() is None
assert manager.on_pre_compress([]) == ""
assert manager.on_turn_start(1, None) is None
with pytest.raises(NotImplementedError):
manager.reload_memory(user_id="u")
with pytest.raises(NotImplementedError):
manager.create_fact("x", user_id="u")
with pytest.raises(NotImplementedError):
manager.delete_fact("x", user_id="u")
with pytest.raises(NotImplementedError):
manager.update_fact("x", user_id="u")
def test_tier1_abstract_enforcement():
"""A backend missing add or get_context cannot instantiate (TypeError at
construction -- memory is persistent state, missing write/read is a severe
bug caught eagerly)."""
class _NoAdd(MemoryManager):
def get_context(self, user_id, *, agent_name=None, thread_id=None) -> str:
return ""
class _NoGet(MemoryManager):
def add(self, thread_id, messages, *, agent_name=None, user_id=None, trace_id=None) -> None:
pass
with pytest.raises(TypeError):
_NoAdd()
with pytest.raises(TypeError):
_NoGet()
def test_invariant_tool_mode_requires_search():
"""mode='tool' + a non-search backend raises at instantiation (the agent
calls memory_search in tool mode, so a non-search backend is a
misconfiguration -- fail fast). Middleware mode is fine for any backend."""
with pytest.raises(ValueError):
_MinimalBackend(backend_config={}, mode="tool")
assert _MinimalBackend(backend_config={}, mode="middleware").mode == "middleware"
def test_invariant_tool_mode_factory_path_raises():
"""The invariant also fires on the factory path (mode='tool' + non-search
backend configured via manager_class)."""
set_memory_config(MemoryConfig(manager_class=f"{__name__}:_MinimalBackend", mode="tool"))
with pytest.raises(ValueError):
get_memory_manager()
class _SearchOverrideForgotFlag(_MinimalBackend):
"""Overrides search() but forgets supports_search=True (flag/impl drift)."""
def search(self, query, top_k=5, *, user_id=None, agent_name=None, category=None):
return []
class _FlagWithoutSearchOverride(_MinimalBackend):
"""Sets supports_search=True without overriding search() (flag/impl drift)."""
supports_search = True
class _ConsistentSearchBackend(_MinimalBackend):
"""Overrides search() AND sets supports_search=True -- consistent, tool-OK."""
supports_search = True
def search(self, query, top_k=5, *, user_id=None, agent_name=None, category=None):
return []
def test_invariant_supports_search_flag_must_match_override():
"""supports_search (ClassVar) must match whether search() is overridden, so
the flag can't drift from the implementation -- caught at instantiation, not
as a misleading tool-mode rejection (override-but-forgot-flag) or a runtime
NotImplementedError on the first memory_search call (flag-without-override)."""
with pytest.raises(ValueError, match="inconsistent"):
_SearchOverrideForgotFlag(backend_config={})
with pytest.raises(ValueError, match="inconsistent"):
_FlagWithoutSearchOverride(backend_config={})
def test_invariant_consistent_search_backend_runs_in_tool_mode():
"""A backend that overrides search() AND sets supports_search=True is
consistent and may run in tool mode (the override is the real capability;
the flag agrees with it)."""
manager = _ConsistentSearchBackend(backend_config={}, mode="tool")
assert manager.mode == "tool"
assert manager.search("q") == []
def test_async_defaults_delegate_to_sync():
"""a* methods default to the sync path (no concurrency benefit); a future
async LLM client overrides without changing the contract."""
manager = _MinimalBackend(backend_config={})
asyncio.run(manager.aadd("t", [], user_id="u"))
assert manager._adds == [("t", "u")]
assert asyncio.run(manager.aget_context("u")) == "ctx:u"
# asearch delegates to search -> raises (default) just like search.
with pytest.raises(NotImplementedError):
asyncio.run(manager.asearch("q"))
def test_callbacks_field_optional_and_noop_default():
"""callbacks is an optional field (default None); the base
MemoryCallbacks.on_memory_llm_call is a no-op, so a backend with no
callbacks runs without tracing."""
assert _MinimalBackend(backend_config={}).callbacks is None
noop = MemoryCallbacks()
# no-op: mutates nothing, raises nothing
noop.on_memory_llm_call({}, thread_id="t", user_id="u", trace_id="tr", model_name="m")
manager = _MinimalBackend(backend_config={}, callbacks=noop)
assert manager.callbacks is noop
def test_from_config_consumes_host_hooks_it_needs():
"""A backend's from_config consumes the host_hooks it needs; the minimal
backend consumes none (ignores callbacks / host_llm_factory / etc.). A real
backend (DeerMem) consumes the ones it uses -- see test_deermem_self_contained."""
manager = _MinimalBackend.from_config(
{"some_key": "v"},
mode="middleware",
callbacks=MemoryCallbacks(),
host_llm_factory=lambda: None,
should_keep_hidden_message=lambda ak: True,
)
assert isinstance(manager, _MinimalBackend)
assert manager.backend_config == {"some_key": "v"}
assert manager.mode == "middleware"
# The minimal backend consumes NO host_hooks (callbacks / host_llm_factory /
# should_keep_hidden_message are all ignored) -- callbacks stays None.
assert manager.callbacks is None
def test_unsupported_op_raises_for_caller_try_except():
"""Callers (router/client/tools) call tier-3 ops directly and catch
NotImplementedError (no more hasattr probing). A minimal backend's
unsupported ops raise, so a caller's try/except degrades cleanly (501 /
fallback / JSON error)."""
manager = _MinimalBackend(backend_config={})
try:
manager.create_fact("x", user_id="u")
raise AssertionError("create_fact should have raised NotImplementedError")
except NotImplementedError:
pass # caller returns 501 / JSON error / falls back
def test_only_add_and_get_context_are_abstract():
"""The tier-1 abstract set is exactly {add, get_context} (plus from_config);
tier-2/3 methods carry defaults so a backend implements only what it supports."""
assert "add" in MemoryManager.__abstractmethods__
assert "get_context" in MemoryManager.__abstractmethods__
assert "from_config" in MemoryManager.__abstractmethods__
# tier-2/3 are NOT abstract (they have defaults)
for non_abstract in ("search", "get_memory", "shutdown_flush", "warm", "create_fact", "on_pre_compress"):
assert non_abstract not in MemoryManager.__abstractmethods__, non_abstract

37
backend/tests/test_memory_manager_pluggable.py
View File

@ -79,18 +79,23 @@ def test_noop_runs_with_empty_memory() -> None:
assert manager.get_memory(user_id="u") == {"facts": []}
def test_internal_capabilities_are_hasattr_probeable() -> None:
"""reload_memory + fact CRUD + warm exist on DeerMem but not on noop (the ABC omits them)."""
set_memory_config(MemoryConfig(manager_class="deermem"))
deermem = get_memory_manager()
for cap in ("warm", "reload_memory", "create_fact", "delete_fact", "update_fact"):
assert hasattr(deermem, cap), cap
reset_memory_manager()
def test_tier3_hooks_have_defaults_noop_inherits() -> None:
"""warm/reload/fact CRUD are tier-3 hooks ON the ABC with defaults (no more
``hasattr`` probing): noop inherits ``warm``=None (nothing to warm) and
fact-CRUD/reload raise ``NotImplementedError``. DeerMem overrides the ones
it supports (covered elsewhere)."""
set_memory_config(MemoryConfig(manager_class="noop"))
noop = get_memory_manager()
for cap in ("warm", "reload_memory", "create_fact", "delete_fact", "update_fact"):
assert not hasattr(noop, cap), cap
assert noop.warm() is None # inherited default (nothing to warm)
with pytest.raises(NotImplementedError):
noop.reload_memory(user_id="u")
with pytest.raises(NotImplementedError):
noop.create_fact("x", user_id="u")
with pytest.raises(NotImplementedError):
noop.delete_fact("x", user_id="u")
with pytest.raises(NotImplementedError):
noop.update_fact("x", user_id="u")
reset_memory_manager()
def test_deermem_search_works_delete_export_are_stubs() -> None:
@ -149,11 +154,13 @@ def test_empty_storage_path_factory_injects_runtime_home(tmp_path, monkeypatch)
assert len(user_dirs) == 1
def test_shutdown_flush_is_on_abc_and_noop_is_noop_success() -> None:
"""``shutdown_flush`` is part of the MemoryManager ABC (every backend must
implement a bounded graceful-shutdown drain); noop has no buffer, so it
returns True immediately."""
assert "shutdown_flush" in MemoryManager.__abstractmethods__
def test_shutdown_flush_has_default_and_noop_is_noop_success() -> None:
"""``shutdown_flush`` is a tier-2 method with a default (True -- backends
without a buffer have nothing to drain), NOT abstract; noop inherits/overrides
to True. Only ``add`` / ``get_context`` are tier-1 abstract."""
assert "add" in MemoryManager.__abstractmethods__
assert "get_context" in MemoryManager.__abstractmethods__
assert "shutdown_flush" not in MemoryManager.__abstractmethods__
reset_memory_manager()
set_memory_config(MemoryConfig(manager_class="noop"))
noop = get_memory_manager()

88
backend/tests/test_memory_router.py
View File

@ -463,3 +463,91 @@ def test_import_memory_scopes_overwrite_to_bound_owner() -> None:
with patch("app.gateway.routers.memory.get_effective_user_id", return_value="real-user"):
asyncio.run(memory.import_memory(payload, _browser_request_with_spoofed_owner_header()))
assert seen["user_id"] == "real-user"
# ── unsupported-backend 501s ────────────────────────────────────────────────
# A minimal backend (only add + get_context) inherits the tier-2/tier-3 default
# raise for get_memory / clear_memory / import_memory / reload_memory. Before
# the contract change these were @abstractmethod (every backend implemented
# them, so the endpoints could never raise); now the endpoints catch
# NotImplementedError -> 501 so an unsupported backend gets a clean "not
# supported" instead of a raw 500 (there is no global NotImplementedError
# handler, so an uncaught raise is a 500).
def _unsupported_manager() -> MagicMock:
"""Mock a minimal backend: read/manage ops raise NotImplementedError."""
mock_mgr = MagicMock()
mock_mgr.get_memory.side_effect = NotImplementedError("get_memory not supported")
mock_mgr.clear_memory.side_effect = NotImplementedError("clear_memory not supported")
mock_mgr.import_memory.side_effect = NotImplementedError("import_memory not supported")
mock_mgr.reload_memory.side_effect = NotImplementedError("reload_memory not supported")
return mock_mgr
def test_get_memory_route_returns_501_for_unsupported_backend() -> None:
app = FastAPI()
app.include_router(memory.router)
with patch("app.gateway.routers.memory.get_memory_manager", return_value=_unsupported_manager()):
with TestClient(app) as client:
response = client.get("/api/memory")
assert response.status_code == 501
assert "not supported" in response.json()["detail"]
def test_export_memory_route_returns_501_for_unsupported_backend() -> None:
app = FastAPI()
app.include_router(memory.router)
with patch("app.gateway.routers.memory.get_memory_manager", return_value=_unsupported_manager()):
with TestClient(app) as client:
response = client.get("/api/memory/export")
assert response.status_code == 501
def test_memory_status_route_returns_501_for_unsupported_backend() -> None:
app = FastAPI()
app.include_router(memory.router)
cfg = SimpleNamespace(
enabled=True,
mode="middleware",
injection_enabled=True,
shutdown_flush_timeout_seconds=30.0,
manager_class="minimal",
backend_config={},
)
with (
patch("app.gateway.routers.memory.get_memory_manager", return_value=_unsupported_manager()),
patch("app.gateway.routers.memory.get_memory_config", return_value=cfg),
):
with TestClient(app) as client:
response = client.get("/api/memory/status")
assert response.status_code == 501
def test_clear_memory_route_returns_501_for_unsupported_backend() -> None:
app = FastAPI()
app.include_router(memory.router)
with patch("app.gateway.routers.memory.get_memory_manager", return_value=_unsupported_manager()):
with TestClient(app) as client:
response = client.delete("/api/memory")
assert response.status_code == 501
def test_import_memory_route_returns_501_for_unsupported_backend() -> None:
app = FastAPI()
app.include_router(memory.router)
with patch("app.gateway.routers.memory.get_memory_manager", return_value=_unsupported_manager()):
with TestClient(app) as client:
response = client.post("/api/memory/import", json=_sample_memory())
assert response.status_code == 501
def test_reload_memory_route_returns_501_when_read_also_unsupported() -> None:
"""reload falls back to get_memory; if both raise (minimal backend), the
fallback surfaces 501 instead of a raw 500 from the uncaught raise."""
app = FastAPI()
app.include_router(memory.router)
with patch("app.gateway.routers.memory.get_memory_manager", return_value=_unsupported_manager()):
with TestClient(app) as client:
response = client.post("/api/memory/reload")
assert response.status_code == 501

19
backend/tests/test_memory_tools.py
View File

@ -92,16 +92,23 @@ class _MockManager:
raise self._raise_on_delete
return {"facts": []}
# Tool uses getattr+callable to probe these; shadow with None to simulate a
# backend that does not expose fact CRUD (e.g. noop) -- getattr() returns
# None and the tool's callable() check fails gracefully.
# fact-CRUD ops are tier-3 hooks that raise NotImplementedError when
# unsupported (the tool catches it -> JSON error). Simulate an unsupported
# backend by replacing each op with a raiser (no more None/hasattr probing).
def _drop_fact_ops(self):
if not self._supports_create:
self.create_fact = None
self.create_fact = self._unsupported("create_fact")
if not self._supports_update:
self.update_fact = None
self.update_fact = self._unsupported("update_fact")
if not self._supports_delete:
self.delete_fact = None
self.delete_fact = self._unsupported("delete_fact")
@staticmethod
def _unsupported(name):
def _raise(*args, **kwargs):
raise NotImplementedError(f"{name} not supported by _MockManager")
return _raise
def _install_manager(monkeypatch, manager):

2
backend/tests/test_memory_updater.py
View File

@ -3,7 +3,7 @@ import pytest
pytest.skip(
"Pending full DI migration: MemoryUpdater now takes (config, storage, llm); "
"module-level funcs are instance methods. Key paths (DI, zero-config, trace_id, "
"tracing_callback, hide_from_ui, LLM update, fact extraction) are covered by "
"callbacks, hide_from_ui, LLM update, fact extraction) are covered by "
"test_deermem_self_contained.py. Full unit-test migration is a follow-up.",
allow_module_level=True,
)