From 4d2f17bd0bb5c6f135f5b945840b95ee026192bb Mon Sep 17 00:00:00 2001 From: Keyang Ru Date: Thu, 30 Oct 2025 12:45:10 -0700 Subject: [PATCH] [router] Function call support for openai router Responses API (#12386) --- .github/workflows/pr-test-rust.yml | 1 + scripts/ci/ci_install_rust.sh | 2 +- sgl-router/src/data_connector/core.rs | 2 +- sgl-router/src/protocols/responses.rs | 18 ++-- .../src/routers/grpc/harmony/responses.rs | 1 + .../src/routers/grpc/responses/conversions.rs | 1 + .../src/routers/grpc/responses/handlers.rs | 1 + .../src/routers/grpc/responses/tool_loop.rs | 1 + .../src/routers/openai/conversations.rs | 62 ++++++++++--- sgl-router/src/routers/openai/router.rs | 86 ++++++++++++++++--- 10 files changed, 144 insertions(+), 31 deletions(-) diff --git a/.github/workflows/pr-test-rust.yml b/.github/workflows/pr-test-rust.yml index 02cf5682c..ce833eb17 100644 --- a/.github/workflows/pr-test-rust.yml +++ b/.github/workflows/pr-test-rust.yml @@ -97,6 +97,7 @@ jobs: run: | source "$HOME/.cargo/env" cd sgl-router/ + rustup component add clippy cargo clippy --all-targets --all-features -- -D warnings - name: Run fmt diff --git a/scripts/ci/ci_install_rust.sh b/scripts/ci/ci_install_rust.sh index ac042fc9a..7f67b820c 100755 --- a/scripts/ci/ci_install_rust.sh +++ b/scripts/ci/ci_install_rust.sh @@ -11,7 +11,7 @@ else fi # Install rustup (Rust installer and version manager) -curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y +curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --default-toolchain 1.90 # Follow the installation prompts, then reload your shell diff --git a/sgl-router/src/data_connector/core.rs b/sgl-router/src/data_connector/core.rs index 95f21a13c..e61b4163b 100644 --- a/sgl-router/src/data_connector/core.rs +++ b/sgl-router/src/data_connector/core.rs @@ -268,7 +268,7 @@ pub fn make_item_id(item_type: &str) -> ConversationItemId { "reasoning" => "rs".to_string(), "mcp_call" => "mcp".to_string(), "mcp_list_tools" => "mcpl".to_string(), - "function_tool_call" => "ftc".to_string(), + "function_call" => "fc".to_string(), other => { // Fallback: first 3 letters of type or "itm" let mut p = other.chars().take(3).collect::(); diff --git a/sgl-router/src/protocols/responses.rs b/sgl-router/src/protocols/responses.rs index 3e1db426d..d09f96790 100644 --- a/sgl-router/src/protocols/responses.rs +++ b/sgl-router/src/protocols/responses.rs @@ -9,7 +9,7 @@ use validator::Validate; // Import shared types from common module use super::common::{ - default_model, default_true, ChatLogProbs, GenerationRequest, PromptTokenUsageInfo, + default_model, default_true, ChatLogProbs, Function, GenerationRequest, PromptTokenUsageInfo, StringOrArray, ToolChoice, UsageInfo, }; @@ -22,8 +22,10 @@ pub struct ResponseTool { #[serde(rename = "type")] pub r#type: ResponseToolType, // Function tool fields (used when type == "function") + // In Responses API, function fields are flattened at the top level + #[serde(flatten)] #[serde(skip_serializing_if = "Option::is_none")] - pub function: Option, + pub function: Option, // MCP-specific fields (used when type == "mcp") #[serde(skip_serializing_if = "Option::is_none")] pub server_url: Option, @@ -123,15 +125,16 @@ pub enum ResponseInputOutputItem { #[serde(rename = "reasoning")] Reasoning { id: String, - #[serde(skip_serializing_if = "Vec::is_empty")] summary: Vec, + #[serde(skip_serializing_if = "Vec::is_empty")] content: Vec, #[serde(skip_serializing_if = "Option::is_none")] status: Option, }, - #[serde(rename = "function_tool_call")] + #[serde(rename = "function_call")] FunctionToolCall { id: String, + call_id: String, name: String, arguments: String, #[serde(skip_serializing_if = "Option::is_none")] @@ -141,6 +144,7 @@ pub enum ResponseInputOutputItem { }, #[serde(rename = "function_call_output")] FunctionCallOutput { + id: Option, call_id: String, output: String, #[serde(skip_serializing_if = "Option::is_none")] @@ -207,15 +211,15 @@ pub enum ResponseOutputItem { #[serde(rename = "reasoning")] Reasoning { id: String, - #[serde(skip_serializing_if = "Vec::is_empty")] summary: Vec, content: Vec, #[serde(skip_serializing_if = "Option::is_none")] status: Option, }, - #[serde(rename = "function_tool_call")] + #[serde(rename = "function_call")] FunctionToolCall { id: String, + call_id: String, name: String, arguments: String, #[serde(skip_serializing_if = "Option::is_none")] @@ -925,6 +929,7 @@ impl ResponseOutputItem { /// Create a new function tool call output item pub fn new_function_tool_call( id: String, + call_id: String, name: String, arguments: String, output: Option, @@ -932,6 +937,7 @@ impl ResponseOutputItem { ) -> Self { Self::FunctionToolCall { id, + call_id, name, arguments, output, diff --git a/sgl-router/src/routers/grpc/harmony/responses.rs b/sgl-router/src/routers/grpc/harmony/responses.rs index dc4ee54be..e82f45852 100644 --- a/sgl-router/src/routers/grpc/harmony/responses.rs +++ b/sgl-router/src/routers/grpc/harmony/responses.rs @@ -912,6 +912,7 @@ fn build_next_request_with_tools( for tool_call in tool_calls { items.push(ResponseInputOutputItem::FunctionToolCall { id: tool_call.id.clone(), + call_id: tool_call.id.clone(), name: tool_call.function.name.clone(), arguments: tool_call .function diff --git a/sgl-router/src/routers/grpc/responses/conversions.rs b/sgl-router/src/routers/grpc/responses/conversions.rs index 147784dac..0058fe5b4 100644 --- a/sgl-router/src/routers/grpc/responses/conversions.rs +++ b/sgl-router/src/routers/grpc/responses/conversions.rs @@ -304,6 +304,7 @@ pub fn chat_to_responses( for tool_call in tool_calls { output.push(ResponseOutputItem::FunctionToolCall { id: tool_call.id.clone(), + call_id: tool_call.id.clone(), name: tool_call.function.name.clone(), arguments: tool_call.function.arguments.clone().unwrap_or_default(), output: None, // Tool hasn't been executed yet diff --git a/sgl-router/src/routers/grpc/responses/handlers.rs b/sgl-router/src/routers/grpc/responses/handlers.rs index 7da43c5a6..4e8b84085 100644 --- a/sgl-router/src/routers/grpc/responses/handlers.rs +++ b/sgl-router/src/routers/grpc/responses/handlers.rs @@ -721,6 +721,7 @@ impl StreamingResponseAccumulator { while self.tool_calls.len() <= index { self.tool_calls.push(ResponseOutputItem::FunctionToolCall { id: String::new(), + call_id: String::new(), name: String::new(), arguments: String::new(), output: None, diff --git a/sgl-router/src/routers/grpc/responses/tool_loop.rs b/sgl-router/src/routers/grpc/responses/tool_loop.rs index eb01001c8..caa689936 100644 --- a/sgl-router/src/routers/grpc/responses/tool_loop.rs +++ b/sgl-router/src/routers/grpc/responses/tool_loop.rs @@ -124,6 +124,7 @@ impl ToolLoopState { self.conversation_history .push(ResponseInputOutputItem::FunctionToolCall { id: call_id.clone(), + call_id: call_id.clone(), name: tool_name.clone(), arguments: args_json_str.clone(), output: Some(output_str.clone()), diff --git a/sgl-router/src/routers/openai/conversations.rs b/sgl-router/src/routers/openai/conversations.rs index 4d754576b..44332f228 100644 --- a/sgl-router/src/routers/openai/conversations.rs +++ b/sgl-router/src/routers/openai/conversations.rs @@ -384,9 +384,9 @@ const SUPPORTED_ITEM_TYPES: &[&str] = &[ "mcp_list_tools", "mcp_call", "item_reference", - // Accepted but not yet implemented (stored, warning returned) - "function_tool_call", + "function_call", "function_call_output", + // Accepted but not yet implemented (stored, warning returned) "file_search_call", "computer_call", "computer_call_output", @@ -936,6 +936,26 @@ fn item_to_json(item: &crate::data_connector::ConversationItem) -> Value { } } } + "function_call" => { + // Extract function_call fields: call_id, name, arguments, output + if let Some(content_obj) = item.content.as_object() { + for field in ["call_id", "name", "arguments", "output"] { + if let Some(value) = content_obj.get(field) { + obj.insert(field.to_string(), value.clone()); + } + } + } + } + "function_call_output" => { + // Extract function_call_output fields: call_id, output + if let Some(content_obj) = item.content.as_object() { + for field in ["call_id", "output"] { + if let Some(value) = content_obj.get(field) { + obj.insert(field.to_string(), value.clone()); + } + } + } + } _ => { // For all other types (message, reasoning, etc.), keep content as-is obj.insert("content".to_string(), item.content.clone()); @@ -1144,7 +1164,7 @@ fn extract_input_items(input: &ResponseInput) -> Result, String> { })) } _ => { - // For other item types (Message, Reasoning, FunctionToolCall), serialize and ensure ID + // For other item types (Message, Reasoning, FunctionToolCall, FunctionCallOutput), serialize and ensure ID let mut value = serde_json::to_value(item) .map_err(|e| format!("Failed to serialize item: {}", e))?; @@ -1157,7 +1177,15 @@ fn extract_input_items(input: &ResponseInput) -> Result, String> { .map(|s| s.is_empty()) .unwrap_or(true) { - obj.insert("id".to_string(), json!(generate_id("item"))); + // Generate ID with appropriate prefix based on type + let item_type = + obj.get("type").and_then(|v| v.as_str()).unwrap_or("item"); + let prefix = match item_type { + "function_call" | "function_call_output" => "fc", + "message" => "msg", + _ => "item", + }; + obj.insert("id".to_string(), json!(generate_id(prefix))); } } @@ -1201,17 +1229,31 @@ async fn link_items_to_conversation( .get("role") .and_then(|v| v.as_str()) .map(String::from); - let content = input_item_value - .get("content") - .cloned() - .unwrap_or(json!([])); + + // For function_call and function_call_output, store the entire item as content + // For message types, extract just the content field + let content = if item_type == "function_call" || item_type == "function_call_output" { + input_item_value.clone() + } else { + input_item_value + .get("content") + .cloned() + .unwrap_or(json!([])) + }; + let status = input_item_value .get("status") .and_then(|v| v.as_str()) .map(String::from); + // Extract the original item ID from input if present + let item_id = input_item_value + .get("id") + .and_then(|v| v.as_str()) + .map(ConversationItemId::from); + let new_item = NewConversationItem { - id: None, // Let storage generate ID + id: item_id, // Preserve ID if present response_id: response_id_opt.clone(), item_type: item_type.to_string(), role, @@ -1252,7 +1294,7 @@ async fn link_items_to_conversation( .cloned() .unwrap_or(json!([])) } else { - // For other types (reasoning, function_tool_call, mcp_call, etc.) + // For other types (reasoning, function_call, function_call_output, mcp_call, etc.) // store the entire item structure output_item_value.clone() }; diff --git a/sgl-router/src/routers/openai/router.rs b/sgl-router/src/routers/openai/router.rs index d884d7cc6..ef0ade353 100644 --- a/sgl-router/src/routers/openai/router.rs +++ b/sgl-router/src/routers/openai/router.rs @@ -810,20 +810,76 @@ impl crate::routers::RouterTrait for OpenAIRouter { Ok(stored_items) => { let mut items: Vec = Vec::new(); for item in stored_items.into_iter() { - // Only use message items for conversation context - // Skip non-message items (reasoning, function calls, etc.) - if item.item_type == "message" { - if let Ok(content_parts) = - serde_json::from_value::>( + // Include messages, function calls, and function call outputs + // Skip reasoning items as they're internal processing details + match item.item_type.as_str() { + "message" => { + match serde_json::from_value::>( item.content.clone(), - ) - { - items.push(ResponseInputOutputItem::Message { - id: item.id.0.clone(), - role: item.role.clone().unwrap_or_else(|| "user".to_string()), - content: content_parts, - status: item.status.clone(), - }); + ) { + Ok(content_parts) => { + items.push(ResponseInputOutputItem::Message { + id: item.id.0.clone(), + role: item + .role + .clone() + .unwrap_or_else(|| "user".to_string()), + content: content_parts, + status: item.status.clone(), + }); + } + Err(e) => { + tracing::error!( + "Failed to deserialize message content: {}", + e + ); + } + } + } + "function_call" => { + // The entire function_call item is stored in content field + match serde_json::from_value::( + item.content.clone(), + ) { + Ok(func_call) => items.push(func_call), + Err(e) => { + tracing::error!( + "Failed to deserialize function_call: {}", + e + ); + } + } + } + "function_call_output" => { + // The entire function_call_output item is stored in content field + tracing::debug!( + "Loading function_call_output from DB - content: {}", + serde_json::to_string_pretty(&item.content) + .unwrap_or_else(|_| "failed to serialize".to_string()) + ); + match serde_json::from_value::( + item.content.clone(), + ) { + Ok(func_output) => { + tracing::debug!( + "Successfully deserialized function_call_output" + ); + items.push(func_output); + } + Err(e) => { + tracing::error!( + "Failed to deserialize function_call_output: {}", + e + ); + } + } + } + "reasoning" => { + // Skip reasoning items - they're internal processing details + } + _ => { + // Skip unknown item types + warn!("Unknown item type in conversation: {}", item.item_type); } } } @@ -889,6 +945,10 @@ impl crate::routers::RouterTrait for OpenAIRouter { // Always set store=false for upstream (we store internally) request_body.store = Some(false); + // Filter out reasoning items from input - they're internal processing details + if let ResponseInput::Items(ref mut items) = request_body.input { + items.retain(|item| !matches!(item, ResponseInputOutputItem::Reasoning { .. })); + } // Convert to JSON and strip SGLang-specific fields let mut payload = match to_value(&request_body) {