A high-throughput and memory-efficient inference and serving engine for LLMs
Mirrored from https://github.com/tduncan-lum/Nvidia-Optimization-Stack---VLLM.git
Synced Nov 7, 10:15 UTC
[Tool Parser] Fix Qwen3Coder streaming parameter loss with speculative decode (#35615)
Signed-off-by: Martin Vit <martin@voipmonitor.org> Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Martin Vit committed
Mar 3, 2026 at 02:40 UTC
8ebd872f50bb35689ae2d0c12010431f1765caac
2 files changed
+175
-244
vllm/entrypoints/openai/chat_completion/serving.py
+16
-6
@@ -1249,13 +1249,23 @@ class OpenAIServingChat(OpenAIServing):
1249
)
1250
1251
# get the expected call based on partial JSON
1252
- # parsing which "autocompletes" the JSON
1253
- expected_call = json.dumps(
1254
- tool_parser.prev_tool_call_arr[index].get(
1255
- "arguments", {}
1256
- ),
1257
- ensure_ascii=False,
1252
+ # parsing which "autocompletes" the JSON.
1253
+ # Tool parsers (e.g. Qwen3Coder) store
1254
+ # arguments as a JSON string in
1255
+ # prev_tool_call_arr. Calling json.dumps()
1256
+ # on an already-serialized string would
1257
+ # double-serialize it (e.g. '{"k":1}' becomes
1258
+ # '"{\\"k\\":1}"'), which then causes the
1259
+ # replace() below to fail and append the
1260
+ # entire double-serialized string as a
1261
+ # spurious final delta.
1262
+ args = tool_parser.prev_tool_call_arr[index].get(
1263
+ "arguments", {}
1264
)
1265
+ if isinstance(args, str):
1266
+ expected_call = args
1267
+ else:
1268
+ expected_call = json.dumps(args, ensure_ascii=False)
1269
1270
# get what we've streamed so far for arguments
1271
# for the current tool
vllm/tool_parsers/qwen3coder_tool_parser.py
+159
-238
@@ -479,20 +479,22 @@ class Qwen3CoderToolParser(ToolParser):
479
self.header_sent = True
480
self.in_function = True
481
482
- # IMPORTANT: Add to prev_tool_call_arr immediately when
483
- # we detect a tool call. This ensures
484
- # finish_reason="tool_calls" even if parsing isn't complete
485
- already_added = any(
486
- tool.get("name") == self.current_function_name
487
- for tool in self.prev_tool_call_arr
482
+ # Always append — each tool call is a separate
483
+ # invocation even if the function name is the same
484
+ # (e.g. two consecutive "read" calls).
485
+ self.prev_tool_call_arr.append(
486
+ {
487
+ "name": self.current_function_name,
488
+ "arguments": "{}",
489
+ }
490
)
489
- if not already_added:
490
- self.prev_tool_call_arr.append(
491
- {
492
- "name": self.current_function_name,
493
- "arguments": "{}", # Placeholder, will be updated later
494
- }
495
- )
491
+
492
+ # Initialize streamed args tracking for this tool.
493
+ # The serving layer reads streamed_args_for_tool to
494
+ # compute remaining arguments at stream end. Without
495
+ # this, IndexError occurs when the serving layer
496
+ # accesses streamed_args_for_tool[index].
497
+ self.streamed_args_for_tool.append("")
498
499
# Send header with function info
500
return DeltaMessage(
@@ -511,9 +513,14 @@ class Qwen3CoderToolParser(ToolParser):
513
514
# We've sent header, now handle function body
515
if self.in_function:
514
- # Send opening brace if not sent yet
515
- if not self.json_started and self.parameter_prefix not in delta_text:
516
+ # Always send opening brace first, regardless of whether
517
+ # parameter_prefix is in the current delta. With speculative
518
+ # decoding, a single delta may contain both the opening brace
519
+ # and parameter data; skipping "{" here would desync
520
+ # json_started from what was actually streamed.
521
+ if not self.json_started:
522
self.json_started = True
523
+ self.streamed_args_for_tool[self.current_tool_index] += "{"
524
return DeltaMessage(
525
tool_calls=[
526
DeltaToolCall(
@@ -523,25 +530,133 @@ class Qwen3CoderToolParser(ToolParser):
530
]
531
)
532
526
- # Make sure json_started is set if we're processing parameters
527
- if not self.json_started:
528
- self.json_started = True
533
+ # Find all parameter start positions in current tool_text
534
+ param_starts = []
535
+ search_idx = 0
536
+ while True:
537
+ search_idx = tool_text.find(self.parameter_prefix, search_idx)
538
+ if search_idx == -1:
539
+ break
540
+ param_starts.append(search_idx)
541
+ search_idx += len(self.parameter_prefix)
542
+
543
+ # Process ALL complete params in a loop (spec decode fix).
544
+ # With speculative decoding a single delta can deliver
545
+ # multiple complete parameters at once. The old single-pass
546
+ # code would process one and ``return None`` if the next was
547
+ # incomplete — skipping any already-complete params that
548
+ # preceded it. Using a loop with ``break`` instead ensures
549
+ # we emit every complete parameter before yielding control.
550
+ json_fragments = []
551
+ while not self.in_param and self.param_count < len(param_starts):
552
+ param_idx = param_starts[self.param_count]
553
+ param_start = param_idx + len(self.parameter_prefix)
554
+ remaining = tool_text[param_start:]
555
+
556
+ if ">" not in remaining:
557
+ break
558
+
559
+ name_end = remaining.find(">")
560
+ current_param_name = remaining[:name_end]
561
+
562
+ value_start = param_start + name_end + 1
563
+ value_text = tool_text[value_start:]
564
+ if value_text.startswith("\n"):
565
+ value_text = value_text[1:]
566
+
567
+ param_end_idx = value_text.find(self.parameter_end_token)
568
+ if param_end_idx == -1:
569
+ next_param_idx = value_text.find(self.parameter_prefix)
570
+ func_end_idx = value_text.find(self.function_end_token)
571
+
572
+ if next_param_idx != -1 and (
573
+ func_end_idx == -1 or next_param_idx < func_end_idx
574
+ ):
575
+ param_end_idx = next_param_idx
576
+ elif func_end_idx != -1:
577
+ param_end_idx = func_end_idx
578
+ else:
579
+ # Fallback for malformed XML where </function>
580
+ # is missing. Use </tool_call> as a delimiter
581
+ # if present in the value so we don't include
582
+ # the closing tag as part of the param value.
583
+ tool_end_in_value = value_text.find(self.tool_call_end_token)
584
+ if tool_end_in_value != -1:
585
+ param_end_idx = tool_end_in_value
586
+ else:
587
+ # Parameter incomplete — break so we still
588
+ # emit any fragments accumulated by earlier
589
+ # loop iterations.
590
+ break
591
+
592
+ if param_end_idx == -1:
593
+ break
594
+
595
+ param_value = value_text[:param_end_idx]
596
+ if param_value.endswith("\n"):
597
+ param_value = param_value[:-1]
598
+
599
+ self.current_param_name = current_param_name
600
+ self.accumulated_params[current_param_name] = param_value
601
+
602
+ param_config = self._get_arguments_config(
603
+ self.current_function_name or "",
604
+ self.streaming_request.tools if self.streaming_request else None,
605
+ )
606
530
- # Check for function end in accumulated text
607
+ converted_value = self._convert_param_value(
608
+ param_value,
609
+ current_param_name,
610
+ param_config,
611
+ self.current_function_name or "",
612
+ )
613
+
614
+ serialized_value = json.dumps(converted_value, ensure_ascii=False)
615
+
616
+ if self.param_count == 0:
617
+ json_fragment = f'"{current_param_name}": {serialized_value}'
618
+ else:
619
+ json_fragment = f', "{current_param_name}": {serialized_value}'
620
+
621
+ self.param_count += 1
622
+ json_fragments.append(json_fragment)
623
+
624
+ if json_fragments:
625
+ combined = "".join(json_fragments)
626
+
627
+ if self.current_tool_index < len(self.streamed_args_for_tool):
628
+ self.streamed_args_for_tool[self.current_tool_index] += combined
629
+ else:
630
+ logger.warning(
631
+ "streamed_args_for_tool out of sync: index=%d len=%d",
632
+ self.current_tool_index,
633
+ len(self.streamed_args_for_tool),
634
+ )
635
+
636
+ return DeltaMessage(
637
+ tool_calls=[
638
+ DeltaToolCall(
639
+ index=self.current_tool_index,
640
+ function=DeltaFunctionCall(arguments=combined),
641
+ )
642
+ ]
643
+ )
644
+
645
+ # Check for function end AFTER processing parameters.
646
+ # This ordering is critical: with speculative decoding a
647
+ # burst can deliver the final parameter value together with
648
+ # </function>. If the close check ran first it would emit
649
+ # "}" and set in_function=False before the parameter loop
650
+ # ever ran, causing the parameter to be silently dropped.
651
if not self.json_closed and self.function_end_token in tool_text:
532
- # Close JSON
652
self.json_closed = True
653
535
- # Extract complete tool call to update
536
- # prev_tool_call_arr with final arguments
537
- # Find the function content
654
func_start = tool_text.find(self.tool_call_prefix) + len(
655
self.tool_call_prefix
656
)
657
func_content_end = tool_text.find(self.function_end_token, func_start)
658
if func_content_end != -1:
659
func_content = tool_text[func_start:func_content_end]
544
- # Parse to get the complete arguments
660
try:
661
parsed_tool = self._parse_xml_function_call(
662
func_content,
@@ -549,16 +664,27 @@ class Qwen3CoderToolParser(ToolParser):
664
if self.streaming_request
665
else None,
666
)
552
- if parsed_tool:
553
- # Update existing entry in
554
- # prev_tool_call_arr with complete args
555
- for i, tool in enumerate(self.prev_tool_call_arr):
556
- if tool.get("name") == parsed_tool.function.name:
557
- args = parsed_tool.function.arguments
558
- self.prev_tool_call_arr[i]["arguments"] = args
559
- break
667
+ if parsed_tool and self.current_tool_index < len(
668
+ self.prev_tool_call_arr
669
+ ):
670
+ self.prev_tool_call_arr[self.current_tool_index][
671
+ "arguments"
672
+ ] = parsed_tool.function.arguments
673
except Exception:
561
- pass # Ignore parsing errors during streaming
674
+ logger.debug(
675
+ "Failed to parse tool call during streaming: %s",
676
+ tool_text,
677
+ exc_info=True,
678
+ )
679
+
680
+ if self.current_tool_index < len(self.streamed_args_for_tool):
681
+ self.streamed_args_for_tool[self.current_tool_index] += "}"
682
+ else:
683
+ logger.warning(
684
+ "streamed_args_for_tool out of sync: index=%d len=%d",
685
+ self.current_tool_index,
686
+ len(self.streamed_args_for_tool),
687
+ )
688
689
result = DeltaMessage(
690
tool_calls=[
@@ -569,215 +695,10 @@ class Qwen3CoderToolParser(ToolParser):
695
]
696
)
697
572
- # Reset state for next tool
698
self.in_function = False
699
self.json_closed = True
700
self.accumulated_params = {}
701
702
return result
703
579
- # Look for parameters
580
- # Find all parameter starts
581
- param_starts = []
582
- idx = 0
583
- while True:
584
- idx = tool_text.find(self.parameter_prefix, idx)
585
- if idx == -1:
586
- break
587
- param_starts.append(idx)
588
- idx += len(self.parameter_prefix)
589
-
590
- # Check if we should start a new parameter
591
- if (
592
- not self.in_param
593
- and self.param_count < len(param_starts)
594
- and len(param_starts) > self.param_count
595
- ):
596
- # Process the next parameter
597
- param_idx = param_starts[self.param_count]
598
- param_start = param_idx + len(self.parameter_prefix)
599
- remaining = tool_text[param_start:]
600
-
601
- if ">" in remaining:
602
- # We have the complete parameter name
603
- name_end = remaining.find(">")
604
- self.current_param_name = remaining[:name_end]
605
-
606
- # Find the parameter value
607
- value_start = param_start + name_end + 1
608
- value_text = tool_text[value_start:]
609
- if value_text.startswith("\n"):
610
- value_text = value_text[1:]
611
-
612
- # Find where this parameter ends
613
- param_end_idx = value_text.find(self.parameter_end_token)
614
- if param_end_idx == -1:
615
- # No closing tag, look for next parameter or
616
- # function end
617
- next_param_idx = value_text.find(self.parameter_prefix)
618
- func_end_idx = value_text.find(self.function_end_token)
619
-
620
- if next_param_idx != -1 and (
621
- func_end_idx == -1 or next_param_idx < func_end_idx
622
- ):
623
- param_end_idx = next_param_idx
624
- elif func_end_idx != -1:
625
- param_end_idx = func_end_idx
626
- else:
627
- # Neither found, check if tool call is complete
628
- if self.tool_call_end_token in tool_text:
629
- # Tool call is complete, so parameter
630
- # must be complete too. Use all
631
- # remaining text before function end
632
- param_end_idx = len(value_text)
633
- else:
634
- # Still streaming, wait for more content
635
- return None
636
-
637
- if param_end_idx != -1:
638
- # Complete parameter found
639
- param_value = value_text[:param_end_idx]
640
- if param_value.endswith("\n"):
641
- param_value = param_value[:-1]
642
-
643
- # Store raw value for later processing
644
- self.accumulated_params[self.current_param_name] = param_value
645
-
646
- # Get parameter configuration for type conversion
647
- param_config = self._get_arguments_config(
648
- self.current_function_name or "",
649
- self.streaming_request.tools
650
- if self.streaming_request
651
- else None,
652
- )
653
-
654
- # Convert param value to appropriate type
655
- converted_value = self._convert_param_value(
656
- param_value,
657
- self.current_param_name,
658
- param_config,
659
- self.current_function_name or "",
660
- )
661
-
662
- # Build JSON fragment based on the converted type
663
- # Use json.dumps to properly serialize the value
664
- serialized_value = json.dumps(
665
- converted_value, ensure_ascii=False
666
- )
667
-
668
- if self.param_count == 0:
669
- json_fragment = (
670
- f'"{self.current_param_name}": {serialized_value}'
671
- )
672
- else:
673
- json_fragment = (
674
- f', "{self.current_param_name}": {serialized_value}'
675
- )
676
-
677
- self.param_count += 1
678
-
679
- return DeltaMessage(
680
- tool_calls=[
681
- DeltaToolCall(
682
- index=self.current_tool_index,
683
- function=DeltaFunctionCall(arguments=json_fragment),
684
- )
685
- ]
686
- )
687
-
688
- # Continue parameter value - Not used in the current implementation
689
- # since we process complete parameters above
690
- if self.in_param:
691
- if self.parameter_end_token in delta_text:
692
- # End of parameter
693
- end_idx = delta_text.find(self.parameter_end_token)
694
- value_chunk = delta_text[:end_idx]
695
-
696
- # Skip past > if at start
697
- if not self.current_param_value and ">" in value_chunk:
698
- gt_idx = value_chunk.find(">")
699
- value_chunk = value_chunk[gt_idx + 1 :]
700
-
701
- if not self.current_param_value and value_chunk.startswith("\n"):
702
- value_chunk = value_chunk[1:]
703
-
704
- # Store complete value
705
- full_value = self.current_param_value + value_chunk
706
- self.accumulated_params[self.current_param_name] = full_value
707
-
708
- # Get parameter configuration for type conversion
709
- param_config = self._get_arguments_config(
710
- self.current_function_name or "",
711
- self.streaming_request.tools
712
- if self.streaming_request
713
- else None,
714
- )
715
-
716
- # Convert the parameter value to the appropriate type
717
- converted_value = self._convert_param_value(
718
- full_value,
719
- self.current_param_name or "",
720
- param_config,
721
- self.current_function_name or "",
722
- )
723
-
724
- # Serialize the converted value
725
- serialized_value = json.dumps(converted_value, ensure_ascii=False)
726
-
727
- # Since we've been streaming the quoted version,
728
- # we need to close it properly
729
- # This is complex - for now just complete the value
730
- self.in_param = False
731
- self.current_param_value = ""
732
-
733
- # Just close the current parameter string
734
- return DeltaMessage(
735
- tool_calls=[
736
- DeltaToolCall(
737
- index=self.current_tool_index,
738
- function=DeltaFunctionCall(
739
- arguments='"'
740
- ), # Close the string quote
741
- )
742
- ]
743
- )
744
- else:
745
- # Continue accumulating value
746
- value_chunk = delta_text
747
-
748
- # Handle first chunk after param name
749
- if not self.current_param_value and ">" in value_chunk:
750
- gt_idx = value_chunk.find(">")
751
- value_chunk = value_chunk[gt_idx + 1 :]
752
-
753
- if not self.current_param_value and value_chunk.startswith("\n"):
754
- value_chunk = value_chunk[1:]
755
-
756
- if value_chunk:
757
- # Stream the escaped delta
758
- prev_escaped = (
759
- json.dumps(self.current_param_value, ensure_ascii=False)[
760
- 1:-1
761
- ]
762
- if self.current_param_value
763
- else ""
764
- )
765
- self.current_param_value += value_chunk
766
- full_escaped = json.dumps(
767
- self.current_param_value, ensure_ascii=False
768
- )[1:-1]
769
- delta_escaped = full_escaped[len(prev_escaped) :]
770
-
771
- if delta_escaped:
772
- return DeltaMessage(
773
- tool_calls=[
774
- DeltaToolCall(
775
- index=self.current_tool_index,
776
- function=DeltaFunctionCall(
777
- arguments=delta_escaped
778
- ),
779
- )
780
- ]
781
- )
782
-
704
return None