From 7e01dc9b163f031962ba454da10baa9f1ba664b4 Mon Sep 17 00:00:00 2001 From: tzt <14718231+flying-travel@user.noreply.gitee.com> Date: Fri, 18 Sep 2026 22:55:49 +0800 Subject: [PATCH] =?UTF-8?q?feat(sense):=20T-X4=20=E5=8F=AF=E8=A7=A3?= =?UTF-8?q?=E9=87=8A=E5=86=B3=E7=AD=96=E7=95=99=E7=97=95=E4=B8=89=E5=85=83?= =?UTF-8?q?=E7=BB=93=E6=9E=84=20+=20=E4=BF=AE=E5=A4=8D=20sense=20live=20?= =?UTF-8?q?=E5=88=86=E6=B5=81=E6=BD=9C=E4=BC=8F=20bug=EF=BC=88=E9=87=87?= =?UTF-8?q?=E7=BA=B3=20ai-model-router=20=E5=86=B3=E7=AD=96=E6=A8=A1?= =?UTF-8?q?=E5=9E=8B=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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） --- gateway/api.py | 3 +- gateway/proxy/routes.py | 77 +++++++++---- gateway/sense/grader.py | 68 ++++++++++++ gateway/sense/store.py | 47 ++++++++ tests/test_sense_decisions.py | 200 ++++++++++++++++++++++++++++++++++ 任务拆解与执行计划.md | 1 + 6 files changed, 371 insertions(+), 25 deletions(-) create mode 100644 tests/test_sense_decisions.py diff --git a/gateway/api.py b/gateway/api.py index 51d4f50..cf07e6b 100644 --- a/gateway/api.py +++ b/gateway/api.py @@ -208,7 +208,8 @@ 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)) from gateway.proxy.routes import install_error_handlers install_error_handlers(app) except Exception as _pe: # pragma: no cover - 代理层装配失败不拖垮主应用 diff --git a/gateway/proxy/routes.py b/gateway/proxy/routes.py index bcfbb22..15e8b94 100644 --- a/gateway/proxy/routes.py +++ b/gateway/proxy/routes.py @@ -46,8 +46,12 @@ 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) -> APIRouter: + """组装代理面路由（唯一组装点）。 + + settings_provider：返回设置对象的回调（api.py 注入 settings_store）。 + 未注入时语义分析器分流不启用（None 安全，既有测试不受影响）。 + """ router = APIRouter(prefix="/proxy") ledger = __import__("gateway.proxy.ledger", fromlist=["Ledger"]).Ledger.init_db(cfg.db_path) # ---------------- 学生面 ---------------- @@ -80,29 +84,32 @@ def build_proxy_router(cfg: ProxyConfig, pool) -> APIRouter: if not _limits().acquire_slot(ctx["key_id"]): raise QuotaError("并发请求已达该 key 上限") - # 语义分析器 live 分流（D-G7：mode=live 才启用；任何异常不影响代理可用性） + # 语义分析器 live 分流（D-G7：mode=live 才启用；任何异常不影响代理可用性）。 + # T-X4 修复：settings_provider 由 api.py 注入（原实现引用未导入的 + # settings_store，NameError 被 except 吞掉，分流从未实际生效）。 tier_used = None - 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：分级故障不影响代理可用性 + if settings_provider is not None: + 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_provider().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) if tier_used: @@ -208,6 +215,28 @@ def build_proxy_router(cfg: ProxyConfig, pool) -> APIRouter: "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, limit: int = 50, offset: int = 0): diff --git a/gateway/sense/grader.py b/gateway/sense/grader.py index f635c3d..724d514 100644 --- a/gateway/sense/grader.py +++ b/gateway/sense/grader.py @@ -21,6 +21,7 @@ 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.store import json_dumps @dataclass @@ -44,6 +45,57 @@ 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 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 + # ---------- 工件 ---------- def register_artifact(self, version: str, kind: str, path: str, metrics: Dict[str, Any], active: bool = False, diff --git a/tests/test_sense_decisions.py b/tests/test_sense_decisions.py new file mode 100644 index 0000000..7b2887b --- /dev/null +++ b/tests/test_sense_decisions.py @@ -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" 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() diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md index 04c7ad1..a88f504 100644 --- a/任务拆解与执行计划.md +++ b/任务拆解与执行计划.md @@ -169,3 +169,4 @@ P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯 | 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 |