feat: streaming (SSE) end-to-end
- Provider::chat_stream; OpenAI provider streams via stream_options.include_usage - pipeline.run_stream: shared route/select, failover before first byte - /v1/chat/completions forwards SSE verbatim with X-Cortiq-* headers; usage tapped for stats - admin /test/stream + live streaming in Playground (stream toggle, 7 langs)
This commit is contained in:
@@ -3,6 +3,16 @@
|
|||||||
All notable changes to this project are documented here.
|
All notable changes to this project are documented here.
|
||||||
The format is based on [Keep a Changelog](https://keepachangelog.com/).
|
The format is based on [Keep a Changelog](https://keepachangelog.com/).
|
||||||
|
|
||||||
|
## [Unreleased]
|
||||||
|
|
||||||
|
### Added
|
||||||
|
- **Streaming (SSE)** end-to-end: `POST /v1/chat/completions` with `"stream": true`
|
||||||
|
forwards provider SSE chunks verbatim, with `X-Cortiq-*` routing headers and token
|
||||||
|
usage tapped from the final chunk for statistics. Failover applies before the first
|
||||||
|
byte. Live streaming output in the admin Playground (`/admin/api/test/stream`).
|
||||||
|
- Cross-platform Windows binaries in releases (`x86_64-pc-windows-msvc`) and a
|
||||||
|
cross-platform CI test matrix (ubuntu/windows/macos); secure token RNG via `getrandom`.
|
||||||
|
|
||||||
## [0.1.0] - 2026-06-30
|
## [0.1.0] - 2026-06-30
|
||||||
|
|
||||||
### Added
|
### Added
|
||||||
|
|||||||
Generated
+23
@@ -76,6 +76,28 @@ dependencies = [
|
|||||||
"rustversion",
|
"rustversion",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "async-stream"
|
||||||
|
version = "0.3.6"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476"
|
||||||
|
dependencies = [
|
||||||
|
"async-stream-impl",
|
||||||
|
"futures-core",
|
||||||
|
"pin-project-lite",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "async-stream-impl"
|
||||||
|
version = "0.3.6"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d"
|
||||||
|
dependencies = [
|
||||||
|
"proc-macro2",
|
||||||
|
"quote",
|
||||||
|
"syn",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "async-trait"
|
name = "async-trait"
|
||||||
version = "0.1.89"
|
version = "0.1.89"
|
||||||
@@ -267,6 +289,7 @@ version = "0.1.0"
|
|||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"arc-swap",
|
"arc-swap",
|
||||||
|
"async-stream",
|
||||||
"async-trait",
|
"async-trait",
|
||||||
"axum",
|
"axum",
|
||||||
"bytes",
|
"bytes",
|
||||||
|
|||||||
@@ -48,6 +48,7 @@ getrandom = "0.2"
|
|||||||
# streaming / async utils
|
# streaming / async utils
|
||||||
futures = "0.3"
|
futures = "0.3"
|
||||||
async-trait = "0.1"
|
async-trait = "0.1"
|
||||||
|
async-stream = "0.3"
|
||||||
bytes = "1"
|
bytes = "1"
|
||||||
tokio-stream = "0.1"
|
tokio-stream = "0.1"
|
||||||
|
|
||||||
|
|||||||
@@ -165,6 +165,7 @@ pub fn api_routes(admin_token: String) -> Router<SharedState> {
|
|||||||
.route("/admin/api/stats", get(get_stats))
|
.route("/admin/api/stats", get(get_stats))
|
||||||
.route("/admin/api/requests", get(get_requests))
|
.route("/admin/api/requests", get(get_requests))
|
||||||
.route("/admin/api/test", post(run_test))
|
.route("/admin/api/test", post(run_test))
|
||||||
|
.route("/admin/api/test/stream", post(run_test_stream))
|
||||||
.route_layer(middleware::from_fn(move |req: Request, next: Next| {
|
.route_layer(middleware::from_fn(move |req: Request, next: Next| {
|
||||||
let expected = admin_token.clone();
|
let expected = admin_token.clone();
|
||||||
async move { auth(expected, req, next).await }
|
async move { auth(expected, req, next).await }
|
||||||
@@ -753,3 +754,38 @@ async fn run_test(State(state): State<SharedState>, Json(b): Json<TestBody>) ->
|
|||||||
Err(e) => ok(json!({ "ok": false, "latency_ms": latency_ms, "error": e.to_string() })),
|
Err(e) => ok(json!({ "ok": false, "latency_ms": latency_ms, "error": e.to_string() })),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Streaming playground: forwards provider SSE chunks; routing decision is exposed
|
||||||
|
/// via X-Cortiq-* response headers (the SPA reads them after the stream ends).
|
||||||
|
async fn run_test_stream(State(state): State<SharedState>, Json(b): Json<TestBody>) -> Response {
|
||||||
|
if b.messages.is_empty() {
|
||||||
|
return ApiError::bad("messages must not be empty").into_response();
|
||||||
|
}
|
||||||
|
let req = ChatRequest {
|
||||||
|
routing: parse_routing(&b.model),
|
||||||
|
messages: b.messages,
|
||||||
|
tools: vec![],
|
||||||
|
params: GenParams {
|
||||||
|
temperature: b.temperature,
|
||||||
|
max_tokens: b.max_tokens,
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
stream: true,
|
||||||
|
meta: RequestMeta {
|
||||||
|
account: "playground".into(),
|
||||||
|
protocol: "playground".into(),
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
};
|
||||||
|
match state.pipeline.run_stream(req, &state).await {
|
||||||
|
Ok((info, stream)) => {
|
||||||
|
let mut headers = crate::protocols::openai_chat::cortiq_headers(&info);
|
||||||
|
headers.insert(
|
||||||
|
header::CONTENT_TYPE,
|
||||||
|
axum::http::HeaderValue::from_static("text/event-stream"),
|
||||||
|
);
|
||||||
|
(headers, axum::body::Body::from_stream(stream)).into_response()
|
||||||
|
}
|
||||||
|
Err(e) => ApiError::bad(e.to_string()).into_response(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+224
-3
@@ -79,9 +79,13 @@ impl Pipeline {
|
|||||||
result
|
result
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn run_inner(&self, req: ChatRequest, state: &SharedState) -> Result<ChatResponse> {
|
/// Stages 1–3: extract text → route → select & capability-filter candidates.
|
||||||
let live = state.live();
|
/// Shared by both the non-streaming and streaming paths.
|
||||||
|
async fn resolve(
|
||||||
|
&self,
|
||||||
|
req: &ChatRequest,
|
||||||
|
live: &crate::state::Live,
|
||||||
|
) -> Result<(RouteDecision, String, Vec<String>)> {
|
||||||
// 1. Extract text for classification
|
// 1. Extract text for classification
|
||||||
let text = self.extract_text(
|
let text = self.extract_text(
|
||||||
&req.messages,
|
&req.messages,
|
||||||
@@ -159,6 +163,12 @@ impl Pipeline {
|
|||||||
.to_string(),
|
.to_string(),
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
Ok((decision, route_source, final_candidates))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn run_inner(&self, req: ChatRequest, state: &SharedState) -> Result<ChatResponse> {
|
||||||
|
let live = state.live();
|
||||||
|
let (decision, route_source, final_candidates) = self.resolve(&req, &live).await?;
|
||||||
|
|
||||||
// 4. Attempt provider call with failover (fall back to alternatives on error)
|
// 4. Attempt provider call with failover (fall back to alternatives on error)
|
||||||
let mut last_err = None;
|
let mut last_err = None;
|
||||||
@@ -210,6 +220,89 @@ impl Pipeline {
|
|||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Streaming variant: returns the routing metadata plus a stream of SSE byte
|
||||||
|
/// chunks (OpenAI wire format). Failover applies only before the first byte —
|
||||||
|
/// once the upstream stream has started it cannot be retried. Statistics are
|
||||||
|
/// recorded when the stream completes (tokens taken from the final usage chunk).
|
||||||
|
pub async fn run_stream(
|
||||||
|
&self,
|
||||||
|
req: ChatRequest,
|
||||||
|
state: &SharedState,
|
||||||
|
) -> Result<(RouteInfo, crate::providers::ChatStream)> {
|
||||||
|
let started = Instant::now();
|
||||||
|
let account = req.meta.account.clone();
|
||||||
|
let protocol = req.meta.protocol.clone();
|
||||||
|
let directive = match &req.routing {
|
||||||
|
RoutingDirective::Pinned { .. } => "pinned",
|
||||||
|
RoutingDirective::Auto { .. } => "auto",
|
||||||
|
}
|
||||||
|
.to_string();
|
||||||
|
|
||||||
|
let live = state.live();
|
||||||
|
let (decision, route_source, final_candidates) = match self.resolve(&req, &live).await {
|
||||||
|
Ok(t) => t,
|
||||||
|
Err(e) => {
|
||||||
|
record_error(state, &account, &protocol, &directive, started, &e);
|
||||||
|
return Err(e);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let mut last_err = None;
|
||||||
|
let mut attempt = 0usize;
|
||||||
|
for model_id in final_candidates {
|
||||||
|
let provider = match live.registry.get(&model_id) {
|
||||||
|
Some(p) => p,
|
||||||
|
None => continue,
|
||||||
|
};
|
||||||
|
tracing::info!(
|
||||||
|
"Streaming request (label: {}, tier: {}) to model '{}' via source '{}'",
|
||||||
|
decision.task_label,
|
||||||
|
decision.complexity_tier,
|
||||||
|
model_id,
|
||||||
|
route_source
|
||||||
|
);
|
||||||
|
match provider.chat_stream(req.clone()).await {
|
||||||
|
Ok(stream) => {
|
||||||
|
let info = RouteInfo {
|
||||||
|
task_label: decision.task_label.clone(),
|
||||||
|
complexity_score: decision.complexity_score,
|
||||||
|
complexity_tier: decision.complexity_tier.clone(),
|
||||||
|
selected_model: model_id,
|
||||||
|
route_source: route_source.clone(),
|
||||||
|
router_request_id: decision.router_request_id.clone(),
|
||||||
|
cost_usd: 0.0,
|
||||||
|
failover: attempt > 0,
|
||||||
|
};
|
||||||
|
let tapped = tap_stream(
|
||||||
|
stream,
|
||||||
|
state.stats.clone(),
|
||||||
|
provider.clone(),
|
||||||
|
info.clone(),
|
||||||
|
account,
|
||||||
|
protocol,
|
||||||
|
directive,
|
||||||
|
started,
|
||||||
|
);
|
||||||
|
return Ok((info, Box::pin(tapped)));
|
||||||
|
}
|
||||||
|
Err(err) => {
|
||||||
|
tracing::warn!(
|
||||||
|
"Provider '{}' stream failed: {:?}. Attempting failover...",
|
||||||
|
model_id,
|
||||||
|
err
|
||||||
|
);
|
||||||
|
last_err = Some(err);
|
||||||
|
attempt += 1;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let e = last_err.unwrap_or_else(|| {
|
||||||
|
GatewayError::UpstreamUnavailable("All candidate models failed".to_string())
|
||||||
|
});
|
||||||
|
record_error(state, &account, &protocol, &directive, started, &e);
|
||||||
|
Err(e)
|
||||||
|
}
|
||||||
|
|
||||||
fn extract_text(&self, messages: &[Message], strategy: &str, max_chars: usize) -> String {
|
fn extract_text(&self, messages: &[Message], strategy: &str, max_chars: usize) -> String {
|
||||||
let raw_text = match strategy {
|
let raw_text = match strategy {
|
||||||
"last_user" => messages
|
"last_user" => messages
|
||||||
@@ -249,3 +342,131 @@ impl Pipeline {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn now_secs() -> u64 {
|
||||||
|
SystemTime::now()
|
||||||
|
.duration_since(UNIX_EPOCH)
|
||||||
|
.unwrap_or_default()
|
||||||
|
.as_secs()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn record_error(
|
||||||
|
state: &SharedState,
|
||||||
|
account: &str,
|
||||||
|
protocol: &str,
|
||||||
|
directive: &str,
|
||||||
|
started: Instant,
|
||||||
|
e: &GatewayError,
|
||||||
|
) {
|
||||||
|
state.stats.record(RequestRecord {
|
||||||
|
ts: now_secs(),
|
||||||
|
account: account.to_string(),
|
||||||
|
protocol: protocol.to_string(),
|
||||||
|
directive: directive.to_string(),
|
||||||
|
task_label: String::new(),
|
||||||
|
tier: String::new(),
|
||||||
|
score: 0.0,
|
||||||
|
model_id: String::new(),
|
||||||
|
route_source: String::new(),
|
||||||
|
prompt_tokens: 0,
|
||||||
|
completion_tokens: 0,
|
||||||
|
cost_usd: 0.0,
|
||||||
|
latency_ms: started.elapsed().as_millis() as u64,
|
||||||
|
outcome: "error".to_string(),
|
||||||
|
failover: false,
|
||||||
|
error: Some(e.to_string()),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Forward a provider SSE stream to the client verbatim while tapping the final
|
||||||
|
/// `usage` chunk; record statistics when the stream completes.
|
||||||
|
#[allow(clippy::too_many_arguments)]
|
||||||
|
fn tap_stream(
|
||||||
|
inner: crate::providers::ChatStream,
|
||||||
|
stats: std::sync::Arc<crate::stats::Stats>,
|
||||||
|
provider: std::sync::Arc<dyn crate::providers::Provider>,
|
||||||
|
info: RouteInfo,
|
||||||
|
account: String,
|
||||||
|
protocol: String,
|
||||||
|
directive: String,
|
||||||
|
started: Instant,
|
||||||
|
) -> impl futures::Stream<Item = Result<bytes::Bytes>> {
|
||||||
|
use futures::StreamExt;
|
||||||
|
async_stream::stream! {
|
||||||
|
let mut inner = inner;
|
||||||
|
let mut buf = String::new();
|
||||||
|
let mut prompt_tokens = 0u32;
|
||||||
|
let mut completion_tokens = 0u32;
|
||||||
|
let mut got_usage = false;
|
||||||
|
let mut delta_chars = 0usize;
|
||||||
|
let mut errored: Option<String> = None;
|
||||||
|
|
||||||
|
while let Some(item) = inner.next().await {
|
||||||
|
match &item {
|
||||||
|
Ok(bytes) => {
|
||||||
|
if let Ok(s) = std::str::from_utf8(bytes) {
|
||||||
|
buf.push_str(s);
|
||||||
|
while let Some(idx) = buf.find("\n\n") {
|
||||||
|
let evt: String = buf.drain(..idx + 2).collect();
|
||||||
|
for line in evt.lines() {
|
||||||
|
if let Some(data) = line.trim_start().strip_prefix("data:") {
|
||||||
|
let data = data.trim();
|
||||||
|
if data.is_empty() || data == "[DONE]" {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if let Ok(v) =
|
||||||
|
serde_json::from_str::<serde_json::Value>(data)
|
||||||
|
{
|
||||||
|
let u = &v["usage"];
|
||||||
|
if !u.is_null() {
|
||||||
|
prompt_tokens = u["prompt_tokens"]
|
||||||
|
.as_u64()
|
||||||
|
.unwrap_or(prompt_tokens as u64)
|
||||||
|
as u32;
|
||||||
|
completion_tokens = u["completion_tokens"]
|
||||||
|
.as_u64()
|
||||||
|
.unwrap_or(completion_tokens as u64)
|
||||||
|
as u32;
|
||||||
|
got_usage = true;
|
||||||
|
}
|
||||||
|
if let Some(c) =
|
||||||
|
v["choices"][0]["delta"]["content"].as_str()
|
||||||
|
{
|
||||||
|
delta_chars += c.chars().count();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(e) => errored = Some(e.to_string()),
|
||||||
|
}
|
||||||
|
yield item;
|
||||||
|
}
|
||||||
|
|
||||||
|
if !got_usage && completion_tokens == 0 {
|
||||||
|
// estimate when the server didn't return usage (~4 chars per token)
|
||||||
|
completion_tokens = (delta_chars / 4) as u32;
|
||||||
|
}
|
||||||
|
let cost = provider.price(prompt_tokens, completion_tokens);
|
||||||
|
stats.record(RequestRecord {
|
||||||
|
ts: now_secs(),
|
||||||
|
account,
|
||||||
|
protocol,
|
||||||
|
directive,
|
||||||
|
task_label: info.task_label.clone(),
|
||||||
|
tier: info.complexity_tier.clone(),
|
||||||
|
score: info.complexity_score,
|
||||||
|
model_id: info.selected_model.clone(),
|
||||||
|
route_source: info.route_source.clone(),
|
||||||
|
prompt_tokens,
|
||||||
|
completion_tokens,
|
||||||
|
cost_usd: cost,
|
||||||
|
latency_ms: started.elapsed().as_millis() as u64,
|
||||||
|
outcome: if errored.is_some() { "error" } else { "ok" }.to_string(),
|
||||||
|
failover: info.failover,
|
||||||
|
error: errored,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -3,11 +3,37 @@
|
|||||||
//! to the pipeline, then converts the canonical [`ChatResponse`] back to OpenAI format.
|
//! to the pipeline, then converts the canonical [`ChatResponse`] back to OpenAI format.
|
||||||
|
|
||||||
use crate::error::{GatewayError, Result};
|
use crate::error::{GatewayError, Result};
|
||||||
use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective};
|
use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RouteInfo, RoutingDirective};
|
||||||
use crate::state::SharedState;
|
use crate::state::SharedState;
|
||||||
|
use axum::response::{IntoResponse, Response};
|
||||||
use axum::{extract::State, routing::post, Json, Router};
|
use axum::{extract::State, routing::post, Json, Router};
|
||||||
use serde::Deserialize;
|
use serde::Deserialize;
|
||||||
|
|
||||||
|
/// Build the `X-Cortiq-*` response headers from routing metadata.
|
||||||
|
pub(crate) fn cortiq_headers(c: &RouteInfo) -> axum::http::HeaderMap {
|
||||||
|
use axum::http::{HeaderMap, HeaderValue};
|
||||||
|
fn put(h: &mut HeaderMap, k: &'static str, v: &str) {
|
||||||
|
if let Ok(val) = HeaderValue::from_str(v) {
|
||||||
|
h.insert(k, val);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let mut h = HeaderMap::new();
|
||||||
|
put(&mut h, "X-Cortiq-Task-Label", &c.task_label);
|
||||||
|
put(
|
||||||
|
&mut h,
|
||||||
|
"X-Cortiq-Complexity-Score",
|
||||||
|
&c.complexity_score.to_string(),
|
||||||
|
);
|
||||||
|
put(&mut h, "X-Cortiq-Complexity-Tier", &c.complexity_tier);
|
||||||
|
put(&mut h, "X-Cortiq-Selected-Model", &c.selected_model);
|
||||||
|
put(&mut h, "X-Cortiq-Route-Source", &c.route_source);
|
||||||
|
put(&mut h, "X-Cortiq-Cost-Usd", &c.cost_usd.to_string());
|
||||||
|
if let Some(id) = &c.router_request_id {
|
||||||
|
put(&mut h, "X-Cortiq-Request-Id", id);
|
||||||
|
}
|
||||||
|
h
|
||||||
|
}
|
||||||
|
|
||||||
pub fn routes() -> Router<SharedState> {
|
pub fn routes() -> Router<SharedState> {
|
||||||
Router::new().route("/v1/chat/completions", post(handler))
|
Router::new().route("/v1/chat/completions", post(handler))
|
||||||
}
|
}
|
||||||
@@ -46,7 +72,7 @@ fn parse_routing(model: &str) -> RoutingDirective {
|
|||||||
async fn handler(
|
async fn handler(
|
||||||
State(state): State<SharedState>,
|
State(state): State<SharedState>,
|
||||||
Json(req): Json<OpenAiChatRequest>,
|
Json(req): Json<OpenAiChatRequest>,
|
||||||
) -> Result<impl axum::response::IntoResponse> {
|
) -> Result<Response> {
|
||||||
// hot protocol toggle: if the adapter is disabled in config — return 404
|
// hot protocol toggle: if the adapter is disabled in config — return 404
|
||||||
if !state.live().cfg.protocols.openai_chat {
|
if !state.live().cfg.protocols.openai_chat {
|
||||||
return Err(GatewayError::InvalidRequest(
|
return Err(GatewayError::InvalidRequest(
|
||||||
@@ -77,52 +103,23 @@ async fn handler(
|
|||||||
},
|
},
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// streaming: forward provider SSE chunks verbatim with X-Cortiq-* headers
|
||||||
if req.stream {
|
if req.stream {
|
||||||
return Err(GatewayError::InvalidRequest(
|
let (info, stream) = state.pipeline.run_stream(canonical, &state).await?;
|
||||||
"Streaming is not yet supported in this version".into(),
|
let mut headers = cortiq_headers(&info);
|
||||||
));
|
headers.insert(
|
||||||
|
axum::http::header::CONTENT_TYPE,
|
||||||
|
axum::http::HeaderValue::from_static("text/event-stream"),
|
||||||
|
);
|
||||||
|
headers.insert(
|
||||||
|
axum::http::header::CACHE_CONTROL,
|
||||||
|
axum::http::HeaderValue::from_static("no-cache"),
|
||||||
|
);
|
||||||
|
return Ok((headers, axum::body::Body::from_stream(stream)).into_response());
|
||||||
}
|
}
|
||||||
|
|
||||||
let resp = state.pipeline.run(canonical, &state).await?;
|
let resp = state.pipeline.run(canonical, &state).await?;
|
||||||
|
let headers = cortiq_headers(&resp.cortiq);
|
||||||
let mut headers = axum::http::HeaderMap::new();
|
|
||||||
headers.insert(
|
|
||||||
"X-Cortiq-Task-Label",
|
|
||||||
axum::http::HeaderValue::from_str(&resp.cortiq.task_label)
|
|
||||||
.unwrap_or(axum::http::HeaderValue::from_static("")),
|
|
||||||
);
|
|
||||||
headers.insert(
|
|
||||||
"X-Cortiq-Complexity-Score",
|
|
||||||
axum::http::HeaderValue::from_str(&resp.cortiq.complexity_score.to_string())
|
|
||||||
.unwrap_or(axum::http::HeaderValue::from_static("0.0")),
|
|
||||||
);
|
|
||||||
headers.insert(
|
|
||||||
"X-Cortiq-Complexity-Tier",
|
|
||||||
axum::http::HeaderValue::from_str(&resp.cortiq.complexity_tier)
|
|
||||||
.unwrap_or(axum::http::HeaderValue::from_static("")),
|
|
||||||
);
|
|
||||||
headers.insert(
|
|
||||||
"X-Cortiq-Selected-Model",
|
|
||||||
axum::http::HeaderValue::from_str(&resp.cortiq.selected_model)
|
|
||||||
.unwrap_or(axum::http::HeaderValue::from_static("")),
|
|
||||||
);
|
|
||||||
headers.insert(
|
|
||||||
"X-Cortiq-Route-Source",
|
|
||||||
axum::http::HeaderValue::from_str(&resp.cortiq.route_source)
|
|
||||||
.unwrap_or(axum::http::HeaderValue::from_static("")),
|
|
||||||
);
|
|
||||||
headers.insert(
|
|
||||||
"X-Cortiq-Cost-Usd",
|
|
||||||
axum::http::HeaderValue::from_str(&resp.cortiq.cost_usd.to_string())
|
|
||||||
.unwrap_or(axum::http::HeaderValue::from_static("0.0")),
|
|
||||||
);
|
|
||||||
if let Some(req_id) = &resp.cortiq.router_request_id {
|
|
||||||
headers.insert(
|
|
||||||
"X-Cortiq-Request-Id",
|
|
||||||
axum::http::HeaderValue::from_str(req_id)
|
|
||||||
.unwrap_or(axum::http::HeaderValue::from_static("")),
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
let mut body = serde_json::json!({
|
let mut body = serde_json::json!({
|
||||||
"id": resp.id,
|
"id": resp.id,
|
||||||
@@ -164,5 +161,5 @@ async fn handler(
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
Ok((headers, Json(body)))
|
Ok((headers, Json(body)).into_response())
|
||||||
}
|
}
|
||||||
|
|||||||
+17
-2
@@ -7,8 +7,16 @@
|
|||||||
pub mod anthropic;
|
pub mod anthropic;
|
||||||
pub mod openai;
|
pub mod openai;
|
||||||
|
|
||||||
|
use crate::error::GatewayError;
|
||||||
use crate::model::{ChatRequest, ChatResponse};
|
use crate::model::{ChatRequest, ChatResponse};
|
||||||
use async_trait::async_trait;
|
use async_trait::async_trait;
|
||||||
|
use bytes::Bytes;
|
||||||
|
use futures::stream::BoxStream;
|
||||||
|
|
||||||
|
/// A streaming chat response: raw Server-Sent-Events byte chunks in OpenAI wire
|
||||||
|
/// format (`data: {json}\n\n` … `data: [DONE]\n\n`). The gateway forwards these
|
||||||
|
/// to the client verbatim while tapping the final `usage` for statistics.
|
||||||
|
pub type ChatStream = BoxStream<'static, crate::error::Result<Bytes>>;
|
||||||
|
|
||||||
/// Model capabilities — checked at selection time (whether tools/vision are needed).
|
/// Model capabilities — checked at selection time (whether tools/vision are needed).
|
||||||
#[derive(Clone, Debug, Default)]
|
#[derive(Clone, Debug, Default)]
|
||||||
@@ -41,6 +49,13 @@ pub trait Provider: Send + Sync {
|
|||||||
/// Non-streaming chat call.
|
/// Non-streaming chat call.
|
||||||
async fn chat(&self, req: ChatRequest) -> crate::error::Result<ChatResponse>;
|
async fn chat(&self, req: ChatRequest) -> crate::error::Result<ChatResponse>;
|
||||||
|
|
||||||
// TODO(v0.1): async fn chat_stream(...) -> Result<ChatStream> (SSE).
|
/// Streaming chat call (SSE). Default: not supported (override per provider).
|
||||||
// TODO(v0.1): async fn embed(...) -> Result<EmbedResponse>.
|
async fn chat_stream(&self, _req: ChatRequest) -> crate::error::Result<ChatStream> {
|
||||||
|
Err(GatewayError::UpstreamUnavailable(format!(
|
||||||
|
"streaming is not supported by provider '{}'",
|
||||||
|
self.id()
|
||||||
|
)))
|
||||||
|
}
|
||||||
|
|
||||||
|
// TODO(v0.2): async fn embed(...) -> Result<EmbedResponse>.
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -154,4 +154,35 @@ impl Provider for OpenAiProvider {
|
|||||||
},
|
},
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn chat_stream(&self, req: ChatRequest) -> Result<super::ChatStream> {
|
||||||
|
use futures::StreamExt;
|
||||||
|
let mut body = self.to_wire(&req);
|
||||||
|
body["stream"] = serde_json::Value::Bool(true);
|
||||||
|
// ask OpenAI-compatible servers to include token usage in the final chunk
|
||||||
|
body["stream_options"] = serde_json::json!({ "include_usage": true });
|
||||||
|
let mut r = self
|
||||||
|
.http
|
||||||
|
.post(format!("{}/chat/completions", self.base_url))
|
||||||
|
.json(&body);
|
||||||
|
if let Some(key) = &self.api_key {
|
||||||
|
r = r.bearer_auth(key);
|
||||||
|
}
|
||||||
|
let resp = r
|
||||||
|
.send()
|
||||||
|
.await
|
||||||
|
.map_err(|e| GatewayError::UpstreamUnavailable(e.to_string()))?;
|
||||||
|
if !resp.status().is_success() {
|
||||||
|
let status = resp.status();
|
||||||
|
let err_body = resp.text().await.unwrap_or_default();
|
||||||
|
return Err(GatewayError::UpstreamUnavailable(format!(
|
||||||
|
"provider {} returned HTTP {}: {}",
|
||||||
|
self.id, status, err_body
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
let stream = resp
|
||||||
|
.bytes_stream()
|
||||||
|
.map(|res| res.map_err(|e| GatewayError::UpstreamUnavailable(e.to_string())));
|
||||||
|
Ok(Box::pin(stream))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Vendored
+48
@@ -91,3 +91,51 @@ export const api = {
|
|||||||
// playground
|
// playground
|
||||||
test: (b) => j("POST", "/test", b),
|
test: (b) => j("POST", "/test", b),
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// Streaming playground: POST /test/stream, calls onDelta(text) per content chunk,
|
||||||
|
// returns the routing decision read from X-Cortiq-* response headers.
|
||||||
|
export async function testStream(body, onDelta) {
|
||||||
|
const r = await fetch(BASE + "/test/stream", {
|
||||||
|
method: "POST",
|
||||||
|
headers: { Authorization: "Bearer " + getToken(), "content-type": "application/json" },
|
||||||
|
body: JSON.stringify(body),
|
||||||
|
});
|
||||||
|
if (r.status === 401) {
|
||||||
|
if (onUnauthorized) onUnauthorized();
|
||||||
|
throw new Error("unauthorized");
|
||||||
|
}
|
||||||
|
if (!r.ok || !r.body) {
|
||||||
|
throw new Error((await r.text()) || r.statusText);
|
||||||
|
}
|
||||||
|
const reader = r.body.getReader();
|
||||||
|
const dec = new TextDecoder();
|
||||||
|
let buf = "";
|
||||||
|
for (;;) {
|
||||||
|
const { done, value } = await reader.read();
|
||||||
|
if (done) break;
|
||||||
|
buf += dec.decode(value, { stream: true });
|
||||||
|
let idx;
|
||||||
|
while ((idx = buf.indexOf("\n\n")) >= 0) {
|
||||||
|
const line = buf.slice(0, idx).split("\n").find((l) => l.startsWith("data:"));
|
||||||
|
buf = buf.slice(idx + 2);
|
||||||
|
if (!line) continue;
|
||||||
|
const data = line.slice(5).trim();
|
||||||
|
if (!data || data === "[DONE]") continue;
|
||||||
|
try {
|
||||||
|
const d = JSON.parse(data).choices?.[0]?.delta?.content;
|
||||||
|
if (d) onDelta(d);
|
||||||
|
} catch {
|
||||||
|
/* ignore malformed chunk */
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
const h = r.headers;
|
||||||
|
return {
|
||||||
|
task_label: h.get("x-cortiq-task-label") || "",
|
||||||
|
tier: h.get("x-cortiq-complexity-tier") || "",
|
||||||
|
score: parseFloat(h.get("x-cortiq-complexity-score") || "0"),
|
||||||
|
selected_model: h.get("x-cortiq-selected-model") || "",
|
||||||
|
route_source: h.get("x-cortiq-route-source") || "",
|
||||||
|
cost_usd: parseFloat(h.get("x-cortiq-cost-usd") || "0"),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|||||||
Vendored
+1
@@ -200,6 +200,7 @@ export default {
|
|||||||
"pg.temp": "Temperatur",
|
"pg.temp": "Temperatur",
|
||||||
"pg.maxTokens": "Max. Tokens",
|
"pg.maxTokens": "Max. Tokens",
|
||||||
"pg.send": "Senden",
|
"pg.send": "Senden",
|
||||||
|
"pg.stream": "Streamen",
|
||||||
"pg.sending": "Wird gesendet…",
|
"pg.sending": "Wird gesendet…",
|
||||||
"pg.answer": "Antwort",
|
"pg.answer": "Antwort",
|
||||||
"pg.empty": "Antwort und Routing-Entscheidung erscheinen hier.",
|
"pg.empty": "Antwort und Routing-Entscheidung erscheinen hier.",
|
||||||
|
|||||||
Vendored
+1
@@ -201,6 +201,7 @@ export default {
|
|||||||
"pg.temp": "Temperature",
|
"pg.temp": "Temperature",
|
||||||
"pg.maxTokens": "Max tokens",
|
"pg.maxTokens": "Max tokens",
|
||||||
"pg.send": "Send",
|
"pg.send": "Send",
|
||||||
|
"pg.stream": "Stream",
|
||||||
"pg.sending": "Sending…",
|
"pg.sending": "Sending…",
|
||||||
"pg.answer": "Answer",
|
"pg.answer": "Answer",
|
||||||
"pg.empty": "The answer and routing decision will appear here.",
|
"pg.empty": "The answer and routing decision will appear here.",
|
||||||
|
|||||||
Vendored
+1
@@ -200,6 +200,7 @@ export default {
|
|||||||
"pg.temp": "Temperatura",
|
"pg.temp": "Temperatura",
|
||||||
"pg.maxTokens": "Tokens máx.",
|
"pg.maxTokens": "Tokens máx.",
|
||||||
"pg.send": "Enviar",
|
"pg.send": "Enviar",
|
||||||
|
"pg.stream": "Streaming",
|
||||||
"pg.sending": "Enviando…",
|
"pg.sending": "Enviando…",
|
||||||
"pg.answer": "Respuesta",
|
"pg.answer": "Respuesta",
|
||||||
"pg.empty": "La respuesta y la decisión de enrutamiento aparecerán aquí.",
|
"pg.empty": "La respuesta y la decisión de enrutamiento aparecerán aquí.",
|
||||||
|
|||||||
Vendored
+1
@@ -200,6 +200,7 @@ export default {
|
|||||||
"pg.temp": "Température",
|
"pg.temp": "Température",
|
||||||
"pg.maxTokens": "Tokens max",
|
"pg.maxTokens": "Tokens max",
|
||||||
"pg.send": "Envoyer",
|
"pg.send": "Envoyer",
|
||||||
|
"pg.stream": "Streaming",
|
||||||
"pg.sending": "Envoi…",
|
"pg.sending": "Envoi…",
|
||||||
"pg.answer": "Réponse",
|
"pg.answer": "Réponse",
|
||||||
"pg.empty": "La réponse et la décision de routage s'afficheront ici.",
|
"pg.empty": "La réponse et la décision de routage s'afficheront ici.",
|
||||||
|
|||||||
Vendored
+1
@@ -200,6 +200,7 @@ export default {
|
|||||||
"pg.temp": "Температура",
|
"pg.temp": "Температура",
|
||||||
"pg.maxTokens": "Макс. токены",
|
"pg.maxTokens": "Макс. токены",
|
||||||
"pg.send": "Отправить",
|
"pg.send": "Отправить",
|
||||||
|
"pg.stream": "Стриминг",
|
||||||
"pg.sending": "Отправка…",
|
"pg.sending": "Отправка…",
|
||||||
"pg.answer": "Ответ",
|
"pg.answer": "Ответ",
|
||||||
"pg.empty": "Ответ и решение роутера появятся здесь.",
|
"pg.empty": "Ответ и решение роутера появятся здесь.",
|
||||||
|
|||||||
Vendored
+1
@@ -200,6 +200,7 @@ export default {
|
|||||||
"pg.temp": "Sıcaklık",
|
"pg.temp": "Sıcaklık",
|
||||||
"pg.maxTokens": "Maks. token",
|
"pg.maxTokens": "Maks. token",
|
||||||
"pg.send": "Gönder",
|
"pg.send": "Gönder",
|
||||||
|
"pg.stream": "Akış",
|
||||||
"pg.sending": "Gönderiliyor…",
|
"pg.sending": "Gönderiliyor…",
|
||||||
"pg.answer": "Yanıt",
|
"pg.answer": "Yanıt",
|
||||||
"pg.empty": "Yanıt ve yönlendirme kararı burada görünecek.",
|
"pg.empty": "Yanıt ve yönlendirme kararı burada görünecek.",
|
||||||
|
|||||||
Vendored
+1
@@ -200,6 +200,7 @@ export default {
|
|||||||
"pg.temp": "温度",
|
"pg.temp": "温度",
|
||||||
"pg.maxTokens": "最大 Token 数",
|
"pg.maxTokens": "最大 Token 数",
|
||||||
"pg.send": "发送",
|
"pg.send": "发送",
|
||||||
|
"pg.stream": "流式",
|
||||||
"pg.sending": "发送中…",
|
"pg.sending": "发送中…",
|
||||||
"pg.answer": "回答",
|
"pg.answer": "回答",
|
||||||
"pg.empty": "回答和路由决策将在此显示。",
|
"pg.empty": "回答和路由决策将在此显示。",
|
||||||
|
|||||||
Vendored
+44
-14
@@ -1,7 +1,7 @@
|
|||||||
// Playground — send a prompt through the live pipeline, inspect routing decision.
|
// Playground — send a prompt through the live pipeline, inspect routing decision.
|
||||||
import { h, mount, money, ms, num } from "../ui.js";
|
import { h, mount, money, ms } from "../ui.js";
|
||||||
import { t } from "../i18n.js";
|
import { t } from "../i18n.js";
|
||||||
import { api } from "../api.js";
|
import { api, testStream } from "../api.js";
|
||||||
import { appState } from "../app.js";
|
import { appState } from "../app.js";
|
||||||
|
|
||||||
export async function renderPlayground() {
|
export async function renderPlayground() {
|
||||||
@@ -16,6 +16,7 @@ export async function renderPlayground() {
|
|||||||
const promptIn = h("textarea", { placeholder: t("pg.promptPlaceholder"), rows: 5 });
|
const promptIn = h("textarea", { placeholder: t("pg.promptPlaceholder"), rows: 5 });
|
||||||
const tempIn = h("input", { type: "number", step: "0.1", value: 0.7 });
|
const tempIn = h("input", { type: "number", step: "0.1", value: 0.7 });
|
||||||
const maxTokIn = h("input", { type: "number", value: 512 });
|
const maxTokIn = h("input", { type: "number", value: 512 });
|
||||||
|
const streamCb = h("input", { type: "checkbox", checked: true });
|
||||||
const profileSel = h("select", {}, ...(meta.profiles || []).map((p) => h("option", { value: p }, p)));
|
const profileSel = h("select", {}, ...(meta.profiles || []).map((p) => h("option", { value: p }, p)));
|
||||||
const modelSel = h("select", {}, ...ids.map((id) => h("option", { value: id }, id)));
|
const modelSel = h("select", {}, ...ids.map((id) => h("option", { value: id }, id)));
|
||||||
|
|
||||||
@@ -62,19 +63,43 @@ export async function renderPlayground() {
|
|||||||
sendBtn.disabled = true;
|
sendBtn.disabled = true;
|
||||||
sendBtn.textContent = t("pg.sending");
|
sendBtn.textContent = t("pg.sending");
|
||||||
mount(out, h("div", { class: "flex" }, h("span", { class: "spinner" }), " ", t("pg.sending")));
|
mount(out, h("div", { class: "flex" }, h("span", { class: "spinner" }), " ", t("pg.sending")));
|
||||||
|
const body = {
|
||||||
|
model,
|
||||||
|
messages,
|
||||||
|
temperature: parseFloat(tempIn.value),
|
||||||
|
max_tokens: parseInt(maxTokIn.value) || null,
|
||||||
|
};
|
||||||
try {
|
try {
|
||||||
const r = await api.test({
|
if (streamCb.checked) {
|
||||||
model,
|
const started = performance.now();
|
||||||
messages,
|
let acc = "";
|
||||||
temperature: parseFloat(tempIn.value),
|
out.textContent = "";
|
||||||
max_tokens: parseInt(maxTokIn.value) || null,
|
const info = await testStream(body, (d) => {
|
||||||
});
|
acc += d;
|
||||||
if (r.ok) {
|
out.textContent = acc;
|
||||||
mount(out, r.answer || h("span", { class: "muted" }, "—"));
|
});
|
||||||
renderDecision(r);
|
out.textContent = acc || "—";
|
||||||
|
renderDecision({
|
||||||
|
cortiq: {
|
||||||
|
task_label: info.task_label,
|
||||||
|
complexity: { score: info.score, tier: info.tier },
|
||||||
|
selected_model: info.selected_model,
|
||||||
|
route_source: info.route_source,
|
||||||
|
cost_usd: info.cost_usd,
|
||||||
|
failover: false,
|
||||||
|
},
|
||||||
|
usage: {},
|
||||||
|
latency_ms: Math.round(performance.now() - started),
|
||||||
|
});
|
||||||
} else {
|
} else {
|
||||||
mount(out, h("div", { class: "callout" }, h("b", {}, t("pg.error") + ": "), r.error || ""));
|
const r = await api.test(body);
|
||||||
mount(decision, h("div", { class: "card-head" }, h("h3", {}, t("pg.decision"))), h("div", { class: "empty" }, "—"));
|
if (r.ok) {
|
||||||
|
mount(out, r.answer || h("span", { class: "muted" }, "—"));
|
||||||
|
renderDecision(r);
|
||||||
|
} else {
|
||||||
|
mount(out, h("div", { class: "callout" }, h("b", {}, t("pg.error") + ": "), r.error || ""));
|
||||||
|
mount(decision, h("div", { class: "card-head" }, h("h3", {}, t("pg.decision"))), h("div", { class: "empty" }, "—"));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
mount(out, h("div", { class: "callout" }, String(e.message || e)));
|
mount(out, h("div", { class: "callout" }, String(e.message || e)));
|
||||||
@@ -126,7 +151,12 @@ export async function renderPlayground() {
|
|||||||
h("label", { class: "field" }, h("span", {}, t("pg.mode")), modeSeg),
|
h("label", { class: "field" }, h("span", {}, t("pg.mode")), modeSeg),
|
||||||
modeExtra,
|
modeExtra,
|
||||||
h("div", { class: "row" }, h("label", { class: "field" }, h("span", {}, t("pg.temp")), tempIn), h("label", { class: "field" }, h("span", {}, t("pg.maxTokens")), maxTokIn)),
|
h("div", { class: "row" }, h("label", { class: "field" }, h("span", {}, t("pg.temp")), tempIn), h("label", { class: "field" }, h("span", {}, t("pg.maxTokens")), maxTokIn)),
|
||||||
h("div", { class: "flex", style: "justify-content:flex-end" }, sendBtn)
|
h(
|
||||||
|
"div",
|
||||||
|
{ class: "flex", style: "justify-content:space-between" },
|
||||||
|
h("label", { class: "check", style: "margin:0" }, streamCb, t("pg.stream")),
|
||||||
|
sendBtn
|
||||||
|
)
|
||||||
),
|
),
|
||||||
h("div", { class: "card" }, h("div", { class: "card-head" }, h("h3", {}, t("pg.answer"))), out)
|
h("div", { class: "card" }, h("div", { class: "card-head" }, h("h3", {}, t("pg.answer"))), out)
|
||||||
),
|
),
|
||||||
|
|||||||
Reference in New Issue
Block a user