Add weighted multi-model routing

app/control_store.py

@@ -51,12 +51,46 @@ def buddy_claim_reserved(record):
 
 def validate_model(source, rule, models=None, known_models=(), *, legacy_scopes=False):
     _identifier(source, "模型规则 ID")
-    if not isinstance(rule, dict) or set(rule) - {"public_id", "upstream_id", "custom", "enabled", "keep_original", "region", "profile", "credential_ids", "credits_override"}:
+    if not isinstance(rule, dict) or set(rule) - {"public_id", "upstream_id", "routes", "custom", "enabled", "keep_original", "region", "profile", "credential_ids", "credits_override"}:
         raise ValueError("模型规则字段无效")
     clean = {"public_id": source, "upstream_id": source, "custom": False,
              "enabled": True, "keep_original": False,
              "region": None, "profile": None, "credential_ids": [], **rule}
     _identifier(clean["upstream_id"], "上游模型 ID")
+    routes = clean.get("routes")
+    if routes is not None:
+        if not isinstance(routes, list) or not routes or len(routes) > 100:
+            raise ValueError("routes 必须是 1–100 条路由列表")
+        normalized_routes = []
+        seen_routes = set()
+        weighted = False
+        for route in routes:
+            if not isinstance(route, dict) or set(route) - {"upstream_id", "weight", "enabled", "source"}:
+                raise ValueError("路由字段无效")
+            route_id = route.get("upstream_id")
+            _identifier(route_id, "路由上游模型 ID")
+            if route_id in seen_routes:
+                raise ValueError("路由上游模型重复")
+            seen_routes.add(route_id)
+            enabled = route.get("enabled", True)
+            if type(enabled) is not bool:
+                raise ValueError("路由 enabled 必须为布尔值")
+            source_kind = route.get("source", "manual")
+            if source_kind not in ("builtin", "cloud", "manual"):
+                raise ValueError("路由 source 无效")
+            weight = route.get("weight")
+            if weight is not None:
+                if isinstance(weight, bool) or not isinstance(weight, (int, float)) \
+                        or not math.isfinite(weight) or weight < 0 or weight > 1000:
+                    raise ValueError("路由 weight 必须是 0–1000 的有限数字")
+                weight = float(weight)
+                weighted = True
+            normalized_routes.append({"upstream_id": route_id, "weight": weight,
+                                      "enabled": enabled, "source": source_kind})
+        if weighted and not any((route["weight"] or 0) > 0 and route["enabled"]
+                                for route in normalized_routes):
+            raise ValueError("加权路由至少需要一条正权重路线")
+        clean["routes"] = normalized_routes
     if type(clean["custom"]) is not bool or (clean["custom"] and clean["keep_original"]):
         raise ValueError("自建模型不能公开内部规则标识")
     _identifier(clean["public_id"], "公开模型 ID")

app/gateway_management.py

@@ -372,9 +372,12 @@ def admin_model_inventory(self):
             rates = model_rates.details(self.gateway, upstream, sorted(served.get(upstream, ())))
             for profile, value in (declared.get(upstream) or {}).items():
                 rates[profile]["declared"] = value
-            result.append({**target, **metadata, **prices, **rule, "id": source, "rate_details": rates,
-                           "available_credentials": len(preview["candidates"]),
-                           "available": bool(preview["candidates"])})
+            row = {**target, **metadata, **prices, **rule, "id": source, "rate_details": rates,
+                   "available_credentials": len(preview["candidates"]),
+                   "available": bool(preview["candidates"])}
+            if "routes" not in rule and source in (self.CONFIG.get("model_route_groups") or {}):
+                row["routes"] = self.CONFIG["model_route_groups"][source]
+            result.append(row)
         return sorted(result, key=lambda row: row["id"])
 
     def admin_rate_probe_model(self, source):

app/media_store.py

@@ -114,13 +114,13 @@ def _create_schema(self):
                 model TEXT, created REAL NOT NULL, updated REAL NOT NULL, status TEXT NOT NULL,
                 media_id TEXT, error TEXT, credit REAL, public_model TEXT, api_key TEXT);
             CREATE INDEX IF NOT EXISTS tasks_created ON tasks(created);
