Skip to content

Factory

bruno_llm.factory.LLMFactory

Factory for creating LLM provider instances.

Provides multiple ways to instantiate providers: - Direct creation with configuration - Environment-based configuration - Fallback chain for resilience

Examples:

>>> # Direct creation
>>> llm = LLMFactory.create(
...     provider="ollama",
...     config={"model": "llama2"}
... )
>>> # From environment variables
>>> llm = LLMFactory.create_from_env(provider="openai")
>>> # With fallback
>>> llm = LLMFactory.create_with_fallback(
...     providers=["openai", "ollama"],
...     configs=[openai_config, ollama_config]
... )
Source code in bruno_llm/factory.py
class LLMFactory:
    """
    Factory for creating LLM provider instances.

    Provides multiple ways to instantiate providers:
    - Direct creation with configuration
    - Environment-based configuration
    - Fallback chain for resilience

    Examples:
        >>> # Direct creation
        >>> llm = LLMFactory.create(
        ...     provider="ollama",
        ...     config={"model": "llama2"}
        ... )

        >>> # From environment variables
        >>> llm = LLMFactory.create_from_env(provider="openai")

        >>> # With fallback
        >>> llm = LLMFactory.create_with_fallback(
        ...     providers=["openai", "ollama"],
        ...     configs=[openai_config, ollama_config]
        ... )
    """

    _providers: dict[str, Callable[..., LLMInterface]] = {}

    @classmethod
    def register(cls, name: str, provider_class: Callable[..., LLMInterface]) -> None:
        """
        Register a provider class with the factory.

        Args:
            name: Provider name (e.g., "ollama", "openai")
            provider_class: Provider class or factory function
        """
        cls._providers[name.lower()] = provider_class

    @classmethod
    def create(
        cls, provider: str, config: Optional[dict[str, Any]] = None, **kwargs: Any
    ) -> LLMInterface:
        """
        Create a provider instance.

        Args:
            provider: Provider name ("ollama", "openai", etc.)
            config: Configuration dictionary (optional)
            **kwargs: Additional arguments passed to provider

        Returns:
            Configured LLMInterface instance

        Raises:
            ConfigurationError: If provider not found or config invalid

        Examples:
            >>> llm = LLMFactory.create("ollama", {"model": "llama2"})
            >>> llm = LLMFactory.create("openai", api_key="sk-...", model="gpt-4")
        """
        provider_lower = provider.lower()

        if provider_lower not in cls._providers:
            available = ", ".join(cls._providers.keys())
            raise ConfigurationError(
                f"Provider '{provider}' not found. Available providers: {available}"
            )

        provider_class = cls._providers[provider_lower]

        # Merge config dict and kwargs
        final_config = config.copy() if config else {}
        final_config.update(kwargs)

        try:
            return provider_class(**final_config)
        except TypeError as e:
            raise ConfigurationError(f"Invalid configuration for provider '{provider}': {e}") from e

    @classmethod
    def create_from_env(cls, provider: str, prefix: Optional[str] = None) -> LLMInterface:
        """
        Create provider from environment variables.

        Environment variables are read using the pattern:
        {PREFIX}_{PROVIDER}_{SETTING}

        Args:
            provider: Provider name
            prefix: Environment variable prefix (default: "BRUNO_LLM")

        Returns:
            Configured LLMInterface instance

        Raises:
            ConfigurationError: If required env vars missing

        Examples:
            >>> # With BRUNO_LLM_OPENAI_API_KEY=sk-...
            >>> # and BRUNO_LLM_OPENAI_MODEL=gpt-4
            >>> llm = LLMFactory.create_from_env("openai")

            >>> # Custom prefix
            >>> # MY_APP_OLLAMA_MODEL=llama2
            >>> llm = LLMFactory.create_from_env("ollama", prefix="MY_APP")
        """
        prefix = prefix or "BRUNO_LLM"
        provider_upper = provider.upper()
        env_prefix = f"{prefix}_{provider_upper}_"

        # Collect all matching environment variables
        config: dict[str, Any] = {}
        for key, value in os.environ.items():
            if key.startswith(env_prefix):
                # Remove prefix and convert to lowercase
                setting_name = key[len(env_prefix) :].lower()
                config[setting_name] = value

        if not config:
            raise ConfigurationError(
                f"No environment variables found for provider '{provider}'. "
                f"Expected variables starting with {env_prefix}"
            )

        return cls.create(provider, config)

    @classmethod
    async def create_with_fallback(
        cls, providers: list[str], configs: Optional[list[dict[str, Any]]] = None
    ) -> LLMInterface:
        """
        Create provider with fallback chain.

        Tries each provider in order until one connects successfully.

        Args:
            providers: List of provider names in priority order
            configs: Optional list of configurations (must match providers length)

        Returns:
            First successfully connected LLMInterface instance

        Raises:
            LLMError: If all providers fail to connect

        Examples:
            >>> llm = await LLMFactory.create_with_fallback(
            ...     providers=["openai", "ollama"],
            ...     configs=[
            ...         {"api_key": "sk-...", "model": "gpt-4"},
            ...         {"model": "llama2"}
            ...     ]
            ... )
        """
        if not providers:
            raise ConfigurationError("No providers specified for fallback")

        if configs and len(configs) != len(providers):
            raise ConfigurationError(
                f"configs length ({len(configs)}) must match providers length ({len(providers)})"
            )

        errors = []

        for i, provider_name in enumerate(providers):
            try:
                config = configs[i] if configs else {}
                provider = cls.create(provider_name, config)

                # Test connection
                if await provider.check_connection():
                    return provider
                else:
                    errors.append(f"{provider_name}: Connection check failed")

            except Exception as e:
                errors.append(f"{provider_name}: {e}")
                continue

        # All providers failed
        error_details = "; ".join(errors)
        raise LLMError(
            f"All providers failed to connect. Tried: {', '.join(providers)}. "
            f"Errors: {error_details}"
        )

    @classmethod
    def list_providers(cls) -> list[str]:
        """
        List all registered providers.

        Returns:
            List of provider names
        """
        return sorted(cls._providers.keys())

    @classmethod
    def is_registered(cls, provider: str) -> bool:
        """
        Check if a provider is registered.

        Args:
            provider: Provider name

        Returns:
            True if provider is registered
        """
        return provider.lower() in cls._providers

