From 05284378d644512a8615c35fc17210fb0a912c81 Mon Sep 17 00:00:00 2001 From: Simo Lin Date: Fri, 5 Dec 2025 08:44:13 -0800 Subject: [PATCH] [model-gateway] move conversation to first class routing (#14506) Co-authored-by: key4ng --- sgl-router/src/routers/mod.rs | 109 -------------- sgl-router/src/routers/openai/router.rs | 116 --------------- sgl-router/src/routers/router_manager.rs | 175 ----------------------- sgl-router/src/server.rs | 82 +++++------ 4 files changed, 36 insertions(+), 446 deletions(-) diff --git a/sgl-router/src/routers/mod.rs b/sgl-router/src/routers/mod.rs index 9ee7aaf0d..41fb34dc6 100644 --- a/sgl-router/src/routers/mod.rs +++ b/sgl-router/src/routers/mod.rs @@ -9,7 +9,6 @@ use axum::{ http::{HeaderMap, StatusCode}, response::{IntoResponse, Response}, }; -use serde_json::Value; use crate::protocols::{ chat::ChatCompletionRequest, @@ -149,112 +148,4 @@ pub trait RouterTrait: Send + Sync + Debug { fn is_pd_mode(&self) -> bool { self.router_type() == "pd" } - - /// Create a new conversation - async fn create_conversation(&self, _headers: Option<&HeaderMap>, _body: &Value) -> Response { - ( - StatusCode::NOT_IMPLEMENTED, - "Conversations not supported by this router", - ) - .into_response() - } - - /// Get a conversation by ID - async fn get_conversation( - &self, - _headers: Option<&HeaderMap>, - _conversation_id: &str, - ) -> Response { - ( - StatusCode::NOT_IMPLEMENTED, - "Conversations not supported by this router", - ) - .into_response() - } - - /// Update a conversation - async fn update_conversation( - &self, - _headers: Option<&HeaderMap>, - _conversation_id: &str, - _body: &Value, - ) -> Response { - ( - StatusCode::NOT_IMPLEMENTED, - "Conversations not supported by this router", - ) - .into_response() - } - - /// Delete a conversation - async fn delete_conversation( - &self, - _headers: Option<&HeaderMap>, - _conversation_id: &str, - ) -> Response { - ( - StatusCode::NOT_IMPLEMENTED, - "Conversations not supported by this router", - ) - .into_response() - } - - /// List items in a conversation - async fn list_conversation_items( - &self, - _headers: Option<&HeaderMap>, - _conversation_id: &str, - _limit: Option, - _order: Option<&str>, - _after: Option<&str>, - ) -> Response { - ( - StatusCode::NOT_IMPLEMENTED, - "Conversations not supported by this router", - ) - .into_response() - } - - /// Create items in a conversation - async fn create_conversation_items( - &self, - _headers: Option<&HeaderMap>, - _conversation_id: &str, - _body: &Value, - ) -> Response { - ( - StatusCode::NOT_IMPLEMENTED, - "Conversations not supported by this router", - ) - .into_response() - } - - /// Get a specific item from a conversation - async fn get_conversation_item( - &self, - _headers: Option<&HeaderMap>, - _conversation_id: &str, - _item_id: &str, - _include: Option>, - ) -> Response { - ( - StatusCode::NOT_IMPLEMENTED, - "Conversations not supported by this router", - ) - .into_response() - } - - /// Delete an item from a conversation - async fn delete_conversation_item( - &self, - _headers: Option<&HeaderMap>, - _conversation_id: &str, - _item_id: &str, - ) -> Response { - ( - StatusCode::NOT_IMPLEMENTED, - "Conversations not supported by this router", - ) - .into_response() - } } diff --git a/sgl-router/src/routers/openai/router.rs b/sgl-router/src/routers/openai/router.rs index d2345722d..6fc878b8f 100644 --- a/sgl-router/src/routers/openai/router.rs +++ b/sgl-router/src/routers/openai/router.rs @@ -1110,120 +1110,4 @@ impl crate::routers::RouterTrait for OpenAIRouter { fn router_type(&self) -> &'static str { "openai" } - - // ============================================================================ - // Conversation API Methods - delegate to conversations module - // ============================================================================ - - async fn create_conversation(&self, _headers: Option<&HeaderMap>, body: &Value) -> Response { - super::conversations::create_conversation( - &self.responses_components.conversation_storage, - body.clone(), - ) - .await - } - - async fn get_conversation( - &self, - _headers: Option<&HeaderMap>, - conversation_id: &str, - ) -> Response { - super::conversations::get_conversation( - &self.responses_components.conversation_storage, - conversation_id, - ) - .await - } - - async fn update_conversation( - &self, - _headers: Option<&HeaderMap>, - conversation_id: &str, - body: &Value, - ) -> Response { - super::conversations::update_conversation( - &self.responses_components.conversation_storage, - conversation_id, - body.clone(), - ) - .await - } - - async fn delete_conversation( - &self, - _headers: Option<&HeaderMap>, - conversation_id: &str, - ) -> Response { - super::conversations::delete_conversation( - &self.responses_components.conversation_storage, - conversation_id, - ) - .await - } - - async fn list_conversation_items( - &self, - _headers: Option<&HeaderMap>, - conversation_id: &str, - limit: Option, - order: Option<&str>, - after: Option<&str>, - ) -> Response { - super::conversations::list_conversation_items( - &self.responses_components.conversation_storage, - &self.responses_components.conversation_item_storage, - conversation_id, - limit, - order, - after, - ) - .await - } - - async fn create_conversation_items( - &self, - _headers: Option<&HeaderMap>, - conversation_id: &str, - body: &Value, - ) -> Response { - super::conversations::create_conversation_items( - &self.responses_components.conversation_storage, - &self.responses_components.conversation_item_storage, - conversation_id, - body.clone(), - ) - .await - } - - async fn get_conversation_item( - &self, - _headers: Option<&HeaderMap>, - conversation_id: &str, - item_id: &str, - include: Option>, - ) -> Response { - super::conversations::get_conversation_item( - &self.responses_components.conversation_storage, - &self.responses_components.conversation_item_storage, - conversation_id, - item_id, - include, - ) - .await - } - - async fn delete_conversation_item( - &self, - _headers: Option<&HeaderMap>, - conversation_id: &str, - item_id: &str, - ) -> Response { - super::conversations::delete_conversation_item( - &self.responses_components.conversation_storage, - &self.responses_components.conversation_item_storage, - conversation_id, - item_id, - ) - .await - } } diff --git a/sgl-router/src/routers/router_manager.rs b/sgl-router/src/routers/router_manager.rs index bac979f88..9ac494ff9 100644 --- a/sgl-router/src/routers/router_manager.rs +++ b/sgl-router/src/routers/router_manager.rs @@ -579,181 +579,6 @@ impl RouterTrait for RouterManager { fn router_type(&self) -> &'static str { "manager" } - - // ============================================================================ - // Conversation API Methods - delegate to selected router - // ============================================================================ - - async fn create_conversation(&self, headers: Option<&HeaderMap>, body: &Value) -> Response { - let router = self.select_router_for_request(headers, None); - if let Some(router) = router { - router.create_conversation(headers, body).await - } else { - ( - StatusCode::NOT_FOUND, - "No router available to create conversation", - ) - .into_response() - } - } - - async fn get_conversation( - &self, - headers: Option<&HeaderMap>, - conversation_id: &str, - ) -> Response { - let router = self.select_router_for_request(headers, None); - if let Some(router) = router { - router.get_conversation(headers, conversation_id).await - } else { - ( - StatusCode::NOT_FOUND, - format!( - "No router available to get conversation '{}'", - conversation_id - ), - ) - .into_response() - } - } - - async fn update_conversation( - &self, - headers: Option<&HeaderMap>, - conversation_id: &str, - body: &Value, - ) -> Response { - let router = self.select_router_for_request(headers, None); - if let Some(router) = router { - router - .update_conversation(headers, conversation_id, body) - .await - } else { - ( - StatusCode::NOT_FOUND, - format!( - "No router available to update conversation '{}'", - conversation_id - ), - ) - .into_response() - } - } - - async fn delete_conversation( - &self, - headers: Option<&HeaderMap>, - conversation_id: &str, - ) -> Response { - let router = self.select_router_for_request(headers, None); - if let Some(router) = router { - router.delete_conversation(headers, conversation_id).await - } else { - ( - StatusCode::NOT_FOUND, - format!( - "No router available to delete conversation '{}'", - conversation_id - ), - ) - .into_response() - } - } - - async fn list_conversation_items( - &self, - headers: Option<&HeaderMap>, - conversation_id: &str, - limit: Option, - order: Option<&str>, - after: Option<&str>, - ) -> Response { - let router = self.select_router_for_request(headers, None); - if let Some(router) = router { - router - .list_conversation_items(headers, conversation_id, limit, order, after) - .await - } else { - ( - StatusCode::NOT_FOUND, - format!( - "No router available to list items for conversation '{}'", - conversation_id - ), - ) - .into_response() - } - } - - async fn create_conversation_items( - &self, - headers: Option<&HeaderMap>, - conversation_id: &str, - body: &Value, - ) -> Response { - let router = self.select_router_for_request(headers, None); - if let Some(router) = router { - router - .create_conversation_items(headers, conversation_id, body) - .await - } else { - ( - StatusCode::NOT_FOUND, - format!( - "No router available to create items for conversation '{}'", - conversation_id - ), - ) - .into_response() - } - } - - async fn get_conversation_item( - &self, - headers: Option<&HeaderMap>, - conversation_id: &str, - item_id: &str, - include: Option>, - ) -> Response { - let router = self.select_router_for_request(headers, None); - if let Some(router) = router { - router - .get_conversation_item(headers, conversation_id, item_id, include) - .await - } else { - ( - StatusCode::NOT_FOUND, - format!( - "No router available to get item '{}' from conversation '{}'", - item_id, conversation_id - ), - ) - .into_response() - } - } - - async fn delete_conversation_item( - &self, - headers: Option<&HeaderMap>, - conversation_id: &str, - item_id: &str, - ) -> Response { - let router = self.select_router_for_request(headers, None); - if let Some(router) = router { - router - .delete_conversation_item(headers, conversation_id, item_id) - .await - } else { - ( - StatusCode::NOT_FOUND, - format!( - "No router available to delete item '{}' from conversation '{}'", - item_id, conversation_id - ), - ) - .into_response() - } - } } impl std::fmt::Debug for RouterManager { diff --git a/sgl-router/src/server.rs b/sgl-router/src/server.rs index bbf946aa3..88e32ce08 100644 --- a/sgl-router/src/server.rs +++ b/sgl-router/src/server.rs @@ -45,7 +45,7 @@ use crate::{ validated::ValidatedJson, worker_spec::{WorkerConfigRequest, WorkerErrorResponse, WorkerInfo}, }, - routers::{router_manager::RouterManager, RouterTrait}, + routers::{conversations, router_manager::RouterManager, RouterTrait}, service_discovery::{start_service_discovery, ServiceDiscoveryConfig}, }; @@ -270,47 +270,32 @@ async fn v1_responses_list_input_items( async fn v1_conversations_create( State(state): State>, - headers: http::HeaderMap, Json(body): Json, ) -> Response { - state - .router - .create_conversation(Some(&headers), &body) - .await + conversations::create_conversation(&state.context.conversation_storage, body).await } async fn v1_conversations_get( State(state): State>, Path(conversation_id): Path, - headers: http::HeaderMap, ) -> Response { - state - .router - .get_conversation(Some(&headers), &conversation_id) - .await + conversations::get_conversation(&state.context.conversation_storage, &conversation_id).await } async fn v1_conversations_update( State(state): State>, Path(conversation_id): Path, - headers: http::HeaderMap, Json(body): Json, ) -> Response { - state - .router - .update_conversation(Some(&headers), &conversation_id, &body) + conversations::update_conversation(&state.context.conversation_storage, &conversation_id, body) .await } async fn v1_conversations_delete( State(state): State>, Path(conversation_id): Path, - headers: http::HeaderMap, ) -> Response { - state - .router - .delete_conversation(Some(&headers), &conversation_id) - .await + conversations::delete_conversation(&state.context.conversation_storage, &conversation_id).await } #[derive(Deserialize, Default)] @@ -323,23 +308,21 @@ struct ListItemsQuery { async fn v1_conversations_list_items( State(state): State>, Path(conversation_id): Path, - headers: http::HeaderMap, Query(ListItemsQuery { limit, order, after, }): Query, ) -> Response { - state - .router - .list_conversation_items( - Some(&headers), - &conversation_id, - limit, - order.as_deref(), - after.as_deref(), - ) - .await + conversations::list_conversation_items( + &state.context.conversation_storage, + &state.context.conversation_item_storage, + &conversation_id, + limit, + order.as_deref(), + after.as_deref(), + ) + .await } #[derive(Deserialize, Default)] @@ -351,36 +334,43 @@ struct GetItemQuery { async fn v1_conversations_create_items( State(state): State>, Path(conversation_id): Path, - headers: http::HeaderMap, Json(body): Json, ) -> Response { - state - .router - .create_conversation_items(Some(&headers), &conversation_id, &body) - .await + conversations::create_conversation_items( + &state.context.conversation_storage, + &state.context.conversation_item_storage, + &conversation_id, + body, + ) + .await } async fn v1_conversations_get_item( State(state): State>, Path((conversation_id, item_id)): Path<(String, String)>, - headers: http::HeaderMap, Query(query): Query, ) -> Response { - state - .router - .get_conversation_item(Some(&headers), &conversation_id, &item_id, query.include) - .await + conversations::get_conversation_item( + &state.context.conversation_storage, + &state.context.conversation_item_storage, + &conversation_id, + &item_id, + query.include, + ) + .await } async fn v1_conversations_delete_item( State(state): State>, Path((conversation_id, item_id)): Path<(String, String)>, - headers: http::HeaderMap, ) -> Response { - state - .router - .delete_conversation_item(Some(&headers), &conversation_id, &item_id) - .await + conversations::delete_conversation_item( + &state.context.conversation_storage, + &state.context.conversation_item_storage, + &conversation_id, + &item_id, + ) + .await } async fn flush_cache(State(state): State>, _req: Request) -> Response {