diff --git a/src/xagent/core/agent/pattern/react/react.py b/src/xagent/core/agent/pattern/react/react.py index e8949eac4..8bf00f4f8 100644 --- a/src/xagent/core/agent/pattern/react/react.py +++ b/src/xagent/core/agent/pattern/react/react.py @@ -385,6 +385,15 @@ async def _run_tool_calling_loop( "tools": tool_schemas or None, "tool_choice": self.tool_choice if tool_schemas else None, } + start_final_answer_stream_immediately = True + if force_final_answer_now: + parser_kwargs = self._forced_final_parser_kwargs( + llm=llm, + tool_schemas=base_tool_schemas, + ) + if parser_kwargs: + llm_kwargs.update(parser_kwargs) + start_final_answer_stream_immediately = False if tool_schemas: answer_streamer = ReActFinalAnswerStreamer(runtime) response = await runtime.run_streaming_llm_call( @@ -393,7 +402,11 @@ async def _run_tool_calling_loop( **llm_kwargs, ) else: - response = await runtime.stream_final_answer(llm, **llm_kwargs) + response = await runtime.stream_final_answer( + llm, + start_immediately=start_final_answer_stream_immediately, + **llm_kwargs, + ) except LLMCallInterrupted: if answer_streamer is not None: await answer_streamer.fail("interrupted during LLM stream") @@ -445,6 +458,8 @@ async def _run_tool_calling_loop( self._remember_tool_call_content(tool_calls, assistant_content) self.status = "acting" self.pending_tool_calls = list(tool_calls) + if force_final_answer_now: + self.force_final_answer_next = False await runtime.checkpoint("after_llm", context=context, pattern=self) pending_result = await self._execute_pending_tool_calls( context=context, @@ -584,6 +599,22 @@ def _schema_tool_names(self, tool_schemas: list[dict[str, Any]]) -> list[str]: names.append(str(function["name"])) return names + def _forced_final_parser_kwargs( + self, + *, + llm: Any, + tool_schemas: list[dict[str, Any]], + ) -> dict[str, Any]: + """Return provider-local parser kwargs for final turns that suppress tools.""" + + if not tool_schemas: + return {} + parse_kwargs = getattr(llm, "deepseek_dsml_parse_kwargs", None) + if not callable(parse_kwargs): + return {} + result = parse_kwargs(tool_schemas) + return result if isinstance(result, dict) else {} + def _tool_decision_groups_for_tools(self, tools: list[Any]) -> dict[str, str]: groups: dict[str, str] = {} for tool in tools: diff --git a/src/xagent/core/agent/runtime.py b/src/xagent/core/agent/runtime.py index a8d1db266..54cdb2596 100644 --- a/src/xagent/core/agent/runtime.py +++ b/src/xagent/core/agent/runtime.py @@ -95,7 +95,13 @@ async def run_llm_call(self, llm: Any, **kwargs: Any) -> Any: finally: self._active_llm_tasks.discard(task) - async def stream_final_answer(self, llm: Any, **kwargs: Any) -> Any: + async def stream_final_answer( + self, + llm: Any, + *, + start_immediately: bool = True, + **kwargs: Any, + ) -> Any: """Stream only the final user-facing answer through the outbound boundary.""" if self.outbound_message_handler is None or not self._has_native_stream_chat( @@ -106,12 +112,13 @@ async def stream_final_answer(self, llm: Any, **kwargs: Any) -> Any: from .pattern.final_answer_stream import FinalAnswerStreamSession stream = FinalAnswerStreamSession(self) - if await stream.start() is None: + if start_immediately and await stream.start() is None: return await self.run_llm_call(llm, **kwargs) + defer_text_delta = not start_immediately async def emit_text_delta(chunk: Any) -> None: delta = self._chunk_text_delta(chunk) - if delta: + if delta and not defer_text_delta: await stream.emit_delta(delta) try: @@ -123,6 +130,13 @@ async def emit_text_delta(chunk: Any) -> None: raise content = self._response_content(response) + if self._response_tool_calls(response): + if stream.started: + await stream.fail( + "Final-answer stream stopped because the response contained tool calls." + ) + return response + await stream.finish(content) return response @@ -373,6 +387,12 @@ def _response_content(self, response: Any) -> str: return str(value) return str(response) + def _response_tool_calls(self, response: Any) -> list[Any]: + if not isinstance(response, dict): + return [] + tool_calls = response.get("tool_calls") + return list(tool_calls) if tool_calls else [] + async def _emit_outbound(self, payload: dict[str, Any]) -> None: if self.outbound_message_handler is not None: await self._maybe_await(self.outbound_message_handler(payload)) diff --git a/src/xagent/core/model/chat/basic/deepseek.py b/src/xagent/core/model/chat/basic/deepseek.py index ea3fcaffb..6489b90d3 100644 --- a/src/xagent/core/model/chat/basic/deepseek.py +++ b/src/xagent/core/model/chat/basic/deepseek.py @@ -4,6 +4,12 @@ from ...providers import is_placeholder_api_key from .base import StreamChunk +from .deepseek_dsml import ( + DEEPSEEK_DSML_PARSE_TOOLS_KWARG, + DeepSeekDSMLParseError, + normalize_deepseek_dsml_response, + stream_chunks_from_chat_result, +) from .openai import PROVIDER_STATE_METADATA_KEY, OpenAICompatibleLLM logger = logging.getLogger(__name__) @@ -131,6 +137,33 @@ def _prepare_deepseek_kwargs( return updated_kwargs + def deepseek_dsml_parse_kwargs( + self, + tools: list[dict[str, Any]] | None, + ) -> dict[str, Any]: + """Return parser-only DSML schema metadata for forced-final recovery. + + ReAct forced-final turns intentionally send ``tools=None`` to DeepSeek. + If DeepSeek still emits DSML tool markup, this hidden kwarg lets this + adapter validate and convert it locally without leaking the schema into + provider request parameters. + """ + + return {DEEPSEEK_DSML_PARSE_TOOLS_KWARG: tools} if tools else {} + + def _pop_deepseek_dsml_parse_tools( + self, + kwargs: dict[str, Any], + ) -> list[dict[str, Any]] | None: + raw_tools = kwargs.pop(DEEPSEEK_DSML_PARSE_TOOLS_KWARG, None) + if raw_tools is None: + return None + if not isinstance(raw_tools, list): + raise DeepSeekDSMLParseError( + f"{DEEPSEEK_DSML_PARSE_TOOLS_KWARG} must be a list of tool schemas." + ) + return raw_tools + def _prepare_messages_for_request( self, messages: List[Dict[str, Any]], @@ -219,6 +252,23 @@ def _normalize_response_format( return response_format, output_config + def _normalize_deepseek_response( + self, + response: Any, + *, + tools: Optional[List[Dict[str, Any]]], + ) -> Any: + normalized_response, parsed_dsml = normalize_deepseek_dsml_response( + response, + tools=tools, + model_name=self._model_name, + ) + if parsed_dsml and isinstance(normalized_response, dict): + provider_state = self._response_provider_state(normalized_response) + if provider_state: + normalized_response[PROVIDER_STATE_METADATA_KEY] = provider_state + return normalized_response + async def chat( self, messages: List[Dict[str, str]], @@ -231,12 +281,13 @@ async def chat( output_config: Optional[Dict[str, Any]] = None, **kwargs: Any, ) -> Any: + dsml_parse_tools = self._pop_deepseek_dsml_parse_tools(kwargs) response_format, output_config = self._normalize_response_format( response_format=response_format, output_config=output_config, ) kwargs = self._prepare_deepseek_kwargs(kwargs=kwargs) - return await super().chat( + result = await super().chat( messages=messages, temperature=temperature, max_tokens=max_tokens, @@ -247,6 +298,10 @@ async def chat( output_config=output_config, **kwargs, ) + return self._normalize_deepseek_response( + result, + tools=tools or dsml_parse_tools, + ) async def stream_chat( self, @@ -260,6 +315,26 @@ async def stream_chat( output_config: Optional[Dict[str, Any]] = None, **kwargs: Any, ) -> AsyncIterator[StreamChunk]: + # DSML recovery needs a complete response. Keep ordinary tool-enabled + # streams native; only forced-final turns opt into buffering with the + # private sentinel because those calls intentionally suppress provider + # tools while still needing local DSML validation. + if DEEPSEEK_DSML_PARSE_TOOLS_KWARG in kwargs: + result = await self.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, + ) + for chunk in stream_chunks_from_chat_result(result): + yield chunk + return + response_format, output_config = self._normalize_response_format( response_format=response_format, output_config=output_config, diff --git a/src/xagent/core/model/chat/basic/deepseek_dsml.py b/src/xagent/core/model/chat/basic/deepseek_dsml.py new file mode 100644 index 000000000..e7c64408e --- /dev/null +++ b/src/xagent/core/model/chat/basic/deepseek_dsml.py @@ -0,0 +1,304 @@ +from __future__ import annotations + +import html +import json +import re +import uuid +from typing import Any, Iterable + +from ..types import ChunkType, StreamChunk + + +class DeepSeekDSMLParseError(RuntimeError): + """Raised when DeepSeek emits DSML text that cannot safely become tool calls.""" + + +DEEPSEEK_DSML_PARSE_TOOLS_KWARG = "_xagent_deepseek_dsml_parse_tools" +_DSML_TOKEN_RE = r"(?:｜｜DSML｜｜|｜DSML｜|\|\|DSML\|\||\|DSML\|)" +_DSML_TOOL_CALLS_BLOCK_RE = re.compile( + rf"<\s*{_DSML_TOKEN_RE}\s*tool_calls\s*>" + rf"(.*?)" + rf"", + re.IGNORECASE | re.DOTALL, +) +_DSML_INVOKE_RE = re.compile( + rf"<\s*{_DSML_TOKEN_RE}\s*invoke\b([^>]*)>" + rf"(.*?)" + rf"", + re.IGNORECASE | re.DOTALL, +) +_DSML_PARAMETER_RE = re.compile( + rf"<\s*{_DSML_TOKEN_RE}\s*parameter\b([^>]*)>" + rf"(.*?)" + rf"", + re.IGNORECASE | re.DOTALL, +) +_DSML_ATTR_RE = re.compile(r"""([A-Za-z_][\w:-]*)\s*=\s*(?:"([^"]*)"|'([^']*)')""") +_DSML_MARKER_RE = re.compile( + rf"<\s*/?\s*{_DSML_TOKEN_RE}\s*(?:tool_calls|invoke|parameter)\b", + re.IGNORECASE, +) + + +def contains_deepseek_dsml_tool_markup(text: Any) -> bool: + """Return true when text contains a DeepSeek DSML tool-call marker.""" + + return isinstance(text, str) and _DSML_MARKER_RE.search(text) is not None + + +def normalize_deepseek_dsml_response( + response: Any, + *, + tools: list[dict[str, Any]] | None, + model_name: str, +) -> tuple[Any, bool]: + """Convert DeepSeek DSML assistant content into Xagent tool-call payloads. + + The OpenAI-compatible base parser can only see official ``tool_calls``. Some + DeepSeek V4 routes leak their DSML tool grammar into ``message.content`` + instead. This helper is deliberately provider-scoped: callers must opt in + from DeepSeek-specific code paths and provide the tool schemas from the + current request. + """ + + if not isinstance(response, dict): + return response, False + + content = response.get("content") + if not isinstance(content, str) or not contains_deepseek_dsml_tool_markup(content): + return response, False + + if response.get("tool_calls"): + normalized = dict(response) + normalized["content"] = _strip_dsml_markup_from_visible_content(content) or None + return normalized, True + + content = _trim_unmatched_dsml_marker_tail(content) + if _DSML_TOOL_CALLS_BLOCK_RE.search(content) is None: + normalized = dict(response) + normalized["content"] = _strip_dsml_markup_from_visible_content(content) or None + return normalized, True + + parsed = parse_deepseek_dsml_tool_calls( + content, + tools=tools, + model_name=model_name, + ) + normalized = dict(response) + normalized["type"] = "tool_call" + normalized["content"] = parsed["content"] or None + normalized["tool_calls"] = parsed["tool_calls"] + return normalized, True + + +def parse_deepseek_dsml_tool_calls( + text: str, + *, + tools: list[dict[str, Any]] | None, + model_name: str, +) -> dict[str, Any]: + """Parse complete DeepSeek DSML ``tool_calls`` blocks from assistant text.""" + + if not tools: + raise DeepSeekDSMLParseError( + f"{model_name} returned DSML tool markup, but no tools were supplied." + ) + + available_tools = _available_tool_names(tools) + if not available_tools: + raise DeepSeekDSMLParseError( + f"{model_name} returned DSML tool markup, but no callable tool schemas were supplied." + ) + + matches, clean_content = _extract_dsml_tool_call_blocks( + text, + model_name=model_name, + ) + + tool_calls: list[dict[str, Any]] = [] + for match in matches: + tool_calls.extend( + _parse_dsml_invokes( + match.group(1), + available_tools=available_tools, + model_name=model_name, + ) + ) + + if not tool_calls: + raise DeepSeekDSMLParseError( + f"{model_name} returned DSML tool markup without any valid invoke blocks." + ) + + return {"content": clean_content, "tool_calls": tool_calls} + + +def stream_chunks_from_chat_result(response: dict[str, Any]) -> Iterable[StreamChunk]: + """Represent a non-streaming DeepSeek chat result as stream chunks. + + ``chat()`` results are always dicts produced by the OpenAI-compatible + response processing, so only the dict shape is handled here. + """ + + content = response.get("content") + if content: + yield StreamChunk( + type=ChunkType.TOKEN, + content=content, + delta=content, + raw=response, + ) + + if response.get("tool_calls"): + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=list(response.get("tool_calls") or []), + finish_reason="tool_calls", + raw=response, + ) + return + + yield StreamChunk(type=ChunkType.END, finish_reason="stop", raw=response) + + +def _extract_dsml_tool_call_blocks( + text: str, + *, + model_name: str, +) -> tuple[list[re.Match[str]], str]: + matches = list(_DSML_TOOL_CALLS_BLOCK_RE.finditer(text)) + if not matches: + raise DeepSeekDSMLParseError( + f"{model_name} returned incomplete DeepSeek DSML tool markup." + ) + + clean_content = _DSML_TOOL_CALLS_BLOCK_RE.sub("", text).strip() + if contains_deepseek_dsml_tool_markup(clean_content): + raise DeepSeekDSMLParseError( + f"{model_name} returned unparsed DeepSeek DSML tool markup." + ) + return matches, clean_content + + +def _strip_dsml_markup_from_visible_content(text: str) -> str: + clean_content = _DSML_TOOL_CALLS_BLOCK_RE.sub("", text).strip() + residual_marker = _DSML_MARKER_RE.search(clean_content) + if residual_marker is None: + return clean_content + return clean_content[: residual_marker.start()].strip() + + +def _trim_unmatched_dsml_marker_tail(text: str) -> str: + complete_blocks = list(_DSML_TOOL_CALLS_BLOCK_RE.finditer(text)) + for marker in _DSML_MARKER_RE.finditer(text): + if not any( + block.start() <= marker.start() < block.end() for block in complete_blocks + ): + return text[: marker.start()].strip() + return text + + +def _parse_dsml_invokes( + block_content: str, + *, + available_tools: set[str], + model_name: str, +) -> list[dict[str, Any]]: + invoke_matches = list(_DSML_INVOKE_RE.finditer(block_content)) + if not invoke_matches: + raise DeepSeekDSMLParseError( + f"{model_name} returned a DSML tool_calls block without invoke blocks." + ) + + tool_calls: list[dict[str, Any]] = [] + for match in invoke_matches: + attrs = _parse_dsml_attrs(match.group(1)) + tool_name = attrs.get("name", "").strip() + if not tool_name: + raise DeepSeekDSMLParseError( + f"{model_name} returned a DSML invoke without a tool name." + ) + if tool_name not in available_tools: + raise DeepSeekDSMLParseError( + f"{model_name} returned a DSML invoke for unknown tool '{tool_name}'." + ) + + arguments = _parse_dsml_parameters( + match.group(2), + model_name=model_name, + tool_name=tool_name, + ) + tool_calls.append( + { + "id": f"call_dsml_{uuid.uuid4().hex[:24]}", + "type": "function", + "function": { + "name": tool_name, + "arguments": json.dumps(arguments, ensure_ascii=False), + }, + } + ) + + return tool_calls + + +def _parse_dsml_parameters( + invoke_content: str, + *, + model_name: str, + tool_name: str, +) -> dict[str, Any]: + arguments: dict[str, Any] = {} + for match in _DSML_PARAMETER_RE.finditer(invoke_content): + attrs = _parse_dsml_attrs(match.group(1)) + name = attrs.get("name", "").strip() + string_attr = attrs.get("string", "").strip().lower() + if not name: + raise DeepSeekDSMLParseError( + f"{model_name} returned a DSML parameter without a name for tool '{tool_name}'." + ) + if name in arguments: + raise DeepSeekDSMLParseError( + f"{model_name} returned duplicate DSML parameter '{name}' for tool '{tool_name}'." + ) + if string_attr not in {"true", "false"}: + raise DeepSeekDSMLParseError( + f"{model_name} returned DSML parameter '{name}' with invalid string attribute." + ) + + raw_value = html.unescape(match.group(2)) + if string_attr == "true": + arguments[name] = raw_value + else: + arguments[name] = _parse_json_value_or_string(raw_value.strip()) + + return arguments + + +def _parse_dsml_attrs(raw_attrs: str) -> dict[str, str]: + attrs: dict[str, str] = {} + for match in _DSML_ATTR_RE.finditer(raw_attrs): + value = match.group(2) if match.group(2) is not None else match.group(3) + attrs[match.group(1).strip().lower()] = html.unescape(value or "") + return attrs + + +def _parse_json_value_or_string(value: str) -> Any: + try: + return json.loads(value) + except json.JSONDecodeError: + return value + + +def _available_tool_names(tools: list[dict[str, Any]]) -> set[str]: + names: set[str] = set() + for tool in tools: + if not isinstance(tool, dict): + continue + function = tool.get("function") + if not isinstance(function, dict): + continue + name = function.get("name") + if isinstance(name, str) and name.strip(): + names.add(name.strip()) + return names diff --git a/tests/core/agent/test_react.py b/tests/core/agent/test_react.py index a9dd56499..c45c689c7 100644 --- a/tests/core/agent/test_react.py +++ b/tests/core/agent/test_react.py @@ -17,6 +17,7 @@ ReActReasoningMode, ToolCallRecord, ) +from xagent.core.model.chat.basic.deepseek_dsml import DEEPSEEK_DSML_PARSE_TOOLS_KWARG from xagent.core.model.chat.types import ChunkType, StreamChunk react_module = importlib.import_module("xagent.core.agent.pattern.react.react") @@ -270,6 +271,69 @@ async def stream_chat(self, **kwargs: Any) -> Any: yield StreamChunk(type=ChunkType.END) +class StreamingDeepSeekForcedFinalToolLLM: + """DeepSeek-like fake that can recover a tool call during a forced-final turn.""" + + def __init__(self, *, forced_final_preamble: str | None = None) -> None: + self.forced_final_preamble = forced_final_preamble + self.stream_calls: list[dict[str, Any]] = [] + self.parser_tool_batches: list[list[dict[str, Any]]] = [] + + def deepseek_dsml_parse_kwargs( + self, + tools: list[dict[str, Any]], + ) -> dict[str, Any]: + self.parser_tool_batches.append(tools) + return {DEEPSEEK_DSML_PARSE_TOOLS_KWARG: tools} + + async def chat(self, **kwargs: Any) -> Any: + raise AssertionError("streaming DeepSeek ReAct path should not call chat()") + + async def stream_chat(self, **kwargs: Any) -> Any: + self.stream_calls.append(kwargs) + call_number = len(self.stream_calls) + + if call_number == 1: + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "id": "call_initial", + "function": { + "name": "calculator", + "arguments": '{"expression":"2+2"}', + }, + } + ], + ) + yield StreamChunk(type=ChunkType.END) + return + + if call_number == 2 and kwargs.get(DEEPSEEK_DSML_PARSE_TOOLS_KWARG): + if self.forced_final_preamble: + yield StreamChunk( + type=ChunkType.TOKEN, + delta=self.forced_final_preamble, + ) + yield StreamChunk( + type=ChunkType.TOOL_CALL, + tool_calls=[ + { + "id": "call_forced_final_recovered", + "function": { + "name": "calculator", + "arguments": '{"expression":"3+3"}', + }, + } + ], + ) + yield StreamChunk(type=ChunkType.END) + return + + yield StreamChunk(type=ChunkType.TOKEN, delta="The final result is 6.") + yield StreamChunk(type=ChunkType.END) + + class StreamingPlainTextFinalAnswerLLM: def __init__(self) -> None: self.stream_calls: list[dict[str, Any]] = [] @@ -708,6 +772,69 @@ async def test_react_pattern_streams_only_final_answer_after_tool_call() -> None ] +@pytest.mark.asyncio +async def test_react_pattern_converts_deepseek_forced_final_dsml_tool_call() -> None: + llm = StreamingDeepSeekForcedFinalToolLLM() + pattern = ReActPattern(max_iterations=4, finalize_after_tool_result=True) + tool = FakeTool() + context = ExecutionContext(system_prompt="You are helpful.", execution_id="task-1") + context.add_user_message("Calculate 2+2 and then 3+3") + 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"] == "The final result is 6." + assert tool.calls == [{"expression": "2+2"}, {"expression": "3+3"}] + assert len(llm.stream_calls) == 3 + assert llm.stream_calls[1]["tools"] is None + assert DEEPSEEK_DSML_PARSE_TOOLS_KWARG in llm.stream_calls[1] + assert [schema["function"]["name"] for schema in llm.parser_tool_batches[0]] == [ + "calculator", + "final_answer", + "send_message", + "ask_user_question", + ] + assert [event["type"] for event in outbound.events] == [ + "final_answer_start", + "final_answer_delta", + "final_answer_end", + ] + assert outbound.events[1]["delta"] == "The final result is 6." + + +@pytest.mark.asyncio +async def test_react_pattern_does_not_finish_forced_final_preamble_before_tool_call() -> ( + None +): + llm = StreamingDeepSeekForcedFinalToolLLM( + forced_final_preamble="I need one more calculation." + ) + pattern = ReActPattern(max_iterations=4, finalize_after_tool_result=True) + tool = FakeTool() + context = ExecutionContext(system_prompt="You are helpful.", execution_id="task-1") + context.add_user_message("Calculate 2+2 and then 3+3") + 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"] == "The final result is 6." + assert tool.calls == [{"expression": "2+2"}, {"expression": "3+3"}] + assert [event["type"] for event in outbound.events] == [ + "final_answer_start", + "final_answer_delta", + "final_answer_end", + ] + assert [event.get("delta") for event in outbound.events] == [ + None, + "The final result is 6.", + None, + ] + + @pytest.mark.asyncio async def test_react_pattern_does_not_stream_plain_text_when_tool_protocol_is_ignored() -> ( None diff --git a/tests/core/model/chat/basic/test_deepseek.py b/tests/core/model/chat/basic/test_deepseek.py index 8c22cc7ad..d2219f6c5 100644 --- a/tests/core/model/chat/basic/test_deepseek.py +++ b/tests/core/model/chat/basic/test_deepseek.py @@ -8,10 +8,39 @@ DEEPSEEK_REASONING_CONTENT_STATE_KEY, DeepSeekLLM, ) +from xagent.core.model.chat.basic.deepseek_dsml import ( + DEEPSEEK_DSML_PARSE_TOOLS_KWARG, + DeepSeekDSMLParseError, +) from xagent.core.model.chat.basic.openai import PROVIDER_STATE_METADATA_KEY, OpenAILLM from xagent.core.model.chat.types import ChunkType +def _browser_extract_text_tool() -> dict: + return { + "type": "function", + "function": { + "name": "browser_extract_text", + "description": "Extract text from a browser session", + "parameters": { + "type": "object", + "properties": {"session_id": {"type": "string"}}, + "required": ["session_id"], + }, + }, + } + + +def _browser_extract_text_dsml() -> str: + return """ +<｜｜DSML｜｜tool_calls> +<｜｜DSML｜｜invoke name="browser_extract_text"> +<｜｜DSML｜｜parameter name="session_id" string="true">821:react_468b98b2 + + +""".strip() + + class TestDeepSeekLLM: @pytest.fixture def llm(self): @@ -414,6 +443,101 @@ async def test_reasoning_content_is_preserved_for_text_response(self, llm, mocke assert result["reasoning_content"] == "Detailed reasoning" assert result["reasoning"] == "Detailed reasoning" + @pytest.mark.asyncio + async def test_raw_dsml_content_is_parsed_as_tool_call(self, llm, mocker): + message = SimpleNamespace( + content=_browser_extract_text_dsml(), + tool_calls=None, + reasoning_content="Need to extract browser text.", + ) + response = SimpleNamespace( + choices=[SimpleNamespace(message=message)], + usage=None, + model_dump=lambda: {"id": "deepseek-dsml"}, + ) + + 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": "Read the page"}], + tools=[_browser_extract_text_tool()], + ) + + assert result["type"] == "tool_call" + assert result["content"] is None + assert result["tool_calls"][0]["function"]["name"] == "browser_extract_text" + assert ( + result["tool_calls"][0]["function"]["arguments"] + == '{"session_id": "821:react_468b98b2"}' + ) + assert result["reasoning_content"] == "Need to extract browser text." + assert result[PROVIDER_STATE_METADATA_KEY] == { + DEEPSEEK_PROVIDER_STATE_NAMESPACE: { + DEEPSEEK_REASONING_CONTENT_STATE_KEY: "Need to extract browser text." + } + } + + @pytest.mark.asyncio + async def test_hidden_dsml_parse_tools_convert_without_provider_tools( + self, llm, mocker + ): + message = SimpleNamespace( + content=_browser_extract_text_dsml(), + tool_calls=None, + reasoning_content=None, + ) + response = SimpleNamespace( + choices=[SimpleNamespace(message=message)], + usage=None, + model_dump=lambda: {"id": "deepseek-dsml"}, + ) + + 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": "Read the page"}], + **{DEEPSEEK_DSML_PARSE_TOOLS_KWARG: [_browser_extract_text_tool()]}, + ) + + assert result["type"] == "tool_call" + assert result["tool_calls"][0]["function"]["name"] == "browser_extract_text" + call_kwargs = mock_client.chat.completions.create.call_args.kwargs + assert "tools" not in call_kwargs + assert DEEPSEEK_DSML_PARSE_TOOLS_KWARG not in call_kwargs + + @pytest.mark.asyncio + async def test_raw_dsml_content_without_tools_is_rejected(self, llm, mocker): + message = SimpleNamespace( + content=_browser_extract_text_dsml(), + tool_calls=None, + reasoning_content=None, + ) + response = SimpleNamespace( + choices=[SimpleNamespace(message=message)], + usage=None, + model_dump=lambda: {"id": "deepseek-dsml"}, + ) + + 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, + ) + + with pytest.raises(DeepSeekDSMLParseError, match="no tools were supplied"): + await llm.chat([{"role": "user", "content": "Read the page"}]) + @pytest.mark.asyncio async def test_reasoning_content_is_preserved_for_tool_calls(self, llm, mocker): tool_call = SimpleNamespace( @@ -807,6 +931,105 @@ async def stream(): end_chunk = next(chunk for chunk in chunks if chunk.type == ChunkType.END) assert end_chunk.raw["reasoning_content"] == "Think first." + @pytest.mark.asyncio + async def test_stream_chat_with_regular_tools_keeps_native_streaming( + self, llm, mocker + ): + async def stream(): + yield SimpleNamespace( + choices=[ + SimpleNamespace( + delta=SimpleNamespace( + reasoning_content=None, + content="Streaming with tools.", + tool_calls=None, + ), + finish_reason=None, + ) + ], + usage=None, + model_dump=lambda: {"id": "content"}, + ) + yield SimpleNamespace( + choices=[ + SimpleNamespace( + delta=SimpleNamespace( + reasoning_content=None, + content=None, + tool_calls=None, + ), + finish_reason="stop", + ) + ], + usage=None, + model_dump=lambda: {"id": "end"}, + ) + + mock_client = mocker.AsyncMock() + mock_client.chat.completions.create.return_value = stream() + mocker.patch( + "xagent.core.model.chat.basic.openai.AsyncOpenAI", + return_value=mock_client, + ) + mocker.patch.object( + llm, + "chat", + side_effect=AssertionError("regular tool streams must not buffer"), + ) + + chunks = [ + chunk + async for chunk in llm.stream_chat( + [{"role": "user", "content": "Read the page"}], + tools=[_browser_extract_text_tool()], + ) + ] + + assert [chunk.delta for chunk in chunks if chunk.type == ChunkType.TOKEN] == [ + "Streaming with tools." + ] + call_kwargs = mock_client.chat.completions.create.call_args.kwargs + assert call_kwargs["stream"] is True + assert call_kwargs["tools"][0]["function"]["name"] == "browser_extract_text" + + @pytest.mark.asyncio + async def test_stream_chat_hidden_dsml_parse_tools_uses_non_streaming_parser( + self, llm, mocker + ): + message = SimpleNamespace( + content=_browser_extract_text_dsml(), + tool_calls=None, + reasoning_content=None, + ) + response = SimpleNamespace( + choices=[SimpleNamespace(message=message)], + usage=None, + model_dump=lambda: {"id": "deepseek-dsml"}, + ) + + 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, + ) + + chunks = [ + chunk + async for chunk in llm.stream_chat( + [{"role": "user", "content": "Read the page"}], + **{DEEPSEEK_DSML_PARSE_TOOLS_KWARG: [_browser_extract_text_tool()]}, + ) + ] + + assert len(chunks) == 1 + assert chunks[0].type == ChunkType.TOOL_CALL + assert chunks[0].tool_calls[0]["function"]["name"] == "browser_extract_text" + call_kwargs = mock_client.chat.completions.create.call_args.kwargs + assert "stream" not in call_kwargs + assert "tools" not in call_kwargs + assert DEEPSEEK_DSML_PARSE_TOOLS_KWARG not in call_kwargs + @pytest.mark.asyncio async def test_list_available_models_returns_curated_v4_models(self): models = await DeepSeekLLM.list_available_models("test-api-key") diff --git a/tests/core/model/chat/basic/test_deepseek_dsml.py b/tests/core/model/chat/basic/test_deepseek_dsml.py new file mode 100644 index 000000000..7e8d73189 --- /dev/null +++ b/tests/core/model/chat/basic/test_deepseek_dsml.py @@ -0,0 +1,348 @@ +import json + +import pytest + +from xagent.core.model.chat.basic.deepseek_dsml import ( + DeepSeekDSMLParseError, + normalize_deepseek_dsml_response, + parse_deepseek_dsml_tool_calls, +) + + +def _tool_schema(name: str) -> dict: + return { + "type": "function", + "function": { + "name": name, + "description": f"{name} tool", + "parameters": {"type": "object", "properties": {}}, + }, + } + + +def _tools(*names: str) -> list[dict]: + return [_tool_schema(name) for name in names] + + +def _dsml_block(token: str, *, tool_name: str = "get_weather") -> str: + return f""" +<{token}tool_calls> +<{token}invoke name="{tool_name}"> +<{token}parameter name="location" string="true">Boston +<{token}parameter name="limit" string="false">5 +<{token}parameter name="options" string="false">{{"fresh":true}} + + +""".strip() + + +@pytest.mark.parametrize( + "token", + [ + "｜DSML｜", + "｜｜DSML｜｜", + "|DSML|", + "||DSML||", + ], +) +def test_parse_deepseek_dsml_tool_calls_supports_known_token_variants(token): + parsed = parse_deepseek_dsml_tool_calls( + _dsml_block(token), + tools=_tools("get_weather"), + model_name="deepseek-v4-flash", + ) + + assert parsed["content"] == "" + assert len(parsed["tool_calls"]) == 1 + tool_call = parsed["tool_calls"][0] + assert tool_call["type"] == "function" + assert tool_call["function"]["name"] == "get_weather" + assert tool_call["id"].startswith("call_dsml_") + assert json.loads(tool_call["function"]["arguments"]) == { + "location": "Boston", + "limit": 5, + "options": {"fresh": True}, + } + + +def test_parse_deepseek_dsml_tool_calls_supports_multiple_invokes(): + token = "｜｜DSML｜｜" + text = f""" +<{token}tool_calls> +<{token}invoke name="get_weather"> +<{token}parameter name="location" string="true">Boston + +<{token}invoke name="browser_extract_text"> +<{token}parameter name="session_id" string="true">821:react_468b98b2 + + +""".strip() + + parsed = parse_deepseek_dsml_tool_calls( + text, + tools=_tools("get_weather", "browser_extract_text"), + model_name="deepseek-v4-flash", + ) + + assert [call["function"]["name"] for call in parsed["tool_calls"]] == [ + "get_weather", + "browser_extract_text", + ] + assert json.loads(parsed["tool_calls"][1]["function"]["arguments"]) == { + "session_id": "821:react_468b98b2" + } + + +def test_normalize_deepseek_dsml_response_strips_markup_from_visible_content(): + response = { + "type": "text", + "content": _dsml_block("｜｜DSML｜｜"), + "raw": {"id": "deepseek-text"}, + } + + normalized, parsed = normalize_deepseek_dsml_response( + response, + tools=_tools("get_weather"), + model_name="deepseek-v4-flash", + ) + + assert parsed is True + assert normalized["type"] == "tool_call" + assert normalized["content"] is None + assert "DSML" not in json.dumps(normalized["content"]) + assert normalized["raw"] == {"id": "deepseek-text"} + + +def test_parse_deepseek_dsml_tool_calls_preserves_surrounding_prose(): + parsed = parse_deepseek_dsml_tool_calls( + f"I will inspect the page first.{_dsml_block('｜｜DSML｜｜')}Then summarize it.", + tools=_tools("get_weather"), + model_name="deepseek-v4-flash", + ) + + assert parsed["content"] == "I will inspect the page first.Then summarize it." + assert parsed["tool_calls"][0]["function"]["name"] == "get_weather" + + +def test_normalize_deepseek_dsml_response_strips_markup_when_tool_calls_exist(): + official_tool_calls = [ + { + "id": "call_1", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{"location":"Boston"}', + }, + } + ] + response = { + "type": "tool_call", + "content": f"Let me check.\n\n{_dsml_block('｜｜DSML｜｜')}", + "tool_calls": official_tool_calls, + } + + normalized, changed = normalize_deepseek_dsml_response( + response, + tools=_tools("get_weather"), + model_name="deepseek-v4-flash", + ) + + assert changed is True + assert normalized["tool_calls"] is official_tool_calls + assert normalized["content"] == "Let me check." + assert "DSML" not in normalized["content"] + + +def test_normalize_deepseek_dsml_response_strips_partial_markup_when_tool_calls_exist(): + official_tool_calls = [ + { + "id": "call_1", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{"location":"Boston"}', + }, + } + ] + response = { + "type": "tool_call", + "content": "Let me check.\n<｜｜DSML｜｜tool_calls>", + "tool_calls": official_tool_calls, + } + + normalized, changed = normalize_deepseek_dsml_response( + response, + tools=None, + model_name="deepseek-v4-flash", + ) + + assert changed is True + assert normalized["tool_calls"] is official_tool_calls + assert normalized["content"] == "Let me check." + assert "DSML" not in normalized["content"] + + +def test_normalize_deepseek_dsml_response_strips_mixed_complete_and_partial_markup(): + official_tool_calls = [ + { + "id": "call_1", + "type": "function", + "function": { + "name": "get_weather", + "arguments": '{"location":"Boston"}', + }, + } + ] + response = { + "type": "tool_call", + "content": ( + f"Let me check.\n\n{_dsml_block('｜｜DSML｜｜')}\n<｜｜DSML｜｜tool_calls>" + ), + "tool_calls": official_tool_calls, + } + + normalized, changed = normalize_deepseek_dsml_response( + response, + tools=None, + model_name="deepseek-v4-flash", + ) + + assert changed is True + assert normalized["tool_calls"] is official_tool_calls + assert normalized["content"] == "Let me check." + assert "DSML" not in normalized["content"] + + +def test_normalize_deepseek_dsml_response_strips_partial_markup_without_tool_calls(): + response = { + "type": "text", + "content": "Let me check.\n<｜｜DSML｜｜tool_calls>", + } + + normalized, changed = normalize_deepseek_dsml_response( + response, + tools=_tools("get_weather"), + model_name="deepseek-v4-flash", + ) + + assert changed is True + assert normalized["type"] == "text" + assert normalized["content"] == "Let me check." + assert "DSML" not in normalized["content"] + assert "tool_calls" not in normalized + + +def test_normalize_deepseek_dsml_response_recovers_complete_block_before_partial_markup(): + response = { + "type": "text", + "content": ( + f"Let me check.\n\n{_dsml_block('｜｜DSML｜｜')}\n<｜｜DSML｜｜tool_calls>" + ), + } + + normalized, changed = normalize_deepseek_dsml_response( + response, + tools=_tools("get_weather"), + model_name="deepseek-v4-flash", + ) + + assert changed is True + assert normalized["type"] == "tool_call" + assert normalized["content"] == "Let me check." + assert "DSML" not in normalized["content"] + assert json.loads(normalized["tool_calls"][0]["function"]["arguments"]) == { + "location": "Boston", + "limit": 5, + "options": {"fresh": True}, + } + + +def test_parse_deepseek_dsml_tool_calls_falls_back_to_string_for_invalid_json_value(): + token = "｜DSML｜" + text = f""" +<{token}tool_calls> +<{token}invoke name="get_weather"> +<{token}parameter name="query" string="false">not json + + +""".strip() + + parsed = parse_deepseek_dsml_tool_calls( + text, + tools=_tools("get_weather"), + model_name="deepseek-v4-flash", + ) + + assert json.loads(parsed["tool_calls"][0]["function"]["arguments"]) == { + "query": "not json" + } + + +def test_parse_deepseek_dsml_tool_calls_preserves_string_parameter_whitespace(): + token = "｜DSML｜" + text = f""" +<{token}tool_calls> +<{token}invoke name="write_file"> +<{token}parameter name="content" string="true"> keep exact text + + + +""".strip() + + parsed = parse_deepseek_dsml_tool_calls( + text, + tools=_tools("write_file"), + model_name="deepseek-v4-flash", + ) + + assert json.loads(parsed["tool_calls"][0]["function"]["arguments"]) == { + "content": " keep exact text\n" + } + + +def test_parse_deepseek_dsml_tool_calls_rejects_duplicate_parameters(): + token = "｜DSML｜" + text = f""" +<{token}tool_calls> +<{token}invoke name="get_weather"> +<{token}parameter name="location" string="true">Boston +<{token}parameter name="location" string="true">Paris + + +""".strip() + + with pytest.raises(DeepSeekDSMLParseError, match="duplicate DSML parameter"): + parse_deepseek_dsml_tool_calls( + text, + tools=_tools("get_weather"), + model_name="deepseek-v4-flash", + ) + + +def test_parse_deepseek_dsml_tool_calls_rejects_unknown_tools(): + with pytest.raises(DeepSeekDSMLParseError, match="unknown tool"): + parse_deepseek_dsml_tool_calls( + _dsml_block("｜DSML｜", tool_name="unknown_tool"), + tools=_tools("get_weather"), + model_name="deepseek-v4-flash", + ) + + +def test_parse_deepseek_dsml_tool_calls_rejects_incomplete_blocks(): + text = '<｜DSML｜tool_calls><｜DSML｜invoke name="get_weather">' + + with pytest.raises(DeepSeekDSMLParseError, match="incomplete"): + parse_deepseek_dsml_tool_calls( + text, + tools=_tools("get_weather"), + model_name="deepseek-v4-flash", + ) + + +def test_parse_deepseek_dsml_tool_calls_rejects_missing_request_tools(): + with pytest.raises(DeepSeekDSMLParseError, match="no tools were supplied"): + parse_deepseek_dsml_tool_calls( + _dsml_block("｜DSML｜"), + tools=None, + model_name="deepseek-v4-flash", + )