tzt/projectAIpopular
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(proxy): T-X2 上游有序降级链 + failover 三元组如实上报（采纳 cortiq tier 链）

Browse Source 
- 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）
This commit is contained in:
tzt
2026-09-18 22:35:53 +08:00
parent aa7cb0704c
commit a043493548
5 changed files with 230 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/proxy/routes.py
+42 -9
View File
@@ -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})
gateway/proxy/upstream.py
+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}")
tests/test_proxy_budget.py
+10 -10
View File
@@ -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})
tests/test_proxy_upstream_chain.py
+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()
任务拆解与执行计划.md
+1
View File
@@ -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 |