diff --git a/lightllm/server/function_call_parser.py b/lightllm/server/function_call_parser.py index dfcb2f8d9e..99a6c5833d 100644 --- a/lightllm/server/function_call_parser.py +++ b/lightllm/server/function_call_parser.py @@ -1759,6 +1759,7 @@ def __init__(self): r"|(?=)|$)", re.DOTALL ) self._normal_text_buffer = "" + self._current_tool_name: Optional[str] = None def has_tool_call(self, text: str) -> bool: return " Optional func_name = function_str[:end_index].strip() tool_indices = self._get_tool_indices(tools) - if func_name not in tool_indices: + if not func_name or (ENABLE_TOOL_NAME_CHECK and func_name not in tool_indices): logger.warning(f"Model attempted to call undefined function: {func_name}") return None @@ -1843,11 +1844,27 @@ def _parse_function_call(self, function_str: str, tools: List[Tool]) -> Optional param_dict[param_name] = self._convert_param_value(param_value, param_name, param_config, func_name) return ToolCallItem( - tool_index=tool_indices[func_name], + tool_index=tool_indices.get(func_name, -1), name=func_name, parameters=json.dumps(param_dict, ensure_ascii=False), ) + def _get_incomplete_function_name(self, text: str) -> Optional[str]: + """Return a complete, valid function name from an unfinished tool-call block.""" + function_start = text.find("", name_start) + if name_end == -1: + return None + + function_name = text[name_start:name_end].strip() + if not function_name or (ENABLE_TOOL_NAME_CHECK and function_name not in self._tool_indices): + return None + return function_name + def _build_partial_arguments_json(self, func_name: str, partial_body: str, tools: List[Tool]) -> Optional[str]: """Build the current argument JSON from a partial XML tool-call body.""" param_matches = self.parameter_regex.findall(partial_body) @@ -1943,6 +1960,31 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami eot_pos = current_text.find(self.eot_token) if eot_pos == -1: + if not self.current_tool_name_sent: + function_name = self._get_incomplete_function_name(current_text) + if function_name is None: + return StreamingParseResult(normal_text=normal_text, calls=calls) + + if self.current_tool_id == -1: + self.current_tool_id = 0 + self.current_tool_name_sent = True + self._current_tool_name = function_name + calls.append( + ToolCallItem( + tool_index=self.current_tool_id, + name=function_name, + parameters="", + ) + ) + else: + # Keep the connection active for every decoded chunk without + # exposing a partial XML value as malformed JSON arguments. + calls.append( + ToolCallItem( + tool_index=self.current_tool_id, + parameters="", + ) + ) return StreamingParseResult(normal_text=normal_text, calls=calls) complete_block = current_text[: eot_pos + len(self.eot_token)] @@ -1956,9 +1998,17 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami item = self._parse_function_call(func_str, tools) if item: item.tool_index = self.current_tool_id + if self.current_tool_name_sent and item.name == self._current_tool_name: + # The head was already emitted. Send the complete arguments as + # one continuation delta and do not repeat the function name. + item.name = None + self.current_tool_name_sent = False + self._current_tool_name = None calls.append(item) self.current_tool_id += 1 + self.current_tool_name_sent = False + self._current_tool_name = None self._buffer = current_text[eot_pos + len(self.eot_token) :].lstrip() diff --git a/unit_tests/server/test_qwen3_coder_stream_fc.py b/unit_tests/server/test_qwen3_coder_stream_fc.py new file mode 100644 index 0000000000..bd8ff6f08c --- /dev/null +++ b/unit_tests/server/test_qwen3_coder_stream_fc.py @@ -0,0 +1,151 @@ +"""Coverage for buffered Qwen3-Coder streaming keepalive deltas.""" + +import json +from types import SimpleNamespace + +from lightllm.server.api_cli import make_argument_parser +from lightllm.server.api_models import Function, Tool +from lightllm.server.function_call_parser import FunctionCallParser, Qwen3CoderDetector + + +def _tool() -> Tool: + return Tool( + type="function", + function=Function( + name="write_file", + description="", + parameters={ + "type": "object", + "properties": {"content": {"type": "string"}}, + }, + ), + ) + + +def test_qwen3_coder_selects_buffered_parser(): + assert FunctionCallParser.ToolCallParserEnum["qwen3_coder"] is Qwen3CoderDetector + assert "qwen3_coder_legacy" not in FunctionCallParser.ToolCallParserEnum + + args = make_argument_parser().parse_args(["--tool_call_parser", "qwen3_coder"]) + assert args.tool_call_parser == "qwen3_coder" + + +def test_qwen3_coder_emits_one_tool_delta_for_every_decode_after_the_name(): + parser = FunctionCallParser([_tool()], "qwen3_coder") + + normal_text, calls = parser.parse_stream_chunk("\n\n\n") + assert normal_text == "" + assert len(calls) == 1 + assert calls[0].tool_index == 0 + assert calls[0].name == "write_file" + assert calls[0].parameters == "" + + streamed_calls = list(calls) + for chunk in ("a", "b", "c" * 100_000): + normal_text, calls = parser.parse_stream_chunk(chunk) + assert normal_text == "" + assert len(calls) == 1 + assert calls[0].tool_index == 0 + assert calls[0].name is None + assert calls[0].parameters == "" + streamed_calls.extend(calls) + + normal_text, calls = parser.parse_stream_chunk("\n\n\n") + assert normal_text == "" + assert len(calls) == 1 + assert calls[0].tool_index == 0 + assert calls[0].name is None + assert json.loads(calls[0].parameters) == {"content": "ab" + "c" * 100_000} + streamed_calls.extend(calls) + + assert [call.name for call in streamed_calls].count("write_file") == 1 + assert json.loads("".join(call.parameters for call in streamed_calls)) == {"content": "ab" + "c" * 100_000} + + +def test_qwen3_coder_accepts_an_undefined_tool_when_name_check_is_disabled(monkeypatch): + import lightllm.server.function_call_parser as parser_module + + monkeypatch.setattr(parser_module, "ENABLE_TOOL_NAME_CHECK", False) + parser = FunctionCallParser([_tool()], "qwen3_coder") + + _, calls = parser.parse_stream_chunk("\n\n") + assert len(calls) == 1 + assert calls[0].name == "unknown" + assert calls[0].parameters == "" + + _, calls = parser.parse_stream_chunk("x\n\n") + assert len(calls) == 1 + assert calls[0].name is None + assert json.loads(calls[0].parameters) == {"value": "x"} + + +def test_qwen3_coder_rejects_an_undefined_tool_when_name_check_is_enabled(monkeypatch): + import lightllm.server.function_call_parser as parser_module + + monkeypatch.setattr(parser_module, "ENABLE_TOOL_NAME_CHECK", True) + parser = FunctionCallParser([_tool()], "qwen3_coder") + + _, calls = parser.parse_stream_chunk("\n\n") + assert calls == [] + + _, calls = parser.parse_stream_chunk("x\n\n") + assert calls == [] + + +def test_api_tool_parser_keeps_its_two_item_return_contract(): + from lightllm.server.api_openai import _process_tools_stream + + request = SimpleNamespace(tools=[_tool()]) + parser_dict = {0: FunctionCallParser(request.tools, "qwen3_coder")} + + result = _process_tools_stream( + 0, + "\n\n\n", + parser_dict, + request, + ) + + assert len(result) == 2 + normal_text, calls = result + assert normal_text == "" + assert len(calls) == 1 + assert calls[0].name == "write_file" + + +def test_empty_arguments_are_preserved_in_the_sse_payload(): + from lightllm.server.api_models import ( + ChatCompletionStreamResponse, + ChatCompletionStreamResponseChoice, + DeltaMessage, + FunctionResponse, + ToolCall, + ) + from lightllm.server.api_openai import _serialize_sse_chunk + + chunk = ChatCompletionStreamResponse( + id="chatcmpl-test", + created=0, + model="test", + choices=[ + ChatCompletionStreamResponseChoice( + index=0, + delta=DeltaMessage( + tool_calls=[ + ToolCall( + index=0, + function=FunctionResponse(arguments=""), + ) + ] + ), + finish_reason=None, + ) + ], + ) + + payload = json.loads(_serialize_sse_chunk(chunk, ("logprobs", "token_ids", "finish_reason"))) + function_delta = payload["choices"][0]["delta"]["tool_calls"][0]["function"] + assert function_delta["arguments"] == ""