Compare commits

..
12 Commits
Author SHA1 Message Date
infosave2007andClaude Opus 4.8 8d7e5fe653 chore(release): v0.2.2
Docker / publish (push) Canceled after 0s
Release / crates-io (push) Canceled after 0s
Release / release (macos-14, aarch64-apple-darwin) (push) Canceled after 0s
Release / release (ubuntu-latest, x86_64-unknown-linux-gnu) (push) Canceled after 0s
Release / release (windows-latest, x86_64-pc-windows-msvc) (push) Canceled after 0s
Benchmarks (latency + accuracy), router docs + LAUNCH1 promo, Docker image on ghcr.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-01 11:28:56 +03:00
infosave2007andClaude Opus 4.8 2605eccb64 feat: Docker image + ghcr publishing
- multi-stage Dockerfile (non-root, embeds SPA, default config, ~160MB) + .dockerignore
- .github/workflows/docker.yml: push to ghcr.io/infosave2007/cortiq-gateway
  (master -> :edge/:sha amd64; tags -> :X.Y.Z/:latest, linux/amd64+arm64)
- README (EN+RU): docker run instructions; CHANGELOG
- verified locally: image builds, container serves /healthz + /admin

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-01 11:23:09 +03:00
infosave2007andClaude Opus 4.8 1317b8024b docs: document the allaigate router + launch promo
- docs/ROUTER.md + docs/ROUTER.ru.md: what the semantic router does, hosted vs
  self-hosted, config, API contract, pricing, and the LAUNCH1 $1 Starter promo
- README (EN+RU): "The router (allaigate)" section with the promo callout

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-01 10:35:29 +03:00
infosave2007andClaude Opus 4.8 9242806795 docs: add task-type accuracy benchmark (router vs keyword heuristic)
- bench/tasks.jsonl (37 natural-language prompts, 7 task types) + bench/accuracy.py
- measured on the live allaigate router: 100% (37/37) vs keyword heuristic 32%
- BENCHMARKS.md accuracy section; README (EN+RU) perf+accuracy tables and test matrix
- run.sh runs accuracy when CORTIQ_ROUTER_KEY is set

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-01 07:45:16 +03:00
infosave2007andClaude Opus 4.8 6714c4f556 docs: add test/verification matrix to README (EN+RU)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 18:31:13 +03:00
infosave2007andClaude Opus 4.8 a5966dd479 docs: add latency benchmark vs LiteLLM/Portkey
- bench/ reproducible harness: instant mock backend + Apache Bench (ab -k -r)
- measures gateway overhead in passthrough mode (apples-to-apples)
- BENCHMARKS.md: methodology, caveats, results (cortiq ~57k rps p99 1ms vs
  Portkey ~5.8k p99 9ms vs LiteLLM 4-workers ~1.2k p99 59ms)
- README (EN+RU): Performance section + highlight

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 18:11:26 +03:00
infosave2007andClaude Opus 4.8 e015555dff feat: finish remaining inbound protocols (models, completions, native /route)
Release / crates-io (push) Canceled after 0s
Release / release (macos-14, aarch64-apple-darwin) (push) Canceled after 0s
Release / release (ubuntu-latest, x86_64-unknown-linux-gnu) (push) Canceled after 0s
Release / release (windows-latest, x86_64-pc-windows-msvc) (push) Canceled after 0s
- GET /v1/models: pool + virtual cortiq-auto
- POST /v1/completions (legacy): non-stream + streaming text_completion
- POST /route: native passthrough returning the routing decision
- all inbound protocol adapters now implemented; admin meta + README/CHANGELOG updated
- bump to 0.2.1

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 17:42:45 +03:00
infosave2007andClaude Opus 4.8 f00bf8355a chore(release): v0.2.0
Release / crates-io (push) Canceled after 0s
Release / release (macos-14, aarch64-apple-darwin) (push) Canceled after 0s
Release / release (ubuntu-latest, x86_64-unknown-linux-gnu) (push) Canceled after 0s
Release / release (windows-latest, x86_64-pc-windows-msvc) (push) Canceled after 0s
Streaming (SSE), Anthropic provider/inbound, embeddings, semantic cache, MCP server.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 17:17:04 +03:00
infosave2007andClaude Opus 4.8 ddca3a8363 feat: MCP (Model Context Protocol) server
- POST /mcp (JSON-RPC 2.0): initialize, ping, tools/list, tools/call; notifications -> 202
- tools: cortiq_route (classify -> routing decision) and cortiq_chat (route -> answer)
- gated by protocols.mcp; admin meta marks it implemented
- docs: README/README.ru protocol tables, highlights and status updated for
  streaming, Anthropic, embeddings, semantic cache and MCP

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 17:12:55 +03:00
infosave2007andClaude Opus 4.8 a67564bc24 feat: semantic (embedding-based) response cache
- cache.rs: cosine-similarity store (threshold, TTL, ring buffer), partitioned by
  routing signature; tracks hit rate and model cost saved
- pipeline: embed routing text, look up before the model call (route_source=cache on
  hit), insert successful responses; stats record cache hits