-            CREATE INDEX IF NOT EXISTS tasks_api_key ON tasks(api_key, created);
         """)
 
     def _migrate_schema(self):
         """Apply additive task metadata and remove the obsolete asset expiry column."""
         version = self._db.execute("PRAGMA user_version").fetchone()[0]
         if version >= SCHEMA_VERSION:
+            self._db.execute("CREATE INDEX IF NOT EXISTS tasks_api_key ON tasks(api_key, created)")
             return
         if version < 1:
             columns = {row[1] for row in self._db.execute("PRAGMA table_info(assets)").fetchall()}

app/model_policy.py

@@ -1,12 +1,16 @@
 """Local model policies layered over account-owned upstream capabilities."""
 from contextvars import ContextVar
+from math import gcd
+import threading
 
 from fastapi import HTTPException
 
 from app import growth_tasks, perf_metrics, travel, embeddings
 from app.model_capabilities import route_metadata
 from app.api_keys import api_principal
 _request_policy = ContextVar("gateway_model_policy", default=None)
+_route_counter_lock = threading.Lock()
+_route_counters = {}
 
 
 def _principal():
@@ -126,13 +130,87 @@ def rule_for(config, source, policy=None):
     return {**default_rule(source), **_policy(config, policy)["models"].get(source, {})}
 
 
-def source_for(config, name):
+def _route_candidates_for_rule(rule: dict, auto_routes=None, counter_key=None) -> list[str]:
+    routes = rule.get("routes")
+    if not isinstance(routes, list) and isinstance(auto_routes, list):
+        routes = auto_routes
+    if not isinstance(routes, list):
+        return [rule["upstream_id"]]
+    enabled = [route for route in routes if route.get("enabled", True)]
+    if not enabled:
+        return []
+    weighted = any(route.get("weight") is not None for route in enabled)
+    if not weighted:
+        return [route["upstream_id"] for route in enabled]
+    weights = [route.get("weight") if route.get("weight") is not None else 1.0 for route in enabled]
+    scale = 1000
+    while scale < 1_000_000 and any(abs(weight * scale - round(weight * scale)) > 1e-9 for weight in weights):
+        scale *= 10
+    units = [max(0, int(round(weight * scale))) for weight in weights]
+    total = sum(units)
+    if total <= 0:
+        return []
+    key = counter_key or rule.get("public_id") or rule.get("upstream_id")
+    step = units[0]
+    for unit in units[1:]:
+        step = gcd(step, unit)
+    with _route_counter_lock:
+        offset = (_route_counters.get(key, 0) * step) % total
+        _route_counters[key] = _route_counters.get(key, 0) + 1
+    cursor = 0
+    selected = 0
+    for index, weight in enumerate(units):
+        cursor += weight
+        if offset < cursor:
+            selected = index
+            break
+    return [enabled[selected]["upstream_id"]] + [route["upstream_id"] for i, route in enumerate(enabled) if i != selected]
+
+
+def route_candidates(config, public_model: str, policy=None) -> list[str]:
+    """Return enabled upstream candidates, preserving order or applying route weights."""
+    state = _request_policy.get()
+    if state and public_model in (state.get("public_model"), state.get("upstream")):
+        return list(state.get("route_candidates") or ())
+    source = source_for(config, public_model, policy)
+    return _route_candidates_for_rule(rule_for(config, source, policy),
+                                      (config.get("model_route_groups") or {}).get(source),
+                                      counter_key=public_model)
+
+
+def route_ids(config, public_model: str, policy=None) -> list[str]:
+    """Return all enabled route IDs without applying weighted selection."""
+    source = source_for(config, public_model, policy)
+    rule = rule_for(config, source, policy)
+    routes = rule.get("routes")
+    if not isinstance(routes, list):
+        routes = (config.get("model_route_groups") or {}).get(source)
+    if not isinstance(routes, list):
+        return [rule["upstream_id"]]
+    return [route["upstream_id"] for route in routes if route.get("enabled", True)]
+
+
+def bind_route(config, upstream: str, index: int | None = None) -> None:
+    """Bind a retry attempt to one upstream while retaining the aggregate policy rule."""
+    state = _request_policy.get()
+    if state is not None:
+        state["upstream"] = upstream
+        if index is not None:
+            state["route_index"] = index
+
+
+def source_for(config, name, policy=None):
     """Resolve a public model name or automatic alias to its policy source."""
     if name == "default-model":
         return "auto"
     if name in (embeddings.MODEL_ID, embeddings.PUBLIC_MODEL_ID):
         return embeddings.MODEL_ID
-    data = snapshot(config)
+    data = policy.data if policy is not None else snapshot(config)
+    aliases = config.get("model_route_aliases") or {}
+    if name in aliases:
+        saved = data.get("models", {}).get(name) or {}
+        if not saved.get("custom") and not isinstance(saved.get("routes"), list):
+            return aliases[name]
     for source, rule in data.get("models", {}).items():
         if rule.get("public_id", source) == name:
             return source
@@ -180,9 +258,10 @@ def resolve(config, public_model):
         return state["upstream"]
     data = snapshot(config)
     source = next((key for key, value in data["models"].items()
-                   if value.get("public_id", key) == public_model),
-                  embeddings.MODEL_ID if public_model in (embeddings.MODEL_ID, embeddings.PUBLIC_MODEL_ID)
-                  else public_model)
+                   if value.get("public_id", key) == public_model), None)
+    if source is None:
+        source = (embeddings.MODEL_ID if public_model in (embeddings.MODEL_ID, embeddings.PUBLIC_MODEL_ID)
+                  else (config.get("model_route_aliases") or {}).get(public_model, public_model))
     rule = {**default_rule(source), **data["models"].get(source, {})}
     if principal is not None and principal.models is not None and _restriction_source(source) not in principal.models:
         raise HTTPException(status_code=403, detail={"error": {
@@ -192,12 +271,17 @@ def resolve(config, public_model):
         _deny("模型已停用")
     if source == public_model and (rule["custom"] or (rule["public_id"] != source and not rule["keep_original"])):
         _deny("原模型 ID 已停用，请使用已配置的对外 ID", "model_renamed")
-    if state is None and rule["upstream_id"] != source:
+    candidates = _route_candidates_for_rule(rule, (config.get("model_route_groups") or {}).get(source),
+                                            counter_key=public_model)
+    if state is None and (not candidates or candidates[0] != source) and rule["upstream_id"] != source:
         raise HTTPException(status_code=503, detail={"error": {"message": "模型路由上下文未初始化",
                             "type": "server_error", "code": "model_policy_context_missing"}})
+    if not candidates:
+        _deny("模型路由已全部停用")
     if state is not None:
-        state.update(source=source, public_model=public_model, upstream=rule["upstream_id"], rule=rule)
-    return rule["upstream_id"]
+        state.update(source=source, public_model=public_model, upstream=candidates[0],
+                     route_candidates=candidates, route_index=0, rule=rule)
+    return candidates[0]
 
 
 def check_resolved(config, upstream):
@@ -257,11 +341,18 @@ def public_details(gateway, region=None):
     entries = pool.entries() if pool is not None else []
     out = []
     for source in dict.fromkeys([*known, *policies]):
+        if source in (config.get("model_route_aliases") or {}):
+            saved = policies.get(source) or {}
+            if not saved.get("custom") and not isinstance(saved.get("routes"), list):
+                continue
         rule = rule_for(config, source, policy)
         upstream = rule["upstream_id"]
-        item = known.get(upstream)
+        route_names = route_ids(config, source, policy)
+        item = next((known.get(route) for route in route_names if known.get(route) is not None), None)
         if not rule["enabled"] or item is None:
             continue
+        if not route_names:
+            continue
         media = bool(item.get("media_kinds"))
         embedding = bool(item.get("embedding"))
         reranking = bool(item.get("rerank"))
@@ -301,9 +392,10 @@ def public_details(gateway, region=None):
                                 and route_allowed(config, entry, upstream, rule=rule, policy=policy)):
                             candidates = [entry]
             else:
-                candidates = [entry for entry in entries if route_allowed(
-                    config, entry, upstream, rule=rule, policy=policy)
-                    and pool._eligible(entry, upstream, region=region)]
+                candidates = [entry for entry in entries if any(
+                    route_allowed(config, entry, route, rule=rule, policy=policy)
+                    and pool._eligible(entry, route, region=region, rule=rule, policy=policy)
+                    for route in route_names)]
         finally:
             _request_policy.reset(token)
         if not candidates and (pool is not None or media or embedding or reranking):
@@ -313,16 +405,18 @@ def public_details(gateway, region=None):
         if accounts is not None:
             for entry in candidates:
                 profile = entry.get("profile")
-                for model in gateway._published_account_scope(
-                        accounts.get(entry.get("account_key")) or {}, "serves", model_id=gateway._upstream_model(upstream, profile)) or []:
-                    if model.get("id") == gateway._upstream_model(upstream, profile):
-                        price = gateway._multiplier_value(model.get("credits"))
-                        if price is not None and price <= prices.get(profile, price):
-                            prices[profile] = price
-                            if model.get("credits_source") == "measured":
-                                measured.add(profile)
-                            else:
-                                measured.discard(profile)
+                for route in route_names:
+                    route_id = gateway._upstream_model(route, profile)
+                    for model in gateway._published_account_scope(
+                            accounts.get(entry.get("account_key")) or {}, "serves", model_id=route_id) or []:
+                        if model.get("id") == route_id:
+                            price = gateway._multiplier_value(model.get("credits"))
+                            if price is not None and price <= prices.get(profile, price):
+                                prices[profile] = price
+                                if model.get("credits_source") == "measured":
+                                    measured.add(profile)
+                                else:
+                                    measured.discard(profile)
         else:
             allowed_profiles = {entry.get("profile") for entry in candidates}
             prices = {profile: price for profile, price in item["credits_by_profile"].items()
@@ -337,6 +431,8 @@ def public_details(gateway, region=None):
             best = min(prices.values(), default=None)
             row = {"id": name, "credits": best, "credits_by_profile": prices,
                    **route_metadata(gateway, upstream, entries=candidates, rule=rule, region=region)}
+            if source in (config.get("model_route_groups") or {}) and "routes" not in rule:
+                row["routes"] = config["model_route_groups"][source]
             for key in ("media_kinds", "media_sources", "media_defaults", "embedding", "rerank"):
                 if key in item:
                     row[key] = item[key]