client_cache

langroid/language_models/client_cache.py

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

wrap_api_key_provider_async(provider)

Wrap a sync API-key provider for use with AsyncOpenAI.

AsyncOpenAI awaits its api_key callable, so a plain sync provider must be wrapped in an async function. The sync provider is run in a worker thread, so a blocking token refresh cannot stall the event loop. Providers must therefore be thread-safe -- the sync client may call them from arbitrary threads anyway.

Parameters:

Name	Type	Description	Default
provider	Callable[[], str]	Callable returning a (possibly short-lived) API key.	required

Returns:

Type	Description
Callable[[], Awaitable[str]]	Async callable returning the same key.

Source code in langroid/language_models/client_cache.py

def wrap_api_key_provider_async(
    provider: Callable[[], str],
) -> Callable[[], Awaitable[str]]:
    """
    Wrap a sync API-key provider for use with ``AsyncOpenAI``.

    ``AsyncOpenAI`` awaits its ``api_key`` callable, so a plain sync
    provider must be wrapped in an async function. The sync provider is run
    in a worker thread, so a blocking token refresh cannot stall the event
    loop. Providers must therefore be thread-safe -- the sync client may
    call them from arbitrary threads anyway.

    Args:
        provider: Callable returning a (possibly short-lived) API key.

    Returns:
        Async callable returning the same key.
    """

    async def _provider() -> str:
        return await asyncio.to_thread(provider)

    return _provider

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	Union[str, Callable[[], str]]	OpenAI API key, or a callable returning a fresh key (for short-lived rotating credentials); a callable is resolved per-request by the OpenAI client and is excluded from the cache key (the cache is keyed on the provider's identity)	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: Union[str, Callable[[], 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, or a callable returning a fresh key
            (for short-lived rotating credentials); a callable is resolved
            per-request by the OpenAI client and is excluded from the
            cache key (the cache is keyed on the provider's identity)
        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

    cache_key = _get_cache_key(
        "openai",
        api_key=_api_key_cache_component(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)

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

        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	Union[str, Callable[[], str]]	OpenAI API key, or a callable returning a fresh key (for short-lived rotating credentials); a callable is resolved per-request by the OpenAI client and is excluded from the cache key (the cache is keyed on the provider's identity)	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: Union[str, Callable[[], 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, or a callable returning a fresh key
            (for short-lived rotating credentials); a callable is resolved
            per-request by the OpenAI client and is excluded from the
            cache key (the cache is keyed on the provider's identity)
        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)

    api_key_arg: Union[str, Callable[[], Awaitable[str]]]
    if callable(api_key):
        # AsyncOpenAI awaits its api_key callable, so wrap the sync provider
        api_key_arg = wrap_api_key_provider_async(api_key)
    else:
        api_key_arg = api_key

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

    cache_key = _get_cache_key(
        "async_openai",
        api_key=_api_key_cache_component(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)

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

        client = AsyncOpenAI(
            api_key=api_key_arg,
            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)