mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-09 16:08:31 +00:00
feat(subagents): system-maintained delegation ledger to stop redundant re-delegation (#3877)
* feat(subagents): add delegations ledger field + reducer to ThreadState * feat(subagents): pure helpers to derive + format the delegation ledger * feat(subagents): DelegationLedgerMiddleware records + injects the ledger * feat(subagents): register DelegationLedgerMiddleware for lead when subagents enabled + docs * add runtime log * chore(subagents): make delegation-ledger injection log production-ready * test(subagents): make delegation-ledger registration tests config-free; refresh ultra replay golden for delegations channel * refactor(subagents): derive TERMINAL_STATUSES from SUBAGENT_STATUS_VALUES + pin it Make thread_state's TERMINAL_STATUSES a frozenset over the status contract's SUBAGENT_STATUS_VALUES instead of a hardcoded literal, so the terminal-status set can never drift from the contract. Add a pinning test asserting the derivation and that the non-terminal "in_progress" stays excluded. Addresses PR #3877 review. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Nan Gao
GitHub
parent
2453718acd
commit
af0d14d296
6 changed files with 412 additions and 9 deletions
|
|
@ -190,8 +190,8 @@ from deerflow.config import get_app_config
|
|||
- System prompt generated by `apply_prompt_template()` with skills, memory, and subagent instructions
|
||||
|
||||
**ThreadState** (`packages/harness/deerflow/agents/thread_state.py`):
|
||||
- Extends `AgentState` with: `sandbox`, `thread_data`, `title`, `artifacts`, `todos`, `uploaded_files`, `viewed_images`
|
||||
- Uses custom reducers: `merge_artifacts` (deduplicate), `merge_viewed_images` (merge/clear)
|
||||
- Extends `AgentState` with: `sandbox`, `thread_data`, `title`, `artifacts`, `todos`, `uploaded_files`, `viewed_images`, `delegations`
|
||||
- Uses custom reducers: `merge_artifacts` (deduplicate), `merge_viewed_images` (merge/clear), `merge_delegations` (upsert subagent delegation ledger by `task_id`; terminal status never downgraded)
|
||||
|
||||
**Runtime Configuration** (via `config.configurable`):
|
||||
- `thinking_enabled` - Enable model's extended thinking
|
||||
|
|
@ -227,13 +227,14 @@ Lead-agent middlewares are assembled in strict order across three functions: the
|
|||
17. **MemoryMiddleware** - Queues conversations for async memory update (filters to user + final AI responses)
|
||||
18. **ViewImageMiddleware** - *(optional, if the model supports vision)* Injects base64 image data before the LLM call
|
||||
19. **DeferredToolFilterMiddleware** - *(optional, if `tool_search.enabled`)* Hides deferred (MCP) tool schemas from the bound model until `tool_search` promotes them (reads per-thread promotions from `ThreadState.promoted`, hash-scoped)
|
||||
20. **SystemMessageCoalescingMiddleware** - Merges every SystemMessage into a single leading SystemMessage per request; provider-agnostic fix for strict backends (vLLM/SGLang/Qwen/Anthropic) that reject non-leading system messages. Touches the per-request payload only (checkpoint state unchanged); on midnight crossings only the latest `dynamic_context_reminder` SystemMessage survives
|
||||
21. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess `task` tool calls to enforce the `MAX_CONCURRENT_SUBAGENTS` limit
|
||||
22. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer
|
||||
23. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits
|
||||
24. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the safety/clarification tail
|
||||
25. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first
|
||||
26. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, interrupts via `Command(goto=END)` (must be last)
|
||||
20. **DelegationLedgerMiddleware** - *(optional, if `subagent_enabled`)* Maintains a system-maintained ledger of delegated subtasks in `ThreadState.delegations` (derives `description` + status from `task` tool calls and their result `subagent_status` on `after_model`) and re-injects it as a hidden `<system-reminder>` on `wrap_model_call` so the lead stops re-delegating the same work. Stored in state (not message history) so it survives summarization; re-injected ephemerally each call so no duplicate snapshots accumulate. Registered before SystemMessageCoalescingMiddleware so its injected SystemMessage is folded into the leading one
|
||||
21. **SystemMessageCoalescingMiddleware** - Merges every SystemMessage into a single leading SystemMessage per request; provider-agnostic fix for strict backends (vLLM/SGLang/Qwen/Anthropic) that reject non-leading system messages. Touches the per-request payload only (checkpoint state unchanged); on midnight crossings only the latest `dynamic_context_reminder` SystemMessage survives
|
||||
22. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess `task` tool calls to enforce the `MAX_CONCURRENT_SUBAGENTS` limit
|
||||
23. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer
|
||||
24. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits
|
||||
25. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the safety/clarification tail
|
||||
26. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first
|
||||
27. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, interrupts via `Command(goto=END)` (must be last)
|
||||
|
||||
### Configuration System
|
||||
|
||||
|
|
|
|||
|
|
@ -348,6 +348,14 @@ def build_middlewares(
|
|||
|
||||
middlewares.append(DeferredToolFilterMiddleware(deferred_setup.deferred_names, deferred_setup.catalog_hash))
|
||||
|
||||
# Maintain + inject the subagent delegation ledger (only when delegation is on).
|
||||
# Registered before coalescing so its injected <system-reminder> SystemMessage
|
||||
# is folded into the single leading SystemMessage for strict backends.
|
||||
if cfg.get("subagent_enabled", False):
|
||||
from deerflow.agents.middlewares.delegation_ledger_middleware import DelegationLedgerMiddleware
|
||||
|
||||
middlewares.append(DelegationLedgerMiddleware())
|
||||
|
||||
# Coalesce every SystemMessage into a single leading one before the request
|
||||
# reaches the provider. Strict backends (vLLM, SGLang, Qwen, Anthropic)
|
||||
# reject non-leading SystemMessages. See system_message_coalescing_middleware.py.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,109 @@
|
|||
"""Lead-agent middleware: a system-maintained ledger of delegated subtasks.
|
||||
|
||||
Issue: the lead repeatedly re-delegated the same research because the context
|
||||
held no durable record of what it had already dispatched (the record was lost
|
||||
to summarization). This middleware keeps that record in ThreadState (which
|
||||
summarization does not touch) and re-injects it into every model call, so the
|
||||
model always sees "already delegated: ..." and stops re-delegating.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import Any
|
||||
|
||||
from langchain.agents.middleware import AgentMiddleware, ModelRequest
|
||||
from langchain_core.messages import AIMessage, SystemMessage, ToolMessage
|
||||
from langgraph.runtime import Runtime
|
||||
|
||||
from deerflow.agents.thread_state import DelegationEntry
|
||||
from deerflow.subagents.status_contract import SUBAGENT_STATUS_KEY, extract_subagent_status
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def extract_delegations(messages: list) -> list[DelegationEntry]:
|
||||
"""Derive delegation entries from the visible message list, in dispatch order.
|
||||
|
||||
A ``task`` tool-call is a dispatch (status "in_progress"); its matching
|
||||
ToolMessage upgrades the status from the structured ``subagent_status`` kwarg,
|
||||
or by parsing the result text as a fallback (same contract the frontend uses).
|
||||
"""
|
||||
by_id: dict[str, DelegationEntry] = {}
|
||||
|
||||
for message in messages:
|
||||
if isinstance(message, AIMessage):
|
||||
for call in message.tool_calls or []:
|
||||
if call.get("name") != "task":
|
||||
continue
|
||||
task_id = call.get("id")
|
||||
if not task_id or task_id in by_id:
|
||||
continue
|
||||
args = call.get("args") or {}
|
||||
by_id[task_id] = {
|
||||
"task_id": task_id,
|
||||
"description": args.get("description") or "",
|
||||
"subagent_type": args.get("subagent_type") or "",
|
||||
"status": "in_progress",
|
||||
}
|
||||
elif isinstance(message, ToolMessage):
|
||||
entry = by_id.get(message.tool_call_id or "")
|
||||
if entry is None:
|
||||
continue
|
||||
status = message.additional_kwargs.get(SUBAGENT_STATUS_KEY)
|
||||
if not status:
|
||||
content = message.content if isinstance(message.content, str) else str(message.content)
|
||||
status = extract_subagent_status(content)
|
||||
if status:
|
||||
entry["status"] = status
|
||||
|
||||
return list(by_id.values())
|
||||
|
||||
|
||||
def format_delegation_block(entries: list[DelegationEntry]) -> str | None:
|
||||
"""Render the ledger as a hidden <system-reminder>, or None when empty."""
|
||||
if not entries:
|
||||
return None
|
||||
|
||||
lines = [
|
||||
"<system-reminder>",
|
||||
"<delegated_subtasks>",
|
||||
"You have ALREADY delegated these subtasks in this run. Do NOT re-delegate the same work; reuse or build on their results instead.",
|
||||
]
|
||||
for entry in entries:
|
||||
lines.append(f"- [{entry['status']}] ({entry['subagent_type']}) {entry['description']}")
|
||||
lines.append("</delegated_subtasks>")
|
||||
lines.append("</system-reminder>")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
class DelegationLedgerMiddleware(AgentMiddleware):
|
||||
"""Maintain (after_model) and inject (wrap_model_call) the delegation ledger."""
|
||||
|
||||
def _derive_update(self, state: Any) -> dict | None:
|
||||
entries = extract_delegations(list(state.get("messages", [])))
|
||||
return {"delegations": entries} if entries else None
|
||||
|
||||
def after_model(self, state: Any, runtime: Runtime | None = None) -> dict | None:
|
||||
return self._derive_update(state)
|
||||
|
||||
async def aafter_model(self, state: Any, runtime: Runtime | None = None) -> dict | None:
|
||||
return self._derive_update(state)
|
||||
|
||||
def _inject(self, request: ModelRequest) -> ModelRequest:
|
||||
entries = list(request.state.get("delegations") or [])
|
||||
block = format_delegation_block(entries)
|
||||
|
||||
if not block:
|
||||
logger.debug("delegation ledger: nothing to inject this call")
|
||||
return request
|
||||
logger.info("delegation ledger: injected %d subtask(s) into the model request", len(entries))
|
||||
reminder = SystemMessage(content=block, additional_kwargs={"hide_from_ui": True})
|
||||
return request.override(messages=[*request.messages, reminder])
|
||||
|
||||
def wrap_model_call(self, request: ModelRequest, handler: Callable[[ModelRequest], Any]) -> Any:
|
||||
return handler(self._inject(request))
|
||||
|
||||
async def awrap_model_call(self, request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[Any]]) -> Any:
|
||||
return await handler(self._inject(request))
|
||||
|
|
@ -2,6 +2,8 @@ from typing import Annotated, NotRequired, TypedDict
|
|||
|
||||
from langchain.agents import AgentState
|
||||
|
||||
from deerflow.subagents.status_contract import SUBAGENT_STATUS_VALUES
|
||||
|
||||
|
||||
class SandboxState(TypedDict):
|
||||
sandbox_id: NotRequired[str | None]
|
||||
|
|
@ -108,6 +110,42 @@ def merge_promoted(existing: PromotedTools | None, new: PromotedTools | None) ->
|
|||
}
|
||||
|
||||
|
||||
# Terminal subagent statuses. Derived from the single source of truth
|
||||
# (SUBAGENT_STATUS_VALUES) so the set can never drift from the status contract:
|
||||
# every value the contract enumerates is terminal, and the only non-terminal
|
||||
# status, "in_progress", is intentionally absent from the contract. merge_delegations
|
||||
# uses this to guard against status downgrades. test_delegation_ledger pins the
|
||||
# derivation so a future contract edit cannot silently desync this set.
|
||||
TERMINAL_STATUSES: frozenset[str] = frozenset(SUBAGENT_STATUS_VALUES)
|
||||
|
||||
|
||||
class DelegationEntry(TypedDict):
|
||||
task_id: str
|
||||
description: str
|
||||
subagent_type: str
|
||||
status: str # "in_progress" or one of TERMINAL_STATUSES
|
||||
|
||||
|
||||
def merge_delegations(
|
||||
existing: list[DelegationEntry] | None,
|
||||
new: list[DelegationEntry] | None,
|
||||
) -> list[DelegationEntry]:
|
||||
"""Reducer for the delegation ledger: upsert by task_id, preserve dispatch order.
|
||||
|
||||
A terminal status is never overwritten by a non-terminal one, so a later
|
||||
re-derivation from a partially-summarized message list cannot regress a
|
||||
finished subtask back to "in_progress".
|
||||
"""
|
||||
merged: dict[str, DelegationEntry] = {}
|
||||
for entry in list(existing or []) + list(new or []):
|
||||
task_id = entry["task_id"]
|
||||
prev = merged.get(task_id)
|
||||
if prev is not None and prev["status"] in TERMINAL_STATUSES and entry["status"] not in TERMINAL_STATUSES:
|
||||
continue
|
||||
merged[task_id] = {**prev, **entry} if prev else dict(entry)
|
||||
return list(merged.values())
|
||||
|
||||
|
||||
class ThreadState(AgentState):
|
||||
sandbox: SandboxStateField
|
||||
thread_data: NotRequired[ThreadDataState | None]
|
||||
|
|
@ -117,3 +155,4 @@ class ThreadState(AgentState):
|
|||
uploaded_files: NotRequired[list[dict] | None]
|
||||
viewed_images: Annotated[dict[str, ViewedImageData], merge_viewed_images] # image_path -> {base64, mime_type}
|
||||
promoted: Annotated[PromotedTools | None, merge_promoted]
|
||||
delegations: Annotated[list[DelegationEntry], merge_delegations]
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@
|
|||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"delegations",
|
||||
"messages",
|
||||
"viewed_images"
|
||||
]
|
||||
|
|
@ -21,6 +22,7 @@
|
|||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"delegations",
|
||||
"messages",
|
||||
"thread_data",
|
||||
"viewed_images"
|
||||
|
|
@ -30,6 +32,7 @@
|
|||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"delegations",
|
||||
"messages",
|
||||
"thread_data",
|
||||
"viewed_images"
|
||||
|
|
@ -39,6 +42,7 @@
|
|||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"delegations",
|
||||
"messages",
|
||||
"thread_data",
|
||||
"viewed_images"
|
||||
|
|
@ -48,6 +52,7 @@
|
|||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"delegations",
|
||||
"messages",
|
||||
"thread_data",
|
||||
"title",
|
||||
|
|
@ -58,6 +63,7 @@
|
|||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"delegations",
|
||||
"messages",
|
||||
"thread_data",
|
||||
"title",
|
||||
|
|
@ -68,6 +74,7 @@
|
|||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"delegations",
|
||||
"messages",
|
||||
"sandbox",
|
||||
"thread_data",
|
||||
|
|
@ -79,6 +86,7 @@
|
|||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"delegations",
|
||||
"messages",
|
||||
"sandbox",
|
||||
"thread_data",
|
||||
|
|
@ -90,6 +98,7 @@
|
|||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"delegations",
|
||||
"messages",
|
||||
"sandbox",
|
||||
"thread_data",
|
||||
|
|
@ -101,6 +110,7 @@
|
|||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"delegations",
|
||||
"messages",
|
||||
"sandbox",
|
||||
"thread_data",
|
||||
|
|
@ -112,6 +122,7 @@
|
|||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"delegations",
|
||||
"messages",
|
||||
"sandbox",
|
||||
"thread_data",
|
||||
|
|
@ -123,6 +134,7 @@
|
|||
"event": "values",
|
||||
"keys": [
|
||||
"artifacts",
|
||||
"delegations",
|
||||
"messages",
|
||||
"sandbox",
|
||||
"thread_data",
|
||||
|
|
|
|||
234
backend/tests/test_delegation_ledger.py
Normal file
234
backend/tests/test_delegation_ledger.py
Normal file
|
|
@ -0,0 +1,234 @@
|
|||
"""Tests for the subagent delegation ledger (parent issue: redundant delegation).
|
||||
|
||||
The ledger is a system-maintained record of "subtasks already delegated + their
|
||||
status", stored in ThreadState (so it survives summarization) and re-injected
|
||||
into context each model call so the lead stops re-delegating the same work.
|
||||
"""
|
||||
|
||||
from langchain_core.messages import AIMessage, ToolMessage
|
||||
|
||||
from deerflow.agents.middlewares.delegation_ledger_middleware import (
|
||||
extract_delegations,
|
||||
format_delegation_block,
|
||||
)
|
||||
from deerflow.agents.thread_state import TERMINAL_STATUSES, merge_delegations
|
||||
from deerflow.subagents.status_contract import SUBAGENT_STATUS_VALUES
|
||||
|
||||
|
||||
def _entry(task_id, status, description="d", subagent_type="general-purpose"):
|
||||
return {"task_id": task_id, "description": description, "subagent_type": subagent_type, "status": status}
|
||||
|
||||
|
||||
def _task_call(task_id, description, subagent_type="general-purpose"):
|
||||
return {"name": "task", "args": {"description": description, "subagent_type": subagent_type}, "id": task_id, "type": "tool_call"}
|
||||
|
||||
|
||||
def test_terminal_statuses_derived_from_status_contract():
|
||||
"""TERMINAL_STATUSES must stay the exact set the status contract enumerates.
|
||||
|
||||
Pins the derivation in thread_state.py: every value the contract declares is a
|
||||
terminal status, and the lone non-terminal status "in_progress" is never part of
|
||||
the contract. If a future contract edit adds a non-terminal value (or otherwise
|
||||
changes the set), this fails loudly instead of letting merge_delegations'
|
||||
downgrade guard silently desync.
|
||||
"""
|
||||
assert TERMINAL_STATUSES == frozenset(SUBAGENT_STATUS_VALUES)
|
||||
assert "in_progress" not in TERMINAL_STATUSES
|
||||
|
||||
|
||||
def test_merge_upserts_by_task_id_preserving_order():
|
||||
existing = [_entry("a", "in_progress"), _entry("b", "in_progress")]
|
||||
new = [_entry("b", "completed"), _entry("c", "in_progress")]
|
||||
|
||||
merged = merge_delegations(existing, new)
|
||||
|
||||
assert [e["task_id"] for e in merged] == ["a", "b", "c"]
|
||||
assert next(e for e in merged if e["task_id"] == "b")["status"] == "completed"
|
||||
|
||||
|
||||
def test_merge_does_not_downgrade_terminal_status():
|
||||
existing = [_entry("a", "completed")]
|
||||
new = [_entry("a", "in_progress")]
|
||||
|
||||
merged = merge_delegations(existing, new)
|
||||
|
||||
assert merged[0]["status"] == "completed"
|
||||
|
||||
|
||||
def test_merge_handles_none_inputs():
|
||||
assert merge_delegations(None, None) == []
|
||||
assert merge_delegations(None, [_entry("a", "in_progress")])[0]["task_id"] == "a"
|
||||
assert merge_delegations([_entry("a", "in_progress")], None)[0]["task_id"] == "a"
|
||||
|
||||
|
||||
def test_extract_records_dispatch_as_in_progress():
|
||||
msgs = [AIMessage(content="", tool_calls=[_task_call("call_1", "Research A")])]
|
||||
|
||||
entries = extract_delegations(msgs)
|
||||
|
||||
assert entries == [{"task_id": "call_1", "description": "Research A", "subagent_type": "general-purpose", "status": "in_progress"}]
|
||||
|
||||
|
||||
def test_extract_updates_status_from_tool_message_kwarg():
|
||||
msgs = [
|
||||
AIMessage(content="", tool_calls=[_task_call("call_1", "Research A")]),
|
||||
ToolMessage(content="Task Succeeded. Result: ok", tool_call_id="call_1", additional_kwargs={"subagent_status": "completed"}),
|
||||
]
|
||||
|
||||
entries = extract_delegations(msgs)
|
||||
|
||||
assert entries[0]["status"] == "completed"
|
||||
|
||||
|
||||
def test_extract_falls_back_to_parsing_content_when_kwarg_absent():
|
||||
msgs = [
|
||||
AIMessage(content="", tool_calls=[_task_call("call_1", "Research A")]),
|
||||
ToolMessage(content="Task failed. Error: boom", tool_call_id="call_1"),
|
||||
]
|
||||
|
||||
entries = extract_delegations(msgs)
|
||||
|
||||
assert entries[0]["status"] == "failed"
|
||||
|
||||
|
||||
def test_extract_ignores_non_task_tool_calls():
|
||||
msgs = [AIMessage(content="", tool_calls=[{"name": "web_search", "args": {}, "id": "x", "type": "tool_call"}])]
|
||||
|
||||
assert extract_delegations(msgs) == []
|
||||
|
||||
|
||||
def test_extract_preserves_dispatch_order_across_batches():
|
||||
msgs = [
|
||||
AIMessage(content="", tool_calls=[_task_call("call_1", "A"), _task_call("call_2", "B")]),
|
||||
AIMessage(content="", tool_calls=[_task_call("call_3", "C")]),
|
||||
]
|
||||
|
||||
assert [e["task_id"] for e in extract_delegations(msgs)] == ["call_1", "call_2", "call_3"]
|
||||
|
||||
|
||||
def test_format_block_lists_entries_and_returns_none_when_empty():
|
||||
assert format_delegation_block([]) is None
|
||||
|
||||
block = format_delegation_block(
|
||||
[
|
||||
{"task_id": "call_1", "description": "Research A", "subagent_type": "general-purpose", "status": "completed"},
|
||||
{"task_id": "call_2", "description": "Research B", "subagent_type": "general-purpose", "status": "in_progress"},
|
||||
]
|
||||
)
|
||||
|
||||
assert "<system-reminder>" in block
|
||||
assert "Research A" in block and "completed" in block
|
||||
assert "Research B" in block and "in_progress" in block
|
||||
assert "re-delegate" in block.lower() or "already delegated" in block.lower()
|
||||
|
||||
|
||||
def test_after_model_returns_derived_delegations():
|
||||
from deerflow.agents.middlewares.delegation_ledger_middleware import DelegationLedgerMiddleware
|
||||
|
||||
mw = DelegationLedgerMiddleware()
|
||||
state = {"messages": [AIMessage(content="", tool_calls=[_task_call("call_1", "Research A")])]}
|
||||
|
||||
update = mw.after_model(state, runtime=None)
|
||||
|
||||
assert update == {"delegations": [{"task_id": "call_1", "description": "Research A", "subagent_type": "general-purpose", "status": "in_progress"}]}
|
||||
|
||||
|
||||
def test_after_model_returns_none_when_no_delegations():
|
||||
from deerflow.agents.middlewares.delegation_ledger_middleware import DelegationLedgerMiddleware
|
||||
|
||||
mw = DelegationLedgerMiddleware()
|
||||
state = {"messages": [AIMessage(content="hi")]}
|
||||
|
||||
assert mw.after_model(state, runtime=None) is None
|
||||
|
||||
|
||||
class _FakeRequest:
|
||||
"""Minimal stand-in for ModelRequest: holds state + messages, supports override()."""
|
||||
|
||||
def __init__(self, state, messages):
|
||||
self.state = state
|
||||
self.messages = messages
|
||||
|
||||
def override(self, *, messages):
|
||||
return _FakeRequest(self.state, messages)
|
||||
|
||||
|
||||
def test_wrap_model_call_injects_ledger_block():
|
||||
from langchain_core.messages import SystemMessage
|
||||
|
||||
from deerflow.agents.middlewares.delegation_ledger_middleware import DelegationLedgerMiddleware
|
||||
|
||||
mw = DelegationLedgerMiddleware()
|
||||
captured = {}
|
||||
|
||||
def handler(req):
|
||||
captured["messages"] = req.messages
|
||||
return "RESPONSE"
|
||||
|
||||
state = {"delegations": [{"task_id": "call_1", "description": "Research A", "subagent_type": "general-purpose", "status": "completed"}]}
|
||||
req = _FakeRequest(state, [AIMessage(content="prev")])
|
||||
|
||||
result = mw.wrap_model_call(req, handler)
|
||||
|
||||
assert result == "RESPONSE"
|
||||
injected = captured["messages"]
|
||||
assert isinstance(injected[-1], SystemMessage)
|
||||
assert "Research A" in injected[-1].content
|
||||
assert len(injected) == 2
|
||||
|
||||
|
||||
def test_wrap_model_call_is_noop_without_delegations():
|
||||
from deerflow.agents.middlewares.delegation_ledger_middleware import DelegationLedgerMiddleware
|
||||
|
||||
mw = DelegationLedgerMiddleware()
|
||||
captured = {}
|
||||
|
||||
def handler(req):
|
||||
captured["messages"] = req.messages
|
||||
return "RESPONSE"
|
||||
|
||||
req = _FakeRequest({"delegations": []}, [AIMessage(content="prev")])
|
||||
|
||||
mw.wrap_model_call(req, handler)
|
||||
|
||||
assert len(captured["messages"]) == 1
|
||||
|
||||
|
||||
def _mw_names(middlewares):
|
||||
return [type(m).__name__ for m in middlewares]
|
||||
|
||||
|
||||
def _explicit_app_config():
|
||||
"""Build a minimal in-memory AppConfig (with one model) so build_middlewares
|
||||
never reads the gitignored, CI-absent config.yaml via get_app_config()."""
|
||||
from deerflow.config.app_config import AppConfig
|
||||
from deerflow.config.model_config import ModelConfig
|
||||
from deerflow.config.sandbox_config import SandboxConfig
|
||||
|
||||
model = ModelConfig(
|
||||
name="test-model",
|
||||
display_name="test-model",
|
||||
description=None,
|
||||
use="langchain_openai:ChatOpenAI",
|
||||
model="test-model",
|
||||
supports_thinking=False,
|
||||
supports_vision=False,
|
||||
)
|
||||
return AppConfig(models=[model], sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"))
|
||||
|
||||
|
||||
def test_middleware_registered_when_subagent_enabled():
|
||||
from deerflow.agents.lead_agent.agent import build_middlewares
|
||||
|
||||
middlewares = build_middlewares({"configurable": {"subagent_enabled": True}}, None, app_config=_explicit_app_config())
|
||||
names = _mw_names(middlewares)
|
||||
assert "DelegationLedgerMiddleware" in names
|
||||
# Must run before coalescing so its injected SystemMessage gets folded in.
|
||||
assert names.index("DelegationLedgerMiddleware") < names.index("SystemMessageCoalescingMiddleware")
|
||||
|
||||
|
||||
def test_middleware_absent_when_subagent_disabled():
|
||||
from deerflow.agents.lead_agent.agent import build_middlewares
|
||||
|
||||
middlewares = build_middlewares({"configurable": {"subagent_enabled": False}}, None, app_config=_explicit_app_config())
|
||||
assert "DelegationLedgerMiddleware" not in _mw_names(middlewares)
|
||||
Loading…
Add table
Add a link
Reference in a new issue