diff --git a/gateway/proxy/routes.py b/gateway/proxy/routes.py index e97c864..a24136a 100644 --- a/gateway/proxy/routes.py +++ b/gateway/proxy/routes.py @@ -217,6 +217,37 @@ def _resolve_entry(pool, model: str, cfg: ProxyConfig) -> Optional[Dict[str, Any 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], cfg: ProxyConfig, ledger, pool): 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 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) if not await asyncio.to_thread( 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: if is_stream: 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, - ledger, model, est, t0) + ledger, model, est, t0, + cache=cache, cacheable=cacheable, + headers=headers) except UpstreamAborted as e: # 流中失败:按已收 usage 结算(无 usage 按字符估算),不缓存(D-P4) 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, - 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} 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, - model, est, t0): + model, est, t0, cache=None, cacheable=False, + headers=None): parts = [] async for raw_bytes in upstream_stream(body, entry, sink, [entry]): 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), 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") + # 缓存准入(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({ "id": f"chatcmpl-{request_id}", "object": "chat.completion", diff --git a/tests/test_proxy_routes_e2e.py b/tests/test_proxy_routes_e2e.py index bbc058c..af4643a 100644 --- a/tests/test_proxy_routes_e2e.py +++ b/tests/test_proxy_routes_e2e.py @@ -100,7 +100,10 @@ def _auth(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): @@ -288,3 +291,34 @@ def test_openai_protocol_compliance_via_httpx(tmp_path): pass mp.reset_pool() 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() diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md index 3e47a88..f1bc678 100644 --- a/任务拆解与执行计划.md +++ b/任务拆解与执行计划.md @@ -141,7 +141,7 @@ P0 完成后的能力:干净的后端抽象 + 可量化的评测 + 可追溯 | T-P3 | 计价+结算:峰谷窗口/毫元整数/黄金用例 ≥10 组 | ✅ 完成 | T-P3 | | T-P4 | 路由端到端:/proxy/v1 非流式+流式+402/429/413 语义 | ✅ 完成 | T-P4 | | 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-P8 | 压测+预算:200 并发流式;P99≤50ms/内存≤1GB/账目零不一致 | ⬜ 待办 | |