fix(summarization): build the summary model off the Gateway event loop (#7874)

Context compaction runs from abefore_model/aafter_model on the shared loop,
and its async candidate loop constructed the summary provider inline.
create_chat_model can read local CLI credentials while it is constructed
(Codex auth.json, Claude Code credential files), so an automatic compaction
stalled every other request for that read.

Split construction out of _model_for and add _amodel_for, which builds the
provider on the assembly pool; the sync path keeps building inline. Anchor
and failure-cache semantics are unchanged, so a broken candidate still falls
through to the next one, and inference still awaits on the loop.

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
yetuge 2026-11-04 21:54:47 +08:00 committed by GitHub
parent 2ceda05bf9
commit a46d9221f8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 166 additions and 27 deletions

View File

@ -33,6 +33,7 @@ from deerflow.extensions.notify import notify_context_compacted
from deerflow.models import create_chat_model
from deerflow.models.reasoning import resolve_reasoning_contract
from deerflow.runtime.events.catalog import MIDDLEWARE_SUMMARIZE_TAG
from deerflow.utils.assembly_io import run_assembly
logger = logging.getLogger(__name__)
_SUMMARY_TRIGGER_MESSAGE_NAME = "summary"
@ -258,18 +259,12 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
deduped.append(name)
return deduped
def _model_for(self, name: str | None) -> Any | None:
"""The nostream summary model for ``name``, built lazily and guarded.
def _build_model(self, name: str | None) -> Any | None:
"""Construct and tag the nostream summary model; ``None`` on failure.
Returns the pre-built anchor when ``name`` matches it (no rebuild), otherwise
constructs and caches. A construction failure is caught and cached as ``None``
so a broken candidate config never escapes the fail-open boundary, is never
retried this turn, and still lets the next candidate run.
A construction failure must not escape the fail-open boundary: the
caller caches the failure for this turn so the next candidate can run.
"""
if name == self._anchor_model_name:
return self._summary_model
if name in self._model_cache:
return self._model_cache[name]
try:
model = create_chat_model(
name=name,
@ -277,10 +272,40 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
app_config=self._app_config,
attach_tracing=False,
)
built = self._tag_nostream(model.with_config(tags=["middleware:summarize"]))
return self._tag_nostream(model.with_config(tags=["middleware:summarize"]))
except Exception:
logger.exception("Failed to build summary model %r; trying the next candidate", name)
built = None
return None
def _model_for(self, name: str | None) -> Any | None:
"""The nostream summary model for ``name`` on the synchronous path.
Returns the pre-built anchor when ``name`` matches it (no rebuild), otherwise
constructs and caches. A construction failure is cached as ``None`` so a
broken candidate config never escapes the fail-open boundary, is never
retried this turn, and still lets the next candidate run.
"""
if name == self._anchor_model_name:
return self._summary_model
if name in self._model_cache:
return self._model_cache[name]
built = self._build_model(name)
self._model_cache[name] = built
return built
async def _amodel_for(self, name: str | None) -> Any | None:
"""The nostream summary model for ``name``, built off the event loop.
``abefore_model``/``aafter_model`` run on the Gateway loop, and provider
construction can read local CLI credentials (Codex auth.json, Claude Code
credential files), so the build runs on the assembly pool. Anchor and
cache semantics stay identical to :meth:`_model_for`.
"""
if name == self._anchor_model_name:
return self._summary_model
if name in self._model_cache:
return self._model_cache[name]
built = await run_assembly(self._build_model, name)
self._model_cache[name] = built
return built
@ -392,7 +417,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
return reused
for index, name in enumerate(names):
text = await self._ainvoke_summary(
self._model_for(name),
await self._amodel_for(name),
prompt,
last=index == len(names) - 1,
model_name=name,

View File

@ -121,20 +121,22 @@ Display summaries use `strip_leading_think_blocks` before limiting text;
### Agent / Tool Assembly Off-Load
One-shot input-polish, follow-up-suggestion, and project-file-summary helpers,
skill moderation (`scan_skill_content`), the goal evaluator, and thread title
generation build providers through `run_assembly` before awaiting inference;
the run worker calls its cached evaluator factory the same way. Provider
construction can read local credentials; keep it off the event loop and the
default executor while preserving request context. Cancellation during
construction must not start inference. A scan without `app_config` (in-graph
`skill_manage`) reads the live config in the same pool, and its fail-closed
fallback reuses that config rather than reading it again on the loop. Other
model construction is separate and not covered here: manual compaction
(`compact_thread_context` builds the summarization middleware and its models
inline) and, inside graph runs, summarization and subagent construction.
skill moderation (`scan_skill_content`), the goal evaluator, thread title
generation, and in-graph context compaction build providers through
`run_assembly` before awaiting inference; the run worker calls its cached
evaluator factory the same way. Provider construction can read local
credentials; keep it off the event loop and the default executor while
preserving request context. Cancellation during construction must not start
inference. A scan without `app_config` (in-graph `skill_manage`) reads the live
config in the same pool, and its fail-closed fallback reuses that config rather
than reading it again on the loop. Other model construction is separate and
not covered here: manual compaction (`compact_thread_context` builds the
summarization middleware and its models inline) and, inside graph runs,
subagent construction.
Coverage: `tests/blocking_io/test_oneshot_llm_offloop.py`,
`tests/blocking_io/test_system_model_offloop.py` and
`tests/blocking_io/test_title_model_offloop.py`.
`tests/blocking_io/test_system_model_offloop.py`,
`tests/blocking_io/test_title_model_offloop.py` and
`tests/blocking_io/test_summarization_model_offloop.py`.
`start_run` also resolves the factory through `run_assembly` before admission:
cold imports cannot occupy the loop or its default pool, and import errors

View File

@ -0,0 +1,108 @@
"""Auto-compaction must build the summary provider off the Gateway event loop.
Compaction runs from ``abefore_model``/``aafter_model`` on the shared loop, and
``create_chat_model`` can read local CLI credentials while it is constructed
(Codex ``auth.json``, Claude Code credential files). The real Codex provider is
built here under the strict gate; tracing and inference stay offline.
"""
import json
import threading
from types import SimpleNamespace
import pytest
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from pydantic import Field
from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware
from deerflow.config.app_config import AppConfig
from deerflow.models import factory
from deerflow.models.openai_codex_provider import CodexChatModel
pytestmark = pytest.mark.asyncio
_SUMMARY = "Configured summary."
class AnchorModel(BaseChatModel):
"""The run model; used here only to prove the fallback was not reached."""
calls: list[str] = Field(default_factory=list)
@property
def _llm_type(self):
return "summarization-offloop-anchor"
def _generate(self, messages, stop=None, run_manager=None, **kwargs):
self.calls.append("anchor")
return ChatResult(generations=[ChatGeneration(message=AIMessage(content="Anchor summary."))])
async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
return self._generate(messages, stop=stop, run_manager=run_manager, **kwargs)
def _codex_config() -> AppConfig:
return AppConfig.model_validate(
{
"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
"models": [{"name": "fixture-codex", "model": "fixture-model", "use": "deerflow.models.openai_codex_provider:CodexChatModel"}],
}
)
@pytest.fixture
def codex(tmp_path, monkeypatch):
"""Real Codex construction (credential file IO included); offline tracing and inference."""
credentials = tmp_path / "auth.json"
credentials.write_text(json.dumps({"access_token": "fixture-token", "account_id": "fixture-account"}), encoding="utf-8")
monkeypatch.setenv("CODEX_AUTH_PATH", str(credentials))
monkeypatch.setattr(factory, "build_tracing_callbacks", lambda **kwargs: [])
state = SimpleNamespace(construction_thread=None, calls=[])
async def invoke(model, messages, config=None, **kwargs):
state.calls.append((model, threading.get_ident()))
return AIMessage(content=_SUMMARY)
monkeypatch.setattr(CodexChatModel, "ainvoke", invoke)
return state
def _middleware(anchor):
return DeerFlowSummarizationMiddleware(
model=anchor,
trigger=("messages", 100),
keep=("messages", 1),
trim_tokens_to_summarize=4096,
summary_prompt="{messages}",
app_config=_codex_config(),
configured_model_name="fixture-codex",
run_model_name="fixture-codex",
anchor_model_name="offline",
)
async def test_auto_compaction_builds_summary_model_off_loop(codex, monkeypatch):
anchor = AnchorModel()
middleware = _middleware(anchor)
real_build = DeerFlowSummarizationMiddleware._build_model
def recording_build(self, name):
codex.construction_thread = threading.get_ident()
return real_build(self, name)
monkeypatch.setattr(DeerFlowSummarizationMiddleware, "_build_model", recording_build)
summary = await middleware._asummarize_with([HumanMessage(content="old question", id="user"), AIMessage(content="old answer", id="assistant")])
# On the loop the gate rejects the credential read, the candidate is cached
# as unavailable, and compaction returns no summary at all.
assert summary == _SUMMARY
assert len(codex.calls) == 1
assert codex.construction_thread is not None
assert codex.construction_thread != threading.get_ident()
# Inference itself still awaits on the loop once the provider exists.
assert codex.calls[0][1] == threading.get_ident()
# The configured model was usable, so the run-model anchor never ran.
assert anchor.calls == []

View File

@ -381,7 +381,11 @@ async def test_async_summarization_observes_each_provider_attempt_and_live_store
middleware.trim_tokens_to_summarize = None
middleware._generation_candidate_names = lambda: ["first", "second"]
models = {"first": _Failing(), "second": _Working()}
middleware._model_for = lambda name: models[name]
async def _amodel_for(name):
return models[name]
middleware._amodel_for = _amodel_for
result = await middleware._asummarize_prompt("prompt", task_store=store)