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..a7933b911596 100644 --- a/backend/python/vllm/test.py +++ b/backend/python/vllm/test.py @@ -664,3 +664,188 @@ 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."] + + 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 + 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"}') 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: