016509c5c6
P1-2 非可重试错误(持续 401/403)也 record_failure，坏 key 可触发熔断。 P1-9 _complete_structured 的 assert 改显式 raise ValueError（-O 安全）。 P2 GatewayRun 抽到 orchestrator/_protocols.py 单点（去 4 处重复）； _is_transient 抽到 adapters/base.py is_transient_by_name（去 3 处重复）； Gemini Protocol 改 async def；gateway._retrying 去无用 async。
188 lines
6.9 KiB
Python
188 lines
6.9 KiB
Python
"""Anthropic（Claude）适配器（ARCH §4.2/§4.4/§4.6）。
|
||
|
||
- 能力：原生结构化输出（经 instructor）、显式前缀缓存（`cache_control` 断点）、思考。
|
||
- 网络客户端注入（`AnthropicClient` Protocol，`AsyncAnthropic` 即满足）——测试注入替身，
|
||
绝不联网；真实 SDK 仅在构造客户端时（apps/api）需要，本模块不硬 import `anthropic`。
|
||
- 瞬时故障（429/超时/5xx/连接错误）翻译为 `TransientProviderError`，交网关退避/回退。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import AsyncIterator
|
||
from typing import TYPE_CHECKING, Any, Protocol, cast
|
||
|
||
import instructor
|
||
from pydantic import BaseModel
|
||
|
||
from ..errors import TransientProviderError
|
||
from ..types import LlmRequest
|
||
from .base import (
|
||
Capabilities,
|
||
ProviderResult,
|
||
ProviderUsage,
|
||
StreamChunk,
|
||
is_transient_by_name,
|
||
)
|
||
|
||
if TYPE_CHECKING:
|
||
from anthropic import AsyncAnthropic
|
||
|
||
# 瞬时错误类名（按名匹配，避免硬依赖 anthropic SDK 类型）。
|
||
_TRANSIENT_NAMES = frozenset(
|
||
{
|
||
"RateLimitError",
|
||
"APITimeoutError",
|
||
"APIConnectionError",
|
||
"InternalServerError",
|
||
"APIStatusError",
|
||
}
|
||
)
|
||
|
||
|
||
class AnthropicClient(Protocol):
|
||
"""`AsyncAnthropic` 满足此 Protocol（仅用到 `messages`）。"""
|
||
|
||
messages: Any
|
||
|
||
|
||
class StructuredAnthropic(Protocol):
|
||
async def create(self, *, response_model: type[BaseModel], **kwargs: Any) -> BaseModel: ...
|
||
|
||
|
||
def _is_transient(exc: Exception) -> bool:
|
||
return is_transient_by_name(exc, _TRANSIENT_NAMES)
|
||
|
||
|
||
def _system_blocks(req: LlmRequest) -> list[dict[str, Any]]:
|
||
"""system 块；缓存断点前的稳定块带 `cache_control`（§4.6）。"""
|
||
out: list[dict[str, Any]] = []
|
||
for b in req.system:
|
||
block: dict[str, Any] = {"type": "text", "text": b.text}
|
||
if b.cache:
|
||
block["cache_control"] = {"type": "ephemeral"}
|
||
out.append(block)
|
||
return out
|
||
|
||
|
||
def _input_text(req: LlmRequest) -> str:
|
||
if isinstance(req.input, str):
|
||
return req.input
|
||
return "\n\n".join(b.text for b in req.input)
|
||
|
||
|
||
def _user_messages(req: LlmRequest) -> list[dict[str, Any]]:
|
||
return [{"role": "user", "content": _input_text(req)}]
|
||
|
||
|
||
def _usage_from(usage: Any) -> ProviderUsage:
|
||
if usage is None:
|
||
return ProviderUsage(input_tokens=0, output_tokens=0)
|
||
return ProviderUsage(
|
||
input_tokens=int(getattr(usage, "input_tokens", 0) or 0),
|
||
output_tokens=int(getattr(usage, "output_tokens", 0) or 0),
|
||
cache_read_tokens=int(getattr(usage, "cache_read_input_tokens", 0) or 0),
|
||
)
|
||
|
||
|
||
def _text_from(resp: Any) -> str:
|
||
parts: list[str] = []
|
||
for block in getattr(resp, "content", []) or []:
|
||
if getattr(block, "type", None) == "text":
|
||
parts.append(getattr(block, "text", "") or "")
|
||
return "".join(parts)
|
||
|
||
|
||
_DEFAULT_MAX_TOKENS = 4096
|
||
|
||
|
||
class AnthropicAdapter:
|
||
def __init__(
|
||
self,
|
||
provider: str,
|
||
client: AnthropicClient,
|
||
*,
|
||
structured_client: StructuredAnthropic | None = None,
|
||
) -> None:
|
||
self.provider = provider
|
||
self._client = client
|
||
self._structured_client = structured_client
|
||
|
||
def capabilities(self) -> Capabilities:
|
||
return Capabilities(structured_output=True, prefix_cache=True, thinking=True)
|
||
|
||
def _structured(self) -> StructuredAnthropic:
|
||
if self._structured_client is None:
|
||
# 注入客户端是 `AnthropicClient` Protocol（保留测试替身缝），
|
||
# 但 instructor.from_anthropic 只重载真实 SDK 的具体类型；
|
||
# 本适配器异步 → 转 `AsyncAnthropic` 选中 AsyncInstructor 重载，
|
||
# 返回的 AsyncInstructor 鸭子匹配 StructuredAnthropic。
|
||
async_client = cast("AsyncAnthropic", self._client)
|
||
self._structured_client = cast(
|
||
"StructuredAnthropic", instructor.from_anthropic(async_client)
|
||
)
|
||
return self._structured_client
|
||
|
||
async def complete(self, req: LlmRequest, model: str) -> ProviderResult:
|
||
try:
|
||
if req.output_schema is not None:
|
||
return await self._complete_structured(req, model)
|
||
return await self._complete_text(req, model)
|
||
except Exception as exc:
|
||
if _is_transient(exc):
|
||
raise TransientProviderError(str(exc), provider=self.provider) from exc
|
||
raise
|
||
|
||
async def _complete_text(self, req: LlmRequest, model: str) -> ProviderResult:
|
||
kwargs: dict[str, Any] = {
|
||
"model": model,
|
||
"max_tokens": req.max_tokens or _DEFAULT_MAX_TOKENS,
|
||
"messages": _user_messages(req),
|
||
}
|
||
system = _system_blocks(req)
|
||
if system:
|
||
kwargs["system"] = system
|
||
resp = await self._client.messages.create(**kwargs)
|
||
usage = _usage_from(getattr(resp, "usage", None))
|
||
return ProviderResult(text=_text_from(resp), usage=usage)
|
||
|
||
async def _complete_structured(self, req: LlmRequest, model: str) -> ProviderResult:
|
||
if req.output_schema is None:
|
||
raise ValueError("_complete_structured called without output_schema")
|
||
kwargs: dict[str, Any] = {
|
||
"model": model,
|
||
"max_tokens": req.max_tokens or _DEFAULT_MAX_TOKENS,
|
||
"messages": _user_messages(req),
|
||
}
|
||
system = _system_blocks(req)
|
||
if system:
|
||
kwargs["system"] = system
|
||
parsed = await self._structured().create(response_model=req.output_schema, **kwargs)
|
||
# instructor.from_anthropic 默认不回 raw usage（按名隐藏）；usage 经流/text 路径覆盖。
|
||
usage = ProviderUsage(input_tokens=0, output_tokens=0)
|
||
return ProviderResult(text=parsed.model_dump_json(), usage=usage, parsed=parsed)
|
||
|
||
async def stream(self, req: LlmRequest, model: str) -> AsyncIterator[StreamChunk]:
|
||
kwargs: dict[str, Any] = {
|
||
"model": model,
|
||
"max_tokens": req.max_tokens or _DEFAULT_MAX_TOKENS,
|
||
"messages": _user_messages(req),
|
||
}
|
||
system = _system_blocks(req)
|
||
if system:
|
||
kwargs["system"] = system
|
||
try:
|
||
async with self._client.messages.stream(**kwargs) as stream:
|
||
async for event in stream:
|
||
if getattr(event, "type", None) == "content_block_delta":
|
||
delta = getattr(event, "delta", None)
|
||
text = getattr(delta, "text", "") if delta is not None else ""
|
||
if text:
|
||
yield StreamChunk(text=text)
|
||
usage = getattr(event, "usage", None)
|
||
if usage is not None:
|
||
yield StreamChunk(usage=_usage_from(usage))
|
||
except Exception as exc:
|
||
if _is_transient(exc):
|
||
raise TransientProviderError(str(exc), provider=self.provider) from exc
|
||
raise
|