Skip to content
Merged
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
22 changes: 8 additions & 14 deletions rust/src/server/src/routes/openai/responses.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ pub mod types;

mod convert;
mod streaming;
mod tools;
mod validate;

use std::convert::Infallible;
Expand Down Expand Up @@ -87,24 +88,20 @@ pub async fn create_responses(
created_at,
);

let items = OutputItemStreamer::new(prepared.include_reasoning, prepared.tool_names);
if stream_requested {
let event_stream = responses_event_stream(
chat_stream,
response,
prepared.include_reasoning,
items,
api_server_options.enable_per_request_metrics,
);
let sse_stream = responses_sse_stream(event_stream).instrument(request_span);
sse_response(sse_stream, api_server_options.sse_keep_alive_interval)
} else {
match collect_responses(
chat_stream,
response,
prepared.include_reasoning,
&api_server_options,
)
.instrument(request_span)
.await
match collect_responses(chat_stream, response, items, &api_server_options)
.instrument(request_span)
.await
{
Ok(response) => Json(response).into_response(),
Err(error) => error.into_response(),
Expand Down Expand Up @@ -141,14 +138,13 @@ fn response_not_found(response_id: &str) -> ApiError {
async fn collect_responses(
mut stream: ChatEventStream,
response: ResponsesResponse,
include_reasoning: bool,
mut items: OutputItemStreamer,
ApiServerOptions {
enable_log_requests,
enable_per_request_metrics,
..
}: &ApiServerOptions,
) -> Result<ResponsesResponse, ApiError> {
let mut items = OutputItemStreamer::new(include_reasoning);
let terminal = loop {
let Some(event) = stream.next().await else {
return Err(server_error!(
Expand Down Expand Up @@ -296,12 +292,10 @@ fn failed_response(
async fn responses_event_stream(
mut stream: impl ChatEventStreamTrait + Unpin,
response: ResponsesResponse,
include_reasoning: bool,
mut items: OutputItemStreamer,
enable_per_request_metrics: bool,
mut y: TryYielder<ResponseStreamEvent, Infallible>,
) -> Result<(), Infallible> {
let mut items = OutputItemStreamer::new(include_reasoning);

let initial = Arc::new(response);
y.yield_ok(ResponseStreamEvent::Created {
response: Arc::clone(&initial),
Expand Down
104 changes: 89 additions & 15 deletions rust/src/server/src/routes/openai/responses/convert.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,12 +16,13 @@ use thiserror_ext::AsReport as _;
use tracing::warn;
use vllm_chat::{
AssistantContentBlock, AssistantToolCall, ChatContent, ChatContentPart, ChatMessage,
ChatOptions, ChatRequest, ChatTool, ChatToolChoice, Error as ChatError, GenerationPromptMode,
ChatOptions, ChatRequest, ChatToolChoice, Error as ChatError, GenerationPromptMode,
ResolvedToolContext,
};
use vllm_text::output::TextDecodeOptions;
use vllm_text::{EffectiveSampling, SamplingParams};

use super::tools::{ResponsesToolNames, ToolIdentity};
use super::types::{
InputTokensDetails, OutputTokensDetails, ResponseInputContentPart, ResponseInputItem,
ResponseInputReasoning, ResponseInputRole, ResponseItemStatus, ResponseMessageContent,
Expand Down Expand Up @@ -70,6 +71,8 @@ pub(crate) struct PreparedResponsesRequest {
pub meta: ResponseMeta,
/// Whether reasoning items appear in the output.
pub include_reasoning: bool,
/// Maps called tool names back to namespaced functions in the output.
pub tool_names: ResponsesToolNames,
pub chat_request: ChatRequest,
}

Expand Down Expand Up @@ -148,11 +151,9 @@ pub(crate) fn prepare_responses_request(
GenerationPromptMode::StartNewAssistant
};

let converted_tools = tools
.iter()
.map(|ResponsesTool::Function(function)| ChatTool::from(function.clone()))
.collect::<Vec<_>>();
let requested_tool_choice = normalize_tool_choice(tool_choice, converted_tools.is_empty())?;
let (converted_tools, tool_names) = ResponsesToolNames::convert(&tools);
let requested_tool_choice =
normalize_tool_choice(tool_choice, &tools, &tool_names, converted_tools.is_empty())?;
let tool_context = ResolvedToolContext::new(
&messages,
converted_tools,
Expand Down Expand Up @@ -302,6 +303,7 @@ pub(crate) fn prepare_responses_request(
request_id,
meta,
include_reasoning,
tool_names,
chat_request,
})
}
Expand Down Expand Up @@ -329,10 +331,12 @@ fn response_prompt_truncation(
/// Resolve the requested tool choice, mirroring the Python
/// `check_tool_usage` validator: without tools, `auto` is normalized to
/// `none` and named function choices are errors; with tools, named choices
/// must exist (enforced by
/// [`ResolvedToolContext`]).
/// must exist and resolve to the name the model sees (see
/// [`ResponsesToolNames::resolve_choice`]).
fn normalize_tool_choice(
tool_choice: Option<ResponseToolChoice>,
tools: &[ResponsesTool],
tool_names: &ResponsesToolNames,
tools_empty: bool,
) -> Result<Option<ChatToolChoice>, ApiError> {
let Some(tool_choice) = tool_choice else {
Expand All @@ -350,7 +354,9 @@ fn normalize_tool_choice(
"Tool choice 'function' not found in 'tools' parameter."
);
}
ChatToolChoice::Function { name }
ChatToolChoice::Function {
name: tool_names.resolve_choice(name, tools)?,
}
}
ResponseToolChoice::Object(ResponseToolChoiceObject::AllowedTools { tools, .. }) => {
bail_invalid_request!(
Expand Down Expand Up @@ -430,9 +436,18 @@ fn convert_input(
}
},
ResponseInputItem::FunctionCall(call) => {
let name = match &call.namespace {
Some(namespace) => ToolIdentity {
namespace: Some(namespace),
name: &call.name,
}
.flat_name()
.into_owned(),
None => call.name,
};
assistant.push_block(AssistantContentBlock::ToolCall(AssistantToolCall {
id: call.call_id,
name: call.name,
name,
arguments: call.arguments,
}));
}
Expand Down Expand Up @@ -776,32 +791,91 @@ fn should_continue_final_message(items: &[ResponseInputItem]) -> bool {
#[cfg(test)]
mod tests {
use serde_json::{Value, json};
use vllm_chat::ChatTokenUsage;
use vllm_chat::{ChatTokenUsage, ChatTool};
use vllm_llm::TokenUsage;

use super::*;
use crate::model_names::ServedModelNames;

#[test]
fn auto_tool_choice_without_tools_normalizes_to_none() {
let tool_choice = serde_json::from_value(json!("auto")).unwrap();

assert_eq!(
normalize_tool_choice(Some(tool_choice), true).unwrap(),
normalize_tool_choice(Some(tool_choice), &[], &ResponsesToolNames::default(), true)
.unwrap(),
Some(ChatToolChoice::None)
);
}

#[test]
fn responses_function_tool_defaults_parameters_to_null() {
let ResponsesTool::Function(function) = serde_json::from_value(json!({
let Ok(ResponsesTool::Function(function)) = serde_json::from_value(json!({
"type": "function",
"name": "lookup",
}))
.expect("deserialize function tool");
})) else {
panic!("deserialize function tool");
};

assert_eq!(ChatTool::from(function).parameters, Value::Null);
}

#[test]
fn namespace_tools_reach_the_model_by_their_flat_names() {
let request: ResponsesRequest = serde_json::from_value(json!({
"input": [
{"role": "user", "content": "Who is Ada?"},
{
"type": "function_call",
"call_id": "call_1",
"name": "get_contact",
"namespace": "crm",
"arguments": "{\"name\":\"Ada\"}"
},
{"type": "function_call_output", "call_id": "call_1", "output": "found"}
],
"tools": [{
"type": "namespace",
"name": "crm",
"description": "CRM tools.",
"tools": [{"type": "function", "name": "get_contact"}]
}],
"tool_choice": {"type": "function", "name": "get_contact"}
}))
.expect("deserialize request");
let served = LoraModelResolution {
model_names: ServedModelNames::new(vec!["test-model".to_string()]),
lora_request: None,
};

let prepared = prepare_responses_request(request, &served, Default::default())
.expect("prepare request");

let tool_context = &prepared.chat_request.tool_context;
let names = tool_context.initial_tools.iter().map(|tool| tool.name.as_str());
assert_eq!(names.collect::<Vec<_>>(), ["crm__get_contact"]);
let flat_choice = ChatToolChoice::Function {
name: "crm__get_contact".to_string(),
};
assert_eq!(tool_context.tool_choice, flat_choice);
assert_eq!(
prepared.meta.tool_choice,
ResponseToolChoice::Object(ResponseToolChoiceObject::Function(FunctionChoice {
name: "crm__get_contact".to_string(),
}))
);
let ChatMessage::Assistant { content } = &prepared.chat_request.messages[1] else {
panic!("assistant turn: {:?}", prepared.chat_request.messages);
};
assert!(
matches!(
content.as_slice(),
[AssistantContentBlock::ToolCall(call)] if call.name == "crm__get_contact"
),
"{content:?}"
);
}

#[test]
fn usage_preserves_reasoning_and_prompt_cache_token_counts() {
let usage = build_usage(&ChatTokenUsage {
Expand Down
29 changes: 19 additions & 10 deletions rust/src/server/src/routes/openai/responses/streaming.rs
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ use vllm_chat::{
AssistantBlockKind, AssistantContentBlock, AssistantToolCall, ChatEvent, FinishReason,
};

use super::tools::{ResponsesToolNames, ToolIdentity};
use super::types::{
AssistantRole, ResponseItemStatus, ResponseOutputContentPart, ResponseOutputItem,
ResponsesResponse, TextPart,
Expand Down Expand Up @@ -175,6 +176,8 @@ pub(crate) struct OutputItemStreamer {
current: Option<OpenItem>,
/// Whether reasoning items appear in the stream and final output.
include_reasoning: bool,
/// Reports calls of flat namespaced tool names with their namespace.
tool_names: ResponsesToolNames,
/// Items completed so far, in output order.
completed_items: Vec<ResponseOutputItem>,
/// Whether the last completed item still owes its `output_item.done`
Expand All @@ -196,16 +199,18 @@ enum OpenItem {
FunctionCall {
item_id: String,
call_id: String,
name: String,
/// The name the model called, flat for a namespaced function.
called_name: String,
arguments: String,
},
}

impl OutputItemStreamer {
pub(crate) fn new(include_reasoning: bool) -> Self {
pub(crate) fn new(include_reasoning: bool, tool_names: ResponsesToolNames) -> Self {
Self {
current: None,
include_reasoning,
tool_names,
completed_items: Vec::new(),
last_item_done_pending: false,
}
Expand Down Expand Up @@ -373,13 +378,15 @@ impl OutputItemStreamer {
self.current = Some(OpenItem::FunctionCall {
item_id: item_id.clone(),
call_id: call_id.clone(),
name: name.to_string(),
called_name: name.to_string(),
arguments: String::new(),
});
let ToolIdentity { namespace, name } = self.tool_names.decode(name);
let item = ResponseOutputItem::FunctionCall {
id: item_id,
call_id,
name: name.to_string(),
namespace: namespace.map(str::to_string),
arguments: String::new(),
status: Some(ResponseItemStatus::InProgress),
};
Expand Down Expand Up @@ -501,23 +508,25 @@ impl OutputItemStreamer {
OpenItem::FunctionCall {
item_id,
call_id,
name,
called_name,
arguments,
} => {
let (call_id, name, arguments) = match final_call {
let (call_id, called_name, arguments) = match final_call {
Some(call) => (call.id.clone(), call.name.clone(), call.arguments.clone()),
None => (call_id, name, arguments),
None => (call_id, called_name, arguments),
};
let ToolIdentity { namespace, name } = self.tool_names.decode(&called_name);
let events = vec![ResponseStreamEvent::FunctionCallArgumentsDone {
item_id: item_id.clone(),
output_index,
name: name.clone(),
name: name.to_string(),
arguments: arguments.clone(),
}];
let item = ResponseOutputItem::FunctionCall {
id: item_id,
call_id,
name,
name: name.to_string(),
namespace: namespace.map(str::to_string),
arguments,
status: Some(ResponseItemStatus::Completed),
};
Expand Down Expand Up @@ -578,7 +587,7 @@ mod tests {

#[test]
fn empty_function_call_emits_arguments_done() {
let mut streamer = OutputItemStreamer::new(true);
let mut streamer = OutputItemStreamer::new(true, ResponsesToolNames::default());
streamer.on_event(&ChatEvent::ToolCallStart {
index: 0,
id: "call_1".to_string(),
Expand Down Expand Up @@ -606,7 +615,7 @@ mod tests {

#[test]
fn empty_reasoning_block_is_not_an_output_item() {
let mut streamer = OutputItemStreamer::new(true);
let mut streamer = OutputItemStreamer::new(true, ResponsesToolNames::default());
for event in [
ChatEvent::BlockStart {
index: 0,
Expand Down
Loading