diff --git a/backend/python/vllm/backend.py b/backend/python/vllm/backend.py index 5d566285765f..41d76af413f4 100644 --- a/backend/python/vllm/backend.py +++ b/backend/python/vllm/backend.py @@ -597,12 +597,18 @@ async def _predict(self, request, context, streaming=False): # Stream the results generated_text = "" last_output = None + # When a tool parser is active for a tool-enabled request, the model's + # raw tool-call markup (e.g. ...) must not be streamed as + # delta.content — clients would otherwise see the unparsed syntax. Buffer + # the text; the final chat_delta below carries the parsed tool_calls (or + # the cleaned content), which the Go side converts to SSE deltas. + has_tool_parser = bool(self.tool_parser_cls and request.Tools) try: async for request_output in outputs: iteration_text = request_output.outputs[0].text last_output = request_output - if streaming: + if streaming and not has_tool_parser: # Remove text already sent as vllm concatenates the text from previous yields delta_iteration_text = iteration_text.removeprefix(generated_text) # Send the partial result diff --git a/backend/python/vllm/test.py b/backend/python/vllm/test.py index 25a7f54e6354..db36babf60bd 100644 --- a/backend/python/vllm/test.py +++ b/backend/python/vllm/test.py @@ -278,4 +278,78 @@ def test_embedding(self): print(err) self.fail("Embedding service failed") finally: - self.tearDown() \ No newline at end of file + self.tearDown() + + def test_streaming_tool_parser_buffering(self): + """ + When a tool parser is active and the request carries tools, streaming + must NOT emit the model's raw tool-call markup as content, and must NOT + duplicate the buffered content. Exercises _predict(streaming=True) with a + mocked engine + tool parser (no server / GPU). + """ + import sys, os, asyncio + from types import SimpleNamespace + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + from backend import BackendServicer + + def make_generate(chunks): + async def gen(*a, **k): + for t in chunks: + yield SimpleNamespace( + outputs=[SimpleNamespace(text=t, token_ids=[1], logprobs=None)], + prompt_token_ids=[0], + ) + return lambda *a, **k: gen() + + def parser_cls(called, content, calls): + class _P: + def __init__(self, tokenizer, tools=None): + pass + def extract_tool_calls(self, c, request=None): + return SimpleNamespace(tools_called=called, content=content, tool_calls=calls) + return _P + + def collect(servicer, req): + async def run(): + return [r async for r in servicer._predict(req, None, streaming=True)] + return asyncio.run(run()) + + def contents(replies): + return [cd.content for r in replies for cd in r.chat_deltas if cd.content] + + tools_json = '[{"type":"function","function":{"name":"calc"}}]' + + # Case 1: model emits a tool call -> no raw markup as content, tool_call present. + s = BackendServicer() + s.reasoning_parser_cls = None + s.tokenizer = None + s.llm = SimpleNamespace(generate=make_generate([ + '\n{"name": "calc"', + '\n{"name": "calc", "arguments": {"x": 1}}\n', + ])) + call = SimpleNamespace(id="call_1", function=SimpleNamespace(name="calc", arguments='{"x": 1}')) + s.tool_parser_cls = parser_cls(True, "", [call]) + req = backend_pb2.PredictOptions(Prompt="x", Tools=tools_json) + replies = collect(s, req) + self.assertFalse( + any(" content once. + s2 = BackendServicer() + s2.reasoning_parser_cls = None + s2.tokenizer = None + s2.llm = SimpleNamespace(generate=make_generate([ + "The capital ", + "The capital of France is Paris.", + ])) + s2.tool_parser_cls = parser_cls(False, "", []) + req2 = backend_pb2.PredictOptions(Prompt="x", Tools=tools_json) + joined = "".join(contents(collect(s2, req2))) + self.assertEqual( + joined.count("The capital of France is Paris."), 1, + f"buffered content was duplicated: {joined!r}", + ) \ No newline at end of file