From 9a848e2c78920ddc664ba2a030eef3ff8bf7e838 Mon Sep 17 00:00:00 2001 From: Jacob Date: Fri, 9 Oct 2026 22:58:34 -0400 Subject: [PATCH] feat: LiquidAI and decisions and system one implementations --- src/lib.rs | 3 + src/server/chat/completions.rs | 1 + src/server/decisions.rs | 160 ++++++++++++++++++++++++ src/server/embeddings.rs | 15 +-- src/server/mod.rs | 5 + src/server/system_one.rs | 215 +++++++++++++++++++++++++++++++++ 6 files changed, 386 insertions(+), 13 deletions(-) create mode 100644 src/server/decisions.rs create mode 100644 src/server/system_one.rs diff --git a/src/lib.rs b/src/lib.rs index 7f4ea4c..eea25f2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -48,6 +48,9 @@ pub struct AppArgs { #[arg(long)] pub hugging_face_api_key: Vec, + #[arg(long)] + pub liquid_ai_api_key: Vec, + #[arg(long)] pub mistral_api_key: Vec, diff --git a/src/server/chat/completions.rs b/src/server/chat/completions.rs index 079f4a9..791d967 100644 --- a/src/server/chat/completions.rs +++ b/src/server/chat/completions.rs @@ -316,6 +316,7 @@ pub async fn create_chat_completion( .build() }) .collect(), + Provider::LiquidAi => unreachable!(), Provider::Mistral => state .app_args .mistral_api_key diff --git a/src/server/decisions.rs b/src/server/decisions.rs new file mode 100644 index 0000000..f61bf6f --- /dev/null +++ b/src/server/decisions.rs @@ -0,0 +1,160 @@ +use crate::CloudflareCredentials; +use crate::openai; +use crate::server::AppState; +use crate::server::Provider; +use crate::server::Upstream; +use crate::server::take_random_by_swapping; +use axum::Json; +use axum::extract::State; +use axum::http::StatusCode; +use axum::response::IntoResponse; +use axum::response::Response; +use axum_extra::routing::TypedPath; +use ecow::EcoString; +use headers::Header; +use headers::authorization::Bearer; +use mitsein::vec1::Vec1; +use reqwest::Certificate; +use secrecy::ExposeSecret; +use serde::Deserialize; +use serde_with::DeserializeFromStr; +use serde_with::SerializeDisplay; +use std::collections::VecDeque; +use strum::EnumString; +use strum::VariantArray; + +#[derive(Deserialize, TypedPath)] +#[typed_path("/v1/decisions")] +pub struct V1DecisionsPath; + +#[derive( + Clone, Debug, EnumString, DeserializeFromStr, strum::Display, SerializeDisplay, VariantArray, +)] +#[strum(serialize_all = "lowercase")] +pub enum Shortcut {} + +impl Shortcut { + fn suggestions(&self) -> Vec<(Provider, EcoString)> { + use Provider as P; + use Shortcut as S; + + match self { + _ => unreachable!() + } + } +} + +pub struct DecisionsResponse(openai::embeddings::CreateEmbeddingResponse); + +impl IntoResponse for DecisionsResponse { + fn into_response(self) -> Response { + let Self(embeddings) = self; + + Json(embeddings).into_response() + } +} + +pub enum DecisionsError { + NoneSucceeded, +} + +impl IntoResponse for DecisionsError { + fn into_response(self) -> Response { + match self { + DecisionsError::NoneSucceeded => StatusCode::SERVICE_UNAVAILABLE.into_response(), + } + } +} + +#[tracing::instrument(skip(state, request))] +pub async fn create_decision( + _: V1DecisionsPath, + State(state): State, + Json(request): Json>, +) -> Result { + let shortcut = request.model.clone(); + let mut model_upstreams = VecDeque::new(); + + for (provider, model) in shortcut.suggestions() { + let upstreams: Vec = match provider { + Provider::CloudflareWorkersAi => unreachable!(), + Provider::GoogleAiStudio => unreachable!(), + Provider::Groq => unreachable!(), + Provider::HuggingFace => unreachable!(), + Provider::LiquidAi => unreachable!(), + Provider::Mistral => unreachable!(), + Provider::OpenRouter => unreachable!(), + Provider::ZAi => unreachable!(), + }; + + if let Ok(upstreams) = Vec1::try_from(upstreams) { + model_upstreams.push_back((model, upstreams)); + } + } + + while let Some((model, upstreams)) = model_upstreams.pop_front() { + let (upstream, upstreams) = take_random_by_swapping(upstreams, &mut rand::rng()); + if let Some(upstreams) = upstreams { + model_upstreams.push_back((model.clone(), upstreams)); + } + + tracing::info!(?model, ?upstream); + + let openai::decisions::DecisionRequest { model: _shortcut, input, questions } = request.clone(); + + let request = openai::decisions::DecisionRequest { model, input, questions }; + + let client = reqwest::Client::builder() + .tls_certs_only( + webpki_root_certs::TLS_SERVER_ROOT_CERTS + .iter() + .filter_map(|cert| Certificate::from_der(cert).ok()), + ) + .build() + .expect("TODO"); + + let mut upstream_request = client.post(upstream.url).json(&request); + + if let Some(api_key) = upstream.api_key { + let authorization = + headers::Authorization::bearer(api_key.expose_secret()).expect("TODO"); + upstream_request = upstream_request.header(headers::Authorization::::name(), { + let mut vec = vec![]; + + authorization.encode(&mut vec); + + let [one] = vec.try_into().expect("TODO"); + + one + }); + } + + let upstream_response = match upstream_request.send().await { + Ok(upstream_response) => upstream_response, + Err(upstream_error) => { + tracing::error!(?upstream_error); + continue; + } + }; + + if let Err(upstream_error) = upstream_response.error_for_status_ref() { + tracing::error!(?upstream_error); + match upstream_response.text().await { + Ok(body) => tracing::error!(?body), + Err(body_error) => tracing::error!(?body_error), + }; + continue; + }; + + let response = match upstream_response.json().await { + Ok(response) => response, + Err(json_error) => { + tracing::error!(?json_error); + continue; + } + }; + + return Ok(DecisionsResponse(response)); + } + Err(DecisionsError::NoneSucceeded) +} diff --git a/src/server/embeddings.rs b/src/server/embeddings.rs index 3319ce4..2dda5d6 100644 --- a/src/server/embeddings.rs +++ b/src/server/embeddings.rs @@ -144,6 +144,7 @@ pub async fn create_embeddings( .build() }) .collect(), + Provider::LiquidAi => unreachable!(), Provider::Mistral => state .app_args .mistral_api_key @@ -168,19 +169,7 @@ pub async fn create_embeddings( .build() }) .collect(), - Provider::ZAi => state - .app_args - .z_ai_api_key - .iter() - .cloned() - .map(|api_key| { - Upstream::builder() - // TODO: doesn't exist - .url("https://api.z.ai/api/paas/v4/embeddings".parse().expect("TODO")) - .api_key(api_key) - .build() - }) - .collect(), + Provider::ZAi => unreachable!(), }; if let Ok(upstreams) = Vec1::try_from(upstreams) { diff --git a/src/server/mod.rs b/src/server/mod.rs index f2ceb36..72c4156 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -10,8 +10,10 @@ use typed_builder::TypedBuilder; use url::Url; mod chat; +mod decisions; mod embeddings; mod models; +mod system_one; #[derive(Debug, Clone, TypedBuilder)] pub struct AppState { @@ -24,6 +26,7 @@ enum Provider { GoogleAiStudio, Groq, HuggingFace, + LiquidAi, Mistral, OpenRouter, ZAi, @@ -52,7 +55,9 @@ struct Upstream { pub fn create_router() -> Router { Router::new() .typed_post(chat::completions::create_chat_completion) + .typed_post(decisions::create_decision) .typed_post(embeddings::create_embeddings) + .typed_post(system_one::create_system_one) .typed_get(models::list_models) } diff --git a/src/server/system_one.rs b/src/server/system_one.rs new file mode 100644 index 0000000..28465cd --- /dev/null +++ b/src/server/system_one.rs @@ -0,0 +1,215 @@ +use crate::CloudflareCredentials; +use crate::openai; +use crate::server::AppState; +use crate::server::Provider; +use crate::server::Upstream; +use crate::server::take_random_by_swapping; +use axum::Json; +use axum::extract::State; +use axum::http::StatusCode; +use axum::response::IntoResponse; +use axum::response::Response; +use axum_extra::routing::TypedPath; +use ecow::EcoString; +use headers::Header; +use headers::authorization::Bearer; +use mitsein::vec1::Vec1; +use reqwest::Certificate; +use secrecy::ExposeSecret; +use serde::Deserialize; +use serde_with::DeserializeFromStr; +use serde_with::SerializeDisplay; +use std::collections::VecDeque; +use strum::EnumString; +use strum::VariantArray; + +#[derive(Deserialize, TypedPath)] +#[typed_path("/v1/systemone")] +pub struct V1SystemOnePath; + +#[derive( + Clone, Debug, EnumString, DeserializeFromStr, strum::Display, SerializeDisplay, VariantArray, +)] +#[strum(serialize_all = "lowercase")] +pub enum Shortcut { + M, +} + +impl Shortcut { + fn suggestions(&self) -> Vec<(Provider, EcoString)> { + use Provider as P; + use Shortcut as S; + + match self { + S::M => { + vec![(P::LiquidAi, "d1:free".into())] + } + } + } +} + +pub struct SystemOneResponse(openai::system_one::SystemOneResponse); + +impl IntoResponse for SystemOneResponse { + fn into_response(self) -> Response { + let Self(system_one) = self; + + Json(system_one).into_response() + } +} + +pub enum SystemOneError { + NoneSucceeded, +} + +impl IntoResponse for SystemOneError { + fn into_response(self) -> Response { + match self { + SystemOneError::NoneSucceeded => StatusCode::SERVICE_UNAVAILABLE.into_response(), + } + } +} + +#[tracing::instrument(skip(state, request))] +pub async fn create_system_one( + _: V1SystemOnePath, + State(state): State, + Json(request): Json>, +) -> Result { + let shortcut = request.model.clone(); + let mut model_upstreams = VecDeque::new(); + + for (provider, model) in shortcut.suggestions() { + let upstreams: Vec = + match provider { + Provider::CloudflareWorkersAi => state + .app_args + .cloudflare_workers_ai_credentials + .iter() + .cloned() + .map( + |CloudflareCredentials { + account_id, + api_key, + }| { + Upstream::builder() + .url( + format!("https://api.cloudflare.com/client/v4/accounts/{account_id}/ai/v1/systemone") + .parse() + .expect("TODO"), + ) + .api_key(api_key) + .build() + }, + ) + .collect(), + Provider::GoogleAiStudio => unreachable!(), + Provider::Groq => unreachable!(), + Provider::HuggingFace => unreachable!(), + Provider::LiquidAi => state + .app_args + .liquid_ai_api_key + .iter() + .cloned() + .map(|api_key| { + Upstream::builder() + .url("https://api.liquid.ai/decisions/v1/systemone".parse().expect("TODO")) + .api_key(api_key) + .build() + }) + .collect(), + Provider::Mistral => unreachable!(), + Provider::OpenRouter => state + .app_args + .openrouter_api_key + .iter() + .cloned() + .map(|api_key| { + Upstream::builder() + .url("https://openrouter.ai/api/v1/systemone".parse().expect("TODO")) + .api_key(api_key) + .build() + }) + .collect(), + Provider::ZAi => unreachable!(), + }; + + if let Ok(upstreams) = Vec1::try_from(upstreams) { + model_upstreams.push_back((model, upstreams)); + } + } + + while let Some((model, upstreams)) = model_upstreams.pop_front() { + let (upstream, upstreams) = take_random_by_swapping(upstreams, &mut rand::rng()); + if let Some(upstreams) = upstreams { + model_upstreams.push_back((model.clone(), upstreams)); + } + + tracing::info!(?model, ?upstream); + + let openai::system_one::SystemOneRequest { model: _shortcut, state, questions } = request.clone(); + + let request = openai::system_one::SystemOneRequest { + model, + state, + questions, + }; + + + let client = reqwest::Client::builder() + .tls_certs_only( + webpki_root_certs::TLS_SERVER_ROOT_CERTS + .iter() + .filter_map(|cert| Certificate::from_der(cert).ok()), + ) + .build() + .expect("TODO"); + + let mut upstream_request = client + .post(upstream.url) + .json(&request); + + if let Some(api_key) = upstream.api_key { + let authorization = + headers::Authorization::bearer(api_key.expose_secret()).expect("TODO"); + upstream_request = upstream_request.header(headers::Authorization::::name(), { + let mut vec = vec![]; + + authorization.encode(&mut vec); + + let [one] = vec.try_into().expect("TODO"); + + one + }); + } + + let upstream_response = match upstream_request.send().await { + Ok(upstream_response) => upstream_response, + Err(upstream_error) => { + tracing::error!(?upstream_error); + continue; + } + }; + + if let Err(upstream_error) = upstream_response.error_for_status_ref() { + tracing::error!(?upstream_error); + match upstream_response.text().await { + Ok(body) => tracing::error!(?body), + Err(body_error) => tracing::error!(?body_error), + }; + continue; + }; + + + let response = match upstream_response.json().await { + Ok(response) => response, + Err(json_error) => { + tracing::error!(?json_error); + continue; + } + }; + + return Ok(SystemOneResponse(response)); + } + Err(SystemOneError::NoneSucceeded) +}