张胜宇
e239eb778b
1) 客服 Agent 四份交付文档 + 构建脚手架：品牌由包装占位 XX科技 / 旧名 南方财富 统一为南方基金（热线 400-889-8899 / 官网 nffund.com），系统名改为「智能服务系统」； 同步追加 §0.4 修订记录行，工程记录行保留原占位字面以支撑硬编码扫描验收。 2) 开发文档：清理 28 份已作废/残留文档（14 份移出归档 + 14 份仓库副本）， 新增《文档规整方案与开发前待决事项-2026-09-17》。 3) 客服agent 四份交付文档首次纳入本分支。
141 lines
5.0 KiB
Python
141 lines
5.0 KiB
Python
"""模型向量化（embedding）能力的契约测试。
|
|
|
|
与文本生成同族：OpenAI-compatible `/embeddings`、密钥只从 `secret_ref` 解析、
|
|
端点按声明顺序尝试、降级时显式标记，任何结构违约都失败关闭。
|
|
"""
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.core.errors import RecoverableAgentError
|
|
from app.service.model_gateway import (
|
|
ModelDispatchService,
|
|
ModelEmbeddingService,
|
|
OpenAICompatibleGateway,
|
|
)
|
|
|
|
|
|
class Endpoint:
|
|
def __init__(self, code: str, *, timeout_ms: int = 1000) -> None:
|
|
self.endpoint_code = code
|
|
self.base_url = "https://model.example/v1"
|
|
self.model_name = "embed-small"
|
|
self.secret_ref = "env:MODEL_TEST_KEY"
|
|
self.timeout_ms = timeout_ms
|
|
|
|
|
|
class SeedResolver:
|
|
"""替身密钥解析器：断言只接受 env: 引用，返回值固定。"""
|
|
|
|
def resolve(self, secret_ref: str) -> str:
|
|
assert secret_ref.startswith("env:")
|
|
return "test-key"
|
|
|
|
|
|
class RecordingGateway:
|
|
"""替身网关：按端点记录调用并返回预设结果或抛错。"""
|
|
|
|
def __init__(self, behaviour: dict[str, Any]) -> None:
|
|
self.behaviour = behaviour
|
|
self.calls: list[str] = []
|
|
|
|
async def generate(self, *, endpoint_code: str, prompt: str, timeout_ms: int) -> str:
|
|
raise AssertionError("embedding 测试不应调用文本生成")
|
|
|
|
async def embed(self, *, endpoint_code: str, text: str, timeout_ms: int) -> list[float]:
|
|
self.calls.append(endpoint_code)
|
|
result = self.behaviour[endpoint_code]
|
|
if isinstance(result, Exception):
|
|
raise result
|
|
return list(result)
|
|
|
|
|
|
async def test_embed_parses_openai_compatible_response() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
assert request.url.path.endswith("/embeddings")
|
|
assert request.headers["authorization"] == "Bearer test-key"
|
|
body = json.loads(request.content)
|
|
assert body["model"] == "embed-small"
|
|
assert body["input"] == "稳健型"
|
|
return httpx.Response(200, json={"data": [{"embedding": [0.1, 0.2, 0.3]}]})
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
gateway = OpenAICompatibleGateway(
|
|
{"ep-1": Endpoint("ep-1")}, secret_resolver=SeedResolver(), client=client
|
|
)
|
|
vector = await gateway.embed(endpoint_code="ep-1", text="稳健型", timeout_ms=1000)
|
|
|
|
assert vector == [0.1, 0.2, 0.3]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"payload",
|
|
[
|
|
{"data": []},
|
|
{"data": [{}]},
|
|
{"data": [{"embedding": []}]},
|
|
{"data": [{"embedding": ["not-a-number"]}]},
|
|
{"unexpected": True},
|
|
],
|
|
)
|
|
async def test_malformed_embedding_response_fails_closed(payload: dict[str, Any]) -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json=payload)
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
gateway = OpenAICompatibleGateway(
|
|
{"ep-1": Endpoint("ep-1")}, secret_resolver=SeedResolver(), client=client
|
|
)
|
|
with pytest.raises(RecoverableAgentError):
|
|
await gateway.embed(endpoint_code="ep-1", text="稳健型", timeout_ms=1000)
|
|
|
|
|
|
async def test_http_failure_is_recoverable() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(503, json={"error": "unavailable"})
|
|
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
|
gateway = OpenAICompatibleGateway(
|
|
{"ep-1": Endpoint("ep-1")}, secret_resolver=SeedResolver(), client=client
|
|
)
|
|
with pytest.raises(RecoverableAgentError):
|
|
await gateway.embed(endpoint_code="ep-1", text="稳健型", timeout_ms=1000)
|
|
|
|
|
|
async def test_dispatch_falls_back_in_declared_order() -> None:
|
|
gateway = RecordingGateway({
|
|
"ep-1": RecoverableAgentError("端点不可用"),
|
|
"ep-2": [0.5, 0.6],
|
|
})
|
|
dispatch = ModelDispatchService(gateway) # type: ignore[arg-type]
|
|
|
|
execution = await dispatch.embed([Endpoint("ep-1"), Endpoint("ep-2")], "稳健型")
|
|
|
|
assert execution.endpoint_code == "ep-2"
|
|
assert execution.vector == [0.5, 0.6]
|
|
assert execution.degraded is True
|
|
assert execution.attempts == 2
|
|
assert gateway.calls == ["ep-1", "ep-2"]
|
|
|
|
|
|
async def test_dispatch_raises_when_all_endpoints_fail() -> None:
|
|
gateway = RecordingGateway({
|
|
"ep-1": RecoverableAgentError("端点不可用"),
|
|
"ep-2": RecoverableAgentError("端点不可用"),
|
|
})
|
|
dispatch = ModelDispatchService(gateway) # type: ignore[arg-type]
|
|
|
|
with pytest.raises(RecoverableAgentError):
|
|
await dispatch.embed([Endpoint("ep-1"), Endpoint("ep-2")], "稳健型")
|
|
|
|
|
|
async def test_embedding_service_rejects_empty_endpoint_list() -> None:
|
|
dispatch = ModelDispatchService(RecordingGateway({})) # type: ignore[arg-type]
|
|
service = ModelEmbeddingService(dispatch)
|
|
|
|
with pytest.raises(RecoverableAgentError):
|
|
await service.embed([], "稳健型")
|