diff --git a/gateway/proxy/routes.py b/gateway/proxy/routes.py index 0c6600d..6b95d61 100644 --- a/gateway/proxy/routes.py +++ b/gateway/proxy/routes.py @@ -30,7 +30,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: @@ -204,7 +205,8 @@ def build_proxy_router(cfg: ProxyConfig, pool) -> APIRouter: 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} + "today": today, + "upstream_failover": upstream_failover_stats()} @router.get("/admin/ledger", tags=["proxy-admin"]) async def admin_ledger(request: Request, student_id: int = 0, @@ -272,6 +274,24 @@ def _budget_headers(budget_mode: str) -> Dict[str, str]: return {"X-Budget-Mode": budget_mode} if budget_mode and budget_mode != "normal" else {} +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.list().get("entries", []) + 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):在池内找恰好低一档/最低档的启用条目。 @@ -427,16 +447,18 @@ 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) 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, - budget_mode=budget_mode) + 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, budget_mode=budget_mode) + headers=headers, budget_mode=budget_mode, + chain=chain) except UpstreamAborted as e: # 流中失败:按已收 usage 结算(无 usage 按字符估算),不缓存(D-P4) usage = sink.get("usage") or _estimate_usage_from_sink(sink) @@ -465,7 +487,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, budget_mode: str = "normal"): + 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(): @@ -473,7 +496,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 @@ -518,9 +542,11 @@ async def _stream_response(body, entry, sink, headers, client_wants_usage, async def _json_response(body, entry, sink, request_id, ctx, cfg, ledger, model, est, t0, cache=None, cacheable=False, - headers=None, budget_mode: str = "normal"): + headers=None, budget_mode: str = "normal", + chain: Optional[List[Dict[str, Any]]] = None): 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 @@ -562,6 +588,13 @@ async def _json_response(body, entry, sink, request_id, ctx, cfg, ledger, ttl_hours=bucket_cfg.ttl_hours) 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", @@ -576,4 +609,4 @@ async def _json_response(body, entry, sink, request_id, ctx, cfg, ledger, "total_tokens": usage.get("in_miss", 0) + usage.get("in_hit", 0) + usage.get("out", 0)}, }, headers={"X-Request-Id": request_id, - **_budget_headers(budget_mode)}) + **_budget_headers(budget_mode), **fb_headers}) diff --git a/gateway/proxy/upstream.py b/gateway/proxy/upstream.py index 3ce91f3..c1898e0 100644 --- a/gateway/proxy/upstream.py +++ b/gateway/proxy/upstream.py @@ -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}") diff --git a/tests/test_proxy_budget.py b/tests/test_proxy_budget.py index 72fd278..3d40d46 100644 --- a/tests/test_proxy_budget.py +++ b/tests/test_proxy_budget.py @@ -77,9 +77,9 @@ def test_budget_mode_missing_student_normal(tmp_path): assert led.budget_mode(999, 100, ts=1789874000.0) == "normal" -def _make_pool() -> PoolStore: +def _make_pool(tmp_path) -> PoolStore: import gateway.model_pool as mp - store = PoolStore() # 不落盘(path=None 仅内存) + 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", @@ -93,30 +93,30 @@ def _make_pool() -> PoolStore: return store -def test_downgrade_optimize_one_tier(): - pool = _make_pool() +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(): - pool = _make_pool() +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(): +def test_downgrade_stops_at_local(tmp_path): """已在最低档:cheap/optimize 均不再降(杜绝反向升档)。""" - pool = _make_pool() + 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(): - pool = _make_pool() +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}) diff --git a/tests/test_proxy_upstream_chain.py b/tests/test_proxy_upstream_chain.py new file mode 100644 index 0000000..fb8951b --- /dev/null +++ b/tests/test_proxy_upstream_chain.py @@ -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() diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md index 5be550b..ef666a6 100644 --- a/任务拆解与执行计划.md +++ b/任务拆解与执行计划.md @@ -166,3 +166,4 @@ P0 完成后的能力:干净的后端抽象 + 可量化的评测 + 可追溯 | 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 |