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
38 changes: 37 additions & 1 deletion config/coordinator/coordinator.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,8 @@ pipeline:
# ec-shared-storage - shared filesystem (default for non-RDMA clusters)
ec_connector: ec-shared-storage

# use_openai_format selects the wire format for the encode and prefill steps.
# use_openai_format selects the wire format for the encode, prefill and
# decode steps.
# true (default) - forward on the client's original path
# (/v1/chat/completions, /v1/completions, /v1/responses,
# /v1/messages).
Expand Down Expand Up @@ -450,3 +451,38 @@ pipeline:
- type: decode
# params:
# kv_connector: kv-nixl # overrides pipeline.kv_connector
#
# output_mode selects what a tokens-in decode leg asks the worker to
# return. Read only when the leg emits the generate format
# (use_openai_format: false); the OpenAI-compatible routes have no
# such field. Both values require a worker that loads a tokenizer:
# vLLM answers 400 under --tokens-only or --skip-tokenizer-init.
# text (default) - token IDs plus the detokenized text.
# derender - token IDs plus the parsed assistant message
# (content, reasoning, tool_calls), so the worker
# applies its own parsers. vLLM refuses it
# together with stream: true, so the leg is sent
# unstreamed and a streaming client receives its
# answer in a single frame. On a worker with a
# tool or reasoning parser configured it also
# needs a parse_context from /render, which this
# pipeline does not send.
# output_mode: text
#
# wrap_chat_completion rebuilds the chat completions envelope around a
# rewritten leg's response, for a client that called
# /v1/chat/completions on a tokens-in pipeline. vLLM serves output_mode
# on the generate envelope rather than returning an OpenAI one, so
# without this the client reads a generate response: choices carry
# token_ids beside their message, and the body names the request
# request_id rather than id. Default false, which forwards the
# generate response as it arrived.
#
# Only a non-streaming chat-completions request is rewritten. A
# streamed answer is a sequence of chat.completion.chunk frames this
# step does not assemble, so it is forwarded unchanged; the other
# client APIs keep the envelope their own schema declares.
#
# The rewritten leg is not streamed to the client but read whole (up
# to 8 MB) so the envelope can be rebuilt around it.
# wrap_chat_completion: false
24 changes: 24 additions & 0 deletions pkg/common/request/apitype.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,30 @@ const (
PathSGLangGenerate = "/generate"
)

// output_mode values of the vLLM generate API, naming what a response carries
// beyond its token IDs. The field defaults to tokens, and a server
// predating it ignores the value and answers with token IDs either way.
//
// OutputModeText adds the detokenized text to each choice. vLLM rejects it
// with 400 on a worker that loads no tokenizer (--tokens-only,
// --skip-tokenizer-init) or whose request sets sampling_params.detokenize to
// false.
//
// OutputModeDerender adds the parsed assistant message (content, reasoning,
// tool_calls) instead, applying the engine's own parsers in the decode call.
// vLLM rejects it with 400 alongside stream: true, and on a server with a tool
// or reasoning parser configured it also requires a parse_context built by
// /render. It postdates OutputModeText, so a worker that knows only the
// earlier modes answers 400 for an unknown mode.
//
// The default, tokens, has no constant here: the coordinator never asks for it.
// Decode streams its response to the client, so a token-IDs-only body would
// reach the caller with no text in it.
const (
OutputModeText = "text"
OutputModeDerender = "derender"
)

