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
    )

    with _client_cache_lock:
        cached_client = _get_cached_client(cache_key)
        if cached_client is not None:
            return cast(OpenAI, cached_client)

        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
        )

        _store_client(cache_key, 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
    )

    with _client_cache_lock:
        cached_client = _get_cached_client(cache_key)
        if cached_client is not None:
            return cast(AsyncOpenAI, cached_client)

        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
        )

        _store_client(cache_key, 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)

    with _client_cache_lock:
        cached_client = _get_cached_client(cache_key)
        if cached_client is not None:
            return cast(Groq, cached_client)

        client = Groq(api_key=api_key)
        _store_client(cache_key, 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)

    with _client_cache_lock:
        cached_client = _get_cached_client(cache_key)
        if cached_client is not None:
            return cast(AsyncGroq, cached_client)

        client = AsyncGroq(api_key=api_key)
        _store_client(cache_key, 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)

    with _client_cache_lock:
        cached_client = _get_cached_client(cache_key)
        if cached_client is not None:
            return cast(Cerebras, cached_client)

        client = Cerebras(api_key=api_key)
        _store_client(cache_key, 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)

    with _client_cache_lock:
        cached_client = _get_cached_client(cache_key)
        if cached_client is not None:
            return cast(AsyncCerebras, cached_client)

        client = AsyncCerebras(api_key=api_key)
        _store_client(cache_key, client)
    return client

prune_cache(max_age_seconds)

Remove cache entries whose last-used time exceeds max_age_seconds.

Evicted clients are not closed here because they may still be serving in-flight requests. Cleanup is handled by the atexit handler and the garbage collector.

Parameters:

Name	Type	Description	Default
max_age_seconds	float	Maximum age (in seconds) for cache entries to keep. Entries older than this value are removed.	required

Returns:

Type	Description
int	Number of cache entries removed.

Source code in langroid/language_models/client_cache.py

def prune_cache(max_age_seconds: float) -> int:
    """
    Remove cache entries whose last-used time exceeds *max_age_seconds*.

    Evicted clients are **not** closed here because they may still be serving
    in-flight requests.  Cleanup is handled by the ``atexit`` handler and the
    garbage collector.

    Args:
        max_age_seconds: Maximum age (in seconds) for cache entries to keep.
            Entries older than this value are removed.

    Returns:
        Number of cache entries removed.
    """
    if max_age_seconds < 0:
        raise ValueError("max_age_seconds must be non-negative")

    now = time.monotonic()

    with _client_cache_lock:
        stale_keys = [
            key
            for key, (_, last_used_at) in _client_cache.items()
            if now - last_used_at > max_age_seconds
        ]

        for key in stale_keys:
            _client_cache.pop(key)

    # Don't close evicted clients here — they may still be serving in-flight
    # requests. The atexit handler and GC will clean them up.

    return len(stale_keys)