Skip to content

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