84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
"""真实网关 /embeddings 端点探测(M2 设计 §11.6;人类默认口径: 实现时探测)。
|
|
|
|
对 .env 的 LLM 源网关发一次真实 embeddings 请求: 支持则记录向量证据,
|
|
不支持(404/翻译为领域错误)则 skip 并把响应记录进 tests/outputs/
|
|
(降级证据)。无 EMBED scope 配置时复用 LLM 源的 base_url/api_key。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
import os
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from dotenv import dotenv_values
|
|
|
|
from polygateway.errors import PolyGatewayError
|
|
from polygateway.transports.openai_compat import OpenAICompatTransport
|
|
from polygateway.types import SourceConfig
|
|
|
|
_ENV = {k: v for k, v in {**dotenv_values(".env"), **os.environ}.items() if v is not None}
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
"LLM__MINIMAX__1__BASE_URL" not in _ENV, reason="缺真实网关配置(.env)"
|
|
)
|
|
|
|
_OUT = Path("tests/outputs/embedding")
|
|
|
|
|
|
def _record(name: str, lines: list[str]) -> Path:
|
|
_OUT.mkdir(parents=True, exist_ok=True)
|
|
path = _OUT / f"{name}_{datetime.now():%Y%m%d_%H%M%S}.md"
|
|
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
return path
|
|
|
|
|
|
async def test_probe_real_gateway_embeddings():
|
|
source = SourceConfig(
|
|
name="probe_1",
|
|
provider="minimax",
|
|
base_url=_ENV["LLM__MINIMAX__1__BASE_URL"],
|
|
api_key=_ENV["LLM__MINIMAX__1__API_KEY"],
|
|
model=_ENV.get("PGW_EMBED_PROBE_MODEL", "text-embedding-v1"),
|
|
timeout_s=30.0,
|
|
est_tokens=8,
|
|
)
|
|
transport = OpenAICompatTransport()
|
|
try:
|
|
result = await transport.embed(
|
|
texts=["polygateway embedding probe"], source=source, call_id="probe"
|
|
)
|
|
except PolyGatewayError as exc:
|
|
path = _record(
|
|
"probe_unsupported",
|
|
[
|
|
"# Embedding 端点探测: 网关不支持",
|
|
f"- base_url: {source.base_url}",
|
|
f"- model: {source.model}",
|
|
f"- 错误分类: {type(exc).__name__}",
|
|
f"- status_code: {exc.status_code}",
|
|
f"- 详情: {exc}",
|
|
"",
|
|
"结论: e2e 按设计 §11.6 降级,embedding 行为由 unit 全覆盖。",
|
|
],
|
|
)
|
|
await transport.aclose()
|
|
pytest.skip(f"网关不支持 embeddings({type(exc).__name__}),证据: {path}")
|
|
else:
|
|
await transport.aclose()
|
|
assert result.dim > 0 and len(result.vectors) == 1
|
|
_record(
|
|
"probe_supported",
|
|
[
|
|
"# Embedding 端点探测: 网关支持",
|
|
f"- base_url: {source.base_url}",
|
|
f"- model: {source.model}",
|
|
f"- dim: {result.dim}",
|
|
f"- usage: {result.prompt_tokens}({result.usage_source})",
|
|
f"- 向量前 5 维: {result.vectors[0][:5]}",
|
|
f"- raw: {dataclasses.asdict(result)['raw']}",
|
|
],
|
|
)
|