register(name, provider_class) classmethod

Register a provider class with the factory.

Parameters:

Name Type Description Default
name str

Provider name (e.g., "ollama", "openai")

required
provider_class Callable[..., LLMInterface]

Provider class or factory function

required
Source code in bruno_llm/factory.py
@classmethod
def register(cls, name: str, provider_class: Callable[..., LLMInterface]) -> None:
    """
    Register a provider class with the factory.

    Args:
        name: Provider name (e.g., "ollama", "openai")
        provider_class: Provider class or factory function
    """
    cls._providers[name.lower()] = provider_class

create(provider, config=None, **kwargs) classmethod

Create a provider instance.

Parameters:

Name Type Description Default
provider str

Provider name ("ollama", "openai", etc.)

required
config Optional[dict[str, Any]]

Configuration dictionary (optional)

None
**kwargs Any

Additional arguments passed to provider

{}

Returns:

Type Description
LLMInterface

Configured LLMInterface instance

Raises:

Type Description
ConfigurationError

If provider not found or config invalid

Examples:

>>> llm = LLMFactory.create("ollama", {"model": "llama2"})
>>> llm = LLMFactory.create("openai", api_key="sk-...", model="gpt-4")
Source code in bruno_llm/factory.py
@classmethod
def create(
    cls, provider: str, config: Optional[dict[str, Any]] = None, **kwargs: Any
) -> LLMInterface:
    """
    Create a provider instance.

    Args:
        provider: Provider name ("ollama", "openai", etc.)
        config: Configuration dictionary (optional)
        **kwargs: Additional arguments passed to provider

    Returns:
        Configured LLMInterface instance

    Raises:
        ConfigurationError: If provider not found or config invalid

    Examples:
        >>> llm = LLMFactory.create("ollama", {"model": "llama2"})
        >>> llm = LLMFactory.create("openai", api_key="sk-...", model="gpt-4")
    """
    provider_lower = provider.lower()

    if provider_lower not in cls._providers:
        available = ", ".join(cls._providers.keys())
        raise ConfigurationError(
            f"Provider '{provider}' not found. Available providers: {available}"
        )

    provider_class = cls._providers[provider_lower]

    # Merge config dict and kwargs
    final_config = config.copy() if config else {}
    final_config.update(kwargs)

    try:
        return provider_class(**final_config)
    except TypeError as e:
        raise ConfigurationError(f"Invalid configuration for provider '{provider}': {e}") from e

create_from_env(provider, prefix=None) classmethod

Create provider from environment variables.

Environment variables are read using the pattern: {PREFIX}{PROVIDER}{SETTING}

Parameters:

Name Type Description Default
provider str

Provider name

required
prefix Optional[str]

Environment variable prefix (default: "BRUNO_LLM")

None

Returns:

Type Description
LLMInterface

Configured LLMInterface instance

Raises:

Type Description
ConfigurationError

If required env vars missing

Examples:

