feat: Anthropic provider/inbound + embeddings
- Anthropic outbound provider: /v1/messages, system extraction, x-api-key + anthropic-version; streaming translates Anthropic SSE -> unified OpenAI wire format - Anthropic inbound adapter (POST /v1/messages): non-streaming JSON + streaming SSE (OpenAI -> Anthropic event translation) - Embeddings: POST /v1/embeddings inbound + Provider::embed(); embedding models resolvable from the pool (registry no longer skips them) - protocols mounted always, gated on live flags; admin meta marks them implemented
This commit is contained in:
@@ -12,6 +12,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/).
|
|||||||
byte. Live streaming output in the admin Playground (`/admin/api/test/stream`).
|
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 Windows binaries in releases (`x86_64-pc-windows-msvc`) and a
|
||||||
cross-platform CI test matrix (ubuntu/windows/macos); secure token RNG via `getrandom`.
|
cross-platform CI test matrix (ubuntu/windows/macos); secure token RNG via `getrandom`.
|
||||||
|
- **Anthropic provider** (outbound): `POST /v1/messages` with `system` extraction,
|
||||||
|
`x-api-key` + `anthropic-version` headers, and streaming that translates Anthropic
|
||||||
|
SSE events into the unified OpenAI wire format.
|
||||||
|
- **Anthropic inbound** adapter: `POST /v1/messages` (non-streaming JSON and streaming
|
||||||
|
SSE), so Claude-native clients can target the gateway.
|
||||||
|
- **Embeddings**: `POST /v1/embeddings` inbound + provider `embed()`; embedding models
|
||||||
|
are now resolvable from the pool.
|
||||||
|
|
||||||
## [0.1.0] - 2026-06-30
|
## [0.1.0] - 2026-06-30
|
||||||
|
|
||||||
|
|||||||
+3
-3
@@ -271,7 +271,7 @@ async fn meta() -> ApiResult {
|
|||||||
ok(json!({
|
ok(json!({
|
||||||
"version": env!("CARGO_PKG_VERSION"),
|
"version": env!("CARGO_PKG_VERSION"),
|
||||||
"providers": ["openai", "anthropic", "ollama", "http"],
|
"providers": ["openai", "anthropic", "ollama", "http"],
|
||||||
"provider_impl": { "openai": true, "ollama": true, "http": true, "anthropic": false },
|
"provider_impl": { "openai": true, "ollama": true, "http": true, "anthropic": true },
|
||||||
"cost_tiers": ["cheap", "mid", "expensive"],
|
"cost_tiers": ["cheap", "mid", "expensive"],
|
||||||
"kinds": ["chat", "embedding"],
|
"kinds": ["chat", "embedding"],
|
||||||
"profiles": ["cost-saver", "balanced", "quality-first"],
|
"profiles": ["cost-saver", "balanced", "quality-first"],
|
||||||
@@ -280,8 +280,8 @@ async fn meta() -> ApiResult {
|
|||||||
"tiers": ["low", "medium", "high"],
|
"tiers": ["low", "medium", "high"],
|
||||||
"caps": ["tools", "vision"],
|
"caps": ["tools", "vision"],
|
||||||
"protocols_impl": {
|
"protocols_impl": {
|
||||||
"openai_chat": true, "openai_completions": false, "openai_embeddings": false,
|
"openai_chat": true, "openai_completions": false, "openai_embeddings": true,
|
||||||
"openai_models": false, "anthropic_messages": false, "mcp": false, "native_passthrough": false
|
"openai_models": false, "anthropic_messages": true, "mcp": false, "native_passthrough": false
|
||||||
},
|
},
|
||||||
"languages": ["en", "ru", "de", "fr", "es", "zh", "tr"],
|
"languages": ["en", "ru", "de", "fr", "es", "zh", "tr"],
|
||||||
}))
|
}))
|
||||||
|
|||||||
@@ -0,0 +1,209 @@
|
|||||||
|
//! Inbound Anthropic Messages adapter: `POST /v1/messages`.
|
||||||
|
//! Translates the Anthropic request into the canonical [`ChatRequest`], runs the
|
||||||
|
//! pipeline, and converts the result back to the Anthropic Messages format
|
||||||
|
//! (non-streaming JSON or streaming SSE).
|
||||||
|
|
||||||
|
use crate::error::{GatewayError, Result};
|
||||||
|
use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective};
|
||||||
|
use crate::protocols::openai_chat::cortiq_headers;
|
||||||
|
use crate::state::SharedState;
|
||||||
|
use axum::response::{IntoResponse, Response};
|
||||||
|
use axum::{extract::State, routing::post, Json, Router};
|
||||||
|
use serde::Deserialize;
|
||||||
|
|
||||||
|
pub fn routes() -> Router<SharedState> {
|
||||||
|
Router::new().route("/v1/messages", post(handler))
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Deserialize)]
|
||||||
|
struct MessagesRequest {
|
||||||
|
model: String,
|
||||||
|
#[serde(default)]
|
||||||
|
system: serde_json::Value,
|
||||||
|
#[serde(default)]
|
||||||
|
messages: Vec<serde_json::Value>,
|
||||||
|
#[serde(default)]
|
||||||
|
max_tokens: Option<u32>,
|
||||||
|
#[serde(default)]
|
||||||
|
temperature: Option<f32>,
|
||||||
|
#[serde(default)]
|
||||||
|
top_p: Option<f32>,
|
||||||
|
#[serde(default)]
|
||||||
|
stream: bool,
|
||||||
|
#[serde(default)]
|
||||||
|
tools: Vec<serde_json::Value>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Anthropic content is a string or an array of blocks; flatten to text.
|
||||||
|
fn content_to_string(v: &serde_json::Value) -> String {
|
||||||
|
if let Some(s) = v.as_str() {
|
||||||
|
return s.to_string();
|
||||||
|
}
|
||||||
|
if let Some(arr) = v.as_array() {
|
||||||
|
return arr
|
||||||
|
.iter()
|
||||||
|
.filter_map(|b| b["text"].as_str())
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
.join("");
|
||||||
|
}
|
||||||
|
String::new()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn parse_routing(model: &str) -> RoutingDirective {
|
||||||
|
if let Some(rest) = model.strip_prefix("cortiq-auto") {
|
||||||
|
RoutingDirective::Auto {
|
||||||
|
profile: rest.strip_prefix(':').map(|p| p.to_string()),
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
RoutingDirective::Pinned {
|
||||||
|
model_id: model.to_string(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn handler(
|
||||||
|
State(state): State<SharedState>,
|
||||||
|
Json(req): Json<MessagesRequest>,
|
||||||
|
) -> Result<Response> {
|
||||||
|
if !state.live().cfg.protocols.anthropic_messages {
|
||||||
|
return Err(GatewayError::InvalidRequest(
|
||||||
|
"anthropic_messages protocol is disabled".into(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
if req.messages.is_empty() {
|
||||||
|
return Err(GatewayError::InvalidRequest(
|
||||||
|
"messages must not be empty".into(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut messages = Vec::new();
|
||||||
|
let sys = content_to_string(&req.system);
|
||||||
|
if !sys.is_empty() {
|
||||||
|
messages.push(Message {
|
||||||
|
role: "system".into(),
|
||||||
|
content: sys,
|
||||||
|
tool_calls: vec![],
|
||||||
|
});
|
||||||
|
}
|
||||||
|
for m in &req.messages {
|
||||||
|
messages.push(Message {
|
||||||
|
role: m["role"].as_str().unwrap_or("user").to_string(),
|
||||||
|
content: content_to_string(&m["content"]),
|
||||||
|
tool_calls: vec![],
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
let canonical = ChatRequest {
|
||||||
|
routing: parse_routing(&req.model),
|
||||||
|
messages,
|
||||||
|
tools: req.tools,
|
||||||
|
params: GenParams {
|
||||||
|
temperature: req.temperature,
|
||||||
|
max_tokens: req.max_tokens,
|
||||||
|
top_p: req.top_p,
|
||||||
|
stop: Vec::new(),
|
||||||
|
passthrough: Default::default(),
|
||||||
|
},
|
||||||
|
stream: req.stream,
|
||||||
|
meta: RequestMeta {
|
||||||
|
protocol: "anthropic_messages".into(),
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
};
|
||||||
|
|
||||||
|
if req.stream {
|
||||||
|
let (info, stream) = state.pipeline.run_stream(canonical, &state).await?;
|
||||||
|
let mut headers = cortiq_headers(&info);
|
||||||
|
headers.insert(
|
||||||
|
axum::http::header::CONTENT_TYPE,
|
||||||
|
axum::http::HeaderValue::from_static("text/event-stream"),
|
||||||
|
);
|
||||||
|
let translated = openai_to_anthropic_sse(stream);
|
||||||
|
return Ok((headers, axum::body::Body::from_stream(translated)).into_response());
|
||||||
|
}
|
||||||
|
|
||||||
|
let resp = state.pipeline.run(canonical, &state).await?;
|
||||||
|
let answer = resp
|
||||||
|
.choices
|
||||||
|
.first()
|
||||||
|
.map(|c| c.message.content.clone())
|
||||||
|
.unwrap_or_default();
|
||||||
|
let headers = cortiq_headers(&resp.cortiq);
|
||||||
|
let body = serde_json::json!({
|
||||||
|
"id": resp.id,
|
||||||
|
"type": "message",
|
||||||
|
"role": "assistant",
|
||||||
|
"model": resp.model_used,
|
||||||
|
"content": [{ "type": "text", "text": answer }],
|
||||||
|
"stop_reason": "end_turn",
|
||||||
|
"stop_sequence": serde_json::Value::Null,
|
||||||
|
"usage": {
|
||||||
|
"input_tokens": resp.usage.prompt_tokens,
|
||||||
|
"output_tokens": resp.usage.completion_tokens,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
Ok((headers, Json(body)).into_response())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Translate the gateway's internal OpenAI-format SSE into Anthropic Messages SSE
|
||||||
|
/// events, so Anthropic-native clients can stream regardless of the upstream provider.
|
||||||
|
fn openai_to_anthropic_sse(
|
||||||
|
stream: crate::providers::ChatStream,
|
||||||
|
) -> impl futures::Stream<Item = Result<bytes::Bytes>> + Send + 'static {
|
||||||
|
use futures::StreamExt;
|
||||||
|
async_stream::stream! {
|
||||||
|
fn evt(event: &str, data: serde_json::Value) -> bytes::Bytes {
|
||||||
|
bytes::Bytes::from(format!("event: {event}\ndata: {data}\n\n"))
|
||||||
|
}
|
||||||
|
|
||||||
|
yield Ok(evt("message_start", serde_json::json!({
|
||||||
|
"type": "message_start",
|
||||||
|
"message": {"id": "msg_gw", "type": "message", "role": "assistant",
|
||||||
|
"content": [], "model": "cortiq-auto", "stop_reason": null,
|
||||||
|
"usage": {"input_tokens": 0, "output_tokens": 0}}
|
||||||
|
})));
|
||||||
|
yield Ok(evt("content_block_start", serde_json::json!({
|
||||||
|
"type": "content_block_start", "index": 0,
|
||||||
|
"content_block": {"type": "text", "text": ""}
|
||||||
|
})));
|
||||||
|
|
||||||
|
let mut buf = String::new();
|
||||||
|
let mut completion_tokens = 0u64;
|
||||||
|
futures::pin_mut!(stream);
|
||||||
|
while let Some(item) = stream.next().await {
|
||||||
|
let bytes = match item {
|
||||||
|
Ok(b) => b,
|
||||||
|
Err(e) => { yield Err(e); break; }
|
||||||
|
};
|
||||||
|
if let Ok(s) = std::str::from_utf8(&bytes) {
|
||||||
|
buf.push_str(s);
|
||||||
|
}
|
||||||
|
while let Some(idx) = buf.find("\n\n") {
|
||||||
|
let chunk: String = buf.drain(..idx + 2).collect();
|
||||||
|
for line in chunk.lines() {
|
||||||
|
let Some(data) = line.trim_start().strip_prefix("data:") else { continue };
|
||||||
|
let data = data.trim();
|
||||||
|
if data.is_empty() || data == "[DONE]" { continue; }
|
||||||
|
let Ok(v) = serde_json::from_str::<serde_json::Value>(data) else { continue };
|
||||||
|
if let Some(u) = v.get("usage").filter(|u| !u.is_null()) {
|
||||||
|
completion_tokens = u["completion_tokens"].as_u64().unwrap_or(completion_tokens);
|
||||||
|
}
|
||||||
|
if let Some(t) = v["choices"][0]["delta"]["content"].as_str() {
|
||||||
|
yield Ok(evt("content_block_delta", serde_json::json!({
|
||||||
|
"type": "content_block_delta", "index": 0,
|
||||||
|
"delta": {"type": "text_delta", "text": t}
|
||||||
|
})));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
yield Ok(evt("content_block_stop", serde_json::json!({"type": "content_block_stop", "index": 0})));
|
||||||
|
yield Ok(evt("message_delta", serde_json::json!({
|
||||||
|
"type": "message_delta",
|
||||||
|
"delta": {"stop_reason": "end_turn", "stop_sequence": null},
|
||||||
|
"usage": {"output_tokens": completion_tokens}
|
||||||
|
})));
|
||||||
|
yield Ok(evt("message_stop", serde_json::json!({"type": "message_stop"})));
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,7 +1,9 @@
|
|||||||
//! Incoming protocol adapters. Each one builds its own axum routes and translates
|
//! Incoming protocol adapters. Each one builds its own axum routes and translates
|
||||||
//! its protocol to/from the canonical model. Enabled via flags in `[protocols]`.
|
//! its protocol to/from the canonical model. Enabled via flags in `[protocols]`.
|
||||||
|
|
||||||
|
pub mod anthropic_messages;
|
||||||
pub mod openai_chat;
|
pub mod openai_chat;
|
||||||
|
pub mod openai_embeddings;
|
||||||
|
|
||||||
use crate::state::SharedState;
|
use crate::state::SharedState;
|
||||||
use axum::Router;
|
use axum::Router;
|
||||||
@@ -10,8 +12,9 @@ use axum::Router;
|
|||||||
/// each handler checks the live `protocols.*` flag from config itself, so
|
/// each handler checks the live `protocols.*` flag from config itself, so
|
||||||
/// toggles in the admin panel take effect without a restart (hot switching).
|
/// toggles in the admin panel take effect without a restart (hot switching).
|
||||||
pub fn build_router() -> Router<SharedState> {
|
pub fn build_router() -> Router<SharedState> {
|
||||||
Router::new().merge(openai_chat::routes())
|
Router::new()
|
||||||
// TODO: openai_completions, openai_embeddings, openai_models,
|
.merge(openai_chat::routes())
|
||||||
// anthropic_messages, mcp, native_passthrough — add here as well,
|
.merge(openai_embeddings::routes())
|
||||||
// with a check of the corresponding flag inside the handler.
|
.merge(anthropic_messages::routes())
|
||||||
|
// TODO: openai_completions, openai_models, mcp, native_passthrough.
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,52 @@
|
|||||||
|
//! Inbound OpenAI Embeddings adapter: `POST /v1/embeddings`.
|
||||||
|
//! Resolves an embedding model (explicit id, or the first `kind = "embedding"`),
|
||||||
|
//! calls the provider, and forwards the OpenAI-style response body.
|
||||||
|
|
||||||
|
use crate::error::{GatewayError, Result};
|
||||||
|
use crate::state::SharedState;
|
||||||
|
use axum::{extract::State, response::IntoResponse, routing::post, Json, Router};
|
||||||
|
use serde::Deserialize;
|
||||||
|
|
||||||
|
pub fn routes() -> Router<SharedState> {
|
||||||
|
Router::new().route("/v1/embeddings", post(handler))
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Deserialize)]
|
||||||
|
struct EmbeddingsRequest {
|
||||||
|
#[serde(default)]
|
||||||
|
model: String,
|
||||||
|
input: serde_json::Value,
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn handler(
|
||||||
|
State(state): State<SharedState>,
|
||||||
|
Json(req): Json<EmbeddingsRequest>,
|
||||||
|
) -> Result<impl IntoResponse> {
|
||||||
|
let live = state.live();
|
||||||
|
if !live.cfg.protocols.openai_embeddings {
|
||||||
|
return Err(GatewayError::InvalidRequest(
|
||||||
|
"openai_embeddings protocol is disabled".into(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
// resolve the embedding model: explicit real id, else first kind == "embedding"
|
||||||
|
let model_id = if !req.model.is_empty()
|
||||||
|
&& req.model != "cortiq-auto"
|
||||||
|
&& live.registry.get(&req.model).is_some()
|
||||||
|
{
|
||||||
|
req.model.clone()
|
||||||
|
} else {
|
||||||
|
live.cfg
|
||||||
|
.models
|
||||||
|
.iter()
|
||||||
|
.find(|m| m.kind == "embedding")
|
||||||
|
.map(|m| m.id.clone())
|
||||||
|
.ok_or_else(|| GatewayError::InvalidRequest("no embedding model configured".into()))?
|
||||||
|
};
|
||||||
|
|
||||||
|
let provider = live.registry.get(&model_id).ok_or_else(|| {
|
||||||
|
GatewayError::UpstreamUnavailable(format!("embedding model '{model_id}' is not available"))
|
||||||
|
})?;
|
||||||
|
let body = provider.embed(req.input).await?;
|
||||||
|
Ok(Json(body))
|
||||||
|
}
|
||||||
+215
-13
@@ -1,23 +1,25 @@
|
|||||||
//! Anthropic Messages provider (v0.2). Translates the canonical ChatRequest to
|
//! Anthropic Messages provider. Translates the canonical [`ChatRequest`] to
|
||||||
//! `POST /v1/messages` and back. The `system` message is extracted into a separate field.
|
//! `POST {base_url}/v1/messages` and back. `system` messages are lifted into the
|
||||||
|
//! top-level `system` field. Streaming translates Anthropic SSE events into the
|
||||||
|
//! OpenAI wire format, so the gateway's unified SSE passthrough works uniformly.
|
||||||
|
|
||||||
use super::{Caps, Provider};
|
use super::{Caps, Provider};
|
||||||
use crate::config::ModelCfg;
|
use crate::config::ModelCfg;
|
||||||
use crate::error::{GatewayError, Result};
|
use crate::error::{GatewayError, Result};
|
||||||
use crate::model::{ChatRequest, ChatResponse};
|
use crate::model::{ChatRequest, ChatResponse, Choice, Message, RouteInfo, Usage};
|
||||||
use async_trait::async_trait;
|
use async_trait::async_trait;
|
||||||
|
|
||||||
|
const ANTHROPIC_VERSION: &str = "2023-06-01";
|
||||||
|
|
||||||
pub struct AnthropicProvider {
|
pub struct AnthropicProvider {
|
||||||
id: String,
|
id: String,
|
||||||
#[allow(dead_code)]
|
|
||||||
base_url: String,
|
base_url: String,
|
||||||
#[allow(dead_code)]
|
|
||||||
model: String,
|
model: String,
|
||||||
#[allow(dead_code)]
|
|
||||||
api_key: Option<String>,
|
api_key: Option<String>,
|
||||||
caps: Caps,
|
caps: Caps,
|
||||||
price_in: f64,
|
price_in: f64,
|
||||||
price_out: f64,
|
price_out: f64,
|
||||||
|
http: reqwest::Client,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl AnthropicProvider {
|
impl AnthropicProvider {
|
||||||
@@ -30,8 +32,61 @@ impl AnthropicProvider {
|
|||||||
caps: Caps::from_list(&cfg.caps),
|
caps: Caps::from_list(&cfg.caps),
|
||||||
price_in: cfg.price_in,
|
price_in: cfg.price_in,
|
||||||
price_out: cfg.price_out,
|
price_out: cfg.price_out,
|
||||||
|
http: reqwest::Client::new(),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Canonical request → Anthropic Messages request body.
|
||||||
|
fn to_wire(&self, req: &ChatRequest, stream: bool) -> serde_json::Value {
|
||||||
|
let mut system = String::new();
|
||||||
|
let mut messages = Vec::new();
|
||||||
|
for m in &req.messages {
|
||||||
|
if m.role == "system" {
|
||||||
|
if !system.is_empty() {
|
||||||
|
system.push('\n');
|
||||||
|
}
|
||||||
|
system.push_str(&m.content);
|
||||||
|
} else {
|
||||||
|
let role = if m.role == "assistant" {
|
||||||
|
"assistant"
|
||||||
|
} else {
|
||||||
|
"user"
|
||||||
|
};
|
||||||
|
messages.push(serde_json::json!({ "role": role, "content": m.content }));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let mut body = serde_json::json!({
|
||||||
|
"model": self.model,
|
||||||
|
"messages": messages,
|
||||||
|
"max_tokens": req.params.max_tokens.unwrap_or(1024),
|
||||||
|
"stream": stream,
|
||||||
|
});
|
||||||
|
if !system.is_empty() {
|
||||||
|
body["system"] = serde_json::Value::String(system);
|
||||||
|
}
|
||||||
|
if let Some(t) = req.params.temperature {
|
||||||
|
body["temperature"] = t.into();
|
||||||
|
}
|
||||||
|
if let Some(p) = req.params.top_p {
|
||||||
|
body["top_p"] = p.into();
|
||||||
|
}
|
||||||
|
if !req.tools.is_empty() {
|
||||||
|
body["tools"] = serde_json::Value::Array(req.tools.clone());
|
||||||
|
}
|
||||||
|
body
|
||||||
|
}
|
||||||
|
|
||||||
|
fn request(&self, body: &serde_json::Value) -> reqwest::RequestBuilder {
|
||||||
|
let mut r = self
|
||||||
|
.http
|
||||||
|
.post(format!("{}/v1/messages", self.base_url))
|
||||||
|
.header("anthropic-version", ANTHROPIC_VERSION)
|
||||||
|
.json(body);
|
||||||
|
if let Some(key) = &self.api_key {
|
||||||
|
r = r.header("x-api-key", key);
|
||||||
|
}
|
||||||
|
r
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
@@ -46,12 +101,159 @@ impl Provider for AnthropicProvider {
|
|||||||
(prompt_tokens as f64 * self.price_in + completion_tokens as f64 * self.price_out)
|
(prompt_tokens as f64 * self.price_in + completion_tokens as f64 * self.price_out)
|
||||||
/ 1_000_000.0
|
/ 1_000_000.0
|
||||||
}
|
}
|
||||||
async fn chat(&self, _req: ChatRequest) -> Result<ChatResponse> {
|
|
||||||
// TODO(v0.2): extract system from messages → "system" field; messages → content blocks;
|
async fn chat(&self, req: ChatRequest) -> Result<ChatResponse> {
|
||||||
// add x-api-key + anthropic-version headers; map response → canonical form.
|
let resp = self
|
||||||
// Returns an error (not a panic) so that failover/playground continue to work normally.
|
.request(&self.to_wire(&req, false))
|
||||||
Err(GatewayError::UpstreamUnavailable(
|
.send()
|
||||||
"anthropic provider not yet implemented (planned v0.2)".to_string(),
|
.await
|
||||||
))
|
.map_err(|e| GatewayError::UpstreamUnavailable(e.to_string()))?;
|
||||||
|
if !resp.status().is_success() {
|
||||||
|
let status = resp.status();
|
||||||
|
let b = resp.text().await.unwrap_or_default();
|
||||||
|
return Err(GatewayError::UpstreamUnavailable(format!(
|
||||||
|
"provider {} returned HTTP {}: {}",
|
||||||
|
self.id, status, b
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
let v: serde_json::Value = resp.json().await.map_err(|e| {
|
||||||
|
GatewayError::Internal(format!("failed to parse anthropic response: {e}"))
|
||||||
|
})?;
|
||||||
|
let content = v["content"]
|
||||||
|
.as_array()
|
||||||
|
.map(|blocks| {
|
||||||
|
blocks
|
||||||
|
.iter()
|
||||||
|
.filter_map(|b| b["text"].as_str())
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
.join("")
|
||||||
|
})
|
||||||
|
.unwrap_or_default();
|
||||||
|
let prompt_tokens = v["usage"]["input_tokens"].as_u64().unwrap_or(0) as u32;
|
||||||
|
let completion_tokens = v["usage"]["output_tokens"].as_u64().unwrap_or(0) as u32;
|
||||||
|
Ok(ChatResponse {
|
||||||
|
id: v["id"].as_str().unwrap_or("").to_string(),
|
||||||
|
model_used: v["model"].as_str().unwrap_or(&self.model).to_string(),
|
||||||
|
choices: vec![Choice {
|
||||||
|
index: 0,
|
||||||
|
message: Message {
|
||||||
|
role: "assistant".into(),
|
||||||
|
content,
|
||||||
|
tool_calls: vec![],
|
||||||
|
},
|
||||||
|
finish_reason: v["stop_reason"].as_str().unwrap_or("stop").to_string(),
|
||||||
|
}],
|
||||||
|
usage: Usage {
|
||||||
|
prompt_tokens,
|
||||||
|
completion_tokens,
|
||||||
|
total_tokens: prompt_tokens + completion_tokens,
|
||||||
|
},
|
||||||
|
cortiq: RouteInfo {
|
||||||
|
task_label: String::new(),
|
||||||
|
complexity_score: 0.0,
|
||||||
|
complexity_tier: String::new(),
|
||||||
|
selected_model: self.id.clone(),
|
||||||
|
route_source: String::new(),
|
||||||
|
router_request_id: None,
|
||||||
|
cost_usd: 0.0,
|
||||||
|
failover: false,
|
||||||
|
},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn chat_stream(&self, req: ChatRequest) -> Result<super::ChatStream> {
|
||||||
|
let resp = self
|
||||||
|
.request(&self.to_wire(&req, true))
|
||||||
|
.send()
|
||||||
|
.await
|
||||||
|
.map_err(|e| GatewayError::UpstreamUnavailable(e.to_string()))?;
|
||||||
|
if !resp.status().is_success() {
|
||||||
|
let status = resp.status();
|
||||||
|
let b = resp.text().await.unwrap_or_default();
|
||||||
|
return Err(GatewayError::UpstreamUnavailable(format!(
|
||||||
|
"provider {} returned HTTP {}: {}",
|
||||||
|
self.id, status, b
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
Ok(Box::pin(anthropic_to_openai_sse(resp.bytes_stream())))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Translate an Anthropic Messages SSE stream into OpenAI `chat.completion.chunk`
|
||||||
|
/// SSE so the gateway's unified streaming path works for Anthropic too.
|
||||||
|
fn anthropic_to_openai_sse(
|
||||||
|
upstream: impl futures::Stream<Item = reqwest::Result<bytes::Bytes>> + Send + 'static,
|
||||||
|
) -> impl futures::Stream<Item = Result<bytes::Bytes>> + Send + 'static {
|
||||||
|
use futures::StreamExt;
|
||||||
|
async_stream::stream! {
|
||||||
|
futures::pin_mut!(upstream);
|
||||||
|
let mut buf = String::new();
|
||||||
|
let mut input_tokens = 0u64;
|
||||||
|
let mut role_sent = false;
|
||||||
|
let emit = |obj: serde_json::Value| bytes::Bytes::from(format!("data: {obj}\n\n"));
|
||||||
|
|
||||||
|
while let Some(chunk) = upstream.next().await {
|
||||||
|
let chunk = match chunk {
|
||||||
|
Ok(c) => c,
|
||||||
|
Err(e) => {
|
||||||
|
yield Err(GatewayError::UpstreamUnavailable(e.to_string()));
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
if let Ok(s) = std::str::from_utf8(&chunk) {
|
||||||
|
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() {
|
||||||
|
let Some(data) = line.trim_start().strip_prefix("data:") else {
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
let data = data.trim();
|
||||||
|
if data.is_empty() {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
let Ok(v) = serde_json::from_str::<serde_json::Value>(data) else {
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
match v["type"].as_str().unwrap_or("") {
|
||||||
|
"message_start" => {
|
||||||
|
input_tokens =
|
||||||
|
v["message"]["usage"]["input_tokens"].as_u64().unwrap_or(0);
|
||||||
|
if !role_sent {
|
||||||
|
role_sent = true;
|
||||||
|
yield Ok(emit(serde_json::json!({
|
||||||
|
"choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": null}]
|
||||||
|
})));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
"content_block_delta" => {
|
||||||
|
if let Some(t) = v["delta"]["text"].as_str() {
|
||||||
|
yield Ok(emit(serde_json::json!({
|
||||||
|
"choices": [{"index": 0, "delta": {"content": t}, "finish_reason": null}]
|
||||||
|
})));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
"message_delta" => {
|
||||||
|
let out = v["usage"]["output_tokens"].as_u64().unwrap_or(0);
|
||||||
|
let stop = v["delta"]["stop_reason"].as_str().unwrap_or("stop").to_string();
|
||||||
|
yield Ok(emit(serde_json::json!({
|
||||||
|
"choices": [{"index": 0, "delta": {}, "finish_reason": stop}]
|
||||||
|
})));
|
||||||
|
yield Ok(emit(serde_json::json!({
|
||||||
|
"choices": [],
|
||||||
|
"usage": {
|
||||||
|
"prompt_tokens": input_tokens,
|
||||||
|
"completion_tokens": out,
|
||||||
|
"total_tokens": input_tokens + out
|
||||||
|
}
|
||||||
|
})));
|
||||||
|
}
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
yield Ok(bytes::Bytes::from("data: [DONE]\n\n"));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -57,5 +57,13 @@ pub trait Provider: Send + Sync {
|
|||||||
)))
|
)))
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO(v0.2): async fn embed(...) -> Result<EmbedResponse>.
|
/// Embeddings call. `input` is the OpenAI `input` field (string or array of
|
||||||
|
/// strings); returns the provider's OpenAI-style embeddings response body.
|
||||||
|
/// Default: not supported (override per provider).
|
||||||
|
async fn embed(&self, _input: serde_json::Value) -> crate::error::Result<serde_json::Value> {
|
||||||
|
Err(GatewayError::UpstreamUnavailable(format!(
|
||||||
|
"embeddings are not supported by provider '{}'",
|
||||||
|
self.id()
|
||||||
|
)))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -185,4 +185,30 @@ impl Provider for OpenAiProvider {
|
|||||||
.map(|res| res.map_err(|e| GatewayError::UpstreamUnavailable(e.to_string())));
|
.map(|res| res.map_err(|e| GatewayError::UpstreamUnavailable(e.to_string())));
|
||||||
Ok(Box::pin(stream))
|
Ok(Box::pin(stream))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn embed(&self, input: serde_json::Value) -> Result<serde_json::Value> {
|
||||||
|
let body = serde_json::json!({ "model": self.model, "input": input });
|
||||||
|
let mut r = self
|
||||||
|
.http
|
||||||
|
.post(format!("{}/embeddings", 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
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
resp.json().await.map_err(|e| {
|
||||||
|
GatewayError::Internal(format!("failed to parse embeddings response: {e}"))
|
||||||
|
})
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -15,9 +15,6 @@ impl Registry {
|
|||||||
pub fn from_config(cfg: &Config, secrets: &SecretStore) -> anyhow::Result<Self> {
|
pub fn from_config(cfg: &Config, secrets: &SecretStore) -> anyhow::Result<Self> {
|
||||||
let mut by_id: HashMap<String, Arc<dyn Provider>> = HashMap::new();
|
let mut by_id: HashMap<String, Arc<dyn Provider>> = HashMap::new();
|
||||||
for m in &cfg.models {
|
for m in &cfg.models {
|
||||||
if m.kind == "embedding" {
|
|
||||||
continue; // embedding models are handled by a separate branch (v0.1)
|
|
||||||
}
|
|
||||||
// provider key: secret store → environment variable
|
// provider key: secret store → environment variable
|
||||||
let api_key = m
|
let api_key = m
|
||||||
.api_key_env
|
.api_key_env
|
||||||
|
|||||||
Reference in New Issue
Block a user