126 lines
4.4 KiB
Python
126 lines
4.4 KiB
Python
"""OpenAI 兼容模型凭证的最小连通测试。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import time
|
||
import wave
|
||
|
||
import httpx
|
||
|
||
from schemas import CredentialTestRequest, CredentialTestResult
|
||
|
||
TEST_TIMEOUT_SECONDS = 10.0
|
||
|
||
|
||
def _endpoint(base_url: str, path: str) -> str:
|
||
return f"{base_url.rstrip('/')}/{path.lstrip('/')}"
|
||
|
||
|
||
def _silent_wav() -> bytes:
|
||
buffer = io.BytesIO()
|
||
with wave.open(buffer, "wb") as wav:
|
||
wav.setnchannels(1)
|
||
wav.setsampwidth(2)
|
||
wav.setframerate(16_000)
|
||
wav.writeframes(b"\x00\x00" * 1_600)
|
||
return buffer.getvalue()
|
||
|
||
|
||
def _error_detail(response: httpx.Response, api_key: str) -> str:
|
||
try:
|
||
body = response.json()
|
||
detail = (
|
||
body.get("error", {}).get("message")
|
||
if isinstance(body, dict) and isinstance(body.get("error"), dict)
|
||
else body.get("detail") if isinstance(body, dict) else None
|
||
)
|
||
except ValueError:
|
||
detail = None
|
||
text = str(detail or response.text or response.reason_phrase).strip()
|
||
return text.replace(api_key, "***")[:300]
|
||
|
||
|
||
async def test_openai_credential(
|
||
config: CredentialTestRequest,
|
||
) -> CredentialTestResult:
|
||
started = time.perf_counter()
|
||
headers = {"Authorization": f"Bearer {config.api_key}"}
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=TEST_TIMEOUT_SECONDS) as client:
|
||
if config.type == "LLM":
|
||
response = await client.post(
|
||
_endpoint(config.api_url, "chat/completions"),
|
||
headers=headers,
|
||
json={
|
||
"model": config.model_id,
|
||
"messages": [{"role": "user", "content": "Reply with OK."}],
|
||
"max_tokens": 1,
|
||
"stream": False,
|
||
},
|
||
)
|
||
elif config.type == "Embedding":
|
||
response = await client.post(
|
||
_endpoint(config.api_url, "embeddings"),
|
||
headers=headers,
|
||
json={"model": config.model_id, "input": "ping"},
|
||
)
|
||
elif config.type == "ASR":
|
||
response = await client.post(
|
||
_endpoint(config.api_url, "audio/transcriptions"),
|
||
headers=headers,
|
||
data={
|
||
"model": config.model_id,
|
||
**({"language": config.language} if config.language else {}),
|
||
},
|
||
files={"file": ("test.wav", _silent_wav(), "audio/wav")},
|
||
)
|
||
elif config.type == "TTS":
|
||
response = await client.post(
|
||
_endpoint(config.api_url, "audio/speech"),
|
||
headers=headers,
|
||
json={
|
||
"model": config.model_id,
|
||
"input": "测试",
|
||
"voice": config.voice,
|
||
"response_format": "pcm",
|
||
"speed": config.speed,
|
||
},
|
||
)
|
||
else:
|
||
return CredentialTestResult(
|
||
ok=False,
|
||
message="暂不支持该资源类型的连通测试",
|
||
detail=f"当前仅支持 LLM、Embedding、ASR、TTS，收到 {config.type}",
|
||
)
|
||
|
||
latency_ms = round((time.perf_counter() - started) * 1000)
|
||
if response.is_success:
|
||
return CredentialTestResult(
|
||
ok=True,
|
||
latency_ms=latency_ms,
|
||
message="连接成功",
|
||
detail=f"OpenAI 兼容接口响应正常（HTTP {response.status_code}）",
|
||
)
|
||
return CredentialTestResult(
|
||
ok=False,
|
||
latency_ms=latency_ms,
|
||
message=f"连接失败（HTTP {response.status_code}）",
|
||
detail=_error_detail(response, config.api_key),
|
||
)
|
||
except httpx.TimeoutException:
|
||
return CredentialTestResult(
|
||
ok=False,
|
||
latency_ms=round((time.perf_counter() - started) * 1000),
|
||
message="连接超时",
|
||
detail=f"服务未在 {TEST_TIMEOUT_SECONDS:g} 秒内响应",
|
||
)
|
||
except httpx.RequestError as exc:
|
||
return CredentialTestResult(
|
||
ok=False,
|
||
latency_ms=round((time.perf_counter() - started) * 1000),
|
||
message="无法连接到模型服务",
|
||
detail=str(exc)[:300],
|
||
)
|