From f783878c02fb250f311ab8d125a231ba4ea3dc4c Mon Sep 17 00:00:00 2001 From: tzt <14718231+flying-travel@user.noreply.gitee.com> Date: Sat, 5 Sep 2026 13:33:58 +0800 Subject: [PATCH] =?UTF-8?q?feat(sense):=20T-G1=20Embedder=EF=BC=88int8=20?= =?UTF-8?q?=E9=87=8F=E5=8C=96/=E9=99=8D=E7=BA=A7=20fail-closed=20+=20/v1/e?= =?UTF-8?q?mbeddings=20=E9=80=8F=E4=BC=A0=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - embedder.py：embed() 调 llama-server /v1/embeddings -> 对称 per-vector int8 量化（scale=127/max|v|，1024 维余弦扰动 ~1e-4 << 1e-2 验收）； 超时/连接/非200/空形状 -> EmbedderDown（D-G4 fail-closed）； quantize_int8/cosine_int8 辅助（int8 直接点积，scale 正标量不改方向） - routes.py 拆双路由组：/sense 前缀组 + /v1 无前缀组（/v1/embeddings OpenAI 兼容透传，embedder 不可用 503）；build_sense_routers 返回列表， api.py 逐个 include；build_sense_router 保留向后兼容 - 测试 +6：量化余弦误差 200 组 <1e-2/范围/零向量/int8 余弦/ 正常量化/三种 fail-closed/透传端点（503+形状+400），全量 381 passed - 待真机项：llama-server embedder 端点冒烟（需用户配置 embedder.base_url） --- gateway/sense/__init__.py | 20 +++-- gateway/sense/embedder.py | 84 ++++++++++++++++++++ gateway/sense/routes.py | 49 +++++++++--- tests/test_sense_embedder.py | 148 +++++++++++++++++++++++++++++++++++ 任务拆解与执行计划.md | 2 +- 5 files changed, 286 insertions(+), 17 deletions(-) create mode 100644 gateway/sense/embedder.py create mode 100644 tests/test_sense_embedder.py diff --git a/gateway/sense/__init__.py b/gateway/sense/__init__.py index c3185e2..9f1eeaa 100644 --- a/gateway/sense/__init__.py +++ b/gateway/sense/__init__.py @@ -1,17 +1,23 @@ -"""语义分析器与三级任务分级（T-G0…T-G8）：/v1/route + /sense/admin/*。 +"""语义分析器与三级任务分级（T-G0…T-G8）：/v1/route + /v1/embeddings + /sense/admin/*。 -组装唯一入口：build_sense_router(cfg) -> APIRouter； -由 gateway.api 在 sense.enabled 时 include（D-G7 关闭时不注册任何路由）。 -灰度：mode=collect（只记不决策）-> shadow（决策只比对）-> live（真分流）； -collect 攒满 min_labels 前不得开 live（晋升门 §6）。 +组装唯一入口：build_sense_routers(cfg) -> [APIRouter, APIRouter] +（/sense 前缀组 + /v1 无前缀组）；由 gateway.api 在 sense.enabled 时逐个 include +（D-G7 关闭时不注册任何路由）。 """ from __future__ import annotations +from typing import List + from fastapi import APIRouter from gateway.sense.config import SenseConfig, build_sense_config # noqa: F401 -def build_sense_router(cfg: SenseConfig) -> APIRouter: - from gateway.sense.routes import build_sense_router as _build +def build_sense_routers(cfg: SenseConfig) -> List[APIRouter]: + from gateway.sense.routes import build_sense_routers as _build return _build(cfg) + + +def build_sense_router(cfg: SenseConfig) -> APIRouter: + """向后兼容：仅 /sense 前缀组（/v1 组见 build_sense_routers）。""" + return build_sense_routers(cfg)[0] diff --git a/gateway/sense/embedder.py b/gateway/sense/embedder.py new file mode 100644 index 0000000..3f9ff43 --- /dev/null +++ b/gateway/sense/embedder.py @@ -0,0 +1,84 @@ +"""Embedder 客户端（T-G1）：llama-server /v1/embeddings + int8 量化 + 降级（D-G4）。 + +- int8 量化：对称 per-vector scale（scale = max|v| / 127），q = round(v × scale)。 + 1024 维余弦扰动 ~1e-4（验收 <1e-2）；存储 1024 维 fp32 4KB -> int8 1KB（D-G6）。 +- 降级（D-G4）：超时/连接失败/非 200/形状不符 -> raise EmbedderDown， + 调用方（grader）退化为规则门 + 默认 T2。 +""" +from __future__ import annotations + +import httpx + +from gateway.sense.config import EmbedderCfg +from gateway.sense.errors import EmbedderDown + +_client: Optional[httpx.AsyncClient] = None + + +def get_client() -> httpx.AsyncClient: + global _client + if _client is None: + _client = httpx.AsyncClient(timeout=5.0) + return _client + + +async def close_client() -> None: + global _client + if _client is not None: + await _client.aclose() + _client = None + + +def quantize_int8(vec: list[float]) -> list[int]: + """对称 per-vector int8 量化：q = round(v × scale)，scale = 127 / max|v|。""" + m = max((abs(v) for v in vec), default=0.0) + if m == 0: + return [0] * len(vec) + scale = 127.0 / m + return [max(-127, min(127, int(round(v * scale)))) for v in vec] + + +def dequantize_int8(q: list[int], scale: float) -> list[float]: + """反量化（需要保留 scale 时使用；检索场景直接用 int8 余弦即可）。""" + if scale == 0: + return [0.0] * len(q) + return [x / scale for x in q] + + +def cosine_int8(a: list[int], b: list[int]) -> float: + """int8 向量余弦相似度（无 scale 版本：scale 为正标量不改变方向）。""" + n = min(len(a), len(b)) + if n == 0: + return 0.0 + dot = na = nb = 0 + for i in range(n): + x, y = a[i], b[i] + dot += x * y + na += x * x + nb += y * y + if na == 0 or nb == 0: + return 0.0 + return max(-1.0, min(1.0, dot / ((na ** 0.5) * (nb ** 0.5)))) + + +async def embed(text: str, cfg: EmbedderCfg) -> list[int]: + """调用 llama-server /v1/embeddings，返回 int8 量化向量（§7 签名）。 + + 失败（超时/连接/非 200/形状不符）-> raise EmbedderDown（D-G4 降级信号）。 + """ + try: + client = get_client() + resp = await client.post( + cfg.base_url.rstrip("/") + "/embeddings", + json={"model": cfg.model, "input": text}, + timeout=cfg.timeout_s) + resp.raise_for_status() + data = resp.json() + vec = data["data"][0]["embedding"] + if not isinstance(vec, list) or not vec: + raise ValueError("空向量") + return quantize_int8([float(v) for v in vec]) + except EmbedderDown: + raise + except Exception as e: # noqa: BLE001 + raise EmbedderDown(f"embedder 不可用: {type(e).__name__}: {e}") from e diff --git a/gateway/sense/routes.py b/gateway/sense/routes.py index d6da685..692c2f5 100644 --- a/gateway/sense/routes.py +++ b/gateway/sense/routes.py @@ -1,21 +1,52 @@ -"""Sense 路由（T-G0 骨架：健康检查；/v1/route 与 /v1/embeddings 在 T-G1/G5 落地）。""" +"""Sense 路由（T-G1）：/sense 前缀组 + /v1 无前缀组（embeddings 透传）。 + +/v1/route（T-G5 grader）后续加入无前缀组；/sense/admin/*（T-G3+）加入前缀组。 +""" from __future__ import annotations -from fastapi import APIRouter +from typing import List + +from fastapi import APIRouter, Request +from fastapi.responses import JSONResponse from gateway.sense.config import SenseConfig +from gateway.sense.embedder import embed +from gateway.sense.errors import EmbedderDown -def build_sense_router(cfg: SenseConfig) -> APIRouter: - """组装 sense 面路由（唯一组装点）。""" - router = APIRouter(prefix="/sense") +def build_sense_routers(cfg: SenseConfig) -> List[APIRouter]: + """组装 sense 路由组：[/sense 前缀组, /v1 无前缀组]。""" + main = APIRouter(prefix="/sense") + v1 = APIRouter() - @router.get("/health", tags=["sense"]) + @main.get("/health", tags=["sense"]) async def health(): """sense 面健康检查（含灰度状态，供看板/运维）。""" return {"enabled": cfg.enabled, "mode": cfg.mode, "embedder": cfg.embedder.base_url} - # /v1/route（T-G5 grader）、/v1/embeddings（T-G1）、/sense/admin/*（T-G3+）后续追加。 - # 注意 /v1/route 不带 /sense 前缀——由 api.py 单独 include 一个无前缀子路由（T-G5）。 - return router + @v1.post("/v1/embeddings", tags=["sense"]) + async def embeddings(request: Request): + """OpenAI 兼容透传 embedder（客户端/代理共用）。 + + 请求：{"model"?, "input": str}；响应：{"object":"list","data":[{"object": + "embedding","index":0,"embedding":[int8...]}],"model":...}。 + embedder 不可用 -> 503（D-G4 fail-closed，客户端可感知降级）。 + """ + try: + body = await request.json() + except Exception: + body = {} + text = str((body or {}).get("input") or "").strip() + if not text: + return JSONResponse({"error": {"message": "input 必填"}}, status_code=400) + model = str((body or {}).get("model") or cfg.embedder.model) + try: + vec = await embed(text, cfg.embedder) + except EmbedderDown as e: + return JSONResponse( + {"error": {"message": str(e), "type": "embedder_down"}}, status_code=503) + return {"object": "list", "model": model, + "data": [{"object": "embedding", "index": 0, "embedding": vec}]} + + return [main, v1] diff --git a/tests/test_sense_embedder.py b/tests/test_sense_embedder.py new file mode 100644 index 0000000..e001d5b --- /dev/null +++ b/tests/test_sense_embedder.py @@ -0,0 +1,148 @@ +"""Embedder 测试（T-G1）：int8 量化余弦误差 / 降级 fail-closed / 透传端点。""" +import asyncio +import json +import math +import random + +import httpx +import pytest + +pytest.importorskip("fastapi") + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from gateway.sense.config import EmbedderCfg, build_sense_config +from gateway.sense.embedder import cosine_int8, embed, quantize_int8 +from gateway.sense.errors import EmbedderDown + + +def test_quantize_roundtrip_cosine_error_small(): + """200 组 1024 维随机向量：量化前后余弦误差 < 1e-2（§11 验收）。""" + random.seed(42) + worst = 0.0 + for _ in range(200): + v = [random.gauss(0, 1) for _ in range(1024)] + q = quantize_int8(v) + scale = (max(abs(x) for x in v) and 127.0 / max(abs(x) for x in v)) or 1.0 + dv = [x / scale for x in q] + err = 1.0 - cosine_int8(q, quantize_int8(v)) # 自反性恒 1；此行检对称 + # 真误差：原向量 vs 反量化向量 + dot = sum(a * b for a, b in zip(v, dv)) + na = math.sqrt(sum(a * a for a in v)) + nb = math.sqrt(sum(b * b for b in dv)) + err = 1.0 - dot / (na * nb) + worst = max(worst, abs(err)) + assert worst < 1e-2 + + +def test_quantize_range_and_zero_vector(): + q = quantize_int8([3.0, -6.0, 0.0]) + assert all(-127 <= x <= 127 for x in q) + assert quantize_int8([0.0] * 4) == [0, 0, 0, 0] # 零向量不崩 + + +def test_cosine_int8_identical_and_orthogonal(): + assert cosine_int8([3, 4], [6, 8]) == pytest.approx(1.0) + assert cosine_int8([1, 0], [0, 1]) == 0.0 + assert cosine_int8([], []) == 0.0 + + +def _cfg(base_url="http://emb"): + return EmbedderCfg(base_url=base_url, model="bge-m3", timeout_s=1.0, dim=8) + + +def test_embed_ok_returns_int8(monkeypatch): + vec = [0.5 * i for i in range(1, 9)] + body = {"data": [{"embedding": vec}]} + + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path == "/embeddings" + payload = json.loads(request.read()) + assert payload["model"] == "bge-m3" and payload["input"] == "hello" + return httpx.Response(200, content=json.dumps(body).encode()) + + import gateway.sense.embedder as em + orig = em._client + em._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + try: + q = asyncio_run(embed("hello", _cfg())) + assert len(q) == 8 and all(-127 <= x <= 127 for x in q) + finally: + em._client = orig + + +def test_embed_down_fail_closed(monkeypatch): + """连接拒绝/非 200/超时 -> EmbedderDown（D-G4 fail-closed）。""" + from gateway.sense import embedder as em + + def handler_500(request): + return httpx.Response(500) + + orig = em._client + em._client = httpx.AsyncClient(transport=httpx.MockTransport(handler_500)) + try: + with pytest.raises(EmbedderDown): + asyncio_run(embed("x", _cfg())) + finally: + em._client = orig + + def handler_timeout(request): + raise httpx.ConnectTimeout("超时") + + em._client = httpx.AsyncClient(transport=httpx.MockTransport(handler_timeout)) + try: + with pytest.raises(EmbedderDown): + asyncio_run(embed("x", _cfg())) + finally: + em._client = orig + + def handler_bad_shape(request): + return httpx.Response(200, content=json.dumps({"data": [{"embedding": []}]}).encode()) + + em._client = httpx.AsyncClient(transport=httpx.MockTransport(handler_bad_shape)) + try: + with pytest.raises(EmbedderDown): + asyncio_run(embed("x", _cfg())) + finally: + em._client = orig + + +def test_v1_embeddings_endpoint(tmp_path): + """enabled 独立挂载：/v1/embeddings 透传（Mock embedder 注入）。""" + tc_app = FastAPI() + cfg = build_sense_config({"sense": {"enabled": True, + "db_path": str(tmp_path / "s.sqlite3")}}) + from gateway.sense.routes import build_sense_routers + app = FastAPI() + for r in build_sense_routers(cfg): + app.include_router(r) + client = TestClient(app) + + # embedder 不可用 -> 503 fail-closed + r = client.post("/v1/embeddings", json={"input": "hello"}) + assert r.status_code == 503 + + # 注入可用 embedder（monkeypatch 内部函数） + import gateway.sense.routes as sr + orig_embed = sr.embed + + async def fake_embed(text, c): + return [1, 2, 3] + + sr.embed = fake_embed + try: + r2 = client.post("/v1/embeddings", json={"input": "hello"}) + assert r2.status_code == 200 + data = r2.json() + assert data["object"] == "list" + assert data["data"][0]["embedding"] == [1, 2, 3] + # 空 input -> 400 + r3 = client.post("/v1/embeddings", json={"input": ""}) + assert r3.status_code == 400 + finally: + sr.embed = orig_embed + + +def asyncio_run(coro): + return asyncio.run(coro) diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md index d5c8f79..55deb26 100644 --- a/任务拆解与执行计划.md +++ b/任务拆解与执行计划.md @@ -155,7 +155,7 @@ P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯 | T | 内容 | 状态 | commit | |---|------|------|--------| | T-G0 | 骨架：gateway/sense/ 包 + sense.sqlite3 DDL + enabled 门控 | ✅ 完成 | c3efa70 | -| T-G1 | Embedder：/v1/embeddings 客户端 + int8 量化 + 降级阶梯 | ⬜ 待办 | | +| T-G1 | Embedder：/v1/embeddings 客户端 + int8 量化 + 降级阶梯 | ✅ 完成 | T-G1 | | T-G2 | 观察埋点：observer + 三消费方埋点（pipeline/proxy/client） | ⬜ 待办 | | | T-G3 | 标签+校准：夜间 true_tier 推导 + split-conformal + 工件表 | ⬜ 待办 | | | T-G3b | KnnHead（架构变体 B）：kNN 投票 + conformal-kNN + 按桶分区/封顶/压缩；hybrid fusion 预留（§14） | ⬜ 待办 | |