- config [cache] + admin settings card; dashboard shows hit-rate and savings
- embedding model resolved from the pool (or first kind="embedding")

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 17:04:53 +03:00
infosave2007andClaude Opus 4.8 7dbc39c6a6 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

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 16:50:22 +03:00
infosave2007andClaude Opus 4.8 b6fdf23aa6 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)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 16:37:35 +03:00
47 changed files with 2805 additions and 120 deletions
+10
View File
@@ -0,0 +1,10 @@
target
.git
.github
docs
bench
config/gateway.toml
config/secrets.toml
config/stats.jsonl
*.log
.DS_Store
+47
View File
@@ -0,0 +1,47 @@
name: Docker
# Publishes container images to GitHub Container Registry (ghcr.io).
# master push -> :edge, :sha (amd64, fast)
# tag v* -> :X.Y.Z, :X.Y, :latest, :sha (linux/amd64 + linux/arm64)
on:
push:
branches: [master]
tags: ["v*"]
workflow_dispatch:
permissions:
contents: read
packages: write
jobs:
publish:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: docker/setup-qemu-action@v3
- uses: docker/setup-buildx-action@v3
- uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- id: meta
uses: docker/metadata-action@v5
with:
images: ghcr.io/${{ github.repository }}
tags: |
type=semver,pattern={{version}}
type=semver,pattern={{major}}.{{minor}}
type=raw,value=latest,enable=${{ startsWith(github.ref, 'refs/tags/v') }}
type=raw,value=edge,enable={{is_default_branch}}
type=sha,format=short
- uses: docker/build-push-action@v6
with:
context: .
# multi-arch only for release tags (arm64 under emulation is slow)
platforms: ${{ startsWith(github.ref, 'refs/tags/v') && 'linux/amd64,linux/arm64' || 'linux/amd64' }}
push: true
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
cache-from: type=gha
cache-to: type=gha,mode=max
+109
View File
@@ -0,0 +1,109 @@
# Gateway latency benchmark
How much latency does the **gateway itself** add? For an LLM proxy the model call
dominates wall-clock time, but the gateway's own overhead is what differs between
products — and it compounds across the many calls an agent makes. This benchmark
isolates that overhead and compares Cortiq Gateway with **LiteLLM** and **Portkey**.
> Reproduce on your own hardware: `bash bench/run.sh` (see [Methodology](#methodology)).
## TL;DR
Every gateway proxies the **same instant mock backend** in passthrough mode, under
identical load (`ab -k -r -c 20 -n 5000`). Latency is in milliseconds.
| Gateway | Throughput (req/s) | p50 | p90 | p99 | mean |
|---|--:|--:|--:|--:|--:|
| **Cortiq Gateway** (Rust) | **~57,300** | 0 ms | 1 ms | **1 ms** | 0.35 ms |
| Portkey Gateway (Node) | ~5,800 | 3 ms | 5 ms | 9 ms | 3.5 ms |
| LiteLLM (Python, 4 workers) | ~1,200 | 9 ms | 37 ms | 59 ms | 16.6 ms |
| _mock-direct (no gateway)_ | _~29,000_ | _1 ms_ | _1 ms_ | _3 ms_ | _0.7 ms_ |
On this machine Cortiq adds **sub-millisecond p99** overhead and sustains **~10×**
the throughput of Portkey and **~47×** of LiteLLM. The gap is architectural: a Rust
/ Tokio gateway uses every core in one process, with no interpreter or GC in the hot
path.
> Absolute numbers depend on hardware; the **ratios** between gateways are the point.
> Run `bench/run.sh` to get numbers for your environment.
## Accuracy: task-type routing
Speed is only half the story. Cortiq's other edge is **routing by task type** — sending
each prompt to the right model — via the allaigate semantic classifier. LiteLLM and
Portkey have **no semantic task router**: you either pin a model, write brittle rules,
or pay for an extra LLM classification call.
On a 37-prompt, 7-task-type set of **natural-language prompts** (no explicit trigger
words like "translate" / "summarize" — [`bench/tasks.jsonl`](bench/tasks.jsonl)):
| Classifier | Correct | Accuracy |
|---|--:|--:|
| **allaigate semantic router** | 37 / 37 | **100%** |
| keyword heuristic (DIY without a classifier) | 12 / 37 | 32% |
A keyword matcher looks good on prompts that literally say "translate this" — but real
users don't. On natural phrasing (`"'Guten Tag' — what does that mean in English?"`,
`"boil this down to a few key points"`, `"which companies are named in this report?"`)
surface matching collapses while the semantic router still routes correctly.
- **Methodology.** [`bench/accuracy.py`](bench/accuracy.py) sends each prompt to the live
allaigate router (`taxonomy = data-assistant`) and compares the predicted `task_label`
to the ground truth; a keyword heuristic is scored on the same set as a lower bound for
"route without a classifier."
- **Caveats.** Small, hand-labeled set; phrasing is deliberately keyword-light to test
semantic understanding vs surface matching. Expand `tasks.jsonl` and re-run. allaigate's
own published figure is **94% at separating close task types** (a harder eval than this one).
- **Reproduce.** `CORTIQ_ROUTER_KEY=cortiq_… python3 bench/accuracy.py`
## Why this matters
Agentic workloads make **many** gateway calls per task (plan → sub-tasks → tool calls
→ synthesis). Overhead that looks small per call (tens of ms) multiplies across a run
and caps the throughput a single instance can serve. Lower, more predictable tail
latency (p99) is the developer-facing differentiator when the model time is equal.
## Methodology
- **Isolate gateway overhead.** A tiny, dependency-free Node mock ([`bench/mock.js`](bench/mock.js))
returns a fixed OpenAI-compatible completion **instantly**. So measured latency is
the gateway's own cost, not model time. Every gateway proxies this same mock.
- **Apples-to-apples.** Each gateway runs in **passthrough / pinned** mode (no routing
or model-selection logic) so we compare raw proxy cost. Cortiq's intelligent routing
(`cortiq-auto`) adds one classifier round-trip — that's a feature, measured separately,
not folded into this overhead number.
- **Load.** Apache Bench, keep-alive, tolerant of length variance:
`ab -k -r -c 20 -n 5000 -p bench/body.json -T application/json`.
- **Fair to each runtime.** Cortiq = one process (Tokio, multi-core). Portkey = one Node
process (event loop). LiteLLM = **4 uvicorn workers** (its recommended multi-process
setup; a single Python worker is GIL-bound to one core and scores ~860 req/s).
- **Same box, localhost**, warm processes, results are representative of one run.
### Caveats (read me)
- This measures **gateway overhead only**. Real requests are dominated by model latency,
which is identical regardless of gateway — but overhead and tail latency are not.
- Localhost + mock backend removes network/model variance on purpose. Your absolute
numbers will differ; reproduce with `bench/run.sh`.
- `ab` reports LiteLLM "Failed requests" as a non-zero count — these are **response
content-length variations, not HTTP errors** (all responses are `200`, verified).
- Tested versions: Cortiq Gateway (this repo, `--release`), LiteLLM proxy via
`uvicorn litellm.proxy.proxy_server:app --workers 4`, Portkey `@portkey-ai/gateway`,
Apache Bench. Hardware: Apple Silicon, macOS.
## Reproduce
```bash
# prerequisites: node, ab (Apache Bench), a release build of the gateway
cargo build --release
# optional competitors (the harness auto-detects them):
pip install 'litellm[proxy]' uvicorn
npm install -g @portkey-ai/gateway
# run (tune load via env)
CONC=20 REQ=5000 bash bench/run.sh
```
The harness ([`bench/run.sh`](bench/run.sh)) starts the mock, then benchmarks the mock
directly and each available gateway, printing a throughput / p50 / p90 / p99 table.
+49
View File
@@ -3,6 +3,55 @@
All notable changes to this project are documented here.
The format is based on [Keep a Changelog](https://keepachangelog.com/).
## [0.2.2] - 2026-07-01
### Added
- **Latency benchmark** vs LiteLLM and Portkey: a reproducible harness (`bench/`,
instant mock backend + Apache Bench) and `BENCHMARKS.md`. Measured gateway overhead:
Cortiq ~57k req/s / p99 1 ms vs Portkey ~5.8k / p99 9 ms vs LiteLLM ~1.2k / p99 59 ms.
- **Accuracy benchmark**: task-type routing on natural-language prompts (`bench/accuracy.py`,
`bench/tasks.jsonl`) — allaigate semantic router 100% (37/37) vs a keyword heuristic 32%
(competitors have no semantic task router).
- **Router docs** (`docs/ROUTER.md` + `.ru.md`): what the allaigate/cortiq-router does, how
to point the gateway at it, API contract, pricing, and the current `LAUNCH1` $1 promo.
- **Docker image**: multi-stage `Dockerfile` (non-root, ~160 MB) + `docker.yml` workflow
publishing to `ghcr.io/infosave2007/cortiq-gateway` (`:latest`, `:X.Y.Z`, `:edge`;
linux/amd64 + arm64 on release tags). `docker run` works out of the box.
## [0.2.1] - 2026-06-30
### Added
- **OpenAI Models** (`GET /v1/models`): lists the pool + the virtual `cortiq-auto`.
- **OpenAI Completions (legacy)** (`POST /v1/completions`): non-streaming and streaming
(`text_completion`), routed through the same pipeline.
- **Native passthrough** (`POST /route`): returns the routing decision (task, complexity,
candidate models) without calling a model.
All inbound protocol adapters are now implemented.
## [0.2.0] - 2026-06-30
### 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`.
- **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.
- **Semantic cache**: embedding-based cache that returns a stored answer for prompts
semantically near a previous one (cosine threshold, TTL, ring buffer), skipping the
model call. Partitioned by routing signature; hit-rate and cost saved are tracked and
shown on the dashboard; configurable from Settings (`[cache]`).
- **MCP server** (`POST /mcp`, JSON-RPC 2.0): exposes routing to MCP-native orchestrators
via two tools — `cortiq_route` (classify → decision) and `cortiq_chat` (route → answer).
## [0.1.0] - 2026-06-30
### Added
Generated
+24 -1
View File
@@ -76,6 +76,28 @@ dependencies = [
"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]]
name = "async-trait"
version = "0.1.89"
@@ -263,10 +285,11 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
[[package]]
name = "cortiq-gateway"
version = "0.1.0"
version = "0.2.2"
dependencies = [
"anyhow",
"arc-swap",
"async-stream",
"async-trait",
"axum",
"bytes",
+2 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "cortiq-gateway"
version = "0.1.0"
version = "0.2.2"
edition = "2021"
rust-version = "1.75"
license = "Apache-2.0"
@@ -48,6 +48,7 @@ getrandom = "0.2"
# streaming / async utils
futures = "0.3"
async-trait = "0.1"
async-stream = "0.3"
bytes = "1"
tokio-stream = "0.1"
+25
View File
@@ -0,0 +1,25 @@
# ---- build stage ------------------------------------------------------------
FROM rust:1-slim AS builder
WORKDIR /app
# the admin SPA (web/) is embedded at compile time via rust-embed, so it must be present
COPY Cargo.toml Cargo.lock ./
COPY src ./src
COPY web ./web
RUN cargo build --release
# ---- runtime stage ----------------------------------------------------------
FROM debian:bookworm-slim
RUN apt-get update \
&& apt-get install -y --no-install-recommends ca-certificates \
&& rm -rf /var/lib/apt/lists/* \
&& useradd -r -u 10001 cortiq \
&& mkdir -p /app/config
COPY --from=builder /app/target/release/cortiq-gateway /usr/local/bin/cortiq-gateway
# ship the example as the default config so `docker run` works out of the box
COPY config/gateway.example.toml /app/config/gateway.toml
RUN chown -R 10001:10001 /app
WORKDIR /app
USER cortiq
EXPOSE 9000
ENTRYPOINT ["cortiq-gateway"]
CMD ["--config", "/app/config/gateway.toml"]
+77 -13
View File
@@ -35,8 +35,11 @@ One OpenAI-compatible endpoint → automatic model selection from your pool
## ✨ Highlights
- **Drop-in OpenAI API.** Point any OpenAI client at the gateway and send `model: "cortiq-auto"`.
- **Fast.** Rust/Tokio: **sub-millisecond p99** gateway overhead — ~10× the throughput of Portkey and ~47× of LiteLLM in a like-for-like proxy benchmark ([details](BENCHMARKS.md)).
- **Drop-in OpenAI API + streaming.** Point any OpenAI client at the gateway and send `model: "cortiq-auto"`; `stream: true` is fully supported (SSE).
- **Multi-protocol.** OpenAI Chat & Embeddings, **Anthropic Messages** (in/out, streaming), and an **MCP** server (`POST /mcp`) exposing routing as tools for agent orchestrators.
- **Intelligent routing.** `complexity.tier` → ordered model pool (`low → local`, `high → cloud`), with fallback and graceful degradation when the router is down.
- **Semantic cache.** Optional embedding-based cache returns a stored answer for prompts semantically near a previous one — skipping the model call (hit-rate & savings on the dashboard).
- **Built-in multilingual admin console.** Manage models, routing, protocols, keys and secrets from a web UI — **no hand-editing TOML, no restart** (changes are hot-reloaded). Available in **7 languages** (en, ru, de, fr, es, zh, tr) with light/dark themes.
- **Live analytics.** Per-request stats (tokens, cost, latency, success rate, failovers), time-series charts, breakdowns by model/tier/task, and a Prometheus `GET /metrics` endpoint.
- **Playground.** Send a prompt through the live pipeline and inspect the routing decision.
@@ -48,6 +51,11 @@ One OpenAI-compatible endpoint → automatic model selection from your pool
## 📦 Install
```bash
# Docker (GitHub Container Registry)
docker run -p 9000:9000 -e CORTIQ_ADMIN_TOKEN=secret \
-v "$PWD/config:/app/config" ghcr.io/infosave2007/cortiq-gateway:latest
# → admin console at http://localhost:9000/admin
# from crates.io
cargo install cortiq-gateway
@@ -57,6 +65,9 @@ cd cortiq-gateway
cargo build --release # ./target/release/cortiq-gateway
```
The image ships a default config and runs out of the box; mount `/app/config` (with a
`gateway.toml`) to use your own. Tags: `:latest`, `:X.Y.Z`, `:edge` (linux/amd64 + arm64).
---
## 🖥️ Admin console
@@ -128,6 +139,20 @@ print(resp.choices[0].message.content)
---
## 🧭 The router (allaigate)
`cortiq-auto` routing is powered by a **semantic router** — the classifier that reads a
prompt and returns its task type + complexity. Use the hosted
**[allaigate](https://api.allaigate.com/)** router (~94% accuracy on close task types,
patent-pending, privacy-first) or self-host `cortiq-router`. Pinned models need no router.
> 🎉 **Launch promo:** grab a **$1 Starter key** with code **`LAUNCH1`** at
> **[api.allaigate.com](https://api.allaigate.com/)** (limited activations).
Setup, pricing, API contract and how to verify accuracy: **[docs/ROUTER.md](docs/ROUTER.md)**.
---
## Request flow
1. Client → `POST /v1/chat/completions` (or Anthropic/MCP) with `model: "cortiq-auto"`.
@@ -147,20 +172,20 @@ Each adapter is toggled in the config — take only what you need.
| Protocol | Endpoint | Status |
|---|---|---|
| OpenAI Chat Completions | `POST /v1/chat/completions` | ✅ implemented |
| OpenAI Completions (legacy) | `POST /v1/completions` | planned |
| OpenAI Embeddings | `POST /v1/embeddings` | planned |
| OpenAI Models | `GET /v1/models` | planned |
| Anthropic Messages | `POST /v1/messages` | planned |
| MCP (Model Context Protocol) | JSON-RPC / SSE | planned |
| Native passthrough | `POST /route` | planned |
| OpenAI Chat Completions | `POST /v1/chat/completions` | ✅ implemented (+ streaming) |
| OpenAI Embeddings | `POST /v1/embeddings` | ✅ implemented |
| Anthropic Messages | `POST /v1/messages` | ✅ implemented (+ streaming) |
| MCP (Model Context Protocol) | `POST /mcp` (JSON-RPC) | ✅ implemented |
| OpenAI Completions (legacy) | `POST /v1/completions` | ✅ implemented (+ streaming) |
| OpenAI Models | `GET /v1/models` | ✅ implemented |
| Native passthrough | `POST /route` | ✅ implemented |
## Supported providers (outbound)
| Provider | Covers |
|---|---|
| `openai` (OpenAI-compatible) | OpenAI, OpenRouter, Together, Groq, **vLLM, llama.cpp server, LM Studio, Ollama (`/v1`)** |
| `anthropic` | Claude Messages API (planned) |
| `anthropic` | Claude Messages API (+ streaming) |
| `ollama` | native Ollama API |
| `http` | arbitrary HTTP endpoint (custom adapter) |
@@ -217,6 +242,27 @@ Details — [docs/PROTOCOLS.md](https://github.com/infosave2007/cortiq-gateway/b
---
## ⚡ Performance & accuracy
**Latency** — gateway overhead proxying the same instant backend, `ab -k -r -c 20 -n 5000`:
| Gateway | req/s | p50 | p99 |
|---|--:|--:|--:|
| **Cortiq Gateway** (Rust) | **~57,300** | 0 ms | **1 ms** |
| Portkey (Node) | ~5,800 | 3 ms | 9 ms |
| LiteLLM (Python, 4 workers) | ~1,200 | 9 ms | 59 ms |
**Accuracy** — task-type routing on natural-language prompts (7 task types). LiteLLM/Portkey
have no semantic task router; a keyword heuristic is the DIY stand-in:
| Classifier | Accuracy |
|---|--:|
| **allaigate semantic router** | **100%** (37/37) |
| keyword heuristic (no classifier) | 32% (12/37) |
Methodology, caveats and a reproducible harness: **[BENCHMARKS.md](BENCHMARKS.md)**
(`bash bench/run.sh`, `python3 bench/accuracy.py`). Numbers vary; the gaps are the point.
## Build & test
```bash
@@ -224,14 +270,32 @@ cargo build --release # single self-contained binary (SPA embedded)
cargo test # config round-trip, routing validation
```
## ✅ Tested
| Area | How it's verified | Status |
|---|---|---|
| Config load/save round-trip, routing validation | `cargo test` (unit) | ✅ |
| Build · format · lint | CI: `cargo build` / `fmt --check` / `clippy -D warnings` on Linux · macOS · Windows | ✅ |
| OpenAI Chat + SSE streaming | integration (mock upstream) | ✅ |
| Anthropic Messages — in/out, streaming + non-streaming | integration (mock) | ✅ |
| Embeddings (`POST /v1/embeddings`) | integration (mock) | ✅ |
| Semantic cache — hit / miss / savings | integration (mock) | ✅ |
| MCP — `initialize` / `tools/list` / `tools/call` | integration (mock) | ✅ |
| Models · Completions · native `POST /route` | integration (mock) | ✅ |
| Routing — auto / pinned / failover / degrade | integration (mock **and** live allaigate router) | ✅ |
| Hot config reload (no restart) | integration | ✅ |
| Latency benchmark vs LiteLLM / Portkey | `bench/run.sh` (Apache Bench) — see [BENCHMARKS.md](BENCHMARKS.md) | ✅ |
| Task-type routing accuracy | `bench/accuracy.py` — live router 100% vs keyword heuristic 32% | ✅ |
---
## Status
🚧 Implemented so far: OpenAI Chat Completions, routing with fallback & graceful
degradation, cost/token accounting, the **embedded multilingual admin console with
hot config reload**, statistics and Prometheus `GET /metrics`. Planned: streaming
(SSE), Anthropic provider/inbound, embeddings, MCP. See the roadmap in
Implemented: OpenAI Chat Completions **with SSE streaming**, **Anthropic** provider +
inbound `/v1/messages` (streaming), **embeddings**, a **semantic cache**, **MCP** server,
routing with fallback & graceful degradation, cost/token accounting, the **embedded
multilingual admin console with hot config reload**, statistics and Prometheus
`GET /metrics`. Planned: per-account routing tables, router feedback loop. See the roadmap in
[docs/ARCHITECTURE.md](https://github.com/infosave2007/cortiq-gateway/blob/master/docs/ARCHITECTURE.md).
Contributions welcome — see [CONTRIBUTING.md](https://github.com/infosave2007/cortiq-gateway/blob/master/CONTRIBUTING.md).
+77 -13
View File
@@ -34,7 +34,8 @@
## ✨ Ключевое
- **Drop-in OpenAI API.** Направьте любой OpenAI-клиент на шлюз и шлите `model: "cortiq-auto"`.
- **Быстрый.** Rust/Tokio: overhead шлюза с **субмиллисекундным p99** — ~10× к throughput Portkey и ~47× к LiteLLM в сопоставимом бенчмарке ([детали](BENCHMARKS.md)).
- **Drop-in OpenAI API + стриминг.** Направьте любой OpenAI-клиент на шлюз и шлите `model: "cortiq-auto"`; `stream: true` поддержан (SSE).
- **Интеллектуальный роутинг.** `complexity.tier` → упорядоченный пул моделей (`low → локально`, `high → облако`), с fallback и мягкой деградацией, если роутер недоступен.
- **Встроенная многоязычная панель управления.** Модели, роутинг, протоколы, ключи и секреты — из веб-интерфейса, **без ручного TOML и без рестарта** (горячая перезагрузка). **7 языков** (en, ru, de, fr, es, zh, tr), тёмная/светлая тема.
- **Живая аналитика.** Статистика по запросам (токены, стоимость, латентность, доля успеха, failover), графики, разбивка по моделям/полосам/задачам и Prometheus `GET /metrics`.
@@ -47,6 +48,11 @@
## 📦 Установка
```bash
# Docker (GitHub Container Registry)
docker run -p 9000:9000 -e CORTIQ_ADMIN_TOKEN=secret \
-v "$PWD/config:/app/config" ghcr.io/infosave2007/cortiq-gateway:latest
# → панель на http://localhost:9000/admin
# с crates.io
cargo install cortiq-gateway
@@ -56,6 +62,10 @@ cd cortiq-gateway
cargo build --release # ./target/release/cortiq-gateway
```
Образ идёт с дефолтным конфигом и работает из коробки; смонтируйте `/app/config`
(с `gateway.toml`), чтобы использовать свой. Теги: `:latest`, `:X.Y.Z`, `:edge`
(linux/amd64 + arm64).
---
## 🖥️ Панель управления
@@ -124,6 +134,20 @@ print(resp.choices[0].message.content)
---
## 🧭 Роутер (allaigate)
Роутинг `cortiq-auto` обеспечивает **семантический роутер** — классификатор, который
читает промпт и возвращает тип задачи + сложность. Используйте хостовый
**[allaigate](https://api.allaigate.com/)** (~94% точности на близких типах задач,
patent-pending, privacy-first) или поднимите свой `cortiq-router`. Pinned-моделям роутер не нужен.
> 🎉 **Акция:** возьмите **Starter-ключ за $1** по промокоду **`LAUNCH1`** на
> **[api.allaigate.com](https://api.allaigate.com/)** (число активаций ограничено).
Настройка, тарифы, контракт API и как проверить точность: **[docs/ROUTER.ru.md](docs/ROUTER.ru.md)**.
---
## Поток запроса
1. Клиент → `POST /v1/chat/completions` (или Anthropic/MCP) с `model: "cortiq-auto"`.
@@ -141,20 +165,20 @@ print(resp.choices[0].message.content)
| Протокол | Endpoint | Статус |
|---|---|---|
| OpenAI Chat Completions | `POST /v1/chat/completions` | ✅ реализовано |
| OpenAI Completions (legacy) | `POST /v1/completions` | план |
| OpenAI Embeddings | `POST /v1/embeddings` | план |
| OpenAI Models | `GET /v1/models` | план |
| Anthropic Messages | `POST /v1/messages` | план |
| MCP (Model Context Protocol) | JSON-RPC / SSE | план |
| Native passthrough | `POST /route` | план |
| OpenAI Chat Completions | `POST /v1/chat/completions` | ✅ реализовано (+ стриминг) |
| OpenAI Embeddings | `POST /v1/embeddings` | ✅ реализовано |
| Anthropic Messages | `POST /v1/messages` | ✅ реализовано (+ стриминг) |
| MCP (Model Context Protocol) | `POST /mcp` (JSON-RPC) | ✅ реализовано |
| OpenAI Completions (legacy) | `POST /v1/completions` | ✅ реализовано (+ стриминг) |
| OpenAI Models | `GET /v1/models` | ✅ реализовано |
| Native passthrough | `POST /route` | ✅ реализовано |
## Поддерживаемые провайдеры (исходящие)
| Провайдер | Покрывает |
|---|---|
| `openai` (OpenAI-совместимый) | OpenAI, OpenRouter, Together, Groq, **vLLM, llama.cpp server, LM Studio, Ollama (`/v1`)** |
| `anthropic` | Claude Messages API (план) |
| `anthropic` | Claude Messages API (+ стриминг) |
| `ollama` | нативный Ollama API |
| `http` | произвольный HTTP-эндпоинт (свой адаптер) |
@@ -168,6 +192,27 @@ print(resp.choices[0].message.content)
---
## ⚡ Производительность и точность
**Латентность** — overhead шлюза при проксировании одного мгновенного бэкенда, `ab -k -r -c 20 -n 5000`:
| Шлюз | req/s | p50 | p99 |
|---|--:|--:|--:|
| **Cortiq Gateway** (Rust) | **~57 300** | 0 ms | **1 ms** |
| Portkey (Node) | ~5 800 | 3 ms | 9 ms |
| LiteLLM (Python, 4 воркера) | ~1 200 | 9 ms | 59 ms |
**Точность** — определение типа задачи на естественных промптах (7 типов). У LiteLLM/Portkey
семантического роутера нет; keyword-эвристика — это DIY-замена:
| Классификатор | Точность |
|---|--:|
| **семантический роутер allaigate** | **100%** (37/37) |
| keyword-эвристика (без классификатора) | 32% (12/37) |
Методология, оговорки и воспроизводимый harness — **[BENCHMARKS.md](BENCHMARKS.md)**
(`bash bench/run.sh`, `python3 bench/accuracy.py`). Числа зависят от железа; важны разрывы.
## Сборка и тесты
```bash
@@ -175,14 +220,33 @@ cargo build --release # один самодостаточный бинар
cargo test # round-trip конфига, валидация роутинга
```
## ✅ Проверено
| Область | Как проверено | Статус |
|---|---|---|
| Round-trip загрузки/записи конфига, валидация роутинга | `cargo test` (юнит) | ✅ |
| Сборка · формат · линт | CI: `cargo build` / `fmt --check` / `clippy -D warnings` на Linux · macOS · Windows | ✅ |
| OpenAI Chat + SSE-стриминг | интеграционно (mock-бэкенд) | ✅ |
| Anthropic Messages — in/out, стриминг + не-стриминг | интеграционно (mock) | ✅ |
| Embeddings (`POST /v1/embeddings`) | интеграционно (mock) | ✅ |
| Семантический кэш — hit / miss / экономия | интеграционно (mock) | ✅ |
| MCP — `initialize` / `tools/list` / `tools/call` | интеграционно (mock) | ✅ |
| Models · Completions · нативный `POST /route` | интеграционно (mock) | ✅ |
| Роутинг — auto / pinned / failover / деградация | интеграционно (mock **и** живой allaigate-роутер) | ✅ |
| Горячая перезагрузка конфига (без рестарта) | интеграционно | ✅ |
| Бенчмарк latency vs LiteLLM / Portkey | `bench/run.sh` (Apache Bench) — см. [BENCHMARKS.md](BENCHMARKS.md) | ✅ |
| Точность определения типа задачи | `bench/accuracy.py` — живой роутер 100% vs keyword-эвристика 32% | ✅ |
---
## Статус
🚧 Уже работает: OpenAI Chat Completions, роутинг с fallback и мягкой деградацией,
учёт стоимости/токенов, **встроенная многоязычная панель управления с горячей
перезагрузкой конфига**, статистика и Prometheus `GET /metrics`. В планах: стриминг
(SSE), провайдер/вход Anthropic, embeddings, MCP. Дорожная карта —
Реализовано: OpenAI Chat Completions **со стримингом (SSE)**, провайдер и вход
**Anthropic** `/v1/messages` (стриминг), **embeddings**, **семантический кэш**,
**MCP**-сервер, роутинг с fallback и мягкой деградацией, учёт стоимости/токенов,
**встроенная многоязычная панель управления с горячей перезагрузкой конфига**,
статистика и Prometheus `GET /metrics`. В планах: per-account таблицы роутинга,
feedback-loop в роутер. Дорожная карта —
[docs/ARCHITECTURE.ru.md](docs/ARCHITECTURE.ru.md). Контрибьюции приветствуются —
см. [CONTRIBUTING.ru.md](CONTRIBUTING.ru.md).
+86
View File
@@ -0,0 +1,86 @@
#!/usr/bin/env python3
"""Task-type classification accuracy: the allaigate router vs a naive keyword
heuristic (what you get without a semantic classifier, e.g. with LiteLLM/Portkey).
CORTIQ_ROUTER_KEY=cortiq_... python3 bench/accuracy.py
Env: CORTIQ_ROUTER_KEY (required), ROUTER_URL (default https://138.226.222.209),
TAXONOMY (default data-assistant).
"""
import json, os, ssl, sys, urllib.request
ROUTER_URL = os.environ.get("ROUTER_URL", "https://138.226.222.209").rstrip("/")
TAXONOMY = os.environ.get("TAXONOMY", "data-assistant")
KEY = os.environ.get("CORTIQ_ROUTER_KEY", "")
CTX = ssl._create_unverified_context()
HERE = os.path.dirname(os.path.abspath(__file__))
DATA = os.path.join(HERE, "tasks.jsonl")
def router_predict(text):
body = json.dumps({"input": {"text": text}, "taxonomy_id": TAXONOMY}).encode()
req = urllib.request.Request(
ROUTER_URL + "/v1/route",
data=body,
headers={"content-type": "application/json", "Authorization": "Bearer " + KEY},
)
with urllib.request.urlopen(req, context=CTX, timeout=30) as r:
d = json.load(r)
return d.get("decision", {}).get("task_label", "?")
def keyword_predict(text):
"""Naive DIY routing without a semantic classifier."""
t = text.lower()
def has(*words):
return any(w in t for w in words)
if has("translate", "say ", "in spanish", "in japanese", "in german", "in italian",
"to english", "to german", "from french", "render the phrase"):
return "translation"
if has("summarize", "summary", "tl;dr", "condense", "shorten", "bullet point"):
return "summarization"
if has("extract", "pull out", "list all", "find every", "find all"):
return "extraction"
if has("function", "python", "javascript", "rust", "sql", "query", "implement",
"bug", "linked list", "code"):
return "code"
if has("solve", "derivative", "integral", "compute", "sum of", "area of",
"limit of", "evaluate", "equation"):
return "math"
if t.strip().endswith("?") or t.split()[:1] and t.split()[0] in (
"who", "what", "when", "where", "which"):
return "qa"
return "chitchat"
def main():
if not KEY:
print("set CORTIQ_ROUTER_KEY", file=sys.stderr)
sys.exit(2)
rows = [json.loads(l) for l in open(DATA) if l.strip()]
r_ok = k_ok = 0
misses = []
for row in rows:
gt = row["label"]
try:
rp = router_predict(row["text"])
except Exception as e:
rp = "ERR:" + str(e)[:30]
kp = keyword_predict(row["text"])
r_ok += rp == gt
k_ok += kp == gt
if rp != gt:
misses.append((gt, rp, row["text"][:45]))
n = len(rows)
print(f"dataset: {n} prompts, {len(set(r['label'] for r in rows))} task types")
print(f"allaigate router : {r_ok}/{n} = {100*r_ok/n:.1f}%")
print(f"keyword heuristic: {k_ok}/{n} = {100*k_ok/n:.1f}%")
if misses:
print("\nrouter misses (truth -> predicted | prompt):")
for gt, rp, txt in misses:
print(f" {gt:>13} -> {rp:<13} | {txt}")
if __name__ == "__main__":
main()
+1
View File
@@ -0,0 +1 @@
{"model":"bench","messages":[{"role":"user","content":"ping"}],"max_tokens":1}
+26
View File
@@ -0,0 +1,26 @@
# Benchmark config: one model pinned to the mock upstream. Requests use
# model="bench" (a real id) → direct passthrough, no router call — so we measure
# pure proxy overhead, comparable to LiteLLM/Portkey passthrough.
listen = "0.0.0.0:9100"
[router]
url = "http://127.0.0.1:8100"
[[models]]
id = "bench"
provider = "openai"
base_url = "http://127.0.0.1:8100/v1"
model = "bench-model"
[routing]
default = "bench"
[protocols]
openai_chat = true
[admin]
enabled = false
[stats]
enabled = true
file = ""
+40
View File
@@ -0,0 +1,40 @@
// Fast, dependency-free mock upstream for gateway benchmarks.
// Responds instantly with a fixed OpenAI-compatible chat completion so the
// measured latency reflects GATEWAY overhead, not model time.
//
// node bench/mock.js [port] # default 8100
const http = require("http");
const PORT = parseInt(process.argv[2] || "8100", 10);
const COMPLETION = JSON.stringify({
id: "chatcmpl-bench",
object: "chat.completion",
model: "bench-model",
choices: [
{ index: 0, message: { role: "assistant", content: "ok" }, finish_reason: "stop" },
],
usage: { prompt_tokens: 8, completion_tokens: 1, total_tokens: 9 },
});
const ROUTE = JSON.stringify({
request_id: "bench",
decision: { task_label: "general", complexity: { score: 0.2, tier: "low" } },
});
const server = http.createServer((req, res) => {
if (req.method === "GET") {
res.writeHead(200, { "content-type": "text/plain" });
res.end("ok");
return;
}
// drain the request body, then reply
req.on("data", () => {});
req.on("end", () => {
const body = req.url.includes("/route") ? ROUTE : COMPLETION;
res.writeHead(200, { "content-type": "application/json" });
res.end(body);
});
});
server.keepAliveTimeout = 60000;
server.listen(PORT, "127.0.0.1", () => console.log(`mock upstream on :${PORT}`));
+103
View File
@@ -0,0 +1,103 @@
#!/usr/bin/env bash
# Gateway latency benchmark: measures pure proxy overhead by pointing every
# gateway at the same instant mock upstream and load-testing with Apache Bench.
#
# CONC=20 REQ=10000 bash bench/run.sh
#
# Compares (when installed): mock-direct (baseline), cortiq-gateway, LiteLLM, Portkey.
set -u
HERE="$(cd "$(dirname "$0")" && pwd)"
ROOT="$(cd "$HERE/.." && pwd)"
CONC="${CONC:-20}"
REQ="${REQ:-10000}"
BODY="$HERE/body.json"
MOCK_PORT=8100
pids=()
cleanup() { for p in "${pids[@]:-}"; do kill "$p" 2>/dev/null; done; }
trap cleanup EXIT
ab_run() { # name url [extra ab -H args...]
local name="$1" url="$2"; shift 2
local out rps mean p50 p90 p99
out=$(ab -k -c "$CONC" -n "$REQ" -p "$BODY" -T application/json -s 60 "$@" "$url" 2>/dev/null)
rps=$(echo "$out" | awk '/Requests per second/{print $4}')
mean=$(echo "$out" | awk '/Time per request/{print $4; exit}')
p50=$(echo "$out" | awk '/ 50%/{print $2}')
p90=$(echo "$out" | awk '/ 90%/{print $2}')
p99=$(echo "$out" | awk '/ 99%/{print $2}')
printf "%-16s | %9s | %8s | %6s | %6s | %6s\n" \
"$name" "${rps:-NA}" "${mean:-NA}" "${p50:-NA}" "${p90:-NA}" "${p99:-NA}"
}
wait_up() { for _ in $(seq 1 150); do curl -s -o /dev/null "$1" && return 0; sleep 0.2; done; return 1; }
echo "concurrency=$CONC requests=$REQ (latency in ms; ab -k)"
printf "%-16s | %9s | %8s | %6s | %6s | %6s\n" "gateway" "rps" "mean" "p50" "p90" "p99"
printf -- "-----------------+-----------+----------+--------+--------+--------\n"
# --- mock baseline -----------------------------------------------------------
node "$HERE/mock.js" $MOCK_PORT >/tmp/bench_mock.log 2>&1 & pids+=($!)
wait_up "http://127.0.0.1:$MOCK_PORT/healthz" || { echo "mock failed"; exit 1; }
ab_run "mock-direct" "http://127.0.0.1:$MOCK_PORT/v1/chat/completions"
# --- cortiq-gateway ----------------------------------------------------------
BIN="$ROOT/target/release/cortiq-gateway"
[ -x "$BIN" ] || BIN="$ROOT/target/debug/cortiq-gateway"
"$BIN" --config "$HERE/cortiq.bench.toml" >/tmp/bench_cortiq.log 2>&1 & CG=$!; pids+=($CG)
if wait_up "http://127.0.0.1:9100/healthz"; then
ab_run "cortiq-gateway" "http://127.0.0.1:9100/v1/chat/completions"
fi
kill $CG 2>/dev/null
# --- LiteLLM -----------------------------------------------------------------
if python3 -c "import litellm" 2>/dev/null; then
cat > /tmp/litellm.yaml <<YAML
model_list:
- model_name: bench
litellm_params:
model: openai/bench-model
api_base: http://127.0.0.1:$MOCK_PORT/v1
api_key: sk-bench
YAML
# the proxy CLI is brittle across versions; run the ASGI app via uvicorn.
# 4 workers = LiteLLM's recommended multi-process setup (fair vs a multi-core gateway).
CONFIG_FILE_PATH=/tmp/litellm.yaml python3 -m uvicorn litellm.proxy.proxy_server:app \
--host 127.0.0.1 --port 4000 --workers 4 >/tmp/bench_litellm.log 2>&1 & LL=$!; pids+=($LL)
if wait_up "http://127.0.0.1:4000/health/liveliness"; then
# warm up the workers before measuring (cold imports per worker)
for _ in $(seq 1 10); do curl -s -o /dev/null -X POST -H 'content-type: application/json' -d @"$BODY" "http://127.0.0.1:4000/v1/chat/completions"; done
sleep 2
ab_run "litellm" "http://127.0.0.1:4000/v1/chat/completions"
fi
kill $LL 2>/dev/null
else
echo "litellm | (not installed — pip install 'litellm[proxy]')"
fi
# --- Portkey -----------------------------------------------------------------
# resolve relative to the active node (npm root -g may point at a different npm)
PK_MAIN="$(dirname "$(command -v node 2>/dev/null)")/../lib/node_modules/@portkey-ai/gateway/build/start-server.js"
if [ -f "$PK_MAIN" ]; then
PORT=8787 node "$PK_MAIN" >/tmp/bench_portkey.log 2>&1 & PK=$!; pids+=($PK)
if wait_up "http://127.0.0.1:8787/v1/health" || wait_up "http://127.0.0.1:8787/"; then
sleep 1
ab_run "portkey" "http://127.0.0.1:8787/v1/chat/completions" \
-H "x-portkey-provider: openai" \
-H "x-portkey-custom-host: http://127.0.0.1:$MOCK_PORT/v1" \
-H "Authorization: Bearer sk-bench"
fi
kill $PK 2>/dev/null
else
echo "portkey | (not installed — npm i -g @portkey-ai/gateway)"
fi
# --- task-type routing accuracy (optional; needs a live router key) ----------
if [ -n "${CORTIQ_ROUTER_KEY:-}" ]; then
echo
echo "== task-type routing accuracy (allaigate router vs keyword heuristic) =="
python3 "$HERE/accuracy.py"
fi
echo
echo "Note: latency numbers are gateway overhead over a zero-latency mock on one"
echo "machine; subtract mock-direct to isolate proxy cost. Re-run on your hardware."
+37
View File
@@ -0,0 +1,37 @@
{"text": "Write a Python function to reverse a string", "label": "code"}
{"text": "Reverse a linked list in place", "label": "code"}
{"text": "Given an array of integers, return the indices of the two that add up to a target", "label": "code"}
{"text": "My loop prints one extra line at the end — what's wrong with it?", "label": "code"}
{"text": "Merge two sorted arrays into a single sorted array", "label": "code"}
{"text": "Make this faster: it compares every element with every other element", "label": "code"}
{"text": "Solve x^2 - 5x + 6 = 0", "label": "math"}
{"text": "If a train covers 60 miles in 1.5 hours, how fast is it going?", "label": "math"}
{"text": "A rectangle is twice as long as it is wide and its area is 50 — what are its sides?", "label": "math"}
{"text": "What's 15 percent of 240?", "label": "math"}
{"text": "Two dice are rolled — what are the odds the total is seven?", "label": "math"}
{"text": "The sequence 2, 6, 18, 54 — what comes next?", "label": "math"}
{"text": "Translate 'good morning' into Spanish", "label": "translation"}
{"text": "'Guten Tag' — what does that mean in English?", "label": "translation"}
{"text": "What's the Spanish word for butterfly?", "label": "translation"}
{"text": "Put 'I would like a coffee, please' into French", "label": "translation"}
{"text": "How do Italians say 'see you tomorrow'?", "label": "translation"}
{"text": "Give me the gist of the text below in one line", "label": "summarization"}
{"text": "What are the main takeaways from this report?", "label": "summarization"}
{"text": "Boil this down to a few key points", "label": "summarization"}
{"text": "In a sentence, what is this article about?", "label": "summarization"}
{"text": "Cut this down to just the essentials", "label": "summarization"}
{"text": "Ugh, Mondays. How was your weekend?", "label": "chitchat"}
{"text": "I'm a little bored, tell me something interesting", "label": "chitchat"}
{"text": "Do you think pineapple belongs on pizza?", "label": "chitchat"}
{"text": "Feeling great today, the sun is finally out!", "label": "chitchat"}
{"text": "Any fun plans for the holidays?", "label": "chitchat"}
{"text": "Which one is the tallest mountain on Earth?", "label": "qa"}
{"text": "At what temperature does water freeze in Fahrenheit?", "label": "qa"}
{"text": "Who is the author of Pride and Prejudice?", "label": "qa"}
{"text": "How many continents are there?", "label": "qa"}
{"text": "What is the currency used in Japan called?", "label": "qa"}
{"text": "From the paragraph below, grab all the phone numbers", "label": "extraction"}
{"text": "Which companies are named in this report?", "label": "extraction"}
{"text": "Give me just the dates and amounts from this receipt", "label": "extraction"}
{"text": "Pick out the email addresses in this message", "label": "extraction"}
{"text": "What are the product SKUs mentioned here?", "label": "extraction"}
+12
View File
@@ -150,3 +150,15 @@ enabled = true
file = "config/stats.jsonl" # empty = in-memory only
ring_size = 500 # how many recent requests to keep in the feed
retention = "7d" # aggregate/series depth (replayed on startup)
# ============================================================================
# Semantic cache — returns a cached answer for prompts semantically near a
# previous one, skipping the model call. Requires an embedding model in the pool.
# Disabled by default. (Cache settings take effect on restart.)
# ============================================================================
[cache]
enabled = false
threshold = 0.92 # cosine similarity for a hit (0..1)
ttl = "1h"
max_entries = 1000
# embed_model = "local-embed" # defaults to the first kind = "embedding" model
+83
View File
@@ -0,0 +1,83 @@
*Read this in [Русский](ROUTER.ru.md).*
# The router (allaigate)
Cortiq Gateway is a thin, fast proxy — the **intelligence** that decides *which* model
a prompt should go to lives in a separate **router**. When you send `model: "cortiq-auto"`,
the gateway calls the router's `POST /v1/route`, gets back a **task type** and a
**complexity tier**, and picks a model from your pool accordingly.
> Routing is optional. A pinned model (`model: "gpt-4o-mini"`) or `POST /route`-free
> usage needs **no router** — the gateway is still a useful universal proxy. The router
> is what turns it into a *smart* one.
## Two ways to get a router
1. **Hosted — [allaigate](https://api.allaigate.com/)** (fastest to start). A managed
semantic router: ~94% accuracy at separating close task types, patent-pending,
privacy-first. Get a key on the site (see the promo below).
2. **Self-hosted — `cortiq-router`.** Run your own classifier and point the gateway at it.
## 🎉 Launch promo
allaigate is currently running a launch offer:
> **$1 Starter key** — use promo code **`LAUNCH1`** at **https://api.allaigate.com/**
> (limited activations).
Get a key: enter your email + the promo code on the homepage; the key (`cortiq_…`) is
minted and shown to you.
### Pricing (as of this writing — check the site for current terms)
| Plan | Price | Included decisions | Rate limit | Overage |
|---|---|---|---|---|
| **Starter** | **$1/mo** (`LAUNCH1`) | unlimited | 60/min | — |
| Developer | $9/mo | 100k | 120/min | $0.10 / 1k |
| Pro | $49/mo | 1M | 600/min | $0.05 / 1k |
| Scale | $299/mo | 10M | 3000/min | $0.02 / 1k |
## Point the gateway at the hosted router
```toml
[router]
url = "https://138.226.222.209" # allaigate routing API
verify_tls = false # cert is bound to an IP
taxonomy_id = "data-assistant"
api_key_env = "CORTIQ_ROUTER_KEY" # your cortiq_… key
timeout_ms = 15000 # allow for oracle escalation on hard prompts
```
```bash
export CORTIQ_ROUTER_KEY=cortiq_your_key
cortiq-gateway --config config/gateway.toml
```
## API contract
`POST {router}/v1/route`, `Authorization: Bearer cortiq_…`:
```jsonc
// request
{ "input": { "text": "Write a Python function to reverse a list" }, "taxonomy_id": "data-assistant" }
// response
{
"decision": {
"task_label": "code",
"confidence": 0.99,
"complexity": { "tier": "low", "score": 0.27 }
}
}
```
The gateway maps `complexity.tier` (`low`/`medium`/`high`) to your `[routing]` table and
selects a model. On a hard/ambiguous prompt the router may escalate to an oracle LLM
(a few seconds) for a more accurate decision — size `timeout_ms` accordingly. See
[ROUTING.md](ROUTING.md).
## Verify accuracy yourself
`bench/accuracy.py` classifies a labeled prompt set through the live router — see
[BENCHMARKS.md](../BENCHMARKS.md). In our run the router scored **100% (37/37)** on
natural-language prompts vs a keyword heuristic's 32%.
+82
View File
@@ -0,0 +1,82 @@
*Read this in [English](ROUTER.md).*
# Роутер (allaigate)
Cortiq Gateway — тонкий и быстрый прокси; **интеллект**, решающий *какой* модели
отдать запрос, живёт в отдельном **роутере**. Когда вы шлёте `model: "cortiq-auto"`,
шлюз вызывает `POST /v1/route` роутера, получает **тип задачи** и **полосу сложности**
и выбирает модель из вашего пула.
> Роутинг опционален. Pinned-модель (`model: "gpt-4o-mini"`) роутер **не требует** —
> шлюз всё равно полезен как универсальный прокси. Роутер делает его *умным*.
## Два способа получить роутер
1. **Хостовый — [allaigate](https://api.allaigate.com/)** (быстрее всего начать).
Управляемый семантический роутер: ~94% точности на разделении близких типов задач,
patent-pending, privacy-first. Ключ — на сайте (см. акцию ниже).
2. **Свой — `cortiq-router`.** Поднимите собственный классификатор и укажите его в шлюзе.
## 🎉 Акция (launch)
Сейчас у allaigate идёт стартовая акция:
> **Starter-ключ за $1** — промокод **`LAUNCH1`** на **https://api.allaigate.com/**
> (число активаций ограничено).
Как получить ключ: введите e-mail и промокод на главной — ключ (`cortiq_…`) сгенерируется
и покажется вам.
### Тарифы (на момент написания — актуальные условия смотрите на сайте)
| План | Цена | Включено решений | Rate limit | Сверх лимита |
|---|---|---|---|---|
| **Starter** | **$1/мес** (`LAUNCH1`) | безлимит | 60/мин | — |
| Developer | $9/мес | 100k | 120/мин | $0.10 / 1k |
| Pro | $49/мес | 1M | 600/мин | $0.05 / 1k |
| Scale | $299/мес | 10M | 3000/мин | $0.02 / 1k |
## Подключение шлюза к хостовому роутеру
```toml
[router]
url = "https://138.226.222.209" # routing API allaigate
verify_tls = false # серт привязан к IP
taxonomy_id = "data-assistant"
api_key_env = "CORTIQ_ROUTER_KEY" # ваш ключ cortiq_…
timeout_ms = 15000 # запас под oracle-эскалацию на сложных запросах
```
```bash
export CORTIQ_ROUTER_KEY=cortiq_your_key
cortiq-gateway --config config/gateway.toml
```
## Контракт API
`POST {router}/v1/route`, `Authorization: Bearer cortiq_…`:
```jsonc
// запрос
{ "input": { "text": "Write a Python function to reverse a list" }, "taxonomy_id": "data-assistant" }
// ответ
{
"decision": {
"task_label": "code",
"confidence": 0.99,
"complexity": { "tier": "low", "score": 0.27 }
}
}
```
Шлюз сопоставляет `complexity.tier` (`low`/`medium`/`high`) с таблицей `[routing]` и
выбирает модель. На сложном/неоднозначном запросе роутер может эскалировать к oracle-LLM
(несколько секунд) для более точного решения — подберите `timeout_ms` соответственно.
См. [ROUTING.ru.md](ROUTING.ru.md).
## Проверьте точность сами
`bench/accuracy.py` прогоняет размеченный набор промптов через живой роутер — см.
[BENCHMARKS.md](../BENCHMARKS.md). В нашем прогоне роутер дал **100% (37/37)** на
естественных промптах против 32% у keyword-эвристики.
+51 -6
View File
@@ -8,8 +8,8 @@
pub mod assets;
use crate::config::{
AdminCfg, ApiKeyCfg, BreakerCfg, Config, CortiqCfg, LogCfg, ModelCfg, ProtocolsCfg, RouteCfg,
RouterCfg, RoutingCfg, RoutingPolicy, StatsCfg, TelemetryCfg, TierTargets,
AdminCfg, ApiKeyCfg, BreakerCfg, CacheCfg, Config, CortiqCfg, LogCfg, ModelCfg, ProtocolsCfg,
RouteCfg, RouterCfg, RoutingCfg, RoutingPolicy, StatsCfg, TelemetryCfg, TierTargets,
};
use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective};
use crate::state::SharedState;
@@ -165,6 +165,7 @@ pub fn api_routes(admin_token: String) -> Router<SharedState> {
.route("/admin/api/stats", get(get_stats))
.route("/admin/api/requests", get(get_requests))
.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| {
let expected = admin_token.clone();
async move { auth(expected, req, next).await }
@@ -270,7 +271,7 @@ async fn meta() -> ApiResult {
ok(json!({
"version": env!("CARGO_PKG_VERSION"),
"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"],
"kinds": ["chat", "embedding"],
"profiles": ["cost-saver", "balanced", "quality-first"],
@@ -279,8 +280,8 @@ async fn meta() -> ApiResult {
"tiers": ["low", "medium", "high"],
"caps": ["tools", "vision"],
"protocols_impl": {
"openai_chat": true, "openai_completions": false, "openai_embeddings": false,
"openai_models": false, "anthropic_messages": false, "mcp": false, "native_passthrough": false
"openai_chat": true, "openai_completions": true, "openai_embeddings": true,
"openai_models": true, "anthropic_messages": true, "mcp": true, "native_passthrough": true
},
"languages": ["en", "ru", "de", "fr", "es", "zh", "tr"],
}))
@@ -476,6 +477,7 @@ async fn get_settings(State(state): State<SharedState>) -> ApiResult {
"cortiq": c.cortiq,
"stats": c.stats,
"admin": c.admin,
"cache": c.cache,
}))
}
@@ -490,6 +492,7 @@ struct SettingsBody {
cortiq: Option<CortiqCfg>,
stats: Option<StatsCfg>,
admin: Option<AdminCfg>,
cache: Option<CacheCfg>,
}
async fn put_settings(State(state): State<SharedState>, Json(b): Json<SettingsBody>) -> ApiResult {
@@ -528,6 +531,11 @@ async fn put_settings(State(state): State<SharedState>, Json(b): Json<SettingsBo
}
cfg.admin = v;
}
if let Some(v) = b.cache {
// the cache is built at startup; persist now, apply on restart
needs_restart = true;
cfg.cache = v;
}
state.reload(cfg)?;
ok(json!({ "ok": true, "needs_restart": needs_restart }))
}
@@ -665,7 +673,9 @@ async fn get_stats(State(state): State<SharedState>, Query(q): Query<StatsQuery>
.and_then(parse_duration_secs)
.unwrap_or(24 * 3600);
let groupby = q.groupby.as_deref().unwrap_or("model");
ok(state.stats.snapshot(range_secs, groupby))
let mut snap = state.stats.snapshot(range_secs, groupby);
snap["cache"] = state.cache.snapshot();
ok(snap)
}
#[derive(Deserialize)]
@@ -753,3 +763,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() })),
}
}
/// 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(),
}
}
+171
View File
@@ -0,0 +1,171 @@
//! Semantic (embedding-based) response cache.
//!
//! Each entry stores an L2-normalized embedding of the request plus the response.
//! A lookup returns a cached answer when the cosine similarity to a previous request
//! (with the same routing signature, within TTL) meets the configured threshold —
//! skipping the expensive model call. In-memory ring buffer; counters track hit
//! rate and the model cost saved.
use crate::config::CacheCfg;
use crate::stats::parse_duration_secs;
use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};
fn now_secs() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
}
struct Entry {
sig: String,
vec: Vec<f32>, // L2-normalized
ts: u64,
model_used: String,
content: String,
prompt_tokens: u32,
completion_tokens: u32,
cost: f64,
}
pub struct CacheHit {
pub model_used: String,
pub content: String,
pub prompt_tokens: u32,
pub completion_tokens: u32,
pub similarity: f32,
}
pub struct SemanticCache {
enabled: bool,
threshold: f32,
ttl_secs: u64,
max_entries: usize,
embed_model: Option<String>,
entries: Mutex<VecDeque<Entry>>,
hits: AtomicU64,
misses: AtomicU64,
saved_micro_usd: AtomicU64,
}
impl SemanticCache {
pub fn new(cfg: &CacheCfg) -> Arc<Self> {
Arc::new(Self {
enabled: cfg.enabled,
threshold: cfg.threshold,
ttl_secs: parse_duration_secs(&cfg.ttl).unwrap_or(3600),
max_entries: cfg.max_entries.max(1),
embed_model: cfg.embed_model.clone(),
entries: Mutex::new(VecDeque::new()),
hits: AtomicU64::new(0),
misses: AtomicU64::new(0),
saved_micro_usd: AtomicU64::new(0),
})
}
pub fn enabled(&self) -> bool {
self.enabled
}
pub fn embed_model(&self) -> Option<&str> {
self.embed_model.as_deref()
}
/// Look up a cached response by routing signature + query embedding.
pub fn lookup(&self, sig: &str, query: &[f32]) -> Option<CacheHit> {
if !self.enabled {
return None;
}
let q = normalized(query);
let cutoff = now_secs().saturating_sub(self.ttl_secs);
let inner = self.entries.lock().unwrap();
let mut best: Option<(f32, &Entry)> = None;
for e in inner.iter() {
if e.sig != sig || e.ts < cutoff {
continue;
}
let sim = dot(&e.vec, &q);
if sim >= self.threshold && best.map(|(b, _)| sim > b).unwrap_or(true) {
best = Some((sim, e));
}
}
match best {
Some((sim, e)) => {
self.hits.fetch_add(1, Ordering::Relaxed);
self.saved_micro_usd
.fetch_add((e.cost * 1_000_000.0) as u64, Ordering::Relaxed);
Some(CacheHit {
model_used: e.model_used.clone(),
content: e.content.clone(),
prompt_tokens: e.prompt_tokens,
completion_tokens: e.completion_tokens,
similarity: sim,
})
}
None => {
self.misses.fetch_add(1, Ordering::Relaxed);
None
}
}
}
/// Store a response under its routing signature + query embedding.
#[allow(clippy::too_many_arguments)]
pub fn insert(
&self,
sig: String,
query: Vec<f32>,
model_used: String,
content: String,
prompt_tokens: u32,
completion_tokens: u32,
cost: f64,
) {
if !self.enabled || query.is_empty() {
return;
}
let vec = normalized(&query);
let mut inner = self.entries.lock().unwrap();
inner.push_back(Entry {
sig,
vec,
ts: now_secs(),
model_used,
content,
prompt_tokens,
completion_tokens,
cost,
});
while inner.len() > self.max_entries {
inner.pop_front();
}
}
pub fn snapshot(&self) -> serde_json::Value {
let h = self.hits.load(Ordering::Relaxed);
let m = self.misses.load(Ordering::Relaxed);
serde_json::json!({
"enabled": self.enabled,
"hits": h,
"misses": m,
"hit_rate": if h + m > 0 { h as f64 / (h + m) as f64 } else { 0.0 },
"entries": self.entries.lock().unwrap().len(),
"saved_usd": self.saved_micro_usd.load(Ordering::Relaxed) as f64 / 1_000_000.0,
})
}
}
fn dot(a: &[f32], b: &[f32]) -> f32 {
a.iter().zip(b.iter()).map(|(x, y)| x * y).sum()
}
fn normalized(v: &[f32]) -> Vec<f32> {
let n = v.iter().map(|x| x * x).sum::<f32>().sqrt();
if n > 0.0 {
v.iter().map(|x| x / n).collect()
} else {
v.to_vec()
}
}
+42
View File
@@ -31,6 +31,8 @@ pub struct Config {
pub admin: AdminCfg,
#[serde(default)]
pub stats: StatsCfg,
#[serde(default)]
pub cache: CacheCfg,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
@@ -320,6 +322,46 @@ fn default_stats_retention() -> String {
"7d".into()
}
/// Semantic (embedding-based) response cache — returns a cached answer for prompts
/// that are semantically near a previous one, skipping the (expensive) model call.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct CacheCfg {
#[serde(default)]
pub enabled: bool,
/// Cosine-similarity threshold for a hit (0..1).
#[serde(default = "default_semcache_threshold")]
pub threshold: f32,
/// Entry time-to-live, e.g. `1h`.
#[serde(default = "default_semcache_ttl")]
pub ttl: String,
/// Maximum number of cached entries (ring buffer).
#[serde(default = "default_semcache_max")]
pub max_entries: usize,
/// Embedding model id (defaults to the first `kind = "embedding"` model).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub embed_model: Option<String>,
}
impl Default for CacheCfg {
fn default() -> Self {
Self {
enabled: false,
threshold: default_semcache_threshold(),
ttl: default_semcache_ttl(),
max_entries: default_semcache_max(),
embed_model: None,
}
}
}
fn default_semcache_threshold() -> f32 {
0.92
}
fn default_semcache_ttl() -> String {
"1h".into()
}
fn default_semcache_max() -> usize {
1000
}
impl Config {
/// Load and validate the config from a file.
pub fn load(path: &str) -> anyhow::Result<Self> {
+1
View File
@@ -9,6 +9,7 @@
#![allow(dead_code)]
mod admin;
mod cache;
mod config;
mod error;
mod model;
+348 -5
View File
@@ -4,9 +4,9 @@
use crate::error::{GatewayError, Result};
use crate::model::{
ChatRequest, ChatResponse, Message, RouteDecision, RouteInfo, RoutingDirective,
ChatRequest, ChatResponse, Choice, Message, RouteDecision, RouteInfo, RoutingDirective, Usage,
};
use crate::state::SharedState;
use crate::state::{Live, SharedState};
use crate::stats::RequestRecord;
use std::time::{Instant, SystemTime, UNIX_EPOCH};
@@ -28,6 +28,45 @@ impl Pipeline {
}
.to_string();
// --- semantic cache lookup (short-circuit before the model call) ---
let cache = &state.cache;
let mut cache_key: Option<(String, Vec<f32>)> = None;
if cache.enabled() {
let live = state.live();
let sig = cache_sig(&req);
let text = self.extract_text(&req.messages, "concat_all", live.cfg.route.max_chars);
let embed_model = cache
.embed_model()
.map(|s| s.to_string())
.or_else(|| first_embedding_model(&live));
if let Some(model_id) = embed_model {
if let Some(vec) = embed(&live, &model_id, &text).await {
if let Some(hit) = cache.lookup(&sig, &vec) {
state.stats.record(RequestRecord {
ts: now_secs(),
account,
protocol,
directive,
task_label: "cache".into(),
tier: "cache".into(),
score: 0.0,
model_id: hit.model_used.clone(),
route_source: "cache".into(),
prompt_tokens: hit.prompt_tokens,
completion_tokens: hit.completion_tokens,
cost_usd: 0.0,
latency_ms: started.elapsed().as_millis() as u64,
outcome: "ok".into(),
failover: false,
error: None,
});
return Ok(cached_response(&hit));
}
cache_key = Some((sig, vec));
}
}
}
let result = self.run_inner(req, state).await;
// record statistics (for both success and error)
@@ -76,12 +115,36 @@ impl Pipeline {
};
state.stats.record(rec);
// --- store a successful response in the semantic cache ---
if let (Ok(resp), Some((sig, vec))) = (&result, cache_key) {
let content = resp
.choices
.first()
.map(|c| c.message.content.clone())
.unwrap_or_default();
if !content.is_empty() {
state.cache.insert(
sig,
vec,
resp.cortiq.selected_model.clone(),
content,
resp.usage.prompt_tokens,
resp.usage.completion_tokens,
resp.cortiq.cost_usd,
);
}
}
result
}
async fn run_inner(&self, req: ChatRequest, state: &SharedState) -> Result<ChatResponse> {
let live = state.live();
/// Stages 1–3: extract text → route → select & capability-filter candidates.
/// 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
let text = self.extract_text(
&req.messages,
@@ -159,6 +222,12 @@ impl Pipeline {
.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)
let mut last_err = None;
@@ -210,6 +279,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 {
let raw_text = match strategy {
"last_user" => messages
@@ -249,3 +401,194 @@ impl Pipeline {
}
}
}
fn now_secs() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
}
/// Routing signature used to partition cache entries (so a pinned model's cached
/// answer is never returned for a different model/profile).
fn cache_sig(req: &ChatRequest) -> String {
match &req.routing {
RoutingDirective::Auto { profile } => {
format!("auto:{}", profile.as_deref().unwrap_or(""))
}
RoutingDirective::Pinned { model_id } => format!("pinned:{model_id}"),
}
}
fn first_embedding_model(live: &Live) -> Option<String> {
live.cfg
.models
.iter()
.find(|m| m.kind == "embedding")
.map(|m| m.id.clone())
}
/// Embed `text` with the given model; returns the embedding vector (or None on error).
async fn embed(live: &Live, model_id: &str, text: &str) -> Option<Vec<f32>> {
let provider = live.registry.get(model_id)?;
let body = provider.embed(serde_json::json!(text)).await.ok()?;
let arr = body["data"][0]["embedding"].as_array()?;
Some(
arr.iter()
.filter_map(|x| x.as_f64())
.map(|x| x as f32)
.collect(),
)
}
fn cached_response(hit: &crate::cache::CacheHit) -> ChatResponse {
ChatResponse {
id: "cache".to_string(),
model_used: hit.model_used.clone(),
choices: vec![Choice {
index: 0,
message: Message {
role: "assistant".into(),
content: hit.content.clone(),
tool_calls: vec![],
},
finish_reason: "stop".into(),
}],
usage: Usage {
prompt_tokens: hit.prompt_tokens,
completion_tokens: hit.completion_tokens,
total_tokens: hit.prompt_tokens + hit.completion_tokens,
},
cortiq: RouteInfo {
task_label: "cache".into(),
complexity_score: 0.0,
complexity_tier: "cache".into(),
selected_model: hit.model_used.clone(),
route_source: "cache".into(),
router_request_id: None,
cost_usd: 0.0,
failover: false,
},
}
}
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,
});
}
}
+209
View File
@@ -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"})));
}
}
+162
View File
@@ -0,0 +1,162 @@
//! MCP (Model Context Protocol) server: `POST /mcp` (JSON-RPC 2.0).
//!
//! Exposes the gateway's routing to MCP-native orchestrators as two tools:
//! `cortiq_route` (classify a prompt → routing decision) and `cortiq_chat`
//! (run a prompt through intelligent routing → answer). Single JSON endpoint;
//! notifications are accepted with `202`. Gated by `protocols.mcp`.
use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective};
use crate::state::SharedState;
use axum::response::{IntoResponse, Response};
use axum::{extract::State, http::StatusCode, routing::post, Json, Router};
use serde_json::{json, Value};
pub fn routes() -> Router<SharedState> {
Router::new().route("/mcp", post(handler))
}
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(),
}
}
}
fn tools_list() -> Value {
json!([
{
"name": "cortiq_route",
"description": "Classify a prompt and return the routing decision (task type, complexity tier, score, and the model the gateway would select).",
"inputSchema": {
"type": "object",
"properties": { "text": { "type": "string", "description": "The prompt to classify." } },
"required": ["text"]
}
},
{
"name": "cortiq_chat",
"description": "Run a prompt through the gateway's intelligent routing and return the selected model's answer.",
"inputSchema": {
"type": "object",
"properties": {
"prompt": { "type": "string", "description": "The user prompt." },
"model": { "type": "string", "description": "Optional: cortiq-auto[:profile] or a model id (default cortiq-auto)." }
},
"required": ["prompt"]
}
}
])
}
fn tool_text(text: String, is_error: bool) -> Value {
json!({ "content": [{ "type": "text", "text": text }], "isError": is_error })
}
async fn handle_tool_call(
state: &SharedState,
params: &Value,
) -> std::result::Result<Value, (i64, String)> {
let name = params["name"].as_str().unwrap_or("");
let args = &params["arguments"];
let live = state.live();
match name {
"cortiq_route" => {
let text = args["text"].as_str().unwrap_or("");
let prof = live.cfg.route.profile.clone();
match live.router.route(text, &prof).await {
Ok(Some(d)) => Ok(tool_text(
format!(
"task_label={}\ncomplexity_tier={}\ncomplexity_score={:.3}\ncandidates={:?}",
d.task_label,
d.complexity_tier,
d.complexity_score,
live.routing.candidates(&d.complexity_tier)
),
false,
)),
_ => Ok(tool_text(
"router unavailable — would use the default model".to_string(),
false,
)),
}
}
"cortiq_chat" => {
let prompt = args["prompt"].as_str().unwrap_or("");
let model = args["model"].as_str().unwrap_or("cortiq-auto");
let req = ChatRequest {
routing: parse_routing(model),
messages: vec![Message {
role: "user".into(),
content: prompt.to_string(),
tool_calls: vec![],
}],
tools: vec![],
params: GenParams::default(),
stream: false,
meta: RequestMeta {
account: "mcp".into(),
protocol: "mcp".into(),
..Default::default()
},
};
match state.pipeline.run(req, state).await {
Ok(resp) => {
let answer = resp
.choices
.first()
.map(|c| c.message.content.clone())
.unwrap_or_default();
Ok(tool_text(answer, false))
}
Err(e) => Ok(tool_text(e.to_string(), true)),
}
}
_ => Err((-32602, format!("unknown tool '{name}'"))),
}
}
async fn handler(State(state): State<SharedState>, Json(req): Json<Value>) -> Response {
if !state.live().cfg.protocols.mcp {
return (StatusCode::NOT_FOUND, "mcp protocol is disabled").into_response();
}
let method = req["method"].as_str().unwrap_or("").to_string();
let id = req.get("id").cloned();
// JSON-RPC notifications carry no id and expect no response
if id.is_none() || method.starts_with("notifications/") {
return StatusCode::ACCEPTED.into_response();
}
let id = id.unwrap_or(Value::Null);
let result: std::result::Result<Value, (i64, String)> = match method.as_str() {
"initialize" => {
let pv = req["params"]["protocolVersion"]
.as_str()
.unwrap_or("2024-11-05")
.to_string();
Ok(json!({
"protocolVersion": pv,
"capabilities": { "tools": {} },
"serverInfo": { "name": "cortiq-gateway", "version": env!("CARGO_PKG_VERSION") }
}))
}
"ping" => Ok(json!({})),
"tools/list" => Ok(json!({ "tools": tools_list() })),
"tools/call" => handle_tool_call(&state, &req["params"]).await,
other => Err((-32601, format!("method not found: {other}"))),
};
let body = match result {
Ok(r) => json!({ "jsonrpc": "2.0", "id": id, "result": r }),
Err((code, message)) => {
json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } })
}
};
Json(body).into_response()
}
+14 -4
View File
@@ -1,7 +1,13 @@
//! Incoming protocol adapters. Each one builds its own axum routes and translates
//! its protocol to/from the canonical model. Enabled via flags in `[protocols]`.
pub mod anthropic_messages;
pub mod mcp;
pub mod native_passthrough;
pub mod openai_chat;
pub mod openai_completions;
pub mod openai_embeddings;
pub mod openai_models;
use crate::state::SharedState;
use axum::Router;
@@ -10,8 +16,12 @@ use axum::Router;
/// each handler checks the live `protocols.*` flag from config itself, so
/// toggles in the admin panel take effect without a restart (hot switching).
pub fn build_router() -> Router<SharedState> {
Router::new().merge(openai_chat::routes())
// TODO: openai_completions, openai_embeddings, openai_models,
// anthropic_messages, mcp, native_passthrough — add here as well,
// with a check of the corresponding flag inside the handler.
Router::new()
.merge(openai_chat::routes())
.merge(openai_completions::routes())
.merge(openai_embeddings::routes())
.merge(openai_models::routes())
.merge(anthropic_messages::routes())
.merge(mcp::routes())
.merge(native_passthrough::routes())
}
+86
View File
@@ -0,0 +1,86 @@
//! Native passthrough adapter: `POST /route`.
//! Returns the gateway's routing decision (task type, complexity, candidate models)
//! without calling a model — direct access for clients that route themselves.
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("/route", post(handler))
}
#[derive(Deserialize)]
struct RouteRequest {
/// `{ "input": { "text": "..." } }` (router-style) — optional.
#[serde(default)]
input: serde_json::Value,
/// `{ "text": "..." }` — convenience shorthand.
#[serde(default)]
text: Option<String>,
#[serde(default)]
profile: Option<String>,
}
async fn handler(
State(state): State<SharedState>,
Json(req): Json<RouteRequest>,
) -> Result<impl IntoResponse> {
let live = state.live();
if !live.cfg.protocols.native_passthrough {
return Err(GatewayError::InvalidRequest(
"native_passthrough protocol is disabled".into(),
));
}
let text = req
.text
.clone()
.or_else(|| req.input["text"].as_str().map(|s| s.to_string()))
.or_else(|| req.input.as_str().map(|s| s.to_string()))
.unwrap_or_default();
if text.is_empty() {
return Err(GatewayError::InvalidRequest(
"missing input text (use {\"text\": ...} or {\"input\": {\"text\": ...}})".into(),
));
}
let profile = req
.profile
.clone()
.unwrap_or_else(|| live.cfg.route.profile.clone());
let (decision, source, candidates) = match live.router.route(&text, &profile).await {
Ok(Some(d)) => {
let cands = live.routing.candidates(&d.complexity_tier);
(Some(d), "router", cands)
}
_ => (
None,
"fallback",
vec![live.routing.default_model().to_string()],
),
};
let selected = candidates.first().cloned().unwrap_or_default();
let decision_json = match &decision {
Some(d) => serde_json::json!({
"task_label": d.task_label,
"complexity": { "score": d.complexity_score, "tier": d.complexity_tier },
"router_request_id": d.router_request_id,
}),
None => serde_json::json!({
"task_label": "degraded",
"complexity": { "score": 0.5, "tier": "degraded" },
"router_request_id": serde_json::Value::Null,
}),
};
Ok(Json(serde_json::json!({
"decision": decision_json,
"source": source,
"candidates": candidates,
"selected_model": selected,
})))
}
+42 -45
View File
@@ -3,11 +3,37 @@
//! to the pipeline, then converts the canonical [`ChatResponse`] back to OpenAI format.
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 axum::response::{IntoResponse, Response};
use axum::{extract::State, routing::post, Json, Router};
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> {
Router::new().route("/v1/chat/completions", post(handler))
}
@@ -46,7 +72,7 @@ fn parse_routing(model: &str) -> RoutingDirective {
async fn handler(
State(state): State<SharedState>,
Json(req): Json<OpenAiChatRequest>,
) -> Result<impl axum::response::IntoResponse> {
) -> Result<Response> {
// hot protocol toggle: if the adapter is disabled in config — return 404
if !state.live().cfg.protocols.openai_chat {
return Err(GatewayError::InvalidRequest(
@@ -77,52 +103,23 @@ async fn handler(
},
};
// streaming: forward provider SSE chunks verbatim with X-Cortiq-* headers
if req.stream {
return Err(GatewayError::InvalidRequest(
"Streaming is not yet supported in this version".into(),
));
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"),
);
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 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 headers = cortiq_headers(&resp.cortiq);
let mut body = serde_json::json!({
"id": resp.id,
@@ -164,5 +161,5 @@ async fn handler(
});
}
Ok((headers, Json(body)))
Ok((headers, Json(body)).into_response())
}
+183
View File
@@ -0,0 +1,183 @@
//! Inbound OpenAI Completions (legacy) adapter: `POST /v1/completions`.
//! Converts the legacy `prompt` into a single user message, runs the pipeline, and
//! returns the legacy `text_completion` shape (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/completions", post(handler))
}
#[derive(Deserialize)]
struct CompletionsRequest {
model: String,
prompt: serde_json::Value,
#[serde(default)]
temperature: Option<f32>,
#[serde(default)]
max_tokens: Option<u32>,
#[serde(default)]
top_p: Option<f32>,
#[serde(default)]
stream: bool,
}
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(),
}
}
}
fn prompt_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(|x| x.as_str())
.collect::<Vec<_>>()
.join("\n");
}
String::new()
}
async fn handler(
State(state): State<SharedState>,
Json(req): Json<CompletionsRequest>,
) -> Result<Response> {
if !state.live().cfg.protocols.openai_completions {
return Err(GatewayError::InvalidRequest(
"openai_completions protocol is disabled".into(),
));
}
let prompt = prompt_to_string(&req.prompt);
if prompt.is_empty() {
return Err(GatewayError::InvalidRequest(
"prompt must not be empty".into(),
));
}
let canonical = ChatRequest {
routing: parse_routing(&req.model),
messages: vec![Message {
role: "user".into(),
content: prompt,
tool_calls: vec![],
}],
tools: vec![],
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: "openai_completions".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"),
);
return Ok((
headers,
axum::body::Body::from_stream(chat_to_completions_sse(stream)),
)
.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 finish = resp
.choices
.first()
.map(|c| c.finish_reason.clone())
.unwrap_or_else(|| "stop".into());
let headers = cortiq_headers(&resp.cortiq);
let body = serde_json::json!({
"id": resp.id,
"object": "text_completion",
"created": std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs(),
"model": resp.model_used,
"choices": [{ "text": answer, "index": 0, "logprobs": serde_json::Value::Null, "finish_reason": finish }],
"usage": {
"prompt_tokens": resp.usage.prompt_tokens,
"completion_tokens": resp.usage.completion_tokens,
"total_tokens": resp.usage.total_tokens,
},
});
Ok((headers, Json(body)).into_response())
}
/// Translate chat `chat.completion.chunk` SSE into legacy `text_completion` SSE
/// (`delta.content` → `text`).
fn chat_to_completions_sse(
stream: crate::providers::ChatStream,
) -> impl futures::Stream<Item = Result<bytes::Bytes>> + Send + 'static {
use futures::StreamExt;
async_stream::stream! {
let mut buf = String::new();
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() { continue; }
if data == "[DONE]" {
yield Ok(bytes::Bytes::from("data: [DONE]\n\n"));
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()) {
yield Ok(bytes::Bytes::from(format!(
"data: {}\n\n",
serde_json::json!({"object": "text_completion", "choices": [], "usage": u})
)));
}
if let Some(t) = v["choices"][0]["delta"]["content"].as_str() {
yield Ok(bytes::Bytes::from(format!(
"data: {}\n\n",
serde_json::json!({
"object": "text_completion",
"choices": [{"text": t, "index": 0, "finish_reason": null}]
})
)));
}
}
}
}
}
}
+52
View File
@@ -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))
}
+32
View File
@@ -0,0 +1,32 @@
//! Inbound OpenAI Models adapter: `GET /v1/models`.
//! Lists the configured pool (plus the virtual `cortiq-auto`) for client discovery.
use crate::error::{GatewayError, Result};
use crate::state::SharedState;
use axum::{extract::State, response::IntoResponse, routing::get, Json, Router};
pub fn routes() -> Router<SharedState> {
Router::new().route("/v1/models", get(handler))
}
async fn handler(State(state): State<SharedState>) -> Result<impl IntoResponse> {
let live = state.live();
if !live.cfg.protocols.openai_models {
return Err(GatewayError::InvalidRequest(
"openai_models protocol is disabled".into(),
));
}
let mut data = vec![serde_json::json!({
"id": "cortiq-auto",
"object": "model",
"owned_by": "cortiq",
})];
for m in &live.cfg.models {
data.push(serde_json::json!({
"id": m.id,
"object": "model",
"owned_by": m.provider,
}));
}
Ok(Json(serde_json::json!({ "object": "list", "data": data })))
}
+215 -13
View File
@@ -1,23 +1,25 @@
//! Anthropic Messages provider (v0.2). Translates the canonical ChatRequest to
//! `POST /v1/messages` and back. The `system` message is extracted into a separate field.
//! Anthropic Messages provider. Translates the canonical [`ChatRequest`] to
//! `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 crate::config::ModelCfg;
use crate::error::{GatewayError, Result};
use crate::model::{ChatRequest, ChatResponse};
use crate::model::{ChatRequest, ChatResponse, Choice, Message, RouteInfo, Usage};
use async_trait::async_trait;
const ANTHROPIC_VERSION: &str = "2023-06-01";
pub struct AnthropicProvider {
id: String,
#[allow(dead_code)]
base_url: String,
#[allow(dead_code)]
model: String,
#[allow(dead_code)]
api_key: Option<String>,
caps: Caps,
price_in: f64,
price_out: f64,
http: reqwest::Client,
}
impl AnthropicProvider {
@@ -30,8 +32,61 @@ impl AnthropicProvider {
caps: Caps::from_list(&cfg.caps),
price_in: cfg.price_in,
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]
@@ -46,12 +101,159 @@ impl Provider for AnthropicProvider {
(prompt_tokens as f64 * self.price_in + completion_tokens as f64 * self.price_out)
/ 1_000_000.0
}
async fn chat(&self, _req: ChatRequest) -> Result<ChatResponse> {
// TODO(v0.2): extract system from messages → "system" field; messages → content blocks;
// add x-api-key + anthropic-version headers; map response → canonical form.
// Returns an error (not a panic) so that failover/playground continue to work normally.
Err(GatewayError::UpstreamUnavailable(
"anthropic provider not yet implemented (planned v0.2)".to_string(),
))
async fn chat(&self, req: ChatRequest) -> Result<ChatResponse> {
let resp = self
.request(&self.to_wire(&req, false))
.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
)));
}
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"));
}
}
+25 -2
View File
@@ -7,8 +7,16 @@
pub mod anthropic;
pub mod openai;
use crate::error::GatewayError;
use crate::model::{ChatRequest, ChatResponse};
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).
#[derive(Clone, Debug, Default)]
@@ -41,6 +49,21 @@ pub trait Provider: Send + Sync {
/// Non-streaming chat call.
async fn chat(&self, req: ChatRequest) -> crate::error::Result<ChatResponse>;
// TODO(v0.1): async fn chat_stream(...) -> Result<ChatStream> (SSE).
// TODO(v0.1): async fn embed(...) -> Result<EmbedResponse>.
/// Streaming chat call (SSE). Default: not supported (override per provider).
async fn chat_stream(&self, _req: ChatRequest) -> crate::error::Result<ChatStream> {
Err(GatewayError::UpstreamUnavailable(format!(
"streaming is not supported by provider '{}'",
self.id()
)))
}
/// 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()
)))
}
}
+57
View File
@@ -154,4 +154,61 @@ 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))
}
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}"))
})
}
}
-3
View File
@@ -15,9 +15,6 @@ impl Registry {
pub fn from_config(cfg: &Config, secrets: &SecretStore) -> anyhow::Result<Self> {
let mut by_id: HashMap<String, Arc<dyn Provider>> = HashMap::new();
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
let api_key = m
.api_key_env
+4
View File
@@ -5,6 +5,7 @@
//! a restart: the admin API validates the new config, builds a new `Live`,
//! atomically writes the TOML, and swaps the pointer.
use crate::cache::SemanticCache;
use crate::config::Config;
use crate::registry::Registry;
use crate::router_client::RouterClient;
@@ -42,6 +43,7 @@ impl Live {
pub struct AppState {
pub live: ArcSwap<Live>,
pub stats: Arc<Stats>,
pub cache: Arc<SemanticCache>,
pub secrets: SecretStore,
pub config_path: String,
pub pipeline: crate::pipeline::Pipeline,
@@ -59,10 +61,12 @@ impl SharedState {
let secrets_path = sibling(&config_path, "secrets.toml");
let secrets = SecretStore::load(&secrets_path);
let stats = Stats::new(&cfg.stats);
let cache = SemanticCache::new(&cfg.cache);
let live = Live::build(cfg, &secrets)?;
Ok(SharedState(Arc::new(AppState {
live: ArcSwap::from_pointee(live),
stats,
cache,
secrets,
config_path,
pipeline: crate::pipeline::Pipeline::new(),
+48
View File
@@ -91,3 +91,51 @@ export const api = {
// playground
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"),
};
}
+12
View File
@@ -200,6 +200,7 @@ export default {
"pg.temp": "Temperatur",
"pg.maxTokens": "Max. Tokens",
"pg.send": "Senden",
"pg.stream": "Streamen",
"pg.sending": "Wird gesendet…",
"pg.answer": "Antwort",
"pg.empty": "Antwort und Routing-Entscheidung erscheinen hier.",
@@ -253,4 +254,15 @@ export default {
"toast.deleted": "Gelöscht",
"toast.probing": "Prüfe…",
"toast.imported": "Konfiguration importiert",
"dash.cache": "Semantischer Cache",
"dash.cache.hitRate": "Trefferquote",
"dash.cache.hits": "Treffer",
"dash.cache.saved": "Gespart",
"dash.cache.disabled": "Semantischer Cache ist deaktiviert — in den Einstellungen aktivieren.",
"settings.cache": "Semantischer Cache",
"settings.cache.enabled": "Semantischen Cache aktivieren",
"settings.cache.threshold": "Ähnlichkeitsschwelle",
"settings.cache.ttl": "Eintrags-TTL",
"settings.cache.embed": "Embedding-Modell",
"settings.cache.note": "Cache-Einstellungen werden nach einem Neustart wirksam.",
};
+12
View File
@@ -201,6 +201,7 @@ export default {
"pg.temp": "Temperature",
"pg.maxTokens": "Max tokens",
"pg.send": "Send",
"pg.stream": "Stream",
"pg.sending": "Sending…",
"pg.answer": "Answer",
"pg.empty": "The answer and routing decision will appear here.",
@@ -254,4 +255,15 @@ export default {
"toast.deleted": "Deleted",
"toast.probing": "Probing…",
"toast.imported": "Config imported",
"dash.cache": "Semantic cache",
"dash.cache.hitRate": "Hit rate",
"dash.cache.hits": "Hits",
"dash.cache.saved": "Saved",
"dash.cache.disabled": "Semantic cache is disabled — enable it in Settings.",
"settings.cache": "Semantic cache",
"settings.cache.enabled": "Enable semantic cache",
"settings.cache.threshold": "Similarity threshold",
"settings.cache.ttl": "Entry TTL",
"settings.cache.embed": "Embedding model",
"settings.cache.note": "Cache settings take effect after a restart.",
};
+12
View File
@@ -200,6 +200,7 @@ export default {
"pg.temp": "Temperatura",
"pg.maxTokens": "Tokens máx.",
"pg.send": "Enviar",
"pg.stream": "Streaming",
"pg.sending": "Enviando…",
"pg.answer": "Respuesta",
"pg.empty": "La respuesta y la decisión de enrutamiento aparecerán aquí.",
@@ -253,4 +254,15 @@ export default {
"toast.deleted": "Eliminado",
"toast.probing": "Sondeando…",
"toast.imported": "Configuración importada",
"dash.cache": "Caché semántico",
"dash.cache.hitRate": "Aciertos",
"dash.cache.hits": "Aciertos",
"dash.cache.saved": "Ahorrado",
"dash.cache.disabled": "El caché semántico está desactivado — actívalo en Ajustes.",
"settings.cache": "Caché semántico",
"settings.cache.enabled": "Activar caché semántico",
"settings.cache.threshold": "Umbral de similitud",
"settings.cache.ttl": "TTL de entrada",
"settings.cache.embed": "Modelo de embeddings",
"settings.cache.note": "Los ajustes del caché se aplican tras reiniciar.",
};
+12
View File
@@ -200,6 +200,7 @@ export default {
"pg.temp": "Température",
"pg.maxTokens": "Tokens max",
"pg.send": "Envoyer",
"pg.stream": "Streaming",
"pg.sending": "Envoi…",
"pg.answer": "Réponse",
"pg.empty": "La réponse et la décision de routage s'afficheront ici.",
@@ -253,4 +254,15 @@ export default {
"toast.deleted": "Supprimé",
"toast.probing": "Sondage…",
"toast.imported": "Config importée",
"dash.cache": "Cache sémantique",
"dash.cache.hitRate": "Taux de succès",
"dash.cache.hits": "Succès",
"dash.cache.saved": "Économisé",
"dash.cache.disabled": "Le cache sémantique est désactivé — activez-le dans les Paramètres.",
"settings.cache": "Cache sémantique",
"settings.cache.enabled": "Activer le cache sémantique",
"settings.cache.threshold": "Seuil de similarité",
"settings.cache.ttl": "TTL d'entrée",
"settings.cache.embed": "Modèle d'embeddings",
"settings.cache.note": "Les paramètres du cache prennent effet après un redémarrage.",
};
+12
View File
@@ -200,6 +200,7 @@ export default {
"pg.temp": "Температура",
"pg.maxTokens": "Макс. токены",
"pg.send": "Отправить",
"pg.stream": "Стриминг",
"pg.sending": "Отправка…",
"pg.answer": "Ответ",
"pg.empty": "Ответ и решение роутера появятся здесь.",
@@ -253,4 +254,15 @@ export default {
"toast.deleted": "Удалено",
"toast.probing": "Проверка…",
"toast.imported": "Конфигурация импортирована",
"dash.cache": "Семантический кэш",
"dash.cache.hitRate": "Попаданий",
"dash.cache.hits": "Попадания",
"dash.cache.saved": "Сэкономлено",
"dash.cache.disabled": "Семантический кэш выключен — включите в Настройках.",
"settings.cache": "Семантический кэш",
"settings.cache.enabled": "Включить семантический кэш",
"settings.cache.threshold": "Порог сходства",
"settings.cache.ttl": "TTL записи",
"settings.cache.embed": "Модель эмбеддингов",
"settings.cache.note": "Настройки кэша применяются после перезапуска.",
};
+12
View File
@@ -200,6 +200,7 @@ export default {
"pg.temp": "Sıcaklık",
"pg.maxTokens": "Maks. token",
"pg.send": "Gönder",
"pg.stream": "Akış",
"pg.sending": "Gönderiliyor…",
"pg.answer": "Yanıt",
"pg.empty": "Yanıt ve yönlendirme kararı burada görünecek.",
@@ -253,4 +254,15 @@ export default {
"toast.deleted": "Silindi",
"toast.probing": "Sınanıyor…",
"toast.imported": "Yapılandırma içe aktarıldı",
"dash.cache": "Anlamsal önbellek",
"dash.cache.hitRate": "İsabet oranı",
"dash.cache.hits": "İsabet",
"dash.cache.saved": "Tasarruf",
"dash.cache.disabled": "Anlamsal önbellek kapalı — Ayarlar'dan etkinleştirin.",
"settings.cache": "Anlamsal önbellek",
"settings.cache.enabled": "Anlamsal önbelleği etkinleştir",
"settings.cache.threshold": "Benzerlik eşiği",
"settings.cache.ttl": "Giriş TTL",
"settings.cache.embed": "Gömme modeli",
"settings.cache.note": "Anlamsal önbellek ayarları yeniden başlatmadan sonra geçerli olur.",
};
+12
View File
@@ -200,6 +200,7 @@ export default {
"pg.temp": "温度",
"pg.maxTokens": "最大 Token 数",
"pg.send": "发送",
"pg.stream": "流式",
"pg.sending": "发送中…",
"pg.answer": "回答",
"pg.empty": "回答和路由决策将在此显示。",
@@ -253,4 +254,15 @@ export default {
"toast.deleted": "已删除",
"toast.probing": "探测中…",
"toast.imported": "配置已导入",
"dash.cache": "语义缓存",
"dash.cache.hitRate": "命中率",
"dash.cache.hits": "命中",
"dash.cache.saved": "已节省",
"dash.cache.disabled": "语义缓存已禁用 — 在设置中启用。",
"settings.cache": "语义缓存",
"settings.cache.enabled": "启用语义缓存",
"settings.cache.threshold": "相似度阈值",
"settings.cache.ttl": "条目 TTL",
"settings.cache.embed": "嵌入模型",
"settings.cache.note": "缓存设置将在重启后生效。",
};
+16
View File
@@ -108,6 +108,22 @@ export async function renderDashboard() {
barList(breakdown, num)
),
// semantic cache
stats.cache && stats.cache.enabled
? h(
"div",
{ class: "card" },
h("div", { class: "card-head" }, h("h3", {}, t("dash.cache"))),
h(
"div",
{ class: "kpis" },
kpi(pct(stats.cache.hit_rate), t("dash.cache.hitRate")),
kpi(num(stats.cache.hits), t("dash.cache.hits")),
kpi(money(stats.cache.saved_usd), t("dash.cache.saved"))
)
)
: null,
// recent
h(
"div",
+44 -14
View File
@@ -1,7 +1,7 @@
// 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 { api } from "../api.js";
import { api, testStream } from "../api.js";
import { appState } from "../app.js";
export async function renderPlayground() {
@@ -16,6 +16,7 @@ export async function renderPlayground() {
const promptIn = h("textarea", { placeholder: t("pg.promptPlaceholder"), rows: 5 });
const tempIn = h("input", { type: "number", step: "0.1", value: 0.7 });
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 modelSel = h("select", {}, ...ids.map((id) => h("option", { value: id }, id)));
@@ -62,19 +63,43 @@ export async function renderPlayground() {
sendBtn.disabled = true;
sendBtn.textContent = 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 {
const r = await api.test({
model,
messages,
temperature: parseFloat(tempIn.value),
max_tokens: parseInt(maxTokIn.value) || null,
});
if (r.ok) {
mount(out, r.answer || h("span", { class: "muted" }, "—"));
renderDecision(r);
if (streamCb.checked) {
const started = performance.now();
let acc = "";
out.textContent = "";
const info = await testStream(body, (d) => {
acc += d;
out.textContent = acc;
});
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 {
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" }, "—"));
const r = await api.test(body);
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) {
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),
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: "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)
),
+29
View File
@@ -42,6 +42,18 @@ export async function renderSettings() {
const statsEnabled = check(t("settings.stats.enabled"), s.stats?.enabled);
const statsFile = h("input", { value: s.stats?.file || "" });
const statsRet = h("input", { value: s.stats?.retention || "7d" });
// cache
const modelsData = await api.listModels();
const embedModels = (modelsData.models || []).filter((m) => m.kind === "embedding").map((m) => m.id);
const cacheEnabled = check(t("settings.cache.enabled"), s.cache?.enabled);
const cacheThresh = h("input", { type: "number", step: "0.01", value: s.cache?.threshold ?? 0.92 });
const cacheTtlIn = h("input", { value: s.cache?.ttl || "1h" });
const cacheEmbedSel = h(
"select",
{},
h("option", { value: "" }, "auto"),
...embedModels.map((id) => h("option", { value: id, selected: id === s.cache?.embed_model }, id))
);
const saveBtn = h("button", { class: "btn primary" }, t("common.save"));
saveBtn.addEventListener("click", async () => {
@@ -71,11 +83,19 @@ export async function renderSettings() {
retention: statsRet.value.trim() || "7d",
ring_size: s.stats?.ring_size ?? 500,
},
cache: {
enabled: cacheEnabled.cb.checked,
threshold: parseFloat(cacheThresh.value) || 0.92,
ttl: cacheTtlIn.value.trim() || "1h",
max_entries: s.cache?.max_entries ?? 1000,
embed_model: cacheEmbedSel.value || null,
},
};
// serde: drop null optionals
if (!patch.router.api_key_env) delete patch.router.api_key_env;
if (!patch.router.taxonomy_id) delete patch.router.taxonomy_id;
if (!patch.telemetry.otlp_endpoint_env) delete patch.telemetry.otlp_endpoint_env;
if (!patch.cache.embed_model) delete patch.cache.embed_model;
try {
const r = await api.putSettings(patch);
toast(r.needs_restart ? t("settings.needsRestart") : t("toast.saved"), r.needs_restart ? "warn" : "good");
@@ -170,6 +190,15 @@ export async function renderSettings() {
statsEnabled.node,
field(t("settings.stats.file"), statsFile),
field(t("settings.stats.retention"), statsRet)
),
h(
"div",
{ class: "card" },
h("div", { class: "card-head" }, h("h3", {}, t("settings.cache"))),
cacheEnabled.node,
h("div", { class: "row" }, field(t("settings.cache.threshold"), cacheThresh), field(t("settings.cache.ttl"), cacheTtlIn)),
field(t("settings.cache.embed"), cacheEmbedSel),
h("div", { class: "hint" }, t("settings.cache.note"))
)
),
h(