From e00e2cd2b13c45b9a701685259f1e1f52a6e3966 Mon Sep 17 00:00:00 2001 From: tzt <14718231+flying-travel@user.noreply.gitee.com> Date: Fri, 18 Sep 2026 23:59:44 +0800 Subject: [PATCH] =?UTF-8?q?feat(proxy):=20=E6=9E=B6=E6=9E=84=E4=B8=8E?= =?UTF-8?q?=E7=AE=97=E6=B3=95=E4=BC=98=E5=8C=96=E4=BA=8C=E8=BD=AE=E2=80=94?= =?UTF-8?q?=E2=80=94sense=20=E8=BF=90=E8=A1=8C=E6=97=B6=E5=8D=95=E4=BE=8B?= =?UTF-8?q?=E6=BF=80=E6=B4=BB=20T-X3=E3=80=81=E6=89=93=E7=A0=81=E5=AF=86?= =?UTF-8?q?=E9=92=A5=E6=B3=84=E6=BC=8F=E4=BF=AE=E5=A4=8D=E3=80=81=E8=B4=A6?= =?UTF-8?q?=E6=9C=AC=E7=83=AD=E8=B7=AF=E5=BE=84=202.4x?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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） --- gateway/api.py | 3 +- gateway/model_pool.py | 12 +++- gateway/proxy/billing.py | 33 +++++++++++ gateway/proxy/ledger.py | 34 +++++++++--- gateway/proxy/routes.py | 115 +++++++++++++++++++++++++++------------ gateway/sense/grader.py | 8 ++- 6 files changed, 156 insertions(+), 49 deletions(-) diff --git a/gateway/api.py b/gateway/api.py index 4e76b36..84714fd 100644 --- a/gateway/api.py +++ b/gateway/api.py @@ -209,7 +209,8 @@ try: _proxy_cfg = build_proxy_config(settings_store().to_dict()) if _proxy_cfg.enabled: app.include_router(build_proxy_router(_proxy_cfg, get_pool(), - settings_provider=settings_store)) + 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 - 代理层装配失败不拖垮主应用 diff --git a/gateway/model_pool.py b/gateway/model_pool.py index a1bc67d..aa49aad 100644 --- a/gateway/model_pool.py +++ b/gateway/model_pool.py @@ -116,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"]: diff --git a/gateway/proxy/billing.py b/gateway/proxy/billing.py index 6b4d608..0d4e63b 100644 --- a/gateway/proxy/billing.py +++ b/gateway/proxy/billing.py @@ -179,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）；按学生过滤走联表常量语句。""" diff --git a/gateway/proxy/ledger.py b/gateway/proxy/ledger.py index 6459ecf..33cfbf4 100644 --- a/gateway/proxy/ledger.py +++ b/gateway/proxy/ledger.py @@ -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() diff --git a/gateway/proxy/routes.py b/gateway/proxy/routes.py index d7c7c62..a944225 100644 --- a/gateway/proxy/routes.py +++ b/gateway/proxy/routes.py @@ -46,11 +46,14 @@ def install_error_handlers(app) -> None: headers={"WWW-Authenticate": "Bearer"} if exc.status_code == 401 else None) -def build_proxy_router(cfg: ProxyConfig, pool, settings_provider=None) -> 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) @@ -87,19 +90,16 @@ def build_proxy_router(cfg: ProxyConfig, pool, settings_provider=None) -> APIRou # 语义分析器 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 - 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 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), - promotion=PromotionTable(sstore)) + _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 ""), @@ -113,17 +113,22 @@ def build_proxy_router(cfg: ProxyConfig, pool, settings_provider=None) -> APIRou 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 入队人工核； - # promo 标签供人工 verdict 回填晋升表（T-X5） + # 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", "promo:" + promotion_label("proxy", "")], @@ -136,7 +141,7 @@ def build_proxy_router(cfg: ProxyConfig, pool, settings_provider=None) -> APIRou 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 @@ -191,32 +196,27 @@ def build_proxy_router(cfg: ProxyConfig, pool, settings_provider=None) -> APIRou @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"] 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, + "revenue_milli": s["revenue_milli"], "cost_milli": s["cost_milli"], + "margin_milli": s["revenue_milli"] - s["cost_milli"], + "by_bucket": s["by_bucket"], "today": today, "upstream_failover": upstream_failover_stats()} @@ -271,6 +271,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)。 @@ -352,7 +361,7 @@ def _fallback_chain(pool, entry: Dict[str, Any], max_total: int = 3) -> List[Dic """ from gateway.model_pool import TIERS tier_rank = {t: i for i, t in enumerate(TIERS)} - rest = [e for e in pool.list().get("entries", []) + 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"))] @@ -383,7 +392,7 @@ def _downgrade_entry(pool, entry: Dict[str, Any], mode: str) -> Optional[Dict[st if target < 0: return None want = TIERS[target] - for e in pool.list().get("entries", []): + 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")): @@ -413,6 +422,38 @@ 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 = {} + + def _cached_charge(body: dict, cfg: ProxyConfig, model: str) -> Dict[str, int]: """缓存命中计费：成本 0，按未命中口径对入/出估 token 收售价（§7）。""" from gateway.proxy.pricing import compute @@ -423,14 +464,16 @@ 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 @@ -441,7 +484,7 @@ async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any], if not _entry_meets(entry, needs): from gateway.model_pool import filter_by_capabilities cap_entries = filter_by_capabilities( - [e for e in pool.list().get("entries", []) + [e for e in _pool_entries(pool) if e.get("enabled") and e.get("backend") not in ("mock",) and e.get("base_url")], need_vision=needs["vision"], need_tools=needs["tools"], diff --git a/gateway/sense/grader.py b/gateway/sense/grader.py index 00de8ba..165e25b 100644 --- a/gateway/sense/grader.py +++ b/gateway/sense/grader.py @@ -229,8 +229,11 @@ class Grader: executed_tier=executed) # ---- 决策留痕（T-X4：reasons / candidate_scores / rejected 三元结构）---- + # 2026-09 优化：同步 sqlite 写移入线程池（原直接跑在事件循环线程上， + # 高并发下阻塞整个网关；观察写本就有批量队列，决策留痕此处补齐同等保护） try: - self.store.insert_decision({ + import asyncio + row = { "ts": ts, "request_id": rid, "consumer": consumer, "decided_tier": tier, "executed_tier": executed, "mode": self.cfg.mode, "fallback": fallback, @@ -240,7 +243,8 @@ class Grader: "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 # 留痕失败不影响决策（观察与代理主链路优先）