feat(routing): per-task rules on top of complexity tiers; hosted router is the default
The router has always returned two signals — task label and complexity tier — but only the tier picked models. [routing.tasks] adds `label → ordered models`: a matching rule's models lead the candidate chain and the tier chain follows as fallback, so pinning translations to a local model never weakens failover. Requests whose chain was headed by a task rule report route_source "router+task-rule". The Routing page grows a per-task editor built from the account's LIVE taxonomy (GET /v1/taxonomies via the new /admin/api/routing/labels proxy; built-in label set when the router is unreachable), with localized task names, each task's share of 7-day traffic, "by complexity (no rule)" as the explicit empty state, and custom labels for other taxonomies. Deleting a model scrubs it from task rules like it always did from tiers. Also fixes the default config pointing at a placeholder localhost:8080 router with a nonexistent taxonomy — a fresh install now targets the hosted allaigate router (IP endpoint, verify_tls off, data-assistant, 12 s oracle timeout) and degrades gracefully until a key is set. Verified end-to-end against the live router: translation → task-rule model, math → tier path. 16 new strings in all 7 languages.
This commit is contained in:
+66
-1
@@ -155,6 +155,7 @@ pub fn api_routes(admin_token: String) -> Router<SharedState> {
|
||||
)
|
||||
.route("/admin/api/models/:id/probe", post(probe_model))
|
||||
.route("/admin/api/routing", get(get_routing).put(put_routing))
|
||||
.route("/admin/api/routing/labels", get(get_routing_labels))
|
||||
.route(
|
||||
"/admin/api/protocols",
|
||||
get(get_protocols).put(put_protocols),
|
||||
@@ -744,7 +745,7 @@ fn strip_id_from_routing(routing: &mut RoutingCfg, id: &str) {
|
||||
if routing.default == id {
|
||||
routing.default = String::new();
|
||||
}
|
||||
for tt in routing.tiers.values_mut() {
|
||||
for tt in routing.tiers.values_mut().chain(routing.tasks.values_mut()) {
|
||||
match tt {
|
||||
crate::config::TierTargets::List(v) => v.retain(|x| x != id),
|
||||
crate::config::TierTargets::One(s) if s == id => {
|
||||
@@ -753,6 +754,11 @@ fn strip_id_from_routing(routing: &mut RoutingCfg, id: &str) {
|
||||
crate::config::TierTargets::One(_) => {}
|
||||
}
|
||||
}
|
||||
// an emptied task rule means "no rule", not "no candidates"
|
||||
routing.tasks.retain(|_, tt| match tt {
|
||||
crate::config::TierTargets::List(v) => !v.is_empty(),
|
||||
crate::config::TierTargets::One(s) => !s.is_empty(),
|
||||
});
|
||||
}
|
||||
|
||||
/// Whether any managed server or static model in `cfg` still uses `path` as its
|
||||
@@ -893,17 +899,70 @@ async fn probe_model(State(state): State<SharedState>, Path(id): Path<String>) -
|
||||
|
||||
async fn get_routing(State(state): State<SharedState>) -> ApiResult {
|
||||
let live = state.live();
|
||||
let tasks: HashMap<String, Vec<String>> = live
|
||||
.cfg
|
||||
.routing
|
||||
.tasks
|
||||
.iter()
|
||||
.map(|(k, v)| {
|
||||
(
|
||||
k.clone(),
|
||||
match v {
|
||||
TierTargets::List(l) => l.clone(),
|
||||
TierTargets::One(s) => vec![s.clone()],
|
||||
},
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
ok(json!({
|
||||
"tiers": tiers_to_map(&live.cfg.routing),
|
||||
"tasks": tasks,
|
||||
"default": live.cfg.routing.default,
|
||||
"policy": live.cfg.routing.policy,
|
||||
}))
|
||||
}
|
||||
|
||||
/// Task labels of the active taxonomy, straight from the router (`GET
|
||||
/// /v1/taxonomies`) — the UI builds the per-task rule editor from this, so a
|
||||
/// custom taxonomy shows its own labels without a gateway update.
|
||||
async fn get_routing_labels(State(state): State<SharedState>) -> ApiResult {
|
||||
let live = state.live();
|
||||
match live.router.taxonomies().await {
|
||||
Ok(Some(v)) => {
|
||||
let want = live.cfg.router.taxonomy_id.clone().unwrap_or_default();
|
||||
let mut labels: Vec<String> = Vec::new();
|
||||
if let Some(arr) = v["taxonomies"].as_array() {
|
||||
for tx in arr {
|
||||
let id = tx["taxonomy_id"].as_str().unwrap_or("");
|
||||
if want.is_empty() || id == want || labels.is_empty() {
|
||||
if let Some(ls) = tx["labels"].as_array() {
|
||||
let this: Vec<String> = ls
|
||||
.iter()
|
||||
.filter_map(|l| l.as_str().map(String::from))
|
||||
.collect();
|
||||
if id == want || labels.is_empty() {
|
||||
labels = this;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if labels.is_empty() {
|
||||
ok(json!({ "labels": Value::Null, "source": "router-empty" }))
|
||||
} else {
|
||||
ok(json!({ "labels": labels, "source": "router" }))
|
||||
}
|
||||
}
|
||||
_ => ok(json!({ "labels": Value::Null, "source": "unavailable" })),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct RoutingBody {
|
||||
#[serde(default)]
|
||||
tiers: HashMap<String, Vec<String>>,
|
||||
#[serde(default)]
|
||||
tasks: HashMap<String, Vec<String>>,
|
||||
default: String,
|
||||
#[serde(default)]
|
||||
policy: RoutingPolicy,
|
||||
@@ -917,6 +976,12 @@ async fn put_routing(State(state): State<SharedState>, Json(body): Json<RoutingB
|
||||
.into_iter()
|
||||
.map(|(k, v)| (k, TierTargets::List(v)))
|
||||
.collect(),
|
||||
tasks: body
|
||||
.tasks
|
||||
.into_iter()
|
||||
.filter(|(_, v)| !v.is_empty()) // empty rule = "по сложности", not a rule
|
||||
.map(|(k, v)| (k, TierTargets::List(v)))
|
||||
.collect(),
|
||||
default: body.default,
|
||||
policy: body.policy,
|
||||
};
|
||||
|
||||
+29
-4
@@ -396,6 +396,12 @@ pub struct RoutingCfg {
|
||||
pub default: String,
|
||||
#[serde(default)]
|
||||
pub policy: RoutingPolicy,
|
||||
/// Per-task rules: `task_label → ordered model list`. A matching rule's
|
||||
/// models go FIRST; the complexity-tier chain follows as fallback, so a
|
||||
/// task pin never weakens failover. Named field, so a tier cannot be
|
||||
/// called "tasks" — serde extracts it before the flattened tier map.
|
||||
#[serde(default, skip_serializing_if = "HashMap::is_empty")]
|
||||
pub tasks: HashMap<String, TierTargets>,
|
||||
}
|
||||
|
||||
/// A routing tier is either a list of model_ids or the string `default`.
|
||||
@@ -724,13 +730,16 @@ impl Default for Config {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
listen: "0.0.0.0:9000".to_string(),
|
||||
// The hosted allaigate router — the thing this gateway is built
|
||||
// around. Without a key the pipeline degrades gracefully to local
|
||||
// models and the console asks for one; nothing breaks.
|
||||
router: RouterCfg {
|
||||
enabled: true,
|
||||
url: "http://localhost:8080".to_string(),
|
||||
timeout_ms: default_router_timeout(),
|
||||
verify_tls: true,
|
||||
url: "https://138.226.222.209".to_string(),
|
||||
timeout_ms: 12_000, // hard prompts escalate to an oracle (~10 s)
|
||||
verify_tls: false, // the hosted endpoint is IP-bound
|
||||
api_key_env: Some("CORTIQ_ROUTER_KEY".to_string()),
|
||||
taxonomy_id: Some("general.task-type".to_string()),
|
||||
taxonomy_id: Some("data-assistant".to_string()),
|
||||
},
|
||||
route: RouteCfg::default(),
|
||||
models: Vec::new(),
|
||||
@@ -738,6 +747,7 @@ impl Default for Config {
|
||||
tiers: HashMap::new(),
|
||||
default: String::new(),
|
||||
policy: RoutingPolicy::default(),
|
||||
tasks: HashMap::new(),
|
||||
},
|
||||
breaker: BreakerCfg::default(),
|
||||
protocols: ProtocolsCfg {
|
||||
@@ -862,6 +872,21 @@ impl Config {
|
||||
}
|
||||
}
|
||||
}
|
||||
for (task, targets) in &self.routing.tasks {
|
||||
let list = match targets {
|
||||
TierTargets::List(v) => v.clone(),
|
||||
TierTargets::One(s) => vec![s.clone()],
|
||||
};
|
||||
for id in &list {
|
||||
if !ids.contains(id.as_str()) {
|
||||
anyhow::bail!(
|
||||
"config: routing.tasks.{} references unknown model id '{}'",
|
||||
task,
|
||||
id
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
+10
-2
@@ -232,8 +232,16 @@ impl Pipeline {
|
||||
match live.router.route(&text, prof).await {
|
||||
Ok(Some(dec)) => {
|
||||
let src = dec.source.clone();
|
||||
let tier = dec.complexity_tier.clone();
|
||||
(dec, src, live.routing.candidates(&tier))
|
||||
let (cands, task_rule) = live
|
||||
.routing
|
||||
.candidates_for(&dec.task_label, &dec.complexity_tier);
|
||||
// surface which rule chose the head of the chain
|
||||
let src = if task_rule {
|
||||
format!("{src}+task-rule")
|
||||
} else {
|
||||
src
|
||||
};
|
||||
(dec, src, cands)
|
||||
}
|
||||
Ok(None) | Err(_) => {
|
||||
tracing::warn!(
|
||||
|
||||
@@ -71,6 +71,23 @@ impl RouterClient {
|
||||
|
||||
/// Classify text. Returns `Ok(None)` on network error/timeout —
|
||||
/// the caller treats this as "router unavailable" and uses the default tier.
|
||||
/// `GET /v1/taxonomies` — the account's taxonomies with their label sets.
|
||||
/// `Ok(None)` when the router is disabled or has no key configured.
|
||||
pub async fn taxonomies(&self) -> anyhow::Result<Option<serde_json::Value>> {
|
||||
if self.url.is_empty() {
|
||||
return Ok(None);
|
||||
}
|
||||
let mut req = self.http.get(format!("{}/v1/taxonomies", self.url));
|
||||
if let Some(k) = &self.api_key {
|
||||
req = req.bearer_auth(k);
|
||||
}
|
||||
let resp = req.send().await?;
|
||||
if !resp.status().is_success() {
|
||||
anyhow::bail!("router /v1/taxonomies: HTTP {}", resp.status());
|
||||
}
|
||||
Ok(Some(resp.json().await?))
|
||||
}
|
||||
|
||||
pub async fn route(&self, text: &str, profile: &str) -> anyhow::Result<Option<RouteDecision>> {
|
||||
let mut body = serde_json::json!({
|
||||
"input": { "text": text },
|
||||
|
||||
+92
-17
@@ -3,38 +3,70 @@
|
||||
|
||||
use crate::config::{Config, TierTargets};
|
||||
|
||||
fn to_list(t: &TierTargets) -> Vec<String> {
|
||||
match t {
|
||||
TierTargets::List(v) => v.clone(),
|
||||
TierTargets::One(s) => vec![s.clone()],
|
||||
}
|
||||
}
|
||||
|
||||
pub struct RoutingTable {
|
||||
tiers: std::collections::HashMap<String, Vec<String>>,
|
||||
/// Per-task rules — a matching rule's models lead the candidate chain.
|
||||
tasks: std::collections::HashMap<String, Vec<String>>,
|
||||
default: String,
|
||||
}
|
||||
|
||||
impl RoutingTable {
|
||||
pub fn from_config(cfg: &Config) -> Self {
|
||||
let tiers = cfg
|
||||
.routing
|
||||
.tiers
|
||||
.iter()
|
||||
.map(|(tier, targets)| {
|
||||
let list = match targets {
|
||||
TierTargets::List(v) => v.clone(),
|
||||
TierTargets::One(s) => vec![s.clone()],
|
||||
};
|
||||
(tier.clone(), list)
|
||||
})
|
||||
.collect();
|
||||
Self {
|
||||
tiers,
|
||||
tiers: cfg
|
||||
.routing
|
||||
.tiers
|
||||
.iter()
|
||||
.map(|(tier, targets)| (tier.clone(), to_list(targets)))
|
||||
.collect(),
|
||||
tasks: cfg
|
||||
.routing
|
||||
.tasks
|
||||
.iter()
|
||||
.map(|(task, targets)| (task.clone(), to_list(targets)))
|
||||
.filter(|(_, v)| !v.is_empty())
|
||||
.collect(),
|
||||
default: cfg.routing.default.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Candidates for `(task, tier)`, in preference order, deduplicated:
|
||||
/// the task rule's models first (when one exists), then the tier chain —
|
||||
/// so a task pin adds preference without ever weakening failover.
|
||||
/// Returns `(candidates, task_rule_matched)`.
|
||||
pub fn candidates_for(&self, task: &str, tier: &str) -> (Vec<String>, bool) {
|
||||
let mut out: Vec<String> = Vec::new();
|
||||
let matched = if let Some(rule) = self.tasks.get(task) {
|
||||
out.extend(rule.iter().cloned());
|
||||
true
|
||||
} else {
|
||||
false
|
||||
};
|
||||
let tier_chain = self
|
||||
.tiers
|
||||
.get(tier)
|
||||
.cloned()
|
||||
.unwrap_or_else(|| vec![self.default.clone()]);
|
||||
for id in tier_chain {
|
||||
if !out.contains(&id) {
|
||||
out.push(id);
|
||||
}
|
||||
}
|
||||
out.retain(|id| !id.is_empty());
|
||||
(out, matched)
|
||||
}
|
||||
|
||||
/// Candidates for the given complexity tier, in preference order.
|
||||
/// If the tier is not defined, the sole candidate is `default`.
|
||||
pub fn candidates(&self, tier: &str) -> Vec<String> {
|
||||
self.tiers
|
||||
.get(tier)
|
||||
.cloned()
|
||||
.unwrap_or_else(|| vec![self.default.clone()])
|
||||
self.candidates_for("", tier).0
|
||||
}
|
||||
|
||||
/// Default model (used when the router is unavailable).
|
||||
@@ -46,3 +78,46 @@ impl RoutingTable {
|
||||
// TODO(v0.2): cost_aware mode and circuit breaker — pick the cheapest available
|
||||
// model no lower than min_class[tier], respecting max_cost_usd_per_request,
|
||||
// and skip models whose circuit breaker is open.
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn table(tasks: &[(&str, &[&str])], tiers: &[(&str, &[&str])], default: &str) -> RoutingTable {
|
||||
RoutingTable {
|
||||
tiers: tiers
|
||||
.iter()
|
||||
.map(|(k, v)| (k.to_string(), v.iter().map(|s| s.to_string()).collect()))
|
||||
.collect(),
|
||||
tasks: tasks
|
||||
.iter()
|
||||
.map(|(k, v)| (k.to_string(), v.iter().map(|s| s.to_string()).collect()))
|
||||
.collect(),
|
||||
default: default.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn task_rule_leads_tier_chain_follows() {
|
||||
let t = table(
|
||||
&[("translation", &["local"])],
|
||||
&[("low", &["cheap", "local"]), ("high", &["big"])],
|
||||
"cheap",
|
||||
);
|
||||
let (c, matched) = t.candidates_for("translation", "high");
|
||||
assert!(matched);
|
||||
assert_eq!(c, vec!["local", "big"]);
|
||||
// dedup: rule model already in the tier chain appears once, first
|
||||
let (c, _) = t.candidates_for("translation", "low");
|
||||
assert_eq!(c, vec!["local", "cheap"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_rule_falls_back_to_tier_then_default() {
|
||||
let t = table(&[], &[("low", &["a"])], "d");
|
||||
assert_eq!(t.candidates_for("qa", "low").0, vec!["a"]);
|
||||
let (c, matched) = t.candidates_for("qa", "unknown-tier");
|
||||
assert!(!matched);
|
||||
assert_eq!(c, vec!["d"]);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user