diff --git a/gateway/api.py b/gateway/api.py index e810170..63d2c6a 100644 --- a/gateway/api.py +++ b/gateway/api.py @@ -201,6 +201,17 @@ try: # Vue SPA 静态资源(html=True:对不存在的路径 fallback 到 index.html,支持 SPA 路由) app.mount("/static", StaticFiles(directory=str(_STATIC_DIR), html=True), name="static") + # 校园 AI 代理层(T-P0;D-P7:proxy.enabled=False 时不注册任何 /proxy 路由, + # 行为与现状逐字节一致——含本 if 判断在内对既有路径零影响) + try: + from gateway.proxy import build_proxy_router + 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())) + except Exception as _pe: # pragma: no cover - 代理层装配失败不拖垮主应用 + print(f"[gateway] 代理层未启用({_pe})") + @app.get("/", response_class=HTMLResponse, tags=["ui"]) async def index(): """Vue SPA 的 index.html(FastAPI API 路由优先,此处仅作 fallback)。""" diff --git a/gateway/proxy/__init__.py b/gateway/proxy/__init__.py new file mode 100644 index 0000000..1759235 --- /dev/null +++ b/gateway/proxy/__init__.py @@ -0,0 +1,17 @@ +"""校园 AI 代理层(T-P0…T-P8):/proxy/v1/* 学生面 + /proxy/admin/* 管理面。 + +组装唯一入口:build_proxy_router(cfg, pool) -> APIRouter; +由 gateway.api 在 proxy.enabled 时 include(D-P7:关闭时不注册任何路由)。 +""" +from __future__ import annotations + +from fastapi import APIRouter + +from gateway.proxy.config import ProxyConfig, build_proxy_config # noqa: F401 + +__all__ = ["build_proxy_router", "build_proxy_config", "ProxyConfig"] + + +def build_proxy_router(cfg: ProxyConfig, pool) -> APIRouter: + from gateway.proxy.routes import build_proxy_router as _build + return _build(cfg, pool) diff --git a/gateway/proxy/auth.py b/gateway/proxy/auth.py new file mode 100644 index 0000000..626a5f2 --- /dev/null +++ b/gateway/proxy/auth.py @@ -0,0 +1,22 @@ +"""鉴权与限流(T-P1 落地;本文件先立签名)。""" +from __future__ import annotations + +from typing import Any, Dict, Optional + + +def issue_key(ledger, student_id: int, rpm_cap: Optional[int] = None, + day_cap_req: Optional[int] = None) -> Dict[str, Any]: + """签发学生代理 key:明文只返回一次(T-P1 实现)。""" + raise NotImplementedError("T-P1") + + +def authenticate(authorization: str, ledger, limits, now: float) -> Dict[str, Any]: + """校验 Bearer key -> 学生上下文(T-P1 实现;带热路径缓存)。""" + raise NotImplementedError("T-P1") + + +class RateLimiter: + """令牌桶 rpm + per-key 并发信号量(T-P1 实现)。""" + + def allow(self, key_id: int, rpm_cap: int) -> bool: + raise NotImplementedError("T-P1") diff --git a/gateway/proxy/config.py b/gateway/proxy/config.py new file mode 100644 index 0000000..9868490 --- /dev/null +++ b/gateway/proxy/config.py @@ -0,0 +1,164 @@ +"""ProxyConfig:从 settings 的 proxy 段构建代理层配置(T-P0)。 + +锁定决策落地: +- D-P1 货币:账本毫元整数。配置里价格是浮点"元/1M tokens"(人类可读), + **加载期一次性转为整数毫元/1M tokens**,运行期纯整数运算(毫元漂移只可能 + 发生在这一次换算,黄金用例锁死)。 +- 售价折扣采用差异化 {in: 0.5, out: 0.8}(设计文档 §2.5 结论:统一 5 折 + 盈亏平衡 h0≈32% 结构性危险;执行版 §4 示例与之冲突,按设计文档取差异化, + 纯配置可热改)。 +- D-P7:enabled=False 时不注册任何 /proxy 路由。 +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Tuple + +# 元 -> 毫元换算系数(D-P1:1 元 = 1000 毫元) +YUAN_TO_MILLI = 1000 + + +@dataclass +class ModelPrice: + """单一模型的整数价格表(毫元 / 1M tokens)。""" + in_miss: int + in_hit: int + out: int + + +@dataclass +class BucketCfg: + """一个课程桶的配置。""" + name: str + system_template: str = "你是校园学习助手。" + doc_prefix_file: Optional[str] = None + doc_version: int = 1 + ttl_hours: int = 72 + + +@dataclass +class ProxyConfig: + """代理层运行配置(全部为加载期定型值)。""" + enabled: bool = False + admin_key: str = "" + db_path: str = "data/proxy.sqlite3" + buckets: Dict[str, BucketCfg] = field(default_factory=dict) + model_prices: Dict[str, ModelPrice] = field(default_factory=dict) + peak_start: str = "08:30" + peak_end: str = "23:59" + offpeak_factor: float = 0.5 + sale_in: float = 0.5 + sale_out: float = 0.8 + rpm_per_key: int = 10 + day_req_cap: int = 200 + concurrent_per_key: int = 2 + max_body_chars: int = 60000 + semcache_enabled: bool = True + sim_threshold: float = 0.92 + max_entries: int = 300000 + promote_frequency: int = 5 + + def bucket(self, name: str) -> BucketCfg: + """取桶配置;未知名回落 default(D-P2 缺省桶)。""" + if name in self.buckets: + return self.buckets[name] + return self.buckets.get("default") or BucketCfg(name="default") + + def price(self, model: str) -> Optional[ModelPrice]: + """取模型价格表;未配置返回 None(调用方应拒绝或用 default 档)。""" + return self.model_prices.get(model) + + +def _to_milli_per_m(yuan_per_m: float) -> int: + """元/1M -> 毫元/1M(四舍五入取整,加载期唯一换算点)。""" + return int(round(float(yuan_per_m) * YUAN_TO_MILLI)) + + +def _parse_hhmm(text: str, fallback: str) -> str: + """校验 HH:MM 形态,非法回落默认。""" + t = str(text or fallback).strip() + parts = t.split(":") + if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit(): + h, m = int(parts[0]), int(parts[1]) + if 0 <= h <= 23 and 0 <= m <= 59: + return f"{h:02d}:{m:02d}" + return fallback + + +def build_proxy_config(settings_dict: Dict[str, Any]) -> ProxyConfig: + """从 settings.to_dict() 的 proxy 段构建 ProxyConfig(缺省值兜底)。 + + settings_dict 为整个设置 dict(含 worker/architect/... 与 proxy 段)。 + """ + raw = settings_dict.get("proxy") or {} + if not isinstance(raw, dict): + raw = {} + + # ---- 桶 ---- + buckets: Dict[str, BucketCfg] = {} + raw_buckets = raw.get("buckets") or {} + if isinstance(raw_buckets, dict): + for name, b in raw_buckets.items(): + if not isinstance(b, dict): + continue + buckets[str(name)] = BucketCfg( + name=str(name), + system_template=str(b.get("system_template") or "你是校园学习助手。"), + doc_prefix_file=(str(b["doc_prefix_file"]) + if b.get("doc_prefix_file") else None), + doc_version=int(b.get("doc_version", 1) or 1), + ttl_hours=int(b.get("ttl_hours", 72) or 72), + ) + if "default" not in buckets: + buckets["default"] = BucketCfg(name="default") + + # ---- 价格表:元/1M 浮点 -> 毫元/1M 整数(D-P1 加载期换算) ---- + model_prices: Dict[str, ModelPrice] = {} + raw_pricing = raw.get("pricing") or {} + if isinstance(raw_pricing, dict): + for model, p in raw_pricing.items(): + if not isinstance(p, dict) or "in_miss" not in p: + continue # peak_window/offpeak_factor/sale_discount 非模型键 + in_miss = _to_milli_per_m(p.get("in_miss", 0)) + in_hit = _to_milli_per_m(p.get("in_hit", 0)) + out = _to_milli_per_m(p.get("out", 0)) + if in_miss <= 0: + continue + if in_hit <= 0: + in_hit = max(1, in_miss // 30) # 缺省 = in_miss × 1/30(D-P3) + model_prices[str(model)] = ModelPrice( + in_miss=in_miss, in_hit=in_hit, out=out) + + # ---- 峰谷窗口 / 系数 ---- + pw = raw_pricing.get("peak_window") or {} if isinstance(raw_pricing, dict) else {} + off = float(raw_pricing.get("offpeak_factor", 0.5)) if isinstance(raw_pricing, dict) else 0.5 + sale = raw_pricing.get("sale_discount") or {} if isinstance(raw_pricing, dict) else {} + + # ---- 限流 / 缓存 ---- + limits = raw.get("limits") or {} + if not isinstance(limits, dict): + limits = {} + sem = raw.get("semcache") or {} + if not isinstance(sem, dict): + sem = {} + + return ProxyConfig( + enabled=bool(raw.get("enabled", False)), + admin_key=str(raw.get("admin_key") or ""), + db_path=str(raw.get("db_path") or "data/proxy.sqlite3"), + buckets=buckets, + model_prices=model_prices, + peak_start=_parse_hhmm(pw.get("start"), "08:30") if isinstance(pw, dict) else "08:30", + peak_end=_parse_hhmm(pw.get("end"), "23:59") if isinstance(pw, dict) else "23:59", + offpeak_factor=max(0.0, min(1.0, off)), + sale_in=max(0.0, min(1.0, float(sale.get("in", 0.5)))), + sale_out=max(0.0, min(1.0, float(sale.get("out", 0.8)))), + 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), + 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), + ) diff --git a/gateway/proxy/errors.py b/gateway/proxy/errors.py new file mode 100644 index 0000000..87d74b7 --- /dev/null +++ b/gateway/proxy/errors.py @@ -0,0 +1,50 @@ +"""代理层错误类型(§5.3 错误码映射)。""" +from __future__ import annotations + + +class ProxyError(Exception): + """代理层错误基类(HTTP 语义由子类表达)。""" + status_code = 500 + code = "proxy_error" + + +class ProxyAuthError(ProxyError): + """无效/注销 key。""" + status_code = 401 + code = "invalid_key" + + +class BalanceError(ProxyError): + """余额或日上限不足(预扣失败)。""" + status_code = 402 + code = "insufficient_balance" + + +class SuspendedError(ProxyError): + """学生账户被停用。""" + status_code = 403 + code = "student_suspended" + + +class QuotaError(ProxyError): + """限流(rpm / 日请求上限 / 并发上限)。""" + status_code = 429 + code = "rate_limited" + + +class BodyTooLargeError(ProxyError): + """请求体超过 max_body_chars。""" + status_code = 413 + code = "body_too_large" + + +class UpstreamError(ProxyError): + """上游失败(首 token 前 failover 均失败)。""" + status_code = 502 + code = "upstream_failed" + + +class AdminAuthError(ProxyError): + """管理面鉴权失败。""" + status_code = 401 + code = "admin_unauthorized" diff --git a/gateway/proxy/ledger.py b/gateway/proxy/ledger.py new file mode 100644 index 0000000..6cb68f0 --- /dev/null +++ b/gateway/proxy/ledger.py @@ -0,0 +1,92 @@ +"""账本(T-P0:DDL 初始化 + 连接纪律;CRUD/预扣在 T-P1 落地)。 + +工程纪律: +- D-P10:sqlite3 是同步库,所有 DB 调用必须经 asyncio.to_thread(由调用方 + routes 层包装;本模块保持同步实现,可测试性好)。连接 check_same_thread=False + + threading.Lock 串行化(沿用 ReviewQueue 模式:每操作新连接 + 全局锁)。 +- WAL 模式(§3)。 +- D-P11:预扣用原子 UPDATE,见 try_hold(T-P1 实现)。 +""" +from __future__ import annotations + +import sqlite3 +import threading +from pathlib import Path +from typing import Any, Dict, List, Optional + +# 四表 + 两索引(§3,一字不改) +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS students( + id INTEGER PRIMARY KEY, name TEXT NOT NULL, class TEXT DEFAULT '', + status TEXT NOT NULL DEFAULT 'active', + balance_milli INTEGER NOT NULL DEFAULT 0, + daily_cap_milli INTEGER NOT NULL DEFAULT 5000, + spent_today_milli INTEGER NOT NULL DEFAULT 0, spent_date TEXT DEFAULT ''); + +CREATE TABLE IF NOT EXISTS proxy_keys( + id INTEGER PRIMARY KEY, key_hash TEXT UNIQUE NOT NULL, key_prefix TEXT NOT NULL, + student_id INTEGER NOT NULL REFERENCES students(id), + created_ts INTEGER NOT NULL, revoked INTEGER NOT NULL DEFAULT 0, + rpm_cap INTEGER NOT NULL DEFAULT 10, day_cap_req INTEGER NOT NULL DEFAULT 200, + req_today INTEGER NOT NULL DEFAULT 0, req_date TEXT DEFAULT ''); + +CREATE TABLE IF NOT EXISTS usage_ledger( + request_id TEXT PRIMARY KEY, ts INTEGER NOT NULL, key_id INTEGER NOT NULL, + model TEXT NOT NULL, bucket TEXT NOT NULL DEFAULT 'default', + in_miss_tok INTEGER NOT NULL DEFAULT 0, in_hit_tok INTEGER NOT NULL DEFAULT 0, + out_tok INTEGER NOT NULL DEFAULT 0, gateway_cached INTEGER NOT NULL DEFAULT 0, + upstream_cost_milli INTEGER NOT NULL DEFAULT 0, charged_milli INTEGER NOT NULL DEFAULT 0, + margin_milli INTEGER NOT NULL DEFAULT 0, ttfb_ms INTEGER, total_ms INTEGER, + status TEXT NOT NULL); + +CREATE TABLE IF NOT EXISTS semcache( + cache_key TEXT PRIMARY KEY, + bucket TEXT NOT NULL, q_norm TEXT NOT NULL, answer TEXT NOT NULL, model TEXT NOT NULL, + created_ts INTEGER NOT NULL, ttl_ts INTEGER NOT NULL, + doc_version INTEGER NOT NULL DEFAULT 1, hits INTEGER NOT NULL DEFAULT 0); +CREATE INDEX IF NOT EXISTS idx_semcache_bucket ON semcache(bucket, ttl_ts); +CREATE INDEX IF NOT EXISTS idx_ledger_ts ON usage_ledger(ts); +""" + +_TABLES = ("students", "proxy_keys", "usage_ledger", "semcache") + + +class Ledger: + """代理层账本(sqlite WAL;同步实现,调用方负责 to_thread)。""" + + def __init__(self, db_path: str | Path): + self.db_path = Path(db_path) + self.db_path.parent.mkdir(parents=True, exist_ok=True) + self._lock = threading.Lock() + self._init_db() + + # ---------- 初始化 ---------- + @classmethod + def init_db(cls, db_path: str | Path) -> "Ledger": + """工厂(§6 签名):建库建表(幂等)。""" + return cls(db_path) + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.db_path, check_same_thread=False) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + return conn + + def _init_db(self) -> None: + with self._lock, self._connect() as conn: + conn.executescript(_SCHEMA) + + # ---------- 自省(T-P0 验收用) ---------- + def table_names(self) -> List[str]: + """列出已建表名(测试验收)。""" + with self._lock, self._connect() as conn: + rows = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'").fetchall() + return [r["name"] for r in rows] + + def index_names(self) -> List[str]: + with self._lock, self._connect() as conn: + rows = conn.execute( + "SELECT name FROM sqlite_master WHERE type='index' AND name LIKE 'idx_%'" + ).fetchall() + return [r["name"] for r in rows] diff --git a/gateway/proxy/normalizer.py b/gateway/proxy/normalizer.py new file mode 100644 index 0000000..6b963ff --- /dev/null +++ b/gateway/proxy/normalizer.py @@ -0,0 +1,19 @@ +"""请求规范化与桶解析(T-P5 落地;本文件先立签名)。""" +from __future__ import annotations + +from typing import Any, Dict + + +def resolve_bucket(body: dict, headers: dict, cfg) -> Any: + """解析课程桶:X-Campus-Bucket 头 > model->bucket 映射 > default(T-P5 实现)。""" + raise NotImplementedError("T-P5") + + +def canonical_hash(bucket: str, doc_version: int, body: dict) -> str: + """规范化哈希:稳定序列化 -> sha256(五规则见 §6;T-P5 实现)。""" + raise NotImplementedError("T-P5") + + +def shape(body: dict, bucket_cfg) -> dict: + """整形上游请求体:[canonical_system] -> [doc_prefix] -> [原 messages](T-P5 实现)。""" + raise NotImplementedError("T-P5") diff --git a/gateway/proxy/pricing.py b/gateway/proxy/pricing.py new file mode 100644 index 0000000..a3da201 --- /dev/null +++ b/gateway/proxy/pricing.py @@ -0,0 +1,17 @@ +"""计价(T-P3 落地;本文件先立签名)。 + +D-P1:compute 输入输出全为整数毫元;配置换算在 config 加载期完成。 +""" +from __future__ import annotations + +from typing import Any, Dict + + +def is_offpeak(ts: float, peak_start: str, peak_end: str) -> bool: + """峰谷窗口判定(本地时区 HH:MM;T-P3 实现)。""" + raise NotImplementedError("T-P3") + + +def compute(usage: Dict[str, int], model: str, ts: float, cfg) -> Dict[str, int]: + """按 usage 计费 -> {upstream_cost_milli, charged_milli, margin_milli}(T-P3 实现)。""" + raise NotImplementedError("T-P3") diff --git a/gateway/proxy/routes.py b/gateway/proxy/routes.py new file mode 100644 index 0000000..ac73e68 --- /dev/null +++ b/gateway/proxy/routes.py @@ -0,0 +1,28 @@ +"""代理面路由(T-P0 骨架:/proxy/v1/models;主对话路由 T-P4 落地)。""" +from __future__ import annotations + +from fastapi import APIRouter, Request + +from gateway.proxy.config import ProxyConfig + + +def build_proxy_router(cfg: ProxyConfig, pool) -> APIRouter: + """组装代理面路由(唯一组装点;pool 为 gateway.model_pool.PoolStore)。""" + router = APIRouter(prefix="/proxy") + + @router.get("/v1/models", tags=["proxy"]) + async def list_models(): + """学生面:池内允许代理的模型列表(OpenAI /models 形状)。""" + entries = pool.list().get("entries", []) + names = [e["model"] for e in entries + if e.get("enabled") and e.get("backend") not in ("mock",)] + # 去重保序 + seen, data = set(), [] + for n in names: + if n not in seen: + seen.add(n) + data.append({"id": n, "object": "model", "owned_by": "campus-proxy"}) + return {"object": "list", "data": data} + + # /v1/chat/completions 与 /admin/* 在 T-P1/T-P4 追加 + return router diff --git a/gateway/proxy/semcache.py b/gateway/proxy/semcache.py new file mode 100644 index 0000000..00fd821 --- /dev/null +++ b/gateway/proxy/semcache.py @@ -0,0 +1,24 @@ +"""语义缓存(T-P6 落地;本文件先立签名)。 + +规格(§6):L1 精确 + L2 字符 2/3-gram 倒排(启动自 sqlite 重建), +加权 Jaccard(3-gram 权 2)+ 共享 gram>=3 门限 + 阈值 0.92, +TTL + LRU(max_entries),L2 命中 promote_frequency 次晋升 L1。 +""" +from __future__ import annotations + +from typing import Any, Dict, Optional + + +class SemanticCache: + """两级语义缓存(T-P6 实现)。""" + + def lookup(self, bucket: str, doc_version: int, norm_hash: str, + norm_text: str, now: float) -> Optional[Dict[str, Any]]: + raise NotImplementedError("T-P6") + + def put(self, bucket: str, doc_version: int, norm_hash: str, + norm_text: str, answer: str, model: str, now: float) -> None: + raise NotImplementedError("T-P6") + + def stats(self) -> Dict[str, Any]: + raise NotImplementedError("T-P6") diff --git a/gateway/proxy/upstream.py b/gateway/proxy/upstream.py new file mode 100644 index 0000000..c6370cc --- /dev/null +++ b/gateway/proxy/upstream.py @@ -0,0 +1,21 @@ +"""上游客户端(T-P2 落地;本文件先立签名)。 + +规格(§6):模块级 httpx.AsyncClient 单例(keepalive,max_connections=100), +超时 connect=10s/read=120s/write=10s/pool=30s;始终注入 +stream_options.include_usage(客户端未要求 usage 时过滤该 chunk 不下发); +首 token 前 failover(D-P4)。 +""" +from __future__ import annotations + +from typing import Any, AsyncIterator, Dict + + +async def stream(body: dict, entry: Dict[str, Any], usage_sink) -> AsyncIterator[bytes]: + """流式派发到上游,逐块 yield(T-P2 实现)。""" + raise NotImplementedError("T-P2") + yield b"" # pragma: no cover + + +def normalize_usage(provider: str, usage_dict: Dict[str, Any]) -> Dict[str, int]: + """三家 usage 字段 -> 统一 {in_miss, in_hit, out}(T-P2 实现)。""" + raise NotImplementedError("T-P2") diff --git a/gateway/settings.py b/gateway/settings.py index 889e8d6..cd177e9 100644 --- a/gateway/settings.py +++ b/gateway/settings.py @@ -48,6 +48,26 @@ DEFAULTS: Dict[str, Any] = { "approval_policy": "dangerous", # 审批策略:off | dangerous(写/编辑/命令询问)| all "approval_timeout_s": 120, # 审批等待超时(超时自动拒绝) }, + # 校园 AI 代理层(T-P0;结构见《实施方案_代理层与缓存层.md》§4) + "proxy": { + "enabled": False, # D-P7:默认关,开启需显式配置(重启生效) + "admin_key": "", # 管理面密钥;空 = 仅 loopback 放行 + "db_path": "data/proxy.sqlite3", + "buckets": { # 桶配置(嵌套对象原样存储) + "default": {"system_template": "你是校园学习助手。", + "doc_prefix_file": None, "doc_version": 1, "ttl_hours": 72}, + }, + "pricing": { # 元/1M tokens(加载期转毫元整数,D-P1) + "deepseek-chat": {"in_miss": 3.0, "in_hit": 0.1, "out": 9.0}, + "peak_window": {"start": "08:30", "end": "23:59"}, + "offpeak_factor": 0.5, + "sale_discount": {"in": 0.5, "out": 0.8}, # 差异化(设计文档 §2.5) + }, + "limits": {"rpm_per_key": 10, "day_req_cap": 200, + "concurrent_per_key": 2, "max_body_chars": 60000}, + "semcache": {"enabled": True, "sim_threshold": 0.92, + "max_entries": 300000, "promote_frequency": 5}, + }, } diff --git a/scripts/serve.py b/scripts/serve.py index d9ff253..c3f26c7 100644 --- a/scripts/serve.py +++ b/scripts/serve.py @@ -19,6 +19,13 @@ def start(port: int): if old: print(f"已有服务运行 (pid={old}),先执行 --stop 再启动。") return + # D-P9:代理层(singleflight/令牌桶/内存 LRU)为进程内状态,多 worker 会静默失效 + for var in ("WEB_CONCURRENCY", "UVICORN_WORKERS"): + v = (os.environ.get(var) or "").strip() + if v.isdigit() and int(v) > 1: + print(f"拒绝启动:{var}={v}。代理层要求单进程(workers=1)," + "扩容路径见《实施方案_代理层与缓存层.md》D-P9。") + return out = open(ROOT / "_gateway.out.log", "ab", buffering=0) err = open(ROOT / "_gateway.err.log", "ab", buffering=0) flags = 0x00000008 | 0x08000000 # DETACHED_PROCESS | CREATE_NO_WINDOW diff --git a/tests/test_proxy_routes.py b/tests/test_proxy_routes.py new file mode 100644 index 0000000..fb30328 --- /dev/null +++ b/tests/test_proxy_routes.py @@ -0,0 +1,87 @@ +"""代理层骨架测试(T-P0):enabled 门控 / DDL / 独立挂载的 models 端点。""" +import pytest + +pytest.importorskip("fastapi") + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +import gateway.api as ga +from gateway.proxy.config import ProxyConfig, build_proxy_config +from gateway.proxy.ledger import Ledger + + +def test_proxy_disabled_by_default_404(): + """D-P7:默认 enabled=False -> 全局 app 不注册任何 /proxy 路由。""" + client = TestClient(ga.app) + assert client.get("/proxy/v1/models").status_code == 404 + assert client.post("/proxy/v1/chat/completions", json={}).status_code == 404 + assert client.get("/proxy/admin/stats").status_code == 404 + + +def test_ddl_creates_tables_and_indexes(tmp_path): + """四表 + 两索引幂等创建(WAL)。""" + led = Ledger.init_db(tmp_path / "proxy.sqlite3") + tables = set(led.table_names()) + for t in ("students", "proxy_keys", "usage_ledger", "semcache"): + assert t in tables + idx = set(led.index_names()) + assert "idx_semcache_bucket" in idx + assert "idx_ledger_ts" in idx + # 幂等重建 + led2 = Ledger.init_db(tmp_path / "proxy.sqlite3") + assert "semcache" in set(led2.table_names()) + + +def test_build_proxy_config_defaults_and_milli_conversion(): + """缺省值兜底 + 元/1M -> 毫元/1M 加载期换算(D-P1)。""" + from gateway.settings import DEFAULTS + cfg = build_proxy_config(DEFAULTS) # DEFAULTS 自带 deepseek-chat 价格示例 + assert cfg.enabled is False + assert "default" in cfg.buckets + assert cfg.sale_in == 0.5 and cfg.sale_out == 0.8 # 差异化定价 + # 自带 DEFAULTS 的 deepseek-chat 价格:3.0 元 -> 3000 毫元 + p = cfg.price("deepseek-chat") + assert p is not None + assert p.in_miss == 3000 + assert p.in_hit == 100 # 0.1 元 -> 100 毫元 + assert p.out == 9000 + + # in_hit 缺省 = in_miss / 30(D-P3) + cfg2 = build_proxy_config({"proxy": { + "pricing": {"some-model": {"in_miss": 3.0, "out": 9.0}}}}) + p2 = cfg2.price("some-model") + assert p2.in_hit == 100 # 3000 // 30 + + # 桶解析回落 default + assert cfg.bucket("ghost").name == "default" + + +def test_enabled_router_lists_pool_models(tmp_path, monkeypatch): + """enabled=True 独立挂载:/proxy/v1/models 返回池内非 mock 模型(OpenAI 形状)。""" + import gateway.model_pool as mp + from gateway.model_pool import PoolStore + mp.reset_pool() + monkeypatch.setattr(mp, "_store", PoolStore(path=tmp_path / "pool.json")) + mp.get_pool().upsert({ + "id": "up1", "name": "云端", "tier": "budget", "backend": "openai", + "base_url": "https://api.example.com", "model": "deepseek-chat", + "price_in": 0.1, "price_out": 0.1, "enabled": True}) + mp.get_pool().upsert({ + "id": "mk", "name": "模拟", "tier": "local", "backend": "mock", + "model": "mock", "enabled": True}) # mock 不应出现 + + cfg = build_proxy_config({"proxy": {"enabled": True, + "db_path": str(tmp_path / "p.sqlite3")}}) + app = FastAPI() + app.include_router(__import__("gateway.proxy", fromlist=["x"]).build_proxy_router( + cfg, mp.get_pool())) + client = TestClient(app) + r = client.get("/proxy/v1/models") + assert r.status_code == 200 + data = r.json() + assert data["object"] == "list" + ids = [m["id"] for m in data["data"]] + assert ids == ["deepseek-chat"] + assert data["data"][0]["owned_by"] == "campus-proxy" + mp.reset_pool() diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md index 79b4b8e..428b12a 100644 --- a/任务拆解与执行计划.md +++ b/任务拆解与执行计划.md @@ -135,7 +135,7 @@ P0 完成后的能力:干净的后端抽象 + 可量化的评测 + 可追溯 | T | 内容 | 状态 | commit | |---|------|------|--------| -| T-P0 | 骨架:gateway/proxy/ 包 + SQLite DDL + enabled 门控挂路由 | ⬜ 待办 | | +| T-P0 | 骨架:gateway/proxy/ 包 + SQLite DDL + enabled 门控挂路由 | ✅ 完成 | T-P0 | | T-P1 | 鉴权+账本:key 签发/令牌桶/四表/request_id 幂等/日限额 | ⬜ 待办 | | | T-P2 | 上游客户端:流式派发+三家 usage 归一化+首 token 前 failover | ⬜ 待办 | | | T-P3 | 计价+结算:峰谷窗口/毫元整数/黄金用例 ≥10 组 | ⬜ 待办 | |