>>> # With BRUNO_LLM_OPENAI_API_KEY=sk-...
>>> # and BRUNO_LLM_OPENAI_MODEL=gpt-4
>>> llm = LLMFactory.create_from_env("openai")
>>> # Custom prefix
>>> # MY_APP_OLLAMA_MODEL=llama2
>>> llm = LLMFactory.create_from_env("ollama", prefix="MY_APP")
Source code in bruno_llm/factory.py
@classmethod
def create_from_env(cls, provider: str, prefix: Optional[str] = None) -> LLMInterface:
    """
    Create provider from environment variables.

    Environment variables are read using the pattern:
    {PREFIX}_{PROVIDER}_{SETTING}

    Args:
        provider: Provider name
        prefix: Environment variable prefix (default: "BRUNO_LLM")

    Returns:
        Configured LLMInterface instance

    Raises:
        ConfigurationError: If required env vars missing

    Examples:
        >>> # With BRUNO_LLM_OPENAI_API_KEY=sk-...
        >>> # and BRUNO_LLM_OPENAI_MODEL=gpt-4
        >>> llm = LLMFactory.create_from_env("openai")

        >>> # Custom prefix
        >>> # MY_APP_OLLAMA_MODEL=llama2
        >>> llm = LLMFactory.create_from_env("ollama", prefix="MY_APP")
    """
    prefix = prefix or "BRUNO_LLM"
    provider_upper = provider.upper()
    env_prefix = f"{prefix}_{provider_upper}_"

    # Collect all matching environment variables
    config: dict[str, Any] = {}
    for key, value in os.environ.items():
        if key.startswith(env_prefix):
            # Remove prefix and convert to lowercase
            setting_name = key[len(env_prefix) :].lower()
            config[setting_name] = value

    if not config:
        raise ConfigurationError(
            f"No environment variables found for provider '{provider}'. "
            f"Expected variables starting with {env_prefix}"
        )

    return cls.create(provider, config)

create_with_fallback(providers, configs=None) async classmethod

Create provider with fallback chain.

Tries each provider in order until one connects successfully.

Parameters:

Name Type Description Default
providers list[str]

List of provider names in priority order

required
configs Optional[list[dict[str, Any]]]

Optional list of configurations (must match providers length)

None

Returns:

Type Description
LLMInterface

First successfully connected LLMInterface instance

Raises:

Type Description
LLMError

If all providers fail to connect

Examples:

>>> llm = await LLMFactory.create_with_fallback(
...     providers=["openai", "ollama"],
...     configs=[
...         {"api_key": "sk-...", "model": "gpt-4"},
...         {"model": "llama2"}
...     ]
... )
Source code in bruno_llm/factory.py
@classmethod
async def create_with_fallback(
    cls, providers: list[str], configs: Optional[list[dict[str, Any]]] = None
) -> LLMInterface:
    """
    Create provider with fallback chain.

    Tries each provider in order until one connects successfully.

    Args:
        providers: List of provider names in priority order
        configs: Optional list of configurations (must match providers length)

    Returns:
        First successfully connected LLMInterface instance

    Raises:
        LLMError: If all providers fail to connect

    Examples:
        >>> llm = await LLMFactory.create_with_fallback(
        ...     providers=["openai", "ollama"],
        ...     configs=[
        ...         {"api_key": "sk-...", "model": "gpt-4"},
        ...         {"model": "llama2"}
        ...     ]
        ... )
    """
    if not providers:
        raise ConfigurationError("No providers specified for fallback")

    if configs and len(configs) != len(providers):
        raise ConfigurationError(
            f"configs length ({len(configs)}) must match providers length ({len(providers)})"
        )

    errors = []

    for i, provider_name in enumerate(providers):
        try:
            config = configs[i] if configs else {}
            provider = cls.create(provider_name, config)

            # Test connection
            if await provider.check_connection():
                return provider
            else:
                errors.append(f"{provider_name}: Connection check failed")

        except Exception as e:
            errors.append(f"{provider_name}: {e}")
            continue

    # All providers failed
    error_details = "; ".join(errors)
    raise LLMError(
        f"All providers failed to connect. Tried: {', '.join(providers)}. "
        f"Errors: {error_details}"
    )

list_providers() classmethod

List all registered providers.

Returns:

Type Description
list[str]

List of provider names

Source code in bruno_llm/factory.py
@classmethod
def list_providers(cls) -> list[str]:
    """
    List all registered providers.

    Returns:
        List of provider names
    """
    return sorted(cls._providers.keys())

is_registered(provider) classmethod

Check if a provider is registered.

Parameters:

Name Type Description Default
provider str

Provider name

required

Returns:

Type Description
bool

True if provider is registered

Source code in bruno_llm/factory.py
@classmethod
def is_registered(cls, provider: str) -> bool:
    """
    Check if a provider is registered.

    Args:
        provider: Provider name

    Returns:
        True if provider is registered
    """
    return provider.lower() in cls._providers