Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions proto/sglang/runtime/v1/sglang.proto
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,8 @@ message GenerateRequest {
optional bool require_reasoning = 15;
optional uint32 max_thinking_tokens = 16;
optional KVHintsEnvelope kv_hints = 17;
// Also return the detokenized text, as HTTP /generate does; off by default.
optional bool return_text = 18;
}

// Versioned KV-hint envelope forwarded from a trusted orchestrator.
Expand All @@ -156,6 +158,8 @@ message GenerateResponse {
repeated int32 output_ids = 1;
map<string, string> meta_info = 2;
bool finished = 3;
// Set only when the request asked for return_text.
optional string text = 4;
}

// ---- Text-based embed (text in, embedding out) ----
Expand Down Expand Up @@ -336,6 +340,8 @@ message OpenAIRequest {
message OpenAIStreamChunk {
bytes json_chunk = 1;
bool finished = 2;
// HTTP status of a non-streaming response, set on its single finished chunk.
optional int32 status_code = 3;
}

message OpenAIResponse {
Expand Down
4 changes: 3 additions & 1 deletion python/sglang/srt/entrypoints/grpc_bridge.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from types import SimpleNamespace
from typing import Any, Awaitable, Callable, Dict, List, Optional

from fastapi.encoders import jsonable_encoder
from pydantic import ValidationError

from sglang.srt.arg_groups.overrides import resolving_view
Expand Down Expand Up @@ -795,7 +796,8 @@ async def _flush_event() -> bool:
elif hasattr(result, "body"):
resp_bytes = result.body
elif isinstance(result, (dict, list)):
resp_bytes = json.dumps(result).encode("utf-8")
# As FastAPI encodes it over HTTP; rerank returns a list of models.
resp_bytes = json.dumps(jsonable_encoder(result)).encode("utf-8")
else:
resp_bytes = str(result).encode("utf-8")
status_code = int(
Expand Down
9 changes: 9 additions & 0 deletions rust/sglang-api-types/src/generated/sglang.runtime.v1.rs
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,9 @@ pub struct GenerateRequest {
pub max_thinking_tokens: ::core::option::Option<u32>,
#[prost(message, optional, tag = "17")]
pub kv_hints: ::core::option::Option<KvHintsEnvelope>,
/// Also return the detokenized text, as HTTP /generate does; off by default.
#[prost(bool, optional, tag = "18")]
pub return_text: ::core::option::Option<bool>,
}
/// Versioned KV-hint envelope forwarded from a trusted orchestrator.
#[derive(Clone, PartialEq, ::prost::Message)]
Expand Down Expand Up @@ -201,6 +204,9 @@ pub struct GenerateResponse {
::std::collections::HashMap<::prost::alloc::string::String, ::prost::alloc::string::String>,
#[prost(bool, tag = "3")]
pub finished: bool,
/// Set only when the request asked for return_text.
#[prost(string, optional, tag = "4")]
pub text: ::core::option::Option<::prost::alloc::string::String>,
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct TextEmbedRequest {
Expand Down Expand Up @@ -416,6 +422,9 @@ pub struct OpenAiStreamChunk {
pub json_chunk: ::prost::alloc::vec::Vec<u8>,
#[prost(bool, tag = "2")]
pub finished: bool,
/// HTTP status of a non-streaming response, set on its single finished chunk.
#[prost(int32, optional, tag = "3")]
pub status_code: ::core::option::Option<i32>,
}
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct OpenAiResponse {
Expand Down
15 changes: 4 additions & 11 deletions rust/sglang-grpc/src/bridge.rs
Original file line number Diff line number Diff line change
Expand Up @@ -234,7 +234,7 @@ impl PyBridge {
));
}

let should_call_python = if abort_all {
if abort_all {
let mut state = lock_or_recover(self.state.as_ref(), "state");
let rids = state
.channels
Expand All @@ -252,24 +252,17 @@ impl PyBridge {
);
}
tracing::debug!(affected, "gRPC abort_all cleared active response channels");
true
} else {
let mut state = lock_or_recover(self.state.as_ref(), "state");
let was_active = remove_channel_refs_locked(&mut state, rid);
if was_active {
if remove_channel_refs_locked(&mut state, rid) {
state
.terminal_errors
.insert(rid.to_string(), TerminalError::Aborted { rid: rid.into() });
} else {
tracing::debug!(rid, "Ignoring abort for inactive gRPC request id");
}
was_active
};

if !should_call_python {
return Ok(());
}

// Abort by engine rid even when no gRPC stream here carries it, as HTTP
// /abort_request does: an OpenAI RPC's request runs under its body's rid.
Python::attach(|py| {
self.runtime_handle
.call_method1(py, "abort", (rid, abort_all))?;
Expand Down
8 changes: 8 additions & 0 deletions rust/sglang-grpc/src/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -396,6 +396,7 @@ impl proto::sglang_service_server::SglangService for SglangServiceImpl {
.clone()
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
let req_dict = build_generate_dict(&rid, &req).map_err(Status::invalid_argument)?;
let return_text = req.return_text.unwrap_or(false);

let mut receiver = self
.bridge
Expand All @@ -415,6 +416,7 @@ impl proto::sglang_service_server::SglangService for SglangServiceImpl {
output_ids: data.output_ids.unwrap_or_default(),
meta_info: data.meta_info,
finished: false,
text: data.text.filter(|_| return_text),
});
}
Ok(Some(ResponseChunk::Finished(data))) => {
Expand All @@ -423,6 +425,7 @@ impl proto::sglang_service_server::SglangService for SglangServiceImpl {
output_ids: data.output_ids.unwrap_or_default(),
meta_info: data.meta_info,
finished: true,
text: data.text.filter(|_| return_text),
});
break;
}
Expand Down Expand Up @@ -991,6 +994,7 @@ impl SglangServiceImpl {
yield Ok(proto::OpenAiStreamChunk {
json_chunk: data.json_bytes.unwrap_or_default(),
finished: false,
status_code: None,
});
}
Ok(Some(ResponseChunk::Finished(data))) => {
Expand All @@ -999,6 +1003,10 @@ impl SglangServiceImpl {
yield Ok(proto::OpenAiStreamChunk {
json_chunk: bytes,
finished: true,
status_code: data
.meta_info
.contains_key("status_code")
.then(|| openai_status_code(&data.meta_info, 200)),
});
break;
}
Expand Down
24 changes: 24 additions & 0 deletions test/registered/unit/entrypoints/test_grpc_bridge.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import asyncio
import enum
import json
import unittest
from types import SimpleNamespace

Expand Down Expand Up @@ -112,6 +113,29 @@ def test_streaming_first_finished_choice_is_not_batch_terminal(self):
self.assertEqual([call[1] for call in callback.calls], [False, False, True])


class TestNativeGrpcOpenAIResponses(CustomTestCase):
def test_rerank_list_of_models_is_serialized_as_over_http(self):
from sglang.srt.entrypoints.openai.protocol import RerankResponse

class _Rerank:
async def handle_request(self, request, raw_request):
return [RerankResponse(score=0.5, document="d", index=0)]

calls = []
handle = RuntimeHandle.__new__(RuntimeHandle)
handle._get_openai_serving = lambda: {"rerank": _Rerank()}
body = b'{"query": "q", "documents": ["d"]}'
callback = lambda payload, **kwargs: calls.append((payload, kwargs))

asyncio.run(
handle._run_openai_request("rerank", body, callback, streaming=False)
)

payload, kwargs = calls[-1]
self.assertEqual(kwargs["status_code"], 200)
self.assertEqual(json.loads(payload)[0]["score"], 0.5)


class TestEngineStateNotifications(CustomTestCase):
def setUp(self):
self.manager = TokenizerManager.__new__(TokenizerManager)
Expand Down
Loading