mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-21 10:15:47 +00:00
fix: ensure visible response after tool runs (#4033)
* fix: ensure visible response after tool runs * fix: clean up terminal response recovery state * fix: bound terminal recovery state
This commit is contained in:
Huixin615
@@ -739,6 +739,8 @@ This is the difference between a chatbot with tool access and an agent with an a
|
||||
|
||||
**Strict Tool-Call Recovery**: When a provider or middleware interrupts a tool-call loop, DeerFlow now strips provider-level raw tool-call metadata on forced-stop assistant messages and injects placeholder tool results for dangling calls before the next model invocation. This keeps OpenAI-compatible reasoning models that strictly validate `tool_call_id` sequences from failing with malformed history errors.
|
||||
|
||||
**Visible Tool-Run Completion**: For interactive turns, DeerFlow retries an empty post-tool final response once, then surfaces a visible error instead of reporting a silent successful run.
|
||||
|
||||
### Long-Term Memory
|
||||
|
||||
Most agents forget everything the moment a conversation ends. DeerFlow remembers.
|
||||
|
||||
+4
-3
@@ -249,9 +249,10 @@ Lead-agent middlewares are assembled in strict order across three functions: the
|
||||
26. **SubagentLimitMiddleware** - *(optional, if `subagent_enabled`)* Truncates excess `task` tool calls to enforce the `MAX_CONCURRENT_SUBAGENTS` limit
|
||||
27. **LoopDetectionMiddleware** - *(optional, if `loop_detection.enabled`)* Detects repeated tool-call loops; hard-stop clears both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer; stamps `loop_capped` via `consume_stop_reason` (#3875 Phase 2), symmetric to `TokenBudgetMiddleware`
|
||||
28. **TokenBudgetMiddleware** - *(optional, if `token_budget.enabled`)* Enforces per-run token limits
|
||||
29. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the safety/clarification tail
|
||||
30. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after custom middlewares so LangChain's reverse-order `after_model` dispatch runs it first
|
||||
31. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, writes a readable `ToolMessage.content` fallback plus structured `ToolMessage.artifact.human_input` request payload, and interrupts via `Command(goto=END)` (must be last). Because this middleware can short-circuit tool execution before LangChain emits `on_tool_end`, `RunJournal` performs a root-run final reconciliation for allowlisted clarification `ToolMessage`s whose `tool_call_id` was produced by the current run, so human-input request cards remain recoverable from `run_events` after checkpoint compaction. Human Input Card replies are submitted as `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden response sources (currently `ask_clarification`) as `llm.human.input`, which preserves answered-card state after compaction without exposing generic internal hidden context.
|
||||
29. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before the terminal-response/safety/clarification tail
|
||||
30. **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
|
||||
31. **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 middlewares so LangChain's reverse-order `after_model` dispatch runs it first
|
||||
32. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, writes a readable `ToolMessage.content` fallback plus structured `ToolMessage.artifact.human_input` request payload, and interrupts via `Command(goto=END)` (must be last). Because this middleware can short-circuit tool execution before LangChain emits `on_tool_end`, `RunJournal` performs a root-run final reconciliation for allowlisted clarification `ToolMessage`s whose `tool_call_id` was produced by the current run, so human-input request cards remain recoverable from `run_events` after checkpoint compaction. Human Input Card replies are submitted as `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden response sources (currently `ask_clarification`) as `llm.human.input`, which preserves answered-card state after compaction without exposing generic internal hidden context.
|
||||
|
||||
### Configuration System
|
||||
|
||||
|
||||
@@ -33,6 +33,7 @@ from deerflow.agents.middlewares.memory_middleware import MemoryMiddleware
|
||||
from deerflow.agents.middlewares.safety_finish_reason_middleware import SafetyFinishReasonMiddleware
|
||||
from deerflow.agents.middlewares.subagent_limit_middleware import SubagentLimitMiddleware
|
||||
from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware, create_summarization_middleware
|
||||
from deerflow.agents.middlewares.terminal_response_middleware import TerminalResponseMiddleware
|
||||
from deerflow.agents.middlewares.title_middleware import TitleMiddleware
|
||||
from deerflow.agents.middlewares.todo_middleware import TodoMiddleware
|
||||
from deerflow.agents.middlewares.token_usage_middleware import TokenUsageMiddleware
|
||||
@@ -369,11 +370,16 @@ def build_middlewares(
|
||||
if custom_middlewares:
|
||||
middlewares.extend(custom_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.
|
||||
middlewares.append(TerminalResponseMiddleware())
|
||||
|
||||
# SafetyFinishReasonMiddleware — suppress tool execution when the provider
|
||||
# safety-terminated the response. Registered after custom middlewares so
|
||||
# that LangChain's reverse-order after_model dispatch runs Safety first;
|
||||
# cleared tool_calls then flow through Loop/Subagent accounting without
|
||||
# firing extra alarms. See safety_finish_reason_middleware.py docstring.
|
||||
# safety-terminated the response. Registered after the terminal-response
|
||||
# and custom middlewares so LangChain's reverse-order after_model dispatch
|
||||
# runs Safety first; cleared tool_calls then flow through the remaining
|
||||
# accounting/terminal guards without firing extra alarms.
|
||||
safety_config = resolved_app_config.safety_finish_reason
|
||||
if safety_config.enabled:
|
||||
middlewares.append(SafetyFinishReasonMiddleware.from_config(safety_config))
|
||||
|
||||
@@ -0,0 +1,214 @@
|
||||
"""Ensure tool-using lead-agent turns end with a visible assistant response."""
|
||||
|
||||
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 langgraph.runtime import Runtime
|
||||
|
||||
from deerflow.agents.middlewares._bounded_dict import BoundedDict
|
||||
|
||||
_RECOVERY_PROMPT = (
|
||||
"<system_reminder>\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"
|
||||
"</system_reminder>"
|
||||
)
|
||||
|
||||
_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
|
||||
|
||||
|
||||
def _tool_result_in_current_turn(messages: list[Any]) -> bool:
|
||||
"""Return whether a tool result follows the latest real user message."""
|
||||
latest_user_index = -1
|
||||
for index, message in enumerate(messages):
|
||||
if not isinstance(message, HumanMessage):
|
||||
continue
|
||||
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)
|
||||
|
||||
@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):
|
||||
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",
|
||||
}
|
||||
)
|
||||
fallback = last.model_copy(
|
||||
update={
|
||||
"content": _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
|
||||
@@ -372,10 +372,18 @@ def test_build_middlewares_uses_resolved_model_name_for_vision(monkeypatch):
|
||||
|
||||
assert any(isinstance(m, lead_agent_module.ViewImageMiddleware) for m in middlewares)
|
||||
# verify the custom middleware is injected correctly.
|
||||
# Chain tail order after the custom middleware is:
|
||||
# ..., custom, SafetyFinishReasonMiddleware, ClarificationMiddleware
|
||||
# so the custom mock sits at index [-3].
|
||||
assert len(middlewares) > 0 and isinstance(middlewares[-3], MagicMock)
|
||||
# With this test's default safety config enabled, the tail order is:
|
||||
# ..., custom, TerminalResponseMiddleware, SafetyFinishReasonMiddleware,
|
||||
# ClarificationMiddleware, so the custom mock sits at index [-4].
|
||||
assert len(middlewares) > 0 and isinstance(middlewares[-4], MagicMock)
|
||||
|
||||
from deerflow.agents.middlewares.clarification_middleware import ClarificationMiddleware
|
||||
from deerflow.agents.middlewares.safety_finish_reason_middleware import SafetyFinishReasonMiddleware
|
||||
from deerflow.agents.middlewares.terminal_response_middleware import TerminalResponseMiddleware
|
||||
|
||||
assert isinstance(middlewares[-3], TerminalResponseMiddleware)
|
||||
assert isinstance(middlewares[-2], SafetyFinishReasonMiddleware)
|
||||
assert isinstance(middlewares[-1], ClarificationMiddleware)
|
||||
|
||||
|
||||
def test_build_middlewares_passes_explicit_app_config_to_shared_factory(monkeypatch):
|
||||
|
||||
@@ -0,0 +1,278 @@
|
||||
from __future__ import annotations
|
||||
|
||||
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.outputs import ChatGeneration, ChatResult
|
||||
from langchain_core.tools import tool
|
||||
|
||||
from deerflow.agents.middlewares.terminal_response_middleware import TerminalResponseMiddleware
|
||||
from deerflow.runtime.runs.worker import _extract_llm_error_fallback_message
|
||||
|
||||
|
||||
@tool
|
||||
def lookup_status() -> str:
|
||||
"""Return a deterministic tool result."""
|
||||
return "tool completed"
|
||||
|
||||
|
||||
class _PostToolResponseModel(BaseChatModel):
|
||||
responses: list[str]
|
||||
call_count: int = 0
|
||||
observed_messages: list[list[Any]] = []
|
||||
|
||||
@property
|
||||
def _llm_type(self) -> str:
|
||||
return "post-tool-response"
|
||||
|
||||
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-1", "name": "lookup_status", "args": {}}],
|
||||
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"})
|
||||
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)
|
||||
|
||||
|
||||
def _agent(model: BaseChatModel):
|
||||
return create_agent(
|
||||
model=model,
|
||||
tools=[lookup_status],
|
||||
middleware=[TerminalResponseMiddleware()],
|
||||
)
|
||||
|
||||
|
||||
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 test_retries_empty_post_tool_response_once_and_returns_model_answer():
|
||||
model = _PostToolResponseModel(responses=["", "The tool completed successfully."])
|
||||
|
||||
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
|
||||
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")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_graph_retries_empty_post_tool_response_once():
|
||||
model = _PostToolResponseModel(responses=["", "Recovered asynchronously."])
|
||||
|
||||
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"]) == []
|
||||
|
||||
|
||||
def test_graph_with_thread_id_only_keeps_recovery_state_across_model_loop():
|
||||
model = _PostToolResponseModel(responses=["", "Recovered without a run id."])
|
||||
|
||||
result = _agent(model).invoke(
|
||||
{"messages": [HumanMessage(content="Check the status")]},
|
||||
context={"thread_id": "thread-only"},
|
||||
)
|
||||
|
||||
assert model.call_count == 3
|
||||
assert result["messages"][-1].content == "Recovered without a run id."
|
||||
assert _empty_terminal_messages(result["messages"]) == []
|
||||
|
||||
|
||||
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():
|
||||
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
|
||||
|
||||
|
||||
def test_tool_call_intent_is_not_treated_as_empty_terminal_response():
|
||||
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"}})()
|
||||
|
||||
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"}),
|
||||
],
|
||||
)
|
||||
def test_invalid_or_legacy_tool_call_intent_is_not_treated_as_empty_terminal_response(message):
|
||||
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 = {
|
||||
"messages": [
|
||||
HumanMessage(content="Check the status"),
|
||||
ToolMessage(content="tool completed", tool_call_id="call-6"),
|
||||
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.after_agent(empty_after_tool, runtime)
|
||||
second = middleware.after_model(empty_after_tool, runtime)
|
||||
assert second is not None and second["jump_to"] == "model"
|
||||
|
||||
|
||||
def test_before_agent_clears_same_run_state_for_resumed_invocation():
|
||||
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"}})()
|
||||
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"}),
|
||||
]
|
||||
}
|
||||
|
||||
assert middleware.after_model(state, runtime) is None
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user