Coverage for agentos/llm/anthropic_provider.py: 18%
159 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 12:29 +0800
1"""
2Anthropic Claude Provider — 基于 httpx 直接调用 Anthropic Messages API。
3零额外依赖，不依赖 anthropic SDK。
4v1.3.36: 首个纯 httpx 实现，支持同步/异步/流式/Function Calling。
5"""
7from __future__ import annotations
9import json
10from collections.abc import Iterator
11from typing import Any
13import httpx
15from agentos.llm.base import (
16 CompletionChoice,
17 CompletionResult,
18 CompletionUsage,
19 LLMProvider,
20 Message,
21 MessageRole,
22 StreamChunk,
23 Tool,
24 ToolCall,
25)
27__all__ = ["AnthropicProvider"]
29ANTHROPIC_API_BASE = "https://api.anthropic.com"
30ANTHROPIC_VERSION = "2023-06-01"
32# USD per 1M tokens (Anthropic pricing as of 2026-07)
33_PRICING: dict[str, tuple[float, float]] = {
34 "claude-sonnet-5-20250630": (3.0, 15.0), # Sonnet 5 — 性价比最高的 Agent 模型
35 "claude-sonnet-5-20250701": (3.0, 15.0), # Sonnet 5 (alternate ID)
36 "claude-sonnet-4-20250514": (3.0, 15.0),
37 "claude-3-5-sonnet-20241022": (3.0, 15.0),
38 "claude-3-5-haiku-20241022": (0.80, 4.0),
39 "claude-3-opus-20240229": (15.0, 75.0),
40 "claude-3-haiku-20240307": (0.25, 1.25),
41 "claude-opus-4-20250514": (15.0, 75.0), # Opus 4
42 "claude-opus-4-5-20251101": (15.0, 75.0), # Opus 4.5
43}
45# 5-series models identified by prefix
46_SONNET5_PREFIXES = ("claude-sonnet-5", "claude-sonnet5", "sonnet-5")
49def _is_sonnet5(model: str) -> bool:
50 return any(model.startswith(p) for p in _SONNET5_PREFIXES) or "sonnet-5" in model
53def _messages_to_anthropic(messages: list[Message]) -> tuple[str | None, list[dict[str, Any]]]:
54 """将 Message 列表转换为 Anthropic Messages API 格式。
55 Returns (system_prompt, api_messages) — Anthropic 的 system 是顶层字段。
56 """
57 system_parts: list[str] = []
58 api_messages: list[dict[str, Any]] = []
60 for m in messages:
61 if m.role == MessageRole.SYSTEM:
62 system_parts.append(m.content)
63 continue
65 entry: dict[str, Any] = {"role": _ROLE_MAP[m.role], "content": m.content}
66 if m.tool_calls:
67 # 将 tool_calls 转成 Anthropic tool_use content blocks
68 content_blocks: list[dict[str, Any]] = []
69 if m.content:
70 content_blocks.append({"type": "text", "text": m.content})
71 for tc in m.tool_calls:
72 content_blocks.append(
73 {
74 "type": "tool_use",
75 "id": tc.id,
76 "name": tc.name,
77 "input": json.loads(tc.arguments),
78 }
79 )
80 entry["content"] = content_blocks
82 if m.role == MessageRole.TOOL and m.tool_call_id:
83 entry["content"] = [
84 {
85 "type": "tool_result",
86 "tool_use_id": m.tool_call_id,
87 "content": m.content,
88 }
89 ]
90 del entry["role"]
91 api_messages.append(entry)
93 system_prompt = "\n".join(system_parts) if system_parts else None
94 return system_prompt, api_messages
97_ROLE_MAP: dict[MessageRole, str] = {
98 MessageRole.USER: "user",
99 MessageRole.ASSISTANT: "assistant",
100 MessageRole.TOOL: "user", # Anthropic 用 user 角色承载 tool_result
101}
104def _tools_to_anthropic(tools: list[Tool] | None) -> list[dict[str, Any]] | None:
105 if not tools:
106 return None
107 result: list[dict[str, Any]] = []
108 for t in tools:
109 fn = t.function
110 props = fn.parameters
111 result.append(
112 {
113 "name": fn.name,
114 "description": fn.description,
115 "input_schema": {
116 "type": "object",
117 "properties": {k: v.as_schema() for k, v in props.items()},
118 "required": fn.required or [k for k, v in props.items() if v.required],
119 },
120 }
121 )
122 return result
125def _parse_anthropic_tool_calls(content_blocks: list[dict[str, Any]]) -> list[ToolCall]:
126 """从 Anthropic content 中提取 tool_use blocks。"""
127 result: list[ToolCall] = []
128 for block in content_blocks:
129 if block.get("type") == "tool_use":
130 result.append(
131 ToolCall(
132 id=block["id"],
133 name=block["name"],
134 arguments=json.dumps(block.get("input", {})),
135 )
136 )
137 return result
140def _parse_text_content(content_blocks: list[dict[str, Any]]) -> str:
141 """提取 text content blocks 中的文本。"""
142 texts: list[str] = []
143 for block in content_blocks:
144 if block.get("type") == "text":
145 texts.append(block.get("text", ""))
146 return "".join(texts)
149def _build_result(data: dict[str, Any], model: str) -> CompletionResult:
150 content = data.get("content", [])
151 if isinstance(content, str):
152 text = content
153 tool_calls = None
154 else:
155 text = _parse_text_content(content)
156 tool_calls = _parse_anthropic_tool_calls(content)
157 if not tool_calls:
158 tool_calls = None
160 choice = CompletionChoice(
161 index=0,
162 message=Message(role=MessageRole.ASSISTANT, content=text, tool_calls=tool_calls),
163 finish_reason=data.get("stop_reason", "end_turn"),
164 )
165 usage_data = data.get("usage", {})
166 tokens = CompletionUsage(
167 prompt_tokens=usage_data.get("input_tokens", 0),
168 completion_tokens=usage_data.get("output_tokens", 0),
169 total_tokens=usage_data.get("input_tokens", 0) + usage_data.get("output_tokens", 0),
170 )
171 if model in _PRICING:
172 in_price, out_price = _PRICING[model]
173 tokens.cost_usd = round(
174 tokens.prompt_tokens / 1_000_000 * in_price
175 + tokens.completion_tokens / 1_000_000 * out_price,
176 6,
177 )
178 return CompletionResult(
179 id=data.get("id", ""),
180 model=model,
181 choices=[choice],
182 usage=tokens,
183 )
186class AnthropicProvider(LLMProvider):
187 """Anthropic Claude Provider — 纯 httpx 实现，零 SDK 依赖。"""
189 def __init__(
190 self,
191 model: str = "claude-sonnet-4-20250514",
192 api_key: str = "",
193 base_url: str = "",
194 timeout: float = 120.0,
195 ):
196 super().__init__(
197 model=model,
198 api_key=api_key,
199 base_url=base_url or ANTHROPIC_API_BASE,
200 )
201 self._timeout = timeout
203 @property
204 def provider_name(self) -> str:
205 return "anthropic"
207 def _headers(self) -> dict[str, str]:
208 return {
209 "x-api-key": self.api_key,
210 "anthropic-version": ANTHROPIC_VERSION,
211 "content-type": "application/json",
212 }
214 def _build_body(
215 self,
216 messages: list[Message],
217 temperature: float,
218 max_tokens: int,
219 top_p: float,
220 stop: list[str] | None,
221 tools: list[Tool] | None,
222 tool_choice: str,
223 ) -> dict[str, Any]:
224 system, api_messages = _messages_to_anthropic(messages)
225 body: dict[str, Any] = {
226 "model": self.model,
227 "messages": api_messages,
228 "max_tokens": max_tokens,
229 "temperature": temperature,
230 }
231 if system:
232 body["system"] = system
233 if top_p < 1.0:
234 body["top_p"] = top_p
235 if stop:
236 body["stop_sequences"] = stop
237 if tools:
238 body["tools"] = _tools_to_anthropic(tools)
239 if tool_choice == "any":
240 body["tool_choice"] = {"type": "any"}
241 elif tool_choice == "auto":
242 body["tool_choice"] = {"type": "auto"}
243 return body
245 def chat(
246 self,
247 messages: list[Message],
248 *,
249 temperature: float = 0.7,
250 max_tokens: int = 4096,
251 top_p: float = 1.0,
252 stop: list[str] | None = None,
253 tools: list[Tool] | None = None,
254 tool_choice: str = "auto",
255 **kwargs: Any,
256 ) -> CompletionResult:
257 url = f"{self.base_url}/v1/messages"
258 body = self._build_body(messages, temperature, max_tokens, top_p, stop, tools, tool_choice)
259 with httpx.Client(timeout=self._timeout) as client:
260 resp = client.post(url, headers=self._headers(), json=body)
261 resp.raise_for_status()
262 return _build_result(resp.json(), self.model)
264 async def achat(
265 self,
266 messages: list[Message],
267 *,
268 temperature: float = 0.7,
269 max_tokens: int = 4096,
270 top_p: float = 1.0,
271 stop: list[str] | None = None,
272 tools: list[Tool] | None = None,
273 tool_choice: str = "auto",
274 **kwargs: Any,
275 ) -> CompletionResult:
276 url = f"{self.base_url}/v1/messages"
277 body = self._build_body(messages, temperature, max_tokens, top_p, stop, tools, tool_choice)
278 async with httpx.AsyncClient(timeout=self._timeout) as client:
279 resp = await client.post(url, headers=self._headers(), json=body)
280 resp.raise_for_status()
281 return _build_result(resp.json(), self.model)
283 def stream(
284 self,
285 messages: list[Message],
286 *,
287 temperature: float = 0.7,
288 max_tokens: int = 4096,
289 tools: list[Tool] | None = None,
290 **kwargs: Any,
291 ) -> Iterator[StreamChunk]:
292 url = f"{self.base_url}/v1/messages"
293 body = self._build_body(messages, temperature, max_tokens, 1.0, None, tools, "auto")
294 body["stream"] = True
295 with httpx.Client(timeout=self._timeout) as client:
296 with client.stream("POST", url, headers=self._headers(), json=body) as resp:
297 resp.raise_for_status()
298 for line in resp.iter_lines():
299 if not line.startswith("data: "):
300 continue
301 data_str = line[6:]
302 if data_str == "[DONE]":
303 break
304 try:
305 event = json.loads(data_str)
306 except json.JSONDecodeError:
307 continue
308 if event.get("type") == "content_block_delta":
309 delta = event.get("delta", {})
310 text = delta.get("text", "")
311 if text:
312 yield StreamChunk(content=text)
313 elif event.get("type") == "message_stop":
314 yield StreamChunk(finish_reason="end_turn")
316 async def astream( # pyright: ignore[reportIncompatibleMethodOverride]
317 self,
318 messages: list[Message],
319 *,
320 temperature: float = 0.7,
321 max_tokens: int = 4096,
322 tools: list[Tool] | None = None,
323 **kwargs: Any,
324 ):
325 url = f"{self.base_url}/v1/messages"
326 body = self._build_body(messages, temperature, max_tokens, 1.0, None, tools, "auto")
327 body["stream"] = True
328 async with httpx.AsyncClient(timeout=self._timeout) as client:
329 async with client.stream("POST", url, headers=self._headers(), json=body) as resp:
330 resp.raise_for_status()
331 async for line in resp.aiter_lines():
332 if not line.startswith("data: "):
333 continue
334 data_str = line[6:]
335 if data_str == "[DONE]":
336 break
337 try:
338 event = json.loads(data_str)
339 except json.JSONDecodeError:
340 continue
341 if event.get("type") == "content_block_delta":
342 delta = event.get("delta", {})
343 text = delta.get("text", "")
344 if text:
345 yield StreamChunk(content=text)
346 elif event.get("type") == "message_stop":
347 yield StreamChunk(finish_reason="end_turn")