434 lines
15 KiB
Python
434 lines
15 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from collections.abc import AsyncIterator
|
||
from typing import Any
|
||
|
||
import httpx
|
||
|
||
from app.ai_provider.config.models import ModelConfig, ProviderConfig
|
||
from app.ai_provider.pipeline import ParameterPipeline, PipelineContext, PipelineFactory
|
||
from app.ai_provider.types.chat import ChatChunk, ChatResult, TokenUsage
|
||
from app.ai_provider.types.errors import AIProviderError
|
||
from app.ai_provider.types.messages import (
|
||
AIMessage,
|
||
BaseMessage,
|
||
SystemMessage,
|
||
ToolMessage,
|
||
)
|
||
from app.core.logging import logger
|
||
|
||
|
||
class AnthropicFormatAdapter:
|
||
"""兼容 Anthropic Messages API 格式的适配器。
|
||
|
||
覆盖通过 Anthropic API 访问的 Claude 模型，以及任何镜像
|
||
Anthropic 消息格式的提供商。
|
||
"""
|
||
|
||
API_VERSION = "2023-06-01"
|
||
|
||
def __init__(
|
||
self,
|
||
provider_config: ProviderConfig,
|
||
pipeline: ParameterPipeline | None = None,
|
||
) -> None:
|
||
self._provider = provider_config
|
||
self._pipeline = pipeline or PipelineFactory.create(provider_config)
|
||
self._client: httpx.AsyncClient | None = None
|
||
|
||
@property
|
||
def _http(self) -> httpx.AsyncClient:
|
||
if self._client is None:
|
||
self._client = httpx.AsyncClient(
|
||
base_url=self._provider.base_url or "",
|
||
timeout=self._provider.timeout_seconds,
|
||
)
|
||
return self._client
|
||
|
||
# ------------------------------------------------------------------
|
||
# 内部方法：消息转换
|
||
# ------------------------------------------------------------------
|
||
|
||
def _convert_message(self, msg: BaseMessage) -> dict[str, Any]:
|
||
"""将单条消息转换为 Anthropic 消息格式。
|
||
|
||
Anthropic 使用 content blocks 数组而非纯字符串。
|
||
System 消息**不在此处处理** —— 它们由 :meth:`build_request`
|
||
提取到顶层的 ``system`` 字段中。
|
||
"""
|
||
if isinstance(msg, ToolMessage):
|
||
return {
|
||
"role": "user",
|
||
"content": [
|
||
{
|
||
"type": "tool_result",
|
||
"tool_use_id": msg.tool_call_id,
|
||
"content": msg.content,
|
||
}
|
||
],
|
||
}
|
||
|
||
if isinstance(msg, AIMessage):
|
||
if msg.raw_content_blocks:
|
||
return {"role": "assistant", "content": msg.raw_content_blocks}
|
||
blocks: list[dict[str, Any]] = []
|
||
if msg.content:
|
||
blocks.append({"type": "text", "text": msg.content})
|
||
for tc in msg.tool_calls:
|
||
blocks.append(
|
||
{
|
||
"type": "tool_use",
|
||
"id": tc.get("id", f"tu_{tc['name']}"),
|
||
"name": tc["name"],
|
||
"input": tc.get("args", {}),
|
||
}
|
||
)
|
||
return {"role": "assistant", "content": blocks}
|
||
|
||
# HumanMessage
|
||
return {
|
||
"role": "user",
|
||
"content": [{"type": "text", "text": msg.content}],
|
||
}
|
||
|
||
@staticmethod
|
||
def _convert_tools(tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""将 OpenAI 兼容格式的工具转换为 Anthropic 格式。
|
||
|
||
OpenAI 格式::
|
||
|
||
{"type": "function", "function": {"name": ..., "parameters": ...}}
|
||
|
||
Anthropic 格式::
|
||
|
||
{"name": ..., "input_schema": ...}
|
||
"""
|
||
result: list[dict[str, Any]] = []
|
||
for tool in tools:
|
||
fn = tool.get("function", tool)
|
||
result.append(
|
||
{
|
||
"name": fn.get("name", ""),
|
||
"description": fn.get("description", ""),
|
||
"input_schema": fn.get("parameters", {"type": "object"}),
|
||
}
|
||
)
|
||
return result
|
||
|
||
# ------------------------------------------------------------------
|
||
# 请求构建
|
||
# ------------------------------------------------------------------
|
||
|
||
def build_request(
|
||
self,
|
||
model: ModelConfig,
|
||
messages: list[BaseMessage],
|
||
params: dict[str, Any] | None = None,
|
||
tools: list[dict[str, Any]] | None = None,
|
||
tool_choice: str | dict[str, Any] | None = None,
|
||
stream: bool = False,
|
||
) -> dict[str, Any]:
|
||
ctx = PipelineContext(
|
||
params=params or {},
|
||
model_config=model,
|
||
provider_config=self._provider,
|
||
)
|
||
pipe_result = self._pipeline.run(ctx)
|
||
|
||
# Extract system messages — Anthropic uses a top-level "system" field
|
||
system_parts: list[str] = []
|
||
non_system: list[BaseMessage] = []
|
||
for m in messages:
|
||
if isinstance(m, SystemMessage):
|
||
system_parts.append(m.content)
|
||
else:
|
||
non_system.append(m)
|
||
|
||
body: dict[str, Any] = {
|
||
"model": model.model,
|
||
"messages": [self._convert_message(m) for m in non_system],
|
||
"stream": stream,
|
||
}
|
||
|
||
if system_parts:
|
||
body["system"] = "\n".join(system_parts)
|
||
|
||
body.update(pipe_result.params)
|
||
|
||
if tools:
|
||
body["tools"] = self._convert_tools(tools)
|
||
if tool_choice:
|
||
body["tool_choice"] = tool_choice
|
||
|
||
# Anthropic requires explicit max_tokens
|
||
if "max_tokens" not in body:
|
||
body["max_tokens"] = model.max_output_tokens or 4096
|
||
|
||
return body
|
||
|
||
def bind_tools(
|
||
self,
|
||
request_body: dict[str, Any],
|
||
tools: list[dict[str, Any]],
|
||
tool_choice: str | dict[str, Any] | None = None,
|
||
) -> None:
|
||
request_body["tools"] = self._convert_tools(tools)
|
||
if tool_choice:
|
||
request_body["tool_choice"] = tool_choice
|
||
|
||
# ------------------------------------------------------------------
|
||
# 认证头
|
||
# ------------------------------------------------------------------
|
||
|
||
def _headers(self, api_key: str) -> dict[str, str]:
|
||
return {
|
||
"x-api-key": api_key,
|
||
"anthropic-version": self.API_VERSION,
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
# ------------------------------------------------------------------
|
||
# 响应解析 —— 非流式
|
||
# ------------------------------------------------------------------
|
||
|
||
@staticmethod
|
||
def _parse_content_blocks(raw: dict[str, Any]) -> tuple[str, list[dict[str, Any]], str]:
|
||
"""解析 Anthropic content blocks 为 (text, tool_calls, reasoning)。"""
|
||
content_blocks = raw.get("content", [])
|
||
text_parts: list[str] = []
|
||
tool_calls: list[dict[str, Any]] = []
|
||
reasoning = ""
|
||
|
||
for block in content_blocks:
|
||
block_type = block.get("type", "")
|
||
if block_type == "text":
|
||
text_parts.append(block.get("text", ""))
|
||
elif block_type == "tool_use":
|
||
tool_calls.append(
|
||
{
|
||
"name": block.get("name", ""),
|
||
"args": block.get("input", {}),
|
||
"id": block.get("id", ""),
|
||
}
|
||
)
|
||
elif block_type == "thinking":
|
||
reasoning += block.get("thinking", "")
|
||
# signature 块是 thinking 的加密签名，不是 reasoning 内容，忽略
|
||
|
||
return "".join(text_parts), tool_calls, reasoning
|
||
|
||
def parse_nonstreaming_response(
|
||
self,
|
||
raw: dict[str, Any],
|
||
model_name: str,
|
||
) -> ChatResult:
|
||
content, tool_calls, reasoning = self._parse_content_blocks(raw)
|
||
|
||
result = ChatResult(
|
||
content=content,
|
||
model_name=model_name,
|
||
)
|
||
if tool_calls:
|
||
result.tool_calls = tool_calls
|
||
if reasoning:
|
||
result.reasoning = reasoning
|
||
|
||
usage = raw.get("usage")
|
||
if usage:
|
||
result.usage = TokenUsage(
|
||
prompt_tokens=usage.get("input_tokens", 0),
|
||
prompt_tokens_cached=usage.get("cache_read_input_tokens", 0),
|
||
completion_tokens=usage.get("output_tokens", 0),
|
||
total_tokens=usage.get("input_tokens", 0) + usage.get("output_tokens", 0),
|
||
)
|
||
|
||
return result
|
||
|
||
# ------------------------------------------------------------------
|
||
# 响应解析 —— 流式（SSE 事件行协议）
|
||
# ------------------------------------------------------------------
|
||
|
||
@staticmethod
|
||
def _parse_stream_event(event_type: str, data: dict[str, Any]) -> ChatChunk:
|
||
"""解析单条 SSE 流事件为 ChatChunk。"""
|
||
chunk = ChatChunk()
|
||
|
||
if event_type == "message_start":
|
||
message = data.get("message", {})
|
||
usage = message.get("usage")
|
||
if usage:
|
||
chunk.usage = TokenUsage(
|
||
prompt_tokens=usage.get("input_tokens", 0),
|
||
prompt_tokens_cached=usage.get("cache_read_input_tokens", 0),
|
||
completion_tokens=usage.get("output_tokens", 0),
|
||
)
|
||
|
||
elif event_type == "content_block_start":
|
||
block = data.get("content_block", {})
|
||
chunk.content_block_index = data.get("index", 0)
|
||
chunk.content_block_start = block
|
||
block_type = block.get("type", "")
|
||
if block_type == "text":
|
||
chunk.delta_content = block.get("text", "")
|
||
elif block_type == "thinking":
|
||
chunk.delta_reasoning = block.get("thinking", "")
|
||
elif block_type == "tool_use":
|
||
chunk.tool_call_chunks.append(
|
||
{
|
||
"index": data.get("index", 0),
|
||
"id": block.get("id"),
|
||
"name": block.get("name"),
|
||
"args": "",
|
||
}
|
||
)
|
||
|
||
elif event_type == "content_block_delta":
|
||
delta = data.get("delta", {})
|
||
chunk.content_block_index = data.get("index", 0)
|
||
chunk.content_block_delta = delta
|
||
delta_type = delta.get("type", "")
|
||
if delta_type == "text_delta":
|
||
chunk.delta_content = delta.get("text", "")
|
||
elif delta_type == "thinking_delta":
|
||
chunk.delta_reasoning = delta.get("thinking", "")
|
||
elif delta_type == "signature_delta":
|
||
# signature 不展示，但由 Agent Runtime 原样保留供下一轮回放。
|
||
pass
|
||
elif delta_type == "input_json_delta":
|
||
chunk.tool_call_chunks.append(
|
||
{
|
||
"index": data.get("index", 0),
|
||
"id": None,
|
||
"name": None,
|
||
"args": delta.get("partial_json", ""),
|
||
}
|
||
)
|
||
|
||
elif event_type == "message_delta":
|
||
delta = data.get("delta", {})
|
||
finish_reason_map = {
|
||
"end_turn": "stop",
|
||
"max_tokens": "length",
|
||
"stop_sequence": "stop",
|
||
"tool_use": "tool_calls",
|
||
}
|
||
stop_reason = delta.get("stop_reason")
|
||
if stop_reason:
|
||
chunk.finish_reason = finish_reason_map.get(stop_reason, stop_reason)
|
||
|
||
usage = data.get("usage")
|
||
if usage:
|
||
chunk.usage = TokenUsage(
|
||
prompt_tokens=0,
|
||
completion_tokens=usage.get("output_tokens", 0),
|
||
)
|
||
|
||
# content_block_stop, message_stop, ping — no content to extract
|
||
|
||
return chunk
|
||
|
||
# ------------------------------------------------------------------
|
||
# 公开 API：chat
|
||
# ------------------------------------------------------------------
|
||
|
||
async def chat(
|
||
self,
|
||
model: ModelConfig,
|
||
messages: list[BaseMessage],
|
||
api_key: str,
|
||
base_url: str | None = None,
|
||
params: dict[str, Any] | None = None,
|
||
tools: list[dict[str, Any]] | None = None,
|
||
tool_choice: str | dict[str, Any] | None = None,
|
||
) -> ChatResult:
|
||
url = f"{base_url or self._provider.base_url}/v1/messages"
|
||
body = self.build_request(
|
||
model,
|
||
messages,
|
||
params=params,
|
||
tools=tools,
|
||
tool_choice=tool_choice,
|
||
)
|
||
|
||
try:
|
||
response = await self._http.post(url, json=body, headers=self._headers(api_key))
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
return self.parse_nonstreaming_response(data, model.name)
|
||
except httpx.HTTPStatusError as exc:
|
||
raise AIProviderError.from_provider(
|
||
message=f"Anthropic API error: {exc.response.status_code} {exc.response.text[:200]}",
|
||
code="PROVIDER_ERROR",
|
||
provider=self._provider.name,
|
||
original=exc,
|
||
) from exc
|
||
|
||
async def chat_stream(
|
||
self,
|
||
model: ModelConfig,
|
||
messages: list[BaseMessage],
|
||
api_key: str,
|
||
base_url: str | None = None,
|
||
params: dict[str, Any] | None = None,
|
||
tools: list[dict[str, Any]] | None = None,
|
||
tool_choice: str | dict[str, Any] | None = None,
|
||
) -> AsyncIterator[ChatChunk]:
|
||
url = f"{base_url or self._provider.base_url}/v1/messages"
|
||
body = self.build_request(
|
||
model,
|
||
messages,
|
||
params=params,
|
||
tools=tools,
|
||
tool_choice=tool_choice,
|
||
stream=True,
|
||
)
|
||
|
||
async with self._http.stream(
|
||
"POST",
|
||
url,
|
||
json=body,
|
||
headers=self._headers(api_key),
|
||
) as response:
|
||
if response.status_code != 200:
|
||
error_text = await response.aread()
|
||
raise AIProviderError.from_provider(
|
||
message=f"Anthropic stream error: {response.status_code} {error_text[:200]}",
|
||
code="PROVIDER_ERROR",
|
||
provider=self._provider.name,
|
||
)
|
||
|
||
current_event = ""
|
||
async for line in response.aiter_lines():
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
if line.startswith("event: "):
|
||
current_event = line[7:]
|
||
elif line.startswith("data: "):
|
||
data_str = line[6:]
|
||
if data_str and current_event:
|
||
try:
|
||
raw = json.loads(data_str)
|
||
yield self._parse_stream_event(current_event, raw)
|
||
except json.JSONDecodeError:
|
||
logger.warning(
|
||
"anthropic_stream_parse_error",
|
||
event=current_event,
|
||
raw=data_str[:100],
|
||
)
|
||
current_event = ""
|
||
|
||
async def close(self) -> None:
|
||
if self._client:
|
||
await self._client.aclose()
|
||
self._client = None
|
||
|
||
|
||
class AnthropicFormatAdapterFactory:
|
||
"""工厂：创建 Anthropic 格式适配器。"""
|
||
|
||
@staticmethod
|
||
def create(provider_config: ProviderConfig) -> AnthropicFormatAdapter:
|
||
return AnthropicFormatAdapter(provider_config=provider_config)
|