feat(proxy): T-P6 缓存分支接线(routes 主时序,M2 完整闭环)

- _run_chat 开头缓存分支:cacheable(stop+单轮)-> canonical_hash ->
  L1/L2 查 -> 命中:try_hold->立即 settle(cost=0, charged=售价口径,
  status=cached, gateway_cached=1) -> SSE 合成回放或 JSON(X-Cache: HIT)
- 未命中流程收尾写缓存(_json_response 尾部,stop 且单轮);
  缓存层任何故障降级直连上游(不影响可用性)
- 缓存命中计费 _cached_charge(成本 0,入按字符估收售价——全毛利杠杆 L0)
- _get_semcache 按 db_path 分实例(测试隔离)+ reset_semcache_instances
- 测试 +1(端到端:首问 miss 打上游/二问 HIT 上游仅 1 次/账目 cached),
  全量 423 passed
This commit is contained in:
tzt
2026-09-05 15:56:17 +08:00
parent e41471c39c
commit c54ad23c88
3 changed files with 145 additions and 6 deletions
+109 -4
View File
@@ -217,6 +217,37 @@ def _resolve_entry(pool, model: str, cfg: ProxyConfig) -> Optional[Dict[str, Any
return None return None
_semcache_instances: Dict[str, Any] = {}
def _get_semcache(cfg: ProxyConfig, ledger):
"""按 db_path 的进程内缓存单例(D-P9 单进程前提;不同库隔离)。"""
inst = _semcache_instances.get(cfg.db_path)
if inst is None:
from gateway.proxy.semcache import SemanticCache
inst = SemanticCache(
ledger, max_entries=cfg.max_entries,
sim_threshold=cfg.sim_threshold,
promote_frequency=cfg.promote_frequency)
_semcache_instances[cfg.db_path] = inst
return inst
def reset_semcache_instances() -> None:
"""测试用:清空缓存单例。"""
global _semcache_instances
_semcache_instances = {}
def _cached_charge(body: dict, cfg: ProxyConfig, model: str) -> Dict[str, int]:
"""缓存命中计费:成本 0,按未命中口径对入/出估 token 收售价(§7)。"""
from gateway.proxy.pricing import compute
usage = {"in_miss": len(json.dumps(body.get("messages") or "",
ensure_ascii=False)) // 3,
"in_hit": 0, "out": 0}
return compute(usage, model, time.time(), cfg)
async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any], async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any],
cfg: ProxyConfig, ledger, pool): cfg: ProxyConfig, ledger, pool):
model = str(body.get("model") or "") model = str(body.get("model") or "")
@@ -231,6 +262,61 @@ async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any],
if isinstance(body.get("stream_options"), dict) else False if isinstance(body.get("stream_options"), dict) else False
is_stream = bool(body.get("stream")) is_stream = bool(body.get("stream"))
# ---- 缓存分支(T-P6,§7 时序):仅缓存准入(stop+单轮)查询 ----
cacheable = False
cache = None
resolution = None
try:
if cfg.semcache_enabled:
from gateway.proxy.normalizer import canonical_hash, is_cacheable
from gateway.proxy.semcache import SemanticCache
bucket_cfg = cfg.bucket(str(headers.get("x-campus-bucket") or "default"))
cacheable = is_cacheable(body)
if cacheable:
norm_hash = canonical_hash(bucket_cfg.name, bucket_cfg.doc_version, body)
cache = _get_semcache(cfg, ledger)
norm_text = json.dumps(body.get("messages") or [], ensure_ascii=False,
sort_keys=True)
hit = cache.lookup(norm_hash, norm_text, bucket_cfg.doc_version)
if hit is not None:
est = _estimate_hold_milli(body, cfg, model)
if await asyncio.to_thread(
ledger.try_hold, request_id, ctx["key_id"],
ctx["student_id"], model, bucket_cfg.name, est, ts):
br = _cached_charge(body, cfg, model)
await asyncio.to_thread(
ledger.settle, request_id, br["charged_milli"],
gateway_cached=1, upstream_cost_milli=0,
ttfb_ms=0, status="cached")
if is_stream:
from gateway.proxy.semcache import synth_sse_chunks
chunks = synth_sse_chunks(hit["answer"], model=model,
request_id=request_id)
async def replay():
for c in chunks:
yield c
return StreamingResponse(
replay(), media_type="text/event-stream",
headers={"Cache-Control": "no-cache",
"X-Cache": "HIT",
"X-Request-Id": request_id})
return JSONResponse({
"id": f"chatcmpl-{request_id}", "object": "chat.completion",
"created": int(time.time()), "model": model,
"choices": [{"index": 0,
"message": {"role": "assistant",
"content": hit["answer"]},
"finish_reason": "stop"}],
"usage": {"prompt_tokens": 0, "completion_tokens": 0,
"total_tokens": 0},
}, headers={"X-Cache": "HIT", "X-Request-Id": request_id})
raise BalanceError("余额或当日额度不足")
except BalanceError:
raise
except Exception:
cache = None # 缓存层故障不影响主流程(降级直连上游)
est = _estimate_hold_milli(body, cfg, model) est = _estimate_hold_milli(body, cfg, model)
if not await asyncio.to_thread( if not await asyncio.to_thread(
ledger.try_hold, request_id, ctx["key_id"], ctx["student_id"], ledger.try_hold, request_id, ctx["key_id"], ctx["student_id"],
@@ -242,9 +328,12 @@ async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any],
try: try:
if is_stream: if is_stream:
return await _stream_response(body, entry, sink, headers, client_wants_usage, return await _stream_response(body, entry, sink, headers, client_wants_usage,
request_id, ctx, cfg, ledger, model, est, t0) request_id, ctx, cfg, ledger, model, est, t0,
cache=cache, cacheable=cacheable)
return await _json_response(body, entry, sink, request_id, ctx, cfg, return await _json_response(body, entry, sink, request_id, ctx, cfg,
ledger, model, est, t0) ledger, model, est, t0,
cache=cache, cacheable=cacheable,
headers=headers)
except UpstreamAborted as e: except UpstreamAborted as e:
# 流中失败:按已收 usage 结算(无 usage 按字符估算),不缓存(D-P4) # 流中失败:按已收 usage 结算(无 usage 按字符估算),不缓存(D-P4)
usage = sink.get("usage") or _estimate_usage_from_sink(sink) usage = sink.get("usage") or _estimate_usage_from_sink(sink)
@@ -272,7 +361,8 @@ def _estimate_usage_from_sink(sink: Dict[str, Any]) -> Dict[str, int]:
async def _stream_response(body, entry, sink, headers, client_wants_usage, async def _stream_response(body, entry, sink, headers, client_wants_usage,
request_id, ctx, cfg, ledger, model, est, t0): request_id, ctx, cfg, ledger, model, est, t0,
cache=None, cacheable=False):
usage = {"in_miss": 0, "in_hit": 0, "out": 0} usage = {"in_miss": 0, "in_hit": 0, "out": 0}
async def gen(): async def gen():
@@ -323,7 +413,8 @@ async def _stream_response(body, entry, sink, headers, client_wants_usage,
async def _json_response(body, entry, sink, request_id, ctx, cfg, ledger, async def _json_response(body, entry, sink, request_id, ctx, cfg, ledger,
model, est, t0): model, est, t0, cache=None, cacheable=False,
headers=None):
parts = [] parts = []
async for raw_bytes in upstream_stream(body, entry, sink, [entry]): async for raw_bytes in upstream_stream(body, entry, sink, [entry]):
line = raw_bytes.decode("utf-8").strip() line = raw_bytes.decode("utf-8").strip()
@@ -353,6 +444,20 @@ async def _json_response(body, entry, sink, request_id, ctx, cfg, ledger,
in_miss_tok=usage.get("in_miss", 0), in_hit_tok=usage.get("in_hit", 0), in_miss_tok=usage.get("in_miss", 0), in_hit_tok=usage.get("in_hit", 0),
out_tok=usage.get("out", 0), upstream_cost_milli=br["upstream_cost_milli"], out_tok=usage.get("out", 0), upstream_cost_milli=br["upstream_cost_milli"],
ttfb_ms=sink.get("ttfb_ms"), total_ms=total_ms, status="ok") ttfb_ms=sink.get("ttfb_ms"), total_ms=total_ms, status="ok")
# 缓存准入(D-P5):stop 且单轮且未命中来的 -> 写缓存
if cache is not None and cacheable and sink.get("finish_reason", "stop") == "stop":
try:
from gateway.proxy.normalizer import canonical_hash
bucket_cfg = cfg.bucket(str(headers.get("x-campus-bucket")
or "default")) if headers else cfg.bucket("default")
ckey = canonical_hash(bucket_cfg.name, bucket_cfg.doc_version, body)
norm_text = json.dumps(body.get("messages") or [], ensure_ascii=False,
sort_keys=True)
cache.put(ckey, norm_text, "".join(parts), model,
doc_version=bucket_cfg.doc_version,
ttl_hours=bucket_cfg.ttl_hours)
except Exception:
pass
return JSONResponse({ return JSONResponse({
"id": f"chatcmpl-{request_id}", "id": f"chatcmpl-{request_id}",
"object": "chat.completion", "object": "chat.completion",
+35 -1
View File
@@ -100,7 +100,10 @@ def _auth(key):
return {"Authorization": f"Bearer {key['key']}"} return {"Authorization": f"Bearer {key['key']}"}
BODY = {"model": "deepseek-chat", "messages": [{"role": "user", "content": "问个问题"}]} BODY = {"model": "deepseek-chat",
"messages": [{"role": "user",
"content": "请详细介绍快速排序算法的原理、复杂度与实现要点,"
"并给出 Python 示例代码与适用场景分析。"}]}
def test_chat_non_stream_end_to_end(tmp_path): def test_chat_non_stream_end_to_end(tmp_path):
@@ -288,3 +291,34 @@ def test_openai_protocol_compliance_via_httpx(tmp_path):
pass pass
mp.reset_pool() mp.reset_pool()
reset_auth_state() reset_auth_state()
def test_cache_hit_second_request(tmp_path):
"""T-P6 端到端:首问打上游并写缓存;同问再答 X-Cache: HIT 且上游仅 1 次。"""
from gateway.proxy.routes import reset_semcache_instances
reset_semcache_instances()
tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
try:
r1 = tc.post("/proxy/v1/chat/completions", json=BODY, headers=_auth(key))
assert r1.status_code == 200
assert r1.headers.get("x-cache") is None # 首问未命中
assert calls["n"] == 1
r2 = tc.post("/proxy/v1/chat/completions", json=BODY, headers=_auth(key))
assert r2.status_code == 200
assert r2.headers.get("x-cache") == "HIT" # 缓存命中
assert calls["n"] == 1 # 上游仍只 1 次
assert r2.json()["choices"][0]["message"]["content"] == "你好,世界"
rid = r2.headers["x-request-id"]
u = ledger.get_usage(rid)
assert u["status"] == "cached" and u["gateway_cached"] == 1
assert u["upstream_cost_milli"] == 0 # 缓存命中成本 0(全毛利)
# 小请求售价取整后可为 0(D-P1 毫元整数语义);大请求 charged>0 由真实流量体现
finally:
reset_semcache_instances()
upmod._client = orig
try:
asyncio_run(hclient.aclose())
except Exception:
pass
mp.reset_pool()
reset_auth_state()
+1 -1
View File
@@ -141,7 +141,7 @@ P0 完成后的能力:干净的后端抽象 + 可量化的评测 + 可追溯
| T-P3 | 计价+结算:峰谷窗口/毫元整数/黄金用例 ≥10 组 | ✅ 完成 | T-P3 | | T-P3 | 计价+结算:峰谷窗口/毫元整数/黄金用例 ≥10 组 | ✅ 完成 | T-P3 |
| T-P4 | 路由端到端:/proxy/v1 非流式+流式+402/429/413 语义 | ✅ 完成 | T-P4 | | T-P4 | 路由端到端:/proxy/v1 非流式+流式+402/429/413 语义 | ✅ 完成 | T-P4 |
| T-P5 | 规范化+桶:稳定哈希/整形顺序/doc_version 失效/多轮不缓存 | ✅ 完成 | T-P5 | | T-P5 | 规范化+桶:稳定哈希/整形顺序/doc_version 失效/多轮不缓存 | ✅ 完成 | T-P5 |
| T-P6 | 语义缓存:L1 精确+L2 n-gram 倒排+singleflight+TTL+失败不缓存 | ✅ 完成 | T-P6 | | T-P6 | 语义缓存:L1 精确+L2 n-gram 倒排+singleflight+TTL+失败不缓存 | ✅ 完成(含 routes 接线) | T-P6 |
| T-P7 | 管理面+前端:stats/ledger 端点+ProxyView 三卡片 | ⬜ 待办 | | | T-P7 | 管理面+前端:stats/ledger 端点+ProxyView 三卡片 | ⬜ 待办 | |
| T-P8 | 压测+预算:200 并发流式;P99≤50ms/内存≤1GB/账目零不一致 | ⬜ 待办 | | | T-P8 | 压测+预算:200 并发流式;P99≤50ms/内存≤1GB/账目零不一致 | ⬜ 待办 | |