Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
182 changes: 182 additions & 0 deletions tests/parser/engine/test_deepseek_long_arg_streaming.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,182 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Streaming regression tests for long DeepSeek DSML parameter values.

A long ``string="true"`` value carries no tag boundary inside it, so the
engine's ``arg_structural_chars`` gate used to suppress every delta of that
value: it was buffered until its closing tag arrived, and the client received
no bytes for the whole generation. For multi-KB values that silence exceeds a
client's read timeout, so the request looks like a dropped stream.

These tests assert the value trickles out while it is generated.
"""

import json
from dataclasses import dataclass

import pytest

from tests.parser.engine.conftest import make_mock_tokenizer
from tests.parser.engine.streaming_helpers import (
collect_tool_arguments,
simulate_tool_streaming,
)
from vllm.parser.deepseek_v32 import (
DSML_FUNC_END,
DSML_FUNC_START,
DeepSeekV32Parser,
)
from vllm.parser.deepseek_v4 import (
DSML_INVOKE_END,
DSML_INVOKE_PREFIX,
DSML_PARAM_CLOSE,
DSML_PARAM_START,
DSML_TOOL_END,
DSML_TOOL_START,
DeepSeekV4Parser,
)
from vllm.parser.deepseek_v41 import (
DSML_INVOKE_END as V41_INVOKE_END,
DSML_INVOKE_PREFIX as V41_INVOKE_PREFIX,
DSML_PARAM_CLOSE as V41_PARAM_CLOSE,
DSML_PARAM_START as V41_PARAM_START,
DSML_TOOL_END as V41_TOOL_END,
DSML_TOOL_START as V41_TOOL_START,
DeepSeekV41Parser,
)

_VOCAB = {
"<think>": 50,
"</think>": 51,
"<|DSML|tool_calls>": 52,
"</|DSML|tool_calls>": 53,
"<|DSML|function_calls>": 54,
"</|DSML|function_calls>": 55,
"<|DSML| calls>": 56,
"</|DSML| calls>": 57,
}

_TOOLS = [
{
"type": "function",
"function": {
"name": "Write",
"description": "Write a file to the given path",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string"},
"content": {"type": "string"},
},
"required": ["path", "content"],
},
},
}
]

# Long enough that buffering it would hold the client silent, and free of
# ``'>'`` so the structural-char gate never opens inside the value.
_BODY = (
"研究报告正文,包含标点、数字 123 与换行,"
"但没有任何标签尖括号。\n"
) * 40

# A value that contains ``'<'`` and a partial close-tag prefix on purpose.
_BODY_WITH_TAGS = "abc<def</ghi</|DSML|par xyz" * 40


@dataclass(frozen=True)
class _Dialect:
name: str
parser_cls: type
tool_start: str
tool_end: str
invoke_prefix: str
invoke_end: str
param_start: str
param_close: str


_DIALECTS = [
_Dialect(
name="deepseek_v4",
parser_cls=DeepSeekV4Parser,
tool_start=DSML_TOOL_START,
tool_end=DSML_TOOL_END,
invoke_prefix=DSML_INVOKE_PREFIX,
invoke_end=DSML_INVOKE_END,
param_start=DSML_PARAM_START,
param_close=DSML_PARAM_CLOSE,
),
_Dialect(
name="deepseek_v41",
parser_cls=DeepSeekV41Parser,
tool_start=V41_TOOL_START,
tool_end=V41_TOOL_END,
invoke_prefix=V41_INVOKE_PREFIX,
invoke_end=V41_INVOKE_END,
param_start=V41_PARAM_START,
param_close=V41_PARAM_CLOSE,
),
_Dialect(
name="deepseek_v32",
parser_cls=DeepSeekV32Parser,
tool_start=DSML_FUNC_START,
tool_end=DSML_FUNC_END,
invoke_prefix=DSML_INVOKE_PREFIX,
invoke_end=DSML_INVOKE_END,
param_start=DSML_PARAM_START,
param_close=DSML_PARAM_CLOSE,
),
]


def _dialect_id(dialect: _Dialect) -> str:
return dialect.name


def _write_call(dialect: _Dialect, body: str) -> str:
return (
f"{dialect.tool_start}{dialect.invoke_prefix}Write\">\n"
f'{dialect.param_start} name="path" string="true">/tmp/a.md'
f"{dialect.param_close}\n"
f'{dialect.param_start} name="content" string="true">{body}'
f"{dialect.param_close}\n"
f"{dialect.invoke_end}{dialect.tool_end}"
)


def _stream(dialect: _Dialect, body: str, request, chunk_size: int = 24):
parser = dialect.parser_cls(make_mock_tokenizer(_VOCAB), tools=_TOOLS)
text = _write_call(dialect, body)
chunks = [text[i : i + chunk_size] for i in range(0, len(text), chunk_size)]
results = simulate_tool_streaming(parser, request, chunks)
fragments = [
tc.function.arguments
for delta, _ in results
if delta and delta.tool_calls
for tc in delta.tool_calls
if tc.function and tc.function.arguments
]
return collect_tool_arguments(results), fragments


@pytest.mark.parametrize("dialect", _DIALECTS, ids=_dialect_id)
def test_long_value_streams_while_generated(dialect, mock_request):
mock_request.tools = _TOOLS
arguments, fragments = _stream(dialect, _BODY, mock_request)

assert json.loads(arguments) == {"path": "/tmp/a.md", "content": _BODY}
# The value must trickle out while it is generated instead of arriving as
# a single block once the closing tag lands.
assert len(fragments) > 20
assert max(len(fragment) for fragment in fragments) < len(_BODY) // 4


@pytest.mark.parametrize("dialect", _DIALECTS, ids=_dialect_id)
def test_value_with_angle_brackets_is_not_corrupted(dialect, mock_request):
mock_request.tools = _TOOLS
arguments, fragments = _stream(dialect, _BODY_WITH_TAGS, mock_request)

assert json.loads(arguments) == {"path": "/tmp/a.md", "content": _BODY_WITH_TAGS}
assert all("<|DSML|parameter" not in fragment for fragment in fragments)
4 changes: 4 additions & 0 deletions vllm/parser/deepseek_v32.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
DSML_PARAM_START,
_dsml_arg_converter,
_unwrap_wrapper_args,
make_param_value_extender,
)
from vllm.parser.engine.events import EventType
from vllm.parser.engine.parser_engine import ParserEngine
Expand Down Expand Up @@ -115,6 +116,9 @@ def deepseek_v32_config() -> ParserEngineConfig:
},
arg_converter=_dsml_arg_converter,
arg_structural_chars=frozenset(">"),
arg_value_extender=make_param_value_extender(
DSML_PARAM_START, DSML_PARAM_CLOSE, (DSML_INVOKE_END,)
),
strip_content_whitespace_with_tools=False,
tool_args_json=False,
)
Expand Down
60 changes: 59 additions & 1 deletion vllm/parser/deepseek_v4.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,18 +21,20 @@
import contextlib
import functools
import json
from collections.abc import Callable
from typing import TYPE_CHECKING

import regex as re

from vllm.parser.engine.events import EventType
from vllm.parser.engine.parser_engine import ParserEngine
from vllm.parser.engine.parser_engine_config import (
ArgValueExtender,
ParserEngineConfig,
ParserState,
Transition,
)
from vllm.tool_parsers.utils import find_tool_properties
from vllm.tool_parsers.utils import find_tool_properties, partial_tag_overlap

if TYPE_CHECKING:
from vllm.tokenizers import TokenizerLike
Expand Down Expand Up @@ -143,6 +145,59 @@ def _unwrap_wrapper_args(
return args_json


def make_param_value_extender(
param_start: str,
param_close: str,
extra_stop_tags: tuple[str, ...] = (),
) -> Callable[[str], ArgValueExtender | None]:
"""Build an ``arg_value_extender`` for one DSML parameter dialect.

``_dsml_arg_converter`` re-serialises the whole parameter list on every
streaming tick, and the engine's ``arg_structural_chars`` gate (``'>'``)
drops every delta that carries no tag boundary. A long ``string="true"``
value contains no ``'>'`` at all, so it used to be buffered until the
closing tag arrived: the client received no bytes for as long as the
value took to generate, which for multi-KB values exceeds the client's
read timeout.

The returned extender re-serialises only the value tail, so the engine can
emit it immediately. A trailing partial tag is held back until it either
completes (``None`` makes the engine fall back to the full converter) or
turns out to be ordinary text.
"""
param_open_re = re.compile(
rf'{re.escape(param_start)}\s+name="[^"]+"(?:\s+string="[^"]*")?\s*>'
)
stop_tags = (param_close, param_start, *extra_stop_tags)

def extender(raw_args: str) -> ArgValueExtender | None:
matches = list(param_open_re.finditer(raw_args))
if not matches:
return None
# Only extend while the last parameter's value is still open.
if any(tag in raw_args[matches[-1].end() :] for tag in stop_tags):
return None
# The engine ran a full conversion right before creating this
# extender, so the value generated so far has already been streamed.
held = ""

def extend(delta: str) -> str | None:
nonlocal held
text = held + delta
if any(tag in text for tag in stop_tags):
return None
keep = max(partial_tag_overlap(text, tag) for tag in stop_tags)
visible = text[:-keep] if keep else text
held = text[len(visible) :]
if not visible:
return ""
return json.dumps(visible, ensure_ascii=False)[1:-1]

return extend

return extender


@functools.cache
def deepseek_v4_config(thinking: bool = False) -> ParserEngineConfig:
return ParserEngineConfig(
Expand Down Expand Up @@ -238,6 +293,9 @@ def deepseek_v4_config(thinking: bool = False) -> ParserEngineConfig:
},
arg_converter=_dsml_arg_converter,
arg_structural_chars=frozenset(">"),
arg_value_extender=make_param_value_extender(
DSML_PARAM_START, DSML_PARAM_CLOSE, (DSML_INVOKE_END,)
),
strip_content_whitespace_with_tools=False,
tool_args_json=False,
)
Expand Down
4 changes: 4 additions & 0 deletions vllm/parser/deepseek_v41.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
_dsml_arg_converter,
_param_patterns,
deepseek_v4_config,
make_param_value_extender,
)
from vllm.parser.engine.parser_engine_config import ParserEngineConfig

Expand Down Expand Up @@ -47,6 +48,9 @@ def deepseek_v41_config(thinking: bool = False) -> ParserEngineConfig:
param_re=_PARAM_RE,
partial_param_re=_PARTIAL_PARAM_RE,
),
arg_value_extender=make_param_value_extender(
DSML_PARAM_START, DSML_PARAM_CLOSE, (DSML_INVOKE_END,)
),
)


Expand Down
Loading