fix(llma): extract converters for providers

Author: Radu-Raicea

posthog/ai/anthropic/__init__.py

@@ -6,6 +6,12 @@
     AsyncAnthropicBedrock,
     AsyncAnthropicVertex,
 )
+from .anthropic_converter import (
+    format_anthropic_response,
+    format_anthropic_input,
+    extract_anthropic_tools,
+    format_anthropic_streaming_content,
+)
 
 __all__ = [
     "Anthropic",
@@ -14,4 +20,8 @@
     "AsyncAnthropicBedrock",
     "AnthropicVertex",
     "AsyncAnthropicVertex",
+    "format_anthropic_response",
+    "format_anthropic_input",
+    "extract_anthropic_tools",
+    "format_anthropic_streaming_content",
 ]

posthog/ai/anthropic/anthropic_converter.py

@@ -0,0 +1,150 @@
+"""
+Anthropic-specific conversion utilities.
+
+This module handles the conversion of Anthropic API responses and inputs
+into standardized formats for PostHog tracking.
+"""
+
+from typing import Any, Dict, List, Optional
+
+from posthog.ai.types import (
+    FormattedContentItem,
+    FormattedFunctionCall,
+    FormattedMessage,
+    FormattedTextContent,
+)
+
+
+def format_anthropic_response(response: Any) -> List[FormattedMessage]:
+    """
+    Format an Anthropic response into standardized message format.
+    
+    Args:
+        response: The response object from Anthropic API
+        
+    Returns:
+        List of formatted messages with role and content
+    """
+    output = []
+    
+    if response is None:
+        return output
+    
+    content: List[FormattedContentItem] = []
+    
+    # Process content blocks from the response
+    if hasattr(response, "content"):
+        for choice in response.content:
+            if (
+                hasattr(choice, "type")
+                and choice.type == "text"
+                and hasattr(choice, "text")
+                and choice.text
+            ):
+                text_content: FormattedTextContent = {
+                    "type": "text",
+                    "text": choice.text
+                }
+                content.append(text_content)
+            elif (
+                hasattr(choice, "type")
+                and choice.type == "tool_use"
+                and hasattr(choice, "name")
+                and hasattr(choice, "id")
+            ):
+                function_call: FormattedFunctionCall = {
+                    "type": "function",
+                    "id": choice.id,
+                    "function": {
+                        "name": choice.name,
+                        "arguments": getattr(choice, "input", {}),
+                    }
+                }
+                content.append(function_call)
+    
+    if content:
+        message: FormattedMessage = {
+            "role": "assistant",
+            "content": content,
+        }
+        output.append(message)
+    
+    return output
+
+
+def format_anthropic_input(messages: List[Dict[str, Any]], system: Optional[str] = None) -> List[FormattedMessage]:
+    """
+    Format Anthropic input messages with optional system prompt.
+    
+    Args:
+        messages: List of message dictionaries
+        system: Optional system prompt to prepend
+        
+    Returns:
+        List of formatted messages
+    """
+    formatted_messages: List[FormattedMessage] = []
+    
+    # Add system message if provided
+    if system is not None:
+        formatted_messages.append({
+            "role": "system",
+            "content": system
+        })
+    
+    # Add user messages
+    if messages:
+        for msg in messages:
+            # Messages are already in the correct format, just ensure type safety
+            formatted_msg: FormattedMessage = {
+                "role": msg.get("role", "user"),
+                "content": msg.get("content", "")
+            }
+            formatted_messages.append(formatted_msg)
+    
+    return formatted_messages
+
+
+def extract_anthropic_tools(kwargs: Dict[str, Any]) -> Optional[Any]:
+    """
+    Extract tool definitions from Anthropic API kwargs.
+    
+    Args:
+        kwargs: Keyword arguments passed to Anthropic API
+        
+    Returns:
+        Tool definitions if present, None otherwise
+    """
+    return kwargs.get("tools", None)
+
+
+def format_anthropic_streaming_content(content_blocks: List[Dict[str, Any]]) -> List[FormattedContentItem]:
+    """
+    Format content blocks from Anthropic streaming response.
+    
+    Used by streaming handlers to format accumulated content blocks.
+    
+    Args:
+        content_blocks: List of content block dictionaries from streaming
+        
+    Returns:
+        List of formatted content items
+    """
+    formatted: List[FormattedContentItem] = []
+    
+    for block in content_blocks:
+        if block.get("type") == "text":
+            text_content: FormattedTextContent = {
+                "type": "text",
+                "text": block.get("text", "")
+            }
+            formatted.append(text_content)
+        elif block.get("type") == "function":
+            function_call: FormattedFunctionCall = {
+                "type": "function",
+                "id": block.get("id"),
+                "function": block.get("function", {})
+            }
+            formatted.append(function_call)
+    
+    return formatted

posthog/ai/gemini/__init__.py

@@ -1,5 +1,9 @@
 from .gemini import Client
