mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 05:06:59 +00:00
fix(harness): stop length-capped Responses tool calls (#6946)
Signed-off-by: CorgiBoyG <111257566+CorgiBoyG@users.noreply.github.com>
This commit is contained in:
RongJie G
GitHub
parent
fd61c48fa2
commit
b0a545ec90
@ -353,6 +353,8 @@ streams simulate chunks from a non-streaming response, while no-tool streams sta
|
||||
|
||||
To route OpenAI models through `/v1/responses`, keep using `langchain_openai:ChatOpenAI` and set `use_responses_api: true` with `output_version: responses/v1`.
|
||||
|
||||
The lead agent also recognizes Responses output-token caps (`status: incomplete` with `incomplete_details.reason: max_output_tokens`), including streaming responses. It suppresses tool execution, removes the suppressed calls from later model requests, and preserves text and reasoning. A capped tool-call response includes a visible notice; other incomplete reasons retain their existing handling.
|
||||
|
||||
When thinking is disabled on a legacy profile, blank `extra_body`, `thinking`, and nested `extra_body.thinking` keys in a `when_thinking_enabled` template are treated as absent during dialect inference. When thinking is enabled, template values are forwarded as configured.
|
||||
|
||||
Models whose provider contract differs from DeerFlow's generic thinking/effort assumptions can declare a per-model mapping-valued `reasoning:` block (thinking `unsupported`/`optional`/`required`, the accepted effort values with aliases and a default, the payload dialect, and the reasoning-history requirement). The setup wizard's Z.AI GLM-5.3-Flash profile uses it: thinking stays on for every foreground and background call, and the effort selector offers the model's own `low`/`high`/`max` levels. Ollama's existing boolean `reasoning: true` remains a native provider setting and is forwarded to ChatOllama. When migrating a profile to a custom effort `path`, remove any old `reasoning_effort` setting from the profile and thinking templates; configuration validation rejects the leftover key. The chat UI drops a remembered provider-specific effort when switching to a legacy model that does not advertise it. Profiles without the block keep their existing provider behavior. See `config.example.yaml` for the shape and the equivalent manual configuration.
|
||||
|
||||
@ -137,7 +137,7 @@ Before changing a later authorization phase, read the [authorization RFC](../../
|
||||
35. **Custom middlewares** - *(optional)* Any `custom_middlewares` passed to `build_middlewares` are injected here, before config-declared extensions and the terminal-response/safety/clarification tail
|
||||
36. **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.
|
||||
37. **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
|
||||
38. **ModelLengthFinishReasonMiddleware** - Match stamps `stop_reason=model_length_capped` and ends tool loop. Suppresses calls, appends notice even with partial text, and stamps `model_length_termination` so downstream guards stand down. Preserves content blocks.
|
||||
38. **ModelLengthFinishReasonMiddleware** - Responses: same-container `status=incomplete` + `incomplete_details.reason=max_output_tokens`. Suppress calls via shared cloning after clearing invalid calls; keep non-call blocks/metadata. Stamp cap markers; add notice.
|
||||
39. **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
|
||||
40. **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: []` — 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. `RunJournal` reconciles visible, unpersisted `ToolMessage`s for current-run lead-agent calls at the next lead-agent `on_chat_model_start` after consumption, or at successful root `on_chain_end`. This preserves any middleware short-circuit, including blocked writes, across later model errors and checkpoint compaction (#4666). `_remember_current_run_tool_calls` excludes subagents; their results stay in `subagent.step`. 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`.
|
||||
|
||||
|
||||
@ -28,13 +28,14 @@ from deerflow.agents.middlewares.model_length_termination_detectors import (
|
||||
default_detectors,
|
||||
)
|
||||
from deerflow.agents.middlewares.model_response import append_visible_text, has_tool_call_intent, has_visible_content
|
||||
from deerflow.agents.middlewares.tool_call_metadata import clone_ai_message_with_tool_calls
|
||||
|
||||
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 _tool_call_summary(message: AIMessage) -> tuple[int, list[str]]:
|
||||
def _tool_call_summary(message: AIMessage, cleaned_content: Any) -> 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 [])]
|
||||
@ -47,8 +48,11 @@ def _tool_call_summary(message: AIMessage) -> tuple[int, list[str]]:
|
||||
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)
|
||||
if not calls and isinstance(message.content, list) and isinstance(cleaned_content, list):
|
||||
# The shared cleanup keeps non-call blocks by identity. Summarize
|
||||
# the removed blocks without duplicating its provider type list.
|
||||
kept_blocks = {id(block) for block in cleaned_content}
|
||||
calls = [block for block in message.content if id(block) not in kept_blocks]
|
||||
|
||||
for call in calls:
|
||||
if not isinstance(call, dict):
|
||||
@ -62,20 +66,6 @@ def _tool_call_summary(message: AIMessage) -> tuple[int, list[str]]:
|
||||
return len(calls), names
|
||||
|
||||
|
||||
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 termination and block truncated tool calls."""
|
||||
|
||||
@ -134,15 +124,16 @@ class ModelLengthFinishReasonMiddleware(AgentMiddleware[AgentState]):
|
||||
},
|
||||
)
|
||||
|
||||
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})
|
||||
# Unlike dangling-call repair, a length stop must also suppress invalid
|
||||
# calls, so clear them before syncing the native provider content.
|
||||
content_source = clone_ai_message_with_tool_calls(last.model_copy(update={"invalid_tool_calls": []}), [])
|
||||
contains_tool_call = has_tool_call_intent(last) or content_source.content is not last.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, [])
|
||||
suppressed_count, suppressed_names = _tool_call_summary(last, content_source.content) 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)
|
||||
@ -154,12 +145,14 @@ class ModelLengthFinishReasonMiddleware(AgentMiddleware[AgentState]):
|
||||
"suppressed_tool_call_count": suppressed_count,
|
||||
"suppressed_tool_call_names": suppressed_names,
|
||||
}
|
||||
replacement = last.model_copy(
|
||||
replacement = content_source.model_copy(
|
||||
update={
|
||||
"content": append_visible_text(content_source, _MODEL_LENGTH_CAPPED_CONTENT),
|
||||
"tool_calls": [],
|
||||
"invalid_tool_calls": [],
|
||||
"additional_kwargs": additional_kwargs,
|
||||
# Keep the provider's termination evidence unchanged.
|
||||
"response_metadata": last.response_metadata,
|
||||
}
|
||||
)
|
||||
return {"messages": [replacement]}
|
||||
|
||||
@ -67,6 +67,26 @@ class OpenAICompatibleLengthDetector:
|
||||
)
|
||||
|
||||
|
||||
class OpenAIResponsesLengthDetector:
|
||||
"""Responses API output-token cap, including ``response.incomplete`` SSE."""
|
||||
|
||||
name = "openai_responses_length"
|
||||
|
||||
def detect(self, message: AIMessage) -> ModelLengthTermination | None:
|
||||
for container_name in ("response_metadata", "additional_kwargs"):
|
||||
metadata = getattr(message, container_name, None)
|
||||
if not isinstance(metadata, dict) or metadata.get("status") != "incomplete":
|
||||
continue
|
||||
details = metadata.get("incomplete_details")
|
||||
if isinstance(details, dict) and details.get("reason") == "max_output_tokens":
|
||||
return ModelLengthTermination(
|
||||
detector=self.name,
|
||||
reason_field="incomplete_details.reason",
|
||||
reason_value="max_output_tokens",
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
class AnthropicMaxTokensDetector:
|
||||
"""Anthropic ``stop_reason == "max_tokens"`` signal."""
|
||||
|
||||
@ -113,6 +133,7 @@ def default_detectors() -> list[ModelLengthTerminationDetector]:
|
||||
OpenAICompatibleLengthDetector(),
|
||||
AnthropicMaxTokensDetector(),
|
||||
GeminiMaxTokensDetector(),
|
||||
OpenAIResponsesLengthDetector(),
|
||||
]
|
||||
|
||||
|
||||
@ -122,5 +143,6 @@ __all__ = [
|
||||
"ModelLengthTermination",
|
||||
"ModelLengthTerminationDetector",
|
||||
"OpenAICompatibleLengthDetector",
|
||||
"OpenAIResponsesLengthDetector",
|
||||
"default_detectors",
|
||||
]
|
||||
|
||||
340
backend/tests/test_model_length_responses.py
Normal file
340
backend/tests/test_model_length_responses.py
Normal file
@ -0,0 +1,340 @@
|
||||
"""Length caps through real Responses serialization and Agent tool execution."""
|
||||
|
||||
import json
|
||||
from types import SimpleNamespace
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from langchain.agents import create_agent
|
||||
from langchain_core.messages import AIMessage, HumanMessage, message_chunk_to_message
|
||||
from langchain_core.tools import tool
|
||||
from langchain_openai import ChatOpenAI
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
|
||||
from deerflow.agents.middlewares.model_length_finish_reason_middleware import (
|
||||
MODEL_LENGTH_CAPPED_STOP_REASON,
|
||||
ModelLengthFinishReasonMiddleware,
|
||||
)
|
||||
|
||||
|
||||
def _response(*, arguments="{}", status="incomplete", reason="max_output_tokens", final=False):
|
||||
output = (
|
||||
[{"type": "message", "id": "msg-final", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "Finished.", "annotations": []}]}]
|
||||
if final
|
||||
else [
|
||||
{"type": "reasoning", "id": "rs-length", "summary": [{"type": "summary_text", "text": "Preparing the result."}]},
|
||||
{"type": "function_call", "id": "fc-length", "call_id": "call-length", "name": "record_write", "arguments": arguments, "status": status},
|
||||
]
|
||||
)
|
||||
return {
|
||||
"id": "resp-final" if final else "resp-length",
|
||||
"object": "response",
|
||||
"created_at": 1,
|
||||
"model": "gpt-6",
|
||||
"status": "completed" if final else status,
|
||||
"error": None,
|
||||
"incomplete_details": None if final or reason is None else {"reason": reason},
|
||||
"output": output,
|
||||
"parallel_tool_calls": True,
|
||||
"tool_choice": "auto",
|
||||
"tools": [],
|
||||
"usage": {"input_tokens": 10, "input_tokens_details": {"cached_tokens": 0}, "output_tokens": 8, "output_tokens_details": {"reasoning_tokens": 2}, "total_tokens": 18},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def responses_transport():
|
||||
requests = []
|
||||
state = SimpleNamespace(arguments="{}", status="incomplete", reason="max_output_tokens")
|
||||
|
||||
def send(request):
|
||||
body = json.loads(request.content)
|
||||
requests.append(body)
|
||||
assert request.url.path == "/v1/responses"
|
||||
input_items = body["input"]
|
||||
pending = {item["call_id"] for item in input_items if item.get("type") == "function_call"}
|
||||
answered = {item["call_id"] for item in input_items if item.get("type") == "function_call_output"}
|
||||
if pending - answered:
|
||||
return httpx.Response(400, json={"error": {"message": "No tool output found for function call call-length", "type": "invalid_request_error"}})
|
||||
payload = _response(arguments=state.arguments, status=state.status, reason=state.reason, final=len(requests) > 1)
|
||||
if body.get("stream"):
|
||||
events = [{"type": "response.created", "response": {**payload, "status": "in_progress", "incomplete_details": None, "output": [], "usage": None}}]
|
||||
for index, item in enumerate(payload["output"]):
|
||||
added = {**item, "arguments": ""} if item["type"] == "function_call" else item
|
||||
events.append({"type": "response.output_item.added", "output_index": index, "item": added})
|
||||
if item["type"] == "function_call":
|
||||
events.append({"type": "response.function_call_arguments.delta", "output_index": index, "item_id": item["id"], "delta": item["arguments"]})
|
||||
elif item["type"] == "message":
|
||||
events.append({"type": "response.output_text.delta", "output_index": index, "content_index": 0, "item_id": item["id"], "delta": item["content"][0]["text"]})
|
||||
events.append({"type": "response.output_item.done", "output_index": index, "item": item})
|
||||
events.append({"type": "response.completed" if payload["status"] == "completed" else "response.incomplete", "response": payload})
|
||||
for sequence_number, event in enumerate(events):
|
||||
event["sequence_number"] = sequence_number
|
||||
return httpx.Response(200, headers={"content-type": "text/event-stream"}, content="".join(f"data: {json.dumps(event)}\n\n" for event in events))
|
||||
return httpx.Response(200, json=payload)
|
||||
|
||||
return httpx.MockTransport(send), requests, state
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def responses_provider(responses_transport):
|
||||
transport, requests, state = responses_transport
|
||||
with httpx.Client(transport=transport) as http:
|
||||
model = ChatOpenAI(model="gpt-6", api_key="offline-not-a-key", use_responses_api=True, http_client=http, max_retries=0, max_tokens=8)
|
||||
yield model, requests, state
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def async_responses_provider(responses_transport):
|
||||
transport, requests, state = responses_transport
|
||||
with httpx.Client(transport=transport) as http:
|
||||
async with httpx.AsyncClient(transport=transport) as async_http:
|
||||
model = ChatOpenAI(model="gpt-6", api_key="offline-not-a-key", use_responses_api=True, http_client=http, http_async_client=async_http, max_retries=0, max_tokens=8)
|
||||
yield model, requests, state
|
||||
|
||||
|
||||
def _message(model, streaming):
|
||||
if not streaming:
|
||||
return model.invoke([HumanMessage("Save the result.")])
|
||||
collected = None
|
||||
for chunk in model.stream([HumanMessage("Save the result.")]):
|
||||
collected = chunk if collected is None else collected + chunk
|
||||
assert collected is not None
|
||||
return message_chunk_to_message(collected)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("streaming", [False, True])
|
||||
def test_native_responses_length_signal_stops_tool_calls(responses_provider, streaming):
|
||||
model, requests, _ = responses_provider
|
||||
message = _message(model, streaming)
|
||||
assert message.response_metadata["status"] == "incomplete"
|
||||
assert message.response_metadata["incomplete_details"] == {"reason": "max_output_tokens"}
|
||||
assert message.tool_calls
|
||||
assert "finish_reason" not in message.response_metadata
|
||||
context = {}
|
||||
update = ModelLengthFinishReasonMiddleware().after_model({"messages": [message]}, SimpleNamespace(context=context))
|
||||
assert update is not None
|
||||
assert context["stop_reason"] == MODEL_LENGTH_CAPPED_STOP_REASON
|
||||
replacement = update["messages"][0]
|
||||
assert replacement.tool_calls == []
|
||||
assert replacement.invalid_tool_calls == []
|
||||
assert len(requests) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("streaming", [False, True])
|
||||
def test_responses_length_cap_prevents_agent_write_and_keeps_history_replayable(responses_provider, streaming):
|
||||
model, requests, _ = responses_provider
|
||||
model.streaming = streaming
|
||||
writes = []
|
||||
|
||||
@tool
|
||||
def record_write() -> str:
|
||||
"""Record one observable write."""
|
||||
writes.append("written")
|
||||
return "Saved."
|
||||
|
||||
graph = create_agent(model, [record_write], middleware=[ModelLengthFinishReasonMiddleware()], checkpointer=InMemorySaver())
|
||||
context = {"thread_id": "length-regression", "run_id": "length-run"}
|
||||
config = {"configurable": {"thread_id": "length-regression"}}
|
||||
result = graph.invoke({"messages": [HumanMessage("Save the result.")]}, config=config, context=context)
|
||||
assert writes == [], "A length-capped provider response must not execute its tool call"
|
||||
assert context["stop_reason"] == MODEL_LENGTH_CAPPED_STOP_REASON
|
||||
assert result["messages"][-1].additional_kwargs["model_length_termination"]["suppressed_tool_call_count"] == 1
|
||||
assert len(requests) == 1
|
||||
graph.invoke({"messages": [HumanMessage("Continue with a new turn.")]}, config=config, context={})
|
||||
assert len(requests) == 2
|
||||
assert not any(item.get("type") in ("function_call", "function_call_output") for item in requests[-1]["input"])
|
||||
assert writes == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("arguments", ["{}", '{"value":'])
|
||||
@pytest.mark.parametrize("streaming", [False, True])
|
||||
def test_length_cleanup_removes_native_frames_before_next_responses_request(responses_provider, arguments, streaming):
|
||||
model, requests, state = responses_provider
|
||||
state.arguments = arguments
|
||||
message = _message(model, streaming)
|
||||
# Isolate cleanup independently of the missing native detector. A provider
|
||||
# adapter may also expose the established Chat Completions length spelling.
|
||||
message = message.model_copy(update={"response_metadata": {**message.response_metadata, "finish_reason": "length"}})
|
||||
original = message.model_dump()
|
||||
update = ModelLengthFinishReasonMiddleware().after_model({"messages": [message]}, SimpleNamespace(context={}))
|
||||
assert update is not None
|
||||
replacement = update["messages"][0]
|
||||
assert replacement.tool_calls == []
|
||||
assert replacement.invalid_tool_calls == []
|
||||
final = model.invoke([HumanMessage("Save the result."), replacement, HumanMessage("Continue.")])
|
||||
assert final.text == "Finished."
|
||||
assert not any(item.get("type") == "function_call" for item in requests[-1]["input"])
|
||||
assert message.model_dump() == original
|
||||
assert any(block.get("type") == "reasoning" for block in replacement.content)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"metadata",
|
||||
[
|
||||
{"status": "completed", "incomplete_details": {"reason": "max_output_tokens"}},
|
||||
{"status": "incomplete", "incomplete_details": {"reason": "content_filter"}},
|
||||
{"status": "in_progress", "incomplete_details": {"reason": "max_output_tokens"}},
|
||||
{"status": "incomplete"},
|
||||
{"incomplete_details": {"reason": "max_output_tokens"}},
|
||||
{"status": "incomplete", "incomplete_details": None},
|
||||
{"status": "incomplete", "incomplete_details": "max_output_tokens"},
|
||||
{"status": "incomplete", "incomplete_details": {"reason": ["max_output_tokens"]}},
|
||||
{"status": ["incomplete"], "incomplete_details": {"reason": "max_output_tokens"}},
|
||||
{"status": "incomplete", "incomplete_details": []},
|
||||
{"status": "incomplete", "incomplete_details": {"reason": "MAX_OUTPUT_TOKENS"}},
|
||||
],
|
||||
)
|
||||
def test_other_responses_statuses_do_not_become_length_caps(metadata):
|
||||
message = AIMessage(content="Ordinary answer.", response_metadata=metadata)
|
||||
context = {}
|
||||
assert ModelLengthFinishReasonMiddleware().after_model({"messages": [message]}, SimpleNamespace(context=context)) is None
|
||||
assert "stop_reason" not in context
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("streaming", [False, True])
|
||||
async def test_async_responses_length_signal_stops_tool_calls(async_responses_provider, streaming):
|
||||
model, requests, _ = async_responses_provider
|
||||
if streaming:
|
||||
collected = None
|
||||
async for chunk in model.astream([HumanMessage("Save the result.")]):
|
||||
collected = chunk if collected is None else collected + chunk
|
||||
assert collected is not None
|
||||
message = message_chunk_to_message(collected)
|
||||
else:
|
||||
message = await model.ainvoke([HumanMessage("Save the result.")])
|
||||
assert message.tool_calls
|
||||
assert message.response_metadata["status"] == "incomplete"
|
||||
context = {}
|
||||
update = await ModelLengthFinishReasonMiddleware().aafter_model({"messages": [message]}, SimpleNamespace(context=context))
|
||||
assert update is not None
|
||||
assert update["messages"][0].tool_calls == []
|
||||
assert context["stop_reason"] == MODEL_LENGTH_CAPPED_STOP_REASON
|
||||
assert len(requests) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("streaming", [False, True])
|
||||
async def test_async_capped_agent_does_not_write_and_resumes_from_checkpoint(async_responses_provider, streaming):
|
||||
model, requests, _ = async_responses_provider
|
||||
model.streaming = streaming
|
||||
writes = []
|
||||
|
||||
@tool
|
||||
async def record_write() -> str:
|
||||
"""Record one observable write."""
|
||||
writes.append("written")
|
||||
return "Saved."
|
||||
|
||||
graph = create_agent(model, [record_write], middleware=[ModelLengthFinishReasonMiddleware()], checkpointer=InMemorySaver())
|
||||
context = {"thread_id": "async-length-regression", "run_id": "async-length-run"}
|
||||
config = {"configurable": {"thread_id": "async-length-regression"}}
|
||||
result = await graph.ainvoke({"messages": [HumanMessage("Save the result.")]}, config=config, context=context)
|
||||
assert writes == []
|
||||
assert context["stop_reason"] == MODEL_LENGTH_CAPPED_STOP_REASON
|
||||
assert result["messages"][-1].additional_kwargs["model_length_termination"]["suppressed_tool_call_names"] == ["record_write"]
|
||||
assert len(requests) == 1
|
||||
continued = await graph.ainvoke({"messages": [HumanMessage("Continue.")]}, config=config, context={})
|
||||
assert continued["messages"][-1].text == "Finished."
|
||||
assert len(requests) == 2
|
||||
assert not any(item.get("type") in ("function_call", "function_call_output") for item in requests[-1]["input"])
|
||||
assert writes == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("streaming", [False, True])
|
||||
def test_completed_responses_agent_executes_tool_normally(responses_provider, streaming):
|
||||
model, requests, state = responses_provider
|
||||
state.status, state.reason = "completed", None
|
||||
model.streaming = streaming
|
||||
writes = []
|
||||
|
||||
@tool
|
||||
def record_write() -> str:
|
||||
"""Record one observable write."""
|
||||
writes.append("written")
|
||||
return "Saved."
|
||||
|
||||
context = {}
|
||||
graph = create_agent(model, [record_write], middleware=[ModelLengthFinishReasonMiddleware()])
|
||||
result = graph.invoke({"messages": [HumanMessage("Save the result.")]}, context=context)
|
||||
assert writes == ["written"]
|
||||
assert result["messages"][-1].text == "Finished."
|
||||
assert len(requests) == 2
|
||||
assert "stop_reason" not in context
|
||||
assert any(item.get("type") == "function_call_output" for item in requests[-1]["input"])
|
||||
|
||||
|
||||
@pytest.mark.parametrize("container", ["response_metadata", "additional_kwargs"])
|
||||
@pytest.mark.parametrize("content", ["Partial answer.", [{"type": "reasoning", "id": "rs-capped", "summary": []}]])
|
||||
def test_native_length_cap_preserves_content_metadata_and_earlier_stop_reason(container, content):
|
||||
metadata = {"status": "incomplete", "incomplete_details": {"reason": "max_output_tokens"}}
|
||||
message = AIMessage(content=content, **{container: metadata})
|
||||
original = message.model_dump()
|
||||
context = {"stop_reason": "token_capped"}
|
||||
middleware = ModelLengthFinishReasonMiddleware()
|
||||
update = middleware.after_model({"messages": [message]}, SimpleNamespace(context=context))
|
||||
assert context["stop_reason"] == "token_capped"
|
||||
if isinstance(content, str):
|
||||
assert update is None
|
||||
else:
|
||||
assert update is not None
|
||||
replacement = update["messages"][0]
|
||||
assert replacement.content[0] == content[0]
|
||||
assert "output limit" in replacement.content[-1]["text"]
|
||||
assert getattr(replacement, container)["incomplete_details"] == metadata["incomplete_details"]
|
||||
assert middleware.after_model({"messages": [replacement]}, SimpleNamespace(context=context)) is None
|
||||
assert message.model_dump() == original
|
||||
fresh_context = {}
|
||||
middleware.after_model({"messages": [message]}, SimpleNamespace(context=fresh_context))
|
||||
assert fresh_context["stop_reason"] == MODEL_LENGTH_CAPPED_STOP_REASON
|
||||
|
||||
|
||||
@pytest.mark.parametrize("swap", [False, True])
|
||||
def test_responses_detector_does_not_combine_unrelated_metadata_containers(swap):
|
||||
status = {"status": "incomplete"}
|
||||
details = {"incomplete_details": {"reason": "max_output_tokens"}}
|
||||
message = AIMessage(content="Ordinary answer.", response_metadata=details if swap else status, additional_kwargs=status if swap else details)
|
||||
context = {}
|
||||
assert ModelLengthFinishReasonMiddleware().after_model({"messages": [message]}, SimpleNamespace(context=context)) is None
|
||||
assert "stop_reason" not in context
|
||||
|
||||
|
||||
@pytest.mark.parametrize("block_type", ["function_call", "custom_tool_call", "tool_call", "tool_call_chunk", "tool_use"])
|
||||
@pytest.mark.parametrize("surface", ["content_only", "invalid", "raw"])
|
||||
def test_length_cleanup_clears_every_tool_surface_and_preserves_non_call_blocks(block_type, surface):
|
||||
call = {"type": block_type, "id": "call-capped", "call_id": "call-capped", "name": "record_write", "arguments": "{}", "input": {}}
|
||||
kept = [{"type": "reasoning", "id": "rs-capped", "summary": []}, {"type": "text", "text": "Partial answer."}, {"type": [], "text": "Malformed type."}]
|
||||
message = AIMessage(
|
||||
id="msg-capped",
|
||||
content=[*kept, call],
|
||||
response_metadata={"status": "incomplete", "incomplete_details": {"reason": "max_output_tokens"}, "model_provider": "openai"},
|
||||
additional_kwargs={"kept": "metadata"},
|
||||
usage_metadata={"input_tokens": 10, "output_tokens": 8, "total_tokens": 18},
|
||||
)
|
||||
if surface == "invalid":
|
||||
message.invalid_tool_calls = [{"type": "invalid_tool_call", "id": "call-capped", "name": "record_write", "args": "{", "error": "Incomplete JSON"}]
|
||||
elif surface == "raw":
|
||||
message = message.model_copy(
|
||||
update={"additional_kwargs": {"kept": "metadata", "tool_calls": [{"id": "call-capped", "function": {"name": "record_write", "arguments": "{"}}], "function_call": {"name": "record_write", "arguments": "{"}}}
|
||||
)
|
||||
original = message.model_dump()
|
||||
middleware = ModelLengthFinishReasonMiddleware()
|
||||
update = middleware.after_model({"messages": [message]}, SimpleNamespace(context={}))
|
||||
assert update is not None
|
||||
replacement = update["messages"][0]
|
||||
assert replacement.content[:-1] == kept
|
||||
assert "output limit" in replacement.content[-1]["text"]
|
||||
assert replacement.tool_calls == replacement.invalid_tool_calls == []
|
||||
assert "tool_calls" not in replacement.additional_kwargs
|
||||
assert "function_call" not in replacement.additional_kwargs
|
||||
assert replacement.additional_kwargs["kept"] == "metadata"
|
||||
assert replacement.additional_kwargs["model_length_termination"]["suppressed_tool_call_names"] == ["record_write"]
|
||||
assert replacement.response_metadata == message.response_metadata
|
||||
assert replacement.usage_metadata == message.usage_metadata
|
||||
assert replacement.id == message.id
|
||||
assert message.model_dump() == original
|
||||
assert middleware.after_model({"messages": [replacement]}, SimpleNamespace(context={})) is None
|
||||
completed = message.model_copy(update={"response_metadata": {"status": "completed"}})
|
||||
assert middleware.after_model({"messages": [completed]}, SimpleNamespace(context={})) is None
|
||||
@ -52,11 +52,12 @@ def test_anthropic_max_tokens_detector_matches_stop_reason():
|
||||
assert hit.reason_value == "max_tokens"
|
||||
|
||||
|
||||
def test_default_detectors_cover_openai_anthropic_and_gemini():
|
||||
def test_default_detectors_cover_openai_anthropic_gemini_and_responses():
|
||||
names = [detector.name for detector in default_detectors()]
|
||||
|
||||
assert names == [
|
||||
"openai_compatible_length",
|
||||
"anthropic_max_tokens",
|
||||
"gemini_max_tokens",
|
||||
"openai_responses_length",
|
||||
]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user