iomgaa
195454d2e3
usage 帧缺失/非法时不再拿 est_tokens(最坏情形上界)当实测值,chat 与 embedding 两处兜底改记 0 并标 unavailable;打捞覆盖加 measured 前置条件, 避免 0/0 被洗成 estimated 而算出假的 0.0。embedding 全批合并扩三态(任一批 不可得 → 整体不可得),_total_cost 遇不可得批整体记 NULL。
395 lines
15 KiB
Python
395 lines
15 KiB
Python
"""Embedding 类型/端口/transport 测试(M2 设计 §7;T8)。
|
|
|
|
蓝本审计: GovDoc retrieval/embedding.py(分批/index 排序/维度校验)与
|
|
VT adapters/embedding.py(归一化);库裁决见设计 §7.3 表。
|
|
"""
|
|
|
|
import dataclasses
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from polygateway.errors import (
|
|
RequestRejectedError,
|
|
ResultInvalidError,
|
|
SourceDeadError,
|
|
TransientError,
|
|
)
|
|
from polygateway.ports import EmbeddingTransport
|
|
from polygateway.transports.openai_compat import OpenAICompatTransport
|
|
from polygateway.types import EmbeddingResponse, EmbeddingTransportResult, SourceConfig
|
|
|
|
|
|
def _src(**overrides):
|
|
base = {
|
|
"name": "e1",
|
|
"provider": "openai",
|
|
"base_url": "https://gw.example/v1",
|
|
"api_key": "sk",
|
|
"model": "embed-1",
|
|
"timeout_s": 10.0,
|
|
"est_tokens": 7,
|
|
}
|
|
base.update(overrides)
|
|
return SourceConfig(**base)
|
|
|
|
|
|
class TestTypes:
|
|
def test_embedding_response_frozen_with_defaults(self):
|
|
resp = EmbeddingResponse(
|
|
vectors=[[0.1, 0.2]],
|
|
dim=2,
|
|
model="m",
|
|
provider="p",
|
|
prompt_tokens=3,
|
|
usage_source="measured",
|
|
latency_ms=10,
|
|
call_id="c",
|
|
source_name="e1",
|
|
)
|
|
assert resp.cost is None
|
|
with pytest.raises(dataclasses.FrozenInstanceError):
|
|
resp.dim = 3
|
|
|
|
def test_transport_result_frozen(self):
|
|
r = EmbeddingTransportResult(
|
|
vectors=[[1.0]], dim=1, prompt_tokens=1, usage_source="measured", raw={}
|
|
)
|
|
with pytest.raises(dataclasses.FrozenInstanceError):
|
|
r.dim = 2
|
|
|
|
|
|
class _DummyEmbedTransport:
|
|
async def embed(self, *, texts, source, call_id):
|
|
raise NotImplementedError
|
|
|
|
|
|
def test_embedding_transport_protocol_runtime_checkable():
|
|
assert isinstance(_DummyEmbedTransport(), EmbeddingTransport)
|
|
assert isinstance(OpenAICompatTransport(), EmbeddingTransport)
|
|
|
|
|
|
def _transport_with(handler):
|
|
return OpenAICompatTransport(
|
|
client_factory=lambda source: httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
)
|
|
|
|
|
|
def _ok_body(vectors, *, usage=None, shuffle=False):
|
|
data = [{"index": i, "embedding": v} for i, v in enumerate(vectors)]
|
|
if shuffle:
|
|
data = list(reversed(data))
|
|
body = {"data": data}
|
|
if usage is not None:
|
|
body["usage"] = usage
|
|
return body
|
|
|
|
|
|
class TestEmbedTransport:
|
|
async def test_sorts_by_index_and_measures_usage(self):
|
|
def handler(request):
|
|
assert request.url.path.endswith("/embeddings")
|
|
payload = json.loads(request.content)
|
|
assert payload == {"model": "embed-1", "input": ["a", "b"]}
|
|
return httpx.Response(
|
|
200,
|
|
json=_ok_body([[1.0, 0.0], [0.0, 1.0]], usage={"prompt_tokens": 5}, shuffle=True),
|
|
)
|
|
|
|
result = await _transport_with(handler).embed(texts=["a", "b"], source=_src(), call_id="c")
|
|
assert result.vectors == [[1.0, 0.0], [0.0, 1.0]] # 乱序响应按 index 重排
|
|
assert result.dim == 2
|
|
assert result.prompt_tokens == 5 and result.usage_source == "measured"
|
|
|
|
async def test_missing_usage_is_unavailable(self):
|
|
"""usage 缺失不再退到 `est_tokens`(夹具填 7),与 chat 同口径记 0 + unavailable。"""
|
|
|
|
def handler(request):
|
|
return httpx.Response(200, json=_ok_body([[1.0]]))
|
|
|
|
result = await _transport_with(handler).embed(texts=["a"], source=_src(), call_id="c")
|
|
assert result.prompt_tokens == 0 and result.usage_source == "unavailable"
|
|
|
|
@pytest.mark.parametrize(
|
|
("status", "exc_type"),
|
|
[(401, SourceDeadError), (400, RequestRejectedError), (500, TransientError)],
|
|
)
|
|
async def test_http_errors_translate(self, status, exc_type):
|
|
def handler(request):
|
|
return httpx.Response(status, text="boom")
|
|
|
|
with pytest.raises(exc_type):
|
|
await _transport_with(handler).embed(texts=["a"], source=_src(), call_id="c")
|
|
|
|
async def test_network_error_is_transient(self):
|
|
def handler(request):
|
|
raise httpx.ConnectError("refused")
|
|
|
|
with pytest.raises(TransientError):
|
|
await _transport_with(handler).embed(texts=["a"], source=_src(), call_id="c")
|
|
|
|
@pytest.mark.parametrize(
|
|
"body",
|
|
[
|
|
{"data": []}, # 空 data
|
|
{"data": [{"index": 0, "embedding": [1.0]}]}, # 数量与输入不符(输入 2 条)
|
|
{
|
|
"data": [
|
|
{"index": 0, "embedding": [1.0, 2.0]},
|
|
{"index": 1, "embedding": [1.0]}, # 维度不一致
|
|
]
|
|
},
|
|
{"nope": True}, # 缺 data
|
|
],
|
|
)
|
|
async def test_malformed_payload_is_result_invalid(self, body):
|
|
def handler(request):
|
|
return httpx.Response(200, json=body)
|
|
|
|
with pytest.raises(ResultInvalidError):
|
|
await _transport_with(handler).embed(texts=["a", "b"], source=_src(), call_id="c")
|
|
|
|
async def test_empty_texts_rejected(self):
|
|
with pytest.raises(ValueError):
|
|
await _transport_with(lambda r: None).embed(texts=[], source=_src(), call_id="c")
|
|
|
|
|
|
# ═══════════ T9: EmbeddingClient 治理循环 ═══════════
|
|
|
|
import asyncio # noqa: E402
|
|
|
|
from polygateway.backends.memory.breaker import InMemoryGate # noqa: E402
|
|
from polygateway.backends.memory.limiter import InMemoryLimiter # noqa: E402
|
|
from polygateway.config import EmbeddingSettings # noqa: E402
|
|
from polygateway.embedding import EmbeddingClient # noqa: E402
|
|
from polygateway.pricing import ModelPrice, PricingTable # noqa: E402
|
|
from polygateway.sources import RoundRobinSelector # noqa: E402
|
|
from polygateway.types import ( # noqa: E402
|
|
BackpressurePolicy,
|
|
BreakerConfig,
|
|
GlobalLimits,
|
|
RetryPolicy,
|
|
)
|
|
|
|
_BREAKER = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
|
|
_NO_GLOBAL = GlobalLimits(max_concurrency=0, rpm=0, tpm=0)
|
|
|
|
|
|
def _vec_for(texts):
|
|
"""确定性向量: 每条 text 一个 [len(text)] 一维向量,便于断言保序。"""
|
|
return EmbeddingTransportResult(
|
|
vectors=[[float(len(t))] for t in texts],
|
|
dim=1,
|
|
prompt_tokens=len(texts),
|
|
usage_source="measured",
|
|
raw={},
|
|
)
|
|
|
|
|
|
class ScriptedEmbedTransport:
|
|
"""按脚本响应: 条目为 Exception / "ok"(按输入生成) / EmbeddingTransportResult / "hang"。"""
|
|
|
|
def __init__(self, script):
|
|
self.script = list(script)
|
|
self.calls = []
|
|
|
|
async def embed(self, *, texts, source, call_id):
|
|
self.calls.append((source.name, list(texts), call_id))
|
|
action = self.script.pop(0)
|
|
if isinstance(action, Exception):
|
|
raise action
|
|
if action == "hang":
|
|
await asyncio.Event().wait()
|
|
if action == "ok":
|
|
return _vec_for(texts)
|
|
return action
|
|
|
|
|
|
class _MemoryRecorder:
|
|
def __init__(self):
|
|
self.rows = []
|
|
|
|
async def record_llm_call(self, **fields):
|
|
self.rows.append(fields)
|
|
|
|
|
|
def _embed_client(sources, script, *, batch_size=2, telemetry=None, **overrides):
|
|
limiter = InMemoryLimiter(
|
|
scope="embed",
|
|
sources={s.name: s for s in sources},
|
|
global_limits=_NO_GLOBAL,
|
|
lease_ttl_s=100.0,
|
|
)
|
|
kwargs = {
|
|
"scope": "embed",
|
|
"sources": sources,
|
|
"selector": RoundRobinSelector(),
|
|
"limiter": limiter,
|
|
"breaker": InMemoryGate(config=_BREAKER),
|
|
"transport": ScriptedEmbedTransport(script),
|
|
"retry": RetryPolicy(max_attempts=3, backoff_base_s=0.001, backoff_max_s=0.01),
|
|
"backpressure": BackpressurePolicy(stall_window_s=300.0, poll_interval_s=0.001),
|
|
"batch_size": batch_size,
|
|
"telemetry": telemetry,
|
|
}
|
|
kwargs.update(overrides)
|
|
client = EmbeddingClient(**kwargs)
|
|
return client, limiter
|
|
|
|
|
|
class TestEmbedBatching:
|
|
async def test_batches_sequential_and_order_preserved(self):
|
|
texts = ["a", "bb", "ccc", "dddd", "eeeee"]
|
|
client, _ = _embed_client([_src()], ["ok", "ok", "ok"], batch_size=2)
|
|
resp = await client.embed(texts)
|
|
transport = client._transport
|
|
assert [len(batch) for _, batch, _ in transport.calls] == [2, 2, 1]
|
|
assert resp.vectors == [[1.0], [2.0], [3.0], [4.0], [5.0]] # 全批拼接保序
|
|
assert resp.prompt_tokens == 5 and resp.dim == 1
|
|
|
|
async def test_empty_input_short_circuits(self):
|
|
client, _ = _embed_client([_src()], [])
|
|
resp = await client.embed([])
|
|
assert resp.vectors == [] and resp.prompt_tokens == 0
|
|
assert client._transport.calls == []
|
|
|
|
async def test_usage_source_aggregates_conservatively(self):
|
|
estimated = EmbeddingTransportResult(
|
|
vectors=[[1.0], [1.0]], dim=1, prompt_tokens=9, usage_source="estimated", raw={}
|
|
)
|
|
client, _ = _embed_client([_src()], ["ok", estimated], batch_size=2)
|
|
resp = await client.embed(["a", "b", "c", "d"])
|
|
assert resp.usage_source == "estimated" # 任一批 estimated 则整体 estimated
|
|
assert resp.prompt_tokens == 2 + 9
|
|
|
|
async def test_unavailable_batch_dominates_and_voids_cost(self):
|
|
"""三态合并优先级(设计 §3.2 #10/#11): 任一批不可得 → 整体不可得且 cost NULL。
|
|
|
|
改前二值合并只看 `estimated`,measured+unavailable 会误标 measured;
|
|
`_total_cost` 逐批求和还会给出一个偏低却看似有效的金额。
|
|
"""
|
|
estimated = EmbeddingTransportResult(
|
|
vectors=[[1.0], [1.0]], dim=1, prompt_tokens=9, usage_source="estimated", raw={}
|
|
)
|
|
unavailable = EmbeddingTransportResult(
|
|
vectors=[[1.0], [1.0]], dim=1, prompt_tokens=0, usage_source="unavailable", raw={}
|
|
)
|
|
client, _ = _embed_client(
|
|
[_src()],
|
|
["ok", estimated, unavailable],
|
|
batch_size=2,
|
|
pricing=PricingTable({"embed-1": ModelPrice(input_per_1m=1.0, output_per_1m=0.0)}),
|
|
)
|
|
resp = await client.embed(["a", "b", "c", "d", "e", "f"])
|
|
assert resp.usage_source == "unavailable" # unavailable 压过 estimated 与 measured
|
|
assert resp.cost is None
|
|
|
|
|
|
class TestEmbedPostProcess:
|
|
async def test_normalize_l2(self):
|
|
raw = EmbeddingTransportResult(
|
|
vectors=[[3.0, 4.0]], dim=2, prompt_tokens=1, usage_source="measured", raw={}
|
|
)
|
|
client, _ = _embed_client([_src()], [raw], normalize=True)
|
|
resp = await client.embed(["x"])
|
|
assert resp.vectors[0] == pytest.approx([0.6, 0.8])
|
|
|
|
async def test_zero_vector_normalize_no_nan(self):
|
|
raw = EmbeddingTransportResult(
|
|
vectors=[[0.0, 0.0]], dim=2, prompt_tokens=1, usage_source="measured", raw={}
|
|
)
|
|
client, _ = _embed_client([_src()], [raw], normalize=True)
|
|
resp = await client.embed(["x"])
|
|
assert resp.vectors[0] == [0.0, 0.0] # max(norm, 1e-12) 防除零(VT 语义)
|
|
|
|
async def test_expected_dim_violation_is_result_invalid(self):
|
|
client, _ = _embed_client([_src()], ["ok"], expected_dim=768)
|
|
with pytest.raises(ResultInvalidError):
|
|
await client.embed(["x"])
|
|
|
|
|
|
class TestEmbedGovernance:
|
|
async def test_transient_retries_then_succeeds(self):
|
|
client, _ = _embed_client(
|
|
[_src()], [TransientError("boom", status_code=500), "ok"], batch_size=8
|
|
)
|
|
resp = await client.embed(["a", "b"])
|
|
assert resp.vectors == [[1.0], [1.0]]
|
|
assert len(client._transport.calls) == 2
|
|
|
|
async def test_source_dead_switches_source(self):
|
|
s1, s2 = _src(name="e1"), _src(name="e2")
|
|
client, _ = _embed_client(
|
|
[s1, s2], [SourceDeadError("401", status_code=401), "ok"], batch_size=8
|
|
)
|
|
await client.embed(["a"])
|
|
assert [name for name, _, _ in client._transport.calls] == ["e1", "e2"]
|
|
|
|
async def test_cancel_releases_permit(self):
|
|
client, limiter = _embed_client([_src(max_concurrency=1)], ["hang"])
|
|
task = asyncio.create_task(client.embed(["a"]))
|
|
while not (await limiter.source_stats("e1")).inflight:
|
|
await asyncio.sleep(0.01)
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
assert (await limiter.source_stats("e1")).inflight == 0
|
|
|
|
|
|
class TestEmbedTelemetry:
|
|
async def test_per_batch_rows_with_digest(self):
|
|
rec = _MemoryRecorder()
|
|
client, _ = _embed_client([_src()], ["ok", "ok"], batch_size=1, telemetry=rec)
|
|
await client.embed(["hello", "x" * 5000], session_id="sess", parent_call_id="pc")
|
|
assert len(rec.rows) == 2 # 每批一行
|
|
row = rec.rows[0]
|
|
assert row["session_id"] == "sess" and row["parent_call_id"] == "pc"
|
|
assert row["completion_tokens"] == 0
|
|
assert row["response"] == "<vectors n=1 dim=1>" # 向量绝不入库
|
|
assert len(rec.rows[1]["messages"]) < 1000 # 长文本截断后入库
|
|
|
|
|
|
class TestEmbeddingSettings:
|
|
_ENV = {
|
|
"EMBED__QWEN__1__BASE_URL": "https://gw.example/v1",
|
|
"EMBED__QWEN__1__API_KEY": "sk-a",
|
|
"EMBED__QWEN__1__MODEL": "text-embedding-v3",
|
|
"EMBED__QWEN__1__TIMEOUT_S": "60",
|
|
"EMBED__RETRY__MAX_ATTEMPTS": "3",
|
|
"EMBED__RETRY__BACKOFF_BASE_S": "1.0",
|
|
"EMBED__RETRY__BACKOFF_MAX_S": "10.0",
|
|
"EMBED__BREAKER__FAIL_THRESHOLD": "5",
|
|
"EMBED__BREAKER__COOLDOWN_S": "60",
|
|
"PGW_CACHE_BACKEND": "none",
|
|
"PGW_TELEMETRY_BACKEND": "none",
|
|
"EMBED__BATCH_SIZE": "64",
|
|
}
|
|
|
|
def test_loads_scope_and_batch(self):
|
|
s = EmbeddingSettings.from_env("EMBED", env=self._ENV)
|
|
assert s.gateway.sources[0].model == "text-embedding-v3"
|
|
assert s.batch_size == 64 and s.normalize is False and s.expected_dim is None
|
|
|
|
def test_batch_size_required_and_positive(self):
|
|
env = {k: v for k, v in self._ENV.items() if k != "EMBED__BATCH_SIZE"}
|
|
with pytest.raises(ValueError, match="BATCH_SIZE"):
|
|
EmbeddingSettings.from_env("EMBED", env=env)
|
|
with pytest.raises(ValueError, match="BATCH_SIZE"):
|
|
EmbeddingSettings.from_env("EMBED", env={**self._ENV, "EMBED__BATCH_SIZE": "0"})
|
|
|
|
def test_optional_normalize_and_dim(self):
|
|
env = {**self._ENV, "EMBED__NORMALIZE": "true", "EMBED__EXPECTED_DIM": "768"}
|
|
s = EmbeddingSettings.from_env("EMBED", env=env)
|
|
assert s.normalize is True and s.expected_dim == 768
|
|
|
|
def test_expected_dim_must_be_positive(self):
|
|
"""env 层的检查保留是为了报错能点出键名(构造期那道点的是字段名)。"""
|
|
with pytest.raises(ValueError, match="EXPECTED_DIM"):
|
|
EmbeddingSettings.from_env("EMBED", env={**self._ENV, "EMBED__EXPECTED_DIM": "0"})
|
|
|
|
def test_from_settings_assembles_client(self):
|
|
s = EmbeddingSettings.from_env("EMBED", env=self._ENV)
|
|
client = EmbeddingClient.from_settings(s)
|
|
assert isinstance(client, EmbeddingClient)
|