chenchenghao/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

[router][grpc] Refactor: Add builders for chat and responses (#12852)

Browse Source
This commit is contained in:
Chang Su
2025-11-07 15:43:32 -08:00
committed by GitHub
parent 55e8e3999c
commit fe19a580fb
19 changed files with 1315 additions and 580 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
sgl-router/src/protocols/builders/chat/mod.rs Normal file
View File

@@ -0,0 +1,7 @@
//! Builders for Chat Completion API response types
pub mod response;
pub mod stream_response;
pub use response::ChatCompletionResponseBuilder;
pub use stream_response::ChatCompletionStreamResponseBuilder;

218
sgl-router/src/protocols/builders/chat/response.rs Normal file
View File

@@ -0,0 +1,218 @@
//! Builder for ChatCompletionResponse
//!
//! Provides an ergonomic fluent API for constructing chat completion responses.
use crate::protocols::{chat::*, common::Usage};
/// Builder for ChatCompletionResponse
///
/// Provides a fluent interface for constructing chat completion responses with sensible defaults.
#[must_use = "Builder does nothing until .build() is called"]
#[derive(Clone, Debug)]
pub struct ChatCompletionResponseBuilder {
id: String,
object: String,
created: u64,
model: String,
choices: Vec<ChatChoice>,
usage: Option<Usage>,
system_fingerprint: Option<String>,
}
impl ChatCompletionResponseBuilder {
/// Create a new builder with required fields
///
/// # Arguments
/// - `id`: Completion ID (e.g., "chatcmpl_abc123")
/// - `model`: Model name used for generation
pub fn new(id: impl Into<String>, model: impl Into<String>) -> Self {
Self {
id: id.into(),
object: "chat.completion".to_string(),
created: chrono::Utc::now().timestamp() as u64,
model: model.into(),
choices: Vec::new(),
usage: None,
system_fingerprint: None,
}
}
/// Copy common fields from a ChatCompletionRequest
///
/// This populates the model field from the request.
pub fn copy_from_request(mut self, request: &ChatCompletionRequest) -> Self {
self.model = request.model.clone();
self
}
/// Set the object type (default: "chat.completion")
pub fn object(mut self, object: impl Into<String>) -> Self {
self.object = object.into();
self
}
/// Set the creation timestamp (default: current time)
pub fn created(mut self, timestamp: u64) -> Self {
self.created = timestamp;
self
}
/// Set the choices
pub fn choices(mut self, choices: Vec<ChatChoice>) -> Self {
self.choices = choices;
self
}
/// Add a single choice
pub fn add_choice(mut self, choice: ChatChoice) -> Self {
self.choices.push(choice);
self
}
/// Set usage information
pub fn usage(mut self, usage: Usage) -> Self {
self.usage = Some(usage);
self
}
/// Set usage if provided (handles Option)
pub fn maybe_usage(mut self, usage: Option<Usage>) -> Self {
if let Some(u) = usage {
self.usage = Some(u);
}
self
}
/// Set system fingerprint if provided (handles Option)
pub fn maybe_system_fingerprint(mut self, fingerprint: Option<impl Into<String>>) -> Self {
if let Some(fp) = fingerprint {
self.system_fingerprint = Some(fp.into());
}
self
}
/// Build the ChatCompletionResponse
pub fn build(self) -> ChatCompletionResponse {
ChatCompletionResponse {
id: self.id,
object: self.object,
created: self.created,
model: self.model,
choices: self.choices,
usage: self.usage,
system_fingerprint: self.system_fingerprint,
}
}
}
// ============================================================================
// Tests
// ============================================================================
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_build_minimal() {
let response = ChatCompletionResponse::builder("chatcmpl_123", "gpt-4").build();
assert_eq!(response.id, "chatcmpl_123");
assert_eq!(response.model, "gpt-4");
assert_eq!(response.object, "chat.completion");
assert!(response.choices.is_empty());
assert!(response.usage.is_none());
assert!(response.system_fingerprint.is_none());
}
#[test]
fn test_build_complete() {
let choice = ChatChoice {
index: 0,
message: ChatCompletionMessage {
role: "assistant".to_string(),
content: Some("Hello!".to_string()),
tool_calls: None,
reasoning_content: None,
},
logprobs: None,
finish_reason: Some("stop".to_string()),
matched_stop: None,
hidden_states: None,
};
let usage = Usage {
prompt_tokens: 10,
completion_tokens: 20,
total_tokens: 30,
completion_tokens_details: None,
};
let response = ChatCompletionResponse::builder("chatcmpl_456", "gpt-4")
.choices(vec![choice.clone()])
.maybe_usage(Some(usage))
.maybe_system_fingerprint(Some("fp_123abc"))
.build();
assert_eq!(response.id, "chatcmpl_456");
assert_eq!(response.choices.len(), 1);
assert_eq!(response.choices[0].index, 0);
assert!(response.usage.is_some());
assert_eq!(response.system_fingerprint.as_ref().unwrap(), "fp_123abc");
}
#[test]
fn test_add_multiple_choices() {
let choice1 = ChatChoice {
index: 0,
message: ChatCompletionMessage {
role: "assistant".to_string(),
content: Some("Option 1".to_string()),
tool_calls: None,
reasoning_content: None,
},
logprobs: None,
finish_reason: Some("stop".to_string()),
matched_stop: None,
hidden_states: None,
};
let choice2 = ChatChoice {
index: 1,
message: ChatCompletionMessage {
role: "assistant".to_string(),
content: Some("Option 2".to_string()),
tool_calls: None,
reasoning_content: None,
},
logprobs: None,
finish_reason: Some("stop".to_string()),
matched_stop: None,
hidden_states: None,
};
let response = ChatCompletionResponse::builder("chatcmpl_789", "gpt-4")
.add_choice(choice1)
.add_choice(choice2)
.build();
assert_eq!(response.choices.len(), 2);
assert_eq!(response.choices[0].index, 0);
assert_eq!(response.choices[1].index, 1);
}
#[test]
fn test_copy_from_request() {
let request = ChatCompletionRequest {
messages: vec![],
model: "gpt-3.5-turbo".to_string(),
..Default::default()
};
let response = ChatCompletionResponse::builder("chatcmpl_101", "gpt-4")
.copy_from_request(&request)
.build();
assert_eq!(response.model, "gpt-3.5-turbo"); // Copied from request
}
}

421
sgl-router/src/protocols/builders/chat/stream_response.rs Normal file
View File

