refactor(gateway): 适配器暴露 probe_connection 公开探测——provider_deps 弃私有 _client 反手(CR-M1.4)

This commit is contained in:
Yaojia Wang
2026-07-08 13:00:50 +02:00
parent 9439b97804
commit 832ab7dfa1
3 changed files with 53 additions and 3 deletions

View File

@@ -14,6 +14,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from ww_config import get_settings from ww_config import get_settings
from ww_db import get_session from ww_db import get_session
from ww_llm_gateway.adapters.base import Capabilities from ww_llm_gateway.adapters.base import Capabilities
from ww_llm_gateway.adapters.openai_compat import OpenAICompatAdapter
from ww_llm_gateway.factory import build_adapter from ww_llm_gateway.factory import build_adapter
from ww_shared import AppError, ErrorCode from ww_shared import AppError, ErrorCode
@@ -85,10 +86,17 @@ class GatewayProviderProbe:
# 经网关工厂构造适配器base_url 单点归网关 build_adapter不在此本地构造 AsyncOpenAI # 经网关工厂构造适配器base_url 单点归网关 build_adapter不在此本地构造 AsyncOpenAI
adapter = build_adapter(provider, api_key=api_key, base_url=base_url) adapter = build_adapter(provider, api_key=api_key, base_url=base_url)
# `_PROVIDER_BASE_URLS` 里的 api_key 型 provider 均映射到 OpenAI 兼容适配器(含
# kimi-code-key 子类);据此收窄类型,用公开 `probe_connection` 探测(不反手私有客户端)。
if not isinstance(adapter, OpenAICompatAdapter):
raise AppError(
ErrorCode.VALIDATION,
f"provider {provider} 不支持 api_key 连接测试",
{"provider": provider},
)
try: try:
# 最小探测:列模型即可验证 Key 有效(不消耗生成额度)。底层 AsyncOpenAI 由适配器持有 # 最小探测:列模型即可验证 Key 有效(不消耗生成额度)。
# (已知 provider 为 OpenAI 兼容,见 _PROVIDER_BASE_URLS读私有客户端属同仓既有约定。 await adapter.probe_connection()
await adapter._client.models.list() # type: ignore[attr-defined] # noqa: SLF001
except Exception as exc: # noqa: BLE001 — 任一失败都映射为 LLM 不可用 except Exception as exc: # noqa: BLE001 — 任一失败都映射为 LLM 不可用
raise AppError( raise AppError(
ErrorCode.LLM_UNAVAILABLE, ErrorCode.LLM_UNAVAILABLE,

View File

@@ -7,6 +7,7 @@ from collections.abc import AsyncIterator
from types import SimpleNamespace from types import SimpleNamespace
from typing import Any, cast from typing import Any, cast
import pytest
from openai import AsyncOpenAI from openai import AsyncOpenAI
from ww_llm_gateway.adapters.openai_compat import OpenAICompatAdapter, _messages from ww_llm_gateway.adapters.openai_compat import OpenAICompatAdapter, _messages
from ww_llm_gateway.types import Block, LlmRequest, Scope from ww_llm_gateway.types import Block, LlmRequest, Scope
@@ -100,3 +101,36 @@ async def test_stream_yields_text_then_final_usage() -> None:
assert "".join(texts) == "他说" assert "".join(texts) == "他说"
assert usage_seen is not None assert usage_seen is not None
assert usage_seen.output_tokens == 2 assert usage_seen.output_tokens == 2
class _FakeModels:
def __init__(self, *, error: Exception | None = None) -> None:
self._error = error
self.calls = 0
async def list(self) -> Any:
self.calls += 1
if self._error is not None:
raise self._error
return SimpleNamespace(data=[])
async def test_probe_connection_lists_models() -> None:
# CR-M1.4:连通性探测经公开 `probe_connection`(不再让调用方反手私有 `_client`)。
models = _FakeModels()
client = SimpleNamespace(models=models)
adapter = OpenAICompatAdapter("deepseek", cast(AsyncOpenAI, client))
await adapter.probe_connection()
assert models.calls == 1
async def test_probe_connection_propagates_failure() -> None:
# 探测失败原样上抛(调用方据此映射 LLM 不可用)。
models = _FakeModels(error=RuntimeError("bad key"))
client = SimpleNamespace(models=models)
adapter = OpenAICompatAdapter("deepseek", cast(AsyncOpenAI, client))
with pytest.raises(RuntimeError):
await adapter.probe_connection()

View File

@@ -103,6 +103,14 @@ class OpenAICompatAdapter:
def capabilities(self) -> Capabilities: def capabilities(self) -> Capabilities:
return Capabilities(structured_output=True, prefix_cache=True, thinking=False) return Capabilities(structured_output=True, prefix_cache=True, thinking=False)
async def probe_connection(self) -> None:
"""最小连通性探测:列模型即验 Key 有效不消耗生成额度失败原样上抛CR-M1.4)。
凭据连接测试的**公开入口**——调用方apps/api provider_deps经此验 Key无需
反手适配器私有客户端。任一异常上抛,由调用方映射为 LLM 不可用。
"""
await self._client.models.list()
def _structured(self) -> StructuredClient: def _structured(self) -> StructuredClient:
if self._structured_client is None: if self._structured_client is None:
# 懒构建:从同一 AsyncOpenAI client patch 出 instructor 客户端。 # 懒构建:从同一 AsyncOpenAI client patch 出 instructor 客户端。