chore: refactor into a base with utility functions

Author: njbrake

src/any_llm/providers/anthropic/anthropic.py

@@ -1,6 +1,5 @@
-import os
 import json
-from typing import Any
+from typing import Any, cast
 
 try:
     from anthropic import Anthropic
@@ -11,205 +10,180 @@
 
 from openai.types.chat.chat_completion import ChatCompletion, Choice
 from openai.types.completion_usage import CompletionUsage
-from openai.types.chat.chat_completion_message import ChatCompletionMessage
-from openai.types.chat.chat_completion_message_tool_call import ChatCompletionMessageToolCall, Function
-from any_llm.provider import Provider, ApiConfig
-from any_llm.exceptions import MissingApiKeyError
 
-# Define a constant for the default max_tokens value
+from any_llm.provider import ApiConfig
+from any_llm.providers.base_framework import (
+    BaseCustomProvider,
+    create_openai_tool_call,
+    create_openai_message,
+    create_openai_completion,
+    convert_openai_tools_to_generic,
+    extract_system_message,
+    remove_unsupported_params,
+)
+
 DEFAULT_MAX_TOKENS = 4096
 
 
-def _convert_kwargs(kwargs: dict[str, Any]) -> dict[str, Any]:
-    """Format the kwargs for Anthropic."""
-    kwargs = kwargs.copy()
-    kwargs.setdefault("max_tokens", DEFAULT_MAX_TOKENS)
-
-    # Convert tools if present
-    if "tools" in kwargs:
-        kwargs["tools"] = _convert_tool_spec(kwargs["tools"])
-
-    # Handle parallel_tool_calls parameter
-    if "parallel_tool_calls" in kwargs:
-        parallel_tool_calls = kwargs.pop("parallel_tool_calls")
-        # If parallel_tool_calls is False, set disable_parallel_tool_use to True
-        if parallel_tool_calls is False:
-            tool_choice = {"type": kwargs.get("tool_choice", "any"), "disable_parallel_tool_use": True}
-            kwargs["tool_choice"] = tool_choice
-        # If parallel_tool_calls is True or not specified, don't set disable_parallel_tool_use
-        # (Anthropic defaults to parallel tool use enabled)
-
-    if "response_format" in kwargs:
-        error_msg = (
-            "response_format is not supported for Anthropic, see their documentation "
-            "for tips on how to achieve structured output: "
-            "https://docs.anthropic.com/en/docs/test-and-evaluate/strengthen-guardrails/increase-consistency#example-standardizing-customer-feedback"
-        )
-        raise ValueError(error_msg)
+class AnthropicProvider(BaseCustomProvider):
+    """
+    Anthropic Provider using enhanced BaseCustomProvider framework.
 
-    return kwargs
+    Handles conversion between OpenAI format and Anthropic's native format.
+    """
 
+    PROVIDER_NAME = "Anthropic"
+    ENV_API_KEY_NAME = "ANTHROPIC_API_KEY"
 
