From 5a90f39920c0b025fdb9bfbbd378b4152b12c837 Mon Sep 17 00:00:00 2001 From: qinxuye Date: Sun, 12 Jul 2026 01:06:23 +0800 Subject: [PATCH] fix: keep ReAct tool calls structured --- src/xagent/core/agent/pattern/react/react.py | 201 ++++++- src/xagent/core/agent/runtime.py | 27 + src/xagent/core/model/chat/basic/deepseek.py | 12 +- .../chat/basic/deepseek_tool_protocol.py | 343 ++++++++++++ .../core/model/chat/basic/openrouter.py | 68 ++- src/xagent/core/model/chat/tool_protocol.py | 39 ++ src/xagent/core/model/chat/types.py | 6 + tests/core/agent/test_dag.py | 25 +- tests/core/agent/test_execution_adapter.py | 6 +- tests/core/agent/test_react.py | 491 +++++++++++++++++- tests/core/model/chat/basic/test_deepseek.py | 37 ++ .../chat/basic/test_deepseek_tool_protocol.py | 424 +++++++++++++++ .../core/model/chat/basic/test_openrouter.py | 60 +++ 13 files changed, 1708 insertions(+), 31 deletions(-) create mode 100644 src/xagent/core/model/chat/basic/deepseek_tool_protocol.py create mode 100644 src/xagent/core/model/chat/tool_protocol.py create mode 100644 tests/core/model/chat/basic/test_deepseek_tool_protocol.py diff --git a/src/xagent/core/agent/pattern/react/react.py b/src/xagent/core/agent/pattern/react/react.py index 9ceb0bf53..8d2eafaea 100644 --- a/src/xagent/core/agent/pattern/react/react.py +++ b/src/xagent/core/agent/pattern/react/react.py @@ -38,6 +38,7 @@ from enum import Enum from typing import Any, cast +from ....model.chat.tool_protocol import get_tool_protocol_error from ...context.enrichment import ( enrich_context_with_memory, enrich_context_with_skill, @@ -48,9 +49,7 @@ from ...result import unwrap_final_answer_content from ...runtime import LLMCallInterrupted, PatternRuntime from ..base import AgentPattern, PatternResult, truncate_prompt_preview -from ..final_answer_stream import ( - ReActFinalAnswerStreamer, -) +from ..final_answer_stream import ReActFinalAnswerStreamer class ReActReasoningMode(str, Enum): @@ -352,7 +351,11 @@ async def _run_tool_calling_loop( and not self.pending_tool_calls and self._latest_tool_result_success(context) ) - tool_schemas = [] if force_final_answer_now else base_tool_schemas + tool_schemas = ( + [self._final_answer_tool_schema()] + if force_final_answer_now + else base_tool_schemas + ) interrupted = await self._interrupt_if_requested( runtime=runtime, context=context, @@ -382,10 +385,17 @@ async def _run_tool_calling_loop( ) answer_streamer: ReActFinalAnswerStreamer | None = None try: + effective_tool_choice = ( + "required" + if force_final_answer_now + else self.tool_choice + if tool_schemas + else None + ) llm_kwargs = { "messages": messages, "tools": tool_schemas or None, - "tool_choice": self.tool_choice if tool_schemas else None, + "tool_choice": effective_tool_choice, } if tool_schemas: answer_streamer = ReActFinalAnswerStreamer(runtime) @@ -411,14 +421,73 @@ async def _run_tool_calling_loop( if answer_streamer is not None: await answer_streamer.fail(str(exc)) raise + self.repeated_tool_decision = None + self.last_response = response + normalized = self._normalize_llm_response(response) + requires_protocol_retry = self._response_requires_tool_protocol_retry( + normalized, + force_final_answer=force_final_answer_now, + ) + end_metadata: dict[str, Any] = {"iteration": iteration} + if requires_protocol_retry: + end_metadata.update( + success=False, + phase="discarded_invalid_tool_protocol", + ) await runtime.on_llm_end( context=context, response=response, - metadata={"iteration": iteration}, + metadata=end_metadata, ) - self.repeated_tool_decision = None - self.last_response = response - normalized = self._normalize_llm_response(response) + if requires_protocol_retry: + if answer_streamer is not None: + await answer_streamer.fail("invalid tool protocol, retrying") + try: + ( + response, + answer_streamer, + ) = await self._retry_tool_protocol_response( + context=context, + llm=llm, + runtime=runtime, + iteration=iteration, + tool_schemas=tool_schemas, + force_final_answer=force_final_answer_now, + ) + except LLMCallInterrupted: + interrupted = await self._interrupt_if_requested( + runtime=runtime, + context=context, + label="during_llm", + ) + if interrupted is not None: + return interrupted + raise + self.last_response = response + normalized = self._normalize_llm_response(response) + if self._response_requires_tool_protocol_retry( + normalized, + force_final_answer=force_final_answer_now, + ): + if answer_streamer is not None: + await answer_streamer.fail("invalid tool protocol after retry") + await runtime.checkpoint( + "invalid_tool_protocol", + context=context, + pattern=self, + metadata={"iteration": iteration}, + ) + return PatternResult( + success=False, + error=( + "The model returned an invalid tool protocol response " + "after one repair attempt." + ), + metadata={ + "iterations": iteration + 1, + "status": "invalid_tool_protocol", + }, + ).to_dict() if force_final_answer_now and not normalized.get("tool_calls"): normalized["done"] = True @@ -516,9 +585,10 @@ def _messages_for_llm( messages = list(context.get_messages_for_llm()) if force_final_answer: instruction = ( - "You have already received the tool result needed for the current " - "step. Do not call tools again. Produce the final answer for this " - "step using the latest tool result. " + "Produce the final user-facing answer by calling the final_answer " + "control tool exactly once using the accumulated conversation and " + "tool results. Do not call any other tool and do not output " + "tool-call markup as plain text. " f"{final_answer_language_rule()}" ) elif has_tools: @@ -578,6 +648,112 @@ def _messages_for_llm( ] return [{"role": "system", "content": instruction}, *messages] + async def _retry_tool_protocol_response( + self, + *, + context: Any, + llm: Any, + runtime: PatternRuntime, + iteration: int, + tool_schemas: list[dict[str, Any]], + force_final_answer: bool, + ) -> tuple[Any, ReActFinalAnswerStreamer]: + tools = ( + [self._final_answer_tool_schema()] if force_final_answer else tool_schemas + ) + messages = self._messages_for_llm( + context, + has_tools=True, + force_final_answer=force_final_answer, + tool_names=self._schema_tool_names(tools), + ) + retry_instruction = ( + "The previous response used an invalid tool protocol. Retry the same " + "turn using native structured tool calls only. Never place one tool " + "invocation or its arguments inside another tool's arguments. If work " + "remains, call the appropriate available work tool directly; call " + "final_answer only when the task is actually complete." + ) + messages[0] = { + **messages[0], + "content": f"{messages[0].get('content', '')}\n\n{retry_instruction}", + } + metadata = { + "iteration": iteration, + "phase": "tool_protocol_retry", + } + await runtime.checkpoint( + "tool_protocol_retry", + context=context, + pattern=self, + metadata=metadata, + ) + await runtime.on_llm_start( + context=context, + messages=messages, + tools=tools, + metadata=metadata, + ) + answer_streamer = ReActFinalAnswerStreamer(runtime) + try: + response = await runtime.run_streaming_llm_call( + llm, + messages=messages, + tools=tools, + tool_choice="required", + on_chunk=answer_streamer.handle_chunk, + ) + except LLMCallInterrupted: + await answer_streamer.fail("interrupted during LLM stream") + raise + except Exception as exc: + await answer_streamer.fail(str(exc)) + await runtime.on_llm_error( + context=context, + error=exc, + metadata=metadata, + ) + raise + normalized = self._normalize_llm_response(response) + retry_is_invalid = self._response_requires_tool_protocol_retry( + normalized, + force_final_answer=force_final_answer, + ) + end_metadata = dict(metadata) + if retry_is_invalid: + end_metadata.update( + success=False, + phase="discarded_invalid_tool_protocol_retry", + ) + await runtime.on_llm_end( + context=context, + response=response, + metadata=end_metadata, + ) + return response, answer_streamer + + def _response_requires_tool_protocol_retry( + self, + normalized: dict[str, Any], + *, + force_final_answer: bool, + ) -> bool: + if get_tool_protocol_error(normalized.get("raw")) is not None: + return True + for tool_call in normalized.get("tool_calls") or []: + if not isinstance(tool_call, dict): + continue + if force_final_answer and tool_call.get("name") != "final_answer": + return True + return False + + def _final_answer_tool_schema(self) -> dict[str, Any]: + for schema in self._builtin_tool_schemas(): + function = schema.get("function") + if isinstance(function, dict) and function.get("name") == "final_answer": + return schema + raise RuntimeError("final_answer control tool schema is unavailable") + def _schema_tool_names(self, tool_schemas: list[dict[str, Any]]) -> list[str]: names: list[str] = [] for schema in tool_schemas: @@ -936,6 +1112,7 @@ def _builtin_tool_schemas(self) -> list[dict[str, Any]]: ), "parameters": { "type": "object", + "additionalProperties": False, "properties": { "response_language": { "type": "string", diff --git a/src/xagent/core/agent/runtime.py b/src/xagent/core/agent/runtime.py index e0a4589e8..ffebc3090 100644 --- a/src/xagent/core/agent/runtime.py +++ b/src/xagent/core/agent/runtime.py @@ -15,6 +15,7 @@ normalize_llm_trace_payload, ) from ..model.chat.basic.base import BaseLLM +from ..model.chat.tool_protocol import TOOL_PROTOCOL_ERROR_KEY from ..model.chat.types import ChunkType from .streaming import merge_streamed_tool_call_arguments @@ -147,10 +148,26 @@ async def consume_stream() -> Any: tool_call_chunks: dict[int, dict[str, Any]] = {} usage_payload: dict[str, Any] = {} provider_payload: dict[str, Any] = {} + protocol_error_payload: dict[str, Any] = {} saw_payload_chunk = False async for chunk in stream_chat(**kwargs): await self._raise_if_interrupted("interrupted during LLM stream") self._raise_for_stream_error(chunk) + is_protocol_error = ( + callable(getattr(chunk, "is_protocol_error", None)) + and chunk.is_protocol_error() + ) + if ( + getattr(chunk, "type", None) == ChunkType.PROTOCOL_ERROR + or is_protocol_error + ): + payload = getattr(chunk, "protocol_error", None) + if isinstance(payload, dict): + protocol_error_payload.update(payload) + saw_payload_chunk = True + if on_chunk is not None: + await self._maybe_await(on_chunk(chunk)) + continue text_delta = self._chunk_text_delta(chunk) if text_delta: saw_payload_chunk = True @@ -170,6 +187,16 @@ async def consume_stream() -> Any: tool_calls = [ tool_call_chunks[index] for index in sorted(tool_call_chunks.keys()) ] + if protocol_error_payload: + protocol_response = { + "type": "tool_protocol_error", + "content": "", + "tool_calls": [], + TOOL_PROTOCOL_ERROR_KEY: protocol_error_payload, + } + if usage_payload: + protocol_response["usage"] = usage_payload + return protocol_response if tool_calls: response: dict[str, Any] = { "content": content, diff --git a/src/xagent/core/model/chat/basic/deepseek.py b/src/xagent/core/model/chat/basic/deepseek.py index ea3fcaffb..d52ad6604 100644 --- a/src/xagent/core/model/chat/basic/deepseek.py +++ b/src/xagent/core/model/chat/basic/deepseek.py @@ -4,6 +4,10 @@ from ...providers import is_placeholder_api_key from .base import StreamChunk +from .deepseek_tool_protocol import ( + adapt_deepseek_stream, + normalize_deepseek_response, +) from .openai import PROVIDER_STATE_METADATA_KEY, OpenAICompatibleLLM logger = logging.getLogger(__name__) @@ -236,7 +240,7 @@ async def chat( output_config=output_config, ) kwargs = self._prepare_deepseek_kwargs(kwargs=kwargs) - return await super().chat( + response = await super().chat( messages=messages, temperature=temperature, max_tokens=max_tokens, @@ -247,6 +251,7 @@ async def chat( output_config=output_config, **kwargs, ) + return normalize_deepseek_response(response, tools=tools) async def stream_chat( self, @@ -265,7 +270,7 @@ async def stream_chat( output_config=output_config, ) kwargs = self._prepare_deepseek_kwargs(kwargs=kwargs) - async for chunk in super().stream_chat( + stream = super().stream_chat( messages=messages, temperature=temperature, max_tokens=max_tokens, @@ -275,7 +280,8 @@ async def stream_chat( thinking=thinking, output_config=output_config, **kwargs, - ): + ) + async for chunk in adapt_deepseek_stream(stream, tools=tools): yield chunk @staticmethod diff --git a/src/xagent/core/model/chat/basic/deepseek_tool_protocol.py b/src/xagent/core/model/chat/basic/deepseek_tool_protocol.py new file mode 100644 index 000000000..38dc1a37e --- /dev/null +++ b/src/xagent/core/model/chat/basic/deepseek_tool_protocol.py @@ -0,0 +1,343 @@ +from __future__ import annotations + +import json +import re +from collections.abc import AsyncIterator +from dataclasses import replace +from typing import Any + +from ..tool_protocol import ( + ToolProtocolViolation, + tool_protocol_error_response, +) +from ..types import ChunkType, StreamChunk + +_PROVIDER = "deepseek" +_SERIALIZED_TOOL_CALL_RE = re.compile( + r"<[^>\n]*dsml[^>\n]*tool_calls", + re.IGNORECASE, +) +_PARTIAL_MARKER_TARGET = "dsmltool_calls" +_PARTIAL_MARKER_SCAN_LIMIT = 64 +_MARKER_SEPARATOR_RE = re.compile(r"[\s|｜]") + + +def normalize_deepseek_response( + response: Any, + *, + tools: list[dict[str, Any]] | None, +) -> Any: + if not tools: + return response + violation = _response_violation(response, tools=tools) + if violation is None: + return response + raw = response.get("raw") if isinstance(response, dict) else None + return tool_protocol_error_response(violation, raw=raw) + + +async def adapt_deepseek_stream( + stream: AsyncIterator[StreamChunk], + *, + tools: list[dict[str, Any]] | None, +) -> AsyncIterator[StreamChunk]: + if not tools: + async for chunk in stream: + yield chunk + return + + text = "" + emitted_text_length = 0 + buffered_tool_chunk: StreamChunk | None = None + terminal_chunk: StreamChunk | None = None + usage_chunks: list[StreamChunk] = [] + violation: ToolProtocolViolation | None = None + withheld_tool_tail = False + + async for chunk in stream: + if chunk.is_token(): + delta = str(chunk.delta or chunk.content or "") + text += delta + if violation is None: + violation = _serialized_content_violation(text) + if violation is not None: + continue + safe_length = _safe_streaming_text_length(text) + if safe_length > emitted_text_length: + safe_delta = text[emitted_text_length:safe_length] + emitted_text_length = safe_length + yield StreamChunk( + type=ChunkType.TOKEN, + content=safe_delta, + delta=safe_delta, + raw=chunk.raw, + ) + continue + + if chunk.is_tool_call(): + buffered_tool_chunk = chunk + if violation is None: + violation = _streaming_tool_call_violation(chunk) + if violation is None: + safe_chunk, withheld_tool_tail = _safe_streaming_tool_chunk(chunk) + yield safe_chunk + continue + if chunk.is_usage(): + usage_chunks.append(chunk) + continue + if chunk.is_end(): + terminal_chunk = chunk + continue + yield chunk + + if violation is None and buffered_tool_chunk is not None: + violation = _response_violation( + { + "content": text, + "tool_calls": buffered_tool_chunk.tool_calls, + }, + tools=tools, + ) + + if violation is not None: + raw = ( + buffered_tool_chunk.raw + if buffered_tool_chunk is not None + else terminal_chunk.raw + if terminal_chunk is not None + else None + ) + yield StreamChunk( + type=ChunkType.PROTOCOL_ERROR, + protocol_error=violation.to_dict(), + raw=raw, + ) + else: + if emitted_text_length < len(text): + delta = text[emitted_text_length:] + yield StreamChunk( + type=ChunkType.TOKEN, + content=delta, + delta=delta, + raw=terminal_chunk.raw if terminal_chunk is not None else None, + ) + if buffered_tool_chunk is not None and withheld_tool_tail: + yield buffered_tool_chunk + elif terminal_chunk is not None: + yield terminal_chunk + + for chunk in usage_chunks: + yield chunk + + +def _response_violation( + response: Any, + *, + tools: list[dict[str, Any]] | None, +) -> ToolProtocolViolation | None: + content = _response_content(response) + violation = _serialized_content_violation(content) + if violation is not None: + return violation + + for tool_call in _response_tool_calls(response): + violation = _tool_call_violation(tool_call, tools=tools) + if violation is not None: + return violation + return None + + +def _serialized_content_violation(content: Any) -> ToolProtocolViolation | None: + if _serialized_tool_call_start(content) is None: + return None + return ToolProtocolViolation( + provider=_PROVIDER, + code="serialized_tool_call_content", + message="DeepSeek returned serialized tool-call markup in assistant content.", + ) + + +def _streaming_tool_call_violation( + chunk: StreamChunk, +) -> ToolProtocolViolation | None: + for tool_call in chunk.tool_calls: + function = _function_payload(tool_call) + name = function.get("name") + arguments = function.get("arguments") + if _contains_serialized_tool_call(arguments): + return ToolProtocolViolation( + provider=_PROVIDER, + code="nested_serialized_tool_call", + message=( + "DeepSeek embedded serialized tool-call markup inside " + f"{name or 'a tool call'!r}." + ), + ) + return None + + +def _safe_streaming_tool_chunk( + chunk: StreamChunk, +) -> tuple[StreamChunk, bool]: + safe_tool_calls: list[dict[str, Any]] = [] + withheld = False + for tool_call in chunk.tool_calls: + if not isinstance(tool_call, dict): + safe_tool_calls.append(tool_call) + continue + safe_tool_call = dict(tool_call) + function = tool_call.get("function") + if isinstance(function, dict): + safe_function = dict(function) + arguments = function.get("arguments") + if isinstance(arguments, str): + safe_length = _safe_streaming_text_length(arguments) + if safe_length < len(arguments): + safe_function["arguments"] = arguments[:safe_length] + withheld = True + safe_tool_call["function"] = safe_function + safe_tool_calls.append(safe_tool_call) + return replace(chunk, tool_calls=safe_tool_calls), withheld + + +def _tool_call_violation( + tool_call: Any, + *, + tools: list[dict[str, Any]] | None, +) -> ToolProtocolViolation | None: + function = _function_payload(tool_call) + name = function.get("name") + if not isinstance(name, str) or not name: + return ToolProtocolViolation( + provider=_PROVIDER, + code="malformed_tool_call", + message="DeepSeek returned a tool call without a function name.", + ) + + arguments = function.get("arguments", {}) + if isinstance(arguments, str): + try: + arguments = json.loads(arguments) + except json.JSONDecodeError: + return ToolProtocolViolation( + provider=_PROVIDER, + code="malformed_tool_arguments", + message=f"DeepSeek returned malformed arguments for {name!r}.", + ) + if not isinstance(arguments, dict): + return ToolProtocolViolation( + provider=_PROVIDER, + code="malformed_tool_arguments", + message=f"DeepSeek returned non-object arguments for {name!r}.", + ) + + if _contains_serialized_tool_call(arguments): + return ToolProtocolViolation( + provider=_PROVIDER, + code="nested_serialized_tool_call", + message=f"DeepSeek embedded serialized tool-call markup inside {name!r}.", + ) + + schemas = _tool_schema_by_name(tools) + if not schemas: + return None + schema = schemas.get(name) + if schema is None: + return ToolProtocolViolation( + provider=_PROVIDER, + code="unavailable_tool_call", + message=f"DeepSeek returned unavailable tool call {name!r}.", + ) + + parameters = schema.get("parameters") + if isinstance(parameters, dict) and parameters.get("additionalProperties") is False: + properties = parameters.get("properties") + allowed = set(properties) if isinstance(properties, dict) else set() + unexpected = set(arguments) - allowed + if unexpected: + return ToolProtocolViolation( + provider=_PROVIDER, + code="unexpected_tool_arguments", + message=( + f"DeepSeek returned unexpected arguments for {name!r}: " + f"{', '.join(sorted(unexpected))}." + ), + ) + return None + + +def _contains_serialized_tool_call(value: Any) -> bool: + if isinstance(value, str): + return _serialized_tool_call_start(value) is not None + if isinstance(value, dict): + return any(_contains_serialized_tool_call(item) for item in value.values()) + if isinstance(value, list): + return any(_contains_serialized_tool_call(item) for item in value) + return False + + +def _serialized_tool_call_start(content: Any) -> int | None: + if not isinstance(content, str): + return None + match = _SERIALIZED_TOOL_CALL_RE.search(content) + return match.start() if match is not None else None + + +def _safe_streaming_text_length(content: str) -> int: + serialized_start = _serialized_tool_call_start(content) + if serialized_start is not None: + return serialized_start + + for match in re.finditer("<", content): + tail_start = match.start() + 1 + tail = content[tail_start : tail_start + _PARTIAL_MARKER_SCAN_LIMIT] + if "\n" in tail or ">" in tail: + continue + normalized_tail = _MARKER_SEPARATOR_RE.sub("", tail).casefold() + if _PARTIAL_MARKER_TARGET.startswith(normalized_tail): + return match.start() + return len(content) + + +def _response_content(response: Any) -> Any: + if isinstance(response, str): + return response + if isinstance(response, dict): + return response.get("content") + return None + + +def _response_tool_calls(response: Any) -> list[Any]: + if isinstance(response, dict): + return list(response.get("tool_calls") or []) + return [] + + +def _function_payload(tool_call: Any) -> dict[str, Any]: + if isinstance(tool_call, dict): + function = tool_call.get("function") + if isinstance(function, dict): + return function + return { + "name": tool_call.get("name"), + "arguments": tool_call.get("args", tool_call.get("arguments", {})), + } + function = getattr(tool_call, "function", None) + return { + "name": getattr(function, "name", None), + "arguments": getattr(function, "arguments", {}), + } + + +def _tool_schema_by_name( + tools: list[dict[str, Any]] | None, +) -> dict[str, dict[str, Any]]: + schemas: dict[str, dict[str, Any]] = {} + for tool in tools or []: + function = tool.get("function") + if not isinstance(function, dict): + continue + name = function.get("name") + if isinstance(name, str) and name: + schemas[name] = function + return schemas diff --git a/src/xagent/core/model/chat/basic/openrouter.py b/src/xagent/core/model/chat/basic/openrouter.py index 5df554b06..db6fd7fc1 100644 --- a/src/xagent/core/model/chat/basic/openrouter.py +++ b/src/xagent/core/model/chat/basic/openrouter.py @@ -1,7 +1,12 @@ -from typing import Any, Dict, List, Optional +from typing import Any, AsyncIterator, Dict, List, Optional from .....config import get_openrouter_official_providers_only from ..timeout_config import TimeoutConfig +from ..types import StreamChunk +from .deepseek_tool_protocol import ( + adapt_deepseek_stream, + normalize_deepseek_response, +) from .openai import OpenAILLM OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1" @@ -51,6 +56,67 @@ def __init__( timeout_config=timeout_config, ) + @property + def _uses_deepseek_tool_protocol(self) -> bool: + return _openrouter_model_author(self._model_name) == "deepseek" + + async def chat( + self, + messages: List[Dict[str, str]], + temperature: Optional[float] = None, + max_tokens: Optional[int] = None, + tools: Optional[List[Dict[str, Any]]] = None, + tool_choice: Optional[str | Dict[str, Any]] = None, + response_format: Optional[Dict[str, Any]] = None, + thinking: Optional[Dict[str, Any]] = None, + output_config: Optional[Dict[str, Any]] = None, + **kwargs: Any, + ) -> Any: + response = await super().chat( + messages=messages, + temperature=temperature, + max_tokens=max_tokens, + tools=tools, + tool_choice=tool_choice, + response_format=response_format, + thinking=thinking, + output_config=output_config, + **kwargs, + ) + if not self._uses_deepseek_tool_protocol: + return response + return normalize_deepseek_response(response, tools=tools) + + async def stream_chat( + self, + messages: List[Dict[str, str]], + temperature: Optional[float] = None, + max_tokens: Optional[int] = None, + tools: Optional[List[Dict[str, Any]]] = None, + tool_choice: Optional[str | Dict[str, Any]] = None, + response_format: Optional[Dict[str, Any]] = None, + thinking: Optional[Dict[str, Any]] = None, + output_config: Optional[Dict[str, Any]] = None, + **kwargs: Any, + ) -> AsyncIterator[StreamChunk]: + stream = super().stream_chat( + messages=messages, + temperature=temperature, + max_tokens=max_tokens, + tools=tools, + tool_choice=tool_choice, + response_format=response_format, + thinking=thinking, + output_config=output_config, + **kwargs, + ) + if not self._uses_deepseek_tool_protocol: + async for chunk in stream: + yield chunk + return + async for chunk in adapt_deepseek_stream(stream, tools=tools): + yield chunk + def _is_official_openrouter_client(self) -> bool: return self.base_url.rstrip("/") == OPENROUTER_BASE_URL diff --git a/src/xagent/core/model/chat/tool_protocol.py b/src/xagent/core/model/chat/tool_protocol.py new file mode 100644 index 000000000..091c4d2ac --- /dev/null +++ b/src/xagent/core/model/chat/tool_protocol.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +from dataclasses import asdict, dataclass +from typing import Any + +TOOL_PROTOCOL_ERROR_KEY = "_xagent_tool_protocol_error" + + +@dataclass(frozen=True) +class ToolProtocolViolation: + provider: str + code: str + message: str + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +def tool_protocol_error_response( + violation: ToolProtocolViolation, + *, + raw: Any = None, +) -> dict[str, Any]: + response: dict[str, Any] = { + "type": "tool_protocol_error", + "content": "", + "tool_calls": [], + TOOL_PROTOCOL_ERROR_KEY: violation.to_dict(), + } + if raw is not None: + response["raw"] = raw + return response + + +def get_tool_protocol_error(response: Any) -> dict[str, Any] | None: + if not isinstance(response, dict): + return None + payload = response.get(TOOL_PROTOCOL_ERROR_KEY) + return dict(payload) if isinstance(payload, dict) else None diff --git a/src/xagent/core/model/chat/types.py b/src/xagent/core/model/chat/types.py index f86190cd7..e50eb1f1f 100644 --- a/src/xagent/core/model/chat/types.py +++ b/src/xagent/core/model/chat/types.py @@ -12,6 +12,7 @@ class ChunkType(Enum): TOOL_CALL = "tool_call" # Tool call USAGE = "usage" # Token usage statistics ERROR = "error" # Error + PROTOCOL_ERROR = "protocol_error" # Provider tool-protocol violation END = "end" # End @@ -36,6 +37,7 @@ class StreamChunk: usage: Dict[str, int] = field(default_factory=dict) finish_reason: str = "" raw: Any = None + protocol_error: Dict[str, Any] = field(default_factory=dict) def is_token(self) -> bool: """Check if this is a token type""" @@ -53,6 +55,10 @@ def is_error(self) -> bool: """Check if this is an error type""" return self.type == ChunkType.ERROR + def is_protocol_error(self) -> bool: + """Check if this is a provider tool-protocol error.""" + return self.type == ChunkType.PROTOCOL_ERROR + def is_end(self) -> bool: """Check if this is an end type""" return self.type == ChunkType.END diff --git a/tests/core/agent/test_dag.py b/tests/core/agent/test_dag.py index 0bd65670c..8712ba142 100644 --- a/tests/core/agent/test_dag.py +++ b/tests/core/agent/test_dag.py @@ -544,8 +544,24 @@ async def chat(self, **kwargs: Any) -> dict[str, Any]: } ], } - if not kwargs.get("tools"): - return "DAG child answer." + if len(kwargs.get("tools") or []) == 1 and has_tool(kwargs, "final_answer"): + return { + "content": "", + "tool_calls": [ + { + "id": "final_1", + "function": { + "name": "final_answer", + "arguments": json.dumps( + { + "response_language": "English", + "answer": "DAG child answer.", + } + ), + }, + } + ], + } self.tool_call_count += 1 if self.tool_call_count > 4: @@ -596,7 +612,10 @@ async def chat(self, **kwargs: Any) -> dict[str, Any]: assert [schema["function"]["name"] for schema in llm.calls[4]["tools"]] == [ "react_decision" ] - assert llm.calls[5]["tools"] is None + assert [schema["function"]["name"] for schema in llm.calls[5]["tools"]] == [ + "final_answer" + ] + assert llm.calls[5]["tool_choice"] == "required" assert [event["type"] for event in outbound.events] == [ "final_answer_start", "final_answer_delta", diff --git a/tests/core/agent/test_execution_adapter.py b/tests/core/agent/test_execution_adapter.py index 192cf81c0..7434d6f0d 100644 --- a/tests/core/agent/test_execution_adapter.py +++ b/tests/core/agent/test_execution_adapter.py @@ -242,8 +242,10 @@ async def test_execution_adapter_routes_single_call_to_one_tool_then_final_answe assert tool.calls == [{"value": "from tool"}] assert llm.calls[0]["tools"][0]["function"]["name"] == "noop" assert llm.calls[0]["tool_choice"] == "required" - assert llm.calls[1]["tools"] is None - assert llm.calls[1]["tool_choice"] is None + assert [schema["function"]["name"] for schema in llm.calls[1]["tools"]] == [ + "final_answer" + ] + assert llm.calls[1]["tool_choice"] == "required" @pytest.mark.asyncio diff --git a/tests/core/agent/test_react.py b/tests/core/agent/test_react.py index c7b7fd807..d83cc18ec 100644 --- a/tests/core/agent/test_react.py +++ b/tests/core/agent/test_react.py @@ -17,6 +17,10 @@ ReActReasoningMode, ToolCallRecord, ) +from xagent.core.model.chat.tool_protocol import ( + ToolProtocolViolation, + tool_protocol_error_response, +) from xagent.core.model.chat.types import ChunkType, StreamChunk react_module = importlib.import_module("xagent.core.agent.pattern.react.react") @@ -279,7 +283,28 @@ async def chat(self, **kwargs: Any) -> Any: async def stream_chat(self, **kwargs: Any) -> Any: self.stream_calls.append(kwargs) - if kwargs.get("tools"): + tool_names = [ + tool["function"]["name"] for tool in list(kwargs.get("tools") or []) + ] + if tool_names == ["final_answer"]: + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "id": "call_final", + "function": { + "name": "final_answer", + "arguments": ( + '{"response_language":"English",' + '"answer":"The result is 4."}' + ), + }, + } + ], + ) + yield StreamChunk(type=ChunkType.END) + return + if tool_names: yield StreamChunk( type=ChunkType.TOOL_CALL, tool_calls=[ @@ -299,6 +324,142 @@ async def stream_chat(self, **kwargs: Any) -> Any: yield StreamChunk(type=ChunkType.END) +class StreamingInvalidToolProtocolFinalAnswerLLM: + def __init__( + self, + *, + structured_work_tool: bool = False, + partial_final_before_work_tool: bool = False, + invalid_retry: bool = False, + ) -> None: + self.structured_work_tool = structured_work_tool + self.partial_final_before_work_tool = partial_final_before_work_tool + self.invalid_retry = invalid_retry + self.stream_calls: list[dict[str, Any]] = [] + + async def chat(self, **kwargs: Any) -> Any: + raise AssertionError("invalid tool protocol path should stay streaming") + + async def stream_chat(self, **kwargs: Any) -> Any: + self.stream_calls.append(kwargs) + call_index = len(self.stream_calls) - 1 + if call_index == 0: + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "id": "call_1", + "function": { + "name": "calculator", + "arguments": '{"expression":"2+2"}', + }, + } + ], + ) + elif call_index == 1: + if self.partial_final_before_work_tool: + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "index": 0, + "id": "call_partial_final", + "function": { + "name": "final_answer", + "arguments": ( + '{"response_language":"English",' + '"answer":"Draft answer"}' + ), + }, + } + ], + ) + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "index": 1, + "id": "call_unavailable_work_tool", + "function": { + "name": "calculator", + "arguments": '{"expression":"3+3"}', + }, + } + ], + ) + elif self.structured_work_tool: + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "id": "call_unavailable_work_tool", + "function": { + "name": "calculator", + "arguments": '{"expression":"3+3"}', + }, + } + ], + ) + else: + yield StreamChunk( + type=ChunkType.PROTOCOL_ERROR, + protocol_error={ + "provider": "deepseek", + "code": "serialized_tool_call_content", + "message": "Invalid provider tool protocol.", + }, + ) + elif self.invalid_retry: + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "index": 0, + "id": "call_retry_partial_final", + "function": { + "name": "final_answer", + "arguments": ( + '{"response_language":"English","answer":"Retry draft"}' + ), + }, + } + ], + ) + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "index": 1, + "id": "call_retry_unavailable_work_tool", + "function": { + "name": "calculator", + "arguments": '{"expression":"4+4"}', + }, + } + ], + ) + else: + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "id": "call_final", + "function": { + "name": "final_answer", + "arguments": json.dumps( + { + "response_language": "Simplified Chinese", + "answer": "目前没有找到可直接下载的音频片段。", + }, + ensure_ascii=False, + ), + }, + } + ], + ) + yield StreamChunk(type=ChunkType.END) + + class StreamingPlainTextFinalAnswerLLM: def __init__(self) -> None: self.stream_calls: list[dict[str, Any]] = [] @@ -450,7 +611,21 @@ async def stream_chat(self, **kwargs: Any) -> Any: ], ) else: - yield StreamChunk(type=ChunkType.TOKEN, delta="Fallback answer.") + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "id": "call_final", + "function": { + "name": "final_answer", + "arguments": ( + '{"response_language":"English",' + '"answer":"Fallback answer."}' + ), + }, + } + ], + ) yield StreamChunk(type=ChunkType.END) @@ -477,6 +652,31 @@ async def chat(self, **kwargs: Any) -> Any: raise +class BlockingProtocolRetryLLM: + def __init__(self) -> None: + self.calls = 0 + self.retry_started = asyncio.Event() + self.cancelled = False + + async def chat(self, **kwargs: Any) -> Any: + del kwargs + self.calls += 1 + if self.calls == 1: + return tool_protocol_error_response( + ToolProtocolViolation( + provider="deepseek", + code="serialized_tool_call_content", + message="Invalid provider tool protocol.", + ) + ) + self.retry_started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + self.cancelled = True + raise + + class FakeTracer: def __init__(self) -> None: self.events: list[tuple[str, dict[str, Any]]] = [] @@ -728,13 +928,237 @@ async def test_react_pattern_streams_only_final_answer_after_tool_call() -> None assert len(llm.calls) == 0 assert len(llm.stream_calls) == 2 assert llm.stream_calls[0]["tools"][0]["function"]["name"] == "calculator" - assert llm.stream_calls[1]["tools"] is None + assert [tool["function"]["name"] for tool in llm.stream_calls[1]["tools"]] == [ + "final_answer" + ] + assert llm.stream_calls[1]["tool_choice"] == "required" + assert [event["type"] for event in outbound.events] == [ + "final_answer_start", + "final_answer_delta", + "final_answer_end", + ] + + +@pytest.mark.parametrize("structured_work_tool", [False, True]) +@pytest.mark.asyncio +async def test_react_retries_invalid_forced_final_as_final_answer_tool( + structured_work_tool: bool, +) -> None: + llm = StreamingInvalidToolProtocolFinalAnswerLLM( + structured_work_tool=structured_work_tool + ) + pattern = ReActPattern(max_iterations=3, finalize_after_tool_result=True) + tool = FakeTool() + context = ExecutionContext(system_prompt="You are helpful.", execution_id="task-1") + context.add_user_message("Find an audio clip") + outbound = OutboundCollector() + runtime = PatternRuntime(execution_id="task-1", outbound_message_handler=outbound) + + result = await pattern.run( + context=context, + tools=[tool], + llm=llm, + runtime=runtime, + ) + + assert result["success"] is True + assert result["response"] == "目前没有找到可直接下载的音频片段。" + assert tool.calls == [{"expression": "2+2"}] + assert len(llm.stream_calls) == 3 + assert [tool["function"]["name"] for tool in llm.stream_calls[1]["tools"]] == [ + "final_answer" + ] + assert llm.stream_calls[1]["tool_choice"] == "required" + retry_tools = llm.stream_calls[2]["tools"] + assert [tool["function"]["name"] for tool in retry_tools] == ["final_answer"] + assert llm.stream_calls[2]["tool_choice"] == "required" + assert ( + "calling the final_answer control tool exactly once" + in (llm.stream_calls[2]["messages"][0]["content"]) + ) + outbound_text = "".join( + str(event.get("delta", "")) + str(event.get("content", "")) + for event in outbound.events + ) + assert "DSML" not in outbound_text + assert [event["type"] for event in outbound.events] == [ + "final_answer_start", + "final_answer_delta", + "final_answer_end", + ] + + +@pytest.mark.asyncio +async def test_react_retries_provider_tool_protocol_error_with_work_tools() -> None: + llm = FakeLLM( + responses=[ + tool_protocol_error_response( + ToolProtocolViolation( + provider="deepseek", + code="nested_serialized_tool_call", + message="Invalid provider tool protocol.", + ) + ), + { + "content": "", + "tool_calls": [ + { + "id": "call_write", + "function": { + "name": "write_file", + "arguments": json.dumps( + { + "file_path": "podcast.md", + "content": "# Podcast script\nComplete script body.", + } + ), + }, + } + ], + }, + { + "content": "", + "tool_calls": [ + { + "id": "call_final", + "function": { + "name": "final_answer", + "arguments": json.dumps( + { + "response_language": "English", + "answer": "The podcast script is ready.", + } + ), + }, + } + ], + }, + ] + ) + pattern = ReActPattern(max_iterations=2) + tool = FakeWriteFileTool() + context = ExecutionContext() + context.add_user_message("Create a podcast script and synthesize it.") + tracer = TraceEventRecorder() + runtime = PatternRuntime(tracer=tracer) + + result = await pattern.run(context=context, tools=[tool], llm=llm, runtime=runtime) + + assert result["success"] is True + assert result["response"] == "The podcast script is ready." + assert tool.calls == [ + { + "file_path": "podcast.md", + "content": "# Podcast script\nComplete script body.", + } + ] + assert len(llm.calls) == 3 + retry_tool_names = [schema["function"]["name"] for schema in llm.calls[1]["tools"]] + assert "write_file" in retry_tool_names + assert "final_answer" in retry_tool_names + assert ( + "If work remains, call the appropriate available work tool directly" + in llm.calls[1]["messages"][0]["content"] + ) + assert all( + message.content != "Invalid provider tool protocol." + for message in context.messages + if isinstance(message.content, str) + ) + llm_end_events = [ + event for event in tracer.events if event["event_type"] == "action_end_llm" + ] + assert llm_end_events[0]["data"]["success"] is False + assert llm_end_events[0]["data"]["phase"] == ("discarded_invalid_tool_protocol") + assert llm_end_events[1]["data"]["success"] is True + + +def test_react_accepts_null_tool_calls_in_forced_final_response() -> None: + pattern = ReActPattern() + + assert not pattern._response_requires_tool_protocol_retry( + {"content": "Done.", "tool_calls": None}, + force_final_answer=True, + ) + + +@pytest.mark.asyncio +async def test_react_closes_invalid_initial_final_answer_stream_before_retry() -> None: + llm = StreamingInvalidToolProtocolFinalAnswerLLM( + partial_final_before_work_tool=True + ) + pattern = ReActPattern(max_iterations=3, finalize_after_tool_result=True) + tool = FakeTool() + context = ExecutionContext(system_prompt="You are helpful.", execution_id="task-1") + context.add_user_message("Find an audio clip") + outbound = OutboundCollector() + runtime = PatternRuntime(execution_id="task-1", outbound_message_handler=outbound) + + result = await pattern.run( + context=context, + tools=[tool], + llm=llm, + runtime=runtime, + ) + + assert result["success"] is True + assert result["response"] == "目前没有找到可直接下载的音频片段。" + assert tool.calls == [{"expression": "2+2"}] assert [event["type"] for event in outbound.events] == [ "final_answer_start", "final_answer_delta", + "final_answer_error", + "final_answer_start", "final_answer_delta", "final_answer_end", ] + assert outbound.events[2]["error"] == "invalid tool protocol, retrying" + + +@pytest.mark.asyncio +async def test_react_closes_invalid_retry_final_answer_stream_before_failure() -> None: + llm = StreamingInvalidToolProtocolFinalAnswerLLM( + structured_work_tool=True, + invalid_retry=True, + ) + pattern = ReActPattern(max_iterations=3, finalize_after_tool_result=True) + tool = FakeTool() + context = ExecutionContext(system_prompt="You are helpful.", execution_id="task-1") + context.add_user_message("Find an audio clip") + outbound = OutboundCollector() + tracer = TraceEventRecorder() + runtime = PatternRuntime( + execution_id="task-1", + tracer=tracer, + outbound_message_handler=outbound, + ) + + result = await pattern.run( + context=context, + tools=[tool], + llm=llm, + runtime=runtime, + ) + + assert result["success"] is False + assert result["status"] == "invalid_tool_protocol" + assert tool.calls == [{"expression": "2+2"}] + assert [event["type"] for event in outbound.events] == [ + "final_answer_start", + "final_answer_delta", + "final_answer_error", + ] + assert outbound.events[2]["error"] == "invalid tool protocol after retry" + invalid_end_events = [ + event + for event in tracer.events + if event["event_type"] == "action_end_llm" + and event["data"].get("success") is False + ] + assert [event["data"]["phase"] for event in invalid_end_events] == [ + "discarded_invalid_tool_protocol", + "discarded_invalid_tool_protocol_retry", + ] @pytest.mark.asyncio @@ -1038,7 +1462,10 @@ async def test_react_pattern_finalizes_with_completion_evidence() -> None: assert result["success"] is True assert result["response"] == "Created output/en_poster.html." assert tool.calls == [{"file_path": "en_poster.html", "content": ""}] - assert llm.calls[1]["tools"] is None + assert [schema["function"]["name"] for schema in llm.calls[1]["tools"]] == [ + "final_answer" + ] + assert llm.calls[1]["tool_choice"] == "required" assert ( "Step completion evidence: The writer returned success=true" in (llm.calls[1]["messages"][0]["content"]) @@ -1119,7 +1546,9 @@ async def test_react_pattern_uses_decision_for_repeated_tools() -> None: decision_schema = llm.calls[2]["tools"][0]["function"]["parameters"] assert "response_language" not in decision_schema["properties"] assert "response_language" not in decision_schema["required"] - assert llm.calls[3]["tools"] is None + assert [schema["function"]["name"] for schema in llm.calls[3]["tools"]] == [ + "final_answer" + ] assert pattern.pending_tool_calls == [] @@ -1215,7 +1644,9 @@ async def test_react_repeated_decision_drains_current_tool_call_batch() -> None: assert tool_result_ids[-2:] == ["search_1", "search_2"] tool_names = [schema["function"]["name"] for schema in llm.calls[1]["tools"]] assert tool_names == ["react_decision"] - assert llm.calls[2]["tools"] is None + assert [schema["function"]["name"] for schema in llm.calls[2]["tools"]] == [ + "final_answer" + ] assert pattern.pending_tool_calls == [] @@ -1303,7 +1734,9 @@ async def test_react_pattern_uses_decision_after_cross_tool_attempts() -> None: decision_prompt = llm.calls[3]["messages"][-1]["content"] assert "3 consecutive work-tool calls without a final answer" in decision_prompt assert "latest work tool was browser_navigate" in decision_prompt - assert llm.calls[4]["tools"] is None + assert [schema["function"]["name"] for schema in llm.calls[4]["tools"]] == [ + "final_answer" + ] @pytest.mark.asyncio @@ -1392,7 +1825,9 @@ async def test_react_pattern_uses_decision_after_tool_group_successes() -> None: "3 consecutive successful calls in the research tool group" in decision_prompt ) assert "latest tool was search_source" in decision_prompt - assert llm.calls[4]["tools"] is None + assert [schema["function"]["name"] for schema in llm.calls[4]["tools"]] == [ + "final_answer" + ] assert pattern.pending_tool_calls == [] @@ -1470,7 +1905,9 @@ async def test_react_pattern_accepts_legacy_auto_reroute_kwarg() -> None: in (llm.calls[2]["messages"][-1]["content"]) ) assert llm.calls[2]["tool_choice"] == "required" - assert llm.calls[3]["tools"] is None + assert [schema["function"]["name"] for schema in llm.calls[3]["tools"]] == [ + "final_answer" + ] assert pattern.repeated_tool_decision is None @@ -1684,7 +2121,9 @@ async def test_react_pattern_repeated_guard_can_fire_twice_in_single_run() -> No assert [schema["function"]["name"] for schema in llm.calls[4]["tools"]] == [ "react_decision" ] - assert llm.calls[5]["tools"] is None + assert [schema["function"]["name"] for schema in llm.calls[5]["tools"]] == [ + "final_answer" + ] @pytest.mark.asyncio @@ -3118,6 +3557,38 @@ async def test_react_pattern_pause_cancels_active_llm_call() -> None: } +@pytest.mark.asyncio +async def test_react_pattern_pause_during_tool_protocol_retry_is_during_llm() -> None: + llm = BlockingProtocolRetryLLM() + runtime = PatternRuntime() + pattern = ReActPattern(max_iterations=3) + context = ExecutionContext() + context.add_user_message("Retry the model") + + task = asyncio.create_task( + pattern.run( + context=context, + tools=[FakeTool()], + llm=llm, + runtime=runtime, + ) + ) + await asyncio.wait_for(llm.retry_started.wait(), timeout=1) + + runtime.request_interrupt("paused during retry") + result = await asyncio.wait_for(task, timeout=1) + + assert llm.cancelled is True + assert result["success"] is False + assert result["status"] == "interrupted" + assert result["interrupt_reason"] == "paused during retry" + assert runtime.last_checkpoint["label"] == "interrupted" + assert runtime.last_checkpoint["metadata"] == { + "safe_point": "during_llm", + "reason": "paused during retry", + } + + @pytest.mark.asyncio async def test_react_pattern_interrupts_at_tool_boundary() -> None: llm = FakeLLM( diff --git a/tests/core/model/chat/basic/test_deepseek.py b/tests/core/model/chat/basic/test_deepseek.py index 8c22cc7ad..d23048f82 100644 --- a/tests/core/model/chat/basic/test_deepseek.py +++ b/tests/core/model/chat/basic/test_deepseek.py @@ -9,6 +9,7 @@ DeepSeekLLM, ) from xagent.core.model.chat.basic.openai import PROVIDER_STATE_METADATA_KEY, OpenAILLM +from xagent.core.model.chat.tool_protocol import get_tool_protocol_error from xagent.core.model.chat.types import ChunkType @@ -69,6 +70,42 @@ def test_structured_output_capabilities(self, llm): def test_deepseek_is_not_openai_subclass(self): assert not issubclass(DeepSeekLLM, OpenAILLM) + @pytest.mark.asyncio + async def test_chat_rejects_serialized_tool_protocol_content(self, llm, mocker): + message = SimpleNamespace( + content="<｜｜DSML｜｜tool_calls>", + tool_calls=None, + reasoning_content=None, + ) + response = SimpleNamespace( + choices=[SimpleNamespace(message=message)], + usage=None, + model_dump=lambda: {"id": "deepseek-invalid-protocol"}, + ) + mock_client = mocker.AsyncMock() + mock_client.chat.completions.create.return_value = response + mocker.patch( + "xagent.core.model.chat.basic.openai.AsyncOpenAI", + return_value=mock_client, + ) + + result = await llm.chat( + [{"role": "user", "content": "Use a tool"}], + tools=[ + { + "type": "function", + "function": { + "name": "search", + "parameters": {"type": "object", "properties": {}}, + }, + } + ], + ) + + error = get_tool_protocol_error(result) + assert error is not None + assert error["code"] == "serialized_tool_call_content" + def test_provider_reasoning_hook_disables_deepseek_thinking(self, llm): assert llm._prepare_provider_reasoning_extra_body( extra_body={"trace_id": "abc"}, diff --git a/tests/core/model/chat/basic/test_deepseek_tool_protocol.py b/tests/core/model/chat/basic/test_deepseek_tool_protocol.py new file mode 100644 index 000000000..eac71d17e --- /dev/null +++ b/tests/core/model/chat/basic/test_deepseek_tool_protocol.py @@ -0,0 +1,424 @@ +from __future__ import annotations + +import json +from collections.abc import AsyncIterator + +import pytest + +from xagent.core.model.chat.basic.deepseek_tool_protocol import ( + adapt_deepseek_stream, + normalize_deepseek_response, +) +from xagent.core.model.chat.tool_protocol import get_tool_protocol_error +from xagent.core.model.chat.types import ChunkType, StreamChunk + + +def _tool_schema( + name: str, + properties: dict[str, dict[str, str]], + *, + additional_properties: bool = True, +) -> dict[str, object]: + return { + "type": "function", + "function": { + "name": name, + "description": f"Call {name}.", + "parameters": { + "type": "object", + "additionalProperties": additional_properties, + "properties": properties, + }, + }, + } + + +FINAL_ANSWER_TOOL = _tool_schema( + "final_answer", + { + "response_language": {"type": "string"}, + "answer": {"type": "string"}, + }, + additional_properties=False, +) +WRITE_FILE_TOOL = _tool_schema( + "write_file", + { + "file_path": {"type": "string"}, + "content": {"type": "string"}, + }, +) + + +def test_deepseek_codec_rejects_serialized_tool_call_content() -> None: + response = { + "type": "text", + "content": ( + "Let me continue.\n\n" + "<｜｜DSML｜｜tool_calls>\n" + '<｜｜DSML｜｜invoke name="fetch_web_content">' + ), + "raw": {"id": "task-722-shape"}, + } + + normalized = normalize_deepseek_response( + response, + tools=[FINAL_ANSWER_TOOL], + ) + + error = get_tool_protocol_error(normalized) + assert error is not None + assert error["provider"] == "deepseek" + assert error["code"] == "serialized_tool_call_content" + assert normalized["tool_calls"] == [] + assert normalized["content"] == "" + + +def test_deepseek_codec_rejects_same_line_serialized_tool_call_content() -> None: + response = { + "type": "text", + "content": "Sure: <｜｜DSML｜｜tool_calls>", + } + + normalized = normalize_deepseek_response( + response, + tools=[FINAL_ANSWER_TOOL], + ) + + error = get_tool_protocol_error(normalized) + assert error is not None + assert error["code"] == "serialized_tool_call_content" + + +def test_deepseek_codec_rejects_nested_work_call_in_final_answer() -> None: + response = { + "type": "tool_call", + "tool_calls": [ + { + "id": "call_malformed_final", + "type": "function", + "function": { + "name": "final_answer", + "arguments": json.dumps( + { + "response_language": "English", + "answer": ( + "I'll write the script.\n\n" + "<｜｜DSML｜｜tool_calls>\n" + '<｜｜DSML｜｜invoke name="write_file">\n' + '<｜｜DSML｜｜parameter name="file_path" ' + 'string="true">podcast.md' + ), + "content": "# Podcast script", + }, + ensure_ascii=False, + ), + }, + } + ], + } + + normalized = normalize_deepseek_response( + response, + tools=[WRITE_FILE_TOOL, FINAL_ANSWER_TOOL], + ) + + error = get_tool_protocol_error(normalized) + assert error is not None + assert error["code"] == "nested_serialized_tool_call" + + +@pytest.mark.parametrize( + ("tool_call", "tools", "expected_code"), + [ + ( + {"function": {"arguments": "{}"}}, + [WRITE_FILE_TOOL], + "malformed_tool_call", + ), + ( + { + "function": { + "name": "write_file", + "arguments": '{"file_path":', + } + }, + [WRITE_FILE_TOOL], + "malformed_tool_arguments", + ), + ( + { + "function": { + "name": "fetch_web_content", + "arguments": "{}", + } + }, + [WRITE_FILE_TOOL], + "unavailable_tool_call", + ), + ( + { + "function": { + "name": "final_answer", + "arguments": json.dumps( + { + "response_language": "English", + "answer": "Done.", + "content": "unexpected", + } + ), + } + }, + [FINAL_ANSWER_TOOL], + "unexpected_tool_arguments", + ), + ], +) +def test_deepseek_codec_reports_structured_tool_violations( + tool_call, tools, expected_code +) -> None: + normalized = normalize_deepseek_response( + {"type": "tool_call", "tool_calls": [tool_call]}, + tools=tools, + ) + + error = get_tool_protocol_error(normalized) + assert error is not None + assert error["code"] == expected_code + + +def test_deepseek_codec_keeps_valid_tool_call() -> None: + response = { + "type": "tool_call", + "tool_calls": [ + { + "id": "call_write", + "type": "function", + "function": { + "name": "write_file", + "arguments": json.dumps( + {"file_path": "podcast.md", "content": "script"} + ), + }, + } + ], + } + + assert normalize_deepseek_response(response, tools=[WRITE_FILE_TOOL]) is response + + +def test_deepseek_codec_is_inactive_without_requested_tools() -> None: + response = { + "type": "text", + "content": "A DSML tool_calls marker can be discussed as ordinary text.", + } + + assert normalize_deepseek_response(response, tools=None) is response + + +@pytest.mark.asyncio +async def test_deepseek_stream_suppresses_serialized_protocol_tokens() -> None: + async def source() -> AsyncIterator[StreamChunk]: + yield StreamChunk( + type=ChunkType.TOKEN, + delta="Let me continue.\n\n<｜｜D", + ) + yield StreamChunk( + type=ChunkType.TOKEN, + delta="SML｜｜tool_calls>\n<｜｜DSML｜｜invoke", + ) + yield StreamChunk(type=ChunkType.END, finish_reason="stop") + + chunks = [ + chunk + async for chunk in adapt_deepseek_stream( + source(), + tools=[FINAL_ANSWER_TOOL], + ) + ] + + streamed_text = "".join(chunk.delta for chunk in chunks if chunk.is_token()) + assert streamed_text == "Let me continue.\n\n" + assert "DSML" not in streamed_text + protocol_errors = [chunk for chunk in chunks if chunk.is_protocol_error()] + assert len(protocol_errors) == 1 + assert protocol_errors[0].protocol_error["code"] == ("serialized_tool_call_content") + + +@pytest.mark.asyncio +async def test_deepseek_stream_suppresses_same_line_split_marker() -> None: + async def source() -> AsyncIterator[StreamChunk]: + yield StreamChunk( + type=ChunkType.TOKEN, + delta="Sure: <｜", + ) + yield StreamChunk( + type=ChunkType.TOKEN, + delta="｜DSML｜｜tool_calls>", + ) + yield StreamChunk(type=ChunkType.END, finish_reason="stop") + + chunks = [ + chunk + async for chunk in adapt_deepseek_stream( + source(), + tools=[FINAL_ANSWER_TOOL], + ) + ] + + streamed_text = "".join(chunk.delta for chunk in chunks if chunk.is_token()) + assert streamed_text == "Sure: " + assert len([chunk for chunk in chunks if chunk.is_protocol_error()]) == 1 + + +@pytest.mark.asyncio +async def test_deepseek_stream_replaces_nested_final_answer_with_protocol_error() -> ( + None +): + malformed_arguments = json.dumps( + { + "response_language": "English", + "answer": ( + "Drafting now.\n\n" + "<｜｜DSML｜｜tool_calls>\n" + '<｜｜DSML｜｜invoke name="write_file">' + ), + "content": "script", + }, + ensure_ascii=False, + ) + + async def source() -> AsyncIterator[StreamChunk]: + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "id": "call_malformed_final", + "type": "function", + "function": { + "name": "final_answer", + "arguments": malformed_arguments, + }, + } + ], + ) + yield StreamChunk(type=ChunkType.END, finish_reason="tool_calls") + + chunks = [ + chunk + async for chunk in adapt_deepseek_stream( + source(), + tools=[WRITE_FILE_TOOL, FINAL_ANSWER_TOOL], + ) + ] + + assert not any(chunk.is_tool_call() for chunk in chunks) + protocol_errors = [chunk for chunk in chunks if chunk.is_protocol_error()] + assert len(protocol_errors) == 1 + assert protocol_errors[0].protocol_error["code"] == ("nested_serialized_tool_call") + + +@pytest.mark.asyncio +async def test_deepseek_stream_forwards_accumulated_valid_tool_calls() -> None: + async def source() -> AsyncIterator[StreamChunk]: + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "index": 0, + "id": "call_write", + "type": "function", + "function": { + "name": "write_file", + "arguments": '{"file_path":', + }, + } + ], + ) + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "index": 0, + "id": "call_write", + "type": "function", + "function": { + "name": "write_file", + "arguments": ('{"file_path":"podcast.md","content":"script"}'), + }, + } + ], + finish_reason="tool_calls", + ) + + chunks = [ + chunk + async for chunk in adapt_deepseek_stream( + source(), + tools=[WRITE_FILE_TOOL], + ) + ] + + tool_chunks = [chunk for chunk in chunks if chunk.is_tool_call()] + assert len(tool_chunks) == 2 + assert tool_chunks[-1].tool_calls[0]["function"]["arguments"] == ( + '{"file_path":"podcast.md","content":"script"}' + ) + + +@pytest.mark.asyncio +async def test_deepseek_stream_withholds_split_nested_marker_from_tool_args() -> None: + async def source() -> AsyncIterator[StreamChunk]: + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "index": 0, + "id": "call_final", + "type": "function", + "function": { + "name": "final_answer", + "arguments": ( + '{"response_language":"English","answer":"Draft: <｜' + ), + }, + } + ], + ) + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "index": 0, + "id": "call_final", + "type": "function", + "function": { + "name": "final_answer", + "arguments": ( + '{"response_language":"English",' + '"answer":"Draft: <｜｜DSML｜｜tool_calls>"}' + ), + }, + } + ], + finish_reason="tool_calls", + ) + + chunks = [ + chunk + async for chunk in adapt_deepseek_stream( + source(), + tools=[FINAL_ANSWER_TOOL], + ) + ] + + forwarded_arguments = [ + chunk.tool_calls[0]["function"]["arguments"] + for chunk in chunks + if chunk.is_tool_call() + ] + assert forwarded_arguments == ['{"response_language":"English","answer":"Draft: '] + assert all("DSML" not in arguments for arguments in forwarded_arguments) + protocol_errors = [chunk for chunk in chunks if chunk.is_protocol_error()] + assert len(protocol_errors) == 1 + assert protocol_errors[0].protocol_error["code"] == ("nested_serialized_tool_call") diff --git a/tests/core/model/chat/basic/test_openrouter.py b/tests/core/model/chat/basic/test_openrouter.py index ec0f400c1..60bc84a31 100644 --- a/tests/core/model/chat/basic/test_openrouter.py +++ b/tests/core/model/chat/basic/test_openrouter.py @@ -5,6 +5,66 @@ import pytest from xagent.core.model.chat.basic.openrouter import OpenRouterLLM +from xagent.core.model.chat.tool_protocol import get_tool_protocol_error + + +@pytest.mark.parametrize( + ("model_name", "expected"), + [ + ("deepseek/deepseek-v4-flash", True), + ("openrouter/deepseek/deepseek-v4-flash", True), + ("anthropic/claude-sonnet-4.6", False), + ], +) +def test_openrouter_uses_deepseek_tool_protocol_only_for_deepseek_models( + model_name, expected +): + llm = OpenRouterLLM(model_name=model_name, api_key="test-key") + + assert llm._uses_deepseek_tool_protocol is expected + + +@pytest.mark.asyncio +async def test_openrouter_deepseek_rejects_serialized_tool_protocol_content( + mocker, +): + message = SimpleNamespace( + content="<｜｜DSML｜｜tool_calls>", + tool_calls=None, + reasoning_content=None, + ) + response = SimpleNamespace( + choices=[SimpleNamespace(message=message)], + usage=None, + model_dump=lambda: {"id": "openrouter-deepseek-invalid-protocol"}, + ) + mock_client = mocker.AsyncMock() + mock_client.chat.completions.create.return_value = response + mocker.patch( + "xagent.core.model.chat.basic.openai.AsyncOpenAI", + return_value=mock_client, + ) + llm = OpenRouterLLM( + model_name="deepseek/deepseek-v4-flash", + api_key="test-key", + ) + + result = await llm.chat( + [{"role": "user", "content": "Use a tool"}], + tools=[ + { + "type": "function", + "function": { + "name": "search", + "parameters": {"type": "object", "properties": {}}, + }, + } + ], + ) + + error = get_tool_protocol_error(result) + assert error is not None + assert error["code"] == "serialized_tool_call_content" @pytest.mark.asyncio