ReferencePython SDK ReferenceNemoguardrailsLlmClients

nemoguardrails.llm.clients.base

View as Markdown

Module Contents

Classes

NameDescription
BaseClient-
HTTPResponse-

Functions

NameDescription
_is_stale_loop_errorReturn True if err is a RuntimeError caused by a stale loop binding.

Data

_STALE_LOOP_MARKERS

log

API

class nemoguardrails.llm.clients.base.BaseClient(
    base_url: str,
    api_key: typing.Optional[str] = None,
    timeout: typing.Optional[float] = None,
    connect_timeout: typing.Optional[float] = None,
    max_retries: int = DEFAULT_MAX_RETRIES,
    custom_headers: typing.Optional[typing.Dict[str, str]] = None,
    custom_query: typing.Optional[typing.Dict[str, typing.Any]] = None,
    http_client: typing.Optional[httpx.AsyncClient] = None
)
_base_url
= base_url.rstrip('/')
_client
AsyncClient
_custom_headers
= custom_headers or {}
_custom_query
= custom_query or {}
_owns_client
= http_client is None
provider_name
Optional[str]
provider_url
Optional[str]
nemoguardrails.llm.clients.base.BaseClient.__aenter__()
async
nemoguardrails.llm.clients.base.BaseClient.__aexit__(
    args = ()
)
async
nemoguardrails.llm.clients.base.BaseClient._apost(
    path: str,
    payload: typing.Dict[str, typing.Any]
) -> nemoguardrails.llm.clients.base.HTTPResponse
async
nemoguardrails.llm.clients.base.BaseClient._apost_stream(
    path: str,
    payload: typing.Dict[str, typing.Any]
) -> typing.AsyncGenerator[nemoguardrails.llm.clients.base.HTTPResponse, None]
async
nemoguardrails.llm.clients.base.BaseClient._build_headers() -> typing.Dict[str, str]
nemoguardrails.llm.clients.base.BaseClient._calculate_retry_delay(
    headers: typing.Any,
    retries_attempted: int
) -> float
staticmethod
nemoguardrails.llm.clients.base.BaseClient._check_sse_error(
    parsed: typing.Any,
    headers: typing.Any,
    ctx: typing.Optional[nemoguardrails.llm.clients._errors.ErrorContext] = None
) -> None
nemoguardrails.llm.clients.base.BaseClient._error_context() -> nemoguardrails.llm.clients._errors.ErrorContext
nemoguardrails.llm.clients.base.BaseClient._should_retry(
    status_code: int,
    headers: typing.Any
) -> bool
staticmethod
nemoguardrails.llm.clients.base.BaseClient._sleep_for_retry(
    retries_attempted: int,
    headers: typing.Any = None
) -> None
async
nemoguardrails.llm.clients.base.BaseClient.close() -> None
async
nemoguardrails.llm.clients.base.BaseClient.is_closed() -> bool
class nemoguardrails.llm.clients.base.HTTPResponse(
    body: typing.Dict[str, typing.Any],
    headers: typing.Mapping[str, str] = dict(),
    status_code: int = 200
)
Dataclass
body
Dict[str, Any]
headers
Mapping[str, str] = field(default_factory=dict)
status_code
int = 200
nemoguardrails.llm.clients.base._is_stale_loop_error(
    err: BaseException
) -> bool

Return True if err is a RuntimeError caused by a stale loop binding.

CPython raises one of several messages when an httpx transport (or any asyncio primitive it owns) is reused on a loop other than the one it was created on:

  • ‘Event loop is closed’ (BaseEventLoop._check_closed) when the original loop has been closed.
  • ‘<asyncio.X object> is bound to a different event loop’ (Lock / Event / Semaphore / Queue) when the original loop is still alive.
  • ‘got Future attached to a different loop’ / ‘Task got Future …’ when a Future created in another loop is awaited.

All three are transient: httpx invalidates the stale primitive on retry and rebuilds it in the running loop.

nemoguardrails.llm.clients.base._STALE_LOOP_MARKERS = ('event loop is closed', 'different loop', 'different event loop')
 
nemoguardrails.llm.clients.base.log = logging.getLogger(__name__)