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)

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

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,
) -> 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

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

    cache_key = _get_cache_key(
        "openai",
        api_key=api_key,
        base_url=base_url,
        organization=organization,
        timeout=timeout,
        default_headers=default_headers,
    )

    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,
    )

    _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)

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

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,
) -> 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

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

    cache_key = _get_cache_key(
        "async_openai",
        api_key=api_key,
        base_url=base_url,
        organization=organization,
        timeout=timeout,
        default_headers=default_headers,
    )

    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,
    )

    _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