@@ -0,0 +1,421 @@
//! Builder for ChatCompletionStreamResponse
//!
//! Provides an ergonomic fluent API for constructing streaming chat completion responses.
use std::borrow::Cow;
use crate::protocols::{
chat::*,
common::{FunctionCallDelta, ToolCallDelta, Usage},
};
/// Builder for ChatCompletionStreamResponse
///
/// Provides a fluent interface for constructing streaming chat completion chunks with sensible defaults.
#[must_use = "Builder does nothing until .build() is called"]
#[derive(Clone, Debug)]
pub struct ChatCompletionStreamResponseBuilder {
id: String,
object: String,
created: u64,
model: String,
choices: Vec<ChatStreamChoice>,
usage: Option<Usage>,
system_fingerprint: Option<String>,
}
impl ChatCompletionStreamResponseBuilder {
/// Create a new builder with required fields
///
/// # Arguments
/// - `id`: Completion ID (e.g., "chatcmpl_abc123")
/// - `model`: Model name used for generation
pub fn new(id: impl Into<String>, model: impl Into<String>) -> Self {
Self {
id: id.into(),
object: "chat.completion.chunk".to_string(),
created: chrono::Utc::now().timestamp() as u64,
model: model.into(),
choices: Vec::new(),
usage: None,
system_fingerprint: None,
}
}
/// Copy common fields from a ChatCompletionRequest
///
/// This populates the model field from the request.
pub fn copy_from_request(mut self, request: &ChatCompletionRequest) -> Self {
self.model = request.model.clone();
self
}
/// Set the object type (default: "chat.completion.chunk")
pub fn object(mut self, object: impl Into<String>) -> Self {
self.object = object.into();
self
}
/// Set the creation timestamp (default: current time)
pub fn created(mut self, timestamp: u64) -> Self {
self.created = timestamp;
self
}
/// Set the choices
pub fn choices(mut self, choices: Vec<ChatStreamChoice>) -> Self {
self.choices = choices;
self
}
/// Add a single choice (delta)
pub fn add_choice(mut self, choice: ChatStreamChoice) -> Self {
self.choices.push(choice);
self
}
/// Set usage information (typically sent in final chunk)
pub fn usage(mut self, usage: Usage) -> Self {
self.usage = Some(usage);
self
}
/// Set system fingerprint if provided (handles Option)
pub fn maybe_system_fingerprint(mut self, fingerprint: Option<impl Into<String>>) -> Self {
if let Some(fp) = fingerprint {
self.system_fingerprint = Some(fp.into());
}
self
}
/// Set usage if provided (handles Option)
pub fn maybe_usage(mut self, usage: Option<Usage>) -> Self {
if let Some(u) = usage {
self.usage = Some(u);
}
self
}
/// Add a choice delta that sets `role` and `content`
pub fn add_choice_content(
mut self,
index: u32,
role: impl Into<String>,
content: impl Into<String>,
) -> Self {
self.choices.push(ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some(role.into()),
content: Some(content.into()),
tool_calls: None,
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
});
self
}
/// Add a choice delta that sets `role`, `content`, and `logprobs`
pub fn add_choice_content_with_logprobs(
mut self,
index: u32,
role: impl Into<String>,
content: impl Into<String>,
logprobs: Option<crate::protocols::common::ChatLogProbs>,
) -> Self {
self.choices.push(ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some(role.into()),
content: Some(content.into()),
tool_calls: None,
reasoning_content: None,
},
logprobs,
finish_reason: None,
matched_stop: None,
});
self
}
/// Add a choice delta that only sets `role`
pub fn add_choice_role(mut self, index: u32, role: impl Into<String>) -> Self {
self.choices.push(ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some(role.into()),
content: None,
tool_calls: None,
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
});
self
}
/// Add a choice delta that appends a tool-call *arguments delta*
/// Uses `Cow` so you can pass `&str` or `String` without extra clones
pub fn add_choice_tool_args(
mut self,
index: u32,
args_delta: impl Into<Cow<'static, str>>,
) -> Self {
self.choices.push(ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: None,
tool_calls: Some(vec![ToolCallDelta {
index: 0,
id: None,
tool_type: None,
function: Some(FunctionCallDelta {
name: None,
arguments: Some(args_delta.into().into_owned()),
}),
}]),
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
});
self
}
/// Add a choice delta that sets reasoning content (for models that stream reasoning)
pub fn add_choice_reasoning(mut self, index: u32, reasoning: impl Into<String>) -> Self {
self.choices.push(ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: None,
tool_calls: None,
reasoning_content: Some(reasoning.into()),
},
logprobs: None,
finish_reason: None,
matched_stop: None,
});
self
}
/// Add a choice delta for tool call with function name and ID
pub fn add_choice_tool_name(
mut self,
index: u32,
tool_call_id: impl Into<String>,
function_name: impl Into<String>,
) -> Self {
self.choices.push(ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: None,
tool_calls: Some(vec![ToolCallDelta {
index: 0,
id: Some(tool_call_id.into()),
tool_type: Some("function".to_string()),
function: Some(FunctionCallDelta {
name: Some(function_name.into()),
arguments: None,
}),
}]),
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
});
self
}
/// Add a choice delta with a pre-constructed ToolCallDelta
/// Useful when you already have a ToolCallDelta object to emit
pub fn add_choice_tool_call_delta(
mut self,
index: u32,
tool_call_delta: ToolCallDelta,
) -> Self {
self.choices.push(ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: None,
tool_calls: Some(vec![tool_call_delta]),
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
});
self
}
/// Add a choice with finish_reason (final chunk)
/// This is used for the last chunk in a stream to signal completion
pub fn add_choice_finish_reason(
mut self,
index: u32,
finish_reason: impl Into<String>,
matched_stop: Option<serde_json::Value>,
) -> Self {
self.choices.push(ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: None,
content: None,
tool_calls: None,
reasoning_content: None,
},
logprobs: None,
finish_reason: Some(finish_reason.into()),
matched_stop,
});
self
}
/// Build the ChatCompletionStreamResponse
pub fn build(self) -> ChatCompletionStreamResponse {
ChatCompletionStreamResponse {
id: self.id,
object: self.object,
created: self.created,
model: self.model,
system_fingerprint: self.system_fingerprint,
choices: self.choices,
usage: self.usage,
}
}
}
// ============================================================================
// Tests
// ============================================================================
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_build_minimal() {
let chunk = ChatCompletionStreamResponseBuilder::new("chatcmpl_123", "gpt-4").build();
assert_eq!(chunk.id, "chatcmpl_123");
assert_eq!(chunk.model, "gpt-4");
assert_eq!(chunk.object, "chat.completion.chunk");
assert!(chunk.choices.is_empty());
assert!(chunk.usage.is_none());
}
#[test]
fn test_with_content_delta() {
let chunk = ChatCompletionStreamResponseBuilder::new("chatcmpl_456", "gpt-4")
.add_choice_content(0, "assistant", "Hello")
.build();
assert_eq!(chunk.choices.len(), 1);
assert_eq!(chunk.choices[0].index, 0);
assert_eq!(chunk.choices[0].delta.content.as_ref().unwrap(), "Hello");
assert_eq!(chunk.choices[0].delta.role.as_ref().unwrap(), "assistant");
assert!(chunk.choices[0].finish_reason.is_none());
}
#[test]
fn test_with_role_delta() {
let chunk = ChatCompletionStreamResponseBuilder::new("chatcmpl_789", "gpt-4")
.add_choice_role(0, "assistant")
.build();
assert_eq!(chunk.choices.len(), 1);
assert_eq!(chunk.choices[0].delta.role.as_ref().unwrap(), "assistant");
assert!(chunk.choices[0].delta.content.is_none());
}
#[test]
fn test_with_finish_reason() {
let chunk = ChatCompletionStreamResponseBuilder::new("chatcmpl_101", "gpt-4")
.add_choice_finish_reason(0, "stop", None)
.build();
assert_eq!(chunk.choices.len(), 1);
assert_eq!(chunk.choices[0].finish_reason.as_ref().unwrap(), "stop");
assert!(chunk.choices[0].delta.content.is_none());
assert!(chunk.choices[0].delta.role.is_none());
}
#[test]
fn test_multiple_deltas() {
let chunk = ChatCompletionStreamResponseBuilder::new("chatcmpl_202", "gpt-4")
.add_choice_role(0, "assistant")
.add_choice_content(0, "assistant", "Hello")
.add_choice_content(0, "assistant", " world")
.add_choice_finish_reason(0, "stop", None)
.build();
assert_eq!(chunk.choices.len(), 4); // role + 2 content + finish
}
#[test]
fn test_with_usage() {
let usage = Usage {
prompt_tokens: 10,
completion_tokens: 20,
total_tokens: 30,
completion_tokens_details: None,
};
let chunk = ChatCompletionStreamResponseBuilder::new("chatcmpl_303", "gpt-4")
.add_choice_finish_reason(0, "stop", None)
.usage(usage)
.build();
assert!(chunk.usage.is_some());
assert_eq!(chunk.usage.as_ref().unwrap().total_tokens, 30);
}
#[test]
fn test_copy_from_request() {
let request = ChatCompletionRequest {
messages: vec![],
model: "gpt-3.5-turbo".to_string(),
..Default::default()
};
let chunk = ChatCompletionStreamResponseBuilder::new("chatcmpl_404", "gpt-4")
.copy_from_request(&request)
.add_choice_content(0, "assistant", "test")
.build();
assert_eq!(chunk.model, "gpt-3.5-turbo"); // Copied from request
}
#[test]
fn test_add_choice_explicit() {
let choice = ChatStreamChoice {
index: 0,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: Some("Hello".to_string()),
tool_calls: None,
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
};
let chunk = ChatCompletionStreamResponseBuilder::new("chatcmpl_505", "gpt-4")
.add_choice(choice)
.build();
assert_eq!(chunk.choices.len(), 1);
assert_eq!(chunk.choices[0].delta.role.as_ref().unwrap(), "assistant");
assert_eq!(chunk.choices[0].delta.content.as_ref().unwrap(), "Hello");
}
}

