From c34e0346e66583d67c59d5e9c247b2fb6883f1ae Mon Sep 17 00:00:00 2001 From: pos-ei-don <1822533+pos-ei-don@users.noreply.github.com> Date: Mon, 15 Jun 2026 17:46:49 +0000 Subject: [PATCH 1/6] fix(vllm): don't stream raw tool-call markup as content when a tool parser is active MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a tool_parser is configured and the request carries tools, the streaming loop emitted every text delta as delta.content — including the model's raw tool-call markup (e.g. ...) — because extract_tool_calls only runs on the full output after the stream. Clients streaming a tool call therefore saw the unparsed tool-call syntax as assistant content. Buffer the text while a tool parser is active for the request; the existing end-of-stream chat_delta already carries the parsed tool_calls (or the cleaned content), which the Go side converts to SSE deltas. Non-tool-parser streaming is unchanged. Add a server-less regression test covering both the tool-call case (no raw markup leaked as content) and the plain-text case (content delivered exactly once — guards against double-emitting the buffered content). Signed-off-by: pos-ei-don <1822533+pos-ei-don@users.noreply.github.com> --- backend/python/vllm/backend.py | 8 +++- backend/python/vllm/test.py | 76 +++++++++++++++++++++++++++++++++- 2 files changed, 82 insertions(+), 2 deletions(-) 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 From 0132e4157ecbe284a431b10cebef035819acaf2f Mon Sep 17 00:00:00 2001 From: pos-ei-don <1822533+pos-ei-don@users.noreply.github.com> Date: Mon, 15 Jun 2026 20:38:00 +0200 Subject: [PATCH 2/6] test(vllm): add expectedFailure test for progressive streaming with tool parser (Case 3, #582) Signed-off-by: pos-ei-don <1822533+pos-ei-don@users.noreply.github.com> --- backend/python/vllm/test.py | 84 +++++++++++++++++++++++++++++++++++++ 1 file changed, 84 insertions(+) diff --git a/backend/python/vllm/test.py b/backend/python/vllm/test.py index db36babf60bd..9003c9ade702 100644 --- a/backend/python/vllm/test.py +++ b/backend/python/vllm/test.py @@ -352,4 +352,88 @@ def contents(replies): self.assertEqual( joined.count("The capital of France is Paris."), 1, f"buffered content was duplicated: {joined!r}", + ) + @unittest.expectedFailure + def test_streaming_tool_parser_progressive_plain_text(self): + """ + Case 3 (TDD — currently fails, defines acceptance criterion for follow-up, see #582): + + When a tool parser is active but the model returns plain text (no tool call), + and the parser implements extract_tool_calls_streaming, tokens should be emitted + progressively — not held until the final chunk. + + Proposed interface: + extract_tool_calls_streaming(delta: str, request=None) + -> SimpleNamespace(is_tool_call_token=bool, content=str) + + Fails on current code because has_tool_parser=True suppresses all intermediate + deltas. Will pass after Option A or B from the follow-up (Issue #582). + """ + 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() + + class StreamingAwareParser: + def __init__(self, tokenizer, tools=None): + pass + + def extract_tool_calls(self, c, request=None): + return SimpleNamespace(tools_called=False, content=c, tool_calls=[]) + + def extract_tool_calls_streaming(self, delta, request=None): + # Plain-text delta — not tool-call markup, pass through as content. + return SimpleNamespace(is_tool_call_token=False, content=delta) + + def collect(servicer, req): + async def run(): + return [r async for r in servicer._predict(req, None, streaming=True)] + return asyncio.run(run()) + + # vLLM yields cumulative text per iteration + s = BackendServicer() + s.reasoning_parser_cls = None + s.tokenizer = None + s.llm = SimpleNamespace(generate=make_generate([ + "Paris ", + "Paris is ", + "Paris is the capital of France.", + ])) + s.tool_parser_cls = StreamingAwareParser + + tools_json = '[{"type":"function","function":{"name":"calc"}}]' + req = backend_pb2.PredictOptions(Prompt="x", Tools=tools_json) + replies = collect(s, req) + + # Key assertion: intermediate replies (before the final chunk) must carry content. + # Currently fails because has_tool_parser=True suppresses all intermediate deltas. + intermediate_content = [ + cd.content + for r in replies[:-1] + for cd in r.chat_deltas + if cd.content + ] + self.assertTrue( + len(intermediate_content) > 0, + "Plain-text response not streamed progressively — " + "all content arrived in the final chunk. " + "Fix: use extract_tool_calls_streaming when available (Option A).", + ) + + # No duplication: assembled content equals the full text exactly once. + assembled = "".join( + cd.content for r in replies for cd in r.chat_deltas if cd.content + ) + self.assertEqual( + assembled, "Paris is the capital of France.", + f"Content wrong or duplicated: {assembled!r}", ) \ No newline at end of file From 4383419850aa2de150229de7816fd13ef2f31520 Mon Sep 17 00:00:00 2001 From: pos-ei-don <1822533+pos-ei-don@users.noreply.github.com> Date: Mon, 15 Jun 2026 20:44:26 +0200 Subject: [PATCH 3/6] =?UTF-8?q?test(vllm):=20add=20Cases=204+5=20=E2=80=94?= =?UTF-8?q?=20marker=20split=20across=20chunks=20+=20false-positive=20pref?= =?UTF-8?q?ix=20(TDD,=20Option=20B=20state=20machine,=20#582)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: pos-ei-don <1822533+pos-ei-don@users.noreply.github.com> --- backend/python/vllm/test.py | 156 ++++++++++++++++++++++++++++++++++++ 1 file changed, 156 insertions(+) diff --git a/backend/python/vllm/test.py b/backend/python/vllm/test.py index 9003c9ade702..913287a7e7ce 100644 --- a/backend/python/vllm/test.py +++ b/backend/python/vllm/test.py @@ -436,4 +436,160 @@ async def run(): self.assertEqual( assembled, "Paris is the capital of France.", f"Content wrong or duplicated: {assembled!r}", + ) + @unittest.expectedFailure + def test_streaming_tool_parser_marker_split_across_chunks(self): + """ + Case 4 (TDD — Option B state machine, marker spans chunk boundary): + + vLLM may yield "Some text \\n{...}\\n" in the next. The state machine must hold + the incomplete prefix ("" is split: first chunk ends with "..." + full_tool = '\n{"name": "calc", "arguments": {"x": 1}}\n' + s = BackendServicer() + s.reasoning_parser_cls = None + s.tokenizer = None + s.llm = SimpleNamespace(generate=make_generate([ + "Some text to help." The state machine + must flush the uncertainty buffer as content once the prefix is disconfirmed. + + Expected: + - All text streams through as content (no tool call). + - Assembled content equals the full sentence exactly once (no duplication). + - No tool_calls in the final reply. + """ + 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() + + class StateMachineParser: + def __init__(self, tokenizer, tools=None): + pass + + def extract_tool_calls(self, c, request=None): + return SimpleNamespace(tools_called=False, content=c, tool_calls=[]) + + def collect(servicer, req): + async def run(): + return [r async for r in servicer._predict(req, None, streaming=True)] + return asyncio.run(run()) + + # "" but completes as "" + full_text = "Let me use a to help." + s = BackendServicer() + s.reasoning_parser_cls = None + s.tokenizer = None + s.llm = SimpleNamespace(generate=make_generate([ + "Let me use a Date: Mon, 15 Jun 2026 22:45:17 +0200 Subject: [PATCH 4/6] feat(vllm): progressive streaming via parser.extract_tool_calls_streaming MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a tool parser is active for a tool-enabled streaming request, #10346 buffers the entire generation and surfaces it on the final chunk to prevent raw tool-call markup from leaking as delta.content. This is correct but turns the request into effectively non-streaming for plain-text responses — the client sees nothing until the model stops. Every concrete tool parser shipped with vLLM 0.23+ already implements extract_tool_calls_streaming (Granite4, Qwen3Coder, DeepSeekV31, Jamba, Ernie45, Hermes2Pro, llama3_json, mistral, …). Use it: instantiate the parser before the streaming loop and call its streaming method per delta, emitting DeltaMessage(content=…) or DeltaMessage(tool_calls=[…]) when the parser is ready. Falls back to the existing #10346 buffer path when: - the parser does not have extract_tool_calls_streaming, OR - extract_tool_calls_streaming raises mid-stream (logged, the rest of the request finishes via post-loop extract_tool_calls). Tests (TestStreamingToolParser): 1. Buffer path: no markup leaked, no content duplication 2. Native streaming: plain-text response streams progressively 3. Native streaming: tool_call structured, no markup leaked 4. Native streaming exception → graceful fallback, no markup, no crash 5. No tool parser → unchanged per-delta content stream E2E verified against qwen3_coder on vLLM 0.23.0 (NVIDIA GB10 / arm64 / CUDA 13). Signed-off-by: pos-ei-don <1822533+pos-ei-don@users.noreply.github.com> --- backend/python/vllm/backend.py | 169 +++++++++++-- backend/python/vllm/test.py | 421 ++++++++++++++------------------- 2 files changed, 327 insertions(+), 263 deletions(-) diff --git a/backend/python/vllm/backend.py b/backend/python/vllm/backend.py index 41d76af413f4..0a84e47165ac 100644 --- a/backend/python/vllm/backend.py +++ b/backend/python/vllm/backend.py @@ -596,29 +596,124 @@ async def _predict(self, request, context, streaming=False): # Stream the results generated_text = "" + generated_token_ids: list[int] = [] 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. + + # Tool-parsing strategy decision (made once, before the loop): + # + # When a tool parser is active, the model's raw tool-call markup + # (e.g. ...) must not be streamed verbatim as delta.content + # — clients would see the unparsed syntax. Two paths: + # + # (A) native streaming via parser.extract_tool_calls_streaming. All + # concrete tool parsers shipped with vLLM 0.23+ implement this + # (Granite4, Qwen3Coder, DeepSeekV31, Jamba, Ernie45, Hermes, + # llama3_json, mistral, …). The parser decides per-delta whether + # to emit content or suppress tool-call markup, and emits a + # structured DeltaMessage(tool_calls=[...]) when a call is ready. + # (B) buffer fallback — used only when the parser surprisingly lacks + # the streaming method or it raises mid-stream. The post-loop + # extract_tool_calls assembles the final chat_delta. Same correctness + # guarantee as a non-streaming response, at the cost of a delayed + # final chunk. has_tool_parser = bool(self.tool_parser_cls and request.Tools) + tp_instance = None + tp_request = None + native_streaming = False + native_streaming_error = False + if has_tool_parser: + try: + tools_for_parser = json.loads(request.Tools) + except json.JSONDecodeError: + tools_for_parser = [] + try: + tp_instance = self.tool_parser_cls(self.tokenizer, tools=tools_for_parser) + except TypeError: + tp_instance = self.tool_parser_cls(self.tokenizer) + # Build a minimal ChatCompletionRequest so the streaming method + # sees the tools list. We do not need any other request fields — + # parsers only read .tools (and sometimes .tool_choice, which we + # leave at default). + try: + from vllm.entrypoints.openai.chat_completion.protocol import ( + ChatCompletionRequest as _CCR, + ) + tp_request = _CCR( + model="local", + messages=[{"role": "user", "content": ""}], + tools=tools_for_parser or None, + ) + except Exception as e: + print(f"Could not build ChatCompletionRequest for streaming parser: {e}", + file=sys.stderr) + tp_request = None + native_streaming = ( + tp_request is not None + and hasattr(tp_instance, "extract_tool_calls_streaming") + ) + try: async for request_output in outputs: iteration_text = request_output.outputs[0].text last_output = request_output - if streaming and not has_tool_parser: - # Remove text already sent as vllm concatenates the text from previous yields + if streaming: delta_iteration_text = iteration_text.removeprefix(generated_text) - # Send the partial result - yield backend_pb2.Reply( - message=bytes(delta_iteration_text, encoding='utf-8'), - chat_deltas=[backend_pb2.ChatDelta(content=delta_iteration_text)], - ) - - # Keep track of text generated + new_token_ids = list(request_output.outputs[0].token_ids) + delta_token_ids = new_token_ids[len(generated_token_ids):] + + if not has_tool_parser: + # Plain streaming — unchanged from pre-tool-parser path. + yield backend_pb2.Reply( + message=bytes(delta_iteration_text, encoding='utf-8'), + chat_deltas=[backend_pb2.ChatDelta(content=delta_iteration_text)], + ) + elif native_streaming and not native_streaming_error: + # (A) Native vLLM extract_tool_calls_streaming. + try: + msg = tp_instance.extract_tool_calls_streaming( + previous_text=generated_text, + current_text=iteration_text, + delta_text=delta_iteration_text, + previous_token_ids=generated_token_ids, + current_token_ids=new_token_ids, + delta_token_ids=delta_token_ids, + request=tp_request, + ) + except Exception as e: + print(f"Streaming tool parser error (falling back to " + f"buffer for the rest of the stream): {e}", + file=sys.stderr) + native_streaming_error = True + msg = None + if msg is not None: + tc_protos = [] + for tc in (msg.tool_calls or []): + fn = tc.function or None + tc_protos.append(backend_pb2.ToolCallDelta( + index=tc.index, + id=tc.id or "", + name=(fn.name if fn and fn.name else "") or "", + arguments=(fn.arguments if fn and fn.arguments else "") or "", + )) + cd_kwargs = {} + if msg.content: + cd_kwargs["content"] = msg.content + if msg.reasoning: + cd_kwargs["reasoning_content"] = msg.reasoning + if tc_protos: + cd_kwargs["tool_calls"] = tc_protos + if cd_kwargs: + yield backend_pb2.Reply( + message=bytes(msg.content or "", encoding='utf-8'), + chat_deltas=[backend_pb2.ChatDelta(**cd_kwargs)], + ) + # (B) buffer fallback — emit nothing during the stream. + # The post-loop extract_tool_calls block builds the final chunk. + + # Keep track of text + token_ids generated generated_text = iteration_text + generated_token_ids = list(request_output.outputs[0].token_ids) finally: await outputs.aclose() @@ -643,16 +738,19 @@ async def _predict(self, request, context, streaming=False): except Exception as e: print(f"Reasoning parser error: {e}", file=sys.stderr) - if self.tool_parser_cls and request.Tools: + # When (A) native streaming ran cleanly, per-delta yields above already + # delivered everything — do NOT extract again on the full text or we'd + # duplicate content/tool_calls into the final chunk. + if has_tool_parser and not (native_streaming and not native_streaming_error): try: - tools = json.loads(request.Tools) - # Some concrete parsers only accept the tokenizer; only the - # abstract base declares the tools kwarg. Try with tools first, - # fall back to tokenizer-only. - try: - tp = self.tool_parser_cls(self.tokenizer, tools=tools) - except TypeError: - tp = self.tool_parser_cls(self.tokenizer) + tp = tp_instance + if tp is None: + # Defensive: tp_instance build failed earlier; reconstruct. + tools = json.loads(request.Tools) + try: + tp = self.tool_parser_cls(self.tokenizer, tools=tools) + except TypeError: + tp = self.tool_parser_cls(self.tokenizer) info = tp.extract_tool_calls(content, request=None) if info.tools_called: content = info.content or "" @@ -665,6 +763,10 @@ async def _predict(self, request, context, streaming=False): )) except Exception as e: print(f"Tool parser error: {e}", file=sys.stderr) + elif native_streaming and not native_streaming_error: + # Per-delta path already emitted content + tool_calls; the final + # chat_delta should carry only metadata (token counts, logprobs). + content = "" # Extract token counts prompt_tokens = 0 @@ -704,7 +806,26 @@ async def _predict(self, request, context, streaming=False): ) if streaming: - # Final chunk with structured data + # Final chunk with structured data. + # + # If we used the buffer fallback (has_tool_parser=True AND native + # streaming did NOT run cleanly) and the parser found no tool call, + # flush the buffered content as ONE content delta — and clear the + # final chat_delta's content so the metadata chunk does not repeat + # what we just sent. This is the plain-text-with-tool-parser path. + buffered_fallback = ( + has_tool_parser + and not (native_streaming and not native_streaming_error) + ) + if buffered_fallback and not tool_calls_proto and content: + yield backend_pb2.Reply( + message=bytes(content, encoding='utf-8'), + chat_deltas=[backend_pb2.ChatDelta(content=content)], + ) + chat_delta = backend_pb2.ChatDelta( + reasoning_content=reasoning_content, + tool_calls=tool_calls_proto, + ) yield backend_pb2.Reply( message=b"", prompt_tokens=prompt_tokens, diff --git a/backend/python/vllm/test.py b/backend/python/vllm/test.py index 913287a7e7ce..d00595f016fc 100644 --- a/backend/python/vllm/test.py +++ b/backend/python/vllm/test.py @@ -280,316 +280,259 @@ def test_embedding(self): finally: 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 + +class TestStreamingToolParser(unittest.TestCase): + """ + Server-less unit tests for the streaming + tool-parser machinery in + BackendServicer._predict. These tests instantiate BackendServicer + directly and mock the vLLM engine + tool parser, so they do not need + a GPU, a model, or a running gRPC server. Kept in a separate class to + avoid the parent setUp() which spawns a subprocess. + + Covers #582 (follow-up to #10346): + 1. Markup-leak prevention with a non-streaming parser (buffer fallback) + 2. No content duplication on the plain-text path with the buffer fallback + 3. Native streaming progressive plain-text emission + 4. Native streaming structured tool_call, no markup leak + 5. Parser exception → graceful fallback to buffer, still no markup + 6. No-tool-parser regression: unchanged per-delta content stream + """ + + @staticmethod + def _make_generate(chunks): + """Build a fake vLLM engine.generate that yields cumulative chunks.""" from types import SimpleNamespace + async def gen(*a, **k): + for i, t in enumerate(chunks): + yield SimpleNamespace( + outputs=[SimpleNamespace( + text=t, + token_ids=list(range(i + 1)), + logprobs=None, + )], + prompt_token_ids=[0], + ) + return lambda *a, **k: gen() + + @staticmethod + def _collect(servicer, req): + import asyncio + async def run(): + return [r async for r in servicer._predict(req, None, streaming=True)] + return asyncio.run(run()) + + def _new_servicer(self): + import sys, os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from backend import BackendServicer + s = BackendServicer() + s.reasoning_parser_cls = None + s.tool_parser_cls = None + s.tokenizer = None + return s - 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() + # ── Case 1+2: parser without streaming method → buffer fallback ── + def test_buffer_path_no_markup_no_duplication(self): + from types import SimpleNamespace - def parser_cls(called, content, calls): + def parser_cls(called, content_text, calls): class _P: def __init__(self, tokenizer, tools=None): pass + # NOTE: NO extract_tool_calls_streaming → takes the buffer path def extract_tool_calls(self, c, request=None): - return SimpleNamespace(tools_called=called, content=content, tool_calls=calls) + return SimpleNamespace( + tools_called=called, content=content_text, 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()) + tools_json = '[{"type":"function","function":{"name":"calc","parameters":{}}}]' - 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([ + # Tool-call case: no raw markup in any delta.content + s = self._new_servicer() + s.llm = SimpleNamespace(generate=self._make_generate([ '\n{"name": "calc"', '\n{"name": "calc", "arguments": {"x": 1}}\n', ])) - call = SimpleNamespace(id="call_1", function=SimpleNamespace(name="calc", arguments='{"x": 1}')) + 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) + replies = self._collect(s, req) + contents = [cd.content for r in replies for cd in r.chat_deltas if cd.content] self.assertFalse( - any(" content once. - s2 = BackendServicer() - s2.reasoning_parser_cls = None - s2.tokenizer = None - s2.llm = SimpleNamespace(generate=make_generate([ + # Plain-text-with-tools case: full content delivered exactly once + s2 = self._new_servicer() + s2.llm = SimpleNamespace(generate=self._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))) + joined = "".join( + cd.content for r in self._collect(s2, req2) + for cd in r.chat_deltas if cd.content + ) self.assertEqual( joined.count("The capital of France is Paris."), 1, - f"buffered content was duplicated: {joined!r}", + f"buffered content duplicated: {joined!r}", ) - @unittest.expectedFailure - def test_streaming_tool_parser_progressive_plain_text(self): - """ - Case 3 (TDD — currently fails, defines acceptance criterion for follow-up, see #582): - - When a tool parser is active but the model returns plain text (no tool call), - and the parser implements extract_tool_calls_streaming, tokens should be emitted - progressively — not held until the final chunk. - - Proposed interface: - extract_tool_calls_streaming(delta: str, request=None) - -> SimpleNamespace(is_tool_call_token=bool, content=str) - Fails on current code because has_tool_parser=True suppresses all intermediate - deltas. Will pass after Option A or B from the follow-up (Issue #582). - """ - import sys, os, asyncio + # ── Case 3: native streaming, progressive plain text ── + def test_native_streaming_progressive_plain_text(self): 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() + class _DeltaMsg: + def __init__(self, content=None, reasoning=None, tool_calls=None): + self.content = content + self.reasoning = reasoning + self.tool_calls = tool_calls or [] - class StreamingAwareParser: + class StreamingParser: def __init__(self, tokenizer, tools=None): pass - def extract_tool_calls(self, c, request=None): - return SimpleNamespace(tools_called=False, content=c, tool_calls=[]) - - def extract_tool_calls_streaming(self, delta, request=None): - # Plain-text delta — not tool-call markup, pass through as content. - return SimpleNamespace(is_tool_call_token=False, content=delta) - - def collect(servicer, req): - async def run(): - return [r async for r in servicer._predict(req, None, streaming=True)] - return asyncio.run(run()) - - # vLLM yields cumulative text per iteration - s = BackendServicer() - s.reasoning_parser_cls = None - s.tokenizer = None - s.llm = SimpleNamespace(generate=make_generate([ + # Should NOT be called when native streaming runs successfully. + raise AssertionError("extract_tool_calls invoked on native-streaming path") + def extract_tool_calls_streaming( + self, previous_text, current_text, delta_text, + previous_token_ids, current_token_ids, delta_token_ids, request, + ): + if not delta_text: + return None + return _DeltaMsg(content=delta_text) + + s = self._new_servicer() + s.llm = SimpleNamespace(generate=self._make_generate([ "Paris ", "Paris is ", "Paris is the capital of France.", ])) - s.tool_parser_cls = StreamingAwareParser - - tools_json = '[{"type":"function","function":{"name":"calc"}}]' - req = backend_pb2.PredictOptions(Prompt="x", Tools=tools_json) - replies = collect(s, req) + s.tool_parser_cls = StreamingParser + req = backend_pb2.PredictOptions( + Prompt="x", + Tools='[{"type":"function","function":{"name":"calc","parameters":{}}}]', + ) + replies = self._collect(s, req) - # Key assertion: intermediate replies (before the final chunk) must carry content. - # Currently fails because has_tool_parser=True suppresses all intermediate deltas. intermediate_content = [ - cd.content - for r in replies[:-1] - for cd in r.chat_deltas - if cd.content + cd.content for r in replies[:-1] for cd in r.chat_deltas if cd.content ] self.assertTrue( len(intermediate_content) > 0, - "Plain-text response not streamed progressively — " - "all content arrived in the final chunk. " - "Fix: use extract_tool_calls_streaming when available (Option A).", + "Plain-text response not streamed progressively (native streaming inactive?)", ) - - # No duplication: assembled content equals the full text exactly once. assembled = "".join( cd.content for r in replies for cd in r.chat_deltas if cd.content ) self.assertEqual( assembled, "Paris is the capital of France.", - f"Content wrong or duplicated: {assembled!r}", + f"Assembled content wrong: {assembled!r}", ) - @unittest.expectedFailure - def test_streaming_tool_parser_marker_split_across_chunks(self): - """ - Case 4 (TDD — Option B state machine, marker spans chunk boundary): - - vLLM may yield "Some text \\n{...}\\n" in the next. The state machine must hold - the incomplete prefix ("" is split: first chunk ends with "..." - full_tool = '\n{"name": "calc", "arguments": {"x": 1}}\n' - s = BackendServicer() - s.reasoning_parser_cls = None - s.tokenizer = None - s.llm = SimpleNamespace(generate=make_generate([ - "Some text " in current_text and not self._emitted: + self._emitted = True + fn = SimpleNamespace(name="calc", arguments='{"x": 1}') + tc = SimpleNamespace(id="call_1", type="function", index=0, function=fn) + return _DeltaMsg(tool_calls=[tc]) + return None + + s = self._new_servicer() + s.llm = SimpleNamespace(generate=self._make_generate([ + '\n', + '\n{"name": "calc"', + '\n{"name": "calc", "arguments": {"x": 1}}\n', ])) - s.tool_parser_cls = StateMachineParser - - tools_json = '[{"type":"function","function":{"name":"calc"}}]' - req = backend_pb2.PredictOptions(Prompt="x", Tools=tools_json) - replies = collect(s, req) - - intermediate_content = [ - cd.content - for r in replies[:-1] - for cd in r.chat_deltas - if cd.content - ] - - # "Some text " must have streamed through before the marker was seen - self.assertTrue( - any("Some text" in c for c in intermediate_content), - "Content before the marker was not streamed progressively.", + s.tool_parser_cls = _ToolCallStreamer + req = backend_pb2.PredictOptions( + Prompt="x", + Tools='[{"type":"function","function":{"name":"calc","parameters":{}}}]', ) + replies = self._collect(s, req) - # The marker prefix "" in c for c in contents), + f"markup leaked as content: {contents!r}", ) + names = [tc.name for r in replies for cd in r.chat_deltas for tc in cd.tool_calls if tc.name] + args = [tc.arguments for r in replies for cd in r.chat_deltas for tc in cd.tool_calls if tc.arguments] + self.assertIn("calc", names, f"tool_call name missing; got {names!r}") + self.assertIn('{"x": 1}', args, f"tool_call args missing; got {args!r}") - # Final chunk must carry the parsed tool_call - names = [tc.name for r in replies for cd in r.chat_deltas for tc in cd.tool_calls] - self.assertIn("calc", names, "Structured tool_call not present in final chunk.") - - @unittest.expectedFailure - def test_streaming_tool_parser_false_positive_marker(self): - """ - Case 5 (TDD — Option B state machine, false-positive marker prefix): - - The model writes text that starts like a tool-call marker but is not one, - e.g. "Let me use a to help." The state machine - must flush the uncertainty buffer as content once the prefix is disconfirmed. - - Expected: - - All text streams through as content (no tool call). - - Assembled content equals the full sentence exactly once (no duplication). - - No tool_calls in the final reply. - """ - import sys, os, asyncio + # ── Case 5: parser exception → fallback to buffer, no leak ── + def test_native_streaming_parser_exception_falls_back_to_buffer(self): from types import SimpleNamespace - sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - from backend import BackendServicer + call = SimpleNamespace(id="call_1", + function=SimpleNamespace(name="calc", arguments='{"x": 1}')) - 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() - - class StateMachineParser: + class _BrokenStreamer: def __init__(self, tokenizer, tools=None): pass - def extract_tool_calls(self, c, request=None): - return SimpleNamespace(tools_called=False, content=c, tool_calls=[]) - - def collect(servicer, req): - async def run(): - return [r async for r in servicer._predict(req, None, streaming=True)] - return asyncio.run(run()) + return SimpleNamespace(tools_called=True, content="", tool_calls=[call]) + def extract_tool_calls_streaming(self, *a, **kw): + raise RuntimeError("simulated parser bug") - # "" but completes as "" - full_text = "Let me use a to help." - s = BackendServicer() - s.reasoning_parser_cls = None - s.tokenizer = None - s.llm = SimpleNamespace(generate=make_generate([ - "Let me use a \n{"name": "calc"', + '\n{"name": "calc", "arguments": {"x": 1}}\n', ])) - s.tool_parser_cls = StateMachineParser - - tools_json = '[{"type":"function","function":{"name":"calc"}}]' - req = backend_pb2.PredictOptions(Prompt="x", Tools=tools_json) - replies = collect(s, req) - - # Full text must be assembled exactly once (uncertainty buffer flushed correctly) - assembled = "".join( - cd.content for r in replies for cd in r.chat_deltas if cd.content + s.tool_parser_cls = _BrokenStreamer + req = backend_pb2.PredictOptions( + Prompt="x", + Tools='[{"type":"function","function":{"name":"calc","parameters":{}}}]', ) - self.assertEqual( - assembled, full_text, - f"False-positive marker mangled or duplicated the content: {assembled!r}", + replies = self._collect(s, req) + + contents = [cd.content for r in replies for cd in r.chat_deltas if cd.content] + self.assertFalse( + any(" Date: Mon, 15 Jun 2026 22:47:21 +0200 Subject: [PATCH 5/6] docs(vllm): add server-side TTFT benchmark for the streaming tool-parser path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-contained stdlib-only script that measures time-to-first-token (TTFT) for the vLLM backend's two streaming scenarios: - tool_call: request mentions a tool; model is expected to call it - plain_text: request offers a tool but explicitly asks for prose Use this to compare: - the buffer-all path (#10346) → plain_text TTFT ≈ total response time - the native-streaming path (this PR) → plain_text TTFT ≈ true first-token time python examples/vllm-bench/ttft_streaming_tool_parser.py \\ --url http://localhost:8080 --model my-coder --runs 3 Lives under examples/ so it does not interfere with the test suite. Signed-off-by: pos-ei-don <1822533+pos-ei-don@users.noreply.github.com> --- examples/vllm-bench/README.md | 44 +++++ .../vllm-bench/ttft_streaming_tool_parser.py | 167 ++++++++++++++++++ 2 files changed, 211 insertions(+) create mode 100644 examples/vllm-bench/README.md create mode 100755 examples/vllm-bench/ttft_streaming_tool_parser.py diff --git a/examples/vllm-bench/README.md b/examples/vllm-bench/README.md new file mode 100644 index 000000000000..908650a359e6 --- /dev/null +++ b/examples/vllm-bench/README.md @@ -0,0 +1,44 @@ +# vLLM streaming + tool-parser benchmark + +A small, self-contained Python script (stdlib only) that measures +time-to-first-token (TTFT) for the vLLM backend's streaming path with +a tool parser configured. + +## Why this exists + +When a vLLM tool parser is active and a streaming chat completion is requested, +LocalAI used to buffer the full generation to prevent raw tool-call markup +(e.g. `...`) from leaking as `delta.content`. That was correct +for tool-call responses, but it turned plain-text responses into effectively +non-streaming — the client received nothing until the model finished. + +With native parser-side streaming (`parser.extract_tool_calls_streaming`, +implemented by every concrete vLLM 0.23+ tool parser), each delta can be +classified per-token: emit as content, emit as a structured tool_call, or +suppress. This benchmark shows the difference. + +## Two scenarios + +| Scenario | Request | Expected outcome | +|---|---|---| +| `tool_call` | "What is the weather in Paris? Please use the tool." | Model calls `get_weather`. `delta.tool_calls` chunks; no content leak. | +| `plain_text` | "Explain in 3 short sentences what a hash table is. Do NOT call any tool." | Model writes prose. With the streaming parser, content streams progressively; without it, the entire response arrives in one chunk. | + +## What the script reports + +For each scenario, across N runs: + +- `ttf_content_s` — time until the first `delta.content` chunk +- `ttf_tool_s` — time until the first `delta.tool_calls` chunk +- `n_content_chunks` — total number of distinct content deltas (1 = bundled, >1 = streamed) +- `n_tool_chunks` — total tool_call deltas +- `total_s` — total wall-clock until `[DONE]` +- `finish_reason` — `tool_calls` / `stop` / `length` + +## Usage + +```bash +python ttft_streaming_tool_parser.py --url http://localhost:8080 --model my-coder --runs 3 +``` + +JSON results are written to `ttft_bench_