14 Commits
Author SHA1 Message Date
tzt 0669129833 chore(proxy): .gitignore 登记 extra/ 参考项目目录(不入库) 2026-09-19 10:18:16 +08:00
tzt aa9db718e6 feat(proxy): T-X13 采纳 ai-model-router 五维评分选模——能力过滤后确定性排序 + 拒绝理由留痕
- routes:能力位硬过滤(T-X6 语义不变)后对存活候选五维评分——
  cost_efficiency=1/(1+均价*2)、capability=上下文余量(不满足为 0)、
  speed/quality=档位秩比互补(local 快 / premium 优)、reliability 权重占位;
  重定向目标从'取首元素'改为'取最高分';降级链按评分降序稳定重排
- 被淘汰候选带人话拒绝理由(vision/tools/上下文窗口三类)写入评分留痕环
  (deque 上限 100),/admin/stats 新增 route_scored 段透出(管理面可观测)
- 硬过滤语义不变:评分只改变链内顺序,不改变谁能存活
- 新增 tests/test_route_score.py 4 项(拒绝理由/排序偏好/同分决胜/留痕)
2026-09-19 10:15:21 +08:00
tzt b2b7a64cb3 feat(proxy): T-X12 采纳 enterprise-ai-gateway 双桶限流——TPM token 桶 + 语义缓存命中率透出
- auth.RateLimiter.allow_tokens:TPM 令牌桶(容量=tpm_cap,连续回流 tpm/60 每秒,
  与 RPM 桶共用 per-key 锁、判定相互独立;tpm_cap<=0 不限)
- ProxyConfig.limits.tpm_per_key(默认 60000,0=不限);chat_completions 在并发槽
  之后做 TPM 预扣判定(est=字符/3 与计费同口径),拒绝时释放并发槽并回 429
  (文案区分并发/RPM/TPM 三种原因)
- semcache.stats 增 misses 与 hit_rate;/admin/stats 新增 semcache 段透出
  (entries/hits_exact/hits_semantic/misses/hit_rate,与账本 h_g 互补)
- 新增 tests/test_tpm_limiter.py 6 项(容量/回流/禁用/双桶独立/key 隔离/命中率)
2026-09-19 10:09:56 +08:00
tzt f04a6c4c43 feat(proxy): T-X11 采纳 cortiq shadow 旁路评审——live T1 本地应答异步云端评审喂晋升表
- routes:live 分流 T1 响应后按 sample_rate 采样旁路——premium 条目以评审提示词
  给本地答案打 PASS/FAIL,经 PromotionTable.observe 写入 T-X5 晋升表
  (打通晋升质量信号的自动来源,夜间 labeler 之外的实时通路)
- 零客户端延迟:asyncio.create_task 旁路 + wait_for 超时 + Semaphore(2) 并发上限;
  上游故障/解析失败记 ERROR,不算 FAIL(评审器不可用不惩罚本地模型);
  仅非流式响应采样(流式无同步答案文本),默认关闭(sense.shadow_review.enabled)
- 模块级计数字典 sampled/pass/fail/error + reset_shadow_review_stats() 测试钩子
- 新增 tests/test_shadow_review.py 5 项(默认关/条目选取/PASS-FAIL-ERROR 解析/
  晋升表喂入/关闭态零副作用),全假上游不依赖真实模型
2026-09-19 10:03:38 +08:00
tzt e5470a3715 feat(proxy): T-X10 采纳 cortiq 语义缓存路由签名分桶——不同路由意图不互串答案
- normalizer.canonical_hash 增加可选 route_sig 段:键 = bucket|doc_version|sig|sha256
  (旧三段格式向后兼容,旧条目随 TTL 自然淘汰)
- ProxyConfig 新增 semcache.route_sig_scope:capabilities(默认,vision/tools 需求
  签名)/ model(按模型隔离)/ none(旧行为);非法值回落 capabilities
- semcache:签名升级为条目属性并分区 L2 语义扫描(仅键分桶不够——语义层仍会
  跨签名命中);签名从缓存键第四段解析,重启重建零 schema 变更;
  晋升别名键携带签名段;route_sig=None 的旧调用零过滤完全兼容
- routes:lookup/put 共用同一 norm_hash(消除 put 侧重复哈希),签名贯穿两层
- 新增 tests/test_route_sig.py 6 项(键格式/键空间分割/scope 三态/两层隔离/
  重建存活/旧调用兼容)
2026-09-19 09:59:15 +08:00
tzt 9ce8718812 chore(proxy): 外部优点融入前置锚点——本轮五项目吸收的回滚基线 2026-09-19 09:16:52 +08:00
tzt e00e2cd2b1 feat(proxy): 架构与算法优化二轮——sense 运行时单例激活 T-X3、打码密钥泄漏修复、账本热路径 2.4x
- gateway/proxy/routes.py:sense 运行时(SenseStore+Grader+晋升表)按配置单例化,
  原每请求重建导致:每请求跑全量 DDL、重读工件文件,且 T-X3 60s 决策缓存随
  Grader 丢弃、命中率恒为 0(本轮最大收益,激活既有已测组件);
  T1 审计抽样修复原 NameError×3 被 except 吞掉(load_config/get_review/request_id
  均未定义、§9.4 从未入队)——request_id 上提、review 队列经 review_getter 注入
  (api.py 反向依赖解除)、sample_rate 走 load_config
- gateway/model_pool.py:新增 usable_entries() 未打码内通道;routes 的降级链/
  预算降档/能力位重定向/档位映射四处改走该通道——修复打码 api_key 流入上游
  Bearer 头导致带密钥条目重定向必 401 的隐性缺陷(HTTP 管理面仍用打码 list())
- gateway/proxy/ledger.py + billing.py:共享持久连接(原每操作新建,全链路
  每请求 3-5 次 connect)+ 日重置 UPDATE 每实例每日一次短路(原每请求全表扫描
  抢写锁);check_and_count A/B 0.75→0.31 ms/op(2.4x);新增 usage_stats()
  SQL 聚合,/admin/stats 从 list_usage(limit=50 万) Python 四遍扫描改为下推聚合
  (微基准 ~28x,随流水线性扩大)
- gateway/sense/grader.py:决策留痕 insert_decision 移入 asyncio.to_thread
  (原同步 sqlite 写直接跑在事件循环线程,高并发阻塞网关;与观察写批量队列同等保护)
- 全量 474 项两轮复核全绿(基线 474)
2026-09-18 23:59:44 +08:00
tzt 781f8aaadf chore(proxy): 优化前置锚点——本轮架构/算法优化的回滚基线 2026-09-18 23:03:04 +08:00
tzt 1aaa05cacc feat(sense): T-X5 云端评判晋升表×conformal 双闸门——本地档自动接管(采纳 cortiq promotion)
- gateway/sense/promotion.py:PromotionTable 状态机(candidate/promoted/demoted)
  * 晋升门:n_total>=n_min 且 通过率>=promote_lb 且 soak 浸泡期满足
  * 退化:promoted 期间通过率<demote_lb 自动降级;降级后可凭数据恢复
  * 一票否决:tier=T3 观察不进通过率统计,已晋升者立即降级
  * 持久化 sense_promotion 表(CREATE IF NOT EXISTS 幂等)
- grader:live 模式对已晋升标签的 T2 决策(无 T3 信号/非 fallback)升级 T1,
  hard_gates.promotion_applied 如实标注;T1/T3/fallback 路径不受影响
- 回填链路:T1 审计抽样带 promo:<label> 标签 -> 人工 verdict approve=ok
  经 /review/{id} 提交时回填晋升表(失败不影响审核主流程)
- 双闸门语义:conformal 阈值保证单条决策风险率,晋升表保证标签级接管节奏;
  collect 数据不足(n_min 未满)不晋升,全部迁移可经 T-X4 留痕审计

pytest 474 passed(T-X4 后 466 + 8)
2026-09-18 23:01:52 +08:00
tzt 7e01dc9b16 feat(sense): T-X4 可解释决策留痕三元结构 + 修复 sense live 分流潜伏 bug(采纳 ai-model-router 决策模型)
- store:新增 sense_decisions 表(CREATE IF NOT EXISTS 幂等迁移),
  insert_decision / list_decisions(JSON 字段反序列化),q_hash 不落原文(D-G6)
- grader:全模式落库决策三元——reasons(§8 判定顺序最小完备集)、
  candidate_scores(线性头 probs)、rejected(落选档位+原因);
  留痕失败静默不影响决策主链路
- 修复:chat_completions 的 sense 分流引用未导入的 settings_store,NameError
  被 except 吞掉导致 D-G7 live 分流从未生效;改为 build_proxy_router 显式注入
  settings_provider(api.py 传 settings_store),缺省 None 行为安全
- 新增 /proxy/admin/sense-decisions 管理面查询端点(sense 未启用返回空集)
- 回归测试:注入 settings_provider 后 x-campus-tier 头出现且 T1 决策经
  管理面可查(修复前该头永远缺失)

pytest 466 passed(T-X6 后 461 + 5)
2026-09-18 22:55:49 +08:00
tzt 5078721df4 feat(proxy): T-X6 模型池能力位过滤——vision/tools/context 硬过滤(采纳 cortiq capabilities)
- model_pool:条目新增 capabilities{vision,tools,context_window},
  归一化缺省全兼容(老条目行为零变化;非法形态回落缺省);
  filter_by_capabilities 硬过滤(context_window=0 视为不限)
- proxy routes:_request_needs 从请求体推断需求(多模态 image_url → vision、
  tools 非空 → tools、上下文需求 = 字符/3 + min(max_tokens,4096) 与预扣同口径);
  条目不满足时在池内重定向到首个合格条目
- 与 T-X1/T-X2 组合语义:能力重定向 → 预算降档 → 降级链,逐级独立不互扰

pytest 461 passed(T-X3 后 455 + 6)
2026-09-18 22:45:50 +08:00
tzt 27a092a3f8 feat(sense): T-X3 路由决策缓存——live 模式 embed+线性头去重(采纳 cortiq 决策哈希缓存)
- gateway/sense/decision_cache.py:DecisionCache(sha256(scope+文本) 键、
  LRU + TTL 60s、4096 条上限、时钟可注入),对齐 auth._AuthCache 进程内模式
- Grader:仅 live 模式缓存纯决策负载(probs/head_version/vec);
  命中跳过 embed + 线性头预测;tier 仍按当前 conformal 阈值即时重算;
  观察落盘(observer.log)不因缓存命中跳过——校准数据完整性不受影响
- collect/shadow 模式不走缓存(校准必须全量);invalidate() 同步清空缓存
  (阈值/工件切换即时生效,TTL 仅兜底陈旧)

pytest 455 passed(T-X2 后 447 + 8)
2026-09-18 22:41:15 +08:00
tzt a043493548 feat(proxy): T-X2 上游有序降级链 + failover 三元组如实上报(采纳 cortiq tier 链)
- routes._fallback_chain:主条目之外的启用真实后端按(档位升序,单价和升序)
  构成有序候选链(≤3),排除 mock/停用/同模型,仅作 D-P4 首 token 前 failover
- _stream_response/_json_response 接入链式候选(原 [entry,entry] 同条目重试升级为真降级链)
- upstream.stream:failover 发生即记 sink[upstream_fallback] 与模块级统计;
  JSON 路径经 X-Upstream-Fallback/Original/Used/Reason 四头如实上报
  (流式路径头已发出不可追溯,由统计聚合暴露)
- /proxy/admin/stats 新增 upstream_failover 块(total + recent 20 条)

pytest 447 passed(T-X1 后 443 + 4)
2026-09-18 22:35:53 +08:00
tzt aa7cb0704c feat(proxy): T-X1 预算四档渐进干预(采纳 ai-model-router budgets 设计)
- billing.BillingMixin 新增 budget_mode(student, est, ts):整数基点判定
  normal(<80%) / optimize(>=80%) / cheap(>=95%) / block(>100%);
  cap<=0 不限额恒 normal;学生缺失 normal(扣费与拒绝权威仍在 try_hold)
- routes._run_chat:optimize/cheap 档自动降档池条目(_downgrade_entry,
  TIERS 排名制降档、杜绝反向升档、跳过停用与 mock),重算 model/est;
  非正常档位经 X-Budget-Mode 响应头如实上报(缓存命中路径同样携带)
- 黄金用例锁死边界:7999/8000/9499/9500/10000/10001bp 九组参数化断言,
  全整数运算无浮点漂移;跨日重置与不限额口径与 try_hold 一致

