From 5392eb813e991b20521575cc705c8380e03d1d3a Mon Sep 17 00:00:00 2001 From: pos-ei-don <1822533+pos-ei-don@users.noreply.github.com> Date: Fri, 23 Oct 2026 23:58:52 +0200 Subject: [PATCH 1/2] fix(vllm): keep reasoning out of streamed content With a reasoning parser configured, the streaming path sent every raw text delta as ChatDelta.content. LocalAI uses the backend's ChatDeltas as they are, so with thinking enabled the reasoning arrived as delta.content, and the final chunk then repeated the whole answer and the reasoning a second time. Non-streaming requests were split correctly. Use vLLM's composed parser (ParserManager.get_parser, vLLM >= 0.30), the one vLLM's own OpenAI server streams with: one instance per request, built with the request's chat_template_kwargs (enable_thinking), fed each delta via parse_delta. Reasoning, content and tool calls go out in their own fields as they are produced; the final chunk carries only token counts and logprobs. If the parser raises mid-stream, the rest of the output is buffered and the final chunk sends what the full-text parse has beyond what was already streamed (suffixes per field and per tool-call index). Where the full parse disagrees with what went out, nothing is guessed and the mismatch is logged. Only active when a reasoning_parser is set; without one, or on a vLLM without ParserManager, streaming is unchanged. Signed-off-by: pos-ei-don <1822533+pos-ei-don@users.noreply.github.com> --- backend/python/vllm/backend.py | 225 ++++++++++++++++++++++++++++++++- backend/python/vllm/test.py | 167 ++++++++++++++++++++++++ 2 files changed, 391 insertions(+), 1 deletion(-) diff --git a/backend/python/vllm/backend.py b/backend/python/vllm/backend.py index 6dabbf4bf687..d69c4035724b 100644 --- a/backend/python/vllm/backend.py +++ b/backend/python/vllm/backend.py @@ -51,6 +51,18 @@ except ImportError: HAS_REASONING_PARSERS = False +# vLLM >= 0.30 composes reasoning and tool parsing into one per-delta parser +# (ParserManager / Parser.parse_delta) — the one its own OpenAI server uses +# for streaming. +try: + from vllm.parser.parser_manager import ParserManager + from vllm.entrypoints.openai.chat_completion.protocol import ( + ChatCompletionRequest as _ParserRequest, + ) + HAS_STREAM_PARSER = True +except ImportError: + HAS_STREAM_PARSER = False + # vLLM >= 0.23 renamed GuidedDecodingParams -> StructuredOutputsParams and the # SamplingParams field guided_decoding -> structured_outputs. try: @@ -307,6 +319,20 @@ async def LoadModel(self, request, context): except Exception as e: print(f"Failed to load reasoning_parser {opts.get('reasoning_parser')}: {e}", file=sys.stderr) + # With a reasoning parser, streaming has to split reasoning from content + # per delta. Use vLLM's composed parser for that (reasoning + tools). + self.stream_parser_cls = None + if HAS_STREAM_PARSER and self.reasoning_parser_cls is not None: + try: + self.stream_parser_cls = ParserManager.get_parser( + tool_parser_name=opts.get("tool_parser") if self.tool_parser_cls else None, + reasoning_parser_name=opts.get("reasoning_parser"), + enable_auto_tools=True, + ) + except Exception as e: + print(f"Streaming parser unavailable, streaming raw deltas: {e!r}", file=sys.stderr) + self.stream_parser_cls = None + print("Model loaded successfully", file=sys.stderr) return backend_pb2.Result(message="Model loaded successfully", success=True) @@ -387,6 +413,7 @@ async def Free(self, request, context): del self.tokenizer self.tool_parser_cls = None self.reasoning_parser_cls = None + self.stream_parser_cls = None gc.collect() try: import torch @@ -634,6 +661,120 @@ def _split_reasoning(rp, generated_text, prompt, reasoning, content): # budget. It is all reasoning, and there is no answer to report. return reasoning or "", content or "" + def _new_stream_parser(self, request, template_kwargs): + """Build vLLM's per-delta parser for one streaming request. + + Same construction as vLLM's OpenAI server: the request's + ``chat_template_kwargs`` (``enable_thinking``) set the parser's initial + state, and the prompt token ids passed to the first ``parse_delta`` + tell it whether the reasoning block is open. Returns ``(None, None)`` + when that is not possible; the caller then streams raw deltas as + before. + """ + cls = getattr(self, "stream_parser_cls", None) + if cls is None: + return None, None + try: + tools = None + if request.Tools: + try: + tools = json.loads(request.Tools) or None + except json.JSONDecodeError: + tools = None + parser_request = _ParserRequest( + model="local", + messages=[{"role": "user", "content": ""}], + tools=tools, + ) + chat_template_kwargs = { + k: v for k, v in (template_kwargs or {}).items() + if k not in ("tokenize", "add_generation_prompt", "tools") + } + parser = cls(self.tokenizer, parser_request.tools, + chat_template_kwargs=chat_template_kwargs) + return parser, parser_request + except Exception as e: + print(f"Could not build streaming parser, streaming raw deltas: {e!r}", + file=sys.stderr) + return None, None + + @staticmethod + def _delta_message_to_reply(msg): + """vLLM ``DeltaMessage`` -> ``Reply``, or ``None`` if it carries nothing.""" + 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 not cd_kwargs: + return None + return backend_pb2.Reply( + message=bytes(msg.content or "", encoding='utf-8'), + chat_deltas=[backend_pb2.ChatDelta(**cd_kwargs)], + ) + + @staticmethod + def _unsent_remainder(sent_content, sent_reasoning, sent_tools, + content, reasoning, tool_calls): + """After the streaming parser failed mid-stream: what is still unsent. + + ``content``/``reasoning``/``tool_calls`` come from parsing the full + text; ``sent_tools`` maps a tool-call index to the name and arguments + already streamed for it. Only suffixes of what was sent are added. If + the full parse disagrees with what went out, nothing is guessed: the + field is left as streamed and the mismatch is logged. + """ + def unsent(field, full, sent): + full = full or "" + if full.startswith(sent): + return full[len(sent):] + print(f"Streamed {field} does not match the final parse; " + f"not sending the rest", file=sys.stderr) + return "" + + remainder = {} + rest = unsent("reasoning", reasoning, sent_reasoning) + if rest: + remainder["reasoning_content"] = rest + if sent_tools and not tool_calls: + # A tool call was streamed but the full text does not parse as + # one: the rest of the content would be raw tool-call markup. + print("Streamed tool call not found in the final parse; " + "not sending the rest", file=sys.stderr) + else: + rest = unsent("content", content, sent_content) + if rest: + remainder["content"] = rest + tc_protos = [] + for i, (call_id, name, arguments) in enumerate(tool_calls or []): + sent = sent_tools.get(i) + if sent is None: + tc_protos.append(backend_pb2.ToolCallDelta( + index=i, id=call_id or "", name=name or "", arguments=arguments or "", + )) + continue + if sent["name"] and name and sent["name"] != name: + print(f"Streamed tool call {i} does not match the final parse; " + f"not sending the rest", file=sys.stderr) + continue + rest = unsent(f"tool call {i} arguments", arguments, sent["arguments"]) + if rest: + tc_protos.append(backend_pb2.ToolCallDelta(index=i, arguments=rest)) + if tc_protos: + remainder["tool_calls"] = tc_protos + return remainder + async def _predict(self, request, context, streaming=False): # Build the sampling parameters sampling_params = self._build_sampling_params(request) @@ -777,12 +918,58 @@ async def _predict(self, request, context, streaming=False): and hasattr(tp_instance, "extract_tool_calls_streaming") ) + # With a reasoning parser, split reasoning, content and tool calls per + # delta, as vLLM's OpenAI server does. Without this the raw text — + # reasoning included — was streamed as content, and LocalAI takes the + # backend's ChatDeltas as they are. + stream_parser, parser_request = ( + self._new_stream_parser(request, template_kwargs) if streaming else (None, None) + ) + stream_parser_error = False + sent_content = "" + sent_reasoning = "" + sent_tools = {} # tool-call index -> {"name", "arguments"} already streamed + try: async for request_output in outputs: iteration_text = request_output.outputs[0].text last_output = request_output - if streaming: + if streaming and stream_parser is not None and not stream_parser_error: + delta_iteration_text = iteration_text.removeprefix(generated_text) + new_token_ids = list(request_output.outputs[0].token_ids) + delta_token_ids = new_token_ids[len(generated_token_ids):] + try: + msg = stream_parser.parse_delta( + delta_text=delta_iteration_text, + delta_token_ids=delta_token_ids, + request=parser_request, + prompt_token_ids=list(request_output.prompt_token_ids or []), + finished=bool( + getattr(request_output, "finished", False) + or getattr(request_output.outputs[0], "finish_reason", None) is not None + ), + ) + except Exception as e: + print(f"Streaming parser error (sending the rest with the " + f"final chunk): {e!r}", file=sys.stderr) + stream_parser_error = True + msg = None + if msg is not None: + reply = self._delta_message_to_reply(msg) + if reply is not None: + sent_content += msg.content or "" + sent_reasoning += msg.reasoning or "" + for tc in (msg.tool_calls or []): + fn = tc.function or None + sent = sent_tools.setdefault(tc.index, {"name": "", "arguments": ""}) + if fn and fn.name and not sent["name"]: + sent["name"] = fn.name # sent once, as in OpenAI deltas + sent["arguments"] += (fn.arguments if fn and fn.arguments else "") or "" + yield reply + elif streaming and stream_parser is not None: + pass # parser failed: buffer, the final chunk carries the rest + elif streaming: delta_iteration_text = iteration_text.removeprefix(generated_text) new_token_ids = list(request_output.outputs[0].token_ids) delta_token_ids = new_token_ids[len(generated_token_ids):] @@ -864,6 +1051,9 @@ async def _predict(self, request, context, streaming=False): except Exception as e: print(f"Reasoning parser error: {e}", file=sys.stderr) + # Content after the reasoning split, before tool-call extraction. + split_content = content + # 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. @@ -936,6 +1126,39 @@ async def _predict(self, request, context, streaming=False): tool_calls=tool_calls_proto, ) + if streaming and stream_parser is not None: + # Everything went out per delta; the final chunk carries only token + # counts and logprobs, or LocalAI would repeat answer and + # reasoning. If the parser failed mid-stream, send what is left. + remainder = {} + if stream_parser_error: + rest_content = split_content + rest_tool_calls = [] + if has_tool_parser and rest_content: + try: + tp = tp_instance or self.tool_parser_cls(self.tokenizer) + info = tp.extract_tool_calls(rest_content, request=None) + if info.tools_called: + rest_content = info.content or "" + rest_tool_calls = [ + (tc.id, tc.function.name, tc.function.arguments) + for tc in info.tool_calls + ] + except Exception as e: + print(f"Tool parser error: {e}", file=sys.stderr) + remainder = self._unsent_remainder( + sent_content, sent_reasoning, sent_tools, + rest_content, reasoning_content, rest_tool_calls, + ) + yield backend_pb2.Reply( + message=bytes(remainder.get("content", ""), encoding='utf-8'), + prompt_tokens=prompt_tokens, + tokens=completion_tokens, + chat_deltas=[backend_pb2.ChatDelta(**remainder)], + logprobs=logprobs_bytes, + ) + return + if streaming: # Final chunk with structured data. # diff --git a/backend/python/vllm/test.py b/backend/python/vllm/test.py index 27a846f0918d..a6efe693ccae 100644 --- a/backend/python/vllm/test.py +++ b/backend/python/vllm/test.py @@ -664,3 +664,170 @@ def __init__(self, tokenizer): self.assertIsInstance( s._new_reasoning_parser({"enable_thinking": False}), _Old, ) + + +class TestStreamingReasoningSplit(unittest.TestCase): + """With a reasoning parser, streaming must split reasoning from content. + + The stream used to send each raw text delta as ChatDelta.content, so with + thinking on the reasoning text arrived as delta.content and the final + chunk repeated it as reasoning_content. These tests replace the engine + and vLLM's composed parser (ParserManager.get_parser) with fakes; no GPU, + model or server needed. + """ + + THINK = ["Let me ", "count the ", "primes.", ""] + ANSWER = ["There are ", "37 primes."] + + class _DeltaMsg: + def __init__(self, content=None, reasoning=None, tool_calls=None): + self.content = content + self.reasoning = reasoning + self.tool_calls = tool_calls or [] + + @classmethod + def _parser_cls(cls, seen=None, fail_after=None): + delta_msg = cls._DeltaMsg + + class _Parser: + """Stands in for vLLM's DelegatingParser: reasoning until the + end tag, then content, then a tool call between tags + (name first, arguments as they arrive).""" + def __init__(self, tokenizer, tools=None, **kwargs): + if seen is not None: + seen.update(kwargs, tools=tools) + self._state = "reasoning" + self._calls = 0 + + def parse_delta(self, delta_text, delta_token_ids, request, + prompt_token_ids=None, *, finished): + from types import SimpleNamespace + self._calls += 1 + if fail_after is not None and self._calls > fail_after: + raise RuntimeError("simulated parser bug") + if self._state == "reasoning": + if "" in delta_text: + self._state = "content" + before = delta_text.split("", 1)[0] + return delta_msg(reasoning=before) if before else None + return delta_msg(reasoning=delta_text) + if delta_text == "": + self._state = "tool" + fn = SimpleNamespace(name="search", arguments="") + return delta_msg(tool_calls=[SimpleNamespace(index=0, id="call_1", function=fn)]) + if self._state == "tool": + if delta_text == "": + return None + fn = SimpleNamespace(name=None, arguments=delta_text) + return delta_msg(tool_calls=[SimpleNamespace(index=0, id=None, function=fn)]) + return delta_msg(content=delta_text) + return _Parser + + class _ToolParser: + """Full-text tool extraction used after a streaming parser failure.""" + def __init__(self, tokenizer, tools=None): + pass + + def extract_tool_calls(self, text, request=None): + from types import SimpleNamespace + if "" not in text: + return SimpleNamespace(tools_called=False, content=text, tool_calls=[]) + before, rest = text.split("", 1) + args = rest.split("", 1)[0] + call = SimpleNamespace(id="call_1", function=SimpleNamespace(name="search", arguments=args)) + return SimpleNamespace(tools_called=True, content=before, tool_calls=[call]) + + class _ReasoningParser: + """Full-text split used after the loop (BaseThinkingReasoningParser shape).""" + start_token = "" + end_token = "" + + def __init__(self, tokenizer, **kwargs): + pass + + def extract_reasoning(self, text, request=None): + if "" in text: + r, c = text.split("", 1) + return r, c + return text, None + + def _servicer(self, stream_parser_cls, pieces=None): + import sys, os + from types import SimpleNamespace + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + from backend import BackendServicer + s = BackendServicer() + s.tokenizer = None + s.tool_parser_cls = self._ToolParser + s.reasoning_parser_cls = self._ReasoningParser + s.stream_parser_cls = stream_parser_cls + s.llm = SimpleNamespace(generate=TestStreamingToolParser._make_generate( + self._cumulative(pieces or (self.THINK + self.ANSWER)))) + return s + + @staticmethod + def _cumulative(pieces): + out, acc = [], "" + for p in pieces: + acc += p + out.append(acc) + return out + + @staticmethod + def _joined(replies, field): + return "".join(getattr(cd, field) for r in replies for cd in r.chat_deltas) + + def test_reasoning_is_not_streamed_as_content(self): + s = self._servicer(self._parser_cls()) + replies = TestStreamingToolParser._collect(s, backend_pb2.PredictOptions(Prompt="x")) + self.assertEqual(self._joined(replies, "content"), "There are 37 primes.") + self.assertEqual(self._joined(replies, "reasoning_content"), "Let me count the primes.") + # Progressive: reasoning and content arrive before the final chunk. + self.assertTrue(any(cd.reasoning_content for r in replies[:-1] for cd in r.chat_deltas)) + self.assertTrue(any(cd.content for r in replies[:-1] for cd in r.chat_deltas)) + # The final chunk carries no text, otherwise LocalAI repeats it. + final = replies[-1].chat_deltas[0] + self.assertEqual((final.content, final.reasoning_content), ("", "")) + + def test_parser_error_mid_stream_sends_the_rest_once(self): + s = self._servicer(self._parser_cls(fail_after=2)) + replies = TestStreamingToolParser._collect(s, backend_pb2.PredictOptions(Prompt="x")) + self.assertEqual(self._joined(replies, "content"), "There are 37 primes.") + self.assertEqual(self._joined(replies, "reasoning_content"), "Let me count the primes.") + + def test_without_stream_parser_streams_raw_deltas(self): + """vLLM without ParserManager: previous behaviour.""" + s = self._servicer(None) + replies = TestStreamingToolParser._collect(s, backend_pb2.PredictOptions(Prompt="x")) + intermediate = "".join(cd.content for r in replies[:-1] for cd in r.chat_deltas) + self.assertEqual(intermediate, "".join(self.THINK + self.ANSWER)) + + def test_parser_gets_chat_template_kwargs_and_tools(self): + seen = {} + s = self._servicer(self._parser_cls(seen=seen)) + tools = '[{"type":"function","function":{"name":"calc","parameters":{}}}]' + parser, parser_request = s._new_stream_parser( + backend_pb2.PredictOptions(Prompt="x", Tools=tools), + {"tokenize": False, "add_generation_prompt": True, "enable_thinking": False, + "tools": [{}]}, + ) + self.assertIsNotNone(parser) + self.assertEqual(seen["chat_template_kwargs"], {"enable_thinking": False}) + self.assertEqual([t.function.name for t in seen["tools"]], ["calc"]) + + def test_parser_error_after_partial_tool_arguments(self): + """The rest of the arguments follows on the same index; no raw + tool-call markup reaches content.""" + pieces = self.THINK + ["Checking.", "", '{"q": ', '"primes"}', ""] + tools = '[{"type":"function","function":{"name":"search","parameters":{}}}]' + for fail_after in (6, 7): # after the name / after part of the arguments + with self.subTest(fail_after=fail_after): + s = self._servicer(self._parser_cls(fail_after=fail_after), pieces) + replies = TestStreamingToolParser._collect( + s, backend_pb2.PredictOptions(Prompt="x", Tools=tools)) + self.assertEqual(self._joined(replies, "content"), "Checking.") + self.assertEqual(self._joined(replies, "reasoning_content"), "Let me count the primes.") + calls = [tc for r in replies for cd in r.chat_deltas for tc in cd.tool_calls] + self.assertEqual({tc.index for tc in calls}, {0}) + self.assertEqual([tc.name for tc in calls if tc.name], ["search"]) + self.assertEqual("".join(tc.arguments for tc in calls), '{"q": "primes"}') From ff0130fa89bcb8c8e9a15f5a9d5e06415c405c53 Mon Sep 17 00:00:00 2001 From: pos-ei-don <1822533+pos-ei-don@users.noreply.github.com> Date: Sat, 24 Oct 2026 19:32:01 +0000 Subject: [PATCH 2/2] test(vllm): isolate streaming parser fixtures Mock the request constructor so fake parser tests can run on older vLLM profiles without ParserManager. Document the streaming version requirement and fallback behavior. Signed-off-by: pos-ei-don <1822533+pos-ei-don@users.noreply.github.com> --- backend/python/vllm/test.py | 18 ++++++++++++++++++ docs/content/features/openai-functions.md | 7 +++++++ 2 files changed, 25 insertions(+) diff --git a/backend/python/vllm/test.py b/backend/python/vllm/test.py index a6efe693ccae..a7933b911596 100644 --- a/backend/python/vllm/test.py +++ b/backend/python/vllm/test.py @@ -679,6 +679,24 @@ class TestStreamingReasoningSplit(unittest.TestCase): THINK = ["Let me ", "count the ", "primes.", ""] ANSWER = ["There are ", "37 primes."] + def setUp(self): + from types import SimpleNamespace + from unittest.mock import patch + import backend + + def parser_request(model, messages, tools=None): + return SimpleNamespace( + model=model, messages=messages, + tools=[SimpleNamespace(function=SimpleNamespace(**tool["function"])) + for tool in tools] if tools else None, + ) + + # These fake-parser tests also run on backend profiles whose vLLM + # predates ParserManager and therefore has no _ParserRequest import. + request_patch = patch.object(backend, "_ParserRequest", parser_request, create=True) + request_patch.start() + self.addCleanup(request_patch.stop) + class _DeltaMsg: def __init__(self, content=None, reasoning=None, tool_calls=None): self.content = content diff --git a/docs/content/features/openai-functions.md b/docs/content/features/openai-functions.md index 5e7a384cad64..d80a91767b9d 100644 --- a/docs/content/features/openai-functions.md +++ b/docs/content/features/openai-functions.md @@ -84,6 +84,13 @@ When you import a vLLM model through the LocalAI gallery, the importer looks up Available tool parsers include `hermes`, `llama3_json`, `llama4_pythonic`, `mistral`, `qwen3_xml`, `deepseek_v3`, `granite4`, `kimi_k2`, `glm45`, and more. Available reasoning parsers include `deepseek_r1`, `qwen3`, `mistral`, `gemma4`, `granite`. See the upstream vLLM documentation for the full list. +With `backend: vllm`, vLLM 0.30 or later, and a configured reasoning parser, +streamed responses separate `delta.reasoning_content`, `delta.content`, and tool +calls as tokens arrive. The final chunk does not repeat the generated text. +If parsing fails during a stream, the backend buffers the remaining output and +sends any matching, unsent suffixes in the final chunk. Older vLLM versions keep +the previous streaming behavior and can include reasoning in `delta.content`. + ### MLX / MLX-VLM MLX backends **auto-detect** the right tool parser by inspecting the model's chat template - you don't need to set anything. Just load an MLX-quantized model that was trained with tool support: