create the ai provider interface and factory

jsonbailey · jsonbailey · commit 82718075cdba · 2025-11-07T21:35:34.000Z
diff --git a/ldai/models.py b/ldai/models.py
@@ -1,6 +1,6 @@
 import warnings
 from dataclasses import dataclass
-from typing import Any, Dict, List, Literal, Optional
+from typing import Any, Dict, List, Literal, Optional, Union
 
 from ldai.tracker import LDAIConfigTracker
 
@@ -334,6 +334,9 @@ class AIAgentConfigRequest:
 # Type alias for multiple agents
 AIAgents = Dict[str, AIAgentConfig]
 
+# Type alias for all AI Config variants
+AIConfigKind = Union[AIAgentConfig, AICompletionConfig, AIJudgeConfig]
+
 
 # ============================================================================
 # Deprecated Type Aliases for Backward Compatibility
diff --git a/ldai/providers/__init__.py b/ldai/providers/__init__.py
@@ -0,0 +1,11 @@
+"""AI Provider interfaces and factory for LaunchDarkly AI SDK."""
+
+from ldai.providers.ai_provider import AIProvider
+from ldai.providers.ai_provider_factory import AIProviderFactory, SupportedAIProvider
+
+__all__ = [
+    'AIProvider',
+    'AIProviderFactory',
+    'SupportedAIProvider',
+]
+
diff --git a/ldai/providers/ai_provider.py b/ldai/providers/ai_provider.py
@@ -0,0 +1,96 @@
+"""Abstract base class for AI providers."""
+
+from abc import ABC, abstractmethod
+from typing import Any, Dict, List, Optional, Union
+
+from ldai.models import AIConfigKind, LDMessage
+from ldai.providers.types import ChatResponse, StructuredResponse
+
+
+class AIProvider(ABC):
+    """
+    Abstract base class for AI providers that implement chat model functionality.
+    
+    This class provides the contract that all provider implementations must follow
+    to integrate with LaunchDarkly's tracking and configuration capabilities.
+    
+    Following the AICHAT spec recommendation to use base classes with non-abstract methods
+    for better extensibility and backwards compatibility.
+    """
+
+    def __init__(self, logger: Optional[Any] = None):
+        """
+        Initialize the AI provider.
+        
+        :param logger: Optional logger for logging provider operations.
+        """
+        self.logger = logger
+
+    async def invoke_model(self, messages: List[LDMessage]) -> ChatResponse:
+        """
+        Invoke the chat model with an array of messages.
+        
+        This method should convert messages to provider format, invoke the model,
+        and return a ChatResponse with the result and metrics.
+        
+        Default implementation takes no action and returns a placeholder response.
+        Provider implementations should override this method.
+        
+        :param messages: Array of LDMessage objects representing the conversation
+        :return: ChatResponse containing the model's response
+        """
+        if self.logger:
+            self.logger.warn('invokeModel not implemented by this provider')
+        
+        from ldai.models import LDMessage
+        from ldai.providers.types import LDAIMetrics
+        
+        return ChatResponse(
+            message=LDMessage(role='assistant', content=''),
+            metrics=LDAIMetrics(success=False, usage=None),
+        )
+
+    async def invoke_structured_model(
+        self,
+        messages: List[LDMessage],
+        response_structure: Dict[str, Any],
+    ) -> StructuredResponse:
+        """
+        Invoke the chat model with structured output support.
+        
+        This method should convert messages to provider format, invoke the model with
+        structured output configuration, and return a structured response.
+        
+        Default implementation takes no action and returns a placeholder response.
+        Provider implementations should override this method.
+        
+        :param messages: Array of LDMessage objects representing the conversation
+        :param response_structure: Dictionary of output configurations keyed by output name
+        :return: StructuredResponse containing the structured data
+        """
+        if self.logger:
+            self.logger.warn('invokeStructuredModel not implemented by this provider')
+        
+        from ldai.providers.types import LDAIMetrics
+        
+        return StructuredResponse(
+            data={},
+            raw_response='',
+            metrics=LDAIMetrics(success=False, usage=None),
+        )
+
+    @staticmethod
+    @abstractmethod
+    async def create(ai_config: AIConfigKind, logger: Optional[Any] = None) -> 'AIProvider':
+        """
+        Static method that constructs an instance of the provider.
+        
+        Each provider implementation must provide their own static create method
+        that accepts an AIConfigKind and returns a configured instance.
+        
+        :param ai_config: The LaunchDarkly AI configuration
+        :param logger: Optional logger for the provider
+        :return: Configured provider instance
+        """
+        raise NotImplementedError('Provider implementations must override the static create method')
+
diff --git a/ldai/providers/ai_provider_factory.py b/ldai/providers/ai_provider_factory.py
@@ -0,0 +1,154 @@
+"""Factory for creating AIProvider instances based on the provider configuration."""
+
+import importlib
+from typing import Any, List, Literal, Optional, Type
+
+from ldai.models import AIConfigKind
+from ldai.providers.ai_provider import AIProvider
+
+
+# List of supported AI providers
+SUPPORTED_AI_PROVIDERS = [
+    # Multi-provider packages should be last in the list
+    'langchain',
+]
+
+# Type representing the supported AI providers
+SupportedAIProvider = Literal['langchain']
+
+
+class AIProviderFactory:
+    """
+    Factory for creating AIProvider instances based on the provider configuration.
+    """
+
+    @staticmethod
+    async def create(
+        ai_config: AIConfigKind,
+        logger: Optional[Any] = None,
+        default_ai_provider: Optional[SupportedAIProvider] = None,
+    ) -> Optional[AIProvider]:
+        """
+        Create an AIProvider instance based on the AI configuration.
+        
+        This method attempts to load provider-specific implementations dynamically.
+        Returns None if the provider is not supported.
+        
+        :param ai_config: The AI configuration
+        :param logger: Optional logger for logging provider initialization
+        :param default_ai_provider: Optional default AI provider to use
+        :return: AIProvider instance or None if not supported
+        """
+        provider_name = ai_config.provider.name.lower() if ai_config.provider else None
+        # Determine which providers to try based on default_ai_provider
+        providers_to_try = AIProviderFactory._get_providers_to_try(default_ai_provider, provider_name)
+
+        # Try each provider in order
+        for provider_type in providers_to_try:
+            provider = await AIProviderFactory._try_create_provider(provider_type, ai_config, logger)
+            if provider:
+                return provider
+
+        # If no provider was successfully created, log a warning
+        if logger:
+            logger.warn(
+                f"Provider is not supported or failed to initialize: {provider_name or 'unknown'}"
+            )
+        return None
+
+    @staticmethod
+    def _get_providers_to_try(
+        default_ai_provider: Optional[SupportedAIProvider],
+        provider_name: Optional[str],
+    ) -> List[SupportedAIProvider]:
+        """
+        Determine which providers to try based on default_ai_provider and provider_name.
+        
+        :param default_ai_provider: Optional default provider to use
+        :param provider_name: Optional provider name from config
+        :return: List of providers to try in order
+        """
+        # If default_ai_provider is set, only try that specific provider
+        if default_ai_provider:
+            return [default_ai_provider]
+
+        # If no default_ai_provider is set, try all providers in order
+        provider_set = set()
+
+        # First try the specific provider if it's supported
+        if provider_name and provider_name in SUPPORTED_AI_PROVIDERS:
+            provider_set.add(provider_name)  # type: ignore
+
+        # Then try multi-provider packages, but avoid duplicates
+        multi_provider_packages: List[SupportedAIProvider] = ['langchain', 'vercel']
+        for provider in multi_provider_packages:
+            provider_set.add(provider)
+
+        return list(provider_set)
+
+    @staticmethod
+    async def _try_create_provider(
+        provider_type: SupportedAIProvider,
+        ai_config: AIConfigKind,
+        logger: Optional[Any] = None,
+    ) -> Optional[AIProvider]:
+        """
+        Try to create a provider of the specified type.
+        
+        :param provider_type: Type of provider to create
+        :param ai_config: AI configuration
+        :param logger: Optional logger
+        :return: AIProvider instance or None if creation failed
+        """
+        provider_mappings = {
+            'openai': ('launchdarkly_server_sdk_ai_openai', 'OpenAIProvider'),
+            'langchain': ('launchdarkly_server_sdk_ai_langchain', 'LangChainProvider'),
+            'vercel': ('launchdarkly_server_sdk_ai_vercel', 'VercelProvider'),
+        }
+
+        if provider_type not in provider_mappings:
+            return None
+
+        package_name, provider_class_name = provider_mappings[provider_type]
+        return await AIProviderFactory._create_provider(
+            package_name, provider_class_name, ai_config, logger
+        )
+
+    @staticmethod
+    async def _create_provider(
+        package_name: str,
+        provider_class_name: str,
+        ai_config: AIConfigKind,
+        logger: Optional[Any] = None,
+    ) -> Optional[AIProvider]:
+        """
+        Create a provider instance dynamically.
+        
+        :param package_name: Name of the package containing the provider
+        :param provider_class_name: Name of the provider class
+        :param ai_config: AI configuration
+        :param logger: Optional logger
+        :return: AIProvider instance or None if creation failed
+        """
+        try:
+            # Try to dynamically import the provider
+            # This will work if the package is installed
+            module = importlib.import_module(package_name)
+            provider_class: Type[AIProvider] = getattr(module, provider_class_name)
+
+            provider = await provider_class.create(ai_config, logger)
+            if logger:
+                logger.debug(
+                    f"Successfully created AIProvider for: {ai_config.provider.name if ai_config.provider else 'unknown'} "
+                    f"with package {package_name}"
+                )
+            return provider
+        except (ImportError, AttributeError, Exception) as error:
+            # If the provider is not available or creation fails, return None
+            if logger:
+                logger.warn(
+                    f"Error creating AIProvider for: {ai_config.provider.name if ai_config.provider else 'unknown'} "
+                    f"with package {package_name}: {error}"
+                )
+            return None
+
diff --git a/ldai/providers/types.py b/ldai/providers/types.py
@@ -0,0 +1,37 @@
+"""Types for AI provider responses."""
+
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+from ldai.models import LDMessage
+from ldai.tracker import TokenUsage
+
+
+@dataclass
+class LDAIMetrics:
+    """
+    Metrics information for AI operations that includes success status and token usage.
+    """
+    success: bool
+    usage: Optional[TokenUsage] = None
+
+
+@dataclass
+class ChatResponse:
+    """
+    Chat response structure.
+    """
+    message: LDMessage
+    metrics: LDAIMetrics
+    evaluations: Optional[List[Any]] = None  # List of JudgeResponse, will be populated later
+
+
+@dataclass
+class StructuredResponse:
+    """
+    Structured response from AI models.
+    """
+    data: Dict[str, Any]
+    raw_response: str
+    metrics: LDAIMetrics
+