// APIType is the inference API a request was sent to. Path and the output
// token cap treat a value outside the constants below as
// APITypeChatCompletions; String reports it as APIType(N).
Expand Down
28 changes: 28 additions & 0 deletions pkg/common/request/constants.go
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,7 @@ const (
FieldURL = "url"
FieldImageURL = "image_url"
FieldStore = "store"
FieldOutputMode = "output_mode"
FieldMMProcessorKwargs = "mm_processor_kwargs"
FieldMediaIOKwargs = "media_io_kwargs"
FieldOutput = "output"
Expand Down Expand Up @@ -118,6 +119,33 @@ const (
FieldCachedTokens = "cached_tokens" //#nosec G101 -- JSON field name, not a credential
)

// Response body fields of the OpenAI-compatible APIs. The chat completions
// envelope names the request it answers by FieldID and labels itself by
// FieldObject; a choice carries its position, its assistant message, and why
// generation stopped. The vLLM generate API shares the choice field names but
// names the request FieldRequestID and carries no object label.
const (
FieldID = "id"
FieldObject = "object"
FieldCreated = "created"
FieldRequestID = "request_id"
FieldChoices = "choices"
FieldMessage = "message"
FieldDelta = "delta"
FieldIndex = "index"
FieldFinishReason = "finish_reason"
)

// ObjectChatCompletion is the object label of a non-streamed chat completions
// response. The streamed counterpart is ObjectChatCompletionChunk.
const (
ObjectChatCompletion = "chat.completion"
ObjectChatCompletionChunk = "chat.completion.chunk"
)

// RoleAssistant is the role of a generated message.
const RoleAssistant = "assistant"

// Server-sent event framing for streamed responses. SGLang sends the bare
// marker as an event payload, the OpenAI APIs send the framed line.
const (
Expand Down
29 changes: 29 additions & 0 deletions pkg/common/request/tokens.go
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,35 @@ func CapSingleToken(body map[string]any, apiType APIType) map[string]any {
return limits
}

// OutputTokenLimit returns the output-token cap apiType's body carries, reading
// the same fields CapSingleToken writes (APIType.tokenLimitFields). ok is false
// when the client set none, which means "the worker's own default".
//
// It is how a leg that rewrites a request into another API's shape carries the
// cap across: the chat, legacy completions, Messages and Responses APIs name it
// at the top level, while the vLLM generate API nests it under sampling_params,
// so the field is not simply copied.
//
// The value is returned as it was decoded rather than as an int: it reaches the
// model server, which validates it, and coercing here would turn a value this
// package cannot read into a cap the client never asked for.
func OutputTokenLimit(body map[string]any, apiType APIType) (any, bool) {
limits := body
if apiType == APITypeVLLMGenerate {
sp, ok := body[FieldSamplingParams].(map[string]any)
if !ok {
return nil, false
}
limits = sp
}
for _, field := range apiType.tokenLimitFields() {
if v, set := fieldValue(limits, field); set {
return v, true
}
}
return nil, false
}

// RejectStatefulResponsesFields reports an error naming the first field it
// finds that depends on state the router does not keep: previous_response_id
// and conversation reference a prior turn, background asks for an async job
Expand Down
9 changes: 5 additions & 4 deletions pkg/coordinator/pipeline/builder/builder.go
Original file line number Diff line number Diff line change
Expand Up @@ -53,11 +53,12 @@ func validatePipeline(p config.PipelineConfig) error {
}

// usesOpenAIFormatParam reports whether stepType reads the use_openai_format
// parameter. Only encode and prefill resolve their body format from it; decode
// and conditional-decode derive it directly from the request's original path
// and reject the key instead of silently ignoring it.
// parameter. encode, prefill and decode resolve their body format from it.
// conditional-decode does not: it probes the client's own route with the body
// the client sent, so it derives its format from the request's original path
// and rejects the key instead of silently ignoring it.
func usesOpenAIFormatParam(stepType string) bool {
return stepType == steps.EncodeStepName || stepType == steps.PrefillStepName
return stepType == steps.EncodeStepName || stepType == steps.PrefillStepName || stepType == steps.DecodeStepName
}

func mergePipelineDefaults(params map[string]any, cfg config.PipelineConfig) map[string]any {
Expand Down
74 changes: 49 additions & 25 deletions pkg/coordinator/pipeline/builder/builder_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -96,33 +96,57 @@ func TestBuildDoesNotInjectUseOpenAIFormatIntoDecodeOrConditionalDecode(t *testi
}
}

// TestBuildRejectsUseOpenAIFormatOverrideInCoordinatorYAML reproduces a
// coordinator.yaml carrying the deprecated per-step override this PR removed:
// decode or conditional-decode setting use_openai_format under their own
// params. Build must fail config loading rather than build a pipeline that
// silently ignores the setting.
// TestBuildRejectsUseOpenAIFormatOverrideInCoordinatorYAML covers a
// coordinator.yaml setting use_openai_format under the params of a step that
// does not read it. conditional-decode probes the client's own route with the
// body the client sent, so the setting cannot apply to it: Build must fail
// rather than build a pipeline that silently ignores it.
func TestBuildRejectsUseOpenAIFormatOverrideInCoordinatorYAML(t *testing.T) {
for _, stepType := range []string{steps.DecodeStepName, steps.ConditionalDecodeStepName} {
t.Run(stepType, func(t *testing.T) {
body := "pipeline:\n" +
" use_openai_format: true\n" +
" steps:\n" +
" - type: " + stepType + "\n" +
" params:\n" +
" use_openai_format: false\n"
path := filepath.Join(t.TempDir(), "coordinator.yaml")
if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
t.Fatalf("writing fixture: %v", err)
}
stepType := steps.ConditionalDecodeStepName
body := "pipeline:\n" +
" use_openai_format: true\n" +
" steps:\n" +
" - type: " + stepType + "\n" +
" params:\n" +
" use_openai_format: false\n"
path := filepath.Join(t.TempDir(), "coordinator.yaml")
if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
t.Fatalf("writing fixture: %v", err)
}

cfg, err := config.Load(path)
if err != nil {
t.Fatalf("Load() error = %v", err)
}
if _, err := Build(cfg, gateway.New(config.GatewayConfig{})); err == nil {
t.Fatalf("Build() expected an error for %q overriding use_openai_format", stepType)
}
})
cfg, err := config.Load(path)
if err != nil {
t.Fatalf("Load() error = %v", err)
}
if _, err := Build(cfg, gateway.New(config.GatewayConfig{})); err == nil {
t.Fatalf("Build() expected an error for %q overriding use_openai_format", stepType)
}
}

// TestBuildAcceptsDecodeUseOpenAIFormatOverride covers the decode step reading
// use_openai_format: a pipeline may set the tokens-in format for the decode leg
// alone, so Build must accept the key under decode's own params.
func TestBuildAcceptsDecodeUseOpenAIFormatOverride(t *testing.T) {
body := "pipeline:\n" +
" use_openai_format: true\n" +
" steps:\n" +
" - type: " + steps.RenderStepName + "\n" +
" params:\n" +
" address: http://render.invalid:8080\n" +
" - type: " + steps.DecodeStepName + "\n" +
" params:\n" +
" use_openai_format: false\n"
path := filepath.Join(t.TempDir(), "coordinator.yaml")
if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
t.Fatalf("writing fixture: %v", err)
}

cfg, err := config.Load(path)
if err != nil {
t.Fatalf("Load() error = %v", err)
}
if _, err := Build(cfg, gateway.New(config.GatewayConfig{})); err != nil {
t.Fatalf("Build() unexpected error: %v", err)
}
}

Expand Down
131 changes: 131 additions & 0 deletions pkg/coordinator/steps/chat_envelope.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,131 @@
/*
Copyright 2026 The llm-d Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package steps

import (
"encoding/json"
"fmt"

reqcommon "github.com/llm-d/llm-d-router/pkg/common/request"
)

// generateOnlyResponseFields are the generate-envelope fields that have no
// place in a chat completions response: the KV and EC handoff descriptors are
// coordination state between the phases, and the rest describe the prompt and
// the engine rather than the answer. They are dropped rather than passed
// through, since a client reading the OpenAI schema has nowhere to put them.
var generateOnlyResponseFields = []string{
reqcommon.FieldRequestID,
reqcommon.FieldKVTransferParams,
reqcommon.FieldECTransferParams,
"prompt_logprobs",
"prompt_token_id_logprobs",
"prompt_token_ids",
"mm_placeholders",
"metrics",
reqcommon.FieldOutputMode,
}

// generateOnlyChoiceFields are the per-choice counterparts: the token IDs the
// text or message already spells out, and the engine-level detail a chat
// completions choice does not declare.
var generateOnlyChoiceFields = []string{
"token_ids",
"routed_experts",
"sampling_mask",
"weight_versions",
}

// wrapChatCompletion rewrites a vLLM generate response into a chat completions
// response. vLLM serves output_mode on the generate envelope rather than
// returning an OpenAI one (the response schema would otherwise depend on a
// request field), so a client that called /v1/chat/completions and had its
// request rewritten to the generate route needs the envelope put back.
//
// It carries over what the two schemas share, drops the generate-only fields,
// and labels the result. A choice's message is moved across as it arrived, so
// the reasoning and tool_calls an output_mode of derender produces survive
// without this package modeling them.
//
// A response whose choices carry text rather than a message (output_mode text)
// is given the message a chat completions client reads, built from that text.
//
// requestID seeds the id field: the generate response's own request_id is the
// one the coordinator sent, so the two agree, and a response that omitted it
// still names the request the client asked about.
//
// wantMode is the output_mode the leg asked for. A worker predating the field
// ignores it and answers with token IDs only, which this rewrite would hand to
// the client as an empty message, so a response not echoing the mode is an
// error rather than an answer.
func wrapChatCompletion(generate []byte, requestID, wantMode string) ([]byte, error) {
var resp map[string]any
if err := json.Unmarshal(generate, &resp); err != nil {
return nil, fmt.Errorf("decoding generate response: %w", err)
}

// Absent means tokens, the default of a worker that knows the field and of
// one that does not.
gotMode, _ := resp[reqcommon.FieldOutputMode].(string)
if gotMode != wantMode {
return nil, fmt.Errorf("worker answered output_mode %q, want %q: it carries no text to answer with",
gotMode, wantMode)
}

id, _ := resp[reqcommon.FieldRequestID].(string)
if id == "" {
id = requestID
}

choices, _ := resp[reqcommon.FieldChoices].([]any)
for i, raw := range choices {
choice, ok := raw.(map[string]any)
if !ok {
continue
}
if _, hasMessage := choice[reqcommon.FieldMessage]; !hasMessage {
text, _ := choice["text"].(string)
choice[reqcommon.FieldMessage] = map[string]any{
reqcommon.FieldRole: reqcommon.RoleAssistant,
reqcommon.FieldContent: text,
}
}
delete(choice, "text")
for _, field := range generateOnlyChoiceFields {
delete(choice, field)
}
choices[i] = choice
}

for _, field := range generateOnlyResponseFields {
delete(resp, field)
}
resp[reqcommon.FieldID] = id
resp[reqcommon.FieldObject] = reqcommon.ObjectChatCompletion
// choices is required by the schema, so an answer that carried none is
// still well-formed rather than missing the key.
if choices == nil {
choices = []any{}
}
resp[reqcommon.FieldChoices] = choices

wrapped, err := json.Marshal(resp)
if err != nil {
return nil, fmt.Errorf("encoding chat completion response: %w", err)
}
return wrapped, nil
}
Loading
Loading