-def _convert_tool_spec(openai_tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    """Convert OpenAI tool specification to Anthropic format."""
-    anthropic_tools = []
+    def _initialize_client(self, config: ApiConfig) -> None:
+        """Initialize the Anthropic client."""
+        self.client = Anthropic(api_key=config.api_key, base_url=config.api_base)
 
-    for tool in openai_tools:
-        if tool.get("type") != "function":
-            continue
+    def _convert_kwargs(self, kwargs: dict[str, Any]) -> dict[str, Any]:
+        """Convert kwargs to Anthropic format."""
+        kwargs = kwargs.copy()
+        kwargs.setdefault("max_tokens", DEFAULT_MAX_TOKENS)
+
+        # Remove unsupported parameters
+        kwargs = remove_unsupported_params(kwargs, ["response_format"])
+
+        # Convert tools if present
+        if "tools" in kwargs:
+            kwargs["tools"] = self._convert_tool_spec(kwargs["tools"])
+
+        # Handle parallel_tool_calls
+        if "parallel_tool_calls" in kwargs:
+            parallel_tool_calls = kwargs.pop("parallel_tool_calls")
+            if parallel_tool_calls is False:
+                tool_choice = {"type": kwargs.get("tool_choice", "any"), "disable_parallel_tool_use": True}
+                kwargs["tool_choice"] = tool_choice
+
+        return kwargs
+
+    def _convert_messages(self, messages: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:
+        """Convert messages to Anthropic format, extracting system message."""
+        # Extract system message using the utility
+        system_message, remaining_messages = extract_system_message(messages)
+
+        converted_messages = []
+        for message in remaining_messages:
+            if message["role"] == "tool":
+                converted_message = {
+                    "role": "user",
+                    "content": [
+                        {
+                            "type": "tool_result",
+                            "tool_use_id": message["tool_call_id"],
+                            "content": message["content"],
+                        }
+                    ],
+                }
+                converted_messages.append(converted_message)
+            elif message["role"] == "assistant" and "tool_calls" in message:
+                message_content = []
+                if message.get("content"):
+                    message_content.append({"type": "text", "text": message["content"]})
+
+                for tool_call in message.get("tool_calls") or []:
+                    message_content.append(
+                        {
+                            "type": "tool_use",
+                            "id": tool_call["id"],
+                            "name": tool_call["function"]["name"],
+                            "input": json.loads(tool_call["function"]["arguments"]),
+                        }
+                    )
+
+                converted_message = {"role": "assistant", "content": message_content}
+                converted_messages.append(converted_message)
+            else:
+                converted_message = {"role": message["role"], "content": message["content"]}
+                converted_messages.append(converted_message)
+
+        return system_message, converted_messages
+
+    def _make_api_call(self, model: str, messages: tuple[str, list[dict[str, Any]]], **kwargs: Any) -> Message:
+        """Make the API call to Anthropic."""
+        system_message, converted_messages = messages
+
+        return self.client.messages.create(
+            model=model,
+            system=system_message,
+            messages=converted_messages,  # type: ignore[arg-type]
+            **kwargs,
+        )
 
-        function = tool["function"]
-        anthropic_tool = {
-            "name": function["name"],
-            "description": function["description"],
-            "input_schema": {
-                "type": "object",
-                "properties": function["parameters"]["properties"],
-                "required": function["parameters"].get("required", []),
-            },
+    def _convert_response(self, response: Message) -> ChatCompletion:
+        """Convert Anthropic response to OpenAI format."""
+        finish_reason_mapping = {
+            "end_turn": "stop",
+            "max_tokens": "length",
+            "tool_use": "tool_calls",
         }
-        anthropic_tools.append(anthropic_tool)
-
-    return anthropic_tools
-
-
-def _convert_messages(messages: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:
-    """Convert messages to Anthropic format, extracting system message."""
-    system_message = ""
-    converted_messages = []
-
-    for message in messages:
-        if message["role"] == "system":
-            system_message = message["content"]
-            continue
-        elif message["role"] == "tool":
-            # Convert tool message to Anthropic format
-            converted_message = {
-                "role": "user",
-                "content": [
-                    {
-                        "type": "tool_result",
-                        "tool_use_id": message["tool_call_id"],
-                        "content": message["content"],
-                    }
-                ],
-            }
-            converted_messages.append(converted_message)
-        elif message["role"] == "assistant" and "tool_calls" in message:
-            # Convert assistant message with tool calls
-            message_content = []
-            if message.get("content"):
-                message_content.append({"type": "text", "text": message["content"]})
-
-            for tool_call in message.get("tool_calls") or []:
-                message_content.append(
-                    {
-                        "type": "tool_use",
-                        "id": tool_call["id"],
-                        "name": tool_call["function"]["name"],
-                        "input": json.loads(tool_call["function"]["arguments"]),
-                    }
-                )
 
-            converted_message = {"role": "assistant", "content": message_content}
-            converted_messages.append(converted_message)
-        else:
-            # Regular message
-            converted_message = {"role": message["role"], "content": message["content"]}
-            converted_messages.append(converted_message)
-
-    return system_message, converted_messages
-
-
-def _convert_response(response: Message) -> ChatCompletion:
-    """Convert Anthropic response directly to OpenAI ChatCompletion format."""
-    # Finish reason mapping
-    finish_reason_mapping = {
-        "end_turn": "stop",
-        "max_tokens": "length",
-        "tool_use": "tool_calls",
-    }
-
-    # Process content blocks
-    tool_calls = []
-    content = ""
-
-    for content_block in response.content:
-        if content_block.type == "text":
-            content = content_block.text
-        elif content_block.type == "tool_use":
-            tool_calls.append(
-                ChatCompletionMessageToolCall(
-                    id=content_block.id,
-                    type="function",
-                    function=Function(name=content_block.name, arguments=json.dumps(content_block.input)),
+        # Process content blocks
+        tool_calls = []
+        content = ""
+
+        for content_block in response.content:
+            if content_block.type == "text":
+                content = content_block.text
+            elif content_block.type == "tool_use":
+                tool_calls.append(
+                    create_openai_tool_call(
+                        tool_call_id=content_block.id,
+                        name=content_block.name,
+                        arguments=json.dumps(content_block.input),
+                    )
                 )
-            )
-
-    # Create the message
-    message = ChatCompletionMessage(
-        content=content or None,
-        role="assistant",
-        tool_calls=tool_calls if tool_calls else None,
-    )
-
-    # Create the choice
-    if not response.stop_reason:
-        response.stop_reason = "end_turn"
-    mapped_finish_reason = finish_reason_mapping.get(response.stop_reason, "stop")
-    choice = Choice(
-        finish_reason=mapped_finish_reason,  # type: ignore
-        index=0,
-        message=message,
-    )
-
-    # Create usage information
-    usage = CompletionUsage(
-        completion_tokens=response.usage.output_tokens,
-        prompt_tokens=response.usage.input_tokens,
-        total_tokens=response.usage.input_tokens + response.usage.output_tokens,
-    )
-
-    # Build the final ChatCompletion object
-    return ChatCompletion(
-        id=response.id,
-        model=response.model,
-        object="chat.completion",
-        created=int(response.created_at.timestamp()) if hasattr(response, "created_at") else 0,
-        choices=[choice],
-        usage=usage,
-    )
-
-
-class AnthropicProvider(Provider):
-    def __init__(self, config: ApiConfig) -> None:
-        """Initialize Anthropic provider."""
-        if not config.api_key:
-            config.api_key = os.getenv("ANTHROPIC_API_KEY")
-        if not config.api_key:
-            raise MissingApiKeyError(
-                "Anthropic",
-                "ANTHROPIC_API_KEY",
-            )
-        self.client = Anthropic(api_key=config.api_key, base_url=config.api_base)
 
-    def completion(
-        self,
-        model: str,
-        messages: list[dict[str, Any]],
-        **kwargs: Any,
-    ) -> ChatCompletion:
-        """Create a chat completion using Anthropic."""
-        kwargs = _convert_kwargs(kwargs)
-        system_message, converted_messages = _convert_messages(messages)
-
-        # Make the request to Anthropic
-        response = self.client.messages.create(
-            model=model,
-            system=system_message,
-            messages=converted_messages,  # type: ignore
-            **kwargs,
+        # Create the message
+        message = create_openai_message(
+            role="assistant",
+            content=content or None,
+            tool_calls=tool_calls if tool_calls else None,
+        )
+
+        # Create the choice
+        mapped_finish_reason = finish_reason_mapping.get(response.stop_reason or "end_turn", "stop")
+        choice = Choice(
+            finish_reason=cast(Any, mapped_finish_reason),
+            index=0,
+            message=message,
+        )
+
+        # Create usage information
+        usage = CompletionUsage(
+            completion_tokens=response.usage.output_tokens,
+            prompt_tokens=response.usage.input_tokens,
+            total_tokens=response.usage.input_tokens + response.usage.output_tokens,
+        )
+
+        return create_openai_completion(
+            id=response.id,
+            model=response.model,
+            choices=[choice],
+            usage=usage,
+            created=int(response.created_at.timestamp()) if hasattr(response, "created_at") else 0,
         )
 
-        # Convert to OpenAI format
-        return _convert_response(response)
+    def _convert_tool_spec(self, openai_tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        """Convert OpenAI tool specification to Anthropic format."""
+        # Use the generic utility first
+        generic_tools = convert_openai_tools_to_generic(openai_tools)
+
+        # Convert to Anthropic-specific format
+        anthropic_tools = []
+        for tool in generic_tools:
+            anthropic_tool = {
+                "name": tool["name"],
+                "description": tool["description"],
+                "input_schema": {
+                    "type": "object",
+                    "properties": tool["parameters"]["properties"],
+                    "required": tool["parameters"].get("required", []),
+                },
+            }
+            anthropic_tools.append(anthropic_tool)
+
+        return anthropic_tools