pytest 443 passed(基线 425 + 18)
2026-09-18 22:28:50 +08:00
26 changed files with 2490 additions and 118 deletions
+3
View File
@@ -79,3 +79,6 @@ bench_pool.json
# 分支快照 worktree 临时目录
.wt/
# 参考项目目录(不入库,仅作设计吸收来源)
extra/
+19 -1
View File
@@ -208,7 +208,9 @@ try:
from gateway.proxy.config import build_proxy_config
_proxy_cfg = build_proxy_config(settings_store().to_dict())
if _proxy_cfg.enabled:
app.include_router(build_proxy_router(_proxy_cfg, get_pool()))
app.include_router(build_proxy_router(_proxy_cfg, get_pool(),
settings_provider=settings_store,
review_getter=get_review))
from gateway.proxy.routes import install_error_handlers
install_error_handlers(app)
except Exception as _pe: # pragma: no cover - 代理层装配失败不拖垮主应用
@@ -450,6 +452,22 @@ try:
raise HTTPException(status_code=400, detail=str(e))
if not ok:
raise HTTPException(status_code=404, detail=f"审核记录不存在或已审核: {review_id}")
# 晋升表回填(T-X5):带 promo 标签的 sense 审计样本,人工 verdict
# approve=ok -> 晋升表累计;失败不影响审核主流程
try:
row = get_review().get(review_id)
promo = next((t[len("promo:"):] for t in (row.get("tags") or [])
if isinstance(t, str) and t.startswith("promo:")), None)
if promo:
from gateway.sense.config import build_sense_config
from gateway.sense.promotion import PromotionTable
from gateway.sense.store import SenseStore
scfg = build_sense_config(settings_store().to_dict())
if scfg.enabled:
PromotionTable(SenseStore.init_db(scfg.db_path)).observe(
promo, ok=(verdict == "approve"))
except Exception:
pass
return {"ok": True, "review_id": review_id, "verdict": verdict}
# ---------------- 模型池(多价位异构模型) ----------------
+55 -2
View File
@@ -13,7 +13,7 @@ import json
import re
import threading
from pathlib import Path
from typing import Any, Dict, Optional
from typing import Any, Dict, List, Optional
_POOL_PATH = Path(__file__).resolve().parent.parent / "config" / "model_pool.json"
@@ -28,6 +28,7 @@ ENTRY_FIELDS = {
"id", "name", "tier", "backend", "base_url", "model", "api_key",
"price_in", "price_out", "temperature", "max_tokens", "enabled",
"provider", "in_hit_price", # 代理层扩展(D-P3):usage 归一化 / 按命中价选上游
"capabilities", # 能力位(T-X6):vision/tools/context_window
}
# 单价默认值($/1M tokens);local 档为 0
@@ -41,6 +42,45 @@ def _empty_pool() -> Dict[str, Any]:
}
def _normalize_capabilities(raw: Any) -> Dict[str, Any]:
"""能力位归一(T-X6):缺省全兼容。
- vision/tools:缺省 True(未声明 = 不设限,老条目行为不变);
- context_window:缺省 0 = 不按上下文长度过滤。
"""
if not isinstance(raw, dict):
return {"vision": True, "tools": True, "context_window": 0}
try:
ctx = int(raw.get("context_window", 0) or 0)
except (TypeError, ValueError):
ctx = 0
return {"vision": bool(raw.get("vision", True)),
"tools": bool(raw.get("tools", True)),
"context_window": max(0, ctx)}
def filter_by_capabilities(entries: List[Dict[str, Any]], *,
need_vision: bool = False, need_tools: bool = False,
min_context_tokens: int = 0) -> List[Dict[str, Any]]:
"""能力位硬过滤(T-X6,采纳 cortiq capabilities 过滤)。
依次校验 vision/tools 声明位与 context_window >= 提示+输出预估
(context_window=0 视为不限)。返回保持原顺序的合格条目子集。
"""
out: List[Dict[str, Any]] = []
for e in entries:
cap = e.get("capabilities") or _normalize_capabilities(None)
if need_vision and not cap.get("vision", True):
continue
if need_tools and not cap.get("tools", True):
continue
ctx = int(cap.get("context_window") or 0)
if min_context_tokens > 0 and 0 < ctx < min_context_tokens:
continue
out.append(e)
return out
class PoolError(ValueError):
"""池条目/角色配置非法。"""
@@ -76,13 +116,23 @@ class PoolStore:
# ---------- 条目 CRUD ----------
def list(self) -> Dict[str, Any]:
"""返回完整池(api_key 打码)。"""
"""返回完整池(api_key 打码,供 HTTP 管理面)。"""
with self._lock:
return {
"roles": dict(self._data["roles"]),
"entries": [self._masked(e) for e in self._data["entries"]],
}
def usable_entries(self) -> List[Dict[str, Any]]:
"""未打码条目快照——仅供代理转发链内部使用(failover/降档/能力重定向),
绝不进入任何 HTTP 响应。
背景:此前降级链/能力重定向从 list() 取打码条目,带 api_key 的条目
被重定向时向上游发送打码密钥(必 401),T-X1/T-X2/T-X6 链路静默劣化。
"""
with self._lock:
return [dict(e) for e in self._data["entries"]]
def get(self, entry_id: str) -> Optional[Dict[str, Any]]:
with self._lock:
for e in self._data["entries"]:
@@ -203,6 +253,7 @@ class PoolStore:
max_tokens = int(entry.get("max_tokens", 4096))
except (TypeError, ValueError):
max_tokens = 4096
capabilities = _normalize_capabilities(entry.get("capabilities"))
return {
"id": eid,
"name": str(entry.get("name") or entry.get("model") or eid),
@@ -220,6 +271,8 @@ class PoolStore:
"provider": (str(entry["provider"]) if entry.get("provider") else "openai"),
"in_hit_price": (float(entry["in_hit_price"]) if entry.get("in_hit_price") is not None
else round(price_in / 30.0, 6)),
# 能力位(T-X6):缺省全兼容(不破坏既有条目/选型行为)
"capabilities": capabilities,
}
@staticmethod
+22
View File
@@ -89,6 +89,7 @@ class RateLimiter:
def __init__(self, concurrent_per_key: int = 2):
self._tokens: Dict[int, tuple[float, float]] = {} # key_id -> (tokens, last_ts)
self._tpm: Dict[int, tuple[float, float]] = {} # key_id -> (tokens, last_ts)
self._locks: Dict[int, threading.Lock] = {}
self._sems: Dict[int, threading.Semaphore] = {}
self._global = threading.Lock()
@@ -108,6 +109,27 @@ class RateLimiter:
self._tokens[key_id] = (tokens - 1.0, now)
return True
def allow_tokens(self, key_id: int, est_tokens: int, tpm_cap: int) -> bool:
"""TPM 令牌桶放行判定(T-X12,采纳 enterprise-ai-gateway 双桶设计)。
容量 = tpm_cap,回流速率 = tpm/60 每秒(连续时间回流补充);
est_tokens 为本次预扣估算(字符/3,与预扣计费同口径)。
tpm_cap <= 0 视为不限流;与 RPM 桶共用 per-key 锁,判定相互独立。
"""
if tpm_cap <= 0:
return True
now = time.monotonic()
with self._global:
lock = self._locks.setdefault(key_id, threading.Lock())
with lock:
tokens, last = self._tpm.get(key_id, (float(tpm_cap), now))
tokens = min(float(tpm_cap), tokens + (now - last) * (tpm_cap / 60.0))
if tokens < est_tokens:
self._tpm[key_id] = (tokens, now)
return False
self._tpm[key_id] = (tokens - est_tokens, now)
return True
def acquire_slot(self, key_id: int) -> bool:
"""并发槽(非阻塞);返回 False = 超并发上限(429)。"""
with self._global:
+72
View File
@@ -21,6 +21,10 @@ _SQL_HOLD = """UPDATE students SET
spent_today_milli = ?, spent_date = ?
WHERE id = ? AND balance_milli >= ?"""
# 预算四档阈值(T-X1,基点 bp:1% = 100bp;整数比较,杜绝浮点边界漂移)
BUDGET_OPTIMIZE_BP = 8000 # >= 80% 建议降一档
BUDGET_CHEAP_BP = 9500 # >= 95% 强制最低档
_SQL_HOLD_MARK = """INSERT INTO usage_ledger
(request_id, ts, key_id, model, bucket, charged_milli, status)
VALUES (?, ?, ?, ?, ?, ?, 'holding')"""
@@ -132,6 +136,41 @@ class BillingMixin:
" WHERE request_id = ?", (status, request_id))
return True
# ---------- 预算四档(T-X1) ----------
def budget_mode(self, student_id: int, est_milli: int, ts: float) -> str:
"""预算档位判定:normal / optimize / cheap / block。
以「当日已 spent + 本次预估」占日上限的比例判定(基点 bp,全整数运算,
无浮点边界漂移;黄金用例锁死):
< 8000bp(80%) -> normal(现行行为)
>= 8000bp(80%) -> optimize(调用方应降一档模型)
>= 9500bp(95%) -> cheap(调用方应强制最低档)
> 10000bp(100%) -> block(try_hold 的硬拒绝语义兜底)
daily_cap_milli <= 0 视为不限额 -> 恒 normal;学生不存在 -> normal
(try_hold 才是扣费与拒绝的唯一权威,本方法只做档位建议)。
"""
today = _today(ts)
with self._lock, self._connect() as conn:
stu = conn.execute(
"SELECT daily_cap_milli, spent_today_milli, spent_date"
" FROM students WHERE id = ?", (student_id,)).fetchone()
if stu is None:
return "normal"
cap = int(stu["daily_cap_milli"] or 0)
if cap <= 0:
return "normal"
spent = int(stu["spent_today_milli"] or 0) \
if stu["spent_date"] == today else 0
projected = spent + max(0, int(est_milli))
projected_bp = projected * 10000 // cap
if projected_bp > 10000:
return "block"
if projected_bp >= BUDGET_CHEAP_BP:
return "cheap"
if projected_bp >= BUDGET_OPTIMIZE_BP:
return "optimize"
return "normal"
# ---------- 流水查询 ----------
def get_usage(self, request_id: str) -> Optional[Dict[str, Any]]:
with self._lock, self._connect() as conn:
@@ -140,6 +179,39 @@ class BillingMixin:
(request_id,)).fetchone()
return dict(row) if row else None
def usage_stats(self) -> Dict[str, Any]:
"""全局流水聚合(§5.2 看板口径,2026-09 优化:SQL 下推替代全量拉取)。
返回 {requests, gateway_cached, in_hit_tok, in_miss_tok, revenue_milli,
cost_milli, by_bucket};by_bucket 键为 bucket 名。
"""
with self._lock, self._connect() as conn:
row = conn.execute(
"SELECT COUNT(*) AS requests,"
" COALESCE(SUM(gateway_cached), 0) AS gateway_cached,"
" COALESCE(SUM(in_hit_tok), 0) AS in_hit_tok,"
" COALESCE(SUM(in_miss_tok), 0) AS in_miss_tok,"
" COALESCE(SUM(charged_milli), 0) AS revenue_milli,"
" COALESCE(SUM(upstream_cost_milli), 0) AS cost_milli"
" FROM usage_ledger").fetchone()
buckets = conn.execute(
"SELECT bucket, COUNT(*) AS requests,"
" COALESCE(SUM(charged_milli), 0) AS revenue_milli,"
" COALESCE(SUM(upstream_cost_milli), 0) AS cost_milli"
" FROM usage_ledger GROUP BY bucket").fetchall()
return {
"requests": row["requests"],
"gateway_cached": row["gateway_cached"],
"in_hit_tok": row["in_hit_tok"],
"in_miss_tok": row["in_miss_tok"],
"revenue_milli": row["revenue_milli"],
"cost_milli": row["cost_milli"],
"by_bucket": {r["bucket"]: {"requests": r["requests"],
"revenue_milli": r["revenue_milli"],
"cost_milli": r["cost_milli"]}
for r in buckets},
}
def list_usage(self, student_id: Optional[int] = None,
limit: int = 50, offset: int = 0) -> List[Dict[str, Any]]:
"""流水分页(可按学生过滤,经其名下 key);按学生过滤走联表常量语句。"""
+9
View File
@@ -52,11 +52,15 @@ class ProxyConfig:
rpm_per_key: int = 10
day_req_cap: int = 200
concurrent_per_key: int = 2
tpm_per_key: int = 60000 # T-X12:每分钟 token 桶(0 = 不限)
max_body_chars: int = 60000
semcache_enabled: bool = True
sim_threshold: float = 0.92
max_entries: int = 300000
promote_frequency: int = 5
# T-X10(采纳 cortiq 路由签名分桶):语义缓存键的路由签名粒度
# none | capabilities(默认:vision/tools 需求不同不互串) | model(按模型隔离)
route_sig_scope: str = "capabilities"
def bucket(self, name: str) -> BucketCfg:
"""取桶配置;未知名回落 default(D-P2 缺省桶)。"""
@@ -156,9 +160,14 @@ def build_proxy_config(settings_dict: Dict[str, Any]) -> ProxyConfig:
rpm_per_key=int(limits.get("rpm_per_key", 10) or 10),
day_req_cap=int(limits.get("day_req_cap", 200) or 200),
concurrent_per_key=int(limits.get("concurrent_per_key", 2) or 2),
tpm_per_key=int(limits.get("tpm_per_key", 60000) or 0),
max_body_chars=int(limits.get("max_body_chars", 60000) or 60000),
semcache_enabled=bool(sem.get("enabled", True)),
sim_threshold=float(sem.get("sim_threshold", 0.92) or 0.92),
max_entries=int(sem.get("max_entries", 300000) or 300000),
promote_frequency=int(sem.get("promote_frequency", 5) or 5),
route_sig_scope=(str(sem.get("route_sig_scope"))
if str(sem.get("route_sig_scope")) in
("none", "capabilities", "model")
else "capabilities"),
)
+25 -9
View File
@@ -8,8 +8,8 @@
工程纪律:
- D-P1 毫元整数,本模块不做任何浮点运算(元换算只发生在入口参数转换)。
- D-P10:同步实现 + 全局锁每操作连接(ReviewQueue 模式),异步调用方经
asyncio.to_thread 包装;WAL 模式(§3)。
- D-P10:同步实现 + 全局锁串行访问(ReviewQueue 模式;2026-09 起复用单条
持久连接,语义不变),异步调用方经 asyncio.to_thread 包装;WAL 模式(§3)。
- 全部数据库访问使用占位符参数化语句,语句为常量,零拼接。
"""
from __future__ import annotations
@@ -74,6 +74,8 @@ class Ledger(BillingMixin):
self.db_path = Path(db_path)
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self._lock = threading.Lock()
self._conn: Optional[sqlite3.Connection] = None
self._reset_date: Optional[str] = None # 日重置短路标记(2026-09 优化)
self._init_db()
# ---------- 初始化 ----------
@@ -83,9 +85,17 @@ class Ledger(BillingMixin):
return cls(db_path)
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path, check_same_thread=False)
conn.row_factory = sqlite3.Row
return conn
"""共享持久连接(2026-09 优化)。
所有操作本就串行于 self._lock 之下,复用单连接语义与逐操作新建等价,
省去每请求 3-5 次 sqlite3.connect(Windows 上开销显著);
`with conn` 只做提交/回滚不关闭,连接持续可用。
"""
if self._conn is None:
conn = sqlite3.connect(self.db_path, check_same_thread=False)
conn.row_factory = sqlite3.Row
self._conn = conn
return self._conn
def _init_db(self) -> None:
with self._lock, self._connect() as conn:
@@ -168,12 +178,18 @@ class Ledger(BillingMixin):
# ---------- 日限额 ----------
def check_and_count(self, key_id: int, student_id: int, now: float) -> bool:
"""日请求限额双检:跨日重置(注入日期)+ 原子计数;超限返回 False(429)。"""
"""日请求限额双检:跨日重置(注入日期)+ 原子计数;超限返回 False(429)。
2026-09 优化:重置 UPDATE 每实例每日只执行一次(原每请求全表扫
students 并抢写锁);单进程部署(D-P9)下语义等价。
"""
today = _today(now)
with self._lock, self._connect() as conn:
conn.execute(
"UPDATE students SET spent_today_milli = 0, spent_date = ?"
" WHERE spent_date IS NOT ?", (today, today))
if self._reset_date != today:
conn.execute(
"UPDATE students SET spent_today_milli = 0, spent_date = ?"
" WHERE spent_date IS NOT ?", (today, today))
self._reset_date = today
row = conn.execute(
"SELECT req_today, req_date, day_cap_req FROM proxy_keys"
" WHERE id = ?", (key_id,)).fetchone()
+7 -1
View File
@@ -83,13 +83,19 @@ def is_cacheable(body: dict) -> bool:
return len(non_system) <= 1
def canonical_hash(bucket: str, doc_version: int, body: dict) -> str:
def canonical_hash(bucket: str, doc_version: int, body: dict,
route_sig: str = "") -> str:
"""规则 3:缓存键 = bucket + '|' + doc_version + '|' + sha256(norm)。
route_sig 非空时插入为第三段(T-X10,采纳 cortiq 路由签名分桶:
不同路由意图——vision/tools 需求或模型——不互串答案);为空时保持
旧三段格式(既有调用与旧缓存条目兼容,旧条目随 TTL 自然淘汰)。
桶模板/资料前缀不参与哈希(由 bucket+doc_version 表达,资料更新 = 版本+1)。
"""
norm = normalize_messages(body)
digest = hashlib.sha256(norm.encode("utf-8")).hexdigest()
if route_sig:
return f"{bucket}|{doc_version}|{route_sig}|{digest}"
return f"{bucket}|{doc_version}|{digest}"
+582 -71
View File
@@ -13,7 +13,8 @@ import asyncio
import json
import time
import uuid
from typing import Any, Dict, Optional
from collections import deque
from typing import Any, Deque, Dict, List, Optional
from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse, StreamingResponse
@@ -30,7 +31,8 @@ from gateway.proxy.errors import (
UpstreamError,
)
from gateway.proxy.pricing import compute
from gateway.proxy.upstream import UpstreamAborted, filter_usage_chunk, stream as upstream_stream
from gateway.proxy.upstream import (UpstreamAborted, failover_stats as upstream_failover_stats,
filter_usage_chunk, stream as upstream_stream)
def install_error_handlers(app) -> None:
@@ -45,8 +47,15 @@ def install_error_handlers(app) -> None:
headers={"WWW-Authenticate": "Bearer"} if exc.status_code == 401 else None)
def build_proxy_router(cfg: ProxyConfig, pool) -> APIRouter:
"""组装代理面路由(唯一组装点)。"""
def build_proxy_router(cfg: ProxyConfig, pool, settings_provider=None,
review_getter=None) -> APIRouter:
"""组装代理面路由(唯一组装点)。
settings_provider:返回设置对象的回调(api.py 注入 settings_store)。
未注入时语义分析器分流不启用(None 安全,既有测试不受影响)。
review_getter:返回人工检验队列的回调(api.py 注入 get_review);
未注入时 T1 审计抽样跳过(不报错)。
"""
router = APIRouter(prefix="/proxy")
ledger = __import__("gateway.proxy.ledger", fromlist=["Ledger"]).Ledger.init_db(cfg.db_path)
# ---------------- 学生面 ----------------
@@ -78,42 +87,64 @@ def build_proxy_router(cfg: ProxyConfig, pool) -> APIRouter:
_limits(), time.time())
if not _limits().acquire_slot(ctx["key_id"]):
raise QuotaError("并发请求已达该 key 上限")
# T-X12(采纳 enterprise-ai-gateway 双桶限流):TPM 桶——RPM 之外加
# token 维度(预扣估算 = 字符/3,与计费同口径;cfg.tpm_per_key=0 不限)
tpm_cap = int(getattr(cfg, "tpm_per_key", 0) or 0)
if tpm_cap > 0:
est_tokens = max(1, len(raw_body) // 3)
if not _limits().allow_tokens(ctx["key_id"], est_tokens, tpm_cap):
_limits().release_slot(ctx["key_id"])
raise QuotaError("该 key 的每分钟 token 额度(TPM)已用尽")
# 语义分析器 live 分流(D-G7:mode=live 才启用;任何异常不影响代理可用性)
# 语义分析器 live 分流(D-G7:mode=live 才启用;任何异常不影响代理可用性)。
# T-X4 修复:settings_provider 由 api.py 注入(原实现引用未导入的
# settings_store,NameError 被 except 吞掉,分流从未实际生效)。
# 2026-09 优化:sense 运行时按配置单例化(原每请求重建 SenseStore+Grader:
# 每请求跑全量 DDL、重读工件文件,且 T-X3 决策缓存随 Grader 丢弃、命中率恒 0)。
tier_used = None
request_id = "px" + uuid.uuid4().hex[:12]
if settings_provider is not None:
try:
from gateway.sense.config import build_sense_config
scfg = build_sense_config(settings_provider().to_dict())
if scfg.enabled and scfg.mode == "live":
_sstore, grader = _get_sense_runtime(scfg)
qtext = "\n".join(str(m.get("content") or "")
for m in (body.get("messages") or []))
d = await grader.decide(qtext or str(body.get("model") or ""),
"proxy")
tier_used = d.tier
# 档位 -> 池条目(§2 消费方表):t1/t2/t3 映射模型替换请求模型
hint = scfg.tier_pool_hint("proxy", tier_used)
e = _entry_by_tier_hint(pool, hint)
if e is not None:
body = {**body, "model": e["model"]}
except Exception:
pass # D-G4:分级故障不影响代理可用性
try:
from gateway.sense.config import build_sense_config
from gateway.sense.grader import Grader
from gateway.sense.observer import get_observer
from gateway.sense.store import SenseStore
scfg = build_sense_config(settings_store().to_dict())
if scfg.enabled and scfg.mode == "live":
sstore = SenseStore.init_db(scfg.db_path)
grader = Grader(scfg, sstore, get_observer(sstore))
qtext = "\n".join(str(m.get("content") or "")
for m in (body.get("messages") or []))
d = await grader.decide(qtext or str(body.get("model") or ""),
"proxy")
tier_used = d.tier
# 档位 -> 池条目(§2 消费方表):t1/t2/t3 映射模型替换请求模型
hint = scfg.tier_pool_hint("proxy", tier_used)
e = _entry_by_tier_hint(pool, hint)
if e is not None:
body = {**body, "model": e["model"]}
except Exception:
pass # D-G4:分级故障不影响代理可用性
try:
resp = await _run_chat(body, dict(request.headers), ctx, cfg, ledger, pool)
resp = await _run_chat(body, dict(request.headers), ctx, cfg, ledger, pool,
request_id=request_id)
if tier_used:
resp.headers["x-campus-tier"] = tier_used
# T1 审计抽样(§9.4):按 review.sample_rate 入队人工核
# T-X11(采纳 cortiq shadow 旁路评审):live T1 本地应答按采样率
# 异步旁路云端评审,喂晋升表(零客户端延迟,任何异常不影响响应)
if tier_used == "T1":
_maybe_shadow_review(body, resp, pool, settings_provider, scfg)
# T1 审计抽样(§9.4):按 review.sample_rate 入队人工核;
# promo 标签供人工 verdict 回填晋升表(T-X5)。
# 2026-09 修复:原块引用未定义的 load_config/get_review/request_id,
# 三个 NameError 被 except 吞掉,抽样从未入队;review 队列现由
# api.py 经 review_getter 注入(避免 proxy -> api 反向依赖)。
try:
import random as _random
from router_system.config import load_config
from gateway.sense.promotion import promotion_label
rate = float(load_config().get("review", {}).get("sample_rate", 0.1))
if _random.random() < rate:
get_review().enqueue(
if review_getter is not None and _random.random() < rate:
review_getter().enqueue(
request_id + "-sense", str(body.get("model") or "proxy"),
"(sense T1 审计抽样)", tags=["sense_t1"],
"(sense T1 审计抽样)",
tags=["sense_t1", "promo:" + promotion_label("proxy", "")],
reason="sense_audit")
except Exception:
pass
@@ -123,7 +154,7 @@ def build_proxy_router(cfg: ProxyConfig, pool) -> APIRouter:
def _entry_by_tier_hint(pool, hint: str) -> Optional[Dict[str, Any]]:
"""按档位名选池条目(tier_hint 优先,缺省按 tier 字段映射)。"""
entries = pool.list().get("entries", [])
entries = _pool_entries(pool)
for e in entries:
if not e.get("enabled") or e.get("backend") == "mock":
continue
@@ -178,33 +209,60 @@ def build_proxy_router(cfg: ProxyConfig, pool) -> APIRouter:
@router.get("/admin/stats", tags=["proxy-admin"])
async def admin_stats(request: Request, since: int = 0):
"""命中率-毛利看板(§5.2 口径):h_g/h_p/revenue/cost/margin/by_bucket。"""
"""命中率-毛利看板(§5.2 口径):h_g/h_p/revenue/cost/margin/by_bucket。
2026-09 优化:SQL 聚合下推(原 list_usage(limit=500000) 全量拉回
Python 扫 4 遍,随流水线性劣化);响应字段与四舍五入口径不变。
"""
_guard_admin(request)
from gateway.proxy.ledgerutil import _today
import time as _t
now = _t.time()
today = _today(now)
# list_usage 返回全列(含 stats 所需字段),复用既有参数化查询
rows = ledger.list_usage(limit=500000)
n = len(rows)
cached = sum(1 for r in rows if r["gateway_cached"])
s = ledger.usage_stats()
n = s["requests"]
cached = s["gateway_cached"]
h_g = cached / n if n else 0.0
in_hit = sum(r["in_hit_tok"] for r in rows)
in_miss = sum(r["in_miss_tok"] for r in rows)
in_hit = s["in_hit_tok"]
in_miss = s["in_miss_tok"]
h_p = in_hit / (in_hit + in_miss) if (in_hit + in_miss) else 0.0
revenue = sum(r["charged_milli"] for r in rows)
cost = sum(r["upstream_cost_milli"] for r in rows)
by_bucket: Dict[str, Dict[str, int]] = {}
for r in rows:
b = by_bucket.setdefault(r["bucket"], {"requests": 0, "revenue_milli": 0,
"cost_milli": 0})
b["requests"] += 1
b["revenue_milli"] += r["charged_milli"]
b["cost_milli"] += r["upstream_cost_milli"]
# T-X12:进程内语义缓存计数透出(entries/hits/misses/hit_rate)
sem_stats = None
if cfg.semcache_enabled:
try:
sem_stats = _get_semcache(cfg, ledger).stats()
except Exception: # noqa: BLE001
sem_stats = None
return {"requests": n, "h_g": round(h_g, 4), "h_p": round(h_p, 4),
"revenue_milli": revenue, "cost_milli": cost,
"margin_milli": revenue - cost, "by_bucket": by_bucket,
"today": today}
"revenue_milli": s["revenue_milli"], "cost_milli": s["cost_milli"],
"margin_milli": s["revenue_milli"] - s["cost_milli"],
"by_bucket": s["by_bucket"],
"semcache": sem_stats,
"route_scored": list(_route_score_events)[-10:][::-1],
"today": today,
"upstream_failover": upstream_failover_stats()}
@router.get("/admin/sense-decisions", tags=["proxy-admin"])
async def admin_sense_decisions(request: Request, limit: int = 50):
"""路由决策留痕查询(T-X4):reasons/candidate_scores/rejected 三元结构。
管理面鉴权同 /admin/stats;sense 未启用时返回空集(不报错,便于前端
统一渲染)。
"""
_guard_admin(request)
try:
from gateway.sense.config import build_sense_config
from gateway.sense.store import SenseStore
if settings_provider is None:
return {"enabled": False, "decisions": []}
scfg = build_sense_config(settings_provider().to_dict())
if not scfg.enabled:
return {"enabled": False, "decisions": []}
sstore = SenseStore.init_db(scfg.db_path)
return {"enabled": True,
"decisions": sstore.list_decisions(limit=limit)}
except Exception:
return {"enabled": False, "decisions": []}
@router.get("/admin/ledger", tags=["proxy-admin"])
async def admin_ledger(request: Request, student_id: int = 0,
@@ -235,6 +293,15 @@ def _limits():
return _AUTH_SINGLETON_LIMITS
def _pool_entries(pool) -> List[Dict[str, Any]]:
"""池条目快照(转发链内部专用):优先未打码通道 usable_entries,
回退打码 list()(兼容测试注入的简化 pool)。HTTP 响应一律用 list()。"""
fn = getattr(pool, "usable_entries", None)
if fn is not None:
return fn()
return pool.list().get("entries", [])
def _estimate_hold_milli(body: dict, cfg: ProxyConfig, model: str) -> int:
"""预扣估算(宁可高估,D-P11):in 按字符/3,out 按 min(max_tokens,4096)。
@@ -267,6 +334,94 @@ def _resolve_entry(pool, model: str, cfg: ProxyConfig) -> Optional[Dict[str, Any
return None
def _budget_headers(budget_mode: str) -> Dict[str, str]:
"""预算档位响应头(normal 不发,避免噪音)。"""
return {"X-Budget-Mode": budget_mode} if budget_mode and budget_mode != "normal" else {}
def _request_needs(body: dict) -> Dict[str, Any]:
"""从请求体推断能力需求(T-X6):多模态图片 -> vision;带 tools -> tools;
上下文需求 = 提示字符/3 + min(max_tokens, 4096)(与预扣估算同口径)。"""
vision = False
for m in body.get("messages") or []:
content = m.get("content")
if isinstance(content, list):
for part in content:
if isinstance(part, dict) and ("image_url" in part
or part.get("type") == "image_url"):
vision = True
prompt_chars = sum(len(str(m.get("content") or ""))
if not isinstance(m.get("content"), list)
else sum(len(str(p.get("text") or "")) for p in m["content"]
if isinstance(p, dict))
for m in (body.get("messages") or []))
est_out = min(int(body.get("max_tokens") or 1024), 4096)
return {"vision": vision,
"tools": bool(body.get("tools")),
"min_context_tokens": prompt_chars // 3 + est_out}
def _entry_meets(entry: Dict[str, Any], needs: Dict[str, Any]) -> bool:
"""条目能力位是否满足请求需求(capabilities 缺省全兼容)。"""
cap = entry.get("capabilities") or {}
if needs["vision"] and not cap.get("vision", True):
return False
if needs["tools"] and not cap.get("tools", True):
return False
ctx = int(cap.get("context_window") or 0)
if needs["min_context_tokens"] > 0 and 0 < ctx < needs["min_context_tokens"]:
return False
return True
def _fallback_chain(pool, entry: Dict[str, Any], max_total: int = 3) -> List[Dict[str, Any]]:
"""上游有序降级链(T-X2,采纳 cortiq tier 链思路)。
主条目之外,取池内其他启用真实后端条目,按 (档位升序, 单价和升序) 排列
——便宜的先顶上;总链长 <= max_total。仅作首 token 前 failover 候选
(D-P4),不做负载均衡(单写者模型)。
"""
from gateway.model_pool import TIERS
tier_rank = {t: i for i, t in enumerate(TIERS)}
rest = [e for e in _pool_entries(pool)
if (e.get("enabled") and e.get("id") != entry.get("id")
and e.get("backend") not in ("mock",) and e.get("base_url")
and e.get("model") != entry.get("model"))]
rest.sort(key=lambda e: (tier_rank.get(e.get("tier"), 99),
float(e.get("price_in") or 0) + float(e.get("price_out") or 0)))
return rest[:max(0, int(max_total) - 1)]
def _downgrade_entry(pool, entry: Dict[str, Any], mode: str) -> Optional[Dict[str, Any]]:
"""预算降档(T-X1):在池内找恰好低一档/最低档的启用条目。
档位序 TIERS = (local, budget, premium):
- optimize:降一档(premium->budget,budget->local);
- cheap:直落最低档 local;
- 已在 local / 找不到该档启用条目 -> None(保持原条目,档位语义照常上报)。
"""
from gateway.model_pool import TIERS
tier = str(entry.get("tier") or "")
if tier not in TIERS:
return None
rank = TIERS.index(tier)
if mode == "cheap":
target = 0
elif mode == "optimize":
target = rank - 1
else:
return None
if target < 0:
return None
want = TIERS[target]
for e in _pool_entries(pool):
if (e.get("enabled") and e.get("id") != entry.get("id")
and e.get("tier") == want
and e.get("backend") not in ("mock",) and e.get("base_url")):
return e
return None
_semcache_instances: Dict[str, Any] = {}
@@ -289,6 +444,287 @@ def reset_semcache_instances() -> None:
_semcache_instances = {}
_sense_runtimes: Dict[tuple, Any] = {}
def _get_sense_runtime(scfg):
"""sense 运行时按配置单例化(2026-09 优化,激活 T-X3 决策缓存)。
键取 SenseConfig 的完整 repr:任何设置变更都会派生新键并重建运行时,
旧实例留在注册表中随设置维度有界增长(单进程部署,D-P9)。
返回 (sstore, grader):Grader 持有的线性头/阈值/60s 决策缓存跨请求复用,
免去每请求的 SenseStore DDL 与工件文件重读。
"""
key = ("sense", repr(scfg))
rt = _sense_runtimes.get(key)
if rt is None:
from gateway.sense.grader import Grader
from gateway.sense.observer import get_observer
from gateway.sense.promotion import PromotionTable
from gateway.sense.store import SenseStore
sstore = SenseStore.init_db(scfg.db_path)
grader = Grader(scfg, sstore, get_observer(sstore),
promotion=PromotionTable(sstore))
rt = (sstore, grader)
_sense_runtimes[key] = rt
return rt
def reset_sense_runtimes() -> None:
"""测试用:清空 sense 运行时单例。"""
global _sense_runtimes
_sense_runtimes = {}
# ---------------------------------------------------------------------------
# T-X11(采纳 cortiq shadow 旁路评审设计):live T1 本地应答的异步云端评审。
# 按采样率旁路:premium 条目给本地答案打 PASS/FAIL,写入 T-X5 晋升表——
# 打通晋升表的质量信号自动来源。零客户端延迟(asyncio.create_task 旁路)。
# ---------------------------------------------------------------------------
_shadow_review_stats: Dict[str, int] = {"sampled": 0, "pass": 0, "fail": 0,
"error": 0, "skipped_busy": 0}
_shadow_sem: Optional[asyncio.Semaphore] = None
_SHADOW_CONCURRENCY = 2 # 旁路评审并发上限(保护上游)
_SHADOW_MAX_ANSWER_CHARS = 4000 # 送审答案截断
def reset_shadow_review_stats() -> None:
"""测试用:清零旁路评审计数并复位并发闸门。"""
global _shadow_review_stats, _shadow_sem
_shadow_review_stats = {"sampled": 0, "pass": 0, "fail": 0,
"error": 0, "skipped_busy": 0}
_shadow_sem = None
def _shadow_review_settings(settings_dict: dict) -> Dict[str, Any]:
"""解析 sense.shadow_review 设置(默认关闭;fail-safe 兜底值)。"""
raw = (settings_dict.get("sense") or {}).get("shadow_review") or {}
if not isinstance(raw, dict):
raw = {}
try:
rate = max(0.0, min(1.0, float(raw.get("sample_rate", 0.1) or 0.1)))
except (TypeError, ValueError):
rate = 0.1
try:
timeout_s = max(1.0, float(raw.get("timeout_s", 20) or 20))
except (TypeError, ValueError):
timeout_s = 20.0
return {"enabled": bool(raw.get("enabled", False)),
"sample_rate": rate, "timeout_s": timeout_s,
"tier": str(raw.get("tier", "premium") or "premium")}
def _pick_review_entry(pool, tier_hint: str) -> Optional[Dict[str, Any]]:
"""选评审条目:优先指定档位,回退任意启用真实后端条目。"""
entries = [e for e in _pool_entries(pool)
if e.get("enabled") and e.get("backend") not in ("mock",)
and e.get("base_url")]
for e in entries:
if (e.get("tier") or "") == tier_hint:
return e
return entries[0] if entries else None
async def _grade_answer(entry, question: str, answer: str, model: str,
timeout_s: float) -> str:
"""用指定池条目评审本地答案,返回 PASS / FAIL / ERROR(上游故障不抛出)。"""
judge_body = {
"model": model,
"messages": [
{"role": "system", "content": "你是严格的答案质量评审员。只输出 PASS 或 FAIL。"},
{"role": "user",
"content": (f"问题:{question}\n\n候选答案:{answer[:_SHADOW_MAX_ANSWER_CHARS]}\n\n"
"该答案是否正确且切题?只输出一个词:PASS 或 FAIL。")},
],
"max_tokens": 8,
}
parts: List[str] = []
sink: Dict[str, Any] = {}
async def _collect():
async for raw in upstream_stream(judge_body, entry, sink, [entry]):
line = raw.decode("utf-8", errors="replace").strip()
for sub in line.split("\n\n"):
if not sub.startswith("data:"):
continue
payload = sub[5:].strip()
if not payload or payload == "[DONE]":
continue
try:
obj = json.loads(payload)
except json.JSONDecodeError:
continue
delta = (obj.get("choices") or [{}])[0].get("delta") or {}
if delta.get("content"):
parts.append(str(delta["content"]))
try:
await asyncio.wait_for(_collect(), timeout=timeout_s)
except Exception: # noqa: BLE001 超时/上游异常:评审失败不算 FAIL
return "ERROR"
text = "".join(parts).strip().upper()
if "PASS" in text:
return "PASS"
if "FAIL" in text:
return "FAIL"
return "ERROR"
async def _shadow_review_task(question: str, answer: str, pool,
settings_dict: dict, promotion,
tier_used: str = "T1") -> None:
"""旁路评审任务体:调用方 create_task 丢弃即可,任何异常不影响主链路。"""
cfgs = _shadow_review_settings(settings_dict)
entry = _pick_review_entry(pool, cfgs["tier"])
if entry is None:
return
global _shadow_sem
if _shadow_sem is None:
_shadow_sem = asyncio.Semaphore(_SHADOW_CONCURRENCY)
try:
async with _shadow_sem:
verdict = await _grade_answer(entry, question, answer,
str(entry.get("model") or ""),
cfgs["timeout_s"])
except Exception: # noqa: BLE001
verdict = "ERROR"
_shadow_review_stats["sampled"] += 1
if verdict in ("PASS", "FAIL"):
_shadow_review_stats["pass" if verdict == "PASS" else "fail"] += 1
try:
from gateway.sense.promotion import promotion_label
promotion.observe(promotion_label("proxy", ""),
ok=(verdict == "PASS"), tier=tier_used)
except Exception: # noqa: BLE001
pass
else:
_shadow_review_stats["error"] += 1
def _maybe_shadow_review(body: dict, resp, pool, settings_provider, scfg) -> None:
"""主链路挂钩:live T1 且命中采样率时旁路评审(仅非流式响应可取答案文本)。"""
try:
import random as _random
cfgs = _shadow_review_settings(settings_provider().to_dict())
if not cfgs["enabled"] or _random.random() >= cfgs["sample_rate"]:
return
if isinstance(resp, StreamingResponse):
return # 流式响应此刻无完整答案文本,跳过(后续迭代可经 sink 采集)
payload = json.loads(bytes(resp.body))
answer = str((((payload.get("choices") or [{}])[0])
.get("message") or {}).get("content") or "")
if not answer:
return
question = "\n".join(str(m.get("content") or "")
for m in (body.get("messages") or []))
_, grader = _get_sense_runtime(scfg)
promotion = grader._promotion # 同包内复用晋升表(T-X5)
if promotion is None:
return
asyncio.get_running_loop().create_task(
_shadow_review_task(question, answer, pool,
settings_provider().to_dict(), promotion))
except Exception: # noqa: BLE001 旁路失败不影响主响应(D-G4 纪律)
pass
def _route_sig(needs: Dict[str, Any], model: str, cfg: ProxyConfig) -> str:
"""路由签名(T-X10,采纳 cortiq 语义缓存路由签名分桶)。
签名进入缓存键,使不同路由意图的请求不互串答案:
- capabilities(默认):vision/tools 需求不同 -> 不同签名(多模态/工具请求
不再命中纯文本缓存答案——正确性优先);
- model:按模型名隔离(更保守,命中率换绝对隔离);
- none:空签名(旧行为,最高命中率)。
"""
scope = getattr(cfg, "route_sig_scope", "capabilities")
if scope == "model":
return f"m:{model}"
if scope == "capabilities":
return f"v{int(bool(needs['vision']))}t{int(bool(needs['tools']))}"
return ""
# ---------------------------------------------------------------------------
# T-X13(采纳 ai-model-router 五维评分 + 硬过滤拒绝理由设计):能力位硬过滤
# (T-X6,语义不变)之后对存活候选做确定性加权评分——改变的是链内顺序而非
# 过滤语义;被淘汰候选带人话拒绝理由进评分留痕(管理面可观测)。
# ---------------------------------------------------------------------------
_ROUTE_SCORE_WEIGHTS = {
"cost_efficiency": 0.30, "capability": 0.25, "speed": 0.20,
"reliability": 0.15, "quality": 0.10,
}
_route_score_events: Deque[Dict[str, Any]] = deque(maxlen=100)
def reset_route_score_events() -> None:
"""测试用:清空评分留痕。"""
_route_score_events.clear()
def _entry_reject_reason(entry: Dict[str, Any], needs: Dict[str, Any]) -> Optional[str]:
"""能力位不满足时的人话拒绝理由(满足则 None)。"""
cap = entry.get("capabilities") or {}
if needs["vision"] and not cap.get("vision", True):
return "模型不支持视觉输入(vision)"
if needs["tools"] and not cap.get("tools", True):
return "模型不支持工具调用(tools)"
ctx = int(cap.get("context_window") or 0)
if needs["min_context_tokens"] > 0 and 0 < ctx < needs["min_context_tokens"]:
return f"上下文窗口 {ctx} 小于需求 {needs['min_context_tokens']}"
return None
def _score_entry(entry: Dict[str, Any], needs: Dict[str, Any]) -> Dict[str, Any]:
"""单候选五维评分(各维归一 0-1;确定性)。
cost_efficiency = 1/(1+均价*2);speed = 1 - 档位秩比(local 最快);
quality = 档位秩比(premium 质量最高的启发式);capability = 满足需求时
按上下文余量给分、不满足为 0;reliability 预留(单写者模型暂无按模型
失败率统计,恒 1.0,权重占位)。
"""
from gateway.model_pool import TIERS
price = max(0.0, (float(entry.get("price_in") or 0)
+ float(entry.get("price_out") or 0)) / 2.0)
try:
rank = TIERS.index(str(entry.get("tier") or ""))
except ValueError:
rank = max(0, len(TIERS) - 1)
span = max(1, len(TIERS) - 1)
cap = entry.get("capabilities") or {}
ctx = int(cap.get("context_window") or 0)
meets = _entry_meets(entry, needs)
headroom = 1.0 if ctx <= 0 else min(1.0, ctx / max(1, needs["min_context_tokens"]))
dims = {
"cost_efficiency": 1.0 / (1.0 + price * 2.0),
"capability": headroom if meets else 0.0,
"speed": 1.0 - rank / span,
"reliability": 1.0,
"quality": rank / span,
}
total = sum(dims[k] * _ROUTE_SCORE_WEIGHTS[k] for k in dims)
return {"model": str(entry.get("model") or ""), "tier": entry.get("tier"),
"total": round(total, 4),
"dims": {k: round(v, 4) for k, v in dims.items()}}
def _rank_candidates(candidates: List[Dict[str, Any]], needs: Dict[str, Any]
) -> List[Dict[str, Any]]:
"""候选按 total 降序(同分按模型名字典序,确定性)。"""
return sorted((_score_entry(e, needs) for e in candidates),
key=lambda s: (-s["total"], s["model"]))
def _record_route_score_event(request_id: str, ranked: List[Dict[str, Any]],
rejected: List[Dict[str, Any]]) -> None:
"""评分留痕:入选排序 + 被淘汰候选的拒绝理由(管理面可观测)。"""
_route_score_events.append({
"ts": int(time.time()), "request_id": request_id,
"ranked": ranked[:3],
"rejected": rejected,
})
def _cached_charge(body: dict, cfg: ProxyConfig, model: str) -> Dict[str, int]:
"""缓存命中计费:成本 0,按未命中口径对入/出估 token 收售价(§7)。"""
from gateway.proxy.pricing import compute
@@ -299,23 +735,67 @@ def _cached_charge(body: dict, cfg: ProxyConfig, model: str) -> Dict[str, int]:
async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any],
cfg: ProxyConfig, ledger, pool):
cfg: ProxyConfig, ledger, pool,
request_id: Optional[str] = None):
model = str(body.get("model") or "")
entry = _resolve_entry(pool, model, cfg)
if entry is None:
return JSONResponse({"error": {"message": f"模型不可用: {model}",
"type": "invalid_request_error"}},
status_code=400)
request_id = "px" + uuid.uuid4().hex[:12]
if request_id is None:
request_id = "px" + uuid.uuid4().hex[:12]
ts = time.time()
client_wants_usage = bool(body.get("stream_options", {}).get("include_usage")) \
if isinstance(body.get("stream_options"), dict) else False
is_stream = bool(body.get("stream"))
# ---- 能力位硬过滤(T-X6):请求需要 vision/tools/上下文而条目不满足时重定向 ----
needs = _request_needs(body)
if not _entry_meets(entry, needs):
from gateway.model_pool import filter_by_capabilities
all_entries = [e for e in _pool_entries(pool)
if e.get("enabled") and e.get("backend") not in ("mock",)
and e.get("base_url")]
cap_entries = filter_by_capabilities(
list(all_entries),
need_vision=needs["vision"], need_tools=needs["tools"],
min_context_tokens=needs["min_context_tokens"])
if cap_entries:
# T-X13:存活候选五维评分,取最高者为重定向目标(不再取首元素)
ranked = _rank_candidates(cap_entries, needs)
best_model = ranked[0]["model"]
entry = next(e for e in cap_entries
if str(e.get("model") or "") == best_model)
model = str(entry.get("model") or model)
body = {**body, "model": model}
_record_route_score_event(
request_id, ranked,
[{"model": str(e.get("model") or ""),
"reason": _entry_reject_reason(e, needs) or "评分落选"}
for e in all_entries if not _entry_meets(e, needs)])
# ---- 预算四档(T-X1):接近日上限渐进降档;>100% 仍由 try_hold 硬拒 ----
budget_mode = "normal"
try:
est_probe = _estimate_hold_milli(body, cfg, model)
budget_mode = await asyncio.to_thread(
ledger.budget_mode, ctx["student_id"], est_probe, ts)
except Exception:
budget_mode = "normal"
if budget_mode in ("optimize", "cheap"):
down = _downgrade_entry(pool, entry, budget_mode)
if down is not None:
entry = down
model = str(entry.get("model") or model)
body = {**body, "model": model}
# ---- 缓存分支(T-P6,§7 时序):仅缓存准入(stop+单轮)查询 ----
# T-X10:缓存键带路由签名(scope 由 cfg.route_sig_scope 决定),且
# lookup/put 共用同一 norm_hash(原先 put 侧重复计算一次)
cacheable = False
cache = None
resolution = None
norm_hash = ""
try:
if cfg.semcache_enabled:
from gateway.proxy.normalizer import canonical_hash, is_cacheable
@@ -323,11 +803,14 @@ async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any],
bucket_cfg = cfg.bucket(str(headers.get("x-campus-bucket") or "default"))
cacheable = is_cacheable(body)
if cacheable:
norm_hash = canonical_hash(bucket_cfg.name, bucket_cfg.doc_version, body)
sig = _route_sig(needs, model, cfg)
norm_hash = canonical_hash(bucket_cfg.name, bucket_cfg.doc_version,
body, sig)
cache = _get_semcache(cfg, ledger)
norm_text = json.dumps(body.get("messages") or [], ensure_ascii=False,
sort_keys=True)
hit = cache.lookup(norm_hash, norm_text, bucket_cfg.doc_version)
hit = cache.lookup(norm_hash, norm_text, bucket_cfg.doc_version,
route_sig=sig)
if hit is not None:
est = _estimate_hold_milli(body, cfg, model)
if await asyncio.to_thread(
@@ -350,7 +833,8 @@ async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any],
replay(), media_type="text/event-stream",
headers={"Cache-Control": "no-cache",
"X-Cache": "HIT",
"X-Request-Id": request_id})
"X-Request-Id": request_id,
**_budget_headers(budget_mode)})
return JSONResponse({
"id": f"chatcmpl-{request_id}", "object": "chat.completion",
"created": int(time.time()), "model": model,
@@ -360,7 +844,8 @@ async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any],
"finish_reason": "stop"}],
"usage": {"prompt_tokens": 0, "completion_tokens": 0,
"total_tokens": 0},
}, headers={"X-Cache": "HIT", "X-Request-Id": request_id})
}, headers={"X-Cache": "HIT", "X-Request-Id": request_id,
**_budget_headers(budget_mode)})
raise BalanceError("余额或当日额度不足")
except BalanceError:
raise
@@ -375,15 +860,25 @@ async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any],
t0 = time.perf_counter()
sink: Dict[str, Any] = {}
chain = _fallback_chain(pool, entry)
# T-X13:降级链按五维评分降序重排(稳定:同分保持原相对序;评分异常回退原链)
try:
totals = {s["model"]: s["total"] for s in _rank_candidates(chain, needs)}
chain = sorted(chain, key=lambda e: (-totals.get(str(e.get("model") or ""), 0.0),
str(e.get("model") or "")))
except Exception: # noqa: BLE001
pass
try:
if is_stream:
return await _stream_response(body, entry, sink, headers, client_wants_usage,
request_id, ctx, cfg, ledger, model, est, t0,
cache=cache, cacheable=cacheable)
cache=cache, cacheable=cacheable,
budget_mode=budget_mode, chain=chain)
return await _json_response(body, entry, sink, request_id, ctx, cfg,
ledger, model, est, t0,
cache=cache, cacheable=cacheable,
headers=headers)
headers=headers, budget_mode=budget_mode,
chain=chain, norm_hash=norm_hash)
except UpstreamAborted as e:
# 流中失败:按已收 usage 结算(无 usage 按字符估算),不缓存(D-P4)
usage = sink.get("usage") or _estimate_usage_from_sink(sink)
@@ -412,7 +907,8 @@ def _estimate_usage_from_sink(sink: Dict[str, Any]) -> Dict[str, int]:
async def _stream_response(body, entry, sink, headers, client_wants_usage,
request_id, ctx, cfg, ledger, model, est, t0,
cache=None, cacheable=False):
cache=None, cacheable=False, budget_mode: str = "normal",
chain: Optional[List[Dict[str, Any]]] = None):
usage = {"in_miss": 0, "in_hit": 0, "out": 0}
async def gen():
@@ -420,7 +916,8 @@ async def _stream_response(body, entry, sink, headers, client_wants_usage,
chunk_id = f"chatcmpl-{request_id}"
created = int(time.time())
try:
async for raw_bytes in upstream_stream(body, entry, sink, [entry]):
async for raw_bytes in upstream_stream(body, entry, sink,
chain if chain is not None else [entry]):
line = raw_bytes.decode("utf-8").strip()
if not line:
continue
@@ -459,14 +956,18 @@ async def _stream_response(body, entry, sink, headers, client_wants_usage,
return StreamingResponse(gen(), media_type="text/event-stream",
headers={"Cache-Control": "no-cache",
"X-Request-Id": request_id})
"X-Request-Id": request_id,
**_budget_headers(budget_mode)})
async def _json_response(body, entry, sink, request_id, ctx, cfg, ledger,
model, est, t0, cache=None, cacheable=False,
headers=None):
headers=None, budget_mode: str = "normal",
chain: Optional[List[Dict[str, Any]]] = None,
norm_hash: str = ""):
parts = []
async for raw_bytes in upstream_stream(body, entry, sink, [entry]):
async for raw_bytes in upstream_stream(body, entry, sink,
chain if chain is not None else [entry]):
line = raw_bytes.decode("utf-8").strip()
if not line:
continue
@@ -495,19 +996,28 @@ async def _json_response(body, entry, sink, request_id, ctx, cfg, ledger,
out_tok=usage.get("out", 0), upstream_cost_milli=br["upstream_cost_milli"],
ttfb_ms=sink.get("ttfb_ms"), total_ms=total_ms, status="ok")
# 缓存准入(D-P5):stop 且单轮且未命中来的 -> 写缓存
if cache is not None and cacheable and sink.get("finish_reason", "stop") == "stop":
# T-X10:键复用 lookup 时的 norm_hash(含路由签名),不再重复计算
if cache is not None and cacheable and norm_hash \
and sink.get("finish_reason", "stop") == "stop":
try:
from gateway.proxy.normalizer import canonical_hash
from gateway.proxy.semcache import _sig_from_key
bucket_cfg = cfg.bucket(str(headers.get("x-campus-bucket")
or "default")) if headers else cfg.bucket("default")
ckey = canonical_hash(bucket_cfg.name, bucket_cfg.doc_version, body)
norm_text = json.dumps(body.get("messages") or [], ensure_ascii=False,
sort_keys=True)
cache.put(ckey, norm_text, "".join(parts), model,
cache.put(norm_hash, norm_text, "".join(parts), model,
doc_version=bucket_cfg.doc_version,
ttl_hours=bucket_cfg.ttl_hours)
ttl_hours=bucket_cfg.ttl_hours,
route_sig=_sig_from_key(norm_hash))
except Exception:
pass
fb = sink.get("upstream_fallback") or {}
fb_headers: Dict[str, str] = {}
if fb.get("used"):
fb_headers = {"X-Upstream-Fallback": "1",
"X-Upstream-Original": str(fb.get("original") or ""),
"X-Upstream-Used": str(fb.get("used_model") or ""),
"X-Upstream-Reason": str(fb.get("reason") or "")[:200]}
return JSONResponse({
"id": f"chatcmpl-{request_id}",
"object": "chat.completion",
@@ -521,4 +1031,5 @@ async def _json_response(body, entry, sink, request_id, ctx, cfg, ledger,
"completion_tokens": usage.get("out", 0),
"total_tokens": usage.get("in_miss", 0) + usage.get("in_hit", 0)
+ usage.get("out", 0)},
}, headers={"X-Request-Id": request_id})
}, headers={"X-Request-Id": request_id,
**_budget_headers(budget_mode), **fb_headers})
+44 -13
View File
@@ -62,10 +62,11 @@ def weighted_jaccard(ga: set, gb: set) -> float:
class CacheEntry:
__slots__ = ("answer", "model", "q_norm", "g", "w", "created_ts", "ttl_ts",
"doc_version", "hits")
"doc_version", "hits", "sig")
def __init__(self, answer: str, model: str, q_norm: str,
created_ts: float, ttl_ts: float, doc_version: int):
created_ts: float, ttl_ts: float, doc_version: int,
sig: str = ""):
self.answer = answer
self.model = model
self.q_norm = q_norm
@@ -75,6 +76,17 @@ class CacheEntry:
self.ttl_ts = ttl_ts
self.doc_version = doc_version
self.hits = 0
self.sig = sig # 路由签名(T-X10:L2 扫描按签名分区)
def _sig_from_key(cache_key: str) -> str:
"""从缓存键解析路由签名(T-X10)。
带签名键 = bucket|doc_version|sig|sha256(4 段);旧三段键无签名返回 ""。
签名段约束不含 '|'(capabilities/model 两种 scope 均满足),故解析无歧义。
"""
parts = cache_key.split("|")
return parts[2] if len(parts) >= 4 else ""
class SemanticCache:
@@ -92,6 +104,7 @@ class SemanticCache:
self._clock = now # 假时钟注入(None = time.time)
self.hits_exact = 0
self.hits_semantic = 0
self.misses = 0
self._rebuild()
# ---------- 时钟 ----------
@@ -106,7 +119,8 @@ class SemanticCache:
return
for r in rows:
entry = CacheEntry(r["answer"], r["model"], r["q_norm"],
r["created_ts"], r["ttl_ts"], r["doc_version"])
r["created_ts"], r["ttl_ts"], r["doc_version"],
sig=_sig_from_key(r["cache_key"]))
entry.hits = r["hits"]
self._index(r["cache_key"], entry, promote=False)
@@ -131,12 +145,18 @@ class SemanticCache:
# ---------- 查询 ----------
def lookup(self, cache_key: str, norm_text: str,
doc_version: int = 1) -> Optional[Dict[str, Any]]:
"""L1 精确 -> L2 语义(§7 签名)。返回 {answer, model, level} 或 None。"""
doc_version: int = 1,
route_sig: Optional[str] = None) -> Optional[Dict[str, Any]]:
"""L1 精确 -> L2 语义(§7 签名)。返回 {answer, model, level} 或 None。
route_sig 非 None 时按签名分区(T-X10):L1 命中要求条目签名一致,
L2 扫描跳过签名不一致的候选(不同路由意图不互串答案);
None = 不校验(旧调用/旧测试完全兼容)。
"""
now = self._now()
# L1
entry = self._l1.get(cache_key)
if entry is not None:
if entry is not None and (route_sig is None or entry.sig == route_sig):
if entry.ttl_ts < now:
self._invalidate_key(cache_key)
else:
@@ -147,6 +167,7 @@ class SemanticCache:
# L2
g = grams(norm_text)
if len(g) < 3:
self.misses += 1
return None
w_q = _weight(g)
candidates: Dict[str, int] = {}
@@ -161,6 +182,8 @@ class SemanticCache:
cand = self._l1.get(key)
if cand is None or cand.ttl_ts < now or cand.doc_version != doc_version:
continue
if route_sig is not None and cand.sig != route_sig:
continue # 签名分区:跨路由意图不参与计分
# 规模上界预筛:w_inter <= lo 且 w_union >= hi,故 score <= lo/hi;
# 严格小于阈值者不可能命中,跳过(不构建相交集合)。
# 注意用严格不等式:lo/hi == 阈值的边界候选仍会进入精确计分,
@@ -174,6 +197,7 @@ class SemanticCache:
best_score = score
best_key = key
if best_key is None or best_score < self.sim_threshold:
self.misses += 1
return None
cand = self._l1[best_key]
cand.hits += 1
@@ -181,9 +205,12 @@ class SemanticCache:
promoted = False
if cand.hits >= self.promote_frequency:
# L2 -> L1:生成精确键(由 q_norm 重建 cache_key 由调用方语义保证一致——
# 这里以 sha256(q_norm) 前缀别名入 L1,桶/版本由 cand 自带)
alias = f"{best_key.split('|')[0]}|{cand.doc_version}|promoted:" \
f"{hashlib.sha256(cand.q_norm.encode()).hexdigest()[:16]}"
# 这里以 sha256(q_norm) 前缀别名入 L1,桶/版本/签名由 cand 自带)
h = hashlib.sha256(cand.q_norm.encode()).hexdigest()[:16]
if cand.sig:
alias = f"{best_key.split('|')[0]}|{cand.doc_version}|{cand.sig}|promoted:{h}"
else:
alias = f"{best_key.split('|')[0]}|{cand.doc_version}|promoted:{h}"
self._index(alias, cand, promote=True)
try:
self.store.promote_semcache(best_key, alias, cand.hits)
@@ -205,11 +232,12 @@ class SemanticCache:
# ---------- 写入 ----------
def put(self, cache_key: str, q_norm: str, answer: str, model: str,
doc_version: int = 1, ttl_hours: int = 72) -> None:
"""写 L1 + 倒排 + sqlite 持久化(§7 签名)。"""
doc_version: int = 1, ttl_hours: int = 72,
route_sig: str = "") -> None:
"""写 L1 + 倒排 + sqlite 持久化(§7 签名)。route_sig 随条目留存供 L2 分区。"""
now = self._now()
entry = CacheEntry(answer, model, q_norm, now, now + ttl_hours * 3600,
doc_version)
doc_version, sig=route_sig)
self._index(cache_key, entry)
try:
self.store.put_semcache(cache_key, "default", q_norm, answer, model,
@@ -219,8 +247,11 @@ class SemanticCache:
pass # 持久化失败不影响内存缓存(可重建)
def stats(self) -> Dict[str, Any]:
total = self.hits_exact + self.hits_semantic + self.misses
return {"entries": len(self._l1), "hits_exact": self.hits_exact,
"hits_semantic": self.hits_semantic}
"hits_semantic": self.hits_semantic, "misses": self.misses,
"hit_rate": round((self.hits_exact + self.hits_semantic) / total, 4)
if total else 0.0}
class SingleFlight:
+47 -2
View File
@@ -11,7 +11,9 @@
from __future__ import annotations
import json
from typing import Any, AsyncIterator, Dict, List, Optional
import time
from collections import deque
from typing import Any, AsyncIterator, Deque, Dict, List, Optional
import httpx
@@ -42,6 +44,36 @@ class UpstreamAborted(Exception):
"""首 token 已下发后上游失败(D-P4:不可 failover,按已收 usage 计费)。"""
# ---------------- failover 统计(T-X2:如实上报,进程内,D-P9 单进程) ----------------
_FAILOVER_EVENTS_MAX = 100
_failover_events: Deque[Dict[str, Any]] = deque(maxlen=_FAILOVER_EVENTS_MAX)
_failover_total = 0
def _record_failover(original: str, failed_model: str, reason: str) -> None:
"""记录一次「首 token 前失败 -> 切换候选」事件。"""
global _failover_total
_failover_total += 1
_failover_events.append({
"ts": int(time.time()), "original": original,
"failed_model": failed_model, "reason": reason,
})
def failover_stats(recent: int = 20) -> Dict[str, Any]:
"""failover 统计(/proxy/admin/stats 聚合暴露)。"""
events = list(_failover_events)
return {"total": _failover_total,
"recent": events[-max(0, int(recent)):][::-1]}
def reset_failover_stats() -> None:
"""测试用:清空 failover 统计。"""
global _failover_total
_failover_events.clear()
_failover_total = 0
def normalize_usage(provider: str, usage_dict: Dict[str, Any]) -> Dict[str, int]:
"""三家 usage 字段 -> 统一 {in_miss, in_hit, out}。
@@ -89,17 +121,22 @@ def filter_usage_chunk(raw_line: str, client_wants_usage: bool) -> Optional[str]
async def stream(body: dict, entry: Dict[str, Any], usage_sink: Dict[str, Any],
fallback_entries: Optional[List[Dict[str, Any]]] = None
) -> AsyncIterator[bytes]:
"""流式派发(§6 签名扩展 failover 链)。
"""流式派发(§6 签名扩展 failover 链;T-X2 扩展降级上报)。
- 始终注入 include_usage;usage chunk 交给 usage_sink(透传过滤由调用方用
filter_usage_chunk 决定)。
- D-P4:首 token 前(未 yield 任何字节)失败 -> 切换 fallback_entries;
已 yield 后失败 -> 抛 UpstreamAborted。
- usage_sink["usage"] 收敛为归一化 dict;sink["ttfb_ms"] 记录首字节耗时。
- T-X2 如实上报:failover 发生时记 sink["upstream_fallback"]
{used, original, used_model, reason},并计入模块级 failover 统计
(failover_stats / reset_failover_stats),/proxy/admin/stats 聚合暴露。
"""
import time as _time
candidates: List[Dict[str, Any]] = [entry] + list(fallback_entries or [])
last_err: Optional[Exception] = None
head_model = str(candidates[0].get("model") or "")
first_failure: Optional[Dict[str, Any]] = None
for cand in candidates:
provider = str(cand.get("provider") or "openai")
@@ -130,10 +167,18 @@ async def stream(body: dict, entry: Dict[str, Any], usage_sink: Dict[str, Any],
first = True
usage_sink["ttfb_ms"] = int((_time.perf_counter() - t0) * 1000)
yield (raw + "\n\n").encode("utf-8")
if first_failure is not None:
first_failure["used_model"] = str(cand.get("model") or "")
usage_sink["upstream_fallback"] = dict(first_failure)
return # 正常结束
except Exception as exc: # noqa: BLE001
if first:
raise UpstreamAborted(str(exc)) from exc
last_err = exc
if first_failure is None:
first_failure = {"used": True, "original": head_model,
"used_model": "", "reason": f"{type(exc).__name__}: {exc}"}
_record_failover(head_model, str(cand.get("model") or ""),
f"{type(exc).__name__}: {exc}")
continue # 首 token 前失败 -> failover
raise UpstreamError(f"上游均不可用: {type(last_err).__name__}: {last_err}")
+74
View File
@@ -0,0 +1,74 @@
"""路由决策缓存(T-X3,采纳 cortiq router_client 的决策哈希缓存思路)。
- key = sha256(scope + 0x00 + 归一文本);LRU + TTL(默认 60s / 4096 条,
对齐 auth._AuthCache 的进程内模式,D-P9 单进程前提)。
- 只缓存**纯决策负载**(如 probs/head_version/vec);调用方的观察与审计
落盘职责不因缓存命中而跳过。
- 时钟可注入(now(monotonic)),供测试确定性。
"""
from __future__ import annotations
import hashlib
import threading
import time
from collections import OrderedDict
from typing import Any, Callable, Optional
_DEFAULT_TTL_S = 60.0
_DEFAULT_MAX = 4096
class DecisionCache:
"""进程内决策缓存:LRU + TTL,键为 (scope, 归一文本) 的 sha256。"""
def __init__(self, maxsize: int = _DEFAULT_MAX, ttl_s: float = _DEFAULT_TTL_S,
now: Optional[Callable[[], float]] = None):
self._max = max(1, int(maxsize))
self._ttl = float(ttl_s)
self._now = now or time.monotonic
self._data: "OrderedDict[str, tuple[float, Any]]" = OrderedDict()
self._lock = threading.Lock()
self.hits = 0
self.misses = 0
@staticmethod
def key(scope: str, text: str) -> str:
"""缓存键:scope 隔离消费方/模式,文本归一由调用方完成。"""
return hashlib.sha256(f"{scope}\x00{text}".encode("utf-8")).hexdigest()
def get(self, scope: str, text: str) -> Optional[Any]:
now = self._now()
k = self.key(scope, text)
with self._lock:
item = self._data.get(k)
if item is None:
self.misses += 1
return None
ts, payload = item
if now - ts > self._ttl:
self._data.pop(k, None)
self.misses += 1
return None
self._data.move_to_end(k)
self.hits += 1
return payload
def put(self, scope: str, text: str, payload: Any) -> None:
k = self.key(scope, text)
with self._lock:
self._data[k] = (self._now(), payload)
self._data.move_to_end(k)
while len(self._data) > self._max:
self._data.popitem(last=False)
def clear(self) -> None:
"""清空条目(阈值/工件版本切换时由 Grader.invalidate 调用)。"""
with self._lock:
self._data.clear()
def stats(self) -> dict:
with self._lock:
return {"entries": len(self._data), "hits": self.hits,
"misses": self.misses,
"hit_rate": round(self.hits / (self.hits + self.misses), 4)
if (self.hits + self.misses) else 0.0}
+124 -19
View File
@@ -17,9 +17,12 @@ from typing import Any, Dict, List, Optional
from gateway.sense.calibrate import load_active
from gateway.sense.classifier import LinearHead
from gateway.sense.config import SenseConfig
from gateway.sense.decision_cache import DecisionCache
from gateway.sense.embedder import embed
from gateway.sense.errors import EmbedderDown
from gateway.sense.features import gate
from gateway.sense.promotion import promotion_label
from gateway.sense.store import json_dumps
@dataclass
@@ -43,16 +46,69 @@ def _rule_tier(feats) -> str:
return "T2"
def _decision_reasons(fallback: bool, probs: Dict[str, float],
th: Dict[str, Any], feats) -> List[str]:
"""决策理由(T-X4):按 §8 判定顺序产出,可解释路由的最小完备集。"""
if fallback:
r = ["fallback:rule_gate"]
if feats.repo_signals:
r.append("repo_signals")
if feats.t1_hard_ok:
r.append("t1_hard_ok")
return r
p1 = probs.get("t1", 0.0) >= float(th.get("t1", 0.9))
p3 = probs.get("t3", 0.0) >= float(th.get("t3", 0.9))
t1_ok = p1 and feats.t1_hard_ok
t3_ok = p3 or feats.repo_signals
r: List[str] = []
if t3_ok and not t1_ok:
r.append("t3_ok")
elif t1_ok:
r.append("t1_ok")
else:
r.append("default_t2")
if p1:
r.append("p1>=threshold")
if not feats.t1_hard_ok:
r.append("gate:t1_hard_ok=0")
if p3:
r.append("p3>=threshold")
if feats.repo_signals:
r.append("repo_signals")
return r
def _rejected_tiers(fallback: bool, probs: Dict[str, float],
th: Dict[str, Any], feats) -> List[Dict[str, str]]:
"""落选档位及原因(T2 为缺省档不记录;fallback 时两条均给 fallback 原因)。"""
if fallback:
return [{"tier": "T1", "reason": "fallback"},
{"tier": "T3", "reason": "fallback"}]
p1 = probs.get("t1", 0.0) >= float(th.get("t1", 0.9))
p3 = probs.get("t3", 0.0) >= float(th.get("t3", 0.9))
t1_ok = p1 and feats.t1_hard_ok
t3_ok = p3 or feats.repo_signals
out: List[Dict[str, str]] = []
if not t3_ok:
out.append({"tier": "T3", "reason": "p3<threshold 且无 repo_signals"})
if not t1_ok:
out.append({"tier": "T1",
"reason": "p1<threshold" if not p1 else "t1_hard_ok=0"})
return out
class Grader:
"""决策器(持有 active 工件缓存;工件/阈值切换后调 invalidate)。"""
def __init__(self, cfg: SenseConfig, store, observer=None):
def __init__(self, cfg: SenseConfig, store, observer=None, promotion=None):
self.cfg = cfg
self.store = store
self.observer = observer
self._promotion = promotion # T-X5:晋升表(可 None)
self._head: Optional[LinearHead] = None
self._head_loaded = False
self._thresholds: Optional[Dict[str, Any]] = None
self._dcache = DecisionCache() # T-X3:live 模式决策缓存
def _load_head(self):
if not self._head_loaded:
@@ -66,10 +122,11 @@ class Grader:
return self._head
def invalidate(self) -> None:
"""工件 promote 后调用(重载 active 工件与阈值)。"""
"""工件 promote 后调用(重载 active 工件与阈值;同步清空决策缓存)。"""
self._head = None
self._head_loaded = False
self._thresholds = None
self._dcache.clear()
def _thresholds_cached(self) -> Dict[str, Any]:
if self._thresholds is None:
@@ -79,32 +136,47 @@ class Grader:
async def decide(self, query_or_messages, consumer: str,
domain: str = "", request_id: str = "",
executed_tier: str = "") -> TierDecision:
"""分级决策(§8 时序;全模式写观察)。"""
"""分级决策(§8 时序;全模式写观察)。
T-X3:live 模式下对 (consumer, 文本) 的纯决策负载(probs/head_version/vec)
做 60s LRU 缓存,命中时跳过 embed + 线性头;观察/审计照常落盘。
collect/shadow 模式不走缓存(校准数据必须全量产出)。
"""
ts = time.time()
rid = request_id or ("rt" + uuid.uuid4().hex[:10])
text = (query_or_messages if isinstance(query_or_messages, str)
else "\n".join(str(m.get("content") or "")
for m in query_or_messages))
feats = gate(query_or_messages, consumer, self.cfg)
probs: Dict[str, float] = {}
head_version = ""
fallback = False
vec: Optional[List[int]] = None
try:
vec = await embed(feats and (query_or_messages
if isinstance(query_or_messages, str)
else "\n".join(
str(m.get("content") or "")
for m in query_or_messages)),
self.cfg.embedder)
except EmbedderDown:
fallback = True
cached = self._dcache.get(consumer, text) \
if self.cfg.mode == "live" else None
if cached is not None:
probs = dict(cached["probs"])
head_version = str(cached["head_version"])
vec = cached["vec"]
else:
try:
vec = await embed(text, self.cfg.embedder)
except EmbedderDown:
fallback = True
head = None if fallback else self._load_head()
if head is None:
fallback = True
head = None if fallback else self._load_head()
if head is None:
fallback = True
if not fallback and vec is not None:
probs = head.predict([float(v) for v in vec])
head_version = head.version
if not fallback and vec is not None:
probs = head.predict([float(v) for v in vec])
head_version = head.version
if self.cfg.mode == "live":
self._dcache.put(consumer, text,
{"probs": dict(probs),
"head_version": head_version,
"vec": vec})
th = self._thresholds_cached()
th_version = str(th.get("version") or "")
@@ -124,6 +196,18 @@ class Grader:
else:
tier = "T2"
# ---- 晋升表应用(T-X5:conformal 之外的统计闸门;仅 live)----
# 仅对 T2 决策且无任何 T3 信号时,允许已晋升标签升级 T1;
# T1/T3 决策与 fallback 路径不受晋升表影响。
promotion_applied = False
if (self._promotion is not None and self.cfg.mode == "live"
and not fallback and tier == "T2"
and not feats.repo_signals
and probs.get("t3", 0.0) < float(th.get("t3", 0.9))):
if self._promotion.is_promoted(promotion_label(consumer, domain)):
tier = "T1"
promotion_applied = True
# ---- mode 裁剪(D-G7)----
if self.cfg.mode == "live":
executed = tier # live:决策即执行
@@ -138,11 +222,32 @@ class Grader:
"intent_blocked": feats.intent_blocked,
"repo_signals": feats.repo_signals,
"over_length": feats.over_length,
"t1_hard_ok": feats.t1_hard_ok},
"t1_hard_ok": feats.t1_hard_ok,
"promotion_applied": promotion_applied},
thresholds_version=th_version, head_version=head_version,
mode=self.cfg.mode, fallback=fallback,
executed_tier=executed)
# ---- 决策留痕(T-X4:reasons / candidate_scores / rejected 三元结构)----
# 2026-09 优化:同步 sqlite 写移入线程池(原直接跑在事件循环线程上,
# 高并发下阻塞整个网关;观察写本就有批量队列,决策留痕此处补齐同等保护)
try:
import asyncio
row = {
"ts": ts, "request_id": rid, "consumer": consumer,
"decided_tier": tier, "executed_tier": executed,
"mode": self.cfg.mode, "fallback": fallback,
"policy_version": f"head:{head_version}|th:{th_version}",
"q_hash": __import__("hashlib").sha256(
text.encode("utf-8")).hexdigest()[:32],
"reasons": json_dumps(_decision_reasons(fallback, probs, th, feats)),
"candidate_scores": json_dumps(probs),
"rejected": json_dumps(_rejected_tiers(fallback, probs, th, feats)),
}
await asyncio.to_thread(self.store.insert_decision, row)
except Exception: # noqa: BLE001
pass # 留痕失败不影响决策(观察与代理主链路优先)
# ---- 观察必写(全模式)----
if self.observer is not None:
self.observer.log(__import__("gateway.sense.observer",
+81
View File
@@ -0,0 +1,81 @@
"""云端评判式晋升表(T-X5,采纳 cortiq promotion.rs,与 conformal 双闸门)。
定位:conformal 阈值给出「分对率 >= 1-α」的统计保证(单条查询粒度),
本表在其之上给出「某类任务(label)可由本地档接管」的**节奏自动化**:
- 观察:人工核验 verdict(approve=ok)等质量信号按 label 累计通过率;
- 晋升:n_total >= n_min 且 通过率 >= promote_lb 且 soak 浸泡期满足
-> state=promoted(此后 grader 对该 label 的 T2 决策可升级 T1);
- 退化:promoted 期间通过率 < demote_lb -> demoted(自动回退分级路径);
- 一票否决:tier=T3(高复杂度/仓库级)观察不进通过率统计,且若已晋升
立即降级——与 cortiq "HIGH tier escalation is served by the cloud" 同构。
灰度纪律(沿用 D-G7):collect 数据不足(n_min 未满)不得晋升;
全部状态迁移可由 T-X4 的决策留痕审计。
"""
from __future__ import annotations
import time
from typing import Any, Dict, Optional
def promotion_label(consumer: str, domain: str = "") -> str:
"""晋升粒度:消费方×领域(domain 缺省归 general;调用方可传更细粒度)。"""
return f"{consumer or 'default'}:{domain or 'general'}"
class PromotionTable:
"""标签晋升状态机(candidate / promoted / demoted;持久化于 sense_promotion)。"""
def __init__(self, store, now=None, n_min: int = 20, promote_lb: float = 0.95,
soak_days: float = 3.0, demote_lb: float = 0.85,
high_tier_veto: bool = True):
self.store = store
self._now = now or (lambda: time.time())
self.n_min = max(1, int(n_min))
self.promote_lb = float(promote_lb)
self.demote_lb = float(demote_lb)
self.soak_s = float(soak_days) * 86400.0
self.high_tier_veto = bool(high_tier_veto)
def observe(self, label: str, ok: bool, tier: str = "",
ts: Optional[float] = None) -> Dict[str, Any]:
"""记录一次质量信号并推进状态机;返回迁移后的行。"""
ts = float(ts) if ts is not None else float(self._now())
row = self.store.get_promotion(label) or {
"label": label, "n_total": 0, "n_ok": 0, "state": "candidate",
"first_ts": int(ts), "promoted_ts": 0, "last_ts": 0}
# 一票否决:T3 观察不进通过率统计;已晋升者立即降级
if self.high_tier_veto and str(tier).upper() == "T3":
row["last_ts"] = int(ts)
if row["state"] == "promoted":
row["state"] = "demoted"
row["promoted_ts"] = 0
self.store.upsert_promotion(row)
return row
row["n_total"] = int(row["n_total"]) + 1
row["n_ok"] = int(row["n_ok"]) + (1 if ok else 0)
row["last_ts"] = int(ts)
rate = row["n_ok"] / row["n_total"] if row["n_total"] else 0.0
soaked = (ts - int(row["first_ts"])) >= self.soak_s
if row["state"] == "promoted":
if row["n_total"] >= self.n_min and rate < self.demote_lb:
row["state"] = "demoted" # 退化自动降级
row["promoted_ts"] = 0
else:
# candidate / demoted 共用同一晋升门(降级后可凭数据恢复)
if row["n_total"] >= self.n_min and rate >= self.promote_lb and soaked:
row["state"] = "promoted"
row["promoted_ts"] = int(ts)
self.store.upsert_promotion(row)
return row
def is_promoted(self, label: str) -> bool:
row = self.store.get_promotion(label)
return bool(row) and row["state"] == "promoted"
def snapshot(self, label: str) -> Optional[Dict[str, Any]]:
return self.store.get_promotion(label)
+79
View File
@@ -7,6 +7,7 @@ WAL;全部访问经全局锁 + 每操作新连接(ReviewQueue 模式),
from __future__ import annotations
import sqlite3
import json
import threading
import time
from pathlib import Path
@@ -29,6 +30,25 @@ CREATE TABLE IF NOT EXISTS sense_artifacts(
version TEXT PRIMARY KEY, kind TEXT NOT NULL,
path TEXT NOT NULL, metrics TEXT NOT NULL, created_ts INTEGER NOT NULL,
active INTEGER DEFAULT 0);
CREATE TABLE IF NOT EXISTS sense_decisions(
id INTEGER PRIMARY KEY, ts INTEGER NOT NULL,
request_id TEXT NOT NULL, consumer TEXT NOT NULL,
decided_tier TEXT NOT NULL, executed_tier TEXT NOT NULL DEFAULT '',
mode TEXT NOT NULL DEFAULT 'collect', fallback INTEGER NOT NULL DEFAULT 0,
policy_version TEXT DEFAULT '', q_hash TEXT DEFAULT '',
reasons TEXT NOT NULL DEFAULT '[]',
candidate_scores TEXT NOT NULL DEFAULT '{}',
rejected TEXT NOT NULL DEFAULT '[]');
CREATE INDEX IF NOT EXISTS idx_dec_ts ON sense_decisions(ts);
CREATE TABLE IF NOT EXISTS sense_promotion(
label TEXT PRIMARY KEY, n_total INTEGER NOT NULL DEFAULT 0,
n_ok INTEGER NOT NULL DEFAULT 0,
state TEXT NOT NULL DEFAULT 'candidate',
first_ts INTEGER NOT NULL DEFAULT 0,
promoted_ts INTEGER NOT NULL DEFAULT 0,
last_ts INTEGER NOT NULL DEFAULT 0);
"""
@@ -131,6 +151,65 @@ class SenseStore:
"DELETE FROM tier_observations WHERE ts < ?", (int(ts),))
return cur.rowcount
# ---------- 决策留痕(T-X4:可解释路由三元结构) ----------
def insert_decision(self, rec: Dict[str, Any]) -> None:
"""写入一条路由决策记录(reasons/candidate_scores/rejected 为 JSON 串)。"""
with self._lock, self._connect() as conn:
conn.execute(
"""INSERT INTO sense_decisions
(ts, request_id, consumer, decided_tier, executed_tier, mode,
fallback, policy_version, q_hash, reasons, candidate_scores,
rejected)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",
(int(rec.get("ts", time.time())), rec["request_id"],
rec["consumer"], rec["decided_tier"],
rec.get("executed_tier", ""), rec.get("mode", "collect"),
1 if rec.get("fallback") else 0,
rec.get("policy_version", ""), rec.get("q_hash", ""),
rec.get("reasons", "[]"), rec.get("candidate_scores", "{}"),
rec.get("rejected", "[]")))
def list_decisions(self, limit: int = 50) -> List[Dict[str, Any]]:
"""最近 N 条决策(新在前;JSON 字段反序列化为对象)。"""
with self._lock, self._connect() as conn:
rows = conn.execute(
"SELECT * FROM sense_decisions ORDER BY id DESC LIMIT ?",
(max(1, int(limit)),)).fetchall()
out: List[Dict[str, Any]] = []
for r in rows:
d = dict(r)
for f in ("reasons", "candidate_scores", "rejected"):
try:
d[f] = json.loads(d.get(f) or ("{}" if f == "candidate_scores" else "[]"))
except json.JSONDecodeError:
pass
out.append(d)
return out
# ---------- 晋升表(T-X5:本地档自动接管的双闸门之一) ----------
def upsert_promotion(self, row: Dict[str, Any]) -> None:
with self._lock, self._connect() as conn:
conn.execute(
"""INSERT OR REPLACE INTO sense_promotion
(label, n_total, n_ok, state, first_ts, promoted_ts, last_ts)
VALUES (?,?,?,?,?,?,?)""",
(row["label"], int(row.get("n_total", 0)),
int(row.get("n_ok", 0)), str(row.get("state", "candidate")),
int(row.get("first_ts", 0)), int(row.get("promoted_ts", 0)),
int(row.get("last_ts", 0))))
def get_promotion(self, label: str) -> Optional[Dict[str, Any]]:
with self._lock, self._connect() as conn:
row = conn.execute(
"SELECT * FROM sense_promotion WHERE label = ?", (label,)).fetchone()
return dict(row) if row else None
def list_promotions(self) -> List[Dict[str, Any]]:
with self._lock, self._connect() as conn:
rows = conn.execute(
"SELECT * FROM sense_promotion ORDER BY last_ts DESC").fetchall()
return [dict(r) for r in rows]
# ---------- 工件 ----------
def register_artifact(self, version: str, kind: str, path: str,
metrics: Dict[str, Any], active: bool = False,
+94
View File
@@ -0,0 +1,94 @@
"""模型池能力位过滤测试(T-X6):归一化缺省 / 硬过滤 / 请求需求推断。"""
import pytest
from gateway.model_pool import PoolStore, filter_by_capabilities
from gateway.proxy.routes import _entry_meets, _request_needs
pytest.importorskip("fastapi")
def _make_pool(tmp_path) -> PoolStore:
store = PoolStore(path=tmp_path / "pool.json")
store.upsert({"id": "text-only", "name": "纯文本", "tier": "budget",
"backend": "openai", "base_url": "http://a", "model": "text-m",
"capabilities": {"vision": False, "tools": True,
"context_window": 8192},
"enabled": True})
store.upsert({"id": "vision", "name": "多模态", "tier": "premium",
"backend": "openai", "base_url": "http://b", "model": "vision-m",
"capabilities": {"vision": True, "tools": True,
"context_window": 32768},
"enabled": True})
store.upsert({"id": "legacy", "name": "老条目", "tier": "local",
"backend": "llama_server", "base_url": "http://c",
"model": "legacy-m", "enabled": True}) # 无 capabilities 字段
return store
def test_capabilities_default_full_compat(tmp_path):
"""未声明能力位的老条目:缺省全兼容(vision/tools True、context 0 不限)。"""
pool = _make_pool(tmp_path)
e = pool.find_by_model("legacy-m")
assert e["capabilities"] == {"vision": True, "tools": True, "context_window": 0}
def test_capabilities_invalid_shapes_fall_back(tmp_path):
pool = _make_pool(tmp_path)
pool.upsert({"id": "weird", "name": "怪", "tier": "budget", "backend": "openai",
"base_url": "http://d", "model": "weird-m",
"capabilities": {"vision": "是", "context_window": "abc"},
"enabled": True})
cap = pool.find_by_model("weird-m")["capabilities"]
assert cap["vision"] is True # 非布尔按缺省
assert cap["context_window"] == 0 # 非整数回落不限
def test_filter_by_capabilities_hard_rules():
entries = [{"capabilities": {"vision": False, "tools": True, "context_window": 8192}},
{"capabilities": {"vision": True, "tools": True, "context_window": 32768}},
{"capabilities": {"vision": True, "tools": False, "context_window": 0}}]
out = filter_by_capabilities(entries, need_vision=True)
assert len(out) == 2
out = filter_by_capabilities(entries, need_vision=True, need_tools=True)
assert len(out) == 1
out = filter_by_capabilities(entries, need_vision=True, need_tools=True,
min_context_tokens=16384)
assert len(out) == 1 and out[0]["capabilities"]["context_window"] == 32768
# context_window=0(不限)不受 min_context 约束
out = filter_by_capabilities(entries, min_context_tokens=999999)
assert len(out) == 1 and out[0]["capabilities"]["context_window"] == 0
def test_request_needs_detection():
body = {"messages": [
{"role": "user", "content": "普通文本问题"},
{"role": "user", "content": [{"type": "text", "text": "看这张图"},
{"type": "image_url",
"image_url": {"url": "data:image/png;base64,x"}}]},
], "tools": [{"type": "function", "function": {"name": "f"}}],
"max_tokens": 3000}
needs = _request_needs(body)
assert needs["vision"] is True and needs["tools"] is True
assert needs["min_context_tokens"] > 3000
plain = _request_needs({"messages": [{"role": "user", "content": "hi"}]})
assert plain["vision"] is False and plain["tools"] is False
assert plain["min_context_tokens"] > 0
def test_entry_meets_defaults_and_limits():
legacy = {"capabilities": {"vision": True, "tools": True, "context_window": 0}}
assert _entry_meets(legacy, {"vision": True, "tools": True,
"min_context_tokens": 999999})
limited = {"capabilities": {"vision": True, "tools": True, "context_window": 4096}}
assert _entry_meets(limited, {"vision": False, "tools": False,
"min_context_tokens": 4096})
assert not _entry_meets(limited, {"vision": False, "tools": False,
"min_context_tokens": 4097})
def test_pool_list_returns_capabilities(tmp_path):
"""list() 打码输出保留 capabilities 字段(前端可见)。"""
pool = _make_pool(tmp_path)
e = next(x for x in pool.list()["entries"] if x["id"] == "vision")
assert e["capabilities"]["vision"] is True
+134
View File
@@ -0,0 +1,134 @@
"""预算四档渐进干预测试(T-X1):budget_mode 黄金边界 + 降档选择器。
黄金用例锁死(整数基点 bp,1% = 100bp):
projected < 8000bp -> normal
8000bp <= projected -> optimize
9500bp <= projected -> cheap
projected > 10000bp -> block(try_hold 硬拒兜底)
"""
import pytest
from gateway.model_pool import PoolStore
from gateway.proxy.billing import BUDGET_CHEAP_BP, BUDGET_OPTIMIZE_BP
from gateway.proxy.ledger import Ledger
from gateway.proxy.ledgerutil import _today
from gateway.proxy.routes import _budget_headers, _downgrade_entry
pytest.importorskip("fastapi")
_TS = 1789874000.0 # 固定时间戳(黄金用例确定性)
def _make_ledger(tmp_path, daily_cap_yuan: float = 10.0) -> Ledger:
led = Ledger.init_db(tmp_path / "budget.sqlite3")
led.upsert_student("张三", "软件2201", balance_yuan=100.0,
daily_cap_yuan=daily_cap_yuan)
return led
def _set_spent(led: Ledger, student_id: int, spent_milli: int,
spent_date: str = "") -> None:
"""直接写当日已花(白盒:budget_mode 的输入口径与 try_hold 一致)。"""
with led._lock, led._connect() as conn:
conn.execute(
"UPDATE students SET spent_today_milli = ?, spent_date = ? WHERE id = ?",
(spent_milli, spent_date, student_id))
def test_threshold_constants_golden():
"""阈值常量黄金锁定:80% / 95%。"""
assert BUDGET_OPTIMIZE_BP == 8000
assert BUDGET_CHEAP_BP == 9500
@pytest.mark.parametrize("spent,est,expected", [
(0, 7999, "normal"), # 7999bp
(0, 8000, "optimize"), # 恰好 8000bp(边界含头)
(0, 9499, "optimize"),
(0, 9500, "cheap"), # 恰好 9500bp(边界含头)
(0, 10000, "cheap"), # 恰好达上限 = cheap(不 block)
(0, 10001, "block"), # 严格超出才 block
(5000, 3000, "optimize"), # 8000bp:spent 与 est 合并计算
(5000, 4500, "cheap"), # 9500bp
(5000, 5001, "block"), # 10001bp
])
def test_budget_mode_golden_boundaries(tmp_path, spent, est, expected):
led = _make_ledger(tmp_path, daily_cap_yuan=10.0) # cap = 10000 毫元
_set_spent(led, 1, spent, spent_date=_today(_TS)) # 记账日 = 判定日(当日口径)
assert led.budget_mode(1, est, ts=_TS) == expected
def test_budget_mode_zero_cap_means_unlimited(tmp_path):
led = _make_ledger(tmp_path, daily_cap_yuan=0.0)
sid = 1
_set_spent(led, sid, 999999, spent_date=_today(_TS))
assert led.budget_mode(sid, 10**9, ts=_TS) == "normal"
def test_budget_mode_spent_resets_by_date(tmp_path):
"""spent_date 非今日 -> 当日已花按 0 计(与 try_hold 口径一致)。"""
led = _make_ledger(tmp_path, daily_cap_yuan=10.0)
_set_spent(led, 1, 9999, spent_date="2000-01-01") # 非今日
assert led.budget_mode(1, 1, ts=1789874000.0) == "normal"
def test_budget_mode_missing_student_normal(tmp_path):
led = Ledger.init_db(tmp_path / "b.sqlite3")
assert led.budget_mode(999, 100, ts=1789874000.0) == "normal"
def _make_pool(tmp_path) -> PoolStore:
import gateway.model_pool as mp
store = PoolStore(path=tmp_path / "pool.json") # 显式隔离,防污染真实池文件
mp.reset_pool()
store.upsert({"id": "p1", "name": "旗舰", "tier": "premium", "backend": "openai",
"base_url": "https://api.example.com", "model": "big-x",
"enabled": True})
store.upsert({"id": "b1", "name": "实惠", "tier": "budget", "backend": "openai",
"base_url": "https://api.example.com/v2", "model": "mid-y",
"enabled": True})
store.upsert({"id": "l1", "name": "本地", "tier": "local",
"backend": "llama_server", "base_url": "http://127.0.0.1:8901",
"model": "qwen-local", "enabled": True})
return store
def test_downgrade_optimize_one_tier(tmp_path):
pool = _make_pool(tmp_path)
premium = pool.find_by_model("big-x")
down = _downgrade_entry(pool, premium, "optimize")
assert down is not None and down["tier"] == "budget"
def test_downgrade_cheap_to_lowest(tmp_path):
pool = _make_pool(tmp_path)
premium = pool.find_by_model("big-x")
down = _downgrade_entry(pool, premium, "cheap")
assert down is not None and down["tier"] == "local"
def test_downgrade_stops_at_local(tmp_path):
"""已在最低档:cheap/optimize 均不再降(杜绝反向升档)。"""
pool = _make_pool(tmp_path)
local = pool.find_by_model("qwen-local")
assert _downgrade_entry(pool, local, "cheap") is None
assert _downgrade_entry(pool, local, "optimize") is None
def test_downgrade_skips_disabled_and_mock(tmp_path):
pool = _make_pool(tmp_path)
pool.upsert({"id": "b2", "name": "停用", "tier": "budget", "backend": "openai",
"base_url": "https://api.example.com/v3", "model": "mid-z",
"enabled": False})
pool.upsert({"id": "mk", "name": "假", "tier": "budget", "backend": "mock",
"model": "mock", "enabled": True})
premium = pool.find_by_model("big-x")
down = _downgrade_entry(pool, premium, "optimize")
assert down is not None and down["id"] == "b1" # 跳过停用与 mock
def test_budget_headers_only_when_abnormal():
assert _budget_headers("normal") == {}
assert _budget_headers("") == {}
assert _budget_headers("optimize") == {"X-Budget-Mode": "optimize"}
assert _budget_headers("cheap") == {"X-Budget-Mode": "cheap"}
+130
View File
@@ -0,0 +1,130 @@
"""上游有序降级链测试(T-X2):链构造排序 + 首 token 前 failover 三元组上报。"""
import asyncio
import json
import httpx
import pytest
from gateway.model_pool import PoolStore
from gateway.proxy.routes import _fallback_chain
from gateway.proxy.upstream import UpstreamError, failover_stats, reset_failover_stats
pytest.importorskip("fastapi")
def _make_pool(tmp_path) -> PoolStore:
store = PoolStore(path=tmp_path / "pool.json")
store.upsert({"id": "p-main", "name": "主", "tier": "premium", "backend": "openai",
"base_url": "http://bad", "model": "big-x", "price_in": 1.0,
"price_out": 2.0, "enabled": True})
store.upsert({"id": "p-cheap", "name": "便宜云", "tier": "budget", "backend": "openai",
"base_url": "http://good", "model": "mid-y", "price_in": 0.1,
"price_out": 0.2, "enabled": True})
store.upsert({"id": "p-local", "name": "本地", "tier": "local",
"backend": "llama_server", "base_url": "http://local:8901",
"model": "qwen-local", "price_in": 0.0, "price_out": 0.0,
"enabled": True})
store.upsert({"id": "p-off", "name": "停用", "tier": "local",
"backend": "openai", "base_url": "http://off", "model": "off-z",
"enabled": False})
store.upsert({"id": "p-mock", "name": "假", "tier": "local", "backend": "mock",
"model": "mock", "enabled": True})
return store
def test_fallback_chain_order_and_exclusions(tmp_path):
"""链序:档位升序 -> 单价和升序;排除 mock/停用/同模型;截断到 max_total。"""
pool = _make_pool(tmp_path)
main = pool.find_by_model("big-x")
chain = _fallback_chain(pool, main)
assert [e["model"] for e in chain] == ["qwen-local", "mid-y"] # local(0) < budget(1)
# 预算降档后的条目作主条目:不回排同档更贵者之外的高价条目在前
short = _fallback_chain(pool, main, max_total=2)
assert [e["model"] for e in short] == ["qwen-local"]
def test_fallback_chain_excludes_same_model(tmp_path):
"""同模型不同条目不进链(避免重复打同一上游)。"""
pool = _make_pool(tmp_path)
pool.upsert({"id": "p-dup", "name": "重", "tier": "budget", "backend": "openai",
"base_url": "http://dup", "model": "mid-y", "enabled": True})
mid = pool.find_by_model("mid-y")
assert all(e["model"] != "mid-y" for e in _fallback_chain(pool, mid))
def test_stream_failover_reports_triplet():
"""主上游首 token 前 500 -> 切换候选成功;sink 与统计如实上报三元组。"""
reset_failover_stats()
calls: list = []
def handler(request: httpx.Request) -> httpx.Response:
calls.append(str(request.url))
if request.url.host == "bad":
return httpx.Response(500, text="boom")
body = ("\n\n".join([
'data: {"choices":[{"delta":{"content":"OK"}}]}',
"data: [DONE]",
]) + "\n\n").encode("utf-8")
return httpx.Response(200, content=body)
import gateway.proxy.upstream as up
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
orig = up._client
up._client = client
primary = {"base_url": "http://bad", "model": "big-x", "provider": "openai"}
fallback = {"base_url": "http://good", "model": "mid-y", "provider": "openai"}
async def run():
sink: dict = {}
chunks = []
async for b in up.stream({"model": "big-x", "messages": []}, primary, sink,
[fallback]):
chunks.append(b)
return sink, chunks
try:
sink, chunks = asyncio.run(run())
finally:
up._client = orig
assert b"OK" in b"".join(chunks)
fb = sink["upstream_fallback"]
assert fb["used"] is True
assert fb["original"] == "big-x"
assert fb["used_model"] == "mid-y"
assert "500" in fb["reason"]
stats = failover_stats()
assert stats["total"] == 1
assert stats["recent"][0]["original"] == "big-x"
reset_failover_stats()
assert failover_stats()["total"] == 0
def test_stream_all_fail_raises_and_counts():
"""全部候选失败 -> UpstreamError;每次首 token 前失败均计入统计。"""
reset_failover_stats()
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(503, text="down")
import gateway.proxy.upstream as up
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
orig = up._client
up._client = client
primary = {"base_url": "http://bad", "model": "big-x", "provider": "openai"}
fallback = {"base_url": "http://good", "model": "mid-y", "provider": "openai"}
async def run():
sink: dict = {}
async for _b in up.stream({"model": "big-x", "messages": []}, primary, sink,
[fallback]):
pass
try:
with pytest.raises(UpstreamError):
asyncio.run(run())
finally:
up._client = orig
assert failover_stats()["total"] == 2 # 两个候选各记一次
reset_failover_stats()
+62
View File
@@ -0,0 +1,62 @@
"""五维评分选模与拒绝理由留痕(T-X13,采纳 ai-model-router scoring 设计)。"""
import gateway.proxy.routes as R
from gateway.proxy.routes import (_entry_reject_reason, _rank_candidates,
_record_route_score_event,
reset_route_score_events)
def _entry(model, tier="budget", price_in=1.0, price_out=2.0,
vision=True, tools=True, ctx=0):
return {"id": model, "model": model, "enabled": True,
"backend": "openai", "base_url": "http://127.0.0.1:9/v1",
"tier": tier, "price_in": price_in, "price_out": price_out,
"capabilities": {"vision": vision, "tools": tools,
"context_window": ctx}}
_NEEDS = {"vision": True, "tools": False, "min_context_tokens": 1000}
def test_reject_reasons_are_human_readable():
"""vision/tools/上下文三类不满足均有对应人话理由;满足则 None。"""
assert "vision" in (_entry_reject_reason(_entry("a", vision=False), _NEEDS) or "")
need_tools = {"vision": False, "tools": True, "min_context_tokens": 1000}
assert "tools" in (_entry_reject_reason(_entry("a", tools=False), need_tools) or "")
small = _entry("a", ctx=512)
assert "512" in (_entry_reject_reason(small, _NEEDS) or "")
assert _entry_reject_reason(_entry("a", ctx=8192), _NEEDS) is None
def test_rank_prefers_capable_and_cheap():
"""不满足需求者 capability=0 沉底;便宜者靠成本效率维胜出。"""
good_cheap = _entry("cheap", tier="budget", price_in=0.1, price_out=0.2, ctx=8192)
good_pricey = _entry("pricey", tier="premium", price_in=8.0, price_out=8.0, ctx=8192)
ranked = _rank_candidates([good_pricey, good_cheap], _NEEDS)
assert ranked[0]["model"] == "cheap"
assert ranked[0]["dims"]["capability"] > 0
assert ranked[0]["dims"]["cost_efficiency"] > ranked[1]["dims"]["cost_efficiency"]
# premium 的质量维更高(启发式)
assert ranked[1]["dims"]["quality"] > ranked[0]["dims"]["quality"]
def test_rank_tie_break_deterministic():
"""同配置同分按模型名字典序。"""
a, b = _entry("zeta"), _entry("alpha")
ranked = _rank_candidates([a, b], _NEEDS)
assert [s["model"] for s in ranked] == ["alpha", "zeta"]
def test_score_event_records_rejections():
"""评分留痕:入选排序 + 被淘汰候选的拒绝理由;reset 清空。"""
reset_route_score_events()
ranked = _rank_candidates([_entry("good", ctx=8192)], _NEEDS)
rejected = [{"model": "bad", "reason": "模型不支持视觉输入(vision)"}]
_record_route_score_event("px-test", ranked, rejected)
_record_route_score_event("px-test-2", ranked, rejected)
assert len(R._route_score_events) == 2
ev = R._route_score_events[-1]
assert ev["request_id"] == "px-test-2"
assert ev["ranked"][0]["model"] == "good"
assert ev["rejected"][0]["reason"].startswith("模型不支持")
reset_route_score_events()
assert len(R._route_score_events) == 0
+87
View File
@@ -0,0 +1,87 @@
"""语义缓存路由签名分桶(T-X10,采纳 cortiq 路由签名设计)。"""
from gateway.proxy.config import build_proxy_config
from gateway.proxy.normalizer import canonical_hash
from gateway.proxy.routes import _route_sig
_BODY = {"model": "m1", "messages": [{"role": "user", "content": "什么是递归"}]}
def test_canonical_hash_legacy_format_without_sig():
"""不带签名保持旧三段格式(旧调用/旧条目兼容)。"""
h = canonical_hash("default", 1, _BODY)
assert h.count("|") == 2
h2 = canonical_hash("default", 1, _BODY, "")
assert h == h2
def test_canonical_hash_with_sig_divides_keyspace():
"""签名不同 -> 键不同:vision 请求与纯文本请求不再共用缓存条目。"""
plain = canonical_hash("default", 1, _BODY)
vision = canonical_hash("default", 1, _BODY, "v1t0")
tools = canonical_hash("default", 1, _BODY, "v0t1")
assert len({plain, vision, tools}) == 3
def test_route_sig_scopes():
"""三种 scope 的签名形态:capabilities 默认 / model 精确 / none 旧格式。"""
cfg = build_proxy_config({"proxy": {"enabled": True}})
needs = {"vision": True, "tools": False, "min_context_tokens": 10}
assert _route_sig(needs, "m1", cfg) == "v1t0"
cfg_model = build_proxy_config({"proxy": {"semcache": {"route_sig_scope": "model"}}})
assert _route_sig(needs, "m1", cfg_model) == "m:m1"
cfg_none = build_proxy_config({"proxy": {"semcache": {"route_sig_scope": "none"}}})
assert _route_sig(needs, "m1", cfg_none) == ""
# 非法 scope 回落 capabilities
cfg_bad = build_proxy_config({"proxy": {"semcache": {"route_sig_scope": "bogus"}}})
assert _route_sig(needs, "m1", cfg_bad) == "v1t0"
def test_sig_isolation_end_to_end_via_semcache(tmp_path):
"""经 SemanticCache 验证:签名分区 L1 精确层与 L2 语义扫描两层。"""
from gateway.proxy.ledger import Ledger
from gateway.proxy.semcache import SemanticCache
led = Ledger.init_db(tmp_path / "l.sqlite3")
cache = SemanticCache(led, max_entries=100, sim_threshold=0.5,
promote_frequency=5)
k_text = canonical_hash("default", 1, _BODY, "v0t0")
k_vision = canonical_hash("default", 1, _BODY, "v1t0")
cache.put(k_text, "什么是递归", "纯文本答案", "m1", doc_version=1,
route_sig="v0t0")
# 相同问题、vision 签名:L1 键不同必不中;L2 语义扫描按签名分区也不中
assert cache.lookup(k_vision, "什么是递归", doc_version=1,
route_sig="v1t0") is None
# 同签名:L1 精确命中
hit = cache.lookup(k_text, "什么是递归", doc_version=1, route_sig="v0t0")
assert hit is not None and hit["level"] == "exact"
def test_sig_partition_survives_rebuild(tmp_path):
"""重启重建(semcache 表无签名列):签名从缓存键解析恢复,分区仍有效。"""
from gateway.proxy.ledger import Ledger
from gateway.proxy.semcache import SemanticCache
led = Ledger.init_db(tmp_path / "l.sqlite3")
c1 = SemanticCache(led, max_entries=100, sim_threshold=0.5, promote_frequency=5)
k = canonical_hash("default", 1, _BODY, "v1t0")
c1.put(k, "什么是递归", "多模态答案", "m1", doc_version=1, route_sig="v1t0")
c2 = SemanticCache(led, max_entries=100, sim_threshold=0.5, promote_frequency=5)
assert c2.lookup(k, "什么是递归", doc_version=1, route_sig="v1t0") is not None
assert c2.lookup(canonical_hash("default", 1, _BODY, "v0t0"),
"什么是递归", doc_version=1, route_sig="v0t0") is None
def test_legacy_calls_without_sig_unchanged(tmp_path):
"""不传签名(route_sig=None)的旧调用零过滤:既有测试行为不变。"""
from gateway.proxy.ledger import Ledger
from gateway.proxy.semcache import SemanticCache
led = Ledger.init_db(tmp_path / "l.sqlite3")
cache = SemanticCache(led, max_entries=100, sim_threshold=0.5, promote_frequency=5)
cache.put("k1", "请解释一下递归函数的概念", "答案", "m1", doc_version=1)
assert cache.lookup("k1", "请解释一下递归函数的概念", doc_version=1) is not None
# 语义路径同样不受影响(近似文本跨键命中)
assert cache.lookup("k2", "请解释一下递归函数的概念呢", doc_version=1) is not None
+157
View File
@@ -0,0 +1,157 @@
"""决策缓存测试(T-X3):LRU+TTL 单元契约 + Grader live 模式接线(embed 去重)。"""
import asyncio
import pytest
from gateway.sense.config import build_sense_config
from gateway.sense.decision_cache import DecisionCache
from gateway.sense.grader import Grader
from gateway.sense.store import SenseStore
# ---------------- DecisionCache 单元 ----------------
class _Clock:
def __init__(self):
self.t = 100.0
def __call__(self):
return self.t
def test_key_scope_and_text_sensitive():
assert DecisionCache.key("proxy", "abc") == DecisionCache.key("proxy", "abc")
assert DecisionCache.key("proxy", "abc") != DecisionCache.key("pipeline", "abc")
assert DecisionCache.key("proxy", "abc") != DecisionCache.key("proxy", "abd")
def test_put_get_hit_miss_counters():
c = DecisionCache()
assert c.get("s", "q") is None
c.put("s", "q", {"tier": "T1"})
assert c.get("s", "q") == {"tier": "T1"}
st = c.stats()
assert st["hits"] == 1 and st["misses"] == 1 and st["entries"] == 1
def test_ttl_expiry_with_fake_clock():
clock = _Clock()
c = DecisionCache(ttl_s=60.0, now=clock)
c.put("s", "q", {"v": 1})
assert c.get("s", "q") == {"v": 1}
clock.t += 61.0 # 超时
assert c.get("s", "q") is None
assert c.stats()["entries"] == 0 # 过期条目被清除
def test_lru_eviction_at_maxsize():
c = DecisionCache(maxsize=2)
c.put("s", "a", 1)
c.put("s", "b", 2)
c.get("s", "a") # a 变为最近使用
c.put("s", "c", 3) # 淘汰 b
assert c.get("s", "a") == 1
assert c.get("s", "b") is None
assert c.get("s", "c") == 3
def test_clear_keeps_counters_but_drops_entries():
c = DecisionCache()
c.put("s", "q", 1)
c.get("s", "q")
c.clear()
assert c.get("s", "q") is None
assert c.stats()["hits"] == 1 # 计数保留(审计口径不回退)
# ---------------- Grader live 模式接线 ----------------
class _CountingObserver:
def __init__(self):
self.logs = []
def log(self, obs):
self.logs.append(obs)
class _FakeHead:
version = "test-head"
def predict(self, vec):
return {"t1": 0.95, "t2": 0.03, "t3": 0.02}
def _make_grader(tmp_path, mode, counter, observer=None):
"""构造测试 Grader:embed 打点计数 + FakeHead 注入。"""
import gateway.sense.grader as gr
cfg = build_sense_config({"sense": {
"enabled": True, "mode": mode,
"db_path": str(tmp_path / "s.sqlite3"),
"models_dir": str(tmp_path / "models")}})
store = SenseStore.init_db(cfg.db_path)
g = Grader(cfg, store, observer or _CountingObserver())
g._load_head = lambda: _FakeHead()
async def fake_embed(text, c):
counter["n"] += 1
return [1, 2, 3]
orig = gr.embed
gr.embed = fake_embed
return g, orig
def test_grader_live_mode_dedupes_embed(tmp_path):
"""live 模式:同文本第二次 decide 命中缓存,embed 只跑一次;观察仍写两条。"""
import gateway.sense.grader as gr
counter = {"n": 0}
observer = _CountingObserver()
g, orig = _make_grader(tmp_path, "live", counter, observer)
try:
async def run():
d1 = await g.decide("什么是递归", "proxy")
d2 = await g.decide("什么是递归", "proxy")
d3 = await g.decide("换一个问题", "proxy")
return d1, d2, d3
d1, d2, d3 = asyncio.run(run())
finally:
gr.embed = orig
assert d1.tier == "T1" and d2.tier == "T1" and d3.tier == "T1"
assert counter["n"] == 2 # 同文本去重:2 个不同文本各 1 次
assert d2.head_version == d1.head_version
assert len(observer.logs) == 3 # 观察不因缓存命中而跳过
def test_grader_collect_mode_no_cache(tmp_path):
"""collect/shadow:校准数据必须全量产出,不走缓存。"""
import gateway.sense.grader as gr
counter = {"n": 0}
g, orig = _make_grader(tmp_path, "collect", counter)
try:
async def run():
await g.decide("同一个问题", "proxy")
await g.decide("同一个问题", "proxy")
asyncio.run(run())
finally:
gr.embed = orig
assert counter["n"] == 2
def test_grader_invalidate_clears_cache(tmp_path):
"""invalidate(工件切换)后缓存清空:同文本重新 embed。"""
import gateway.sense.grader as gr
counter = {"n": 0}
g, orig = _make_grader(tmp_path, "live", counter)
try:
async def run():
await g.decide("什么是递归", "proxy")
g.invalidate()
await g.decide("什么是递归", "proxy")
asyncio.run(run())
finally:
gr.embed = orig
assert counter["n"] == 2
+200
View File
@@ -0,0 +1,200 @@
"""决策留痕测试(T-X4):store 三元结构落库 + grader 留痕 + admin 查询端点。"""
import asyncio
import pytest
pytest.importorskip("fastapi")
from fastapi import FastAPI
from fastapi.testclient import TestClient
import gateway.model_pool as mp
import gateway.sense.grader as gr
from gateway.sense.config import build_sense_config
from gateway.sense.grader import Grader
from gateway.sense.store import SenseStore
from gateway.proxy.config import build_proxy_config
from gateway.proxy.routes import build_proxy_router
class _CountingObserver:
def __init__(self):
self.logs = []
def log(self, obs):
self.logs.append(obs)
class _FakeHead:
version = "test-head"
def predict(self, vec):
return {"t1": 0.95, "t2": 0.03, "t3": 0.03}
def _make_grader(tmp_path, mode="live", embed_raises=False):
cfg = build_sense_config({"sense": {
"enabled": True, "mode": mode,
"db_path": str(tmp_path / "s.sqlite3"),
"models_dir": str(tmp_path / "models")}})
store = SenseStore.init_db(cfg.db_path)
g = Grader(cfg, store, _CountingObserver())
g._load_head = lambda: _FakeHead()
async def fake_embed(text, c):
if embed_raises:
raise __import__("gateway.sense.errors",
fromlist=["EmbedderDown"]).EmbedderDown("down")
return [1, 2, 3]
orig = gr.embed
gr.embed = fake_embed
return g, store, orig
def test_store_decision_roundtrip(tmp_path):
store = SenseStore.init_db(tmp_path / "s.sqlite3")
store.insert_decision({
"ts": 1789874000, "request_id": "rt1", "consumer": "proxy",
"decided_tier": "T1", "executed_tier": "T1", "mode": "live",
"fallback": False, "policy_version": "head:test|th:conservative",
"q_hash": "abc123",
"reasons": '["t1_ok", "p1>=threshold"]',
"candidate_scores": '{"t1": 0.95, "t2": 0.03, "t3": 0.03}',
"rejected": '[{"tier": "T3", "reason": "p3<threshold 且无 repo_signals"}]',
})
rows = store.list_decisions(limit=10)
assert len(rows) == 1
r = rows[0]
assert r["decided_tier"] == "T1" and r["mode"] == "live"
assert r["reasons"][0] == "t1_ok"
assert r["candidate_scores"]["t1"] == 0.95
assert r["rejected"][0]["tier"] == "T3"
def test_grader_records_decision_triplet(tmp_path):
"""正常决策:reasons/candidate_scores/rejected 全量落库。"""
g, store, orig = _make_grader(tmp_path)
try:
asyncio.run(g.decide("什么是递归", "proxy"))
finally:
gr.embed = orig
rows = store.list_decisions()
assert len(rows) == 1
r = rows[0]
assert r["decided_tier"] == "T1"
assert "t1_ok" in r["reasons"] and "p1>=threshold" in r["reasons"]
assert r["rejected"][0]["tier"] == "T3"
assert r["candidate_scores"]["t1"] == 0.95
assert r["q_hash"] and len(r["q_hash"]) == 32
def test_grader_records_fallback_reasons(tmp_path):
"""embedder 故障:fallback 规则门决策同样留痕并标注 fallback。"""
g, store, orig = _make_grader(tmp_path, embed_raises=True)
try:
d = asyncio.run(g.decide("什么是递归", "proxy"))
finally:
gr.embed = orig
assert d.fallback is True
rows = store.list_decisions()
assert len(rows) == 1
r = rows[0]
assert r["fallback"] == 1
assert "fallback:rule_gate" in r["reasons"]
assert {x["reason"] for x in r["rejected"]} == {"fallback"}
def test_admin_sense_decisions_endpoint(tmp_path, monkeypatch):
"""管理面查询端点:sense 未注入/未启用 -> enabled False;不报错。"""
import gateway.model_pool as mp
mp.reset_pool()
monkeypatch.setattr(mp, "_store",
__import__("gateway.model_pool",
fromlist=["PoolStore"]).PoolStore(
path=tmp_path / "pool.json"))
cfg = build_proxy_config({"proxy": {"enabled": True,
"db_path": str(tmp_path / "p.sqlite3")}})
app = FastAPI()
app.include_router(build_proxy_router(cfg, mp.get_pool()))
client = TestClient(app)
r = client.get("/proxy/admin/sense-decisions")
assert r.status_code == 200
body = r.json()
assert body["enabled"] is False and body["decisions"] == []
def test_proxy_routes_sense_live_regression(tmp_path, monkeypatch):
"""T-X4 回归:settings_provider 注入后,sense live 分流真正生效
(修复前 NameError 被 except 吞掉,x-campus-tier 永远缺失)。"""
import httpx
import gateway.proxy.upstream as up
from gateway.proxy.auth import issue_key
from gateway.proxy.ledger import Ledger
sense_dir = tmp_path / "sense"
cfg = build_proxy_config({"proxy": {
"enabled": True, "db_path": str(tmp_path / "p.sqlite3"),
"pricing": {"m1": {"in_miss": 1.0, "out": 2.0}}}})
# 签发学生 key(代理面鉴权链:Bearer 形态 -> 哈希 -> 日限额 -> rpm)
led = Ledger.init_db(tmp_path / "p.sqlite3")
sid = led.upsert_student("测试生", "软件2201", balance_yuan=10.0,
daily_cap_yuan=10.0)
issued = issue_key(led, sid, rpm_cap=100, day_cap_req=1000)
auth_header = {"Authorization": f"Bearer {issued['key']}"}
import gateway.model_pool as mp
mp.reset_pool()
pool = mp.PoolStore(path=tmp_path / "pool.json")
pool.upsert({"id": "t1e", "name": "本地", "tier": "local",
"backend": "openai", "base_url": "http://mockup",
"model": "local-m", "enabled": True})
class _Settings:
def to_dict(self):
return {"sense": {
"enabled": True, "mode": "live",
"db_path": str(sense_dir / "s.sqlite3"),
"models_dir": str(sense_dir / "models"),
"consumers": {"proxy": {"t1": "local", "t2": "budget",
"t3": "premium"}}}}
app = FastAPI()
app.include_router(build_proxy_router(cfg, pool, settings_provider=_Settings))
def handler(request: httpx.Request) -> httpx.Response:
body = ("\n\n".join([
'data: {"choices":[{"delta":{"content":"hi"}}]}',
"data: [DONE]",
]) + "\n\n").encode("utf-8")
return httpx.Response(200, content=body)
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
orig_client = up._client
up._client = client
async def fake_embed(text, c):
return [1, 2, 3]
orig_embed = gr.embed
gr.embed = fake_embed
try:
c = TestClient(app)
r = c.post("/proxy/v1/chat/completions",
json={"model": "local-m",
"messages": [{"role": "user", "content": "什么是递归"}]},
headers=auth_header)
assert r.status_code == 200
# 短文本 + t1_hard_ok + p1>=0.9 -> T1 生效并替换为 local 档模型
assert r.headers.get("x-campus-tier") == "T1"
assert r.json()["model"] == "local-m"
# 决策留痕可经管理面查询
d = c.get("/proxy/admin/sense-decisions")
assert d.status_code == 200
decisions = d.json()["decisions"]
assert len(decisions) == 1 and decisions[0]["decided_tier"] == "T1"
finally:
up._client = orig_client
gr.embed = orig_embed
mp.reset_pool()
+178
View File
@@ -0,0 +1,178 @@
"""晋升表测试(T-X5):candidate→promoted→demoted→恢复状态机 + 一票否决 +
grader 晋升应用(T2→T1)与 T3/闸门互斥。"""
import asyncio
import pytest
from gateway.sense.config import build_sense_config
from gateway.sense.grader import Grader
from gateway.sense.promotion import PromotionTable, promotion_label
from gateway.sense.store import SenseStore
class _Clock:
def __init__(self, t=1789874000.0):
self.t = t
def __call__(self):
return self.t
def _table(tmp_path, **kw):
store = SenseStore.init_db(tmp_path / "s.sqlite3")
clock = _Clock()
table = PromotionTable(store, now=clock, **kw)
return store, clock, table
def test_promotion_requires_min_and_soak(tmp_path):
"""n_min 未满不晋升; soak 期未满不晋升;两者满足才 promoted。"""
store, clock, table = _table(tmp_path, n_min=5, promote_lb=0.9,
soak_days=3.0)
label = "proxy:general"
for _ in range(5):
table.observe(label, ok=True, ts=clock.t) # 通过率 1.0 但浸泡期未满
assert not table.is_promoted(label)
clock.t += 4 * 86400.0 # 浸泡期满足
table.observe(label, ok=True, ts=clock.t)
assert table.is_promoted(label)
def test_promotion_demotes_on_degradation(tmp_path):
"""promoted 后通过率跌破 demote_lb -> 自动降级。"""
store, clock, table = _table(tmp_path, n_min=5, promote_lb=0.9,
demote_lb=0.85, soak_days=0.0)
label = "proxy:general"
for _ in range(6):
table.observe(label, ok=True, ts=clock.t)
assert table.is_promoted(label)
for _ in range(6):
clock.t += 60
table.observe(label, ok=False, ts=clock.t) # 通过率跌至 6/12 = 0.5
assert not table.is_promoted(label)
assert table.snapshot(label)["state"] == "demoted"
def test_promotion_recovery_after_demotion(tmp_path):
"""demoted 后凭数据恢复:再次满足晋升门 -> promoted。"""
store, clock, table = _table(tmp_path, n_min=4, promote_lb=0.9,
demote_lb=0.85, soak_days=0.0)
label = "proxy:general"
for _ in range(4):
table.observe(label, ok=True, ts=clock.t)
assert table.is_promoted(label)
for _ in range(4):
clock.t += 60
table.observe(label, ok=False, ts=clock.t) # 跌破
assert table.snapshot(label)["state"] == "demoted"
for _ in range(20):
clock.t += 60
table.observe(label, ok=True, ts=clock.t) # 20/24 ≈ 0.83... 继续
# 24 条中 20 ok = 0.833 < 0.9,再补 ok 至 >0.9
for _ in range(6):
clock.t += 60
table.observe(label, ok=True, ts=clock.t)
# 30 条 26 ok ≈ 0.867,仍 <0.9;再补
for _ in range(6):
clock.t += 60
table.observe(label, ok=True, ts=clock.t)
# 36 条 32 ok ≈ 0.889;最后一条后 >= 0.9
table.observe(label, ok=True, ts=clock.t + 60) # 37 条 33 ok ≈ 0.892
for _ in range(4):
table.observe(label, ok=True, ts=clock.t + 120)
# 41 条 37 ok ≈ 0.902 >= 0.9
assert table.snapshot(label)["state"] == "promoted"
def test_high_tier_veto(tmp_path):
"""T3 观察:不计入通过率;已晋升立即降级。"""
store, clock, table = _table(tmp_path, n_min=3, promote_lb=0.9, soak_days=0.0)
label = "proxy:general"
for _ in range(3):
table.observe(label, ok=True, ts=clock.t)
assert table.is_promoted(label)
before = table.snapshot(label)["n_total"]
table.observe(label, ok=True, tier="T3", ts=clock.t + 10)
snap = table.snapshot(label)
assert snap["n_total"] == before # T3 不进统计
assert snap["state"] == "demoted" # 一票否决
def test_promotion_label_format():
assert promotion_label("proxy", "") == "proxy:general"
assert promotion_label("proxy", "code") == "proxy:code"
# ---------------- grader 晋升应用 ----------------
class _FakeHeadMid:
version = "test-head"
def predict(self, vec):
return {"t1": 0.50, "t2": 0.47, "t3": 0.03} # 低于阈值 -> default T2
class _NoopObserver:
def log(self, obs):
pass
def _make(tmp_path, table):
cfg = build_sense_config({"sense": {
"enabled": True, "mode": "live",
"db_path": str(tmp_path / "s.sqlite3"),
"models_dir": str(tmp_path / "models")}})
store = SenseStore.init_db(cfg.db_path)
g = Grader(cfg, store, _NoopObserver(), promotion=table)
g._load_head = lambda: _FakeHeadMid()
async def fake_embed(text, c):
return [1, 2, 3]
import gateway.sense.grader as gr
orig = gr.embed
gr.embed = fake_embed
return g, orig
def test_grader_applies_promotion_t2_to_t1(tmp_path):
"""已晋升标签:T2 决策升级 T1 且 hard_gates 标注 promotion_applied。"""
store = SenseStore.init_db(tmp_path / "s.sqlite3")
clock = _Clock()
table = PromotionTable(store, now=clock, n_min=2, promote_lb=0.9, soak_days=0.0)
label = promotion_label("proxy", "general")
table.observe(label, ok=True, ts=clock.t)
table.observe(label, ok=True, ts=clock.t)
g, orig = _make(tmp_path, table)
import gateway.sense.grader as gr
try:
d = asyncio.run(g.decide("什么是递归", "proxy"))
finally:
gr.embed = orig
assert d.tier == "T1"
assert d.hard_gates["promotion_applied"] is True
def test_grader_skips_promotion_when_not_promoted(tmp_path):
"""未晋升:同样的 T2 决策保持 T2。"""
store = SenseStore.init_db(tmp_path / "s2.sqlite3")
table = PromotionTable(store, now=_Clock(), n_min=100, soak_days=0.0)
g, orig = _make(tmp_path, table)
import gateway.sense.grader as gr
try:
d = asyncio.run(g.decide("什么是递归", "proxy"))
finally:
gr.embed = orig
assert d.tier == "T2"
assert d.hard_gates["promotion_applied"] is False
def test_grader_promotion_not_applied_without_table(tmp_path):
"""未接晋升表(向后兼容):行为与既有决策一致。"""
g, orig = _make(tmp_path, None)
import gateway.sense.grader as gr
try:
d = asyncio.run(g.decide("什么是递归", "proxy"))
finally:
gr.embed = orig
assert d.tier == "T2"
+128
View File
@@ -0,0 +1,128 @@
"""shadow 旁路评审单元测试(T-X11,采纳 cortiq shadow bypass 设计)。
全部走 monkeypatch 假上游,不依赖真实模型/API key。
"""
import asyncio
import json
import pytest
import gateway.proxy.routes as R
from gateway.proxy.routes import (_grade_answer, _pick_review_entry,
_shadow_review_settings,
reset_shadow_review_stats)
from gateway.sense.promotion import PromotionTable, promotion_label
@pytest.fixture(autouse=True)
def _reset():
reset_shadow_review_stats()
yield
reset_shadow_review_stats()
def _fake_upstream(responses):
"""构造假 upstream_stream:按调用顺序吐预定 verdict 文本(SSE 形态)。"""
calls = {"n": 0}
def _stream(body, entry, sink, chain):
async def gen():
idx = min(calls["n"], len(responses) - 1)
calls["n"] += 1
verdict = responses[idx]
chunk = {"choices": [{"delta": {"content": verdict}}]}
yield (f"data: {json.dumps(chunk)}\n\n").encode("utf-8")
yield b"data: [DONE]\n\n"
return gen()
return _stream, calls
def _entry(tier="premium", model="cloud-x"):
return {"id": "e1", "enabled": True, "backend": "openai",
"base_url": "http://127.0.0.1:9/v1", "model": model, "tier": tier}
def test_settings_default_disabled_and_clamped():
"""默认关闭;采样率/超时非法值被夹取。"""
s = _shadow_review_settings({})
assert s["enabled"] is False and s["sample_rate"] == 0.1
assert _shadow_review_settings({"sense": {"shadow_review": {
"enabled": True, "sample_rate": 5, "timeout_s": -3}}})["sample_rate"] == 1.0
assert _shadow_review_settings({"sense": {"shadow_review": {
"enabled": True, "timeout_s": -3}}})["timeout_s"] == 1.0
def test_pick_review_entry_prefers_tier():
"""评审条目优先指定档位,缺失时回退任意启用条目。"""
pool_entries = [_entry("budget", "b1"), _entry("premium", "p1")]
class _P:
@staticmethod
def usable_entries():
return list(pool_entries)
assert _pick_review_entry(_P, "premium")["model"] == "p1"
assert _pick_review_entry(_P, "local")["model"] in ("b1", "p1")
def test_grade_answer_parses_pass_fail_and_error(monkeypatch):
"""PASS/FAIL 解析;上游异常回 ERROR 而非抛出。"""
fake, calls = _fake_upstream(["PASS", "FAIL"])
monkeypatch.setattr(R, "upstream_stream", fake)
e = _entry()
assert asyncio.run(_grade_answer(e, "问", "答", "cloud-x", 5)) == "PASS"
assert asyncio.run(_grade_answer(e, "问", "答", "cloud-x", 5)) == "FAIL"
def _boom(*a, **k):
async def gen():
raise RuntimeError("上游炸了")
yield b"" # pragma: no cover
return gen()
monkeypatch.setattr(R, "upstream_stream", _boom)
assert asyncio.run(_grade_answer(e, "问", "答", "cloud-x", 5)) == "ERROR"
assert calls["n"] == 2
def test_shadow_review_task_feeds_promotion(tmp_path, monkeypatch):
"""PASS/FAIL 写入晋升表(observe 计数),ERROR 只计错误不进通过率。"""
from gateway.sense.store import SenseStore
sstore = SenseStore.init_db(str(tmp_path / "sense.sqlite3"))
promo = PromotionTable(sstore, now=lambda: 1_000_000.0)
label = promotion_label("proxy", "")
fake, _ = _fake_upstream(["PASS", "PASS", "FAIL", "垃圾输出"])
monkeypatch.setattr(R, "upstream_stream", fake)
entry = _entry()
sd = {"sense": {"shadow_review": {"enabled": True, "sample_rate": 1.0}}}
async def _run():
for _ in range(4):
await R._shadow_review_task("问题", "本地答案", _FixedPool(entry),
sd, promo)
asyncio.run(_run())
assert R._shadow_review_stats == {"sampled": 4, "pass": 2, "fail": 1,
"error": 1, "skipped_busy": 0}
row = sstore.get_promotion(label)
assert row["n_total"] == 3 and row["n_ok"] == 2 # ERROR 不进通过率
class _FixedPool:
def __init__(self, entries):
self._e = entries if isinstance(entries, list) else [entries]
def usable_entries(self):
return list(self._e)
def test_maybe_shadow_review_disabled_is_noop(monkeypatch):
"""默认关闭:不建任务、不计采样。"""
class _SP:
@staticmethod
def to_dict():
return {"sense": {"shadow_review": {"enabled": False}}}
payload = json.dumps({"choices": [{"message": {"content": "本地答案"}}]}).encode()
resp = type("JR", (), {"body": payload})()
before = dict(R._shadow_review_stats)
R._maybe_shadow_review({}, resp, _FixedPool(_entry()), _SP, scfg=None)
assert R._shadow_review_stats == before
+70
View File
@@ -0,0 +1,70 @@
"""TPM 双桶限流与缓存命中率透出(T-X12,采纳 enterprise-ai-gateway 设计)。"""
import time
from gateway.proxy.auth import RateLimiter
from gateway.proxy.semcache import SemanticCache
def test_tpm_bucket_capacity_and_consume():
"""容量内放行并扣减;超过余量拒绝。"""
rl = RateLimiter()
assert rl.allow_tokens(1, 100, 1000) is True
assert rl.allow_tokens(1, 500, 1000) is True
# 已扣 600,剩 400 + 回流(瞬时近似 0):700 > 400+eps -> 拒
assert rl.allow_tokens(1, 700, 1000) is False
def test_tpm_bucket_continuous_refill(monkeypatch):
"""连续时间回流:elapsed * tpm/60 补充(容量封顶)。"""
rl = RateLimiter()
clock = {"t": 100.0}
monkeypatch.setattr(time, "monotonic", lambda: clock["t"])
assert rl.allow_tokens(2, 1000, 2000) is True # 剩 1000
clock["t"] = 130.0 # 30s 回流 1000 -> 满 2000
assert rl.allow_tokens(2, 1500, 2000) is True # 扣 1500 剩 500
clock["t"] = 145.0 # 15s 回流 500 -> 1000
assert rl.allow_tokens(2, 1000, 2000) is True # 精确扣空
assert rl.allow_tokens(2, 1, 2000) is False # 空桶拒绝
def test_tpm_disabled_when_cap_zero():
"""tpm_cap <= 0 = 不限流(恒放行)。"""
rl = RateLimiter()
assert rl.allow_tokens(3, 10**9, 0) is True
def test_tpm_independent_from_rpm_bucket():
"""双桶相互独立:打满 TPM 不影响 RPM 放行,RPM 打空不影响 TPM 记账。"""
rl = RateLimiter()
assert rl.allow_tokens(4, 5000, 5000) is True # TPM 打空
assert rl.allow(4, 10) is True # RPM 仍放行
for _ in range(9): # 共 10 次 = 打满 RPM
assert rl.allow(4, 10) is True
assert rl.allow(4, 10) is False # RPM 空桶
assert rl.allow_tokens(4, 1, 5000) is False # TPM 仍空(未被 RPM 动过)
def test_tpm_per_key_isolation():
"""按 key 隔离:一个 key 打满不影响另一个。"""
rl = RateLimiter()
assert rl.allow_tokens(5, 5000, 5000) is True
assert rl.allow_tokens(6, 5000, 5000) is True
assert rl.allow_tokens(5, 5000, 5000) is False
def test_semcache_stats_hit_rate():
"""stats 透出 misses 与 hit_rate(命中率口径 = hits / (hits + misses))。"""
class _S:
def semcache_rows(self, limit=0):
return []
def put_semcache(self, *a, **k):
pass
c = SemanticCache(_S(), max_entries=10, sim_threshold=0.5, promote_frequency=5)
c.put("k1", "请解释一下递归函数的概念", "答案", "m", doc_version=1)
assert c.lookup("k1", "请解释一下递归函数的概念", doc_version=1) is not None
assert c.lookup("k9", "完全无关的问题内容呢", doc_version=1) is None
s = c.stats()
assert s["hits_exact"] == 1 and s["misses"] == 1
assert s["hit_rate"] == 0.5
+7
View File
@@ -165,3 +165,10 @@ P0 完成后的能力:干净的后端抽象 + 可量化的评测 + 可追溯
| T-G7 | 审计+前端:ReviewQueue 抽样 + tier 指标卡 + 客户端来源显示/一键升级 | ✅ 完成 | T-G7 |
| T-G8 | 实验:E-G1/E-G3 报告;(可选)LoraRemote + E-G2 线性 vs LoRA | ✅ 完成 | T-G8 |
| OPT-1 | 分支推进:语义缓存 L2 查找 3.39x(免并集计分+预筛)+ 安全加固(15 高危清零:SSRF/路径穿越/假凭据) | ✅ 完成 | ad3bf41 |
| T-X1 | 预算四档渐进干预(外部采纳 ai-model-router):budget_mode 整数基点判定(80/95/100%)+ optimize/cheap 自动降档 + X-Budget-Mode 上报;黄金用例锁边界 | ✅ 完成 | T-X1 |
| T-X2 | 上游有序降级链(外部采纳 cortiq tier 链):池内候选按档位/单价排序、首 token 前 failover、X-Upstream-Fallback 三元组响应头 + admin/stats failover 聚合 | ✅ 完成 | T-X2 |
| T-X3 | 路由决策缓存(外部采纳 cortiq 决策哈希缓存):DecisionCache(sha256+LRU+TTL60s,4096 条)前置 grader live 模式,embed+线性头去重;观察/审计不跳过、invalidate 同步清空、collect/shadow 不缓存 | ✅ 完成 | T-X3 |
| T-X6 | 模型池能力位过滤(外部采纳 cortiq capabilities):条目 capabilities{vision,tools,context_window}(缺省全兼容)+ filter_by_capabilities 硬过滤 + 代理请求需求推断(图片→vision/tools/上下文)与重定向 | ✅ 完成 | T-X6 |
| T-X4 | 可解释决策留痕(外部采纳 ai-model-router 决策三元):sense_decisions 表(reasons/candidate_scores/rejected)+ grader 全模式落库 + /proxy/admin/sense-decisions 查询;附带修复 settings_provider 未注入导致 sense live 分流失效的潜伏 bug | ✅ 完成 | T-X4 |
| T-X5 | 云端评判晋升表×conformal 双闸门(外部采纳 cortiq promotion):PromotionTable 状态机(n_min/通过率/soak 晋升、退化降级、T3 一票否决)+ grader live 模式 T2→T1 应用 + 人工核验 verdict 回填(promo 标签) | ✅ 完成 | T-X5 |
| T-X7~X9 | P2 立项不实施:多协议透传(/v1/messages)/ 工具输出清洗沙箱(受限 DSL)/ 网关内置 A/B 实验框架 | 📋 立项 | 待排期 |