58 lines
2.0 KiB
Python
58 lines
2.0 KiB
Python
"""Anthropic Messages API protocol."""
|
|
|
|
from typing import Any
|
|
|
|
from agenteval.models import ModelCapability, ModelProtocol
|
|
|
|
from .base import ModelProtocolAdapter, ProtocolAdapterError
|
|
|
|
|
|
class AnthropicAdapter(ModelProtocolAdapter):
|
|
protocol = ModelProtocol.ANTHROPIC
|
|
supported_capabilities = frozenset({ModelCapability.CHAT})
|
|
|
|
def headers(self, api_key: str | None) -> dict[str, str]:
|
|
headers = super().headers(api_key)
|
|
headers["anthropic-version"] = "2023-06-01"
|
|
if api_key:
|
|
headers["x-api-key"] = api_key
|
|
return headers
|
|
|
|
def chat_payload(
|
|
self,
|
|
model_name: str | None,
|
|
messages: list[dict[str, str]],
|
|
temperature: float,
|
|
) -> dict[str, Any]:
|
|
system_parts: list[str] = []
|
|
api_messages: list[dict[str, str]] = []
|
|
for message in messages:
|
|
role = message.get("role", "user")
|
|
content = message.get("content", "")
|
|
if role in {"system", "developer"}:
|
|
system_parts.append(content)
|
|
else:
|
|
api_messages.append({"role": "assistant" if role == "assistant" else "user", "content": content})
|
|
payload: dict[str, Any] = {
|
|
"model": self.require_model(model_name),
|
|
"max_tokens": 1024,
|
|
"messages": api_messages,
|
|
"temperature": temperature,
|
|
}
|
|
if system_parts:
|
|
payload["system"] = "\n\n".join(system_parts)
|
|
return payload
|
|
|
|
def parse_chat(self, data: dict[str, Any]) -> str:
|
|
blocks = data.get("content")
|
|
if not isinstance(blocks, list):
|
|
raise ProtocolAdapterError("Anthropic 接口返回格式不正确")
|
|
content = "\n".join(
|
|
str(block.get("text", ""))
|
|
for block in blocks
|
|
if isinstance(block, dict) and block.get("type") == "text" and block.get("text")
|
|
)
|
|
if not content:
|
|
raise ProtocolAdapterError("模型接口返回内容为空")
|
|
return content
|