diff --git a/backend/packages/harness/deerflow/agents/lead_agent/agent.py b/backend/packages/harness/deerflow/agents/lead_agent/agent.py index 41c8708c5..b2620bd5b 100644 --- a/backend/packages/harness/deerflow/agents/lead_agent/agent.py +++ b/backend/packages/harness/deerflow/agents/lead_agent/agent.py @@ -716,15 +716,15 @@ def build_middlewares( if configured_middlewares: middlewares.extend(configured_middlewares) - # A provider may return an empty AIMessage after tool execution. Retry the - # final response once, then persist a visible error fallback rather than - # allowing LangChain's no-tool-call router to end a silent successful run. + # LLMErrorHandlingMiddleware gives a run one model-boundary retry for a true + # empty stop. Keep a terminal fallback for post-tool responses that still have + # no user-visible text, without adding a graph-level recovery turn. middlewares.append(TerminalResponseMiddleware()) # A provider may also cap the final assistant response at the model output - # limit. Preserve the assistant content unchanged, but stamp a run-level - # stop_reason so Gateway consumers can tell a length-capped completion from - # a clean one. + # limit. Detector-matched caps stamp stop_reason=model_length_capped, + # suppress that response's tool calls, and append a length notice when no + # visible text was produced. middlewares.append(ModelLengthFinishReasonMiddleware()) # SafetyFinishReasonMiddleware — suppress tool execution when the provider diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md index a2c85e905..c19e4e1e4 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md @@ -70,7 +70,7 @@ strict providers reject. their narrower discovery allowlists never rebuild the shared thread view or force eager sandbox acquisition. 7. **DanglingToolCallMiddleware** - Injects placeholder ToolMessages for AIMessage tool_calls that lack responses (e.g., user interruption), preserving raw provider tool-call payloads in `additional_kwargs["tool_calls"]`; malformed tool-call names and arguments are sanitized in the model-bound request so strict OpenAI-compatible providers do not reject the next request -8. **LLMErrorHandlingMiddleware** - Converts provider/model failures to recoverable assistant errors. Async cancellation at admission, provider execution, retry events, or backoff releases only the call's own half-open probe (ownership assigned under the circuit lock), then propagates unchanged, without retry or failure accounting. +8. **LLMErrorHandlingMiddleware** - Converts provider/model failures to recoverable assistant errors. Normal completions without visible text or tool-call intent (including whitespace/reasoning-only responses) get at most one retry per run, then a marked visible fallback; empties never count toward the circuit breaker. Cancellation during admission, execution, retry events or backoff releases only this call's half-open probe (assigned under the circuit lock), then propagates unchanged without retry or failure accounting. 9. **Authorization / GuardrailMiddleware** - Up to two independent pre-tool-call gates run here. When `authorization.enabled`, the `AuthorizationProvider` instance already used for Layer 1 capability filtering is wrapped by `GuardrailAuthorizationAdapter` and reused for Layer 2 execution checks. A generated `tool_search` bypasses the adapter's second provider call only when the current build has a concrete deferred setup; its catalog was already filtered by Layer 1, and an ordinary same-named tool without that deferred setup receives no exemption. When `guardrails.enabled`, the explicitly configured `GuardrailProvider` is appended after authorization and still evaluates every call, including `tool_search`. Authorization therefore runs outermost and can deny before an external guardrail call; both use the existing middleware's fail-closed, audit, sync/async, and error-`ToolMessage` behavior. See the authorization RFC and [docs/GUARDRAILS.md](../../../../../docs/GUARDRAILS.md). Every guardrail decision path publishes a neutral @@ -120,7 +120,7 @@ Before changing a later authorization phase, read the [authorization RFC](../../ 30. **TokenBudgetMiddleware** - `token_budget.enabled`: shares run-ID budgets across continuations; missing/invalid IDs clear invocation state. 31. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before config-declared extensions and the terminal-response/safety/clarification tail 32. **Configured extension middlewares** - `extensions.middlewares` in `config.yaml` or `extensions_config.json` optionally accepts `module.path:ClassName` strings or `{class, kwargs}` objects. `deerflow.reflection.resolve_class` loads `AgentMiddleware` classes; import, class, and constructor errors fail agent creation. `kwargs` must be JSON-compatible; YAML dates/timestamps become ISO strings. Order: built-ins/custom and loop/token guards → extensions → terminal-response/safety/clarification tail. Subagents share the list before their safety tail; separate lead/subagent lists are unsupported. Trusted operator config only: paths instantiate arbitrary code. Gateway skill/MCP toggles preserve it in raw JSON; adding an API write path requires explicit trust-boundary review. -33. **TerminalResponseMiddleware** - When a provider returns an empty terminal `AIMessage` after tool execution, injects a hidden recovery prompt and retries the model once; a second empty response is replaced in checkpoint state by a visible error fallback marked for the run worker, so the run finishes as an error instead of a silent success -34. **ModelLengthFinishReasonMiddleware** - Records `stop_reason=model_length_capped` when provider-specific length detectors match a terminal `AIMessage` without tool-call intent (`finish_reason=length` / `MAX_TOKENS`, or `stop_reason=max_tokens`), preserving the original assistant content and never reparsing textual tool-call-like envelopes +33. **TerminalResponseMiddleware** - After tools following the latest real user message, an assistant response without visible text or tool intent gets a marked visible fallback in the same step. Preserves content blocks; no graph retry or separate length-reason vocabulary. +34. **ModelLengthFinishReasonMiddleware** - A length-detector match stamps `stop_reason=model_length_capped` and ends the tool loop. Suppresses all structured/raw calls and native `tool_use` blocks, even fully parsed calls. Keeps text/thinking blocks; appends a length notice if no visible text exists. Audit metadata keeps detector, reason, call count/names, never suppressed arguments. 35. **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 terminal-response/custom/configured middlewares so LangChain's reverse-order `after_model` dispatch runs it first 36. **ClarificationMiddleware** - Intercepts `ask_clarification`, writes a readable `ToolMessage.content` fallback plus a structured `ToolMessage.artifact.human_input` payload, and interrupts via `Command(goto=END)` (must be last). `after_model` drops same-turn sibling tool calls so they cannot run before the user answers; a malformed `ask_clarification` parked on `invalid_tool_calls` is the same stop signal. `disable_clarification` runs keep the siblings. Payloads are versioned — legacy `free_text`/`choice_with_other` stay `version: 1`; the v2 `form` mode (from `fields`) is `version: 2` so older frontends reject it and fall back to plain text. Field normalization is deterministic and lives in the middleware (it short-circuits before tool execution, so tool-arg typing gives no runtime validation), and it is atomic: any structurally broken entry — non-dict, bad/duplicate name, a name colliding with a JS `Object.prototype` member (`__proto__`/`constructor`), or exceeding the caps (16 fields / 24 options per field / 200 chars per text / `MAX_FORM_SERIALIZED_BYTES` = 16KB UTF-8, the per-item caps alone admitting forms whose IM text fallback overruns channel limits) — degrades the whole form to the legacy option/free-text modes, so a card never renders "complete" while missing a field. Benign issues degrade locally (unknown types — incl. unhashable JSON like `type: []`, which must not raise from the membership probe — and option-less selects become `text`); options are trimmed/deduped with blanks dropped (form- and top-level) since the frontend rejects blank labels. XML-to-dict option payloads are recursively flattened from dict/list containers in source order, scalar leaves kept, residual XML tags stripped before that trimming. Checkboxes are booleans defaulting to "no"; `required` on one means consent semantics. The response protocol is unchanged (v1 `text`/`option`): form cards submit a text summary as `response_kind: "text"`, so journal persistence needs no new allowlist entries. Because this middleware can short-circuit before `on_tool_end`, `RunJournal` does a root-run reconciliation for `ToolMessage`s whose `tool_call_id` came from the current run, so cards survive checkpoint compaction. That reconciliation is **not** `ask_clarification`-only — any middleware that answers a tool call has the same gap, and a result the user saw must not vanish on reload (#4666 — `ReadBeforeWriteMiddleware` blocked-write errors reached the UI but not the event store). It is bounded by three conditions, not a name allowlist: the message is user-visible, the call belongs to this run's **lead agent** (`_remember_current_run_tool_calls` records lead-agent calls only; subagent results stay in `subagent.step`), and it is not already persisted. Human Input Card replies are `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden sources (currently `ask_clarification`) as `llm.human.input`. diff --git a/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py index 7df38847b..653a37409 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py @@ -22,12 +22,58 @@ from langchain.agents.middleware.types import ( from langchain_core.messages import AIMessage from langgraph.errors import GraphBubbleUp +from deerflow.agents.middlewares.model_response import append_visible_text, finish_reason, has_tool_call_intent, has_visible_content, last_ai_message from deerflow.config.app_config import AppConfig from deerflow.models.request_admission import AdmissionError from deerflow.utils.custom_events import aemit_custom_event, emit_custom_event logger = logging.getLogger(__name__) +_EMPTY_RESPONSE_RETRY_CONTEXT_KEY = "__empty_response_retry_consumed" +_EMPTY_RESPONSE_RETRY_CONSUMED = object() +_NON_CIRCUIT_FAILURE_REASONS = {"burst_rate", "empty_response"} + + +class EmptyModelResponseError(RuntimeError): + """The model completed normally without producing persistent content.""" + + code = "EMPTY_RESPONSE" + + def __init__( + self, + message: str = "Model returned a completed response with no content", + *, + response_message: AIMessage | None = None, + ) -> None: + super().__init__(message) + self.response_message = response_message + + +def _raise_for_empty_response(response: ModelCallResult) -> None: + """在响应写入图状态前把零内容 stop 转换为可重试错误。""" + message = last_ai_message(response) + if message is None: + raise EmptyModelResponseError() + if has_visible_content(message) or has_tool_call_intent(message): + return + reason = finish_reason(message) + if reason in (None, "", "stop", "end_turn"): + raise EmptyModelResponseError(response_message=message) + + +def _consume_empty_response_retry(request: ModelRequest) -> bool: + """Consume the one empty-response retry budget stored in run context.""" + runtime = getattr(request, "runtime", None) + context = getattr(runtime, "context", None) + if not isinstance(context, dict): + # Direct middleware calls without runtime context retain one retry per call. + return True + if context.get(_EMPTY_RESPONSE_RETRY_CONTEXT_KEY) is _EMPTY_RESPONSE_RETRY_CONSUMED: + return False + context[_EMPTY_RESPONSE_RETRY_CONTEXT_KEY] = _EMPTY_RESPONSE_RETRY_CONSUMED + return True + + _RETRIABLE_STATUS_CODES = {408, 409, 425, 429, 500, 502, 503, 504} _BUSY_PATTERNS = ( "server busy", @@ -95,7 +141,9 @@ _BURST_PATTERNS = ( # value of 2 means "1 first attempt + 1 retry" (the CR-requested # "keep one retry" behavior). _RETRY_BUDGET_OVERRIDES: dict[str, int] = { + "EmptyModelResponseError": 2, "StreamChunkTimeoutError": 2, + "ReadTimeout": 2, } # Per-reason retry budget overrides, applied in addition to the per-exception @@ -413,6 +461,12 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): self._circuit_probe_in_flight = False self._circuit_probe_token: object | None = None + def release_policy_parameters(self) -> dict[str, object]: + return { + "empty_response_retry_limit": 1, + "empty_response_retry_scope": "run", + } + def _max_attempts_for(self, exc: BaseException, reason: str = "transient") -> int: """Return the effective max attempt count for this exception. @@ -512,6 +566,8 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): error_code = _extract_error_code(exc) status_code = _extract_status_code(exc) + if isinstance(exc, EmptyModelResponseError): + return True, "empty_response" if _matches_any(lowered, _QUOTA_PATTERNS) or _matches_any(str(error_code).lower(), _QUOTA_PATTERNS): return False, "quota" if _matches_any(lowered, _AUTH_PATTERNS): @@ -529,6 +585,11 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): "APITimeoutError", "APIConnectionError", "InternalServerError", + "ReadTimeout", + "ConnectTimeout", + "WriteTimeout", + "PoolTimeout", + "TimeoutException", "ReadError", # httpx.ReadError: connection dropped mid-stream "RemoteProtocolError", # httpx: server closed connection unexpectedly "StreamChunkTimeoutError", # langchain-openai: chunk gap exceeded stream_chunk_timeout @@ -660,6 +721,7 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): reason_text = { "busy": "provider is busy", "burst_rate": "provider is throttling request burst rate", + "empty_response": "provider returned an empty response", }.get(reason, "provider request failed temporarily") # ``max_attempts`` is the *effective* budget for this call (from # ``_max_attempts_for``), not the configured ceiling: a burst-rate call @@ -678,16 +740,25 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): error_type: str, reason: str, detail: str, + response_message: AIMessage | None = None, ) -> AIMessage: - return AIMessage( - content=content, - additional_kwargs={ + additional_kwargs = dict(response_message.additional_kwargs or {}) if response_message is not None else {} + additional_kwargs.update( + { "deerflow_error_fallback": True, "error_type": error_type, "error_reason": reason, "error_detail": detail, - }, + } ) + if response_message is not None: + return response_message.model_copy( + update={ + "content": append_visible_text(response_message, content), + "additional_kwargs": additional_kwargs, + } + ) + return AIMessage(content=content, additional_kwargs=additional_kwargs) def _build_user_message(self, exc: BaseException, reason: str) -> str: detail = _extract_error_detail(exc) @@ -697,6 +768,8 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): return "The configured LLM provider rejected the request because authentication or access is invalid. Please check the provider credentials and try again." if reason == "burst_rate": return "The configured LLM provider is temporarily throttling requests because the request rate increased too quickly (burst-rate limit). Please wait a moment and try again." + if reason == "empty_response": + return "The configured LLM provider returned an empty response after one automatic retry. Please continue the conversation or use a different model." if reason in {"busy", "transient"}: # Stream-drop failures (chunk-gap timeout, peer-closed connection, # raw read error) almost always point at a single oversized @@ -721,6 +794,7 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): error_type=type(exc).__name__, reason=reason, detail=_extract_error_detail(exc), + response_message=exc.response_message if isinstance(exc, EmptyModelResponseError) else None, ) def _build_retry_event( @@ -804,6 +878,7 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): while True: try: response = self._bounded_model_call_sync(request, handler) + _raise_for_empty_response(response) self._record_success() return response except GraphBubbleUp: @@ -813,7 +888,10 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): except Exception as exc: retriable, reason = self._classify_error(exc) max_attempts = self._max_attempts_for(exc, reason) - if retriable and attempt < max_attempts: + should_retry = retriable and attempt < max_attempts + if should_retry and reason == "empty_response": + should_retry = _consume_empty_response_retry(request) + if should_retry: wait_ms = self._build_retry_delay_ms(prev_delay_ms, exc, reason) prev_delay_ms = wait_ms logger.warning( @@ -833,14 +911,10 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): _extract_error_detail(exc), exc_info=exc, ) - if retriable and reason != "burst_rate": + if retriable and reason not in _NON_CIRCUIT_FAILURE_REASONS: self._record_failure() else: - # Non-retriable, OR burst_rate (a transient provider - # slope-throttle, not "provider down"): release the half-open - # probe without recording a failure so the circuit doesn't - # trip and fast-fail ALL calls for the recovery window - the - # exact self-inflicted outage #4290 is trying to prevent. + # These outcomes do not show that the provider is broadly unavailable. self._release_half_open_probe() return self._build_user_fallback_message(exc, reason) @@ -865,6 +939,7 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): while True: try: response = await self._bounded_model_call(request, handler) + _raise_for_empty_response(response) self._record_success() return response except GraphBubbleUp: @@ -874,7 +949,10 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): except Exception as exc: retriable, reason = self._classify_error(exc) max_attempts = self._max_attempts_for(exc, reason) - if retriable and attempt < max_attempts: + should_retry = retriable and attempt < max_attempts + if should_retry and reason == "empty_response": + should_retry = _consume_empty_response_retry(request) + if should_retry: wait_ms = self._build_retry_delay_ms(prev_delay_ms, exc, reason) prev_delay_ms = wait_ms logger.warning( @@ -894,14 +972,10 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]): _extract_error_detail(exc), exc_info=exc, ) - if retriable and reason != "burst_rate": + if retriable and reason not in _NON_CIRCUIT_FAILURE_REASONS: self._record_failure() else: - # Non-retriable, OR burst_rate (a transient provider - # slope-throttle, not "provider down"): release the half-open - # probe without recording a failure so the circuit doesn't - # trip and fast-fail ALL calls for the recovery window - the - # exact self-inflicted outage #4290 is trying to prevent. + # These outcomes do not show that the provider is broadly unavailable. self._release_half_open_probe() return self._build_user_fallback_message(exc, reason) except asyncio.CancelledError: diff --git a/backend/packages/harness/deerflow/agents/middlewares/model_length_finish_reason_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/model_length_finish_reason_middleware.py index 49650e469..b5cbcd617 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/model_length_finish_reason_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/model_length_finish_reason_middleware.py @@ -1,22 +1,12 @@ -"""Surface provider length-capped model responses as run stop reasons. +"""Surface provider length-capped responses and block truncated tool calls. Background — see issue bytedance/deer-flow#4271. Some providers stop generation because the output budget is exhausted and surface that through ``finish_reason='length'`` while still returning assistant -content. DeerFlow should preserve that content for audit, but it should not -silently treat the run as an uncapped clean completion when the provider has -explicitly signaled truncation. - -This middleware keeps that boundary narrow: -- it only marks a run-level stop reason when the final AIMessage is capped - by a provider length signal and still has visible content; -- it never rewrites the assistant content or reparses XML-like text into a - tool call; -- it ignores any response that still carries tool-call intent, malformed - tool-call metadata, or no visible content, so only terminal assistant - responses with visible content can be marked capped. - +content. DeerFlow preserves visible content, adds a deterministic notice when +no visible answer was produced, and drops tool calls that may have been +truncated at the output boundary before they can execute. """ from __future__ import annotations @@ -34,44 +24,70 @@ from deerflow.agents.middlewares.model_length_termination_detectors import ( ModelLengthTerminationDetector, default_detectors, ) +from deerflow.agents.middlewares.model_response import append_visible_text, has_tool_call_intent, has_visible_content MODEL_LENGTH_CAPPED_STOP_REASON = "model_length_capped" +_MODEL_LENGTH_CAPPED_CONTENT = "The model reached its output limit before producing a complete final response. Please continue the conversation to resume." logger = logging.getLogger(__name__) -def _has_tool_call_intent_or_error(message: AIMessage) -> bool: - if message.tool_calls or getattr(message, "invalid_tool_calls", None): - return True +def _tool_call_summary(message: AIMessage) -> tuple[int, list[str]]: + """Count suppressed tool calls and return their deduplicated names.""" + names: list[str] = [] + structured_calls: list[Any] = [*(message.tool_calls or []), *(getattr(message, "invalid_tool_calls", None) or [])] additional_kwargs = message.additional_kwargs or {} - return bool(additional_kwargs.get("tool_calls") or additional_kwargs.get("function_call")) + if structured_calls: + # LangChain commonly keeps both parsed calls and their raw provider copy. + calls = structured_calls + else: + calls = list(additional_kwargs.get("tool_calls") or []) + function_call = additional_kwargs.get("function_call") + if not calls and isinstance(function_call, dict): + calls.append(function_call) + if not calls: + calls = _anthropic_tool_use_blocks(message) + + for call in calls: + if not isinstance(call, dict): + continue + name = call.get("name") + function = call.get("function") + if not isinstance(name, str) and isinstance(function, dict): + name = function.get("name") + if isinstance(name, str) and name and name not in names: + names.append(name) + return len(calls), names -def _has_visible_content(message: AIMessage) -> bool: - content = message.content - if isinstance(content, str): - return bool(content.strip()) - if isinstance(content, list): - for block in content: - if isinstance(block, str) and block.strip(): - return True - if isinstance(block, dict) and block.get("type") in {"text", "output_text"}: - text = block.get("text") - if isinstance(text, str) and text.strip(): - return True - return False +def _anthropic_tool_use_blocks(message: AIMessage) -> list[dict[str, Any]]: + """Extract native Anthropic tool-use blocks retained in message content.""" + if not isinstance(message.content, list): + return [] + return [block for block in message.content if isinstance(block, dict) and block.get("type") == "tool_use"] + + +def _without_anthropic_tool_use_blocks(content: Any) -> Any: + """Remove native tool calls that cannot receive a matching tool result.""" + if not isinstance(content, list): + return content + return [block for block in content if not (isinstance(block, dict) and block.get("type") == "tool_use")] class ModelLengthFinishReasonMiddleware(AgentMiddleware[AgentState]): - """Record provider length caps for terminal assistant responses with content. - - If the last AIMessage still carries tool-call intent, this middleware - leaves it alone and lets the normal tool-handling path decide what to do. - """ + """Record provider length termination and block truncated tool calls.""" def __init__(self, detectors: list[ModelLengthTerminationDetector] | None = None) -> None: super().__init__() self._detectors: list[ModelLengthTerminationDetector] = list(detectors) if detectors else default_detectors() + def release_policy_parameters(self) -> dict[str, object]: + from deerflow_extension_api import canonical_hash + + return { + "suppress_truncated_tool_calls": True, + "empty_content_fallback_hash": canonical_hash(_MODEL_LENGTH_CAPPED_CONTENT), + } + def _detect(self, message: AIMessage) -> ModelLengthTermination | None: for detector in self._detectors: try: @@ -89,11 +105,6 @@ class ModelLengthFinishReasonMiddleware(AgentMiddleware[AgentState]): return None last = messages[-1] - if _has_tool_call_intent_or_error(last): - return None - if not _has_visible_content(last): - return None - termination = self._detect(last) if termination is None: return None @@ -119,7 +130,36 @@ class ModelLengthFinishReasonMiddleware(AgentMiddleware[AgentState]): "stamped_stop_reason": stamped_stop_reason, }, ) - return None + + contains_tool_call = has_tool_call_intent(last) or bool(_anthropic_tool_use_blocks(last)) + cleaned_content = _without_anthropic_tool_use_blocks(last.content) if contains_tool_call else last.content + content_source = last.model_copy(update={"content": cleaned_content}) + contains_visible_content = has_visible_content(content_source) + if not contains_tool_call and contains_visible_content: + return None + + additional_kwargs = dict(last.additional_kwargs or {}) + suppressed_count, suppressed_names = _tool_call_summary(last) if contains_tool_call else (0, []) + if contains_tool_call: + # Tool arguments may be incomplete at a max-token boundary. + additional_kwargs.pop("tool_calls", None) + additional_kwargs.pop("function_call", None) + additional_kwargs["model_length_termination"] = { + "detector": termination.detector, + "reason_field": termination.reason_field, + "reason_value": termination.reason_value, + "suppressed_tool_call_count": suppressed_count, + "suppressed_tool_call_names": suppressed_names, + } + replacement = last.model_copy( + update={ + "content": (cleaned_content if contains_visible_content else append_visible_text(content_source, _MODEL_LENGTH_CAPPED_CONTENT)), + "tool_calls": [], + "invalid_tool_calls": [], + "additional_kwargs": additional_kwargs, + } + ) + return {"messages": [replacement]} @override def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None: diff --git a/backend/packages/harness/deerflow/agents/middlewares/model_response.py b/backend/packages/harness/deerflow/agents/middlewares/model_response.py new file mode 100644 index 000000000..74a8fb637 --- /dev/null +++ b/backend/packages/harness/deerflow/agents/middlewares/model_response.py @@ -0,0 +1,61 @@ +"""Shared model response content and termination classification.""" + +from __future__ import annotations + +from typing import Any + +from langchain_core.messages import AIMessage + + +def last_ai_message(response: Any) -> AIMessage | None: + """Return the last assistant message from a middleware model result.""" + if isinstance(response, AIMessage): + return response + result = getattr(response, "result", None) + if isinstance(result, (list, tuple)): + return next((message for message in reversed(result) if isinstance(message, AIMessage)), None) + return None + + +def has_tool_call_intent(message: AIMessage) -> bool: + """Return whether parsed or provider-raw tool-call intent is present.""" + if message.tool_calls or getattr(message, "invalid_tool_calls", None): + return True + additional_kwargs = message.additional_kwargs or {} + return bool(additional_kwargs.get("tool_calls") or additional_kwargs.get("function_call")) + + +def has_visible_content(message: AIMessage) -> bool: + """Return whether a message contains non-whitespace user-visible text.""" + content = message.content + if isinstance(content, str): + return bool(content.strip()) + if not isinstance(content, list): + return False + + for block in content: + if isinstance(block, str) and block.strip(): + return True + if not isinstance(block, dict) or block.get("type") not in {"text", "output_text"}: + continue + text = block.get("text") + if isinstance(text, str) and text.strip(): + return True + return False + + +def append_visible_text(message: AIMessage, text: str) -> Any: + """Append a visible text block without dropping existing content blocks.""" + if isinstance(message.content, list): + return [*message.content, {"type": "text", "text": text}] + return text + + +def finish_reason(message: AIMessage) -> str | None: + """Read and normalize common provider termination-reason fields.""" + for metadata in (message.response_metadata or {}, message.additional_kwargs or {}): + for field in ("finish_reason", "stop_reason"): + value = metadata.get(field) + if isinstance(value, str): + return value.strip().lower() + return None diff --git a/backend/packages/harness/deerflow/agents/middlewares/terminal_response_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/terminal_response_middleware.py index e47b7956b..01da29936 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/terminal_response_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/terminal_response_middleware.py @@ -1,57 +1,17 @@ -"""Ensure tool-using lead-agent turns end with a visible assistant response.""" +"""Prevent an empty post-tool terminal response from becoming silent success.""" from __future__ import annotations -import threading -from collections.abc import Awaitable, Callable from typing import Any, override from langchain.agents import AgentState from langchain.agents.middleware import AgentMiddleware -from langchain.agents.middleware.types import ModelCallResult, ModelRequest, ModelResponse, hook_config -from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage, ToolMessage +from langchain_core.messages import AIMessage, HumanMessage, ToolMessage from langgraph.runtime import Runtime -from deerflow.agents.middlewares._bounded_dict import BoundedDict +from deerflow.agents.middlewares.model_response import append_visible_text, has_tool_call_intent, has_visible_content -_RECOVERY_PROMPT = ( - "\n" - "Your previous response after the tool execution was empty. Review the tool results " - "already present in the conversation and provide a concise, user-visible final response. " - "Do not call another tool unless it is strictly necessary.\n" - "" -) - -_FALLBACK_CONTENT = "The model completed the tool run but returned no final response, including after one automatic retry. Please try again or use a different model." - -_TOOL_CALL_FINISH_REASONS = {"tool_calls", "function_call"} - - -def _has_visible_content(message: AIMessage) -> bool: - """Return whether an AI message contains user-visible text.""" - content = message.content - if isinstance(content, str): - return bool(content.strip()) - if isinstance(content, list): - for block in content: - if isinstance(block, str) and block.strip(): - return True - if isinstance(block, dict) and block.get("type") in {"text", "output_text"}: - text = block.get("text") - if isinstance(text, str) and text.strip(): - return True - return False - - -def _has_tool_call_intent_or_error(message: AIMessage) -> bool: - """Keep tool routing and malformed tool-call handling out of this guard.""" - if message.tool_calls or getattr(message, "invalid_tool_calls", None): - return True - additional_kwargs = message.additional_kwargs or {} - if additional_kwargs.get("tool_calls") or additional_kwargs.get("function_call"): - return True - response_metadata = message.response_metadata or {} - return response_metadata.get("finish_reason") in _TOOL_CALL_FINISH_REASONS +_FALLBACK_CONTENT = "The model completed the tool run but returned no final response. Please try again or use a different model." def _tool_result_in_current_turn(messages: list[Any]) -> bool: @@ -63,161 +23,52 @@ def _tool_result_in_current_turn(messages: list[Any]) -> bool: if (message.additional_kwargs or {}).get("hide_from_ui"): continue latest_user_index = index - # Scope: #4027 covers interactive post-tool turns. Scheduled/internal - # invocations without a real HumanMessage need a separate terminal-success - # invariant rather than being inferred from arbitrary historical tools. if latest_user_index == -1: return False return any(isinstance(message, ToolMessage) for message in messages[latest_user_index + 1 :]) class TerminalResponseMiddleware(AgentMiddleware[AgentState]): - """Retry one empty post-tool response, then persist a visible error fallback.""" - - def __init__(self) -> None: - super().__init__() - self._lock = threading.Lock() - self._retry_counts: BoundedDict[tuple[str, str], int] = BoundedDict(1000) - self._pending_prompts: BoundedDict[tuple[str, str], bool] = BoundedDict(1000) + """Last-resort fallback after model-boundary empty-response recovery.""" def release_policy_parameters(self) -> dict[str, object]: from deerflow_extension_api import canonical_hash return { - "post_tool_empty_retry_limit": 1, - "recovery_prompt_hash": canonical_hash(_RECOVERY_PROMPT), + "post_tool_empty_retry_limit": 0, "fallback_content_hash": canonical_hash(_FALLBACK_CONTENT), } - @staticmethod - def _key(runtime: Runtime) -> tuple[str, str]: - context = getattr(runtime, "context", None) - if isinstance(context, dict): - thread_id = str(context.get("thread_id") or "unknown-thread") - run_id = str(context.get("run_id") or context.get("run_attempt_id") or id(runtime)) - return thread_id, run_id - # Defensive fallback for tests/custom embeddings. Production Gateway - # runs always provide thread_id and run_id in Runtime.context. - return "unknown-thread", str(id(runtime)) - - def _clear(self, runtime: Runtime) -> None: - key = self._key(runtime) - with self._lock: - self._retry_counts.pop(key, None) - self._pending_prompts.pop(key, None) - - def _clear_other_runs(self, runtime: Runtime) -> None: - thread_id, run_id = self._key(runtime) - with self._lock: - stale = [key for key in self._retry_counts if key[0] == thread_id and key[1] != run_id] - for key in stale: - self._retry_counts.pop(key, None) - self._pending_prompts.pop(key, None) - def _apply(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None: messages = list(state.get("messages") or []) if not messages or not isinstance(messages[-1], AIMessage): return None last = messages[-1] - if _has_visible_content(last) or _has_tool_call_intent_or_error(last): + if has_visible_content(last) or has_tool_call_intent(last): return None if not _tool_result_in_current_turn(messages): return None - key = self._key(runtime) - with self._lock: - # The recovery budget is once per run, not once per empty message. - # A retry that calls another tool must not refresh the budget and - # create an unbounded empty -> retry -> tool loop. - retry_count = self._retry_counts.get(key, 0) - if retry_count == 0: - self._retry_counts[key] = 1 - self._pending_prompts[key] = True - - if retry_count == 0: - # The next model call gets a new message id. Remove this empty - # terminal message now so a successful recovery does not leave it - # in checkpoint history or future model context. - message_updates = [RemoveMessage(id=last.id)] if last.id else [] - return {"messages": message_updates, "jump_to": "model"} - additional_kwargs = dict(last.additional_kwargs or {}) additional_kwargs.update( { "deerflow_error_fallback": True, - "error_reason": "Model returned an empty terminal response after one retry", + "error_reason": "Model returned an empty terminal response", } ) fallback = last.model_copy( update={ - "content": _FALLBACK_CONTENT, + "content": append_visible_text(last, _FALLBACK_CONTENT), "additional_kwargs": additional_kwargs, } ) return {"messages": [fallback]} - def _augment_request(self, request: ModelRequest) -> ModelRequest: - key = self._key(request.runtime) - with self._lock: - pending = key in self._pending_prompts - self._pending_prompts.pop(key, None) - if not pending: - return request - reminder = HumanMessage( - content=_RECOVERY_PROMPT, - name="terminal_response_recovery", - additional_kwargs={"hide_from_ui": True}, - ) - return request.override(messages=[*request.messages, reminder]) - - @override - def before_agent(self, state: AgentState, runtime: Runtime) -> dict | None: - self._clear_other_runs(runtime) - # A prior invocation can bypass after_agent via Command(goto=END). - # Reset the same run id here so resume starts with a fresh one-retry - # budget; internal jump_to=model loops do not re-run before_agent. - self._clear(runtime) - return None - - @override - async def abefore_agent(self, state: AgentState, runtime: Runtime) -> dict | None: - self._clear_other_runs(runtime) - self._clear(runtime) - return None - - @hook_config(can_jump_to=["model"]) @override def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None: return self._apply(state, runtime) - @hook_config(can_jump_to=["model"]) @override async def aafter_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None: return self._apply(state, runtime) - - @override - def wrap_model_call( - self, - request: ModelRequest, - handler: Callable[[ModelRequest], ModelResponse], - ) -> ModelCallResult: - return handler(self._augment_request(request)) - - @override - async def awrap_model_call( - self, - request: ModelRequest, - handler: Callable[[ModelRequest], Awaitable[ModelResponse]], - ) -> ModelCallResult: - return await handler(self._augment_request(request)) - - @override - def after_agent(self, state: AgentState, runtime: Runtime) -> dict | None: - self._clear(runtime) - return None - - @override - async def aafter_agent(self, state: AgentState, runtime: Runtime) -> dict | None: - self._clear(runtime) - return None diff --git a/backend/packages/harness/deerflow/agents/middlewares/todo_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/todo_middleware.py index 3961d7213..75d230191 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/todo_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/todo_middleware.py @@ -297,6 +297,9 @@ class TodoMiddleware(TodoListMiddleware): if not last_ai or _has_tool_call_intent_or_error(last_ai): return None + if (last_ai.additional_kwargs or {}).get("deerflow_error_fallback"): + return None + # 3. Allow exit when all todos are completed or there are no todos. todos: list[Todo] = state.get("todos") or [] # type: ignore[assignment] if not todos or all(t.get("status") == "completed" for t in todos): diff --git a/backend/packages/harness/deerflow/models/assistant_payload_replay.py b/backend/packages/harness/deerflow/models/assistant_payload_replay.py index cffba5e99..c5c08d3cb 100644 --- a/backend/packages/harness/deerflow/models/assistant_payload_replay.py +++ b/backend/packages/harness/deerflow/models/assistant_payload_replay.py @@ -23,12 +23,6 @@ def restore_assistant_payloads( restore: AssistantPayloadRestorer, ) -> None: """Restore provider-specific fields onto serialized assistant payloads.""" - if len(payload_messages) == len(original_messages): - for payload_msg, orig_msg in zip(payload_messages, original_messages): - if payload_msg.get("role") == "assistant" and isinstance(orig_msg, AIMessage): - restore(payload_msg, orig_msg) - return - ai_messages = [m for m in original_messages if isinstance(m, AIMessage)] assistant_payloads = [m for m in payload_messages if m.get("role") == "assistant"] used_ai_indexes: set[int] = set() diff --git a/backend/packages/harness/deerflow/models/factory.py b/backend/packages/harness/deerflow/models/factory.py index 1bc837ab7..6cfc26369 100644 --- a/backend/packages/harness/deerflow/models/factory.py +++ b/backend/packages/harness/deerflow/models/factory.py @@ -340,7 +340,11 @@ def create_chat_model(name: str | None = None, thinking_enabled: bool = False, * _warn_unknown_model_settings(model_class, name, model_settings_from_config) - model_instance = model_class(**kwargs, **model_settings_from_config) + # 配置提供默认值，调用方显式传入的非空参数统一覆盖配置。 + # 先合并再展开，避免同名字段通过两个 **dict 传入时触发 TypeError。 + effective_model_settings = dict(model_settings_from_config) + effective_model_settings.update({key: value for key, value in kwargs.items() if value is not None}) + model_instance = model_class(**effective_model_settings) if translate_context_window: # Applied *after* construction and merged into the provider's inferred diff --git a/backend/packages/harness/deerflow/models/patched_deepseek.py b/backend/packages/harness/deerflow/models/patched_deepseek.py index 341ff9b6b..177d0ab98 100644 --- a/backend/packages/harness/deerflow/models/patched_deepseek.py +++ b/backend/packages/harness/deerflow/models/patched_deepseek.py @@ -10,11 +10,45 @@ on all assistant messages when thinking mode is enabled. from typing import Any from langchain_core.language_models import LanguageModelInput +from langchain_core.messages import AIMessage from langchain_deepseek import ChatDeepSeek from deerflow.models.assistant_payload_replay import restore_assistant_payloads, restore_reasoning_content +def _thinking_enabled(*sources: Any) -> bool: + """Return whether the request explicitly enables DeepSeek thinking mode.""" + for source in sources: + if not isinstance(source, dict): + continue + thinking = source.get("thinking") + if isinstance(thinking, dict) and thinking.get("type") == "enabled": + return True + extra_body = source.get("extra_body") + if isinstance(extra_body, dict): + nested = extra_body.get("thinking") + if isinstance(nested, dict) and nested.get("type") == "enabled": + return True + return False + + +def _restore_deepseek_assistant_payload( + payload_msg: dict[str, Any], + orig_msg: AIMessage, + *, + thinking_enabled: bool, +) -> None: + """Restore assistant history and required thinking-mode placeholders.""" + restore_reasoning_content(payload_msg, orig_msg) + has_tool_calls = bool(payload_msg.get("tool_calls")) + if has_tool_calls and payload_msg.get("content") is None: + # DeepSeek requires an empty string, rather than null, for tool-call history. + payload_msg["content"] = "" + if thinking_enabled and has_tool_calls and "reasoning_content" not in payload_msg: + # Thinking-mode tool turns require this field even when no reasoning was emitted. + payload_msg["reasoning_content"] = "" + + class PatchedChatDeepSeek(ChatDeepSeek): """ChatDeepSeek with proper reasoning_content preservation. @@ -44,16 +78,25 @@ class PatchedChatDeepSeek(ChatDeepSeek): Overrides the parent method to inject reasoning_content from additional_kwargs into assistant messages in the payload. """ - # Get the original messages before conversion original_messages = self._convert_input(input_).to_messages() + request_messages = [message for message in original_messages if not (isinstance(message, AIMessage) and (message.additional_kwargs or {}).get("deerflow_error_fallback"))] # Call parent to get the base payload - payload = super()._get_request_payload(input_, stop=stop, **kwargs) + payload = super()._get_request_payload(request_messages, stop=stop, **kwargs) + request_thinking_enabled = _thinking_enabled( + payload, + kwargs, + {"extra_body": getattr(self, "extra_body", None)}, + ) restore_assistant_payloads( payload.get("messages", []), - original_messages, - restore_reasoning_content, + request_messages, + lambda payload_msg, orig_msg: _restore_deepseek_assistant_payload( + payload_msg, + orig_msg, + thinking_enabled=request_thinking_enabled, + ), ) return payload diff --git a/backend/tests/test_llm_error_handling_middleware.py b/backend/tests/test_llm_error_handling_middleware.py index 2da3d752b..42f2587d4 100644 --- a/backend/tests/test_llm_error_handling_middleware.py +++ b/backend/tests/test_llm_error_handling_middleware.py @@ -9,10 +9,16 @@ from types import SimpleNamespace from typing import Any import pytest -from langchain_core.messages import AIMessage +from langchain.agents import create_agent +from langchain.agents.middleware.types import ModelResponse +from langchain_core.language_models import BaseChatModel +from langchain_core.messages import AIMessage, HumanMessage +from langchain_core.outputs import ChatGeneration, ChatResult +from langchain_core.tools import StructuredTool from langgraph.errors import GraphBubbleUp from deerflow.agents.middlewares.llm_error_handling_middleware import ( + EmptyModelResponseError, LLMErrorHandlingMiddleware, ) from deerflow.config.app_config import AppConfig, LlmCallConfig @@ -235,6 +241,405 @@ def test_sync_model_call_uses_retry_after_header(monkeypatch: pytest.MonkeyPatch assert [event["type"] for event in events] == ["llm_retry"] +def test_sync_empty_stop_retries_before_response_is_returned(monkeypatch: pytest.MonkeyPatch) -> None: + """An empty stop is retried before the failed attempt reaches graph state.""" + middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1) + attempts = 0 + monkeypatch.setattr("time.sleep", lambda _delay: None) + + def handler(_request) -> AIMessage: + nonlocal attempts + attempts += 1 + if attempts == 1: + return AIMessage(content="", response_metadata={"finish_reason": "stop"}) + return AIMessage(content="recovered", response_metadata={"finish_reason": "stop"}) + + result = middleware.wrap_model_call(SimpleNamespace(), handler) + + assert isinstance(result, AIMessage) + assert result.content == "recovered" + assert attempts == 2 + + +def test_persistent_empty_stop_returns_explicit_error_fallback(monkeypatch: pytest.MonkeyPatch) -> None: + middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1) + attempts = 0 + monkeypatch.setattr("time.sleep", lambda _delay: None) + + def handler(_request) -> AIMessage: + nonlocal attempts + attempts += 1 + return AIMessage(content="", response_metadata={"finish_reason": "stop"}) + + result = middleware.wrap_model_call(SimpleNamespace(), handler) + + assert isinstance(result, AIMessage) + assert attempts == 2 + assert result.additional_kwargs["deerflow_error_fallback"] is True + assert result.additional_kwargs["error_reason"] == "empty_response" + assert result.additional_kwargs["error_type"] == "EmptyModelResponseError" + assert "empty response" in str(result.content).lower() + + +@pytest.mark.anyio +async def test_async_empty_stop_retries_once(monkeypatch: pytest.MonkeyPatch) -> None: + middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1) + attempts = 0 + + async def fake_sleep(_delay: float) -> None: + return None + + async def handler(_request) -> AIMessage: + nonlocal attempts + attempts += 1 + if attempts == 1: + return AIMessage(content="", response_metadata={"finish_reason": "stop"}) + return AIMessage(content="recovered asynchronously", response_metadata={"finish_reason": "stop"}) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + result = await middleware.awrap_model_call(SimpleNamespace(), handler) + + assert result.content == "recovered asynchronously" + assert attempts == 2 + + +@pytest.mark.parametrize( + "message", + [ + AIMessage( + content="", + tool_calls=[{"id": "call-1", "name": "bash", "args": {}}], + response_metadata={"finish_reason": "tool_calls"}, + ), + AIMessage(content="", response_metadata={"finish_reason": "length"}), + ], +) +def test_nonempty_or_non_stop_response_is_not_classified_as_empty(message: AIMessage) -> None: + middleware = _build_middleware(retry_max_attempts=3) + attempts = 0 + + def handler(_request) -> AIMessage: + nonlocal attempts + attempts += 1 + return message + + result = middleware.wrap_model_call(SimpleNamespace(), handler) + + assert result is message + assert attempts == 1 + + +@pytest.mark.parametrize( + "message", + [ + AIMessage(content=" ", response_metadata={"finish_reason": "stop"}), + AIMessage(content="", additional_kwargs={"reasoning_content": "thinking"}, response_metadata={"finish_reason": "stop"}), + AIMessage(content=[{"type": "thinking", "thinking": "thinking"}], response_metadata={"finish_reason": "stop"}), + AIMessage(content=[{"type": "reasoning", "reasoning": "thinking"}], response_metadata={"finish_reason": "stop"}), + ], +) +def test_nonvisible_stop_response_retries_then_returns_marked_fallback( + message: AIMessage, + monkeypatch: pytest.MonkeyPatch, +) -> None: + middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1) + attempts = 0 + monkeypatch.setattr("time.sleep", lambda _delay: None) + + def handler(_request) -> AIMessage: + nonlocal attempts + attempts += 1 + return message + + result = middleware.wrap_model_call(SimpleNamespace(), handler) + + assert attempts == 2 + assert result.additional_kwargs["deerflow_error_fallback"] is True + assert result.additional_kwargs["error_reason"] == "empty_response" + + +def test_empty_response_fallback_preserves_reasoning_payload(monkeypatch: pytest.MonkeyPatch) -> None: + middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1) + thinking_block = {"type": "thinking", "thinking": "internal reasoning"} + message = AIMessage( + content=[thinking_block], + additional_kwargs={"reasoning_content": "provider reasoning"}, + response_metadata={"finish_reason": "stop"}, + ) + monkeypatch.setattr("time.sleep", lambda _delay: None) + + result = middleware.wrap_model_call(SimpleNamespace(), lambda _request: message) + + assert result.content[0] == thinking_block + assert result.content[-1]["type"] == "text" + assert "returned an empty response" in result.content[-1]["text"] + assert result.additional_kwargs["reasoning_content"] == "provider reasoning" + assert result.response_metadata == message.response_metadata + + +def test_empty_model_response_container_retries_before_graph_state(monkeypatch: pytest.MonkeyPatch) -> None: + """生产环境的 ModelResponse.result 结构也必须在模型边界完成判空。""" + middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1) + attempts = 0 + monkeypatch.setattr("time.sleep", lambda _delay: None) + + def handler(_request) -> ModelResponse: + nonlocal attempts + attempts += 1 + if attempts == 1: + return ModelResponse(result=[AIMessage(content="", response_metadata={"finish_reason": "stop"})]) + return ModelResponse(result=[AIMessage(content="recovered", response_metadata={"finish_reason": "stop"})]) + + result = middleware.wrap_model_call(SimpleNamespace(), handler) + + assert isinstance(result, ModelResponse) + assert result.result[-1].content == "recovered" + assert attempts == 2 + + +def test_empty_response_error_uses_one_retry_budget() -> None: + middleware = _build_middleware(retry_max_attempts=3) + + assert middleware._max_attempts_for(EmptyModelResponseError()) == 2 + assert middleware.release_policy_parameters() == { + "empty_response_retry_limit": 1, + "empty_response_retry_scope": "run", + } + + +def test_empty_response_retry_budget_is_shared_across_model_calls_in_one_run(monkeypatch: pytest.MonkeyPatch) -> None: + middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1) + request = SimpleNamespace(runtime=SimpleNamespace(context={"thread_id": "thread-1", "run_id": "run-1"})) + first_attempts = 0 + second_attempts = 0 + monkeypatch.setattr("time.sleep", lambda _delay: None) + + def first_handler(_request) -> AIMessage: + nonlocal first_attempts + first_attempts += 1 + if first_attempts == 1: + return AIMessage(content="", response_metadata={"finish_reason": "stop"}) + return AIMessage( + content="", + tool_calls=[{"id": "call-1", "name": "bash", "args": {}}], + response_metadata={"finish_reason": "tool_calls"}, + ) + + def second_handler(_request) -> AIMessage: + nonlocal second_attempts + second_attempts += 1 + return AIMessage(content="", response_metadata={"finish_reason": "stop"}) + + first_result = middleware.wrap_model_call(request, first_handler) + second_result = middleware.wrap_model_call(request, second_handler) + + assert first_result.tool_calls + assert first_attempts == 2 + assert second_attempts == 1 + assert second_result.additional_kwargs["error_reason"] == "empty_response" + + +def test_caller_cannot_preconsume_empty_response_retry_budget(monkeypatch: pytest.MonkeyPatch) -> None: + middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1) + request = SimpleNamespace(runtime=SimpleNamespace(context={"__empty_response_retry_consumed": True})) + attempts = 0 + monkeypatch.setattr("time.sleep", lambda _delay: None) + + def empty_handler(_request) -> AIMessage: + nonlocal attempts + attempts += 1 + return AIMessage(content="", response_metadata={"finish_reason": "stop"}) + + result = middleware.wrap_model_call(request, empty_handler) + + assert attempts == 2 + assert result.additional_kwargs["error_reason"] == "empty_response" + + +@pytest.mark.anyio +async def test_async_empty_response_retry_budget_is_shared_across_model_calls_in_one_run( + monkeypatch: pytest.MonkeyPatch, +) -> None: + middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1) + request = SimpleNamespace(runtime=SimpleNamespace(context={"thread_id": "thread-async", "run_id": "run-async"})) + attempts = 0 + + async def fake_sleep(_delay: float) -> None: + return None + + async def handler(_request) -> AIMessage: + nonlocal attempts + attempts += 1 + return AIMessage(content="", response_metadata={"finish_reason": "stop"}) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + first_result = await middleware.awrap_model_call(request, handler) + second_result = await middleware.awrap_model_call(request, handler) + + assert attempts == 3 + assert first_result.additional_kwargs["error_reason"] == "empty_response" + assert second_result.additional_kwargs["error_reason"] == "empty_response" + + +def test_empty_response_exhaustion_does_not_trip_circuit_breaker(monkeypatch: pytest.MonkeyPatch) -> None: + middleware = _build_middleware( + circuit_failure_threshold=2, + retry_max_attempts=2, + retry_base_delay_ms=1, + retry_cap_delay_ms=1, + ) + monkeypatch.setattr("time.sleep", lambda _delay: None) + + def empty_handler(_request) -> AIMessage: + return AIMessage(content="", response_metadata={"finish_reason": "stop"}) + + for index in range(3): + request = SimpleNamespace(runtime=SimpleNamespace(context={"run_id": f"run-{index}"})) + result = middleware.wrap_model_call(request, empty_handler) + assert result.additional_kwargs["error_reason"] == "empty_response" + + healthy_calls = 0 + + def healthy_handler(_request) -> AIMessage: + nonlocal healthy_calls + healthy_calls += 1 + return AIMessage(content="healthy") + + result = middleware.wrap_model_call( + SimpleNamespace(runtime=SimpleNamespace(context={"run_id": "run-healthy"})), + healthy_handler, + ) + + assert middleware._circuit_failure_count == 0 + assert middleware._circuit_state == "closed" + assert healthy_calls == 1 + assert result.content == "healthy" + + +def test_empty_response_exhaustion_releases_half_open_probe(monkeypatch: pytest.MonkeyPatch) -> None: + middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=1, retry_cap_delay_ms=1) + middleware._circuit_state = "half_open" + assert middleware._check_circuit() is False + assert middleware._circuit_probe_in_flight is True + monkeypatch.setattr("time.sleep", lambda _delay: None) + monkeypatch.setattr(middleware, "_check_circuit", lambda: False) + + def empty_handler(_request) -> AIMessage: + return AIMessage(content="", response_metadata={"finish_reason": "stop"}) + + result = middleware.wrap_model_call( + SimpleNamespace(runtime=SimpleNamespace(context={"run_id": "half-open-run"})), + empty_handler, + ) + + assert result.additional_kwargs["error_reason"] == "empty_response" + assert middleware._circuit_state == "half_open" + assert middleware._circuit_probe_in_flight is False + + +class _EmptyThenRecoveredGraphModel(BaseChatModel): + call_count: int = 0 + + @property + def _llm_type(self) -> str: + return "empty-then-recovered" + + def bind_tools(self, tools, **kwargs): + return self + + def _generate(self, messages, stop=None, run_manager=None, **kwargs): + self.call_count += 1 + content = "" if self.call_count == 1 else "recovered through graph stream" + message = AIMessage(content=content, response_metadata={"finish_reason": "stop"}) + return ChatResult(generations=[ChatGeneration(message=message)]) + + async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs): + return self._generate(messages, stop=stop, run_manager=run_manager, **kwargs) + + +@pytest.mark.anyio +async def test_empty_response_recovery_runs_through_create_agent_astream(monkeypatch: pytest.MonkeyPatch) -> None: + """Exercise the same create_agent().astream() path used by Gateway.""" + model = _EmptyThenRecoveredGraphModel() + middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=1, retry_cap_delay_ms=1) + + async def fake_sleep(_delay: float) -> None: + return None + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + agent = create_agent(model=model, tools=[], middleware=[middleware], context_schema=dict) + states = [ + state + async for state in agent.astream( + {"messages": [HumanMessage(content="hello")]}, + stream_mode="values", + context={"thread_id": "stream-thread", "run_id": "stream-run"}, + ) + ] + + assert model.call_count == 2 + assert states[-1]["messages"][-1].content == "recovered through graph stream" + + +class _EmptyRetryToolThenEmptyGraphModel(BaseChatModel): + call_count: int = 0 + + @property + def _llm_type(self) -> str: + return "empty-retry-tool-then-empty" + + def bind_tools(self, tools, **kwargs): + return self + + def _generate(self, messages, stop=None, run_manager=None, **kwargs): + self.call_count += 1 + if self.call_count == 2: + message = AIMessage( + content="", + tool_calls=[{"id": "call-probe", "name": "probe", "args": {}}], + response_metadata={"finish_reason": "tool_calls"}, + ) + else: + message = AIMessage(content="", response_metadata={"finish_reason": "stop"}) + return ChatResult(generations=[ChatGeneration(message=message)]) + + async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs): + return self._generate(messages, stop=stop, run_manager=run_manager, **kwargs) + + +@pytest.mark.anyio +async def test_empty_response_retry_budget_survives_real_agent_tool_loop(monkeypatch: pytest.MonkeyPatch) -> None: + """真实工具循环中，整个 run 只能消费一次空响应重试。""" + model = _EmptyRetryToolThenEmptyGraphModel() + middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=1, retry_cap_delay_ms=1) + tool_invocations: list[str] = [] + + def probe() -> str: + tool_invocations.append("probe") + return "probe-result" + + async def fake_sleep(_delay: float) -> None: + return None + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + tool = StructuredTool.from_function(probe, name="probe", description="Record one deterministic invocation") + agent = create_agent(model=model, tools=[tool], middleware=[middleware], context_schema=dict) + states = [ + state + async for state in agent.astream( + {"messages": [HumanMessage(content="use the probe")]}, + stream_mode="values", + context={"thread_id": "tool-loop-thread", "run_id": "tool-loop-run"}, + ) + ] + + final_message = states[-1]["messages"][-1] + assert model.call_count == 3 + assert tool_invocations == ["probe"] + assert final_message.additional_kwargs["deerflow_error_fallback"] is True + assert final_message.additional_kwargs["error_reason"] == "empty_response" + + def test_sync_retry_event_preserves_langgraph_control_flow(monkeypatch: pytest.MonkeyPatch) -> None: middleware = _build_middleware() @@ -782,6 +1187,48 @@ class _StreamChunkTimeoutError(Exception): _StreamChunkTimeoutError.__name__ = "StreamChunkTimeoutError" +class _ReadTimeoutError(Exception): + pass + + +_ReadTimeoutError.__name__ = "ReadTimeout" + + +def test_read_timeout_is_retried_and_exhaustion_returns_marked_fallback( + monkeypatch: pytest.MonkeyPatch, +) -> None: + middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1) + attempts = 0 + monkeypatch.setattr("time.sleep", lambda _delay: None) + + def handler(_request) -> AIMessage: + nonlocal attempts + attempts += 1 + raise _ReadTimeoutError("no bytes received before read deadline") + + result = middleware.wrap_model_call(SimpleNamespace(), handler) + + assert attempts == 2 + assert result.additional_kwargs["error_reason"] == "transient" + assert result.additional_kwargs["error_type"] == "ReadTimeout" + + +def test_empty_model_result_is_retried_before_fallback(monkeypatch: pytest.MonkeyPatch) -> None: + middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=1, retry_cap_delay_ms=1) + attempts = 0 + monkeypatch.setattr("time.sleep", lambda _delay: None) + + def handler(_request) -> ModelResponse: + nonlocal attempts + attempts += 1 + return ModelResponse(result=[]) + + result = middleware.wrap_model_call(SimpleNamespace(), handler) + + assert attempts == 2 + assert result.additional_kwargs["error_reason"] == "empty_response" + + def test_classify_error_stream_chunk_timeout_is_retriable() -> None: """StreamChunkTimeoutError must be classified as transient/retriable.""" middleware = _build_middleware() diff --git a/backend/tests/test_middleware_release_policy.py b/backend/tests/test_middleware_release_policy.py index 1618962f0..efc34ccf7 100644 --- a/backend/tests/test_middleware_release_policy.py +++ b/backend/tests/test_middleware_release_policy.py @@ -136,6 +136,20 @@ def _make_terminal_response_middleware(): return TerminalResponseMiddleware() +def _make_llm_error_handling_middleware(): + from deerflow.agents.middlewares.llm_error_handling_middleware import LLMErrorHandlingMiddleware + from deerflow.config.app_config import AppConfig + from deerflow.config.sandbox_config import SandboxConfig + + return LLMErrorHandlingMiddleware(app_config=AppConfig(sandbox=SandboxConfig(use="test"))) + + +def _make_model_length_finish_reason_middleware(): + from deerflow.agents.middlewares.model_length_finish_reason_middleware import ModelLengthFinishReasonMiddleware + + return ModelLengthFinishReasonMiddleware() + + def _make_todo_middleware(): from deerflow.agents.middlewares.todo_middleware import TodoMiddleware @@ -219,6 +233,12 @@ _MIDDLEWARE_DECLARATIONS = [ ("deerflow.agents.middlewares.loop_detection_middleware", "LoopDetectionMiddleware", _make_loop_detection_middleware), ("deerflow.agents.middlewares.subagent_limit_middleware", "SubagentLimitMiddleware", _make_subagent_limit_middleware), ("deerflow.agents.middlewares.terminal_response_middleware", "TerminalResponseMiddleware", _make_terminal_response_middleware), + ("deerflow.agents.middlewares.llm_error_handling_middleware", "LLMErrorHandlingMiddleware", _make_llm_error_handling_middleware), + ( + "deerflow.agents.middlewares.model_length_finish_reason_middleware", + "ModelLengthFinishReasonMiddleware", + _make_model_length_finish_reason_middleware, + ), # DeerFlow's own subclass, not the LangChain base class re-exported into # this module under the same import path (TodoListMiddleware). ("deerflow.agents.middlewares.todo_middleware", "TodoMiddleware", _make_todo_middleware), diff --git a/backend/tests/test_model_factory.py b/backend/tests/test_model_factory.py index dd816bbd0..1f05aeb04 100644 --- a/backend/tests/test_model_factory.py +++ b/backend/tests/test_model_factory.py @@ -1264,6 +1264,48 @@ def test_no_duplicate_kwarg_when_reasoning_effort_in_config_and_thinking_disable assert captured.get("reasoning_effort") == "minimal" +@pytest.mark.parametrize( + ("runtime_effort", "expected_effort"), + [ + (None, "high"), + ("low", "low"), + ], +) +def test_runtime_reasoning_effort_merges_with_profile_without_duplicate_kwarg( + monkeypatch, + runtime_effort, + expected_effort, +): + model = ModelConfig( + name="deepseek-reasoner", + display_name="DeepSeek Reasoner", + description=None, + use="deerflow.models.patched_deepseek:PatchedChatDeepSeek", + model="deepseek-reasoner", + reasoning_effort="high", + supports_thinking=True, + supports_reasoning_effort=True, + supports_vision=False, + ) + cfg = _make_app_config([model]) + captured: dict = {} + + class CapturingModel(FakeChatModel): + def __init__(self, **kwargs): + captured.update(kwargs) + BaseChatModel.__init__(self, **kwargs) + + _patch_factory(monkeypatch, cfg, model_class=CapturingModel) + + factory_module.create_chat_model( + name="deepseek-reasoner", + thinking_enabled=True, + reasoning_effort=runtime_effort, + ) + + assert captured["reasoning_effort"] == expected_effort + + @pytest.mark.parametrize( ("profile", "thinking_enabled", "requested_effort", "expected_effort"), [ diff --git a/backend/tests/test_model_length_finish_reason_middleware.py b/backend/tests/test_model_length_finish_reason_middleware.py index eedaedeb4..a62e2f2cc 100644 --- a/backend/tests/test_model_length_finish_reason_middleware.py +++ b/backend/tests/test_model_length_finish_reason_middleware.py @@ -3,8 +3,10 @@ import logging from unittest.mock import MagicMock +from langchain_anthropic import ChatAnthropic from langchain_core.messages import AIMessage, HumanMessage +from deerflow.agents.middlewares.dangling_tool_call_middleware import DanglingToolCallMiddleware from deerflow.agents.middlewares.model_length_finish_reason_middleware import ( MODEL_LENGTH_CAPPED_STOP_REASON, ModelLengthFinishReasonMiddleware, @@ -110,11 +112,28 @@ def test_length_cap_detection_logs_observability_fields(caplog): assert record.stamped_stop_reason is True -def test_finish_reason_length_with_tool_calls_passes_through(): +def test_finish_reason_length_drops_potentially_truncated_tool_calls(): mw = ModelLengthFinishReasonMiddleware() runtime = _runtime() msg = AIMessage( - content="", + content=[ + {"type": "text", "text": "partial answer"}, + { + "type": "tool_use", + "id": "call_write_1", + "name": "write_file", + "input": {"path": "/mnt/user-data/outputs/report.md"}, + }, + ], + additional_kwargs={ + "tool_calls": [ + { + "id": "call_write_1", + "type": "function", + "function": {"name": "write_file", "arguments": '{"path":"/tmp/report.md"'}, + } + ] + }, tool_calls=[ { "id": "call_write_1", @@ -125,17 +144,151 @@ def test_finish_reason_length_with_tool_calls_passes_through(): response_metadata={"finish_reason": "length"}, ) - assert mw._apply({"messages": [msg]}, runtime) is None - assert "stop_reason" not in runtime.context + result = mw._apply({"messages": [msg]}, runtime) + + assert result is not None + replacement = result["messages"][0] + assert replacement.tool_calls == [] + assert replacement.invalid_tool_calls == [] + assert replacement.content == [{"type": "text", "text": "partial answer"}] + assert "tool_calls" not in replacement.additional_kwargs + assert replacement.additional_kwargs["model_length_termination"]["suppressed_tool_call_count"] == 1 + assert replacement.additional_kwargs["model_length_termination"]["suppressed_tool_call_names"] == ["write_file"] + assert runtime.context["stop_reason"] == MODEL_LENGTH_CAPPED_STOP_REASON -def test_empty_finish_reason_length_passes_through_for_terminal_response_recovery(): +def test_anthropic_content_only_tool_use_is_removed_before_next_request(): + mw = ModelLengthFinishReasonMiddleware() + runtime = _runtime() + msg = AIMessage( + content=[ + { + "type": "tool_use", + "id": "call_write_1", + "name": "write_file", + "input": {"path": "/mnt/user-data/outputs/report.md"}, + } + ], + response_metadata={"stop_reason": "max_tokens"}, + ) + + result = mw._apply({"messages": [msg]}, runtime) + + assert result is not None + replacement = result["messages"][0] + assert all(block.get("type") != "tool_use" for block in replacement.content) + metadata = replacement.additional_kwargs["model_length_termination"] + assert metadata["suppressed_tool_call_count"] == 1 + assert metadata["suppressed_tool_call_names"] == ["write_file"] + assert msg.content[0]["type"] == "tool_use" + + messages = [HumanMessage("write a report"), replacement, HumanMessage("continue")] + repaired = DanglingToolCallMiddleware()._build_patched_messages(messages) or messages + payload = ChatAnthropic(model="claude-sonnet-4-5", api_key="test")._get_request_payload(repaired) + assistant_message = next(item for item in payload["messages"] if item["role"] == "assistant") + assert all(block.get("type") != "tool_use" for block in assistant_message["content"]) + + +def test_anthropic_thinking_is_preserved_when_native_tool_use_is_removed(): + mw = ModelLengthFinishReasonMiddleware() + runtime = _runtime() + thinking_block = { + "type": "thinking", + "thinking": "Need to write the file.", + "signature": "signed", + } + msg = AIMessage( + content=[ + thinking_block, + { + "type": "tool_use", + "id": "call_write_1", + "name": "write_file", + "input": {"path": "/mnt/user-data/outputs/report.md"}, + }, + ], + response_metadata={"stop_reason": "max_tokens"}, + ) + + result = mw._apply({"messages": [msg]}, runtime) + + assert result is not None + content = result["messages"][0].content + assert thinking_block in content + assert all(block.get("type") != "tool_use" for block in content) + assert content[-1]["type"] == "text" + assert "output limit" in content[-1]["text"] + + +def test_finish_reason_length_suppresses_complete_tool_call_as_safety_policy(): + """Even parsed arguments cannot be proven complete after a length cap.""" + mw = ModelLengthFinishReasonMiddleware() + runtime = _runtime() + msg = AIMessage( + content="", + tool_calls=[ + { + "id": "call_write_complete", + "name": "write_file", + "args": {"path": "/mnt/user-data/outputs/report.md", "content": "complete"}, + } + ], + response_metadata={"finish_reason": "length"}, + ) + + result = mw._apply({"messages": [msg]}, runtime) + + assert result is not None + replacement = result["messages"][0] + assert replacement.tool_calls == [] + assert replacement.additional_kwargs["model_length_termination"]["suppressed_tool_call_names"] == ["write_file"] + assert "output limit" in str(replacement.content) + + +def test_empty_finish_reason_length_gets_visible_capped_message(): mw = ModelLengthFinishReasonMiddleware() runtime = _runtime() msg = AIMessage(content="", response_metadata={"finish_reason": "length"}) - assert mw._apply({"messages": [msg]}, runtime) is None - assert "stop_reason" not in runtime.context + result = mw._apply({"messages": [msg]}, runtime) + + assert result is not None + replacement = result["messages"][0] + assert "output limit" in replacement.content + assert replacement.response_metadata["finish_reason"] == "length" + assert runtime.context["stop_reason"] == MODEL_LENGTH_CAPPED_STOP_REASON + + +def test_reasoning_only_length_preserves_reasoning_when_adding_visible_message(): + mw = ModelLengthFinishReasonMiddleware() + runtime = _runtime() + msg = AIMessage( + content="", + additional_kwargs={"reasoning_content": "internal reasoning"}, + response_metadata={"finish_reason": "length"}, + ) + + result = mw._apply({"messages": [msg]}, runtime) + + assert result is not None + replacement = result["messages"][0] + assert replacement.additional_kwargs["reasoning_content"] == "internal reasoning" + assert "output limit" in replacement.content + + +def test_thinking_blocks_are_preserved_when_length_notice_is_appended(): + mw = ModelLengthFinishReasonMiddleware() + runtime = _runtime() + thinking_block = {"type": "thinking", "thinking": "internal reasoning"} + msg = AIMessage(content=[thinking_block], response_metadata={"finish_reason": "length"}) + + result = mw._apply({"messages": [msg]}, runtime) + + assert result is not None + content = result["messages"][0].content + assert content[0] == thinking_block + assert content[-1]["type"] == "text" + assert "output limit" in content[-1]["text"] def test_existing_stop_reason_is_not_overwritten(caplog): diff --git a/backend/tests/test_patched_deepseek.py b/backend/tests/test_patched_deepseek.py index 695a74066..532e16c93 100644 --- a/backend/tests/test_patched_deepseek.py +++ b/backend/tests/test_patched_deepseek.py @@ -83,21 +83,36 @@ def _make_payload_message(role: str, content: str | None = None, tool_calls: lis return msg +_TOOL_SPEC = [{"type": "function", "function": {"name": "bash", "parameters": {}}}] + + def test_reasoning_content_injected_into_assistant_message(): """reasoning_content from additional_kwargs is restored in the payload.""" - model = _make_model() + model = _make_model(extra_body={"thinking": {"type": "enabled"}}) human = HumanMessage(content="What is 2+2?") ai = AIMessage( - content="4", + content="", additional_kwargs={"reasoning_content": "Let me think: 2+2=4"}, + tool_calls=[{"id": "call-1", "name": "calculator", "args": {"expression": "2+2"}}], ) base_payload = { + "tools": _TOOL_SPEC, "messages": [ _make_payload_message("user", "What is 2+2?"), - _make_payload_message("assistant", "4"), - ] + _make_payload_message( + "assistant", + None, + tool_calls=[ + { + "id": "call-1", + "type": "function", + "function": {"name": "calculator", "arguments": '{"expression":"2+2"}'}, + } + ], + ), + ], } with patch.object(type(model).__bases__[0], "_get_request_payload", return_value=base_payload): @@ -107,6 +122,7 @@ def test_reasoning_content_injected_into_assistant_message(): assistant_msg = next(m for m in payload["messages"] if m["role"] == "assistant") assert assistant_msg["reasoning_content"] == "Let me think: 2+2=4" + assert assistant_msg["content"] == "" def test_no_reasoning_content_is_noop(): @@ -132,22 +148,104 @@ def test_no_reasoning_content_is_noop(): assert "reasoning_content" not in assistant_msg +def test_thinking_tool_call_without_reasoning_gets_empty_placeholder(): + model = _make_model(extra_body={"thinking": {"type": "enabled"}}) + human = HumanMessage(content="Check the repository") + ai = AIMessage( + content="", + tool_calls=[{"id": "call-1", "name": "bash", "args": {"command": "git status"}}], + ) + base_payload = { + "messages": [ + _make_payload_message("user", "Check the repository"), + _make_payload_message( + "assistant", + None, + tool_calls=[ + { + "id": "call-1", + "type": "function", + "function": {"name": "bash", "arguments": '{"command":"git status"}'}, + } + ], + ), + ] + } + + with patch.object(type(model).__bases__[0], "_get_request_payload", return_value=base_payload): + with patch.object(model, "_convert_input") as mock_convert: + mock_convert.return_value = MagicMock(to_messages=lambda: [human, ai]) + payload = model._get_request_payload([human, ai]) + + assistant_msg = payload["messages"][1] + assert assistant_msg["content"] == "" + assert assistant_msg["reasoning_content"] == "" + + +def test_disabled_thinking_tool_call_does_not_invent_reasoning_placeholder(): + """关闭思考时不为原本不存在的 reasoning_content 扩展协议形状。""" + model = _make_model(extra_body={"thinking": {"type": "disabled"}}) + ai = AIMessage( + content="", + tool_calls=[{"id": "call-1", "name": "bash", "args": {"command": "pwd"}}], + ) + base_payload = { + "messages": [ + _make_payload_message( + "assistant", + None, + tool_calls=[ + { + "id": "call-1", + "type": "function", + "function": {"name": "bash", "arguments": '{"command":"pwd"}'}, + } + ], + ) + ] + } + + with patch.object(type(model).__bases__[0], "_get_request_payload", return_value=base_payload): + with patch.object(model, "_convert_input") as mock_convert: + mock_convert.return_value = MagicMock(to_messages=lambda: [ai]) + payload = model._get_request_payload([ai]) + + assistant_msg = payload["messages"][0] + assert assistant_msg["content"] == "" + assert "reasoning_content" not in assistant_msg + + def test_reasoning_content_multi_turn(): """All assistant turns each get their own reasoning_content.""" - model = _make_model() + model = _make_model(extra_body={"thinking": {"type": "enabled"}}) human1 = HumanMessage(content="Step 1?") - ai1 = AIMessage(content="A1", additional_kwargs={"reasoning_content": "Thought1"}) + ai1 = AIMessage( + content="", + additional_kwargs={"reasoning_content": "Thought1"}, + tool_calls=[{"id": "call-1", "name": "lookup", "args": {}}], + ) human2 = HumanMessage(content="Step 2?") ai2 = AIMessage(content="A2", additional_kwargs={"reasoning_content": "Thought2"}) base_payload = { + "tools": _TOOL_SPEC, "messages": [ _make_payload_message("user", "Step 1?"), - _make_payload_message("assistant", "A1"), + _make_payload_message( + "assistant", + None, + tool_calls=[ + { + "id": "call-1", + "type": "function", + "function": {"name": "lookup", "arguments": "{}"}, + } + ], + ), _make_payload_message("user", "Step 2?"), _make_payload_message("assistant", "A2"), - ] + ], } with patch.object(type(model).__bases__[0], "_get_request_payload", return_value=base_payload): @@ -162,19 +260,34 @@ def test_reasoning_content_multi_turn(): def test_positional_fallback_when_count_differs(): """Falls back to positional matching when payload/original message counts differ.""" - model = _make_model() + model = _make_model(extra_body={"thinking": {"type": "enabled"}}) human = HumanMessage(content="hi") - ai = AIMessage(content="hello", additional_kwargs={"reasoning_content": "My reasoning"}) + ai = AIMessage( + content="", + additional_kwargs={"reasoning_content": "My reasoning"}, + tool_calls=[{"id": "call-1", "name": "lookup", "args": {}}], + ) # Simulate count mismatch: payload has 3 messages, original has 2 extra_system = _make_payload_message("system", "You are helpful.") base_payload = { + "tools": _TOOL_SPEC, "messages": [ extra_system, _make_payload_message("user", "hi"), - _make_payload_message("assistant", "hello"), - ] + _make_payload_message( + "assistant", + None, + tool_calls=[ + { + "id": "call-1", + "type": "function", + "function": {"name": "lookup", "arguments": "{}"}, + } + ], + ), + ], } with patch.object(type(model).__bases__[0], "_get_request_payload", return_value=base_payload): @@ -184,3 +297,45 @@ def test_positional_fallback_when_count_differs(): assistant_msg = next(m for m in payload["messages"] if m["role"] == "assistant") assert assistant_msg["reasoning_content"] == "My reasoning" + + +def test_thinking_does_not_replay_reasoning_for_assistant_without_tool_calls(): + model = _make_model(extra_body={"thinking": {"type": "enabled"}}) + human = HumanMessage(content="continue") + ai = AIMessage(content="local response") + base_payload = { + "tools": _TOOL_SPEC, + "messages": [ + _make_payload_message("assistant", "local response"), + _make_payload_message("user", "continue"), + ], + } + + with patch.object(type(model).__bases__[0], "_get_request_payload", return_value=base_payload): + with patch.object(model, "_convert_input") as mock_convert: + mock_convert.return_value = MagicMock(to_messages=lambda: [ai, human]) + payload = model._get_request_payload([ai, human]) + + assistant_msg = next(m for m in payload["messages"] if m["role"] == "assistant") + assert "reasoning_content" not in assistant_msg + + +def test_legacy_fallback_assistant_is_removed_before_deepseek_replay(): + model = _make_model(extra_body={"thinking": {"type": "enabled"}}) + fallback = AIMessage( + content="temporary provider error", + additional_kwargs={"deerflow_error_fallback": True}, + ) + human = HumanMessage(content="retry") + base_payload = { + "tools": _TOOL_SPEC, + "messages": [_make_payload_message("user", "retry")], + } + + with patch.object(type(model).__bases__[0], "_get_request_payload", return_value=base_payload): + with patch.object(model, "_convert_input") as mock_convert: + mock_convert.return_value = MagicMock(to_messages=lambda: [fallback, human]) + payload = model._get_request_payload([fallback, human]) + + assert payload["messages"] == base_payload["messages"] + assert not any(message.get("role") == "assistant" for message in payload["messages"]) diff --git a/backend/tests/test_terminal_response_middleware.py b/backend/tests/test_terminal_response_middleware.py index e8151e742..5e73a8924 100644 --- a/backend/tests/test_terminal_response_middleware.py +++ b/backend/tests/test_terminal_response_middleware.py @@ -5,7 +5,7 @@ from typing import Any import pytest from langchain.agents import create_agent from langchain_core.language_models import BaseChatModel -from langchain_core.messages import AIMessage, HumanMessage, ToolMessage +from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage, ToolMessage from langchain_core.outputs import ChatGeneration, ChatResult from langchain_core.tools import tool @@ -20,9 +20,8 @@ def lookup_status() -> str: class _PostToolResponseModel(BaseChatModel): - responses: list[str] + response: AIMessage call_count: int = 0 - observed_messages: list[list[Any]] = [] @property def _llm_type(self) -> str: @@ -32,7 +31,6 @@ class _PostToolResponseModel(BaseChatModel): return self def _generate(self, messages, stop=None, run_manager=None, **kwargs): - self.observed_messages.append(list(messages)) self.call_count += 1 if self.call_count == 1: message = AIMessage( @@ -41,46 +39,7 @@ class _PostToolResponseModel(BaseChatModel): response_metadata={"finish_reason": "tool_calls"}, ) else: - message = AIMessage( - content=self.responses[self.call_count - 2], - response_metadata={"finish_reason": "stop"}, - ) - return ChatResult(generations=[ChatGeneration(message=message)]) - - async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs): - return self._generate(messages, stop=stop, run_manager=run_manager, **kwargs) - - -class _PerRunRetryBudgetModel(BaseChatModel): - call_count: int = 0 - observed_messages: list[list[Any]] = [] - - @property - def _llm_type(self) -> str: - return "per-run-retry-budget" - - def bind_tools(self, tools, **kwargs): - return self - - def _generate(self, messages, stop=None, run_manager=None, **kwargs): - self.observed_messages.append(list(messages)) - self.call_count += 1 - if self.call_count == 1: - message = AIMessage( - content="", - tool_calls=[{"id": "call-budget-1", "name": "lookup_status", "args": {}}], - response_metadata={"finish_reason": "tool_calls"}, - ) - elif self.call_count == 2: - message = AIMessage(content="", response_metadata={"finish_reason": "stop"}) - elif self.call_count == 3: - message = AIMessage( - content="I need one more status check.", - tool_calls=[{"id": "call-budget-2", "name": "lookup_status", "args": {}}], - response_metadata={"finish_reason": "tool_calls"}, - ) - else: - message = AIMessage(content="", response_metadata={"finish_reason": "stop"}) + message = self.response return ChatResult(generations=[ChatGeneration(message=message)]) async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs): @@ -95,184 +54,137 @@ def _agent(model: BaseChatModel): ) -def _empty_terminal_messages(messages: list[Any]) -> list[AIMessage]: - return [message for message in messages if isinstance(message, AIMessage) and not message.tool_calls and not message.invalid_tool_calls and not str(message.content).strip()] +def _runtime(run_id: str = "run-1"): + return type("RuntimeStub", (), {"context": {"thread_id": "thread-1", "run_id": run_id}})() -def test_retries_empty_post_tool_response_once_and_returns_model_answer(): - model = _PostToolResponseModel(responses=["", "The tool completed successfully."]) +def test_empty_post_tool_response_becomes_fallback_without_graph_retry(): + model = _PostToolResponseModel(response=AIMessage(content="", response_metadata={"finish_reason": "stop"})) result = _agent(model).invoke( {"messages": [HumanMessage(content="Check the status")]}, context={"thread_id": "thread-1", "run_id": "run-1"}, ) - assert model.call_count == 3 - final = result["messages"][-1] - assert isinstance(final, AIMessage) - assert final.content == "The tool completed successfully." - assert _empty_terminal_messages(result["messages"]) == [] - assert any(isinstance(message, HumanMessage) and message.name == "terminal_response_recovery" and message.additional_kwargs.get("hide_from_ui") is True for message in model.observed_messages[-1]) - assert not any(isinstance(message, HumanMessage) and message.name == "terminal_response_recovery" for message in result["messages"]) - - -def test_second_empty_post_tool_response_becomes_visible_error_fallback(): - model = _PostToolResponseModel(responses=["", ""]) - - result = _agent(model).invoke( - {"messages": [HumanMessage(content="Check the status")]}, - context={"thread_id": "thread-2", "run_id": "run-2"}, - ) - - assert model.call_count == 3 + assert model.call_count == 2 final = result["messages"][-1] assert isinstance(final, AIMessage) assert "returned no final response" in str(final.content) assert final.additional_kwargs["deerflow_error_fallback"] is True - assert _empty_terminal_messages(result["messages"]) == [] - assert _extract_llm_error_fallback_message(result) == ("Model returned an empty terminal response after one retry") + assert _extract_llm_error_fallback_message(result) == "Model returned an empty terminal response" + assert not any(isinstance(message, RemoveMessage) for message in result["messages"]) @pytest.mark.asyncio -async def test_async_graph_retries_empty_post_tool_response_once(): - model = _PostToolResponseModel(responses=["", "Recovered asynchronously."]) +async def test_async_empty_post_tool_response_becomes_fallback_without_graph_retry(): + model = _PostToolResponseModel(response=AIMessage(content="", response_metadata={"finish_reason": "stop"})) result = await _agent(model).ainvoke( {"messages": [HumanMessage(content="Check the status")]}, context={"thread_id": "thread-async", "run_id": "run-async"}, ) - assert model.call_count == 3 - assert result["messages"][-1].content == "Recovered asynchronously." - assert _empty_terminal_messages(result["messages"]) == [] + assert model.call_count == 2 + assert result["messages"][-1].additional_kwargs["deerflow_error_fallback"] is True -def test_graph_with_thread_id_only_keeps_recovery_state_across_model_loop(): - model = _PostToolResponseModel(responses=["", "Recovered without a run id."]) +def test_direct_fallback_replaces_same_message_without_remove_or_jump(): + middleware = TerminalResponseMiddleware() + empty = AIMessage(id="empty-1", content="", response_metadata={"finish_reason": "stop"}) + state = { + "messages": [ + HumanMessage(content="Check the status"), + ToolMessage(content="tool completed", tool_call_id="call-1"), + empty, + ] + } - result = _agent(model).invoke( - {"messages": [HumanMessage(content="Check the status")]}, - context={"thread_id": "thread-only"}, - ) + result = middleware.after_model(state, _runtime()) - assert model.call_count == 3 - assert result["messages"][-1].content == "Recovered without a run id." - assert _empty_terminal_messages(result["messages"]) == [] + assert result is not None + assert "jump_to" not in result + assert len(result["messages"]) == 1 + replacement = result["messages"][0] + assert isinstance(replacement, AIMessage) + assert replacement.id == "empty-1" + assert replacement.additional_kwargs["deerflow_error_fallback"] is True + assert not isinstance(replacement, RemoveMessage) -def test_recovery_budget_is_once_per_run_even_when_retry_calls_another_tool(): - model = _PerRunRetryBudgetModel() - - result = _agent(model).invoke( - {"messages": [HumanMessage(content="Check the status twice")]}, - context={"thread_id": "thread-budget", "run_id": "run-budget"}, - ) - - assert model.call_count == 4 - final = result["messages"][-1] - assert final.additional_kwargs["deerflow_error_fallback"] is True - assert _empty_terminal_messages(result["messages"]) == [] - recovery_prompt_count = sum(1 for request_messages in model.observed_messages for message in request_messages if isinstance(message, HumanMessage) and message.name == "terminal_response_recovery") - assert recovery_prompt_count == 1 - - -def test_empty_response_without_tool_result_is_not_retried(): +def test_empty_response_without_tool_result_is_not_handled_by_terminal_guard(): middleware = TerminalResponseMiddleware() message = AIMessage(content="", response_metadata={"finish_reason": "stop"}) state = {"messages": [HumanMessage(content="Hello"), message]} - runtime = type("RuntimeStub", (), {"context": {"thread_id": "thread-3", "run_id": "run-3"}})() - assert middleware.after_model(state, runtime) is None + assert middleware.after_model(state, _runtime()) is None -def test_tool_call_intent_is_not_treated_as_empty_terminal_response(): +def test_tool_call_is_not_treated_as_empty_terminal(): middleware = TerminalResponseMiddleware() message = AIMessage( content="", tool_calls=[{"id": "call-2", "name": "lookup_status", "args": {}}], response_metadata={"finish_reason": "tool_calls"}, ) - state = {"messages": [HumanMessage(content="Hello"), message]} - runtime = type("RuntimeStub", (), {"context": {"thread_id": "thread-4", "run_id": "run-4"}})() + state: dict[str, list[Any]] = { + "messages": [ + HumanMessage(content="Check the status"), + ToolMessage(content="tool completed", tool_call_id="call-2"), + message, + ] + } - assert middleware.after_model(state, runtime) is None + assert middleware.after_model(state, _runtime()) is None @pytest.mark.parametrize( "message", [ - AIMessage(content="", invalid_tool_calls=[{"id": "bad-1", "name": "lookup_status", "args": "{"}]), - AIMessage(content="", additional_kwargs={"function_call": {"name": "lookup_status", "arguments": "{}"}}), - AIMessage(content="", response_metadata={"finish_reason": "function_call"}), + AIMessage(content=" ", response_metadata={"finish_reason": "stop"}), + AIMessage( + content="", + additional_kwargs={"reasoning_content": "thinking"}, + response_metadata={"finish_reason": "stop"}, + ), + AIMessage(content="", response_metadata={"finish_reason": "length"}), ], ) -def test_invalid_or_legacy_tool_call_intent_is_not_treated_as_empty_terminal_response(message): +def test_nonvisible_post_tool_response_becomes_terminal_fallback(message: AIMessage): middleware = TerminalResponseMiddleware() - state = {"messages": [HumanMessage(content="Hello"), message]} - runtime = type("RuntimeStub", (), {"context": {"thread_id": "thread-5", "run_id": "run-5"}})() - - assert middleware.after_model(state, runtime) is None - - -def test_after_agent_clears_retry_state_for_the_run(): - middleware = TerminalResponseMiddleware() - runtime = type("RuntimeStub", (), {"context": {"thread_id": "thread-6", "run_id": "run-6"}})() - empty_after_tool = { + state: dict[str, list[Any]] = { "messages": [ HumanMessage(content="Check the status"), - ToolMessage(content="tool completed", tool_call_id="call-6"), - AIMessage(content="", response_metadata={"finish_reason": "stop"}), + ToolMessage(content="tool completed", tool_call_id="call-2"), + message, ] } - first = middleware.after_model(empty_after_tool, runtime) - assert first is not None and first["jump_to"] == "model" - middleware.after_agent(empty_after_tool, runtime) - second = middleware.after_model(empty_after_tool, runtime) - assert second is not None and second["jump_to"] == "model" + result = middleware.after_model(state, _runtime()) + + assert result is not None + replacement = result["messages"][0] + assert "returned no final response" in str(replacement.content) + assert replacement.additional_kwargs["deerflow_error_fallback"] is True + if "reasoning_content" in message.additional_kwargs: + assert replacement.additional_kwargs["reasoning_content"] == "thinking" -def test_before_agent_clears_same_run_state_for_resumed_invocation(): +def test_thinking_blocks_are_preserved_when_terminal_fallback_is_appended(): middleware = TerminalResponseMiddleware() - runtime = type("RuntimeStub", (), {"context": {"thread_id": "thread-7", "run_id": "run-7"}})() - empty_after_tool = { - "messages": [ - HumanMessage(content="Check the status"), - ToolMessage(content="tool completed", tool_call_id="call-7"), - AIMessage(content="", response_metadata={"finish_reason": "stop"}), - ] - } - - first = middleware.after_model(empty_after_tool, runtime) - assert first is not None and first["jump_to"] == "model" - middleware.before_agent(empty_after_tool, runtime) - resumed = middleware.after_model(empty_after_tool, runtime) - assert resumed is not None and resumed["jump_to"] == "model" - - -def test_tool_history_without_real_user_message_does_not_trigger_recovery(): - middleware = TerminalResponseMiddleware() - runtime = type("RuntimeStub", (), {"context": {"thread_id": "thread-8", "run_id": "run-8"}})() + thinking_block = {"type": "thinking", "thinking": "internal reasoning"} + message = AIMessage(content=[thinking_block], response_metadata={"finish_reason": "stop"}) state = { "messages": [ - HumanMessage(content="internal", additional_kwargs={"hide_from_ui": True}), - ToolMessage(content="tool completed", tool_call_id="call-8"), - AIMessage(content="", response_metadata={"finish_reason": "stop"}), + HumanMessage(content="Check the status"), + ToolMessage(content="tool completed", tool_call_id="call-2"), + message, ] } - assert middleware.after_model(state, runtime) is None + result = middleware.after_model(state, _runtime()) - -def test_abandoned_run_state_is_bounded(): - middleware = TerminalResponseMiddleware() - - for index in range(1001): - key = (f"thread-{index}", f"run-{index}") - middleware._retry_counts[key] = 1 - middleware._pending_prompts[key] = True - - assert len(middleware._retry_counts) == 1000 - assert len(middleware._pending_prompts) == 1000 - assert ("thread-0", "run-0") not in middleware._retry_counts - assert ("thread-0", "run-0") not in middleware._pending_prompts + assert result is not None + content = result["messages"][0].content + assert content[0] == thinking_block + assert content[-1]["type"] == "text" + assert "returned no final response" in content[-1]["text"]