mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-19 19:16:17 +00:00
fix(subagents): scale max_turns into the graph's super-step budget (#5485)
* fix(subagents): scale max_turns into the graph's super-step budget
max_turns was handed to LangGraph as recursion_limit, but the two count
different things. recursion_limit counts super-steps, one per graph node,
and create_agent compiles a node for every middleware lifecycle hook, so
one turn costs before_model + model + after_model + tools nodes — seven to
eight through the subagent chain. The built-in general-purpose agent's
max_turns=150 therefore bought about 18 tool-using turns before failing as
turn_capped, and every middleware added to the chain shrank the effective
budget again.
Resolve the limit from the chain each subagent was actually assembled with
(subagents/turn_budget.py) instead of passing the turn count through, so
raising max_turns buys the turns it names.
No config keys or defaults changed; existing max_turns values now grant
their full budget, bounded as before by subagents.timeout_seconds and
subagents.token_budget.
* fix(subagents): warn when a counted hook can jump the agent loop
Review follow-up. The per-turn cost is a flat multiplier over the straight
before_model -> model -> tools loop. A hook that declares can_jump_to and
returns {"jump_to": ...} re-enters the loop without traversing tools,
spending another before_model + model + after_model pass that buys no tool
result, so the resolved limit becomes a lower bound rather than an exact
budget — silently re-creating the short budget this translation fixes.
Measured against a compiled graph: with one jumping after_model hook,
three tool turns need the resolved limit plus one jump pass, and the run
raises GraphRecursionError at the resolved limit.
How often a jump fires is data-dependent and unbounded, so it cannot be
folded into the arithmetic. find_jumping_hooks reports the condition off
the same __can_jump_to__ attribute the factory reads, and the executor
warns when a counted hook declares one. Nothing in today's subagent chain
does, so this changes no budget.
* fix(subagents): detect jumps declared on agent-level hooks too
Review follow-up. find_jumping_hooks exempted before_agent/after_agent on
the grounds that a jump out of them lands in the loop the budget already
pays for. That does not hold on langchain 1.3.14:
- after_agent jumps re-enter the loop after it finished, and the hook runs
again on the next exit, so the extra passes are unbounded. Even
jump_to "end" is routed to exit_node, the head of the after_agent chain,
so it reruns the chain; destinations are no safe filter.
- a before_agent hook that stages a tool call and jumps to tools runs a
tools step no model turn paid for. It is O(1), but the resolved limit
has zero headroom, so one step caps the last turn.
The detector now scans every hook pair the factory wires jump edges for.
The compiled-graph pin is parametrized over after_model->model,
after_agent->model, after_agent->end and before_agent->tools, each raising
GraphRecursionError at the resolved limit and completing once the jump's
cost is added. No middleware in the subagent chain declares a jump on any
hook, so this changes no budget.
This commit is contained in:
Hyeonsang Cho
GitHub
parent
d87a1c1193
commit
53798b44cd
12
CHANGELOG.md
12
CHANGELOG.md
@ -582,6 +582,18 @@ This section accumulates work toward the **2.1.0** milestone
|
||||
|
||||
### Fixed
|
||||
|
||||
- **subagents:** Give `max_turns` the meaning operators read it as. It was
|
||||
handed to LangGraph as `recursion_limit`, which counts super-steps — one per
|
||||
graph node — while `create_agent` compiles a node for every middleware
|
||||
lifecycle hook, so one turn cost seven to eight steps through the subagent
|
||||
chain and the built-in `general-purpose` agent's `max_turns=150` bought about
|
||||
18 tool-using turns before failing as `turn_capped`. Every middleware added
|
||||
to the chain shrank the effective budget again. The executor now scales the
|
||||
configured turn count by the per-turn node count of the chain it actually
|
||||
assembled, so raising `max_turns` buys the turns it names. No config keys
|
||||
changed; existing `max_turns` values now grant their full budget, which can
|
||||
make a previously truncated subagent run longer, bounded as before by
|
||||
`subagents.timeout_seconds` and `subagents.token_budget`.
|
||||
- **scheduler:** Enforce the global `max_concurrent_runs` budget on SQLite,
|
||||
which previously only held on Postgres. Claiming a queued occurrence counts
|
||||
the executing rows and then promotes one row to `launching`, and Postgres
|
||||
|
||||
@ -397,6 +397,14 @@
|
||||
|
||||
### 修复
|
||||
|
||||
- **子智能体：** `max_turns` 现在真正表示运维人员理解的"轮次"。此前它被直接当作 LangGraph 的
|
||||
`recursion_limit` 传入，而后者统计的是 super-step——每个图节点一步，且 `create_agent` 会为每个
|
||||
中间件生命周期钩子编译出一个节点，因此在子智能体的中间件链上一轮要花掉 7~8 步：内置
|
||||
`general-purpose` 的 `max_turns=150` 实际只买到约 18 轮带工具调用的轮次，随后以 `turn_capped`
|
||||
结束；每往链上加一个中间件，有效预算还会再缩水一次。现在执行器会按实际组装出的中间件链的
|
||||
每轮节点数来换算配置的轮次，调高 `max_turns` 就能得到它所声明的轮次。没有配置项变化；既有的
|
||||
`max_turns` 取值现在会获得完整预算，因此原先被截断的子智能体运行可能变长，其上界仍由
|
||||
`subagents.timeout_seconds` 与 `subagents.token_budget` 约束。
|
||||
- **调度器：** 在 SQLite 上同样强制执行全局 `max_concurrent_runs`，此前该上限只在 Postgres 上成立。
|
||||
认领排队中的 occurrence 时，会先统计正在执行的行，再把其中一行提升为 `launching`，Postgres 用
|
||||
advisory lock 将这两步串行化。而 SQLite 的 deferred 事务直到那条提升用的 UPDATE 才占用 writer，
|
||||
|
||||
@ -29,7 +29,7 @@ executions are not checked, and acceptance never changes automatic retry policy.
|
||||
**Acceptance checklist path portability**: Paths are host-independent and raw `..` fails closed. Drive/UNC absolutes retain their class across `ntpath` normalization and cannot cross the root; drive-relative, shell-dependent provider/PSDrive, and POSIX-rooted `cd` on Windows are unprovable. Selection overlap separates pytest node IDs, normalizes safe `.`/duplicate separators, rejects `..`, and applies Windows casing to drive, UNC, PSDrive, or Windows-context paths; provider-qualified drive/UNC roots remain paths until a later node-ID split. POSIX paths and all node IDs keep their required case sensitivity. Ambiguous trailing-dot/space or 8.3 components, different volume IDs, cross-family pairs, and absolute/relative PSDrive pairs fail closed without execution/filesystem provenance; only same-volume, same-root-form paths compare lexically. Without shell provenance, raw commands reject backslashes; cmd `%VAR%`/`!VAR!`, `^`, `#`; Bash braces/tilde; and PowerShell splatting, typographic quotes, or unquoted parentheses. A lone `%` remains eligible. POSIX-only markers must short-circuit on Windows because pytest evaluates them at import.
|
||||
**Events**: `task_started`, `task_running`, `task_completed`/`task_failed`/`task_timed_out`
|
||||
**Handled LLM failures**: `LLMErrorHandlingMiddleware` deliberately converts provider/model exceptions into an `AIMessage` so the graph can end cleanly, stamping `additional_kwargs.deerflow_error_fallback=true` plus error metadata. Clean graph termination does not imply subagent success: `SubagentExecutor` inspects the last assistant message at terminalization and maps a marked fallback to `SubagentStatus.FAILED`, which then emits `task_failed` and the existing structured `subagent_error`. Only the marker is authoritative — error-looking assistant prose without it remains a normal completed result, so neither the executor nor frontend parses display text as a status protocol.
|
||||
**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.)
|
||||
**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**: `max_turns` counts turns, `recursion_limit` counts super-steps — one per graph node, and `create_agent` compiles a node per middleware hook — so `turn_budget.py::resolve_recursion_limit` scales the turn budget by the assembled chain's depth (`test_subagent_turn_budget.py` pins it against the real compiled graph); a verbatim hand-off bought `general-purpose` ~18 of its 150 turns. Exhausting it 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, #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. `subagent_run_event` rejects malformed chunks that lack a non-empty `task_id`; running chunks additionally require a non-negative integer `message_index` and a message object, so persisted records always satisfy the required lifecycle envelope. `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` applies the subagent name allow/deny list and assembly-time authorization before calling the shared `assemble_deferred_tools`, 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(...)`. Runtime skill policy is intentionally later and dynamic: `tool_search` may disclose/promote catalog metadata, but `SkillToolPolicyMiddleware` still removes or blocks any promoted business tool omitted by the active skill. 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
|
||||
|
||||
@ -22,9 +22,13 @@ class SubagentConfig:
|
||||
disabled for this subagent. Skill bodies and their allowed-tools
|
||||
policies take effect only after activation/loading at runtime.
|
||||
model: Model to use - 'inherit' uses parent's model.
|
||||
max_turns: Maximum agent turns before stopping. Built-in agents use the
|
||||
value set here (general-purpose=150, bash=60) unless the global
|
||||
``subagents.max_turns`` is set.
|
||||
max_turns: Maximum agent turns — model call plus the tools it runs —
|
||||
before stopping. Built-in agents use the value set here
|
||||
(general-purpose=150, bash=60) unless the global
|
||||
``subagents.max_turns`` is set. ``turn_budget.py`` converts this
|
||||
into the LangGraph ``recursion_limit`` that buys that many turns
|
||||
through the assembled middleware chain; it is not passed through as
|
||||
a super-step count.
|
||||
timeout_seconds: Bare fallback execution-time cap. For built-in agents the
|
||||
effective limit is the global ``subagents.timeout_seconds`` (default
|
||||
1800 = 30 min), layered on by the registry; this 900 only applies
|
||||
|
||||
@ -52,6 +52,7 @@ from deerflow.subagents.report_contract import (
|
||||
)
|
||||
from deerflow.subagents.step_events import capture_new_step_messages
|
||||
from deerflow.subagents.token_collector import SubagentTokenCollector
|
||||
from deerflow.subagents.turn_budget import find_jumping_hooks, resolve_recursion_limit
|
||||
from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, ensure_trace_context, resolve_trace_id
|
||||
from deerflow.tracing import build_tracing_callbacks, inject_langfuse_metadata
|
||||
from deerflow.utils.messages import message_content_to_text
|
||||
@ -921,6 +922,12 @@ class SubagentExecutor:
|
||||
# not just the first — because the v2 contract advertises more than one
|
||||
# cap reason.
|
||||
self._stop_reason_middlewares: list[Any] = []
|
||||
# LangGraph super-step budget that buys ``config.max_turns`` turns,
|
||||
# resolved in ``_create_agent`` once the middleware chain — and with it
|
||||
# the compiled graph's per-turn node count — is known. Stays ``None``
|
||||
# until then; ``_aexecute`` falls back to the raw turn count so a test
|
||||
# double replacing ``_create_agent`` still produces a runnable config.
|
||||
self._recursion_limit: int | None = None
|
||||
# What this subagent was assembled from, published to extension
|
||||
# observers at the end of ``_create_agent``. The prompt and skill set
|
||||
# are captured while ``_build_initial_state`` renders them because
|
||||
@ -992,6 +999,7 @@ class SubagentExecutor:
|
||||
# a list (not ``next(...)``) so every guard is checked and a later one
|
||||
# is picked up automatically.
|
||||
self._stop_reason_middlewares = [m for m in middlewares if hasattr(m, "consume_stop_reason")]
|
||||
self._recursion_limit = self._resolve_recursion_limit(middlewares)
|
||||
|
||||
# system_prompt is included in initial state messages (see _build_initial_state)
|
||||
# to avoid multiple SystemMessages which some LLM APIs don't support.
|
||||
@ -1013,6 +1021,44 @@ class SubagentExecutor:
|
||||
)
|
||||
return agent
|
||||
|
||||
def _resolve_recursion_limit(self, middlewares: list[Any]) -> int:
|
||||
"""Translate ``max_turns`` into the super-step budget it actually means.
|
||||
|
||||
``max_turns`` is the operator-facing policy ("how many times may this
|
||||
agent think and act"), while LangGraph's ``recursion_limit`` counts
|
||||
graph nodes. ``create_agent`` compiles one node per middleware
|
||||
lifecycle hook, so the two differ by the depth of the assembled chain —
|
||||
see ``turn_budget.py`` for the arithmetic. Resolving it here, from the
|
||||
chain this subagent was actually built with, keeps the budget stable as
|
||||
middlewares are added and lets a per-agent chain differ.
|
||||
"""
|
||||
recursion_limit = resolve_recursion_limit(self.config.max_turns, middlewares)
|
||||
logger.debug(
|
||||
"[trace=%s] Subagent %s turn budget: max_turns=%s -> recursion_limit=%s (%d middlewares)",
|
||||
self.trace_id,
|
||||
self.config.name,
|
||||
self.config.max_turns,
|
||||
recursion_limit,
|
||||
len(middlewares),
|
||||
)
|
||||
# A hook that declares ``can_jump_to`` — agent-level hooks included —
|
||||
# can leave the straight path through the graph, spending super-steps
|
||||
# the flat per-turn cost does not model, so the budget silently becomes
|
||||
# a lower bound. No middleware in the subagent chain declares one today;
|
||||
# say so loudly if that changes, rather than letting runs quietly cap
|
||||
# short again.
|
||||
jumping_hooks = find_jumping_hooks(middlewares)
|
||||
if jumping_hooks:
|
||||
logger.warning(
|
||||
"[trace=%s] Subagent %s has jump-declaring middleware hooks (%s); recursion_limit=%s is a lower bound for max_turns=%s, so the run may cap early",
|
||||
self.trace_id,
|
||||
self.config.name,
|
||||
", ".join(f"{name}.{hook}" for name, hook in jumping_hooks),
|
||||
recursion_limit,
|
||||
self.config.max_turns,
|
||||
)
|
||||
return recursion_limit
|
||||
|
||||
def _describe_assembly(
|
||||
self,
|
||||
*,
|
||||
@ -1450,7 +1496,10 @@ class SubagentExecutor:
|
||||
# namespace. Business consumers receive thread_id via ``context``
|
||||
# below instead.
|
||||
run_config: RunnableConfig = {
|
||||
"recursion_limit": self.config.max_turns,
|
||||
# Super-steps, not turns: ``_create_agent`` (just above) scaled
|
||||
# ``max_turns`` by the compiled chain's per-turn node count.
|
||||
# Unset only when that method was replaced by a test double.
|
||||
"recursion_limit": self._recursion_limit if self._recursion_limit is not None else self.config.max_turns,
|
||||
"callbacks": [collector],
|
||||
"tags": [collector_caller],
|
||||
}
|
||||
@ -1604,14 +1653,15 @@ class SubagentExecutor:
|
||||
)
|
||||
|
||||
except GraphRecursionError:
|
||||
# ``recursion_limit`` on run_config == ``self.config.max_turns``
|
||||
# (set above). Hitting it means the subagent exhausted its turn
|
||||
# budget. Route into the additive ``stop_reason`` channel (#3875
|
||||
# Phase 2) rather than a dedicated status enum (which would break v1
|
||||
# contract consumers). If the run streamed usable partial work,
|
||||
# surface it as ``completed``; otherwise ``failed``. Either way the
|
||||
# lead can tell "out of budget" from "broken subagent" without
|
||||
# parsing result text.
|
||||
# ``recursion_limit`` on run_config is ``self.config.max_turns``
|
||||
# scaled into super-steps (set above), so hitting it means the
|
||||
# subagent exhausted its turn budget — report the turn count, which
|
||||
# is the number the operator configured. Route into the additive
|
||||
# ``stop_reason`` channel (#3875 Phase 2) rather than a dedicated
|
||||
# status enum (which would break v1 contract consumers). If the run
|
||||
# streamed usable partial work, surface it as ``completed``;
|
||||
# otherwise ``failed``. Either way the lead can tell "out of budget"
|
||||
# from "broken subagent" without parsing result text.
|
||||
#
|
||||
# Prefer a guard's stop reason if one already fired this run: a
|
||||
# token-budget / loop hard-stop strips tool_calls to force a final
|
||||
|
||||
138
backend/packages/harness/deerflow/subagents/turn_budget.py
Normal file
138
backend/packages/harness/deerflow/subagents/turn_budget.py
Normal file
@ -0,0 +1,138 @@
|
||||
"""Translating a subagent's turn budget into a LangGraph recursion limit.
|
||||
|
||||
``max_turns`` is a per-agent policy an operator writes in ``config.yaml``:
|
||||
how many times this agent may think and act before it is cut off. LangGraph's
|
||||
``recursion_limit`` counts something else — super-steps, one per graph node
|
||||
executed — and ``create_agent`` compiles every middleware lifecycle hook into
|
||||
its own node (``{middleware}.before_model``, ``{middleware}.after_model``), so
|
||||
one turn costs
|
||||
|
||||
before_model nodes + ``model`` + after_model nodes + ``tools``
|
||||
|
||||
super-steps. The agent-level hooks add ``before_agent + after_agent`` on top,
|
||||
once per invocation rather than once per turn.
|
||||
|
||||
Passing ``max_turns`` straight through as ``recursion_limit`` therefore divides
|
||||
the operator's budget by the depth of the middleware chain — the subagent chain
|
||||
compiles seven to eight loop nodes, so ``max_turns=150`` bought roughly eighteen
|
||||
turns — and silently shrinks every agent's budget again each time a middleware
|
||||
is added. This module does the translation instead, deriving the multiplier from
|
||||
the chain that was actually assembled.
|
||||
|
||||
Hook participation is read the way LangChain's own factory reads it: a
|
||||
class-level identity check against :class:`AgentMiddleware`'s base
|
||||
implementation. A middleware that leaves a hook alone costs nothing, and an
|
||||
extension middleware wrapped by ``IsolatedMiddleware`` — which mirrors that
|
||||
identity precisely so LangChain sees the wrapper as the middleware it wraps —
|
||||
is counted exactly like its inner middleware.
|
||||
|
||||
The per-turn cost is a flat multiplier, which models the straight
|
||||
``before_agent -> (before_model -> model -> tools)* -> after_agent`` path and
|
||||
nothing else. A hook that declares ``can_jump_to`` and returns
|
||||
``{"jump_to": ...}`` leaves that path: out of a model hook it re-enters the loop
|
||||
without traversing ``tools``, spending another ``before_model + model +
|
||||
after_model`` pass that buys no tool result; out of ``after_agent`` it re-enters
|
||||
the loop after it has finished (or, for ``end``, reruns the ``after_agent``
|
||||
chain), and the hook runs again on the next exit; out of ``before_agent`` a
|
||||
``tools`` jump runs a ``tools`` step no turn paid for. Any of these makes the
|
||||
resolved limit a lower bound rather than an exact budget. How often a jump fires
|
||||
is data-dependent and unbounded, so it cannot be folded into the arithmetic;
|
||||
:func:`find_jumping_hooks` exposes the condition instead, and the subagent
|
||||
executor warns when a hook declares one. No middleware in today's subagent
|
||||
chain does.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Sequence
|
||||
from typing import Any
|
||||
|
||||
from langchain.agents.middleware import AgentMiddleware
|
||||
|
||||
# The nodes every turn traverses whatever the middleware chain looks like:
|
||||
# LangChain's ``model`` node, and the ``tools`` node the loop returns through.
|
||||
# The turn that answers without tool calls skips ``tools`` but instead pays the
|
||||
# graph's entry step, so the per-turn cost is the same either way — which is why
|
||||
# this is a flat multiplier and not a multiplier plus a one-off correction.
|
||||
_LOOP_NODES_PER_TURN = 2
|
||||
|
||||
# Sync/async hook pairs, in the grouping LangChain compiles them: one node per
|
||||
# pair, whichever side (or both) a middleware overrides.
|
||||
_LOOP_HOOK_PAIRS = (("before_model", "abefore_model"), ("after_model", "aafter_model"))
|
||||
_INVOCATION_HOOK_PAIRS = (("before_agent", "abefore_agent"), ("after_agent", "aafter_agent"))
|
||||
|
||||
# Where LangChain's ``hook_config(can_jump_to=...)`` decorator leaves its
|
||||
# declaration; the factory reads the same attribute off the same hook methods.
|
||||
_JUMP_DECLARATION_ATTR = "__can_jump_to__"
|
||||
|
||||
|
||||
def _implements(middleware: Any, hook_pair: tuple[str, str]) -> bool:
|
||||
"""Whether *middleware* overrides either side of one sync/async hook pair.
|
||||
|
||||
An object that does not carry the hook at all compiles to no node, so it
|
||||
counts as not implementing it — LangChain's own check assumes an
|
||||
:class:`AgentMiddleware` subclass and would raise on anything else.
|
||||
"""
|
||||
middleware_type = type(middleware)
|
||||
for hook in hook_pair:
|
||||
implementation = getattr(middleware_type, hook, None)
|
||||
if implementation is not None and implementation is not getattr(AgentMiddleware, hook, None):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _count_nodes(middlewares: Sequence[Any], hook_pairs: tuple[tuple[str, str], ...]) -> int:
|
||||
"""Nodes *middlewares* contribute across *hook_pairs* (one per implemented pair)."""
|
||||
return sum(1 for middleware in middlewares for hook_pair in hook_pairs if _implements(middleware, hook_pair))
|
||||
|
||||
|
||||
def count_turn_steps(middlewares: Sequence[Any]) -> int:
|
||||
"""Super-steps one agent turn costs with *middlewares* in the chain."""
|
||||
return _count_nodes(middlewares, _LOOP_HOOK_PAIRS) + _LOOP_NODES_PER_TURN
|
||||
|
||||
|
||||
def count_invocation_steps(middlewares: Sequence[Any]) -> int:
|
||||
"""Super-steps spent once per invocation, outside the agent loop."""
|
||||
return _count_nodes(middlewares, _INVOCATION_HOOK_PAIRS)
|
||||
|
||||
|
||||
def find_jumping_hooks(middlewares: Sequence[Any]) -> list[tuple[str, str]]:
|
||||
"""Hooks that declare a jump, as ``(middleware, hook)`` name pairs.
|
||||
|
||||
A jump spends super-steps the flat multiplier does not model, so a non-empty
|
||||
result means :func:`resolve_recursion_limit` is a lower bound. Every
|
||||
node-creating hook is inspected — the four LangChain's factory wires jump
|
||||
edges for — because none is safe to exempt: ``after_agent`` can re-enter the
|
||||
loop unboundedly, and even a once-per-invocation ``before_agent`` jump to
|
||||
``tools`` costs one unpriced step, which is enough to cap the last turn of an
|
||||
exact budget. It is the declaration that is reported, not the destinations:
|
||||
``end`` is not a safe exit either, since out of ``after_agent`` LangChain
|
||||
routes it back to the head of the ``after_agent`` chain. The declaration is
|
||||
read off the same attribute, on the same overridden methods, that LangChain's
|
||||
factory reads it from.
|
||||
"""
|
||||
jumping: list[tuple[str, str]] = []
|
||||
for middleware in middlewares:
|
||||
middleware_type = type(middleware)
|
||||
for hook_pair in _LOOP_HOOK_PAIRS + _INVOCATION_HOOK_PAIRS:
|
||||
for hook in hook_pair:
|
||||
# No identity check against the base method is needed here, unlike
|
||||
# in the node counting above: only ``hook_config`` writes this
|
||||
# attribute, and it is never on ``AgentMiddleware``'s own hooks, so
|
||||
# a middleware that leaves the hook alone reads back as no jump.
|
||||
if getattr(getattr(middleware_type, hook, None), _JUMP_DECLARATION_ATTR, None):
|
||||
jumping.append((middleware_type.__name__, hook))
|
||||
return jumping
|
||||
|
||||
|
||||
def resolve_recursion_limit(max_turns: int, middlewares: Sequence[Any]) -> int:
|
||||
"""The ``recursion_limit`` that buys *max_turns* turns through this chain.
|
||||
|
||||
A non-positive ``max_turns`` is clamped to one turn: LangGraph rejects a
|
||||
``recursion_limit`` below 1, and a misconfigured budget should still let the
|
||||
agent answer once rather than fail the run before it starts.
|
||||
|
||||
The result is exact for a chain with no jump-declaring hooks and a lower bound
|
||||
otherwise; see :func:`find_jumping_hooks`.
|
||||
"""
|
||||
return max(1, max_turns) * count_turn_steps(middlewares) + count_invocation_steps(middlewares)
|
||||
@ -390,6 +390,103 @@ class TestAgentConstruction:
|
||||
assert captured["agent"]["tools"] == []
|
||||
assert captured["agent"]["system_prompt"] is None # system_prompt is merged into initial state messages
|
||||
|
||||
def test_create_agent_scales_max_turns_into_a_super_step_budget(
|
||||
self,
|
||||
classes,
|
||||
base_config,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
):
|
||||
"""Regression: ``max_turns`` used to be passed to LangGraph verbatim.
|
||||
|
||||
``recursion_limit`` counts graph nodes and ``create_agent`` compiles one
|
||||
per middleware lifecycle hook, so a verbatim hand-off bought roughly
|
||||
``max_turns / chain_depth`` turns — about 18 of the built-in
|
||||
``general-purpose`` agent's 150.
|
||||
"""
|
||||
from langchain.agents.middleware import AgentMiddleware
|
||||
|
||||
from deerflow.subagents import executor as executor_module
|
||||
|
||||
SubagentExecutor = classes["SubagentExecutor"]
|
||||
|
||||
class _AfterModel(AgentMiddleware):
|
||||
def after_model(self, state, runtime):
|
||||
return None
|
||||
|
||||
middlewares = [_AfterModel(), _AfterModel()]
|
||||
|
||||
monkeypatch.setattr(executor_module, "create_chat_model", lambda **kwargs: object())
|
||||
monkeypatch.setattr(executor_module, "create_agent", lambda **kwargs: object())
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"deerflow.agents.middlewares.tool_error_handling_middleware",
|
||||
_module(
|
||||
"deerflow.agents.middlewares.tool_error_handling_middleware",
|
||||
build_subagent_runtime_middlewares=lambda **kwargs: middlewares,
|
||||
),
|
||||
)
|
||||
|
||||
executor = SubagentExecutor(
|
||||
config=base_config,
|
||||
tools=[],
|
||||
app_config=SimpleNamespace(models=[SimpleNamespace(name="default-model")]),
|
||||
parent_model="parent-model",
|
||||
)
|
||||
executor._create_agent()
|
||||
|
||||
# model + tools + two after_model nodes, once per turn.
|
||||
assert executor._recursion_limit == base_config.max_turns * 4
|
||||
|
||||
def test_create_agent_warns_when_a_counted_hook_can_jump(
|
||||
self,
|
||||
classes,
|
||||
base_config,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
caplog,
|
||||
):
|
||||
"""A jump re-enters the loop without traversing ``tools``.
|
||||
|
||||
The flat per-turn cost does not model that, so the resolved limit becomes
|
||||
a lower bound. Nothing in today's subagent chain declares a jump; if one
|
||||
ever does, the run must not quietly cap short the way it did before this
|
||||
translation existed.
|
||||
"""
|
||||
import logging
|
||||
|
||||
from langchain.agents.middleware import AgentMiddleware, hook_config
|
||||
|
||||
from deerflow.subagents import executor as executor_module
|
||||
|
||||
SubagentExecutor = classes["SubagentExecutor"]
|
||||
|
||||
class _Jumper(AgentMiddleware):
|
||||
@hook_config(can_jump_to=["model"])
|
||||
def after_model(self, state, runtime):
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(executor_module, "create_chat_model", lambda **kwargs: object())
|
||||
monkeypatch.setattr(executor_module, "create_agent", lambda **kwargs: object())
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"deerflow.agents.middlewares.tool_error_handling_middleware",
|
||||
_module(
|
||||
"deerflow.agents.middlewares.tool_error_handling_middleware",
|
||||
build_subagent_runtime_middlewares=lambda **kwargs: [_Jumper()],
|
||||
),
|
||||
)
|
||||
|
||||
executor = SubagentExecutor(
|
||||
config=base_config,
|
||||
tools=[],
|
||||
app_config=SimpleNamespace(models=[SimpleNamespace(name="default-model")]),
|
||||
parent_model="parent-model",
|
||||
)
|
||||
with caplog.at_level(logging.WARNING, logger=executor_module.logger.name):
|
||||
executor._create_agent()
|
||||
|
||||
assert "_Jumper.after_model" in caplog.text
|
||||
assert "lower bound" in caplog.text
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_load_skills_uses_explicit_app_config_for_skill_storage(
|
||||
self,
|
||||
@ -2122,6 +2219,56 @@ class TestAsyncExecutionPath:
|
||||
assert "regression-skill" in system_messages[0].content
|
||||
assert "Skill instruction text" not in system_messages[0].content
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_aexecute_sends_the_resolved_recursion_limit_to_the_graph(self, classes, base_config):
|
||||
"""The run config carries the scaled super-step budget ``_create_agent`` resolved."""
|
||||
from langchain_core.messages import AIMessage
|
||||
|
||||
SubagentExecutor = classes["SubagentExecutor"]
|
||||
captured_configs: list[dict] = []
|
||||
|
||||
async def capturing_astream(state, *, config, **kwargs):
|
||||
captured_configs.append(config)
|
||||
yield {"messages": [AIMessage(content="Done", id="msg-1")]}
|
||||
|
||||
mock_agent = MagicMock()
|
||||
mock_agent.astream = capturing_astream
|
||||
|
||||
executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread")
|
||||
|
||||
def create_agent_resolving_budget(*args, **kwargs):
|
||||
executor._recursion_limit = 77
|
||||
return mock_agent
|
||||
|
||||
with patch.object(executor, "_create_agent", side_effect=create_agent_resolving_budget):
|
||||
await executor._aexecute("Do something")
|
||||
|
||||
assert captured_configs[0]["recursion_limit"] == 77
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_aexecute_falls_back_to_the_turn_count_when_the_chain_is_unknown(self, classes, base_config, mock_agent, msg):
|
||||
"""A test double replacing ``_create_agent`` leaves no chain to measure.
|
||||
|
||||
The run must still get a usable limit rather than ``None``, which
|
||||
LangGraph would reject.
|
||||
"""
|
||||
SubagentExecutor = classes["SubagentExecutor"]
|
||||
captured_configs: list[dict] = []
|
||||
|
||||
async def capturing_astream(state, *, config, **kwargs):
|
||||
captured_configs.append(config)
|
||||
yield {"messages": [msg.ai("Done", msg_id="msg-1")]}
|
||||
|
||||
mock_agent.astream = capturing_astream
|
||||
|
||||
executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread")
|
||||
|
||||
with patch.object(executor, "_create_agent", return_value=mock_agent):
|
||||
await executor._aexecute("Do something")
|
||||
|
||||
assert executor._recursion_limit is None
|
||||
assert captured_configs[0]["recursion_limit"] == base_config.max_turns
|
||||
|
||||
|
||||
class TestSkillAllowedTools:
|
||||
@pytest.mark.anyio
|
||||
|
||||
281
backend/tests/test_subagent_turn_budget.py
Normal file
281
backend/tests/test_subagent_turn_budget.py
Normal file
@ -0,0 +1,281 @@
|
||||
"""Tests for translating a subagent turn budget into a LangGraph recursion limit.
|
||||
|
||||
Covers:
|
||||
- Per-turn and per-invocation node counting from a middleware chain
|
||||
- Sync-only, async-only, and both-sided hooks each costing exactly one node
|
||||
- ``resolve_recursion_limit`` arithmetic, including a non-positive budget
|
||||
- A pin test that compiles real ``create_agent`` graphs and binary-searches the
|
||||
smallest ``recursion_limit`` that completes N turns, so the formula is checked
|
||||
against the installed LangChain rather than against its documentation
|
||||
- Jump detection across every hook LangChain wires a jump edge for, and the
|
||||
shortfall each kind of jump causes against a compiled graph
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from langchain.agents import create_agent
|
||||
from langchain.agents.middleware import AgentMiddleware, hook_config
|
||||
from langchain_core.language_models import BaseChatModel
|
||||
from langchain_core.messages import AIMessage
|
||||
from langchain_core.outputs import ChatGeneration, ChatResult
|
||||
from langchain_core.tools import tool
|
||||
from langgraph.errors import GraphRecursionError
|
||||
|
||||
from deerflow.subagents.turn_budget import count_invocation_steps, count_turn_steps, find_jumping_hooks, resolve_recursion_limit
|
||||
|
||||
|
||||
def _middleware(name: str, *hooks: str) -> AgentMiddleware:
|
||||
"""Build a no-op middleware overriding exactly *hooks*."""
|
||||
namespace = {hook: (lambda self, state, runtime: None) for hook in hooks}
|
||||
return type(name, (AgentMiddleware,), namespace)()
|
||||
|
||||
|
||||
class TestNodeCounting:
|
||||
def test_bare_chain_costs_model_plus_tools(self):
|
||||
assert count_turn_steps([]) == 2
|
||||
assert count_invocation_steps([]) == 0
|
||||
|
||||
def test_each_model_hook_adds_one_node(self):
|
||||
chain = [_middleware("Before", "before_model"), _middleware("After", "after_model")]
|
||||
|
||||
assert count_turn_steps(chain) == 4
|
||||
|
||||
def test_one_middleware_implementing_both_model_hooks_adds_two_nodes(self):
|
||||
chain = [_middleware("Both", "before_model", "after_model")]
|
||||
|
||||
assert count_turn_steps(chain) == 4
|
||||
|
||||
def test_async_only_hook_costs_the_same_as_sync(self):
|
||||
"""LangChain compiles one node per sync/async pair, whichever side exists."""
|
||||
sync_only = [_middleware("Sync", "after_model")]
|
||||
async_only = [_middleware("Async", "aafter_model")]
|
||||
both_sides = [_middleware("Both", "after_model", "aafter_model")]
|
||||
|
||||
assert count_turn_steps(sync_only) == count_turn_steps(async_only) == count_turn_steps(both_sides) == 3
|
||||
|
||||
def test_agent_hooks_are_charged_once_per_invocation_not_per_turn(self):
|
||||
chain = [_middleware("Lifecycle", "before_agent", "after_agent")]
|
||||
|
||||
assert count_turn_steps(chain) == 2
|
||||
assert count_invocation_steps(chain) == 2
|
||||
|
||||
def test_middleware_without_lifecycle_hooks_is_free(self):
|
||||
"""A wrap-only middleware runs inside the model node, so it adds none."""
|
||||
|
||||
class WrapOnly(AgentMiddleware):
|
||||
def wrap_model_call(self, request, handler):
|
||||
return handler(request)
|
||||
|
||||
assert count_turn_steps([WrapOnly()]) == 2
|
||||
assert count_invocation_steps([WrapOnly()]) == 0
|
||||
|
||||
def test_object_without_hooks_contributes_nothing(self):
|
||||
"""A duck-typed entry compiles to no node, so a missing hook is not an override."""
|
||||
assert count_turn_steps([object()]) == 2
|
||||
assert count_invocation_steps([object()]) == 0
|
||||
|
||||
|
||||
class TestResolveRecursionLimit:
|
||||
def test_scales_turns_by_the_assembled_chain(self):
|
||||
"""Regression: passing ``max_turns`` through divided the budget by chain depth."""
|
||||
chain = [
|
||||
_middleware("Before", "before_model"),
|
||||
_middleware("After1", "after_model"),
|
||||
_middleware("After2", "after_model"),
|
||||
_middleware("After3", "after_model"),
|
||||
_middleware("After4", "after_model"),
|
||||
]
|
||||
|
||||
assert resolve_recursion_limit(150, chain) == 150 * 7
|
||||
|
||||
def test_adds_invocation_steps_on_top_of_the_turn_budget(self):
|
||||
chain = [_middleware("Loop", "before_model"), _middleware("Lifecycle", "before_agent", "after_agent")]
|
||||
|
||||
assert resolve_recursion_limit(10, chain) == 10 * 3 + 2
|
||||
|
||||
def test_non_positive_budget_still_buys_one_turn(self):
|
||||
"""LangGraph rejects a limit below 1; a misconfigured budget must not fail the run."""
|
||||
assert resolve_recursion_limit(0, []) == 2
|
||||
assert resolve_recursion_limit(-5, []) == 2
|
||||
|
||||
|
||||
class TestFindJumpingHooks:
|
||||
"""A jump re-enters the loop without traversing ``tools``.
|
||||
|
||||
The flat per-turn multiplier does not model that, and how often a jump fires
|
||||
is data-dependent, so the condition is reported rather than priced in.
|
||||
"""
|
||||
|
||||
def test_reports_the_middleware_and_hook_that_declares_a_jump(self):
|
||||
class Jumper(AgentMiddleware):
|
||||
@hook_config(can_jump_to=["model"])
|
||||
def after_model(self, state, runtime):
|
||||
return None
|
||||
|
||||
assert find_jumping_hooks([Jumper()]) == [("Jumper", "after_model")]
|
||||
|
||||
def test_reports_an_async_hook_declaration(self):
|
||||
class AsyncJumper(AgentMiddleware):
|
||||
@hook_config(can_jump_to=["end"])
|
||||
async def abefore_model(self, state, runtime):
|
||||
return None
|
||||
|
||||
assert find_jumping_hooks([AsyncJumper()]) == [("AsyncJumper", "abefore_model")]
|
||||
|
||||
def test_plain_hooks_declare_no_jump(self):
|
||||
chain = [_middleware("Before", "before_model"), _middleware("After", "after_model")]
|
||||
|
||||
assert find_jumping_hooks(chain) == []
|
||||
|
||||
@pytest.mark.parametrize("hook", ["before_agent", "abefore_agent", "after_agent", "aafter_agent"])
|
||||
def test_agent_level_hooks_are_reported(self, hook):
|
||||
"""They run once per invocation, but a jump out of them is not free.
|
||||
|
||||
``after_agent`` can re-enter the loop after it finished, and a
|
||||
``before_agent`` jump to ``tools`` runs a step no turn paid for.
|
||||
"""
|
||||
|
||||
async def async_hook(self, state, runtime):
|
||||
return None
|
||||
|
||||
def sync_hook(self, state, runtime):
|
||||
return None
|
||||
|
||||
implementation = async_hook if hook in {"abefore_agent", "aafter_agent"} else sync_hook
|
||||
LifecycleJumper = type("LifecycleJumper", (AgentMiddleware,), {hook: hook_config(can_jump_to=["model"])(implementation)})
|
||||
|
||||
assert find_jumping_hooks([LifecycleJumper()]) == [("LifecycleJumper", hook)]
|
||||
|
||||
def test_object_without_hooks_is_not_reported(self):
|
||||
assert find_jumping_hooks([object()]) == []
|
||||
|
||||
|
||||
class _ScriptedModel(BaseChatModel):
|
||||
"""Emits ``turns - 1`` tool calls, then a final text answer."""
|
||||
|
||||
turns: int
|
||||
calls: int = 0
|
||||
|
||||
@property
|
||||
def _llm_type(self) -> str:
|
||||
return "scripted"
|
||||
|
||||
def bind_tools(self, tools, **kwargs):
|
||||
return self
|
||||
|
||||
def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult:
|
||||
self.calls += 1
|
||||
if self.calls < self.turns:
|
||||
message = AIMessage(content="", tool_calls=[{"name": "ping", "args": {}, "id": f"call-{self.calls}"}])
|
||||
else:
|
||||
message = AIMessage(content="done")
|
||||
return ChatResult(generations=[ChatGeneration(message=message)])
|
||||
|
||||
|
||||
@tool
|
||||
def ping() -> str:
|
||||
"""Return a fixed token."""
|
||||
return "pong"
|
||||
|
||||
|
||||
def _smallest_limit_completing(chain: list[AgentMiddleware], turns: int) -> int:
|
||||
"""Binary-search the smallest ``recursion_limit`` that completes *turns* turns."""
|
||||
low, high = 1, 512
|
||||
while low < high:
|
||||
candidate = (low + high) // 2
|
||||
agent = create_agent(model=_ScriptedModel(turns=turns), tools=[ping], middleware=chain, checkpointer=False)
|
||||
try:
|
||||
agent.invoke({"messages": [("user", "go")]}, {"recursion_limit": candidate})
|
||||
except GraphRecursionError:
|
||||
low = candidate + 1
|
||||
else:
|
||||
high = candidate
|
||||
return low
|
||||
|
||||
|
||||
class TestFormulaMatchesCompiledGraph:
|
||||
"""Pins the arithmetic against the installed LangChain, not against its docs.
|
||||
|
||||
If LangChain changes how it compiles middleware hooks into nodes, the
|
||||
subagent turn budget silently drifts again. These cases fail instead.
|
||||
"""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"hooks",
|
||||
[
|
||||
(),
|
||||
(("Before", "before_model"),),
|
||||
(("After", "after_model"),),
|
||||
(("Before", "before_model"), ("After1", "after_model"), ("After2", "after_model")),
|
||||
(("Lifecycle", "before_agent", "after_agent"), ("After", "after_model")),
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("turns", [1, 3])
|
||||
def test_resolved_limit_is_exactly_what_the_graph_needs(self, hooks, turns):
|
||||
chain = [_middleware(*spec) for spec in hooks]
|
||||
|
||||
assert resolve_recursion_limit(turns, chain) == _smallest_limit_completing(chain, turns)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("hook", "destination", "count_model_calls", "jump_cost"),
|
||||
[
|
||||
# Re-enters the model without traversing ``tools``: another
|
||||
# before_model + model + after_model pass, here model + after_model.
|
||||
pytest.param("after_model", "model", False, 2, id="after_model-to-model"),
|
||||
# Re-enters the loop after it finished: a model pass, then the
|
||||
# after_agent node again on the way out.
|
||||
pytest.param("after_agent", "model", False, 2, id="after_agent-to-model"),
|
||||
# ``end`` routes back to the head of the after_agent chain, so even
|
||||
# the exit is not free — why detection ignores destinations.
|
||||
pytest.param("after_agent", "end", False, 1, id="after_agent-to-end"),
|
||||
# Once per invocation, but the staged tool call runs a ``tools``
|
||||
# step outside any model turn.
|
||||
pytest.param("before_agent", "tools", True, 1, id="before_agent-to-tools"),
|
||||
],
|
||||
)
|
||||
def test_a_jumping_hook_makes_the_limit_a_lower_bound(self, hook, destination, count_model_calls, jump_cost):
|
||||
"""Why ``find_jumping_hooks`` exists, pinned against a real graph.
|
||||
|
||||
The loop cases count progress off the ToolMessages actually in state, so
|
||||
a jump that re-enters the model buys no progress and is pure overhead,
|
||||
the retry/repair shape. The ``before_agent`` case stages a tool call the
|
||||
model never made, so there the model counts its own calls instead —
|
||||
otherwise the staged tool result would pass for one of its turns.
|
||||
"""
|
||||
|
||||
class _ToolProgressModel(_ScriptedModel):
|
||||
def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult:
|
||||
executed = sum(1 for message in messages if getattr(message, "type", None) == "tool")
|
||||
if executed < self.turns:
|
||||
message = AIMessage(content="", tool_calls=[{"name": "ping", "args": {}, "id": f"call-{executed}"}])
|
||||
else:
|
||||
message = AIMessage(content="done")
|
||||
return ChatResult(generations=[ChatGeneration(message=message)])
|
||||
|
||||
def jump_once(self, state, runtime):
|
||||
if self.jumped:
|
||||
return None
|
||||
self.jumped = True
|
||||
update = {"jump_to": destination}
|
||||
if destination == "tools":
|
||||
update["messages"] = [AIMessage(content="", tool_calls=[{"name": "ping", "args": {}, "id": "staged"}])]
|
||||
return update
|
||||
|
||||
_JumpOnce = type(
|
||||
"_JumpOnce",
|
||||
(AgentMiddleware,),
|
||||
{"jumped": False, hook: hook_config(can_jump_to=[destination])(jump_once)},
|
||||
)
|
||||
|
||||
tool_turns = 3
|
||||
budget_turns = tool_turns + 1 # the tool turns plus the turn that answers
|
||||
|
||||
def run(limit):
|
||||
model = _ScriptedModel(turns=budget_turns) if count_model_calls else _ToolProgressModel(turns=tool_turns)
|
||||
create_agent(model=model, tools=[ping], middleware=[_JumpOnce()], checkpointer=False).invoke({"messages": [("user", "go")]}, {"recursion_limit": limit})
|
||||
|
||||
resolved = resolve_recursion_limit(budget_turns, [_JumpOnce()])
|
||||
with pytest.raises(GraphRecursionError):
|
||||
run(resolved)
|
||||
run(resolved + jump_cost)
|
||||
|
||||
assert find_jumping_hooks([_JumpOnce()]) == [("_JumpOnce", hook)]
|
||||
@ -1708,7 +1708,9 @@ subagent_runtime:
|
||||
# # Default timeout (seconds) for built-in subagents (default: 1800 = 30 min).
|
||||
# # Custom agents use their own timeout_seconds (default 900) unless overridden.
|
||||
# timeout_seconds: 1800
|
||||
# # Optional global max-turn override for all subagents.
|
||||
# # Optional global max-turn override for all subagents. One turn is one
|
||||
# # model call plus the tools it runs; the turn count is converted into
|
||||
# # LangGraph's step budget, so this is the number of turns you actually get.
|
||||
# # Built-in defaults: general-purpose=150, bash=60. Leave unset to keep them.
|
||||
# # max_turns: 120
|
||||
#
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user