[Render] Add reasoning/tool parsing to /derender + fix byte-fallback FFFD (#45919)

Signed-off-by: aoshen524 <aoshen524@gmail.com>
Co-authored-by: Martin Hickey <martin.hickey@ie.ibm.com>
This commit is contained in:
aoshen02
2026-06-20 19:43:32 -04:00
committed by GitHub
co-authored by Martin Hickey
parent 6e919960af
commit f57ac274b2
4 changed files with 601 additions and 37 deletions
@@ -8,6 +8,7 @@ import pytest
import pytest_asyncio
from tests.utils import RemoteLaunchRenderServer
from vllm.tokenizers import get_tokenizer
MODEL_NAME = "hmellor/tiny-random-LlamaForCausalLM"
@@ -486,3 +487,438 @@ async def test_derender_completion_kv_transfer_params_passthrough(client):
)
assert response.status_code == 200
assert response.json()["kv_transfer_params"] == kv
# ---------------------------------------------------------------------------
# E2E: render -> derender roundtrip with parser (reasoning + tool calls)
# ---------------------------------------------------------------------------
PARSER_MODEL = "deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B"
_E2E_TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather for a city",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
},
},
}
]
@pytest.fixture(scope="module")
def parser_server():
args = [
"--enable-auto-tool-choice",
"--tool-call-parser",
"hermes",
"--reasoning-parser",
"deepseek_r1",
]
with RemoteLaunchRenderServer(PARSER_MODEL, args) as remote_server:
yield remote_server
@pytest_asyncio.fixture
async def parser_client(parser_server):
async with httpx.AsyncClient(
base_url=parser_server.url_for(""), timeout=60.0
) as http_client:
yield http_client
@pytest.fixture(scope="module")
def parser_tokenizer():
return get_tokenizer(PARSER_MODEL)
def _encode(tokenizer, text: str) -> list[int]:
return tokenizer.encode(text, add_special_tokens=False)
def _decoded(tokenizer, token_ids: list[int]) -> str:
return tokenizer.decode(token_ids, skip_special_tokens=True)
def _require_markers_survive(tokenizer, text: str, *markers: str) -> list[int]:
"""Encode text and skip the test if any marker is lost in roundtrip."""
ids = _encode(tokenizer, text)
decoded = tokenizer.decode(ids, skip_special_tokens=False)
for m in markers:
if m not in decoded:
pytest.skip(f"Marker {m!r} lost in encode->decode roundtrip")
return ids
async def _e2e_render_chat(
client: httpx.AsyncClient,
model: str,
messages: list[dict],
) -> dict:
resp = await client.post(
"/v1/chat/completions/render",
json={"model": model, "messages": messages},
)
assert resp.status_code == 200, resp.text
return resp.json()
def _e2e_generate_response(
token_ids: list[int],
request_id: str = "chatcmpl-e2e-test",
) -> dict:
return {
"request_id": request_id,
"choices": [
{
"index": 0,
"token_ids": token_ids,
"finish_reason": "stop",
}
],
}
@pytest.mark.asyncio
async def test_e2e_plain_roundtrip(parser_client, parser_tokenizer):
"""Plain text without reasoning markers roundtrips correctly."""
messages = [{"role": "user", "content": "What is 2+2?"}]
gen_req = await _e2e_render_chat(parser_client, PARSER_MODEL, messages)
answer = "The answer is four."
output_ids = _encode(parser_tokenizer, answer)
expected = _decoded(parser_tokenizer, output_ids)
resp = await parser_client.post(
"/v1/chat/completions/derender",
json={
"model": PARSER_MODEL,
"generate_response": _e2e_generate_response(output_ids),
"prompt_tokens": len(gen_req["token_ids"]),
},
)
assert resp.status_code == 200, resp.text
content = resp.json()["choices"][0]["message"]["content"]
assert content == expected
@pytest.mark.asyncio
async def test_e2e_token_identity(parser_client, parser_tokenizer):
"""encode(derender(token_ids)) == token_ids (RL invariant)."""
messages = [{"role": "user", "content": "Hi"}]
gen_req = await _e2e_render_chat(parser_client, PARSER_MODEL, messages)
answer = "Hello! How can I help?"
output_ids = _encode(parser_tokenizer, answer)
resp = await parser_client.post(
"/v1/chat/completions/derender",
json={
"model": PARSER_MODEL,
"generate_response": _e2e_generate_response(output_ids),
"prompt_tokens": len(gen_req["token_ids"]),
},
)
assert resp.status_code == 200
content = resp.json()["choices"][0]["message"]["content"]
re_encoded = _encode(parser_tokenizer, content)
assert output_ids == re_encoded
@pytest.mark.asyncio
async def test_e2e_non_ascii_roundtrip(parser_client, parser_tokenizer):
"""CJK + emoji roundtrip without U+FFFD."""
messages = [{"role": "user", "content": "Reply in Chinese"}]
gen_req = await _e2e_render_chat(parser_client, PARSER_MODEL, messages)
answer = "你好世界 😀"
output_ids = _encode(parser_tokenizer, answer)
resp = await parser_client.post(
"/v1/chat/completions/derender",
json={
"model": PARSER_MODEL,
"generate_response": _e2e_generate_response(output_ids),
"prompt_tokens": len(gen_req["token_ids"]),
},
)
assert resp.status_code == 200
content = resp.json()["choices"][0]["message"]["content"]
assert "�" not in content
@pytest.mark.asyncio
async def test_e2e_parsed_reasoning(parser_client, parser_tokenizer):
"""<think>...</think> splits into reasoning + content."""
messages = [{"role": "user", "content": "What is 2+3?"}]
gen_req = await _e2e_render_chat(parser_client, PARSER_MODEL, messages)
reasoning_text = "The user wants 2 plus 3. That is 5."
answer_text = "The answer is 5."
output_text = f"<think>{reasoning_text}</think>{answer_text}"
output_ids = _require_markers_survive(parser_tokenizer, output_text, "</think>")
resp = await parser_client.post(
"/v1/chat/completions/derender",
json={
"model": PARSER_MODEL,
"generate_response": _e2e_generate_response(output_ids),
"prompt_tokens": len(gen_req["token_ids"]),
"chat_request": {
"model": PARSER_MODEL,
"messages": messages,
"include_reasoning": True,
},
},
)
assert resp.status_code == 200, resp.text
msg = resp.json()["choices"][0]["message"]
assert msg["reasoning"] is not None
assert reasoning_text in msg["reasoning"]
assert answer_text in msg["content"]
assert "<think>" not in msg["content"]
@pytest.mark.asyncio
async def test_e2e_parsed_tool_call(parser_client, parser_tokenizer):
"""<tool_call> extracted into tool_calls field."""
messages = [{"role": "user", "content": "Weather in Paris?"}]
gen_req = await _e2e_render_chat(parser_client, PARSER_MODEL, messages)
output_text = (
"<think>Let me check the weather.</think>"
'<tool_call>\n{"name": "get_weather", '
'"arguments": {"city": "Paris"}}\n</tool_call>'
)
output_ids = _require_markers_survive(
parser_tokenizer,
output_text,
"</think>",
"<tool_call>",
"</tool_call>",
)
resp = await parser_client.post(
"/v1/chat/completions/derender",
json={
"model": PARSER_MODEL,
"generate_response": _e2e_generate_response(output_ids),
"prompt_tokens": len(gen_req["token_ids"]),
"chat_request": {
"model": PARSER_MODEL,
"messages": messages,
"tools": _E2E_TOOLS,
"tool_choice": "auto",
},
},
)
assert resp.status_code == 200, resp.text
choice = resp.json()["choices"][0]
assert choice["message"]["tool_calls"]
assert choice["message"]["tool_calls"][0]["function"]["name"] == "get_weather"
@pytest.mark.asyncio
async def test_e2e_parsed_reasoning_and_tool_call(parser_client, parser_tokenizer):
"""Reasoning + tool call in the same output."""
messages = [{"role": "user", "content": "Weather in Paris?"}]
gen_req = await _e2e_render_chat(parser_client, PARSER_MODEL, messages)
reasoning_text = "I should look up the weather."
tool_text = (
'<tool_call>\n{"name": "get_weather", '
'"arguments": {"city": "Paris"}}\n</tool_call>'
)
output_text = f"<think>{reasoning_text}</think>{tool_text}"
output_ids = _require_markers_survive(
parser_tokenizer, output_text, "</think>", "<tool_call>"
)
resp = await parser_client.post(
"/v1/chat/completions/derender",
json={
"model": PARSER_MODEL,
"generate_response": _e2e_generate_response(output_ids),
"prompt_tokens": len(gen_req["token_ids"]),
"chat_request": {
"model": PARSER_MODEL,
"messages": messages,
"tools": _E2E_TOOLS,
"tool_choice": "auto",
"include_reasoning": True,
},
},
)
assert resp.status_code == 200, resp.text
choice = resp.json()["choices"][0]
assert choice["message"]["reasoning"] is not None
assert reasoning_text in choice["message"]["reasoning"]
assert choice["message"]["tool_calls"]
@pytest.mark.asyncio
async def test_e2e_no_chat_request_fallback(parser_client, parser_tokenizer):
"""Without chat_request, derender falls back to plain detokenization."""
messages = [{"role": "user", "content": "Hello"}]
gen_req = await _e2e_render_chat(parser_client, PARSER_MODEL, messages)
answer = "Hi there!"
output_ids = _encode(parser_tokenizer, answer)
resp = await parser_client.post(
"/v1/chat/completions/derender",
json={
"model": PARSER_MODEL,
"generate_response": _e2e_generate_response(output_ids),
"prompt_tokens": len(gen_req["token_ids"]),
},
)
assert resp.status_code == 200
content = resp.json()["choices"][0]["message"]["content"]
assert "Hi" in content
# ---------------------------------------------------------------------------
# E2E: HarmonyParser + GPT-OSS
# ---------------------------------------------------------------------------
HARMONY_MODEL = "openai/gpt-oss-20b"
def _ensure_harmony_vocab():
"""Pre-cache the o200k_base BPE file needed by openai-harmony.
The Rust tiktoken-rs backend downloads from Azure Blob Storage, which
may be unreachable in some environments. When the cache is cold we
fetch the file ourselves and place it in ``/tmp/tiktoken-rs-cache/``
using the SHA-1(URL) filename that tiktoken-rs expects.
"""
import hashlib
import urllib.request
from pathlib import Path
url = "https://openaipublic.blob.core.windows.net/encodings/o200k_base.tiktoken"
cache_dir = Path("/tmp/tiktoken-rs-cache")
cache_key = hashlib.sha1(url.encode()).hexdigest()
cache_file = cache_dir / cache_key
if not cache_file.exists():
cache_dir.mkdir(parents=True, exist_ok=True)
urllib.request.urlretrieve(url, cache_file)
@pytest.fixture(scope="module")
def harmony_server():
_ensure_harmony_vocab()
args = [
"--trust-remote-code",
"--enable-auto-tool-choice",
"--tool-call-parser",
"openai",
"--reasoning-parser",
"openai_gptoss",
]
with RemoteLaunchRenderServer(HARMONY_MODEL, args) as remote_server:
yield remote_server
@pytest_asyncio.fixture
async def harmony_client(harmony_server):
async with httpx.AsyncClient(
base_url=harmony_server.url_for(""), timeout=60.0
) as http_client:
yield http_client
@pytest.fixture(scope="module")
def harmony_tokenizer():
return get_tokenizer(HARMONY_MODEL, trust_remote_code=True)
def _harmony_extract_assistant_ids(
tokenizer, assistant_msg: dict, user_content: str = "test"
) -> list[int]:
"""Extract assistant token IDs via apply_chat_template diff."""
prompt = [{"role": "user", "content": user_content}]
full = prompt + [assistant_msg]
text_prompt = tokenizer.apply_chat_template(
prompt, add_generation_prompt=True, tokenize=False
)
text_full = tokenizer.apply_chat_template(
full, add_generation_prompt=False, tokenize=False
)
prompt_ids = tokenizer.encode(text_prompt)
full_ids = tokenizer.encode(text_full)
assistant_ids = list(full_ids[len(prompt_ids) :])
if not assistant_ids:
pytest.skip("Could not extract assistant tokens for Harmony")
return assistant_ids
@pytest.mark.asyncio
async def test_e2e_harmony_plain_roundtrip(harmony_client, harmony_tokenizer):
"""GPT-OSS content-only roundtrip."""
messages = [{"role": "user", "content": "What is 2+2?"}]
gen_req = await _e2e_render_chat(harmony_client, HARMONY_MODEL, messages)
assistant_msg = {"role": "assistant", "content": "Four."}
output_ids = _harmony_extract_assistant_ids(harmony_tokenizer, assistant_msg)
resp = await harmony_client.post(
"/v1/chat/completions/derender",
json={
"model": HARMONY_MODEL,
"generate_response": _e2e_generate_response(output_ids),
"prompt_tokens": len(gen_req["token_ids"]),
"chat_request": {
"model": HARMONY_MODEL,
"messages": messages,
},
},
)
assert resp.status_code == 200, resp.text
content = resp.json()["choices"][0]["message"]["content"]
assert content is not None and len(content) > 0
assert "Four" in content
@pytest.mark.asyncio
async def test_e2e_harmony_reasoning(harmony_client, harmony_tokenizer):
"""GPT-OSS reasoning: analysis channel extracted."""
messages = [{"role": "user", "content": "Add 2 and 3."}]
gen_req = await _e2e_render_chat(harmony_client, HARMONY_MODEL, messages)
reasoning_text = "The user wants 2 plus 3."
answer_text = "The answer is 5."
assistant_msg = {
"role": "assistant",
"thinking": reasoning_text,
"content": answer_text,
}
output_ids = _harmony_extract_assistant_ids(harmony_tokenizer, assistant_msg)
decoded = harmony_tokenizer.decode(output_ids)
if reasoning_text not in decoded:
pytest.skip("Harmony template did not render thinking")
resp = await harmony_client.post(
"/v1/chat/completions/derender",
json={
"model": HARMONY_MODEL,
"generate_response": _e2e_generate_response(output_ids),
"prompt_tokens": len(gen_req["token_ids"]),
"chat_request": {
"model": HARMONY_MODEL,
"messages": messages,
"include_reasoning": True,
},
},
)
assert resp.status_code == 200, resp.text
msg = resp.json()["choices"][0]["message"]
assert msg["reasoning"] is not None
assert reasoning_text in msg["reasoning"]
assert answer_text in (msg["content"] or "")
+1 -1
View File
@@ -455,7 +455,7 @@ async def init_render_app_state(
enable_auto_tools=args.enable_auto_tool_choice,
exclude_tools_when_tool_choice_none=args.exclude_tools_when_tool_choice_none,
tool_parser=args.tool_call_parser,
reasoning_parser=args.structured_outputs_config.reasoning_parser,
reasoning_parser=args.reasoning_parser,
default_chat_template_kwargs=args.default_chat_template_kwargs,
log_error_stack=args.log_error_stack,
)
+8 -4
View File
@@ -219,10 +219,14 @@ class GenerateResponse(BaseModel):
class DerenderChatRequest(BaseModel):
"""Request for the /v1/chat/completions/derender endpoint.
"""Request for the /v1/chat/completions/derender endpoint (non-streaming).
Wraps a GenerateResponse and caller-supplied metadata needed to produce
a fully-formed ChatCompletionResponse without a GPU.
Wraps a complete GenerateResponse and caller-supplied metadata needed to
produce a fully-formed ChatCompletionResponse without a GPU.
Streaming derender would require a separate endpoint design with
incremental token delivery, ``OutputProcessor``-based detokenization,
and ``parser.parse_delta()`` instead of ``parser.parse()``.
"""
model: str
@@ -244,7 +248,7 @@ class DerenderChatRequest(BaseModel):
class DerenderCompletionRequest(BaseModel):
"""Request for the /v1/completions/derender endpoint.
"""Request for the /v1/completions/derender endpoint (non-streaming).
Parallel to DerenderChatRequest but handles the multi-prompt completions
case: one GenerateResponse per prompt, mirroring the list[GenerateRequest]
+156 -32
View File
@@ -27,6 +27,7 @@ from vllm.entrypoints.openai.completion.protocol import (
)
from vllm.entrypoints.openai.engine.protocol import (
ErrorResponse,
ToolCall,
UsageInfo,
)
from vllm.entrypoints.openai.engine.serving import resolve_token_id_placeholder
@@ -43,7 +44,6 @@ from vllm.entrypoints.serve.disagg.protocol import (
DerenderChatRequest,
DerenderCompletionRequest,
GenerateRequest,
GenerateResponseChoice,
MultiModalFeatures,
PlaceholderRangeInfo,
)
@@ -76,21 +76,83 @@ from vllm.utils.mistral import mt as _mt
logger = init_logger(__name__)
def _parse_token_id_placeholder(token: str) -> int | None:
"""Extract token ID from a 'token_id:N' placeholder string."""
if not token.startswith("token_id:"):
return None
try:
return int(token[len("token_id:") :])
except ValueError:
return None
def _correct_decoded_token(
token_id: int, context_token_ids: list[int], tokenizer: TokenizerLike
) -> str:
"""Use preceding tokens as context to fix U+FFFD from byte-fallback.
Mirrors LogprobsProcessor._correct_decoded_token in v1/engine/logprobs.py.
"""
max_ctx = min(len(context_token_ids), 4)
for num_ctx in range(1, max_ctx + 1):
context = context_token_ids[-num_ctx:]
full_decoded = tokenizer.decode(context + [token_id])
if full_decoded.endswith("�"):
continue
clean_end = len(context)
for j in range(len(context) - 1, -1, -1):
if tokenizer.decode([context[j]]).endswith("�"):
clean_end = j
else:
break
clean_prefix = tokenizer.decode(context[:clean_end]) if clean_end > 0 else ""
if full_decoded.startswith(clean_prefix):
return full_decoded[len(clean_prefix) :]
common_len = 0
for a, b in zip(clean_prefix, full_decoded):
if a != b:
break
common_len += 1
return full_decoded[common_len:]
return ""
def _resolve_logprobs(
logprobs: ChatCompletionLogProbs, tokenizer: TokenizerLike
) -> ChatCompletionLogProbs:
"""Resolve all token_id:N placeholders in a ChatCompletionLogProbs object."""
"""Resolve token_id:N placeholders in a ChatCompletionLogProbs object."""
if logprobs.content is None:
return logprobs
context_token_ids: list[int] = []
resolved_content = []
for entry in logprobs.content:
token_str, token_bytes = resolve_token_id_placeholder(entry.token, tokenizer)
sampled_id = _parse_token_id_placeholder(entry.token)
if token_str.endswith("�") and sampled_id is not None:
token_str = _correct_decoded_token(sampled_id, context_token_ids, tokenizer)
token_bytes = list(token_str.encode("utf-8"))
resolved_top = []
for top in entry.top_logprobs:
top_str, top_bytes = resolve_token_id_placeholder(top.token, tokenizer)
top_id = _parse_token_id_placeholder(top.token)
if top_str.endswith("�") and top_id is not None:
top_str = _correct_decoded_token(top_id, context_token_ids, tokenizer)
top_bytes = list(top_str.encode("utf-8"))
resolved_top.append(
top.model_copy(update={"token": top_str, "bytes": top_bytes})
)
resolved_content.append(
entry.model_copy(
update={
@@ -100,6 +162,10 @@ def _resolve_logprobs(
}
)
)
if sampled_id is not None:
context_token_ids.append(sampled_id)
return ChatCompletionLogProbs(content=resolved_content)
@@ -136,30 +202,6 @@ def _convert_chat_logprobs_to_completion_logprobs(
)
def _build_chat_choice(
choice: GenerateResponseChoice, tokenizer: TokenizerLike
) -> ChatCompletionResponseChoice:
"""Detokenize and resolve logprobs for a single GenerateResponseChoice.
Raises:
ValueError: if choice.token_ids is empty or None.
"""
if not choice.token_ids:
raise ValueError(f"choice {choice.index} has empty or null token_ids")
decoded_text = tokenizer.decode(choice.token_ids, skip_special_tokens=True)
resolved_logprobs = (
_resolve_logprobs(choice.logprobs, tokenizer)
if choice.logprobs is not None
else None
)
return ChatCompletionResponseChoice(
index=choice.index,
message=ChatMessage(role="assistant", content=decoded_text),
logprobs=resolved_logprobs,
finish_reason=choice.finish_reason,
)
class OpenAIServingRender:
def __init__(
self,
@@ -536,9 +578,12 @@ class OpenAIServingRender:
) -> ChatCompletionResponse | ErrorResponse:
"""Postprocess a GenerateResponse into a ChatCompletionResponse.
This is the symmetric inverse of render_chat_request: it detokenizes
output token IDs, resolves token_id:N logprob placeholders, and
formats the result as an OpenAI-compatible chat completion response.
Non-streaming only: expects the complete GenerateResponse with all
token IDs present. Uses ``parser.parse()`` for one-shot extraction.
When ``request.chat_request`` is provided, the parser splits the
output into (reasoning, content, tool_calls). Otherwise falls
back to plain detokenization.
"""
error_check_ret = await self._check_model(request)
if error_check_ret is not None:
@@ -546,11 +591,89 @@ class OpenAIServingRender:
tokenizer = self.renderer.get_tokenizer()
gen = request.generate_response
chat_request = request.chat_request
choices: list[ChatCompletionResponseChoice] = []
try:
for choice in gen.choices:
choices.append(_build_chat_choice(choice, tokenizer))
if not choice.token_ids:
raise ValueError(
f"choice {choice.index} has empty or null token_ids"
)
resolved_logprobs = (
_resolve_logprobs(choice.logprobs, tokenizer)
if choice.logprobs is not None
else None
)
if self.parser is not None and chat_request is not None:
# Parser path: decode with special tokens preserved
# so the parser can see markers like </think>,
# <tool_call>, or Harmony channel tokens.
decoded_text = tokenizer.decode(
choice.token_ids, skip_special_tokens=False
)
chat_template_kwargs: dict[str, Any] = {}
if not self.use_harmony:
chat_template_kwargs = (
chat_request.build_chat_params(
self.chat_template,
self.chat_template_content_format,
)
.with_defaults(self.default_chat_template_kwargs)
.chat_template_kwargs
)
parser = self.parser(
tokenizer,
chat_request.tools,
chat_template_kwargs=chat_template_kwargs,
)
reasoning, content, tool_calls = parser.parse(
decoded_text,
chat_request,
enable_auto_tools=self.enable_auto_tools,
model_output_token_ids=choice.token_ids,
)
if not getattr(chat_request, "include_reasoning", True):
reasoning = None
tc_items = (
[
ToolCall(
id=random_uuid(),
function=tc,
)
for tc in tool_calls
]
if tool_calls
else []
)
message = ChatMessage(
role="assistant",
reasoning=reasoning,
content=content,
tool_calls=tc_items,
)
else:
# No parser: plain detokenization.
decoded_text = tokenizer.decode(
choice.token_ids, skip_special_tokens=True
)
message = ChatMessage(role="assistant", content=decoded_text)
choices.append(
ChatCompletionResponseChoice(
index=choice.index,
message=message,
logprobs=resolved_logprobs,
finish_reason=choice.finish_reason,
)
)
except ValueError as exc:
return self.create_error_response(str(exc))
@@ -587,8 +710,9 @@ class OpenAIServingRender:
) -> CompletionResponse | ErrorResponse:
"""Postprocess a list of GenerateResponses into a CompletionResponse.
Mirrors the multi-prompt completions case: one GenerateResponse per
prompt, parallel to the list[GenerateRequest] from /v1/completions/render.
Non-streaming only. Mirrors the multi-prompt completions case: one
GenerateResponse per prompt, parallel to the list[GenerateRequest]
from /v1/completions/render.
"""
error_check_ret = await self._check_model(request)
if error_check_ret is not None: