feat(benchmark): add offline memory vs project-context comparison suite (part of #7884) (#8058)

* feat(benchmark): add offline memory vs project-context comparison suite (part of #7884)

* fix(benchmark): route prompts through DynamicContextMiddleware, flatten rows, and share test run

* refactor(benchmark): run through local Gateway and simplify first-pass suite (#7884)

* docs(agents): keep context memory benchmark guidance within chain budget

* fix(benchmark): verify DeerMem scope rejection

---------

Co-authored-by: Totoro <msy626836554@gmail.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Wistom Zhang 2026-11-10 14:45:56 +08:00 committed by GitHub
parent 10bbcc92de
commit 1fbc06279a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 519 additions and 0 deletions

View File

@ -43,6 +43,9 @@ vars; `summarize` and pytest are offline. See its README for the protocol.
Benchmarks in `scripts/benchmark/` must be standalone and reproducible. Import
production functions; never duplicate them or introduce an alternative runtime.
`scripts/benchmark/context_memory_comparison/` runs an offline Gateway with
SQLite. Keep original- and fresh-thread results separate.
Sandbox benchmarks reuse production lifecycle. Reject negative/NaN timing values
before provider creation, export no secrets, and fail on final cleanup errors.
See the sandbox benchmark README for supported options and tests.

View File

@ -0,0 +1,34 @@
# Offline memory and project context comparison
This synthetic benchmark for [#7884](https://github.com/bytedance/deer-flow/issues/7884)
starts a disposable Gateway in process with SQLite. It creates projects, shelf
documents, threads and memory through the real API. A fixed model records the
messages passed to it, and DeerMem receives scripted extraction responses.
Run this command from the backend directory of a Git checkout.
```sh
uv run python -m scripts.benchmark.context_memory_comparison run-offline --output-dir /tmp/context-memory
```
The command runs the committed fixtures and writes one `report.json` with
the source revision, benchmark and fixture hashes, and the observations below.
- Two projects and an unassigned thread must see only their own context and
the default memory bucket. A custom agent must see its own memory bucket.
The project-only check requires a successful update with exactly one fact
rejected for scope (`scope_rejections` in the report). The fact must also be
absent from default memory and from the other thread's prompt.
- Replacement and deletion are observed in the store, in the original thread's
next turn, and in a fresh thread. A first-turn canary verifies checkpoint reuse.
The fresh thread is a control. If the original thread keeps a memory block
from its checkpoint, the report records
the stale fact and marks that check failed.
- Each recorded turn includes character-based token estimates for the
`<memory>`, `<project>` and `<documents>` blocks in its actual model request.
The default and custom-agent observations identify the memory bucket being
measured. These are estimates for the injected blocks.
The exit code is 0 when all behavioral checks pass and 1 when any fail.
The benchmark uses synthetic data and fixed model responses. Gateway state is
temporary, and the report contains canary IDs and counts rather than full prompts.

View File

@ -0,0 +1 @@
"""DeerFlow Issue #7884 Offline Memory vs Project Context Comparison Benchmark."""

View File

@ -0,0 +1,4 @@
from .cli import main
if __name__ == "__main__":
raise SystemExit(main())

View File

@ -0,0 +1,27 @@
from __future__ import annotations
import argparse
import json
from collections.abc import Sequence
from pathlib import Path
from tempfile import TemporaryDirectory
from .runner import run_offline
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Compare DeerMem and project context through an offline Gateway.")
parser.add_argument("command", choices=["run-offline"])
parser.add_argument("--output-dir", type=Path, required=True)
args = parser.parse_args(argv)
report = run_offline()
output_dir = args.output_dir
output_dir.mkdir(parents=True, exist_ok=True)
target = output_dir / "report.json"
with TemporaryDirectory(dir=output_dir) as staging:
path = Path(staging) / target.name
path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
path.replace(target)
print(f"Wrote {target}")
return 0 if report["passed"] else 1

View File

@ -0,0 +1,28 @@
{
"dataset": "context-memory-synthetic-v1",
"question": "Which context applies here?",
"thread_history_control": "BENCH-THREAD-HISTORY",
"default_fact": "BENCH-DEFAULT-PREFERENCE",
"agent": "bench-agent",
"agent_fact": "BENCH-AGENT-PREFERENCE",
"project_only_fact": "BENCH-PROJECT-ONLY-PREFERENCE",
"projects": {
"alpha": {
"instructions": "BENCH-ALPHA-INSTRUCTIONS",
"document": "BENCH-ALPHA-DOCUMENT.md"
},
"beta": {
"instructions": "BENCH-BETA-INSTRUCTIONS",
"document": "BENCH-BETA-DOCUMENT.md"
}
},
"invalidation": {
"replacement": {
"old": "BENCH-OLD-REPLACEMENT",
"new": "BENCH-NEW-REPLACEMENT"
},
"deletion": {
"old": "BENCH-OLD-DELETION"
}
}
}

View File

@ -0,0 +1,305 @@
"""Run synthetic prompt checks through the local Gateway and real SQLite."""
from __future__ import annotations
import hashlib
import json
import os
import re
import subprocess
from contextlib import chdir
from contextvars import ContextVar
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Any, ClassVar
from unittest.mock import patch
from fastapi.testclient import TestClient
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_core.outputs import ChatGeneration, ChatResult
ROOT = Path(__file__).resolve().parent
MANIFEST = ROOT / "manifest.json"
REPLY = "Offline benchmark response."
DAY = "2026-11-09, Monday"
_PROJECT_SCOPE_TRACE = "benchmark-project-scope"
_EMPTY_UPDATE = {"user": {}, "history": {}, "newFacts": [], "factsToRemove": []}
_SCRIPTED_UPDATE: ContextVar[dict[str, Any] | None] = ContextVar("offline_memory_update", default=None)
class OfflineModel(BaseChatModel):
"""Capture agent requests and return fixed replies for both model roles."""
model: str
temperature: float = 0.0
requests: ClassVar[list[list[BaseMessage]]] = []
@property
def _llm_type(self) -> str:
return "context-memory-offline"
def bind_tools(self, tools, **kwargs):
return self
def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult:
if self.model == "extract":
content = json.dumps(_SCRIPTED_UPDATE.get() or _EMPTY_UPDATE)
else:
self.requests.append(list(messages))
content = REPLY
return ChatResult(generations=[ChatGeneration(message=AIMessage(content=content))])
def _request(client: TestClient, method: str, path: str, **kwargs) -> dict:
response = client.request(method, path, **kwargs)
response.raise_for_status()
return response.json()
def _thread(client: TestClient, project_id: str | None = None) -> str:
return _request(client, "POST", "/api/threads", json={"project_id": project_id})["thread_id"]
def _prompt(client: TestClient, thread_id: str, question: str, assistant_id: str = "lead_agent") -> list[BaseMessage]:
OfflineModel.requests.clear()
result = _request(
client,
"POST",
f"/api/threads/{thread_id}/runs/wait",
json={
"assistant_id": assistant_id,
"input": {"messages": [{"role": "user", "content": question}]},
"config": {"configurable": {"model_name": "record", "thinking_enabled": False}},
},
)
messages = result.get("messages") or []
if len(OfflineModel.requests) != 1 or not messages or messages[-1].get("content") != REPLY:
raise RuntimeError("Gateway run did not complete with exactly one recorded model request")
return OfflineModel.requests.pop()
def _present(text: str, canaries: set[str]) -> list[str]:
return sorted(canary for canary in canaries if canary in text)
def _observe(messages: list[BaseMessage], canaries: set[str], expected: set[str]) -> dict:
from deerflow.agents.memory.backends.deermem.deermem.core.prompt import _count_tokens
from deerflow.projects.context import is_project_context_message
present = _present("\n".join(str(message.content) for message in messages), canaries)
blocks = {"memory": [], "project": [], "documents": []}
for message in messages:
if not isinstance(message, HumanMessage) or not isinstance(message.content, str):
continue
if message.content.startswith("<memory>"):
blocks["memory"].append(message.content)
if is_project_context_message(message):
for tag in ("project", "documents"):
blocks[tag].extend(re.findall(rf"<{tag}(?:\s[^>]*)?>.*?</{tag}>", message.content, flags=re.DOTALL))
return {
"passed": set(present) == expected,
"present": present,
"context_token_estimate": {name: sum(_count_tokens(block, use_tiktoken=False) for block in values) for name, values in blocks.items()},
}
def _extract(thread_id: str, instruction: str, *, new_facts: list[dict], removals: list[dict], trace_id: str | None = None) -> bool:
from deerflow.agents.memory import get_memory_manager
from deerflow.agents.memory.backends.deermem.deermem.core.paths import DEFAULT_AGENT_BUCKET
from deerflow.runtime.user_context import DEFAULT_USER_ID
token = _SCRIPTED_UPDATE.set({**_EMPTY_UPDATE, "newFacts": new_facts, "factsToRemove": removals})
try:
# Execute the production write synchronously so its outcome precedes
# the next Gateway run. Ordinary background extractions are no-ops.
return get_memory_manager()._updater.update_memory(
[HumanMessage(content=instruction), AIMessage(content="Understood.")],
thread_id=thread_id,
user_id=DEFAULT_USER_ID,
agent_name=DEFAULT_AGENT_BUCKET,
trace_id=trace_id,
bypass_watermark=True,
)
finally:
_SCRIPTED_UPDATE.reset(token)
def _fact(content: str) -> dict:
return {"content": content, "category": "preference", "confidence": 0.99}
def _extracted_fact(content: str, scope: str) -> dict:
return {**_fact(content), "scope": scope, "durability": "durable", "authority": "descriptive"}
def _scope_isolation(client: TestClient, fixture: dict, project_extraction_metrics: dict) -> dict:
default_fact = fixture["default_fact"]
agent = fixture["agent"]
agent_fact = fixture["agent_fact"]
rejected = fixture["project_only_fact"]
_request(client, "POST", "/api/memory/facts", json=_fact(default_fact))
_request(client, "POST", "/api/agents", json={"name": agent, "model": "record", "soul": "Synthetic benchmark agent."})
_request(client, "POST", "/api/memory/facts", params={"agent_name": agent}, json=_fact(agent_fact))
threads = {}
project_canaries = {}
for label, project in fixture["projects"].items():
project_id = _request(client, "POST", "/api/projects", json={"name": label, "instructions": project["instructions"]})["id"]
document = project["document"]
_request(client, "POST", f"/api/projects/{project_id}/documents", files={"file": (document, document.encode("utf-8"), "text/plain")})
threads[label] = _thread(client, project_id)
project_canaries[label] = {project["instructions"], document}
canaries = {default_fact, agent_fact, rejected}.union(*project_canaries.values())
question = fixture["question"]
observations = {label: _observe(_prompt(client, thread, question), canaries, {default_fact, *project_canaries[label]}) for label, thread in threads.items()}
observations["unassigned"] = _observe(_prompt(client, _thread(client), question), canaries, {default_fact})
observations["custom_agent"] = _observe(_prompt(client, _thread(client), question, agent), canaries, {agent_fact})
updated = _extract(
threads["alpha"],
"Apply this preference only in the current project.",
new_facts=[_extracted_fact(rejected, "project")],
removals=[],
trace_id=_PROJECT_SCOPE_TRACE,
)
project_fact = {"update_succeeded": updated, "passed": False}
if updated:
scope_rejections = project_extraction_metrics.get("scope_gate_rejections", {}).get("facts", {}).get("scope", 0)
stored = bool(_present(json.dumps(_request(client, "GET", "/api/memory")), {rejected}))
beta = _observe(_prompt(client, threads["beta"], question), canaries, {default_fact, *project_canaries["beta"]})
project_fact.update({"scope_rejections": scope_rejections, "stored": stored, "other_thread": beta, "passed": scope_rejections == 1 and not stored and beta["passed"]})
return {
"passed": all(turn["passed"] for turn in observations.values()) and project_fact["passed"],
"threads": observations,
"project_only_fact": project_fact,
}
def _invalidation(client: TestClient, fixture: dict, case: dict) -> dict:
from deerflow.agents.memory.backends.deermem.deermem.core.storage import create_empty_memory
old = case["old"]
new = case.get("new")
control = fixture["default_fact"]
history = fixture["thread_history_control"]
canaries = {old, control, history}
after = {control}
removal = {"id": "superseded-fact", "scope": "user", "reason": "The user invalidated the old preference."}
new_facts = []
if new is not None:
canaries.add(new)
after.add(new)
removal["replacementFactIndex"] = 0
new_facts.append(_extracted_fact(new, "user"))
seed = create_empty_memory()
seed["facts"] = [
{**_fact(old), "id": removal["id"], "source": "synthetic"},
{**_fact(control), "id": "control-fact", "source": "synthetic"},
]
_request(client, "POST", "/api/memory/import", json=seed)
original = _thread(client)
question = fixture["question"]
before = _observe(_prompt(client, original, f"{question} {history}"), canaries, {old, control, history})
updated = _extract(original, "Forget the old preference.", new_facts=new_facts, removals=[removal])
if not updated:
return {"passed": False, "update_succeeded": False, "before": before}
stored = _present(json.dumps(_request(client, "GET", "/api/memory")), canaries)
same = _observe(_prompt(client, original, question), canaries, after | {history})
fresh = _observe(_prompt(client, _thread(client), question), canaries, after)
return {
"passed": before["passed"] and set(stored) == after and same["passed"] and fresh["passed"],
"update_succeeded": True,
"before": before,
"stored": stored,
"same_thread": same,
"new_thread": fresh,
}
def _config() -> dict:
return {
# DeerMem uses the first configured model for extraction. Runs
# explicitly select the recording model so these calls stay separate.
"models": [{"name": name, "use": f"{__name__}:OfflineModel", "model": name, "temperature": 0, "supports_tool_calling": True} for name in ("extract", "record")],
"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
"database": {"backend": "sqlite", "sqlite_dir": "db"},
"run_events": {"backend": "db"},
"log_level": "WARNING",
"title": {"enabled": False},
"summarization": {"enabled": False},
"agents_api": {"enabled": True},
"memory": {
"backend_config": {
"token_counting": "char",
"retrieval_adapter": "",
"staleness_review_enabled": False,
"consolidation_enabled": False,
},
},
"skills": {"path": "skills"},
}
def run_offline() -> dict:
"""Use a disposable Gateway config and data directory in a standalone process."""
fixture_bytes = MANIFEST.read_bytes()
fixture = json.loads(fixture_bytes)
canaries = [fixture[key] for key in ("default_fact", "agent_fact", "project_only_fact", "thread_history_control")]
canaries.extend(value for project in fixture["projects"].values() for value in project.values())
canaries.extend(value for case in fixture["invalidation"].values() for value in case.values())
if len(set(canaries)) != len(canaries):
raise ValueError("Synthetic canaries must be distinct")
config_bytes = json.dumps(_config(), sort_keys=True).encode("utf-8")
revision = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
project_extraction_metrics: dict[str, Any] = {}
def record_project_extraction(payload: dict[str, Any]) -> None:
# The explicit project update has its own trace. Ignore background extractions.
if payload.get("trace_id") == _PROJECT_SCOPE_TRACE:
project_extraction_metrics.update(payload)
with TemporaryDirectory(prefix="deerflow-context-memory-", ignore_cleanup_errors=True) as directory:
root = Path(directory).resolve()
(root / "config.yaml").write_bytes(config_bytes)
(root / "extensions.json").write_text("{}", encoding="utf-8")
(root / "empty.env").touch()
env = {key: os.environ[key] for key in ("PATH", "SYSTEMROOT", "WINDIR") if key in os.environ}
env.update(
DEER_FLOW_AUTH_DISABLED="1",
DEER_FLOW_HOME=str(root / "home"),
DEER_FLOW_PROJECT_ROOT=str(root),
DEER_FLOW_CONFIG_PATH=str(root / "config.yaml"),
DEER_FLOW_EXTENSIONS_CONFIG_PATH=str(root / "extensions.json"),
DEER_FLOW_ENV_FILE=str(root / "empty.env"),
)
with patch.dict(os.environ, env, clear=True), chdir(root):
from app.gateway.app import app
with (
patch("deerflow.agents.memory.manager._host_default_extraction_callback", new=record_project_extraction),
patch("deerflow.agents.middlewares.dynamic_context_middleware._format_current_date", return_value=DAY),
TestClient(app) as client,
):
scope = _scope_isolation(client, fixture, project_extraction_metrics)
invalidation = {name: _invalidation(client, fixture, case) for name, case in fixture["invalidation"].items()}
return {
"source_revision": revision,
"runner_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
"fixture_sha256": hashlib.sha256(fixture_bytes).hexdigest(),
"config_sha256": hashlib.sha256(config_bytes).hexdigest(),
"dataset": fixture["dataset"],
"prompt_path": "Gateway runs/wait to recording model",
"checkpoint_backend": "sqlite",
"token_estimate_method": "character estimate of injected blocks",
"date": DAY,
"passed": scope["passed"] and all(case["passed"] for case in invalidation.values()),
"scope_isolation": scope,
"invalidation": invalidation,
}

View File

@ -0,0 +1,117 @@
"""Regression checks for the offline benchmark's actual CLI and report."""
import json
import os
import subprocess
import sys
from pathlib import Path
import pytest
BACKEND = Path(__file__).resolve().parents[1]
MODULE = "scripts.benchmark.context_memory_comparison"
MANIFEST = BACKEND / "scripts/benchmark/context_memory_comparison/manifest.json"
def _run(tmp_path: Path, entrypoint: list[str]) -> tuple[int, dict]:
# Run the standalone Gateway in a fresh process. In particular, pytest's
# global stubs and another test's app config must not leak into this run.
env = {
**os.environ,
"PYTHONPATH": os.pathsep.join(str(BACKEND / path) for path in (".", "packages/harness", "packages/extension-api")),
}
result = subprocess.run(
[sys.executable, *entrypoint, "run-offline", "--output-dir", str(tmp_path)],
cwd=BACKEND,
env=env,
capture_output=True,
text=True,
timeout=60,
)
report = tmp_path / "report.json"
assert report.is_file(), result.stderr
assert {path.name for path in tmp_path.iterdir()} == {"report.json"}
return result.returncode, json.loads(report.read_text(encoding="utf-8"))
def test_offline_cli_records_real_gateway_context(tmp_path: Path) -> None:
fixture = json.loads(MANIFEST.read_text(encoding="utf-8"))
code, report = _run(tmp_path, ["-m", MODULE])
assert report["checkpoint_backend"] == "sqlite"
assert report["dataset"] == fixture["dataset"]
scope = report["scope_isolation"]
assert scope["passed"]
control = fixture["default_fact"]
for label, project in fixture["projects"].items():
turn = scope["threads"][label]
assert set(turn["present"]) == {control, project["instructions"], project["document"]}
assert all(turn["context_token_estimate"][tag] > 0 for tag in ("memory", "project", "documents"))
for label, fact in (("unassigned", control), ("custom_agent", fixture["agent_fact"])):
turn = scope["threads"][label]
assert turn["present"] == [fact]
assert turn["context_token_estimate"]["memory"] > 0
assert turn["context_token_estimate"]["project"] == turn["context_token_estimate"]["documents"] == 0
rejected = scope["project_only_fact"]
assert rejected["update_succeeded"] and rejected["passed"]
assert rejected["scope_rejections"] == 1
assert not rejected["stored"]
assert fixture["project_only_fact"] not in rejected["other_thread"]["present"]
history = fixture["thread_history_control"]
assert set(report["invalidation"]) == {"deletion", "replacement"}
for name, case in report["invalidation"].items():
data = fixture["invalidation"][name]
expected = {control}
if "new" in data:
expected.add(data["new"])
assert case["before"]["present"] == sorted({control, data["old"], history})
assert case["update_succeeded"]
assert set(case["stored"]) == set(case["new_thread"]["present"]) == expected
assert history in case["same_thread"]["present"]
# Pin the benchmark's verdict to the captured messages so a future
# production invalidation fix can change the observed behavior.
assert case["passed"] == (set(case["same_thread"]["present"]) == expected | {history})
assert report["passed"] == all(case["passed"] for case in report["invalidation"].values())
assert code == (0 if report["passed"] else 1)
@pytest.mark.parametrize("failure_mode", ["extraction_error", "no_scope_rejection"])
def test_scope_isolation_requires_a_scope_rejection(tmp_path: Path, failure_mode: str) -> None:
driver = """
import sys
from scripts.benchmark.context_memory_comparison import runner
from scripts.benchmark.context_memory_comparison.cli import main
generate = runner.OfflineModel._generate
failure_mode = sys.argv.pop(1)
def change_project_extraction(self, *args, **kwargs):
update = runner._SCRIPTED_UPDATE.get()
if self.model == "extract" and update and any(fact["scope"] == "project" for fact in update["newFacts"]):
if failure_mode == "extraction_error":
raise OSError("Synthetic extraction failure")
token = runner._SCRIPTED_UPDATE.set({**update, "newFacts": []})
try:
return generate(self, *args, **kwargs)
finally:
runner._SCRIPTED_UPDATE.reset(token)
return generate(self, *args, **kwargs)
runner.OfflineModel._generate = change_project_extraction
raise SystemExit(main())
"""
code, report = _run(tmp_path, ["-c", driver, failure_mode])
assert code == 1
scope = report["scope_isolation"]
assert all(turn["passed"] for turn in scope["threads"].values())
project_fact = scope["project_only_fact"]
assert not project_fact["passed"]
if failure_mode == "extraction_error":
assert not project_fact["update_succeeded"]
else:
assert project_fact["update_succeeded"]
assert not project_fact["stored"] and project_fact["other_thread"]["passed"]
assert project_fact["scope_rejections"] == 0
assert not scope["passed"] and not report["passed"]