Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/clients/anthropic.py: 15%
169 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Anthropic Claude LLM client implementation.
3Production-ready implementation with Anthropic API integration.
4"""
6from __future__ import annotations
8from collections.abc import AsyncGenerator, AsyncIterator
9from datetime import UTC, datetime
10from typing import Any
12from lexigram.ai.llm.clients._message_utils import serialize_content_for_anthropic
13from lexigram.ai.llm.clients._tools_utils import parse_json_arguments
14from lexigram.ai.llm.clients.base import AbstractLLMClient
15from lexigram.ai.llm.config import ClientConfig
16from lexigram.ai.llm.exceptions import (
17 LLMAuthenticationError,
18 LLMContentFilterError,
19 LLMError,
20 LLMModelNotFoundError,
21 LLMQuotaExceededError,
22 LLMRateLimitError,
23)
24from lexigram.ai.llm.types import (
25 AIError,
26 ChatMessage,
27 Completion,
28 FunctionCall,
29 Role,
30 StreamChunk,
31 ThinkingResult,
32 TokenUsage,
33 ToolCall,
34)
35from lexigram.contracts.core import HealthCheckResult, HealthStatus
36from lexigram.result import Err, Ok, Result
37from lexigram.serialization import dumps_str
40class AnthropicClient(AbstractLLMClient):
41 """Anthropic Claude LLM client implementation.
43 Conforms to: :class:`~lexigram.contracts.ai.LLMClientProtocol` protocol via structural typing.
45 Supports Claude 3 (Opus, Sonnet, Haiku) models with:
46 - Streaming responses
47 - Tool calling
48 - Vision capabilities
49 - Automatic retry and error handling
51 Example:
52 >>> from lexigram.ai import ClientConfig
53 >>> config = ClientConfig(provider="anthropic", model="claude-3-sonnet-20240229")
54 >>> client = AnthropicClient(config)
55 >>> completion = await client.complete([
56 ... ChatMessage(role="user", content="Hello!")
57 ... ])
58 """
60 def __init__(self, config: ClientConfig):
61 """Initialize Anthropic client.
63 Args:
64 config: LLM configuration
66 Raises:
67 ImportError: If anthropic package is not installed
68 """
69 super().__init__(config=config)
71 try:
72 from anthropic import AsyncAnthropic
73 except ImportError as e:
74 raise ImportError(
75 "Anthropic client requires 'anthropic' package. "
76 "Install with: pip install lexigram-ai-llm[anthropic]",
77 ) from e
79 api_key = config.api_key.get_secret_value() if config.api_key else None
80 self.client = AsyncAnthropic(
81 api_key=api_key,
82 base_url=config.api_base,
83 timeout=config.timeout,
84 )
86 async def _do_complete(
87 self,
88 messages: list[ChatMessage],
89 **kwargs: Any,
90 ) -> Result[Completion, LLMError]:
91 """Generate completion from messages.
93 Args:
94 messages: Chat messages
95 **kwargs: Additional Anthropic API parameters
97 Returns:
98 ``Ok(Completion)`` on success. ``Err(LLMError)`` for recoverable
99 failures (rate limit, quota, content filter, model not found).
101 Raises:
102 LLMAuthenticationError: If API key is invalid.
103 AIError: For unexpected infrastructure failures.
104 """
105 try:
106 # Extract system message if present
107 system_msg = None
108 conv_messages = []
109 for msg in messages:
110 if msg.role == Role.SYSTEM:
111 system_msg = msg.content
112 else:
113 conv_messages.append(self._convert_message(msg))
115 # Build thinking param; suppress temperature (incompatible with thinking)
116 params = {
117 "model": kwargs.pop("model", self.config.model),
118 "messages": conv_messages,
119 "max_tokens": kwargs.pop("max_tokens", self.config.max_tokens or 1024),
120 **kwargs,
121 }
122 self._apply_thinking(params)
123 if "thinking" not in params:
124 params["temperature"] = kwargs.pop(
125 "temperature", self.config.temperature
126 )
128 tools = params.pop("tools", None)
129 if tools:
130 converted_tools = [_tool_to_anthropic(t) for t in tools]
131 params["tools"] = [t for t in converted_tools if t.get("name")]
133 if system_msg:
134 params["system"] = system_msg
136 # Make API call
137 response = await self.client.messages.create(**params)
139 # Parse content blocks: separate thinking blocks from text blocks
140 content = ""
141 thinking: ThinkingResult | None = None
142 thinking_parts: list[str] = []
143 thinking_signature: str | None = None
144 tool_calls: list[ToolCall] = []
145 for block in response.content:
146 block_type = getattr(block, "type", None)
147 if block_type == "thinking":
148 thinking_parts.append(getattr(block, "thinking", "") or "")
149 thinking_signature = getattr(block, "signature", None)
150 elif block_type == "text" or (
151 block_type is None and hasattr(block, "text")
152 ):
153 content = getattr(block, "text", "")
154 elif block_type == "tool_use":
155 tool_calls.append(
156 ToolCall(
157 id=getattr(block, "id", ""),
158 type="function",
159 function=FunctionCall(
160 name=getattr(block, "name", ""),
161 arguments=dumps_str(getattr(block, "input", {})),
162 ),
163 )
164 )
165 if thinking_parts:
166 thinking = ThinkingResult(
167 content="".join(thinking_parts),
168 signature=thinking_signature,
169 )
171 return Ok(
172 Completion(
173 content=content,
174 model=response.model,
175 finish_reason=response.stop_reason,
176 thinking=thinking,
177 tool_calls=tool_calls or None,
178 usage=TokenUsage(
179 prompt_tokens=response.usage.input_tokens,
180 completion_tokens=response.usage.output_tokens,
181 total_tokens=response.usage.input_tokens
182 + response.usage.output_tokens,
183 ),
184 metadata={
185 "id": response.id,
186 "type": response.type,
187 },
188 timestamp=datetime.now(UTC),
189 )
190 )
192 except (ValueError, RuntimeError, OSError, ConnectionError) as e:
193 return self._handle_error_as_result(e)
195 async def _do_stream_chat(
196 self,
197 messages: list[ChatMessage],
198 **kwargs: Any,
199 ) -> Result[AsyncIterator[StreamChunk], LLMError]:
200 """Start a streaming completion.
202 Args:
203 messages: Chat messages
204 **kwargs: Additional Anthropic API parameters
206 Returns:
207 ``Ok(AsyncIterator[StreamChunk])`` on successful setup.
208 ``Err(LLMError)`` for recoverable connection failures.
209 """
210 try:
211 # Extract system message if present
212 system_msg = None
213 conv_messages = []
214 for msg in messages:
215 if msg.role == Role.SYSTEM:
216 system_msg = msg.content
217 else:
218 conv_messages.append(self._convert_message(msg))
220 # Build thinking param; suppress temperature (incompatible with thinking)
221 params = {
222 "model": kwargs.pop("model", self.config.model),
223 "messages": conv_messages,
224 "max_tokens": kwargs.pop("max_tokens", self.config.max_tokens or 1024),
225 **kwargs,
226 }
227 self._apply_thinking(params)
228 if "thinking" not in params:
229 params["temperature"] = kwargs.pop(
230 "temperature", self.config.temperature
231 )
233 if system_msg:
234 params["system"] = system_msg
236 import asyncio
238 stream_ctx = self.client.messages.stream(**params)
239 if asyncio.iscoroutine(stream_ctx):
240 stream_ctx = await stream_ctx
242 return Ok(self._stream_impl(stream_ctx))
244 except (ValueError, RuntimeError, OSError, ConnectionError) as e:
245 return self._handle_error_as_result(e)
247 async def _stream_impl(self, stream_ctx: Any) -> AsyncGenerator[StreamChunk, None]:
248 """Yield StreamChunk objects from an established Anthropic stream context.
250 When ``thinking`` config is set on the config, uses the raw event
251 stream to capture ``thinking_delta`` events alongside ``text_delta``
252 events. Otherwise falls back to the simpler ``text_stream`` path.
253 """
254 index = 0
255 use_thinking = self.config.thinking is not None
256 try:
257 async with stream_ctx as stream:
258 if use_thinking:
259 # Raw event iteration to capture thinking and text deltas
260 async for event in stream:
261 event_type = getattr(event, "type", None)
262 if event_type != "content_block_delta":
263 continue
264 delta = event.delta
265 delta_type = getattr(delta, "type", None)
266 if delta_type == "thinking_delta":
267 yield StreamChunk(
268 thinking_delta=getattr(delta, "thinking", None),
269 is_thinking=True,
270 model=self.config.model,
271 finish_reason=None,
272 index=index,
273 )
274 index += 1
275 elif delta_type == "text_delta":
276 yield StreamChunk(
277 delta=getattr(delta, "text", None),
278 is_thinking=False,
279 model=self.config.model,
280 finish_reason=None,
281 index=index,
282 )
283 index += 1
284 else:
285 iter_source = getattr(stream, "text_stream", stream)
286 async for raw_text in iter_source:
287 text = (
288 raw_text.decode()
289 if isinstance(raw_text, bytes)
290 else raw_text
291 )
292 yield StreamChunk(
293 delta=text,
294 model=self.config.model,
295 finish_reason=None,
296 index=index,
297 )
298 index += 1
299 except (ValueError, RuntimeError, OSError, ConnectionError) as e:
300 err = self._handle_error_as_result(e)
301 raise err if isinstance(err, BaseException) else LLMError(str(err)) from e
303 async def _do_chat(
304 self,
305 messages: list[ChatMessage],
306 tools: list[ToolCall] | None = None,
307 **kwargs: Any,
308 ) -> Result[Completion, LLMError]:
309 """Generate completion with Anthropic tool/function calling.
311 Tool calling is handled by :meth:`_do_complete` via
312 ``complete(..., tools=...)``; this method forwards the tool
313 descriptors to keep the ``chat`` code path consistent.
315 Args:
316 messages: Chat messages.
317 tools: Optional tool/function descriptors.
318 **kwargs: Additional Anthropic API parameters.
320 Returns:
321 ``Ok(Completion)`` on success. ``Err(LLMError)`` for recoverable
322 failures.
323 """
324 return await self._do_complete(messages, tools=tools, **kwargs)
326 def _apply_thinking(self, params: dict[str, Any]) -> None:
327 """Inject Anthropic extended-thinking parameters into the API payload.
329 When ``suppress`` is set, returns immediately — Anthropic's default is
330 no thinking, so suppression simply means not injecting the parameter.
332 Args:
333 params: Mutable API payload dict.
334 """
335 if self.config.thinking is None:
336 return
337 if self.config.thinking.suppress:
338 return
339 params["thinking"] = {
340 "type": "enabled",
341 "budget_tokens": self.config.thinking.budget_tokens,
342 }
343 params.pop("temperature", None)
345 def _convert_message(self, msg: ChatMessage) -> dict[str, Any]:
346 """Convert ChatMessage to Anthropic format.
348 Uses ``serialize_content_for_anthropic`` to convert multimodal message
349 content to Anthropic-compatible block format. When ``thinking_blocks``
350 is populated (multi-turn with extended thinking), prepends thinking blocks
351 to the serialized content.
353 Tool messages (``Role.TOOL``) become user turns with a ``tool_result``
354 block; assistant turns that requested tools gain ``tool_use`` blocks so
355 tool conversations round-trip correctly.
357 Args:
358 msg: ChatMessage to convert
360 Returns:
361 Anthropic message dict with ``role`` and ``content`` keys
362 """
363 if msg.role == Role.TOOL:
364 return {
365 "role": "user",
366 "content": [
367 {
368 "type": "tool_result",
369 "tool_use_id": msg.tool_call_id or "",
370 "content": _tool_result_text(msg.content),
371 }
372 ],
373 }
375 role = "user" if msg.role == Role.USER else "assistant"
376 tool_use_blocks: list[dict[str, Any]] = []
377 for call in msg.tool_calls or []:
378 if call.function is None:
379 continue
380 tool_use_blocks.append(
381 {
382 "type": "tool_use",
383 "id": call.id,
384 "name": call.function.name,
385 "input": parse_json_arguments(call.function.arguments),
386 }
387 )
389 if msg.thinking_blocks:
390 # Only include serialized_content if it's non-trivially empty
391 # (empty string content produces [{"type": "text", "text": ""}] which Anthropic rejects)
392 if msg.content in ("", []):
393 content: list[dict[str, Any]] = list(msg.thinking_blocks)
394 else:
395 content = list(msg.thinking_blocks) + serialize_content_for_anthropic(
396 msg.content
397 )
398 if tool_use_blocks:
399 content.extend(tool_use_blocks)
400 return {"role": role, "content": content}
402 if msg.content in ("", []):
403 return {
404 "role": role,
405 "content": tool_use_blocks,
406 }
408 return {
409 "role": role,
410 "content": serialize_content_for_anthropic(msg.content) + tool_use_blocks,
411 }
413 def _handle_error_as_result(self, error: Exception) -> Result[Any, LLMError]:
414 """Map a caught exception to ``Err`` (recoverable) or re-raise (infra)."""
415 err_str = str(error).lower()
416 if "authentication" in err_str or "api key" in err_str:
417 raise LLMAuthenticationError(
418 f"Anthropic authentication failed: {error}"
419 ) from error
420 if "rate limit" in err_str:
421 return Err(LLMRateLimitError(f"Anthropic rate limit exceeded: {error}"))
422 if "quota" in err_str or "billing" in err_str or "credit" in err_str:
423 return Err(LLMQuotaExceededError(f"Anthropic quota exceeded: {error}"))
424 if "content" in err_str and ("filter" in err_str or "policy" in err_str):
425 return Err(LLMContentFilterError(f"Anthropic content filter: {error}"))
426 if "model" in err_str and (
427 "not found" in err_str or "does not exist" in err_str
428 ):
429 return Err(LLMModelNotFoundError(f"Anthropic model not found: {error}"))
430 raise AIError(f"Anthropic infrastructure error: {error}") from error
432 async def close(self) -> None:
433 """Close the Anthropic client."""
434 if getattr(self, "client", None) and not self._closed:
435 await self.client.close()
436 await super().close()
438 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
439 """Perform health check.
441 Returns:
442 Structured health check result.
443 """
444 if self._closed:
445 return HealthCheckResult(
446 component="llm.anthropic",
447 status=HealthStatus.UNHEALTHY,
448 error="Client is closed",
449 )
451 return HealthCheckResult(
452 component="llm.anthropic",
453 status=HealthStatus.HEALTHY,
454 details={
455 "provider": "anthropic",
456 "model": self.config.model,
457 },
458 )
461# ──────────────────────────────────────────────────────────────────────
462# Private helpers
463# ──────────────────────────────────────────────────────────────────────
466def _tool_to_anthropic(tool: Any) -> dict[str, Any]:
467 """Convert a tool descriptor to Anthropic ``tool_use`` format.
469 Supports objects that expose a ``__tool_schema__`` class attribute
470 (Lexigram tool convention) as well as plain dicts in OpenAI tool format.
472 Args:
473 tool: A class with ``__tool_schema__``, or a dict with a ``function``
474 key in OpenAI tool format.
476 Returns:
477 Anthropic tool dict with ``name``, ``description``, and
478 ``input_schema`` keys.
479 """
480 if hasattr(tool, "__tool_schema__"):
481 schema: dict[str, Any] = tool.__tool_schema__
482 return {
483 "name": schema["name"],
484 "description": schema.get("description", ""),
485 "input_schema": schema.get(
486 "parameters", {"type": "object", "properties": {}}
487 ),
488 }
489 if isinstance(tool, dict):
490 func = tool.get("function", tool)
491 return {
492 "name": func.get("name", ""),
493 "description": func.get("description", ""),
494 "input_schema": func.get(
495 "parameters", {"type": "object", "properties": {}}
496 ),
497 }
498 return {
499 "name": getattr(tool, "name", str(tool)),
500 "description": getattr(tool, "description", ""),
501 "input_schema": getattr(tool, "parameters", None)
502 or {"type": "object", "properties": {}},
503 }
506def _tool_result_text(content: Any) -> str:
507 """Extract plain text from tool-result content.
509 Args:
510 content: Message content (str or list of content parts).
512 Returns:
513 Joined text string.
514 """
515 if isinstance(content, str):
516 return content
517 blocks = serialize_content_for_anthropic(content)
518 return " ".join(str(b.get("text", "")) for b in blocks if b.get("type") == "text")