client_cache

langroid/language_models/client_cache.py

Client caching/singleton pattern for LLM clients to prevent connection pool exhaustion.

get_openai_client(api_key, base_url=None, organization=None, timeout=120.0, default_headers=None, http_client=None, http_client_config=None)

Get or create a singleton OpenAI client with the given configuration.

Parameters:

Name	Type	Description	Default
api_key	str	OpenAI API key	required
base_url	Optional[str]	Optional base URL for API	None
organization	Optional[str]	Optional organization ID	None
timeout	Union[float, Timeout]	Request timeout	120.0
default_headers	Optional[Dict[str, str]]	Optional default headers	None
http_client	Optional[Any]	Optional httpx.Client instance	None
http_client_config	Optional[Dict[str, Any]]	Optional config dict for creating httpx.Client	None

Returns:

Type	Description
OpenAI	OpenAI client instance

Source code in langroid/language_models/client_cache.py

def get_openai_client(
    api_key: str,
    base_url: Optional[str] = None,
    organization: Optional[str] = None,
    timeout: Union[float, Timeout] = 120.0,
    default_headers: Optional[Dict[str, str]] = None,
    http_client: Optional[Any] = None,
    http_client_config: Optional[Dict[str, Any]] = None,
) -> OpenAI:
    """
    Get or create a singleton OpenAI client with the given configuration.

    Args:
        api_key: OpenAI API key
        base_url: Optional base URL for API
        organization: Optional organization ID
        timeout: Request timeout
        default_headers: Optional default headers
        http_client: Optional httpx.Client instance
        http_client_config: Optional config dict for creating httpx.Client

    Returns:
        OpenAI client instance
    """
    if isinstance(timeout, (int, float)):
        timeout = Timeout(timeout)

    # If http_client is provided directly, don't cache (complex object)
    if http_client is not None:
        client = OpenAI(
            api_key=api_key,
            base_url=base_url,
            organization=organization,
            timeout=timeout,
            default_headers=default_headers,
            http_client=http_client,
        )
        _all_clients.add(client)
        return client

    # If http_client_config is provided, create client from config and cache
    created_http_client = None
    if http_client_config is not None:
        try:
            from httpx import Client

            created_http_client = Client(**http_client_config)
        except ImportError:
            raise ValueError(
                "httpx is required to use http_client_config. "
                "Install it with: pip install httpx"
            )

    cache_key = _get_cache_key(
        "openai",
        api_key=api_key,
        base_url=base_url,
        organization=organization,
        timeout=timeout,
        default_headers=default_headers,
        http_client_config=http_client_config,  # Include config in cache key
    )

    if cache_key in _client_cache:
        return cast(OpenAI, _client_cache[cache_key])

    client = OpenAI(
        api_key=api_key,
        base_url=base_url,
        organization=organization,
        timeout=timeout,
        default_headers=default_headers,
        http_client=created_http_client,  # Use the client created from config
    )

    _client_cache[cache_key] = client
    _all_clients.add(client)
    return client

get_async_openai_client(api_key, base_url=None, organization=None, timeout=120.0, default_headers=None, http_client=None, http_client_config=None)

Get or create a singleton AsyncOpenAI client with the given configuration.

Parameters:

Name	Type	Description	Default
api_key	str	OpenAI API key	required
base_url	Optional[str]	Optional base URL for API	None
organization	Optional[str]	Optional organization ID	None
timeout	Union[float, Timeout]	Request timeout	120.0
default_headers	Optional[Dict[str, str]]	Optional default headers	None
http_client	Optional[Any]	Optional httpx.AsyncClient instance	None
http_client_config	Optional[Dict[str, Any]]	Optional config dict for creating httpx.AsyncClient	None

Returns:

Type	Description
AsyncOpenAI	AsyncOpenAI client instance

Source code in langroid/language_models/client_cache.py