27
sgl-router/src/protocols/builders/mod.rs Normal file
View File

@@ -0,0 +1,27 @@
//! Builder patterns for protocol response types
//!
//! This module provides ergonomic builders for response types with many optional fields.
//! Builders help avoid telescoping constructors and make construction intent clear.
//!
//! # Organization
//!
//! Builders are organized by API:
//! - `chat/` - Chat Completion API builders (response, stream_response)
//! - `responses/` - Responses API builder (response)
//!
//! # Optional Fields
//!
//! For optional fields, builders provide `maybe_*` methods that handle `Option<T>` directly:
//! ```ignore
//! builder
//! .field(value)
//! .maybe_optional_field(optional_value) // Accepts Option<T>
//! .build()
//! ```
pub mod chat;
pub mod responses;
// Re-export all builders for convenient access
pub use chat::{ChatCompletionResponseBuilder, ChatCompletionStreamResponseBuilder};
pub use responses::ResponsesResponseBuilder;

5
sgl-router/src/protocols/builders/responses/mod.rs Normal file
View File

@@ -0,0 +1,5 @@
//! Builders for Responses API response types
pub mod response;
pub use response::ResponsesResponseBuilder;

411
sgl-router/src/protocols/builders/responses/response.rs Normal file
View File

@@ -0,0 +1,411 @@
//! Builder for ResponsesResponse
//!
//! Provides an ergonomic fluent API for constructing ResponsesResponse instances.
use std::collections::HashMap;
use serde_json::Value;
use crate::protocols::responses::*;
/// Builder for ResponsesResponse
///
/// Provides a fluent interface for constructing responses with sensible defaults.
#[must_use = "Builder does nothing until .build() is called"]
#[derive(Clone, Debug)]
pub struct ResponsesResponseBuilder {
id: String,
object: String,
created_at: i64,
status: ResponseStatus,
error: Option<Value>,
incomplete_details: Option<Value>,
instructions: Option<String>,
max_output_tokens: Option<u32>,
model: String,
output: Vec<ResponseOutputItem>,
parallel_tool_calls: bool,
previous_response_id: Option<String>,
reasoning: Option<ReasoningInfo>,
store: bool,
temperature: Option<f32>,
text: Option<TextConfig>,
tool_choice: String,
tools: Vec<ResponseTool>,
top_p: Option<f32>,
truncation: Option<String>,
usage: Option<ResponsesUsage>,
user: Option<String>,
safety_identifier: Option<String>,
metadata: HashMap<String, Value>,
}
impl ResponsesResponseBuilder {
/// Create a new builder with required fields
///
/// # Arguments
/// - `id`: Response ID (e.g., "resp_abc123")
/// - `model`: Model name used for generation
pub fn new(id: impl Into<String>, model: impl Into<String>) -> Self {
Self {
id: id.into(),
object: "response".to_string(),
created_at: chrono::Utc::now().timestamp(),
status: ResponseStatus::InProgress,
error: None,
incomplete_details: None,
instructions: None,
max_output_tokens: None,
model: model.into(),
output: Vec::new(),
parallel_tool_calls: true,
previous_response_id: None,
reasoning: None,
store: true,
temperature: None,
text: None,
tool_choice: "auto".to_string(),
tools: Vec::new(),
top_p: None,
truncation: None,
usage: None,
user: None,
safety_identifier: None,
metadata: HashMap::new(),
}
}
/// Copy common fields from a ResponsesRequest
///
/// This populates fields like instructions, max_output_tokens, temperature, etc.
/// from the original request, making it easy to construct a response that mirrors
/// the request parameters.
///
/// Note: `safety_identifier` is intentionally NOT copied as it is for content moderation
/// and should be set independently from the request's `user` field (which is for billing/tracking).
pub fn copy_from_request(mut self, request: &ResponsesRequest) -> Self {
self.instructions = request.instructions.clone();
self.max_output_tokens = request.max_output_tokens;
self.parallel_tool_calls = request.parallel_tool_calls.unwrap_or(true);
self.previous_response_id = request.previous_response_id.clone();
self.store = request.store.unwrap_or(true);
self.temperature = request.temperature;
self.tool_choice = if let Some(ref tc) = request.tool_choice {
serde_json::to_string(tc).unwrap_or_else(|_| "auto".to_string())
} else {
"auto".to_string()
};
self.tools = request.tools.clone().unwrap_or_default();
self.top_p = request.top_p;
self.user = request.user.clone();
self.metadata = request.metadata.clone().unwrap_or_default();
self
}
/// Set the object type (default: "response")
pub fn object(mut self, object: impl Into<String>) -> Self {
self.object = object.into();
self
}
/// Set the creation timestamp (default: current time)
pub fn created_at(mut self, timestamp: i64) -> Self {
self.created_at = timestamp;
self
}
/// Set the response status
pub fn status(mut self, status: ResponseStatus) -> Self {
self.status = status;
self
}
/// Set error information (if status is failed)
pub fn error(mut self, error: Value) -> Self {
self.error = Some(error);
self
}
/// Set incomplete details (if response was truncated)
pub fn incomplete_details(mut self, details: Value) -> Self {
self.incomplete_details = Some(details);
self
}
/// Set system instructions
pub fn instructions(mut self, instructions: impl Into<String>) -> Self {
self.instructions = Some(instructions.into());
self
}
/// Set max output tokens
pub fn max_output_tokens(mut self, tokens: u32) -> Self {
self.max_output_tokens = Some(tokens);
self
}
/// Set output items
pub fn output(mut self, output: Vec<ResponseOutputItem>) -> Self {
self.output = output;
self
}
/// Add a single output item
pub fn add_output(mut self, item: ResponseOutputItem) -> Self {
self.output.push(item);
self
}
/// Set whether parallel tool calls are enabled
pub fn parallel_tool_calls(mut self, enabled: bool) -> Self {
self.parallel_tool_calls = enabled;
self
}
/// Set previous response ID (if continuation)
pub fn previous_response_id(mut self, id: impl Into<String>) -> Self {
self.previous_response_id = Some(id.into());
self
}
/// Set reasoning information
pub fn reasoning(mut self, reasoning: ReasoningInfo) -> Self {
self.reasoning = Some(reasoning);
self
}
/// Set whether the response is stored
pub fn store(mut self, store: bool) -> Self {
self.store = store;
self
}
/// Set temperature setting
pub fn temperature(mut self, temperature: f32) -> Self {
self.temperature = Some(temperature);
self
}
/// Set text format settings if provided (handles Option)
pub fn maybe_text(mut self, text: Option<TextConfig>) -> Self {
if let Some(t) = text {
self.text = Some(t);
}
self
}
/// Set tool choice setting
pub fn tool_choice(mut self, tool_choice: impl Into<String>) -> Self {
self.tool_choice = tool_choice.into();
self
}
/// Set available tools
pub fn tools(mut self, tools: Vec<ResponseTool>) -> Self {
self.tools = tools;
self
}
/// Set top-p setting
pub fn top_p(mut self, top_p: f32) -> Self {
self.top_p = Some(top_p);
self
}
/// Set truncation strategy
pub fn truncation(mut self, truncation: impl Into<String>) -> Self {
self.truncation = Some(truncation.into());
self
}
/// Set usage information
pub fn usage(mut self, usage: ResponsesUsage) -> Self {
self.usage = Some(usage);
self
}
/// Set usage if provided (handles Option)
pub fn maybe_usage(mut self, usage: Option<ResponsesUsage>) -> Self {
if let Some(u) = usage {
self.usage = Some(u);
}
self
}
/// Copy from request if provided (handles Option)
pub fn maybe_copy_from_request(mut self, request: Option<&ResponsesRequest>) -> Self {
if let Some(req) = request {
self = self.copy_from_request(req);
}
self
}
/// Set user identifier
pub fn user(mut self, user: impl Into<String>) -> Self {
self.user = Some(user.into());
self
}
/// Set safety identifier
pub fn safety_identifier(mut self, identifier: impl Into<String>) -> Self {
self.safety_identifier = Some(identifier.into());
self
}
/// Set metadata
pub fn metadata(mut self, metadata: HashMap<String, Value>) -> Self {
self.metadata = metadata;
self
}
/// Add a single metadata entry
pub fn add_metadata(mut self, key: impl Into<String>, value: Value) -> Self {
self.metadata.insert(key.into(), value);
self
}
/// Build the ResponsesResponse
pub fn build(self) -> ResponsesResponse {
ResponsesResponse {
id: self.id,
object: self.object,
created_at: self.created_at,
status: self.status,
error: self.error,
incomplete_details: self.incomplete_details,
instructions: self.instructions,
max_output_tokens: self.max_output_tokens,
model: self.model,
output: self.output,
parallel_tool_calls: self.parallel_tool_calls,
previous_response_id: self.previous_response_id,
reasoning: self.reasoning,
store: self.store,
temperature: self.temperature,
text: self.text,
tool_choice: self.tool_choice,
tools: self.tools,
top_p: self.top_p,
truncation: self.truncation,
usage: self.usage,
user: self.user,
safety_identifier: self.safety_identifier,
metadata: self.metadata,
}
}
}
// ============================================================================
// Tests
// ============================================================================
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_build_minimal() {
let response = ResponsesResponse::builder("resp_123", "gpt-4").build();
assert_eq!(response.id, "resp_123");
assert_eq!(response.model, "gpt-4");
assert_eq!(response.object, "response");
assert_eq!(response.status, ResponseStatus::InProgress);
assert!(response.output.is_empty());
assert!(response.parallel_tool_calls);
assert!(response.store);
}
#[test]
fn test_build_complete() {
let response = ResponsesResponse::builder("resp_123", "gpt-4")
.status(ResponseStatus::Completed)
.instructions("You are a helpful assistant")
.max_output_tokens(1000)
.temperature(0.7)
.top_p(0.9)
.parallel_tool_calls(false)
.store(false)
.build();
assert_eq!(response.status, ResponseStatus::Completed);
assert_eq!(
response.instructions.as_ref().unwrap(),
"You are a helpful assistant"
);
assert_eq!(response.max_output_tokens, Some(1000));
assert_eq!(response.temperature, Some(0.7));
assert_eq!(response.top_p, Some(0.9));
assert!(!response.parallel_tool_calls);
assert!(!response.store);
}
#[test]
fn test_copy_from_request() {
let request = ResponsesRequest {
model: "gpt-4".to_string(),
input: ResponseInput::Text("test".to_string()),
instructions: Some("Be helpful".to_string()),
max_output_tokens: Some(500),
temperature: Some(0.8),
top_p: Some(0.95),
parallel_tool_calls: Some(false),
store: Some(false),
user: Some("user_123".to_string()),
metadata: Some(HashMap::from([(
"key".to_string(),
serde_json::json!("value"),
)])),
..Default::default()
};
let response = ResponsesResponse::builder("resp_456", "gpt-4")
.copy_from_request(&request)
.status(ResponseStatus::Completed)
.build();
assert_eq!(response.instructions.as_ref().unwrap(), "Be helpful");
assert_eq!(response.max_output_tokens, Some(500));
assert_eq!(response.temperature, Some(0.8));
assert_eq!(response.top_p, Some(0.95));
assert!(!response.parallel_tool_calls);
assert!(!response.store);
assert_eq!(response.user.as_ref().unwrap(), "user_123");
assert_eq!(
response.metadata.get("key").unwrap(),
&serde_json::json!("value")
);
}
#[test]
fn test_add_output_items() {
let response = ResponsesResponse::builder("resp_789", "gpt-4")
.add_output(ResponseOutputItem::Message {
id: "msg_1".to_string(),
role: "assistant".to_string(),
content: vec![],
status: "completed".to_string(),
})
.add_output(ResponseOutputItem::Message {
id: "msg_2".to_string(),
role: "assistant".to_string(),
content: vec![],
status: "completed".to_string(),
})
.build();
assert_eq!(response.output.len(), 2);
}
#[test]
fn test_add_metadata() {
let response = ResponsesResponse::builder("resp_101", "gpt-4")
.add_metadata("key1", serde_json::json!("value1"))
.add_metadata("key2", serde_json::json!(42))
.build();
assert_eq!(response.metadata.len(), 2);
assert_eq!(response.metadata.get("key1").unwrap(), "value1");
assert_eq!(response.metadata.get("key2").unwrap(), 42);
}
}