-from .gemini_converter import format_gemini_input, FormattedMessage
+from .gemini_converter import (
+    format_gemini_input,
+    format_gemini_response,
+    extract_gemini_tools,
+)
 
 
 # Create a genai-like module for perfect drop-in replacement
@@ -9,4 +13,10 @@ class _GenAI:
 
 genai = _GenAI()
 
-__all__ = ["Client", "genai", "format_gemini_input", "FormattedMessage"]
+__all__ = [
+    "Client",
+    "genai",
+    "format_gemini_input",
+    "format_gemini_response",
+    "extract_gemini_tools",
+]

posthog/ai/gemini/gemini_converter.py

@@ -1,11 +1,18 @@
 """
-Gemini input format converter module.
+Gemini-specific conversion utilities.
 
-This module handles the conversion of various Gemini input formats into a standardized
-format for PostHog tracking. It eliminates code duplication between gemini.py and utils.py.
+This module handles the conversion of Gemini API responses and inputs
+into standardized formats for PostHog tracking.
 """
 
-from typing import Any, Dict, List, TypedDict, Union
+from typing import Any, Dict, List, Optional, TypedDict, Union
+
+from posthog.ai.types import (
+    FormattedContentItem,
+    FormattedFunctionCall,
+    FormattedMessage,
+    FormattedTextContent,
+)
 
 
 class GeminiPart(TypedDict, total=False):
@@ -21,12 +28,6 @@ class GeminiMessage(TypedDict, total=False):
     text: str
 
 
-class FormattedMessage(TypedDict):
-    """Standardized message format for PostHog tracking."""
-    role: str
-    content: str
-
-
 def _extract_text_from_parts(parts: List[Any]) -> str:
     """
     Extract and concatenate text from a parts array.
@@ -131,6 +132,85 @@ def _format_object_message(item: Any) -> FormattedMessage:
     return {"role": "user", "content": str(item)}
 
 
+def format_gemini_response(response: Any) -> List[FormattedMessage]:
+    """
+    Format a Gemini response into standardized message format.
+    
+    Args:
+        response: The response object from Gemini API
+        
+    Returns:
+        List of formatted messages with role and content
+    """
+    output = []
+    
+    if response is None:
+        return output
+    
+    if hasattr(response, "candidates") and response.candidates:
+        for candidate in response.candidates:
+            if hasattr(candidate, "content") and candidate.content:
+                content: List[FormattedContentItem] = []
+                
+                if hasattr(candidate.content, "parts") and candidate.content.parts:
+                    for part in candidate.content.parts:
+                        if hasattr(part, "text") and part.text:
+                            text_content: FormattedTextContent = {
+                                "type": "text",
+                                "text": part.text
+                            }
+                            content.append(text_content)
+                        elif hasattr(part, "function_call") and part.function_call:
+                            function_call = part.function_call
+                            func_content: FormattedFunctionCall = {
+                                "type": "function",
+                                "function": {
+                                    "name": function_call.name,
+                                    "arguments": function_call.args,
+                                }
+                            }
+                            content.append(func_content)
+                
+                if content:
+                    message: FormattedMessage = {
+                        "role": "assistant",
+                        "content": content,
+                    }
+                    output.append(message)
+            
+            elif hasattr(candidate, "text") and candidate.text:
+                message: FormattedMessage = {
+                    "role": "assistant",
+                    "content": [{"type": "text", "text": candidate.text}],
+                }
+                output.append(message)
+    
+    elif hasattr(response, "text") and response.text:
+        message: FormattedMessage = {
+            "role": "assistant",
+            "content": [{"type": "text", "text": response.text}],
+        }
+        output.append(message)
+    
+    return output
+
+
+def extract_gemini_tools(kwargs: Dict[str, Any]) -> Optional[Any]:
+    """
+    Extract tool definitions from Gemini API kwargs.
+    
+    Args:
+        kwargs: Keyword arguments passed to Gemini API
+        
+    Returns:
+        Tool definitions if present, None otherwise
+    """
+    if "config" in kwargs and hasattr(kwargs["config"], "tools"):
+        return kwargs["config"].tools
+    
+    return None
+
+
 def format_gemini_input(contents: Any) -> List[FormattedMessage]:
     """
     Format Gemini input contents into standardized message format for PostHog tracking.

posthog/ai/openai/__init__.py

@@ -1,5 +1,20 @@
 from .openai import OpenAI
 from .openai_async import AsyncOpenAI
 from .openai_providers import AsyncAzureOpenAI, AzureOpenAI
+from .openai_converter import (
+    format_openai_response,
+    format_openai_input,
+    extract_openai_tools,
+    format_openai_streaming_content,
+)
 
-__all__ = ["OpenAI", "AsyncOpenAI", "AzureOpenAI", "AsyncAzureOpenAI"]
+__all__ = [
+    "OpenAI",
+    "AsyncOpenAI",
+    "AzureOpenAI",
+    "AsyncAzureOpenAI",
+    "format_openai_response",
+    "format_openai_input",
+    "extract_openai_tools",
+    "format_openai_streaming_content",
+]