def get_async_openai_client(
    api_key: str,
    base_url: Optional[str] = None,
    organization: Optional[str] = None,
    timeout: Union[float, Timeout] = 120.0,
    default_headers: Optional[Dict[str, str]] = None,
    http_client: Optional[Any] = None,
    http_client_config: Optional[Dict[str, Any]] = None,
) -> AsyncOpenAI:
    """
    Get or create a singleton AsyncOpenAI client with the given configuration.

    Args:
        api_key: OpenAI API key
        base_url: Optional base URL for API
        organization: Optional organization ID
        timeout: Request timeout
        default_headers: Optional default headers
        http_client: Optional httpx.AsyncClient instance
        http_client_config: Optional config dict for creating httpx.AsyncClient

    Returns:
        AsyncOpenAI client instance
    """
    if isinstance(timeout, (int, float)):
        timeout = Timeout(timeout)

    # If http_client is provided directly, don't cache (complex object)
    if http_client is not None:
        client = AsyncOpenAI(
            api_key=api_key,
            base_url=base_url,
            organization=organization,
            timeout=timeout,
            default_headers=default_headers,
            http_client=http_client,
        )
        _all_clients.add(client)
        return client

    # If http_client_config is provided, create async client from config and cache
    created_http_client = None
    if http_client_config is not None:
        try:
            from httpx import AsyncClient

            created_http_client = AsyncClient(**http_client_config)
        except ImportError:
            raise ValueError(
                "httpx is required to use http_client_config. "
                "Install it with: pip install httpx"
            )

    cache_key = _get_cache_key(
        "async_openai",
        api_key=api_key,
        base_url=base_url,
        organization=organization,
        timeout=timeout,
        default_headers=default_headers,
        http_client_config=http_client_config,  # Include config in cache key
    )

    if cache_key in _client_cache:
        return cast(AsyncOpenAI, _client_cache[cache_key])

    client = AsyncOpenAI(
        api_key=api_key,
        base_url=base_url,
        organization=organization,
        timeout=timeout,
        default_headers=default_headers,
        http_client=created_http_client,  # Use the client created from config
    )

    _client_cache[cache_key] = client
    _all_clients.add(client)
    return client

get_groq_client(api_key)

Get or create a singleton Groq client with the given configuration.

Parameters:

Name	Type	Description	Default
api_key	str	Groq API key	required

Returns:

Type	Description
Groq	Groq client instance

Source code in langroid/language_models/client_cache.py

def get_groq_client(api_key: str) -> Groq:
    """
    Get or create a singleton Groq client with the given configuration.

    Args:
        api_key: Groq API key

    Returns:
        Groq client instance
    """
    cache_key = _get_cache_key("groq", api_key=api_key)

    if cache_key in _client_cache:
        return cast(Groq, _client_cache[cache_key])

    client = Groq(api_key=api_key)
    _client_cache[cache_key] = client
    _all_clients.add(client)
    return client

get_async_groq_client(api_key)

Get or create a singleton AsyncGroq client with the given configuration.

Parameters:

Name	Type	Description	Default
api_key	str	Groq API key	required

Returns:

Type	Description
AsyncGroq	AsyncGroq client instance

Source code in langroid/language_models/client_cache.py

def get_async_groq_client(api_key: str) -> AsyncGroq:
    """
    Get or create a singleton AsyncGroq client with the given configuration.

    Args:
        api_key: Groq API key

    Returns:
        AsyncGroq client instance
    """
    cache_key = _get_cache_key("async_groq", api_key=api_key)

    if cache_key in _client_cache:
        return cast(AsyncGroq, _client_cache[cache_key])

    client = AsyncGroq(api_key=api_key)
    _client_cache[cache_key] = client
    _all_clients.add(client)
    return client

get_cerebras_client(api_key)

Get or create a singleton Cerebras client with the given configuration.

Parameters:

Name	Type	Description	Default
api_key	str	Cerebras API key	required

Returns:

Type	Description
Cerebras	Cerebras client instance

Source code in langroid/language_models/client_cache.py

def get_cerebras_client(api_key: str) -> Cerebras:
    """
    Get or create a singleton Cerebras client with the given configuration.

    Args:
        api_key: Cerebras API key

    Returns:
        Cerebras client instance
    """
    cache_key = _get_cache_key("cerebras", api_key=api_key)

    if cache_key in _client_cache:
        return cast(Cerebras, _client_cache[cache_key])

    client = Cerebras(api_key=api_key)
    _client_cache[cache_key] = client
    _all_clients.add(client)
    return client

get_async_cerebras_client(api_key)

Get or create a singleton AsyncCerebras client with the given configuration.

Parameters:

Name	Type	Description	Default
api_key	str	Cerebras API key	required

Returns:

Type	Description
AsyncCerebras	AsyncCerebras client instance

Source code in langroid/language_models/client_cache.py

def get_async_cerebras_client(api_key: str) -> AsyncCerebras:
    """
    Get or create a singleton AsyncCerebras client with the given configuration.

    Args:
        api_key: Cerebras API key

    Returns:
        AsyncCerebras client instance
    """
    cache_key = _get_cache_key("async_cerebras", api_key=api_key)

    if cache_key in _client_cache:
        return cast(AsyncCerebras, _client_cache[cache_key])

    client = AsyncCerebras(api_key=api_key)
    _client_cache[cache_key] = client
    _all_clients.add(client)
    return client