25
sgl-router/src/protocols/chat.rs
View File

@@ -8,7 +8,10 @@ use super::{
common::*,
sampling_params::{validate_top_k_value, validate_top_p_value},
};
use crate::protocols::validated::Normalizable;
use crate::protocols::{
builders::{ChatCompletionResponseBuilder, ChatCompletionStreamResponseBuilder},
validated::Normalizable,
};
// ============================================================================
// Chat Messages
@@ -639,6 +642,16 @@ pub struct ChatCompletionResponse {
pub system_fingerprint: Option<String>,
}
impl ChatCompletionResponse {
/// Create a new builder for ChatCompletionResponse
pub fn builder(
id: impl Into<String>,
model: impl Into<String>,
) -> ChatCompletionResponseBuilder {
ChatCompletionResponseBuilder::new(id, model)
}
}
/// Response message structure for ChatCompletionResponse (different from request ChatMessage)
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct ChatCompletionMessage {
@@ -680,6 +693,16 @@ pub struct ChatCompletionStreamResponse {
pub usage: Option<Usage>,
}
impl ChatCompletionStreamResponse {
/// Create a new builder for ChatCompletionStreamResponse
pub fn builder(
id: impl Into<String>,
model: impl Into<String>,
) -> ChatCompletionStreamResponseBuilder {
ChatCompletionStreamResponseBuilder::new(id, model)
}
}
/// Delta structure for streaming chat completion responses
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct ChatMessageDelta {

1
sgl-router/src/protocols/mod.rs
View File

@@ -1,6 +1,7 @@
// Protocol definitions and validation for various LLM APIs
// This module provides a structured approach to handling different API protocols
pub mod builders;
pub mod chat;
pub mod classify;
pub mod common;

8
sgl-router/src/protocols/responses.rs
View File

@@ -12,6 +12,7 @@ use super::common::{
default_model, default_true, ChatLogProbs, Function, GenerationRequest, PromptTokenUsageInfo,
StringOrArray, ToolChoice, UsageInfo,
};
use crate::protocols::builders::ResponsesResponseBuilder;
// ============================================================================
// Response Tools (MCP and others)
@@ -273,7 +274,7 @@ pub enum Truncation {
Disabled,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ResponseStatus {
Queued,
@@ -976,6 +977,11 @@ fn default_tool_choice() -> String {
}
impl ResponsesResponse {
/// Create a builder for constructing a ResponsesResponse
pub fn builder(id: impl Into<String>, model: impl Into<String>) -> ResponsesResponseBuilder {
ResponsesResponseBuilder::new(id, model)
}
/// Check if the response is complete
pub fn is_complete(&self) -> bool {
matches!(self.status, ResponseStatus::Completed)

38
sgl-router/src/routers/grpc/common/response_formatting.rs
View File

@@ -4,14 +4,7 @@
//! - Usage calculation from gRPC responses
//! - ChatCompletionResponse construction
use crate::{
grpc_client::proto,
protocols::{
chat::{ChatChoice, ChatCompletionResponse},
common::Usage,
},
routers::grpc::context::DispatchMetadata,
};
use crate::{grpc_client::proto, protocols::common::Usage};
/// Build usage information from collected gRPC responses
///
@@ -34,32 +27,3 @@ pub fn build_usage(responses: &[proto::GenerateComplete]) -> Usage {
completion_tokens_details: None,
}
}
/// Build final ChatCompletionResponse from processed choices
///
/// Constructs the OpenAI-compatible response object with all metadata.
///
/// # Arguments
/// * `choices` - Processed chat choices (after parsing, logprobs, etc.)
/// * `dispatch` - Dispatch metadata (request_id, created timestamp, etc.)
/// * `model` - Model name to include in response
/// * `usage` - Token usage information
///
/// # Returns
/// Complete ChatCompletionResponse ready to send to client
pub fn build_chat_response(
choices: Vec<ChatChoice>,
dispatch: &DispatchMetadata,
model: String,
usage: Usage,
) -> ChatCompletionResponse {
ChatCompletionResponse {
id: dispatch.request_id.clone(),
object: "chat.completion".to_string(),
created: dispatch.created,
model,
choices,
usage: Some(usage),
system_fingerprint: dispatch.weight_version.clone(),
}
}

47
sgl-router/src/routers/grpc/common/responses/streaming.rs
View File

@@ -617,45 +617,14 @@ impl ResponseStreamEventEmitter {
ResponsesUsage::Classic(usage_info)
});
// Get original request fields or use defaults
let req = self.original_request.as_ref();
// Convert tool_choice to String
let tool_choice = req
.and_then(|r| r.tool_choice.as_ref())
.map(|tc| serde_json::to_string(tc).unwrap_or_else(|_| "auto".to_string()))
.unwrap_or_else(|| "auto".to_string());
ResponsesResponse {
id: self.response_id.clone(),
object: "response".to_string(),
created_at: self.created_at as i64,
status: ResponseStatus::Completed,
error: None,
incomplete_details: None,
instructions: req.and_then(|r| r.instructions.clone()),
max_output_tokens: req.and_then(|r| r.max_output_tokens),
model: self.model.clone(),
output,
parallel_tool_calls: req.and_then(|r| r.parallel_tool_calls).unwrap_or(true),
previous_response_id: req.and_then(|r| r.previous_response_id.clone()),
reasoning: None, // TODO: Extract from output items if needed
store: req.and_then(|r| r.store).unwrap_or(true),
temperature: req.and_then(|r| r.temperature),
text: None,
tool_choice,
tools: req
.map(|r| r.tools.clone().unwrap_or_default())
.unwrap_or_default(),
top_p: req.and_then(|r| r.top_p),
truncation: None, // Convert from Truncation to String if needed
usage: responses_usage,
metadata: req
.map(|r| r.metadata.clone().unwrap_or_default())
.unwrap_or_default(),
user: req.and_then(|r| r.user.clone()),
safety_identifier: None,
}
// Build response using builder
ResponsesResponse::builder(&self.response_id, &self.model)
.created_at(self.created_at as i64)
.status(ResponseStatus::Completed)
.output(output)
.maybe_copy_from_request(self.original_request.as_ref())
.maybe_usage(responses_usage)
.build()
}
/// Emit reasoning item wrapper events (added + done)

56
sgl-router/src/routers/grpc/harmony/processor.rs
View File

@@ -99,14 +99,14 @@ impl HarmonyResponseProcessor {
let usage = response_formatting::build_usage(&all_responses);
// Final ChatCompletionResponse
let response = response_formatting::build_chat_response(
choices,
&dispatch,
chat_request.model.clone(),
usage,
);
Ok(response)
Ok(
ChatCompletionResponse::builder(&dispatch.request_id, &chat_request.model)
.created(dispatch.created)
.choices(choices)
.usage(usage)
.maybe_system_fingerprint(dispatch.weight_version.clone())
.build(),
)
}
}
@@ -253,42 +253,20 @@ impl HarmonyResponseProcessor {
}
// Build ResponsesResponse with all required fields
let response = ResponsesResponse {
id: dispatch.request_id.clone(),
object: "response".to_string(),
created_at: dispatch.created as i64,
status: ResponseStatus::Completed,
error: None,
incomplete_details: None,
instructions: responses_request.instructions.clone(),
max_output_tokens: responses_request.max_output_tokens,
model: responses_request.model.clone(),
output,
parallel_tool_calls: responses_request.parallel_tool_calls.unwrap_or(true),
previous_response_id: responses_request.previous_response_id.clone(),
reasoning: None, // Set by caller if needed
store: responses_request.store.unwrap_or(true),
temperature: responses_request.temperature,
text: responses_request.text.clone(),
tool_choice: responses_request
.tool_choice
.as_ref()
.map(|tc| serde_json::to_string(tc).unwrap_or_else(|_| "auto".to_string()))
.unwrap_or_else(|| "auto".to_string()),
tools: responses_request.tools.clone().unwrap_or_default(),
top_p: responses_request.top_p,
truncation: None,
usage: Some(ResponsesUsage::Modern(ResponseUsage {
let response = ResponsesResponse::builder(&dispatch.request_id, &responses_request.model)
.copy_from_request(&responses_request)
.created_at(dispatch.created as i64)
.status(ResponseStatus::Completed)
.output(output)
.maybe_text(responses_request.text.clone())
.usage(ResponsesUsage::Modern(ResponseUsage {
input_tokens: usage.prompt_tokens,
output_tokens: usage.completion_tokens,
total_tokens: usage.total_tokens,
input_tokens_details: None,
output_tokens_details: None,
})),
user: None,
safety_identifier: responses_request.user.clone(),
metadata: responses_request.metadata.clone().unwrap_or_default(),
};
}))
.build();
Ok(ResponsesIterationResult::Completed {
response: Box::new(response),

40
sgl-router/src/routers/grpc/harmony/responses.rs
View File

@@ -1040,7 +1040,6 @@ async fn execute_without_mcp_streaming(
/// Build ResponsesResponse with tool calls (MCP and/or function tools)
///
/// ResponsesResponse with tool calls
/// TODO: Refactor to use builder pattern
#[allow(clippy::too_many_arguments)]
fn build_tool_response(
mcp_tool_calls: Vec<ToolCall>,
@@ -1119,42 +1118,19 @@ fn build_tool_response(
.unwrap()
.as_secs() as i64;
ResponsesResponse {
id: request_id,
object: "response".to_string(),
created_at,
status: ResponseStatus::Completed,
error: None,
incomplete_details: None,
instructions: responses_request.instructions.clone(),
max_output_tokens: responses_request.max_output_tokens,
model: responses_request.model.clone(),
output,
parallel_tool_calls: responses_request.parallel_tool_calls.unwrap_or(true),
previous_response_id: responses_request.previous_response_id.clone(),
reasoning: None,
store: responses_request.store.unwrap_or(true),
temperature: responses_request.temperature,
text: None,
tool_choice: responses_request
.tool_choice
.as_ref()
.map(|tc| to_string(tc).unwrap_or_else(|_| "auto".to_string()))
.unwrap_or_else(|| "auto".to_string()),
tools: responses_request.tools.clone().unwrap_or_default(),
top_p: responses_request.top_p,
truncation: None,
usage: Some(ResponsesUsage::Modern(ResponseUsage {
ResponsesResponse::builder(&request_id, &responses_request.model)
.copy_from_request(&responses_request)
.created_at(created_at)
.status(ResponseStatus::Completed)
.output(output)
.usage(ResponsesUsage::Modern(ResponseUsage {
input_tokens: usage.prompt_tokens,
output_tokens: usage.completion_tokens,
total_tokens: usage.total_tokens,
input_tokens_details: None,
output_tokens_details: None,
})),
user: None,
safety_identifier: responses_request.user.clone(),
metadata: responses_request.metadata.clone().unwrap_or_default(),
}
}))
.build()
}
/// Execute MCP tools and collect results

150
sgl-router/src/routers/grpc/harmony/streaming.rs
View File

@@ -449,26 +449,14 @@ impl HarmonyStreamingProcessor {
) -> Result<(), String> {
// On first chunk, emit role announcement separately
if is_first {
let role_chunk = ChatCompletionStreamResponse {
id: dispatch.request_id.clone(),
object: "chat.completion.chunk".to_string(),
created: dispatch.created,
model: original_request.model.clone(),
system_fingerprint: dispatch.weight_version.clone(),
choices: vec![ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: Some(String::new()),
tool_calls: None,
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
}],
usage: None,
};
let role_chunk = ChatCompletionStreamResponse::builder(
&dispatch.request_id,
&original_request.model,
)
.created(dispatch.created)
.add_choice_role(index, "assistant")
.maybe_system_fingerprint(dispatch.weight_version.clone())
.build();
let chunk_json = serde_json::to_string(&role_chunk)
.map_err(|e| format!("JSON serialization error: {}", e))?;
@@ -497,21 +485,18 @@ impl HarmonyStreamingProcessor {
};
// Build and emit chunk
let chunk = ChatCompletionStreamResponse {
id: dispatch.request_id.clone(),
object: "chat.completion.chunk".to_string(),
created: dispatch.created,
model: original_request.model.clone(),
system_fingerprint: dispatch.weight_version.clone(),
choices: vec![ChatStreamChoice {
index,
delta: chat_delta,
logprobs: None,
finish_reason: None,
matched_stop: None,
}],
usage: None,
};
let chunk =
ChatCompletionStreamResponse::builder(&dispatch.request_id, &original_request.model)
.created(dispatch.created)
.add_choice(ChatStreamChoice {
index,
delta: chat_delta,
logprobs: None,
finish_reason: None,
matched_stop: None,
})
.maybe_system_fingerprint(dispatch.weight_version.clone())
.build();
let chunk_json = serde_json::to_string(&chunk)
.map_err(|e| format!("JSON serialization error: {}", e))?;
@@ -532,26 +517,12 @@ impl HarmonyStreamingProcessor {
original_request: &ChatCompletionRequest,
tx: &mpsc::UnboundedSender<Result<Bytes, io::Error>>,
) -> Result<(), String> {
let chunk = ChatCompletionStreamResponse {
id: dispatch.request_id.clone(),
object: "chat.completion.chunk".to_string(),
created: dispatch.created,
model: original_request.model.clone(),
system_fingerprint: dispatch.weight_version.clone(),
choices: vec![ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: None,
content: None,
tool_calls: None,
reasoning_content: None,
},
logprobs: None,
finish_reason: Some(finish_reason.to_string()),
matched_stop: matched_stop.cloned(),
}],
usage: None,
};
let chunk =
ChatCompletionStreamResponse::builder(&dispatch.request_id, &original_request.model)
.created(dispatch.created)
.add_choice_finish_reason(index, finish_reason, matched_stop.cloned())
.maybe_system_fingerprint(dispatch.weight_version.clone())
.build();
let chunk_json = serde_json::to_string(&chunk)
.map_err(|e| format!("JSON serialization error: {}", e))?;
@@ -571,20 +542,17 @@ impl HarmonyStreamingProcessor {
original_request: &ChatCompletionRequest,
tx: &mpsc::UnboundedSender<Result<Bytes, io::Error>>,
) -> Result<(), String> {
let usage_chunk = ChatCompletionStreamResponse {
id: dispatch.request_id.clone(),
object: "chat.completion.chunk".to_string(),
created: dispatch.created,
model: original_request.model.clone(),
system_fingerprint: dispatch.weight_version.clone(),
choices: vec![],
usage: Some(Usage {
prompt_tokens,
completion_tokens,
total_tokens: prompt_tokens + completion_tokens,
completion_tokens_details: None,
}),
};
let usage_chunk =
ChatCompletionStreamResponse::builder(&dispatch.request_id, &original_request.model)
.created(dispatch.created)
.usage(Usage {
prompt_tokens,
completion_tokens,
total_tokens: prompt_tokens + completion_tokens,
completion_tokens_details: None,
})
.maybe_system_fingerprint(dispatch.weight_version.clone())
.build();
let chunk_json = serde_json::to_string(&usage_chunk)
.map_err(|e| format!("JSON serialization error: {}", e))?;
@@ -1111,38 +1079,18 @@ impl HarmonyStreamingProcessor {
// The caller will build it from the SSE events
// Return a placeholder Completed result (caller ignores these fields in streaming mode)
Ok(ResponsesIterationResult::Completed {
response: Box::new(ResponsesResponse {
id: emitter.response_id.clone(),
object: "response".to_string(),
created_at: 0,
status: ResponseStatus::Completed,
error: None,
incomplete_details: None,
instructions: None,
max_output_tokens: None,
model: String::new(),
output: vec![],
parallel_tool_calls: true,
previous_response_id: None,
reasoning: None,
store: true,
temperature: None,
text: None,
tool_choice: "auto".to_string(),
tools: vec![],
top_p: None,
truncation: None,
user: None,
safety_identifier: None,
metadata: HashMap::new(),
usage: Some(ResponsesUsage::Modern(ResponseUsage {
input_tokens: prompt_tokens,
output_tokens: completion_tokens,
total_tokens: prompt_tokens + completion_tokens,
input_tokens_details: None,
output_tokens_details: None,
})),
}),
response: Box::new(
ResponsesResponse::builder(&emitter.response_id, "")
.status(ResponseStatus::Completed)
.usage(ResponsesUsage::Modern(ResponseUsage {
input_tokens: prompt_tokens,
output_tokens: completion_tokens,
total_tokens: prompt_tokens + completion_tokens,
input_tokens_details: None,
output_tokens_details: None,
}))
.build(),
),
usage: Usage {
prompt_tokens,
completion_tokens,

22
sgl-router/src/routers/grpc/regular/processor.rs
View File

@@ -197,16 +197,14 @@ impl ResponseProcessor {
};
// Step 6: Build ChatChoice
let choice = ChatChoice {
Ok(ChatChoice {
index: index as u32,
message: chat_message,
logprobs,
finish_reason: Some(final_finish_reason_str.to_string()),
matched_stop,
hidden_states: None,
};
Ok(choice)
})
}
/// Process non-streaming chat response (collects all responses and builds final response)
@@ -289,14 +287,14 @@ impl ResponseProcessor {
let usage = response_formatting::build_usage(&all_responses);
// Build final ChatCompletionResponse
let response = response_formatting::build_chat_response(
choices,
&dispatch,
dispatch.model.clone(),
usage,
);
Ok(response)
Ok(
ChatCompletionResponse::builder(&dispatch.request_id, &dispatch.model)
.created(dispatch.created)
.choices(choices)
.usage(usage)
.maybe_system_fingerprint(dispatch.weight_version.clone())
.build(),
)
}
/// Parse tool calls using model-specific parser

37
sgl-router/src/routers/grpc/regular/responses/conversions.rs
View File

@@ -12,7 +12,7 @@ use crate::{
chat::{ChatCompletionRequest, ChatCompletionResponse, ChatMessage, UserMessageContent},
common::{
FunctionCallResponse, JsonSchemaFormat, ResponseFormat, StreamOptions, ToolCall,
ToolChoice, UsageInfo,
UsageInfo,
},
responses::{
ResponseContentPart, ResponseInput, ResponseInputOutputItem, ResponseOutputItem,
@@ -352,32 +352,15 @@ pub fn chat_to_responses(
});
// Generate response
Ok(ResponsesResponse {
id: response_id_override.unwrap_or_else(|| chat_resp.id.clone()),
object: "response".to_string(),
created_at: chat_resp.created as i64,
status,
error: None,
incomplete_details: None,
instructions: original_req.instructions.clone(),
max_output_tokens: original_req.max_output_tokens,
model: chat_resp.model.clone(),
output,
parallel_tool_calls: original_req.parallel_tool_calls.unwrap_or(true),
previous_response_id: original_req.previous_response_id.clone(),
reasoning: None, // TODO: Map reasoning effort if needed
store: original_req.store.unwrap_or(true),
temperature: original_req.temperature,
text: original_req.text.clone(),
tool_choice: ToolChoice::serialize_to_string(&original_req.tool_choice),
tools: original_req.tools.clone().unwrap_or_default(),
top_p: original_req.top_p,
truncation: None,
usage,
user: None,
safety_identifier: original_req.user.clone(),
metadata: original_req.metadata.clone().unwrap_or_default(),
})
let response_id = response_id_override.unwrap_or_else(|| chat_resp.id.clone());
Ok(ResponsesResponse::builder(&response_id, &chat_resp.model)
.copy_from_request(original_req)
.created_at(chat_resp.created as i64)
.status(status)
.output(output)
.maybe_text(original_req.text.clone())
.maybe_usage(usage)
.build())
}
#[cfg(test)]

36
sgl-router/src/routers/grpc/regular/responses/handlers.rs
View File

@@ -58,7 +58,7 @@ use crate::{
},
protocols::{
chat::{self, ChatCompletionStreamResponse},
common::{self, ToolChoice},
common::{self},
responses::{
self, ResponseContentPart, ResponseInput, ResponseInputOutputItem, ResponseOutputItem,
ResponseReasoningContent, ResponseStatus, ResponsesRequest, ResponsesResponse,
@@ -147,7 +147,6 @@ pub async fn route_responses(
route_responses_streaming(ctx, request, headers, model_id).await
} else {
// Generate response ID for synchronous execution
// TODO: we may remove this when we have builder pattern for responses
let response_id = Some(format!("resp_{}", Uuid::new_v4()));
route_responses_sync(ctx, request, headers, model_id, response_id).await
}
@@ -621,32 +620,13 @@ impl StreamingResponseAccumulator {
ResponsesUsage::Classic(usage_info)
});
ResponsesResponse {
id: self.response_id,
object: "response".to_string(),
created_at: self.created_at,
status,
error: None,
incomplete_details: None,
instructions: self.original_request.instructions.clone(),
max_output_tokens: self.original_request.max_output_tokens,
model: self.model,
output,
parallel_tool_calls: self.original_request.parallel_tool_calls.unwrap_or(true),
previous_response_id: self.original_request.previous_response_id.clone(),
reasoning: None,
store: self.original_request.store.unwrap_or(true),
temperature: self.original_request.temperature,
text: None,
tool_choice: ToolChoice::serialize_to_string(&self.original_request.tool_choice),
tools: self.original_request.tools.clone().unwrap_or_default(),
top_p: self.original_request.top_p,
truncation: None,
usage,
user: None,
safety_identifier: self.original_request.user.clone(),
metadata: self.original_request.metadata.clone().unwrap_or_default(),
}
ResponsesResponse::builder(&self.response_id, &self.model)
.copy_from_request(&self.original_request)
.created_at(self.created_at)
.status(status)
.output(output)
.maybe_usage(usage)
.build()
}
}

14
sgl-router/src/routers/grpc/regular/responses/tool_loop.rs
View File

@@ -1148,12 +1148,8 @@ impl ChatResponseAccumulator {
tool_calls_vec.sort_by_key(|(index, _)| *index);
let tool_calls: Vec<_> = tool_calls_vec.into_iter().map(|(_, call)| call).collect();
ChatCompletionResponse {
id: self.id,
object: "chat.completion".to_string(),
created: chrono::Utc::now().timestamp() as u64,
model: self.model,
choices: vec![ChatChoice {
ChatCompletionResponse::builder(&self.id, &self.model)
.choices(vec![ChatChoice {
index: 0,
message: ChatCompletionMessage {
role: "assistant".to_string(),
@@ -1173,9 +1169,7 @@ impl ChatResponseAccumulator {
logprobs: None,
matched_stop: None,
hidden_states: None,
}],
usage: None,
system_fingerprint: None,
}
}])
.build()
}
}

332
sgl-router/src/routers/grpc/regular/streaming.rs
View File

@@ -19,12 +19,10 @@ use tracing::{debug, error, warn};
use crate::{
grpc_client::{proto, sglang_scheduler::AbortOnDropStream},
protocols::{
chat::{
ChatCompletionRequest, ChatCompletionStreamResponse, ChatMessageDelta, ChatStreamChoice,
},
chat::{ChatCompletionRequest, ChatCompletionStreamResponse},
common::{
ChatLogProbs, FunctionCallDelta, StringOrArray, Tool, ToolCallDelta, ToolChoice,
ToolChoiceValue, Usage,
FunctionCallDelta, StringOrArray, Tool, ToolCallDelta, ToolChoice, ToolChoiceValue,
Usage,
},
generate::GenerateRequest,
},
@@ -292,26 +290,11 @@ impl StreamingProcessor {
// Send first chunk with role
if is_firsts.get(&index).copied().unwrap_or(true) {
let first_chunk = ChatCompletionStreamResponse {
id: request_id.clone(),
object: "chat.completion.chunk".to_string(),
created,
model: model.clone(),
system_fingerprint: system_fingerprint.map(|s| s.to_string()),
choices: vec![ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: None,
tool_calls: None,
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
}],
usage: None,
};
let first_chunk = ChatCompletionStreamResponse::builder(request_id, model)
.created(created)
.add_choice_role(index, "assistant")
.maybe_system_fingerprint(system_fingerprint.map(|s| s.to_string()))
.build();
Self::format_sse_chunk_into(&mut sse_buffer, &first_chunk);
tx.send(Ok(Bytes::from(sse_buffer.clone())))
.map_err(|_| "Failed to send first chunk".to_string())?;
@@ -399,15 +382,17 @@ impl StreamingProcessor {
// Regular content emission
if !delta.is_empty() {
let content_chunk = Self::create_content_chunk(
delta,
index,
request_id,
model,
created,
system_fingerprint,
choice_logprobs,
);
let content_chunk =
ChatCompletionStreamResponse::builder(request_id, model)
.created(created)
.add_choice_content_with_logprobs(
index,
"assistant",
delta,
choice_logprobs,
)
.maybe_system_fingerprint(system_fingerprint)
.build();
Self::format_sse_chunk_into(&mut sse_buffer, &content_chunk);
tx.send(Ok(Bytes::from(sse_buffer.clone())))
.map_err(|_| "Failed to send content chunk".to_string())?;
@@ -423,26 +408,14 @@ impl StreamingProcessor {
let stream_buffer = stream_buffers.entry(index).or_default();
stream_buffer.push_str(&text);
let content_chunk = ChatCompletionStreamResponse {
id: request_id.clone(),
object: "chat.completion.chunk".to_string(),
created,
model: model.clone(),
system_fingerprint: system_fingerprint.map(|s| s.to_string()),
choices: vec![ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: Some(text),
tool_calls: None,
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
}],
usage: None,
};
let content_chunk =
ChatCompletionStreamResponse::builder(request_id, model)
.created(created)
.add_choice_content(index, "assistant", text)
.maybe_system_fingerprint(
system_fingerprint.map(|s| s.to_string()),
)
.build();
let sse_chunk =
serde_json::to_string(&content_chunk).map_err(|e| {
@@ -498,26 +471,11 @@ impl StreamingProcessor {
}),
};
let tool_chunk = ChatCompletionStreamResponse {
id: request_id.clone(),
object: "chat.completion.chunk".to_string(),
created,
model: model.clone(),
system_fingerprint: system_fingerprint.map(|s| s.to_string()),
choices: vec![ChatStreamChoice {
index: *index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: None,
tool_calls: Some(vec![tool_call_delta]),
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
}],
usage: None,
};
let tool_chunk = ChatCompletionStreamResponse::builder(request_id, model)
.created(created)
.add_choice_tool_call_delta(*index, tool_call_delta)
.maybe_system_fingerprint(system_fingerprint.map(|s| s.to_string()))
.build();
let sse_chunk = serde_json::to_string(&tool_chunk)
.map_err(|e| format!("Failed to serialize tool chunk: {}", e))?;
@@ -538,26 +496,11 @@ impl StreamingProcessor {
let matched_stop_value = matched_stops.get(index).and_then(|v| v.clone());
let finish_chunk = ChatCompletionStreamResponse {
id: request_id.clone(),
object: "chat.completion.chunk".to_string(),
created,
model: model.clone(),
system_fingerprint: system_fingerprint.map(|s| s.to_string()),
choices: vec![ChatStreamChoice {
index: *index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: None,
tool_calls: None,
reasoning_content: None,
},
logprobs: None,
finish_reason: Some(final_finish_reason),
matched_stop: matched_stop_value,
}],
usage: None,
};
let finish_chunk = ChatCompletionStreamResponse::builder(request_id, model)
.created(created)
.add_choice_finish_reason(*index, final_finish_reason, matched_stop_value)
.maybe_system_fingerprint(system_fingerprint.map(|s| s.to_string()))
.build();
let sse_chunk = serde_json::to_string(&finish_chunk)
.map_err(|e| format!("Failed to serialize finish chunk: {}", e))?;
@@ -571,20 +514,16 @@ impl StreamingProcessor {
let total_prompt: u32 = prompt_tokens.values().sum();
let total_completion: u32 = completion_tokens.values().sum();
let usage_chunk = ChatCompletionStreamResponse {
id: request_id.clone(),
object: "chat.completion.chunk".to_string(),
created,
model: model.clone(),
system_fingerprint: system_fingerprint.map(|s| s.to_string()),
choices: vec![],
usage: Some(Usage {
let usage_chunk = ChatCompletionStreamResponse::builder(request_id, model)
.created(created)
.usage(Usage {
prompt_tokens: total_prompt,
completion_tokens: total_completion,
total_tokens: total_prompt + total_completion,
completion_tokens_details: None,
}),
};
})
.maybe_system_fingerprint(system_fingerprint.map(|s| s.to_string()))
.build();
let sse_chunk = serde_json::to_string(&usage_chunk)
.map_err(|e| format!("Failed to serialize usage chunk: {}", e))?;
@@ -971,7 +910,7 @@ impl StreamingProcessor {
);
// Send final chunk with finish_reason
let finish_response = serde_json::json!({
let finish_response = json!({
"text": accumulated_text,
"output_ids": complete.output_ids[complete.output_ids.len().saturating_sub(1)..].to_vec(),
"meta_info": {
@@ -1082,26 +1021,13 @@ impl StreamingProcessor {
normal_text,
}) => {
let chunk = if !reasoning_text.is_empty() {
Some(ChatCompletionStreamResponse {
id: request_id.to_string(),
object: "chat.completion.chunk".to_string(),
created,
model: model.to_string(),
system_fingerprint: system_fingerprint.map(|s| s.to_string()),
choices: vec![ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: None,
tool_calls: None,
reasoning_content: Some(reasoning_text),
},
logprobs: None,
finish_reason: None,
matched_stop: None,
}],
usage: None,
})
Some(
ChatCompletionStreamResponse::builder(request_id, model)
.created(created)
.add_choice_reasoning(index, reasoning_text)
.maybe_system_fingerprint(system_fingerprint.map(|s| s.to_string()))
.build(),
)
} else {
None
};
@@ -1145,66 +1071,24 @@ impl StreamingProcessor {
history_tool_calls_count,
);
chunks.push(ChatCompletionStreamResponse {
id: request_id.to_string(),
object: "chat.completion.chunk".to_string(),
created,
model: model.to_string(),
system_fingerprint: system_fingerprint.map(|s| s.to_string()),
choices: vec![ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: None,
tool_calls: Some(vec![ToolCallDelta {
index: 0,
id: Some(tool_call_id),
tool_type: Some("function".to_string()),
function: Some(FunctionCallDelta {
name: Some(function.name.clone()),
arguments: None,
}),
}]),
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
}],
usage: None,
});
chunks.push(
ChatCompletionStreamResponse::builder(request_id, model)
.created(created)
.add_choice_tool_name(index, tool_call_id, function.name.clone())
.maybe_system_fingerprint(system_fingerprint.map(|s| s.to_string()))
.build(),
);
}
// Emit arguments delta
if !delta.is_empty() {
chunks.push(ChatCompletionStreamResponse {
id: request_id.to_string(),
object: "chat.completion.chunk".to_string(),
created,
model: model.to_string(),
system_fingerprint: system_fingerprint.map(|s| s.to_string()),
choices: vec![ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: None,
tool_calls: Some(vec![ToolCallDelta {
index: 0,
id: None,
tool_type: None,
function: Some(FunctionCallDelta {
name: None,
arguments: Some(delta.to_string()),
}),
}]),
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
}],
usage: None,
});
chunks.push(
ChatCompletionStreamResponse::builder(request_id, model)
.created(created)
.add_choice_tool_args(index, delta.to_string())
.maybe_system_fingerprint(system_fingerprint.map(|s| s.to_string()))
.build(),
);
}
}
@@ -1256,26 +1140,13 @@ impl StreamingProcessor {
Ok(StreamingParseResult { normal_text, calls }) => {
// Emit normal text if present
if !normal_text.is_empty() {
chunks.push(ChatCompletionStreamResponse {
id: request_id.to_string(),
object: "chat.completion.chunk".to_string(),
created,
model: model.to_string(),
system_fingerprint: system_fingerprint.map(|s| s.to_string()),
choices: vec![ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: Some(normal_text),
tool_calls: None,
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
}],
usage: None,
});
chunks.push(
ChatCompletionStreamResponse::builder(request_id, model)
.created(created)
.add_choice_content(index, "assistant", normal_text)
.maybe_system_fingerprint(system_fingerprint.map(|s| s.to_string()))
.build(),
);
}
// Emit tool call chunks
@@ -1311,26 +1182,13 @@ impl StreamingProcessor {
}),
};
chunks.push(ChatCompletionStreamResponse {
id: request_id.to_string(),
object: "chat.completion.chunk".to_string(),
created,
model: model.to_string(),
system_fingerprint: system_fingerprint.map(|s| s.to_string()),
choices: vec![ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: None,
tool_calls: Some(vec![tool_call_delta]),
reasoning_content: None,
},
logprobs: None,
finish_reason: None,
matched_stop: None,
}],
usage: None,
});
chunks.push(
ChatCompletionStreamResponse::builder(request_id, model)
.created(created)
.add_choice_tool_call_delta(index, tool_call_delta)
.maybe_system_fingerprint(system_fingerprint.map(|s| s.to_string()))
.build(),
);
}
return chunks;
@@ -1359,38 +1217,6 @@ impl StreamingProcessor {
}
buffer.extend_from_slice(b"\n\n");
}
/// Create a content chunk response
fn create_content_chunk(
content: String,
index: u32,
request_id: &str,
model: &str,
created: u64,
system_fingerprint: Option<&str>,
logprobs: Option<ChatLogProbs>,
) -> ChatCompletionStreamResponse {
ChatCompletionStreamResponse {
id: request_id.to_string(),
object: "chat.completion.chunk".to_string(),
created,
model: model.to_string(),
system_fingerprint: system_fingerprint.map(|s| s.to_string()),
choices: vec![ChatStreamChoice {
index,
delta: ChatMessageDelta {
role: Some("assistant".to_string()),
content: Some(content),
tool_calls: None,
reasoning_content: None,
},
logprobs,
finish_reason: None,
matched_stop: None,
}],
usage: None,
}
}
}
/// Build SSE response with proper headers