From fe19a580fb9c9babb2394a9ce4ae4191d29b7a15 Mon Sep 17 00:00:00 2001 From: Chang Su Date: Fri, 7 Nov 2025 15:43:32 -0800 Subject: [PATCH] [router][grpc] Refactor: Add builders for chat and responses (#12852) --- sgl-router/src/protocols/builders/chat/mod.rs | 7 + .../src/protocols/builders/chat/response.rs | 218 +++++++++ .../builders/chat/stream_response.rs | 421 ++++++++++++++++++ sgl-router/src/protocols/builders/mod.rs | 27 ++ .../src/protocols/builders/responses/mod.rs | 5 + .../protocols/builders/responses/response.rs | 411 +++++++++++++++++ sgl-router/src/protocols/chat.rs | 25 +- sgl-router/src/protocols/mod.rs | 1 + sgl-router/src/protocols/responses.rs | 8 +- .../grpc/common/response_formatting.rs | 38 +- .../grpc/common/responses/streaming.rs | 47 +- .../src/routers/grpc/harmony/processor.rs | 56 +-- .../src/routers/grpc/harmony/responses.rs | 40 +- .../src/routers/grpc/harmony/streaming.rs | 150 ++----- .../src/routers/grpc/regular/processor.rs | 22 +- .../grpc/regular/responses/conversions.rs | 37 +- .../grpc/regular/responses/handlers.rs | 36 +- .../grpc/regular/responses/tool_loop.rs | 14 +- .../src/routers/grpc/regular/streaming.rs | 332 ++++---------- 19 files changed, 1315 insertions(+), 580 deletions(-) create mode 100644 sgl-router/src/protocols/builders/chat/mod.rs create mode 100644 sgl-router/src/protocols/builders/chat/response.rs create mode 100644 sgl-router/src/protocols/builders/chat/stream_response.rs create mode 100644 sgl-router/src/protocols/builders/mod.rs create mode 100644 sgl-router/src/protocols/builders/responses/mod.rs create mode 100644 sgl-router/src/protocols/builders/responses/response.rs diff --git a/sgl-router/src/protocols/builders/chat/mod.rs b/sgl-router/src/protocols/builders/chat/mod.rs new file mode 100644 index 000000000..61386b505 --- /dev/null +++ b/sgl-router/src/protocols/builders/chat/mod.rs @@ -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; diff --git a/sgl-router/src/protocols/builders/chat/response.rs b/sgl-router/src/protocols/builders/chat/response.rs new file mode 100644 index 000000000..7267e2ee2 --- /dev/null +++ b/sgl-router/src/protocols/builders/chat/response.rs @@ -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, + usage: Option, + system_fingerprint: Option, +} + +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, model: impl Into) -> 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) -> 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) -> 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) -> 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>) -> 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 + } +} diff --git a/sgl-router/src/protocols/builders/chat/stream_response.rs b/sgl-router/src/protocols/builders/chat/stream_response.rs new file mode 100644 index 000000000..9afe486aa --- /dev/null +++ b/sgl-router/src/protocols/builders/chat/stream_response.rs @@ -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, + usage: Option, + system_fingerprint: Option, +} + +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, model: impl Into) -> 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) -> 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) -> 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>) -> 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) -> 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, + content: impl Into, + ) -> 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, + content: impl Into, + logprobs: Option, + ) -> 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) -> 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>, + ) -> 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) -> 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, + function_name: impl Into, + ) -> 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, + matched_stop: Option, + ) -> 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"); + } +} diff --git a/sgl-router/src/protocols/builders/mod.rs b/sgl-router/src/protocols/builders/mod.rs new file mode 100644 index 000000000..920844124 --- /dev/null +++ b/sgl-router/src/protocols/builders/mod.rs @@ -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` directly: +//! ```ignore +//! builder +//! .field(value) +//! .maybe_optional_field(optional_value) // Accepts Option +//! .build() +//! ``` + +pub mod chat; +pub mod responses; + +// Re-export all builders for convenient access +pub use chat::{ChatCompletionResponseBuilder, ChatCompletionStreamResponseBuilder}; +pub use responses::ResponsesResponseBuilder; diff --git a/sgl-router/src/protocols/builders/responses/mod.rs b/sgl-router/src/protocols/builders/responses/mod.rs new file mode 100644 index 000000000..aca9d2492 --- /dev/null +++ b/sgl-router/src/protocols/builders/responses/mod.rs @@ -0,0 +1,5 @@ +//! Builders for Responses API response types + +pub mod response; + +pub use response::ResponsesResponseBuilder; diff --git a/sgl-router/src/protocols/builders/responses/response.rs b/sgl-router/src/protocols/builders/responses/response.rs new file mode 100644 index 000000000..50d3f1bc5 --- /dev/null +++ b/sgl-router/src/protocols/builders/responses/response.rs @@ -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, + incomplete_details: Option, + instructions: Option, + max_output_tokens: Option, + model: String, + output: Vec, + parallel_tool_calls: bool, + previous_response_id: Option, + reasoning: Option, + store: bool, + temperature: Option, + text: Option, + tool_choice: String, + tools: Vec, + top_p: Option, + truncation: Option, + usage: Option, + user: Option, + safety_identifier: Option, + metadata: HashMap, +} + +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, model: impl Into) -> 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) -> 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) -> 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) -> 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) -> 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) -> 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) -> Self { + self.tool_choice = tool_choice.into(); + self + } + + /// Set available tools + pub fn tools(mut self, tools: Vec) -> 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) -> 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) -> 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) -> Self { + self.user = Some(user.into()); + self + } + + /// Set safety identifier + pub fn safety_identifier(mut self, identifier: impl Into) -> Self { + self.safety_identifier = Some(identifier.into()); + self + } + + /// Set metadata + pub fn metadata(mut self, metadata: HashMap) -> Self { + self.metadata = metadata; + self + } + + /// Add a single metadata entry + pub fn add_metadata(mut self, key: impl Into, 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); + } +} diff --git a/sgl-router/src/protocols/chat.rs b/sgl-router/src/protocols/chat.rs index 54ee58dba..8d8de1e5c 100644 --- a/sgl-router/src/protocols/chat.rs +++ b/sgl-router/src/protocols/chat.rs @@ -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, } +impl ChatCompletionResponse { + /// Create a new builder for ChatCompletionResponse + pub fn builder( + id: impl Into, + model: impl Into, + ) -> 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, } +impl ChatCompletionStreamResponse { + /// Create a new builder for ChatCompletionStreamResponse + pub fn builder( + id: impl Into, + model: impl Into, + ) -> ChatCompletionStreamResponseBuilder { + ChatCompletionStreamResponseBuilder::new(id, model) + } +} + /// Delta structure for streaming chat completion responses #[derive(Debug, Clone, Deserialize, Serialize)] pub struct ChatMessageDelta { diff --git a/sgl-router/src/protocols/mod.rs b/sgl-router/src/protocols/mod.rs index d9d2526b9..813ff48df 100644 --- a/sgl-router/src/protocols/mod.rs +++ b/sgl-router/src/protocols/mod.rs @@ -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; diff --git a/sgl-router/src/protocols/responses.rs b/sgl-router/src/protocols/responses.rs index ed3ba3a30..16e40dcd9 100644 --- a/sgl-router/src/protocols/responses.rs +++ b/sgl-router/src/protocols/responses.rs @@ -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, model: impl Into) -> ResponsesResponseBuilder { + ResponsesResponseBuilder::new(id, model) + } + /// Check if the response is complete pub fn is_complete(&self) -> bool { matches!(self.status, ResponseStatus::Completed) diff --git a/sgl-router/src/routers/grpc/common/response_formatting.rs b/sgl-router/src/routers/grpc/common/response_formatting.rs index 8cd66520d..241cafa89 100644 --- a/sgl-router/src/routers/grpc/common/response_formatting.rs +++ b/sgl-router/src/routers/grpc/common/response_formatting.rs @@ -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, - 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(), - } -} diff --git a/sgl-router/src/routers/grpc/common/responses/streaming.rs b/sgl-router/src/routers/grpc/common/responses/streaming.rs index a44d9ffe6..47bea0d91 100644 --- a/sgl-router/src/routers/grpc/common/responses/streaming.rs +++ b/sgl-router/src/routers/grpc/common/responses/streaming.rs @@ -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) diff --git a/sgl-router/src/routers/grpc/harmony/processor.rs b/sgl-router/src/routers/grpc/harmony/processor.rs index 6d6ae8fa8..c4255220c 100644 --- a/sgl-router/src/routers/grpc/harmony/processor.rs +++ b/sgl-router/src/routers/grpc/harmony/processor.rs @@ -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), diff --git a/sgl-router/src/routers/grpc/harmony/responses.rs b/sgl-router/src/routers/grpc/harmony/responses.rs index 8989c4539..e669ca5b0 100644 --- a/sgl-router/src/routers/grpc/harmony/responses.rs +++ b/sgl-router/src/routers/grpc/harmony/responses.rs @@ -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, @@ -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 diff --git a/sgl-router/src/routers/grpc/harmony/streaming.rs b/sgl-router/src/routers/grpc/harmony/streaming.rs index cae9093cf..5e45f22a5 100644 --- a/sgl-router/src/routers/grpc/harmony/streaming.rs +++ b/sgl-router/src/routers/grpc/harmony/streaming.rs @@ -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<(), 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<(), 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, diff --git a/sgl-router/src/routers/grpc/regular/processor.rs b/sgl-router/src/routers/grpc/regular/processor.rs index 59e95084a..d63183d2c 100644 --- a/sgl-router/src/routers/grpc/regular/processor.rs +++ b/sgl-router/src/routers/grpc/regular/processor.rs @@ -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 diff --git a/sgl-router/src/routers/grpc/regular/responses/conversions.rs b/sgl-router/src/routers/grpc/regular/responses/conversions.rs index 89a66bbb2..802eb73f9 100644 --- a/sgl-router/src/routers/grpc/regular/responses/conversions.rs +++ b/sgl-router/src/routers/grpc/regular/responses/conversions.rs @@ -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)] diff --git a/sgl-router/src/routers/grpc/regular/responses/handlers.rs b/sgl-router/src/routers/grpc/regular/responses/handlers.rs index 09e44eb51..4b6267bef 100644 --- a/sgl-router/src/routers/grpc/regular/responses/handlers.rs +++ b/sgl-router/src/routers/grpc/regular/responses/handlers.rs @@ -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() } } diff --git a/sgl-router/src/routers/grpc/regular/responses/tool_loop.rs b/sgl-router/src/routers/grpc/regular/responses/tool_loop.rs index 3595d281d..2a8715805 100644 --- a/sgl-router/src/routers/grpc/regular/responses/tool_loop.rs +++ b/sgl-router/src/routers/grpc/regular/responses/tool_loop.rs @@ -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() } } diff --git a/sgl-router/src/routers/grpc/regular/streaming.rs b/sgl-router/src/routers/grpc/regular/streaming.rs index 8c8ec1105..f68993617 100644 --- a/sgl-router/src/routers/grpc/regular/streaming.rs +++ b/sgl-router/src/routers/grpc/regular/streaming.rs @@ -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, - ) -> 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