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

[model-gateway] move conversation to first class routing (#14506)

Browse Source 
Co-authored-by: key4ng <rukeyang@gmail.com>
This commit is contained in:
Simo Lin
2025-12-05 08:44:13 -08:00
committed by GitHub
parent a89045603b
commit 05284378d6
4 changed files with 36 additions and 446 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
sgl-router/src/routers/mod.rs
View File

@@ -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<usize>,
_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<Vec<String>>,
) -> 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()
}
}

116
sgl-router/src/routers/openai/router.rs
View File

@@ -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<usize>,
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<Vec<String>>,
) -> 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
}
}

175
sgl-router/src/routers/router_manager.rs
View File

@@ -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<usize>,
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<Vec<String>>,
) -> 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 {

82
sgl-router/src/server.rs
View File

@@ -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<Arc<AppState>>,
headers: http::HeaderMap,
Json(body): Json<Value>,
) -> 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<Arc<AppState>>,
Path(conversation_id): Path<String>,
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<Arc<AppState>>,
Path(conversation_id): Path<String>,
headers: http::HeaderMap,
Json(body): Json<Value>,
) -> 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<Arc<AppState>>,
Path(conversation_id): Path<String>,
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<Arc<AppState>>,
Path(conversation_id): Path<String>,
headers: http::HeaderMap,
Query(ListItemsQuery {
limit,
order,
after,
}): Query<ListItemsQuery>,
) -> 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<Arc<AppState>>,
Path(conversation_id): Path<String>,
headers: http::HeaderMap,
Json(body): Json<Value>,
) -> 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<Arc<AppState>>,
Path((conversation_id, item_id)): Path<(String, String)>,
headers: http::HeaderMap,
Query(query): Query<GetItemQuery>,
) -> 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<Arc<AppState>>,
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<Arc<AppState>>, _req: Request) -> Response {