mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-26 07:57:57 +00:00
fix(subagents): inject durable context before compaction (#4040)
* fix(subagents): inject durable context before compaction * fix(subagents): coalesce system messages after durable-context injection Address #4040 review: - append SystemMessageCoalescingMiddleware innermost on the subagent chain so the SystemMessage(authority) DurableContextMiddleware injects is merged into one leading system_message; otherwise the durable fix trades #4039's assistant-first 400 for a duplicate-system 400 on strict backends - add a two-system regression guard driving the real builder output through a strict model; assert exactly one leading SystemMessage - assert single-leading-system in the compaction integration test too - update the middleware count/last-element assertion (coalescer is now unconditionally last, removing the summarization-dependence ambiguity) - compare _skills_root against posixpath.normpath(container_path) - document the coalescer on the subagent chain in backend/AGENTS.md
This commit is contained in:
Nan Gao
GitHub
parent
79611673d7
commit
8fbf101de3
@ -707,7 +707,7 @@ Use `/compact` in the Web UI composer to summarize older context for the current
|
||||
|
||||
Complex tasks rarely fit in a single pass. DeerFlow decomposes them.
|
||||
|
||||
The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions. Sub-agents run in parallel when possible, report back structured results, and the lead agent synthesizes everything into a coherent output. When token usage tracking is enabled, completed sub-agent usage is attributed back to the dispatching step.
|
||||
The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions. Sub-agents run in parallel when possible, report back structured results, and the lead agent synthesizes everything into a coherent output. Long-running sub-agents compact older history when summarization is enabled and re-inject the summary as guarded, hidden durable context before continuing, so recent assistant/tool activity remains grounded in the task. When token usage tracking is enabled, completed sub-agent usage is attributed back to the dispatching step.
|
||||
|
||||
This is how DeerFlow handles tasks that take minutes to hours: a research task might fan out into a dozen sub-agents, each exploring a different angle, then converge into a single report — or a website — or a slide deck with generated visuals. One harness, many hands.
|
||||
|
||||
|
||||
@ -236,7 +236,7 @@ Lead-agent middlewares are assembled in strict order across three functions: the
|
||||
|
||||
14. **DynamicContextMiddleware** - Injects the current date (and optionally memory) as a `<system-reminder>` into the first HumanMessage, keeping the base system prompt fully static for prefix-cache reuse
|
||||
15. **SkillActivationMiddleware** - Detects strict `/skill-name task` syntax on the latest real user message, resolves only enabled and runtime-allowed skills, injects the `SKILL.md` body as hidden current-turn context, and records a `middleware:skill_activation` audit event
|
||||
16. **DurableContextMiddleware** - Captures `task` delegations into `ThreadState.delegations` (including in-progress dispatches and terminal result summaries) and loaded skill-file references (name/path/description, parsed in-memory - not the body) into `ThreadState.skill_context` before summarization can compact the paired tool-call/result messages, then projects durable context into each model request. Static authority rules are injected as a `SystemMessage`; untrusted field values (`summary_text`, delegation results, skill descriptions) are injected separately as a hidden `HumanMessage` data block so compressed history, delegated work, and which skills are active stay visible without being stored as `messages` or promoted to system-role instructions.
|
||||
16. **DurableContextMiddleware** - Captures `task` delegations into `ThreadState.delegations` (including in-progress dispatches and terminal result summaries) and loaded skill-file references (name/path/description, parsed in-memory - not the body) into `ThreadState.skill_context` before summarization can compact the paired tool-call/result messages, then projects durable context into each model request. Static authority rules are injected as a `SystemMessage`; untrusted field values (`summary_text`, delegation results, skill descriptions) are injected separately as a hidden `HumanMessage` data block so compressed history, delegated work, and which skills are active stay visible without being stored as `messages` or promoted to system-role instructions. `build_subagent_runtime_middlewares` also attaches this middleware immediately before subagent summarization so a compacted `summary_text` is projected ahead of a preserved assistant/tool tail instead of leaving strict providers with an assistant-first request.
|
||||
17. **SummarizationMiddleware** - *(optional, if enabled)* Context reduction when approaching token limits
|
||||
18. **TodoListMiddleware** - *(optional, if `is_plan_mode`)* Task tracking with the `write_todos` tool
|
||||
19. **TokenUsageMiddleware** - *(optional, if `token_usage.enabled`)* Records token usage metrics; subagent usage is merged back into the dispatching AIMessage by message position
|
||||
@ -380,7 +380,7 @@ Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runti
|
||||
**Flow**: `task()` tool → `SubagentExecutor` → background thread → poll 5s → SSE events → result
|
||||
**Events**: `task_started`, `task_running`, `task_completed`/`task_failed`/`task_timed_out`
|
||||
**Guardrail caps & `stop_reason` (#3875 Phase 2)**: three independent axes can end a subagent run early, and all now surface *why* through one additive field rather than a new status enum. **Turn axis**: `recursion_limit` on the subagent `run_config` equals `max_turns`, so exhausting the turn budget raises `GraphRecursionError` from `agent.astream`; `executor.py::_aexecute` catches it specifically (before the generic `except Exception`). **Token axis**: `TokenBudgetMiddleware` is attached per-agent via `build_subagent_runtime_middlewares` from `subagents.token_budget` (default `max_tokens` **coupled to `summarization.enabled`** — 1,000,000 when subagent summarization is on, 2,000,000 when off, warn at 0.7, hard-stop at 1.0; a user-set budget always wins regardless of the switch — #3875 Phase 3; a backstop against a subagent that burns tokens on trivial work). It does *not* raise: at the hard-stop threshold it strips the in-flight turn's tool calls, forces `finish_reason="stop"`, and lets the run complete naturally with a final answer. **Loop axis**: `LoopDetectionMiddleware` (attached at the same point) catches repeated identical tool-call sets — or one tool *type* called many times with varying args — and its hard-stop likewise strips `tool_calls` and forces a final answer without raising, recording `loop_capped`. Each guard exposes its cap on a per-`run_id` `consume_stop_reason(run_id)` accessor; `_aexecute` collects **every** middleware with that method (duck-typed via `hasattr`, so the executor has no import coupling to the guard classes) and surfaces the first non-`None` reason — adding a future guard needs no executor change. **Surfacing**: whichever axis fired, `_aexecute` stamps a normal status plus an additive reason — `completed` + `stop_reason=token_capped|turn_capped|loop_capped` when a usable final answer (or partial recovered from the last streamed chunk via `_extract_final_result` → `utils/messages.py::message_content_to_text`, returning a `"No response Generated"` sentinel when no text survived) was produced; `failed` + `stop_reason=turn_capped` when nothing usable survived. `SubagentResult.stop_reason` flows through `task_tool.py::_task_result_command` → `format_subagent_result_message` (renders `Task Succeeded (capped: ...)` / `Task failed (capped: ...)`) and `make_subagent_additional_kwargs`, which stamps the additive `subagent_stop_reason` key alongside the normal `subagent_status`. **Why additive, not an enum**: a new status value would break v1 consumers; an optional field is ignored by older frontends and ledger readers, so the cross-language contract (`contracts/subagent_status_contract.json` v2 + `subagents/status_contract.py` + `frontend/.../subtask-result.ts`, pinned by `test_status_values_match_contract` / `test_stop_reason_values_match_contract`) stays backward-compatible. The durable delegation ledger captures `stop_reason` onto the entry and renders model-facing guidance ("hit a guardrail cap with a partial result; reuse it, retry tighter, or raise the per-agent budget (`max_turns` / `token_budget`)") so the lead reuses a capped completion knowingly instead of mistaking it for a clean one. (Phase 1 shipped this surfacing as a `MAX_TURNS_REACHED` status enum in #3949; Phase 2 replaced that enum with the additive `stop_reason` field per the agreed design — the `max_turns_reached` status value and `SubagentStatus.MAX_TURNS_REACHED` are gone.)
|
||||
**Context compaction (#3875 Phase 3)**: subagents inherit `DeerFlowSummarizationMiddleware` via `build_subagent_runtime_middlewares`, gated on the **same** `summarization.enabled` switch the lead reads (one config covers both chains; trigger/keep/model/prompt come from the shared `summarization` config so they cannot drift). The factory is called with `skip_memory_flush=True` on the subagent path: the lead's `memory_flush_hook` (attached when `memory.enabled`) flushes pre-compaction messages into durable memory keyed by `thread_id`, and subagents share the parent's `thread_id`, so without skipping the hook a subagent's internal turns would pollute the **parent** thread's durable memory. Placement differs from the lead chain (lead appends summarization *before* the guard trio; subagent appends it *after*) — benign because the middleware implements only `before_model` (compaction) with no `after_model`/`consume_stop_reason`, so it cannot disturb the Phase 2 guard-cap stop-reason channel. Compaction rewrites the messages channel via `RemoveMessage(id=REMOVE_ALL_MESSAGES)`, which shrinks `len(messages)` below the step-capture cursor mid-run; `capture_new_step_messages` (see Step capture below) resets the cursor to the new tail on contraction so steps appended after the compaction point are not silently dropped.
|
||||
**Context compaction (#3875 Phase 3, #4039)**: subagents inherit `DeerFlowSummarizationMiddleware` via `build_subagent_runtime_middlewares`, gated on the **same** `summarization.enabled` switch the lead reads (one config covers both chains; trigger/keep/model/prompt come from the shared `summarization` config so they cannot drift). The subagent builder attaches `DurableContextMiddleware` immediately before summarization, using the same skills path/read-tool settings as the lead chain. Compaction stores the generated summary in `ThreadState.summary_text` rather than as a `messages` item; the durable-context wrapper therefore projects it into the next model request as guarded hidden human data. This is required when a message-count keep policy preserves only an assistant tool-call plus its tool results: without the injected summary the next request begins with assistant/tool history and strict OpenAI-compatible providers can reject it. Because `DurableContextMiddleware` inserts a second `SystemMessage(authority_contract)` after the subagent's leading system prompt, the builder also appends `SystemMessageCoalescingMiddleware` innermost (mirroring the lead chain, appended after the optional summarization middleware so it is unconditionally last) to merge every `SystemMessage` into one leading `system_message` — otherwise the durable fix would trade #4039's assistant-first HTTP 400 for a duplicate-system 400 on the same strict backends (#4040). The factory is called with `skip_memory_flush=True` on the subagent path: the lead's `memory_flush_hook` (attached when `memory.enabled`) flushes pre-compaction messages into durable memory keyed by `thread_id`, and subagents share the parent's `thread_id`, so without skipping the hook a subagent's internal turns would pollute the **parent** thread's durable memory. Placement differs from the lead chain (lead appends summarization *before* the guard trio; subagent appends it *after*) — benign because the middleware implements only `before_model` (compaction) with no `after_model`/`consume_stop_reason`, so it cannot disturb the Phase 2 guard-cap stop-reason channel. Compaction rewrites the messages channel via `RemoveMessage(id=REMOVE_ALL_MESSAGES)`, which shrinks `len(messages)` below the step-capture cursor mid-run; `capture_new_step_messages` (see Step capture below) resets the cursor to the new tail on contraction so steps appended after the compaction point are not silently dropped.
|
||||
**Step capture & persistence (#3779)**: `executor.py` captures both assistant turns (`AIMessage`) **and** tool outputs (`ToolMessage`) via `subagents/step_events.py::capture_new_step_messages`, which walks the *newly-appended tail* of each `stream_mode="values"` chunk (not just `messages[-1]`) so a multi-tool-call turn — where LangGraph's `ToolNode` appends several `ToolMessage`s in one super-step — keeps every tool output instead of dropping all but the last. `runtime/runs/worker.py::_SubagentEventBuffer` additionally persists these `task_*` custom events to the `RunEventStore` as `subagent.start`/`subagent.step`/`subagent.end` (`category="subagent"`, `task_id` in `metadata`). It **batches** writes via `put_batch` (flushing on a terminal `subagent.end`, at `FLUSH_THRESHOLD` events, and in the worker's `finally`) rather than one `put()` per step, since `put()` is a documented low-frequency path (per-thread advisory lock per call) and a deep subagent (`max_turns=150`) emits hundreds of steps on the hot stream loop. `build_subagent_step` caps both the per-step `text` and each tool call's serialized `args` at `SUBAGENT_STEP_MAX_CHARS` (flagged `truncated` / `args_truncated`) so a large `write_file`/`bash` payload can't produce an unbounded row. The dedicated category keeps them out of `list_messages` (the thread feed) while `list_events` returns them for the frontend's fetch-on-expand backfill. `list_events` accepts `task_id` (filters on `metadata["task_id"]` — SQL-side in `DbRunEventStore` via `event_metadata["task_id"].as_string()`, in-memory in the JSONL/memory stores) plus an `after_seq` forward cursor, so the card pages through one subagent's steps without the run-wide `limit` truncating the tail (no schema migration: the filter rides the existing run-scoped index). `step_events.py` is a pure, unit-tested layer (`build_subagent_step` / `subagent_run_event`). **History contraction (#3875 Phase 3)**: `capture_new_step_messages` assumes append-only growth, but `DeerFlowSummarizationMiddleware` rewrites the messages channel via `RemoveMessage(id=REMOVE_ALL_MESSAGES)`, shrinking `len(messages)` below the cursor mid-run. On contraction (`total < processed_count`) the cursor resets to the new tail; `capture_step_message`'s id/content dedup prevents re-emitting pre-compaction steps, so steps appended after the compaction point are still captured instead of being dropped until `total` overtakes the stale cursor.
|
||||
**Deferred MCP tools** (if `tool_search.enabled`): `SubagentExecutor._build_initial_state` assembles deferral after policy filtering via the shared `assemble_deferred_tools` (fail-closed), appends the `tool_search` tool, injects the `<available-deferred-tools>` section into the subagent's `SystemMessage`, and threads the setup to `_create_agent`, which attaches `McpRoutingMiddleware` (when PR1 routing metadata matches deferred tools) before `DeferredToolFilterMiddleware` through `build_subagent_runtime_middlewares(...)`. Subagents thus withhold full MCP schemas until promotion, same as the lead agent; each task run gets a fresh `ThreadState` so promotion is isolated per run
|
||||
**Checkpointer isolation**: Subagent graphs are compiled with `checkpointer=False` to avoid inheriting the parent run's checkpointer, since subagents are one-shot and never resume.
|
||||
|
||||
@ -379,6 +379,22 @@ def build_subagent_runtime_middlewares(
|
||||
|
||||
middlewares.append(SafetyFinishReasonMiddleware.from_config(safety_config))
|
||||
|
||||
# DurableContextMiddleware (#4039) — summarization stores compacted history in the
|
||||
# ``summary_text`` state channel instead of writing a summary message back
|
||||
# into ``messages``. Mirror the lead chain so subagents project that summary
|
||||
# into subsequent model requests; otherwise a message-count keep policy can
|
||||
# leave an assistant tool-call + tool-result tail with no leading user
|
||||
# context, which strict providers reject. The same middleware also keeps
|
||||
# skill references durable when their original read results are compacted.
|
||||
from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware
|
||||
|
||||
middlewares.append(
|
||||
DurableContextMiddleware(
|
||||
skills_container_path=app_config.skills.container_path,
|
||||
skill_file_read_tool_names=app_config.summarization.skill_file_read_tool_names,
|
||||
)
|
||||
)
|
||||
|
||||
# DeerFlowSummarizationMiddleware — subagents inherit none of the lead's
|
||||
# context compaction today (#3875 Phase 3): a deep-research subagent
|
||||
# (``max_turns`` up to 150) can accumulate >1M cumulative input before
|
||||
@ -420,4 +436,20 @@ def build_subagent_runtime_middlewares(
|
||||
if summarization_middleware is not None:
|
||||
middlewares.append(summarization_middleware)
|
||||
|
||||
# SystemMessageCoalescingMiddleware (#4040) — DurableContextMiddleware above
|
||||
# inserts a second ``SystemMessage(authority_contract)`` after the leading
|
||||
# system prompt (subagents carry their prompt as a leading ``SystemMessage``
|
||||
# in ``messages``, not via ``create_agent(system_prompt=...)``). Two system
|
||||
# messages — or a non-leading one — are exactly what the strict backends this
|
||||
# targets (vLLM/SGLang/Qwen/Anthropic) reject, so the durable fix would trade
|
||||
# #4039's assistant-first 400 for a duplicate-system 400. Mirror the lead
|
||||
# chain: append the coalescer innermost so it merges every SystemMessage into
|
||||
# one leading ``system_message`` on the outgoing request. It only rewrites the
|
||||
# per-request payload (no ``after_model``/``consume_stop_reason``), so it is
|
||||
# inert to the Phase 2 guard-cap channel, and must sit inner of
|
||||
# DurableContextMiddleware to observe the injected system message.
|
||||
from deerflow.agents.middlewares.system_message_coalescing_middleware import SystemMessageCoalescingMiddleware
|
||||
|
||||
middlewares.append(SystemMessageCoalescingMiddleware())
|
||||
|
||||
return middlewares
|
||||
|
||||
@ -1,3 +1,4 @@
|
||||
import posixpath
|
||||
import sys
|
||||
from types import ModuleType, SimpleNamespace
|
||||
|
||||
@ -148,18 +149,29 @@ def test_build_subagent_runtime_middlewares_threads_app_config_to_llm_middleware
|
||||
# ToolErrorHandling)
|
||||
# + 1 ReadBeforeWriteMiddleware + 1 LoopDetectionMiddleware
|
||||
# + 1 TokenBudgetMiddleware (subagents.token_budget enabled by default, #3875 Phase 2)
|
||||
# + 1 SafetyFinishReasonMiddleware (all enabled by default).
|
||||
# + 1 SafetyFinishReasonMiddleware + 1 DurableContextMiddleware
|
||||
# + 1 SystemMessageCoalescingMiddleware (all enabled by default).
|
||||
from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware
|
||||
from deerflow.agents.middlewares.safety_finish_reason_middleware import SafetyFinishReasonMiddleware
|
||||
from deerflow.agents.middlewares.system_message_coalescing_middleware import SystemMessageCoalescingMiddleware
|
||||
from deerflow.agents.middlewares.token_budget_middleware import TokenBudgetMiddleware
|
||||
from deerflow.agents.middlewares.tool_output_budget_middleware import ToolOutputBudgetMiddleware
|
||||
|
||||
assert len(middlewares) == 13
|
||||
assert len(middlewares) == 15
|
||||
assert isinstance(middlewares[0], FakeMiddleware) # InputSanitizationMiddleware stub
|
||||
assert isinstance(middlewares[1], ToolOutputBudgetMiddleware)
|
||||
assert any(isinstance(m, ToolErrorHandlingMiddleware) for m in middlewares)
|
||||
# The token-budget backstop is attached by default so the cap engages (#3875).
|
||||
assert any(isinstance(m, TokenBudgetMiddleware) for m in middlewares)
|
||||
assert isinstance(middlewares[-1], SafetyFinishReasonMiddleware)
|
||||
assert any(isinstance(m, SafetyFinishReasonMiddleware) for m in middlewares)
|
||||
# DurableContextMiddleware is present but not last: the coalescer (#4040) is
|
||||
# appended innermost so it can merge the SystemMessage DurableContext injects.
|
||||
# The coalescer is appended unconditionally (after the optional summarization
|
||||
# middleware), so it is the last element regardless of summarization.enabled —
|
||||
# unlike DurableContextMiddleware, which is only last when summarization is off.
|
||||
durable_idx = next(i for i, m in enumerate(middlewares) if isinstance(m, DurableContextMiddleware))
|
||||
assert isinstance(middlewares[-1], SystemMessageCoalescingMiddleware)
|
||||
assert durable_idx < len(middlewares) - 1
|
||||
|
||||
|
||||
def test_tool_progress_middleware_is_outer_relative_to_error_handling(monkeypatch: pytest.MonkeyPatch):
|
||||
@ -628,14 +640,19 @@ def test_subagent_runtime_middlewares_place_loop_detection_before_safety_finish(
|
||||
assert loop_idx < safety_idx
|
||||
|
||||
|
||||
def test_subagent_runtime_middlewares_attach_summarization_when_enabled(monkeypatch):
|
||||
"""Subagents must inherit the lead's DeerFlowSummarizationMiddleware so a
|
||||
long-running deep-research subagent compacts its context before it
|
||||
accumulates pathological input (#3875 Phase 3). Gated on the SAME
|
||||
``summarization.enabled`` switch the lead reads — one config covers both
|
||||
chains. The factory is patched at its source module so the test does not
|
||||
depend on a real chat model; only the gating + wiring is asserted."""
|
||||
def test_subagent_runtime_middlewares_attach_durable_context_before_summarization(monkeypatch):
|
||||
"""Subagents must project ``summary_text`` back into model requests after
|
||||
compaction, just like the lead agent does.
|
||||
|
||||
Without ``DurableContextMiddleware``, a message-count keep policy can
|
||||
retain only an assistant tool-call plus its tool results. The summary is
|
||||
stored in ``ThreadState.summary_text`` but never reaches the next request,
|
||||
so strict providers reject the assistant-first history. The durable
|
||||
context layer must use the same skill settings as the lead chain and run
|
||||
before summarization.
|
||||
"""
|
||||
from deerflow.agents.middlewares import summarization_middleware as sm
|
||||
from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware
|
||||
|
||||
sentinel = object()
|
||||
captured: dict[str, object] = {}
|
||||
@ -662,7 +679,168 @@ def test_subagent_runtime_middlewares_attach_summarization_when_enabled(monkeypa
|
||||
# skip_memory_flush=True so subagent-internal turns are not flushed into the
|
||||
# PARENT thread's durable memory (#3875 Phase 3 review).
|
||||
assert captured["skip_memory_flush"] is True
|
||||
assert sentinel in middlewares
|
||||
durable = [middleware for middleware in middlewares if isinstance(middleware, DurableContextMiddleware)]
|
||||
assert len(durable) == 1
|
||||
# ``_skills_root`` is ``posixpath.normpath(container_path)``, so compare against
|
||||
# the normalized form — a trailing slash / ``.`` / ``..`` in config would fail
|
||||
# a raw equality even though the wiring is correct.
|
||||
assert durable[0]._skills_root == posixpath.normpath(app_config.skills.container_path)
|
||||
assert durable[0]._skill_read_tool_names == frozenset(app_config.summarization.skill_file_read_tool_names)
|
||||
assert middlewares.index(durable[0]) < middlewares.index(sentinel)
|
||||
|
||||
|
||||
def test_subagent_compaction_injects_summary_before_assistant_tool_tail(monkeypatch):
|
||||
"""A three-tool turn with ``keep=4`` must remain provider-valid.
|
||||
|
||||
This reproduces the production failure shape: compaction preserves an
|
||||
assistant tool-call plus three tool results while removing the original
|
||||
system/user messages. The subagent chain must inject the generated summary
|
||||
as durable human context before that tail reaches the model.
|
||||
"""
|
||||
from langchain.agents import create_agent
|
||||
from langchain_core.language_models import BaseChatModel
|
||||
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
|
||||
from langchain_core.outputs import ChatGeneration, ChatResult
|
||||
|
||||
from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware
|
||||
from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware
|
||||
from deerflow.agents.middlewares.system_message_coalescing_middleware import SystemMessageCoalescingMiddleware
|
||||
from deerflow.agents.thread_state import ThreadState
|
||||
from deerflow.config.summarization_config import ContextSize, SummarizationConfig
|
||||
|
||||
class _StaticModel(BaseChatModel):
|
||||
text: str
|
||||
require_durable_summary: bool = False
|
||||
|
||||
@property
|
||||
def _llm_type(self) -> str:
|
||||
return "static"
|
||||
|
||||
def bind_tools(self, tools, **kwargs):
|
||||
return self
|
||||
|
||||
def _generate(self, messages, stop=None, run_manager=None, **kwargs):
|
||||
if self.require_durable_summary:
|
||||
first_ai = next(i for i, message in enumerate(messages) if isinstance(message, AIMessage))
|
||||
durable = [(i, message) for i, message in enumerate(messages) if isinstance(message, HumanMessage) and message.additional_kwargs.get("durable_context_data")]
|
||||
assert durable, "compacted summary must be injected into the subagent request"
|
||||
assert durable[0][0] < first_ai, "durable summary must precede the assistant/tool tail"
|
||||
assert "COMPRESSED_SUBAGENT_HISTORY" in durable[0][1].content
|
||||
# DurableContext injects a SystemMessage(authority); without the
|
||||
# coalescer the request would carry it as a second/non-leading
|
||||
# system message, which strict providers reject (#4040). Assert the
|
||||
# outgoing request is provider-valid: a single leading SystemMessage.
|
||||
system_indices = [i for i, message in enumerate(messages) if isinstance(message, SystemMessage)]
|
||||
assert system_indices == [0], f"request must have exactly one leading SystemMessage, got {system_indices}"
|
||||
return ChatResult(generations=[ChatGeneration(message=AIMessage(content=self.text))])
|
||||
|
||||
summary_model = _StaticModel(text="COMPRESSED_SUBAGENT_HISTORY")
|
||||
strict_model = _StaticModel(text="final answer", require_durable_summary=True)
|
||||
monkeypatch.setattr(
|
||||
"deerflow.agents.middlewares.summarization_middleware.create_chat_model",
|
||||
lambda **kwargs: summary_model,
|
||||
)
|
||||
|
||||
app_config = _make_app_config().model_copy(
|
||||
update={
|
||||
"summarization": SummarizationConfig(
|
||||
enabled=True,
|
||||
trigger=ContextSize(type="messages", value=5),
|
||||
keep=ContextSize(type="messages", value=4),
|
||||
)
|
||||
}
|
||||
)
|
||||
runtime_middlewares = build_subagent_runtime_middlewares(
|
||||
app_config=app_config,
|
||||
model_name="test-model",
|
||||
agent_name="general-purpose",
|
||||
)
|
||||
compaction_middlewares = [middleware for middleware in runtime_middlewares if isinstance(middleware, (DurableContextMiddleware, DeerFlowSummarizationMiddleware, SystemMessageCoalescingMiddleware))]
|
||||
agent = create_agent(
|
||||
model=strict_model,
|
||||
tools=[],
|
||||
middleware=compaction_middlewares,
|
||||
state_schema=ThreadState,
|
||||
)
|
||||
|
||||
tool_calls = [{"name": "web_search", "args": {"query": f"q{i}"}, "id": f"call_{i}", "type": "tool_call"} for i in range(3)]
|
||||
seed = [
|
||||
SystemMessage(content="subagent instructions", id="system"),
|
||||
HumanMessage(content="research three regions", id="human"),
|
||||
AIMessage(content="searching", tool_calls=tool_calls, id="assistant"),
|
||||
*[ToolMessage(content=f"result {i}", tool_call_id=f"call_{i}", id=f"tool_{i}") for i in range(3)],
|
||||
]
|
||||
|
||||
result = agent.invoke({"messages": seed})
|
||||
|
||||
assert result["summary_text"] == "COMPRESSED_SUBAGENT_HISTORY"
|
||||
assert result["messages"][-1].content == "final answer"
|
||||
|
||||
|
||||
def test_subagent_chain_coalesces_durable_authority_system_message(monkeypatch):
|
||||
"""The durable-context authority SystemMessage must not survive as a second one.
|
||||
|
||||
Subagents carry their system prompt as a leading ``SystemMessage`` in
|
||||
``messages`` (``create_agent(system_prompt=None)``), and
|
||||
``DurableContextMiddleware`` inserts ``SystemMessage(authority_contract)``
|
||||
directly after it whenever durable data (summary / delegations / skills) is
|
||||
present. That leaves two adjacent system messages — the exact non-leading /
|
||||
duplicate-system shape strict OpenAI-compatible providers reject and the
|
||||
same #4039 failure class the durable fix set out to avoid.
|
||||
|
||||
``build_subagent_runtime_middlewares`` must therefore pair durable context
|
||||
with ``SystemMessageCoalescingMiddleware`` (#4040). This drives the real
|
||||
builder output through a strict model and asserts the outgoing request keeps
|
||||
exactly one leading ``SystemMessage``. Remove the coalescer from the builder
|
||||
and the model sees ``[System(base), System(authority), ...]`` and this fails.
|
||||
"""
|
||||
from langchain.agents import create_agent
|
||||
from langchain_core.language_models import BaseChatModel
|
||||
from langchain_core.messages import AIMessage, SystemMessage, ToolMessage
|
||||
from langchain_core.outputs import ChatGeneration, ChatResult
|
||||
|
||||
from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware
|
||||
from deerflow.agents.middlewares.system_message_coalescing_middleware import SystemMessageCoalescingMiddleware
|
||||
from deerflow.agents.thread_state import ThreadState
|
||||
|
||||
seen: dict[str, list[int]] = {}
|
||||
|
||||
class _StrictModel(BaseChatModel):
|
||||
@property
|
||||
def _llm_type(self) -> str:
|
||||
return "strict"
|
||||
|
||||
def bind_tools(self, tools, **kwargs):
|
||||
return self
|
||||
|
||||
def _generate(self, messages, stop=None, run_manager=None, **kwargs):
|
||||
seen["system_indices"] = [i for i, message in enumerate(messages) if isinstance(message, SystemMessage)]
|
||||
return ChatResult(generations=[ChatGeneration(message=AIMessage(content="ok"))])
|
||||
|
||||
app_config = _make_app_config()
|
||||
runtime_middlewares = build_subagent_runtime_middlewares(
|
||||
app_config=app_config,
|
||||
model_name="test-model",
|
||||
agent_name="general-purpose",
|
||||
)
|
||||
# Isolate the two middlewares under test, preserving builder order. The
|
||||
# coalescer must come after (inner of) durable context to observe the
|
||||
# injected system message.
|
||||
chain = [m for m in runtime_middlewares if isinstance(m, (DurableContextMiddleware, SystemMessageCoalescingMiddleware))]
|
||||
assert [type(m).__name__ for m in chain] == ["DurableContextMiddleware", "SystemMessageCoalescingMiddleware"]
|
||||
|
||||
agent = create_agent(model=_StrictModel(), tools=[], middleware=chain, state_schema=ThreadState)
|
||||
|
||||
# A leading system prompt plus an assistant tool-call tail, with a summary
|
||||
# already in state so durable context injects its authority SystemMessage.
|
||||
seed = [
|
||||
SystemMessage(content="subagent instructions", id="system"),
|
||||
AIMessage(content="searching", tool_calls=[{"name": "web_search", "args": {"query": "x"}, "id": "call_0", "type": "tool_call"}], id="assistant"),
|
||||
ToolMessage(content="result", tool_call_id="call_0", id="tool_0"),
|
||||
]
|
||||
agent.invoke({"messages": seed, "summary_text": "COMPRESSED_SUBAGENT_HISTORY"})
|
||||
|
||||
assert seen["system_indices"] == [0], f"request must have a single leading SystemMessage, got {seen['system_indices']}"
|
||||
|
||||
|
||||
def test_subagent_runtime_middlewares_omit_summarization_when_factory_returns_none(monkeypatch):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user