lpgcast.llm.client

The model client: one JSON object per request, from any OpenAI-compatible endpoint.

:meth:LLMClient.acomplete_json is the only entry point. It serves a cached response when there is one, and otherwise calls the endpoint — asking for strict json_schema output first and falling back to json_object when the endpoint rejects it — then parses the reply tolerantly, re-asking with the parse error shown when the reply is not a JSON object.

Without an API key no request is made: only cached responses answer.

Attributes

Mode = Literal['json_schema', 'json_object'] module-attribute

OPENAI_BASE_URL = 'https://api.openai.com/v1' module-attribute

RETRY_BACKOFF_BASE_SECONDS = 0.5 module-attribute

RETRY_BACKOFF_MAX_SECONDS = 8.0 module-attribute

SCHEMA_INSTRUCTION = 'Reply with one JSON object and nothing else. It must conform to this JSON Schema:\n' module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

ChatMessage

Bases: TypedDict

One chat message.

Source code in lpgcast/llm/client.py

class ChatMessage(TypedDict):
    """One chat message."""

    role: Literal["system", "user", "assistant"]
    content: str

Attributes

content instance-attribute

role instance-attribute

JsonCompletion

Bases: BaseModel

The parsed reply to one :meth:LLMClient.acomplete_json request.

Source code in lpgcast/llm/client.py

class JsonCompletion(BaseModel):
    """The parsed reply to one :meth:`LLMClient.acomplete_json` request."""

    data: dict[str, Any] = Field(description="The parsed JSON object.")
    content: str = Field(description="Raw text of the reply that parsed.")
    mode: Mode = Field(description="Response format that reply was requested in.")
    cached: bool = Field(description="Whether that reply came from the cache.")
    attempts: int = Field(description="Requests made, counting re-asks.")
    usage: TokenUsage = Field(
        default_factory=TokenUsage, description="Tokens over all attempts."
    )

Attributes

attempts = Field(description='Requests made, counting re-asks.') class-attribute instance-attribute

cached = Field(description='Whether that reply came from the cache.') class-attribute instance-attribute

content = Field(description='Raw text of the reply that parsed.') class-attribute instance-attribute

data = Field(description='The parsed JSON object.') class-attribute instance-attribute

mode = Field(description='Response format that reply was requested in.') class-attribute instance-attribute

usage = Field(default_factory=TokenUsage, description='Tokens over all attempts.') class-attribute instance-attribute

LLMClient

A JSON-object client over openai.AsyncOpenAI.

Parameters:

Name	Type	Description	Default
settings	LLMSettings | None	The endpoint and model. Defaults to the environment.	None
cache	LLMCache | None	The response cache. Defaults to the environment's cache settings.	None
http_client	AsyncClient | None	An httpx.AsyncClient for the SDK, e.g. with a mock transport in tests.	None

Source code in lpgcast/llm/client.py

class LLMClient:
    """A JSON-object client over ``openai.AsyncOpenAI``.

    Args:
        settings: The endpoint and model. Defaults to the environment.
        cache: The response cache. Defaults to the environment's cache settings.
        http_client: An ``httpx.AsyncClient`` for the SDK, e.g. with a mock
            transport in tests.
    """

    def __init__(
        self,
        settings: LLMSettings | None = None,
        cache: LLMCache | None = None,
        *,
        http_client: httpx.AsyncClient | None = None,
    ) -> None:
        self.settings = settings or LLMSettings()
        self.cache = cache if cache is not None else LLMCache()
        self.usage = UsageTracker()
        self._http_client = http_client
        self._provider_client: openai.AsyncOpenAI | None = None
        self._json_schema_rejected = False

    @classmethod
    def from_settings(
        cls, settings: LPGCastSettings, *, http_client: httpx.AsyncClient | None = None
    ) -> LLMClient:
        """Build a client and its cache from top-level settings."""
        return cls(settings.llm, LLMCache(settings.cache), http_client=http_client)

    def _charge(self, apply: Callable[[UsageTracker], None]) -> None:
        apply(self.usage)
        scoped = active_tracker()
        if scoped is not None and scoped is not self.usage:
            apply(scoped)

    def _count(self, *counters: str, n: int = 1) -> None:
        def _apply(tracker: UsageTracker) -> None:
            for counter in counters:
                tracker.incr(counter, n)

        self._charge(_apply)

    def _provider(self) -> openai.AsyncOpenAI:
        if self._provider_client is None:
            key = self.settings.api_key
            if key is None or not key.get_secret_value():
                raise LLMConfigurationError(
                    "No API key: set LPGCAST_LLM_API_KEY (any value for a local "
                    "endpoint). Without one only cached responses are served."
                )
            self._provider_client = openai.AsyncOpenAI(
                api_key=key.get_secret_value(),
                base_url=self.settings.base_url or OPENAI_BASE_URL,
                timeout=self.settings.request_timeout_seconds,
                max_retries=self.settings.transport_retries,
                http_client=self._http_client,
            )
        return self._provider_client

    def _mode(self) -> Mode:
        requested = self.settings.response_format
        if requested == "json_object" or (
            requested == "auto" and self._json_schema_rejected
        ):
            return "json_object"
        return "json_schema"

    def _request(
        self,
        messages: list[ChatMessage],
        json_schema: dict[str, Any],
        schema_name: str,
        mode: Mode,
    ) -> dict[str, Any]:
        wire: list[ChatMessage] = list(messages)
        if mode == "json_object":
            schema_text = json.dumps(json_schema, separators=(",", ":"))
            wire.append({"role": "system", "content": SCHEMA_INSTRUCTION + schema_text})
            response_format: dict[str, Any] = {"type": "json_object"}
        else:
            response_format = {
                "type": "json_schema",
                "json_schema": {
                    "name": schema_name,
                    "schema": json_schema,
                    "strict": True,
                },
            }
        request: dict[str, Any] = {
            "model": self.settings.model,
            "messages": wire,
            "response_format": response_format,
        }
        if self.settings.temperature is not None:
            request["temperature"] = self.settings.temperature
        if self.settings.reasoning_effort is not None:
            request["reasoning_effort"] = self.settings.reasoning_effort
        if self.settings.max_output_tokens is not None:
            # OpenAI's reasoning models refuse `max_tokens`; other compatible
            # servers do not all read `max_completion_tokens`.
            name = (
                "max_completion_tokens"
                if self.settings.base_url is None
                else "max_tokens"
            )
            request[name] = self.settings.max_output_tokens
        return request

    async def _call(
        self,
        provider: openai.AsyncOpenAI,
        messages: list[ChatMessage],
        json_schema: dict[str, Any],
        schema_name: str,
    ) -> CachedResponse:
        while True:
            mode = self._mode()
            request = self._request(messages, json_schema, schema_name, mode)
            try:
                async with _inflight_semaphore(self.settings.max_inflight):
                    response = await provider.chat.completions.create(**request)
            except openai.APITimeoutError as error:
                self._count("calls_failed", "timeouts")
                raise LLMRequestTimeoutError(
                    f"Request exceeded {self.settings.request_timeout_seconds}s "
                    f"({self.settings.model})"
                ) from error
            except (openai.BadRequestError, openai.UnprocessableEntityError) as error:
                if (
                    mode == "json_schema"
                    and self.settings.response_format == "auto"
                    and _mentions(error, _SCHEMA_REJECTION_MARKERS)
                    and not _mentions(error, _UNIT_FAULT_MARKERS)
                ):
                    self._json_schema_rejected = True
                    self._count("json_schema_fallbacks")
                    logger.warning(
                        "%s rejected json_schema output; using json_object for "
                        "the rest of this run: %s",
                        self.settings.model,
                        error,
                    )
                    continue
                self._count("calls_failed")
                if _is_configuration_fault(error):
                    self._count("rejected")
                    raise LLMConfigurationError(
                        f"{self.settings.model} rejected the request: {error}"
                    ) from error
                raise
            except (
                openai.AuthenticationError,
                openai.PermissionDeniedError,
                openai.NotFoundError,
            ) as error:
                self._count("calls_failed", "rejected")
                raise LLMConfigurationError(
                    f"{self.settings.model} at {self.settings.base_url or 'OpenAI'} "
                    f"rejected the request: {error}"
                ) from error
            except openai.RateLimitError:
                self._count("calls_failed", "rate_limited")
                raise
            except openai.APIError:
                self._count("calls_failed")
                raise

            choice = response.choices[0] if response.choices else None
            content = (choice.message.content or "") if choice is not None else ""
            refusal = choice.message.refusal if choice is not None else None
            usage = (
                token_usage_from_openai(response.usage)
                if response.usage is not None
                else TokenUsage()
            )
            sent = sum(len(message["content"]) for message in request["messages"])
            self._charge(
                partial(
                    UsageTracker.add_call,
                    chars_sent=sent,
                    chars_received=len(content),
                    usage=usage,
                )
            )
            return CachedResponse(
                content=content,
                mode=mode,
                model=response.model,
                finish_reason=choice.finish_reason if choice is not None else None,
                refusal=refusal or None,
                usage=None if usage.is_empty() else usage,
            )

    async def _complete_raw(
        self,
        messages: list[ChatMessage],
        json_schema: dict[str, Any],
        schema_name: str,
        *,
        backoff_attempt: int = 0,
    ) -> tuple[CachedResponse, bool]:
        if not self.settings.model:
            raise LLMConfigurationError("No model: set LPGCAST_LLM_MODEL.")
        key = cache_key(messages, json_schema, self.settings)
        hit = await self.cache.aget(key)
        if hit is not None:
            self._charge(lambda t: t.add_hit(hit.usage))
            return hit, True
        if self.cache.require_hit:
            raise CacheMissError(
                "The cache is set to require hits and this request is not cached "
                f"({self.cache.dir}). The prompt, the schema or a model setting "
                "changed since the cache was filled."
            )
        provider = self._provider()
        if backoff_attempt:
            await asyncio.sleep(_backoff_seconds(backoff_attempt))
        entry = await self._call(provider, messages, json_schema, schema_name)
        await self.cache.aset(key, entry)
        return entry, False

    async def acomplete_json(
        self,
        messages: Sequence[ChatMessage],
        json_schema: dict[str, Any],
        *,
        schema_name: str = "instance",
    ) -> JsonCompletion:
        """Ask for one JSON object conforming to *json_schema*.

        The reply is parsed, not validated: conformance to the schema is the
        caller's check. A reply that is not a JSON object is re-asked up to
        ``parse_retries`` times with the error shown, and given up on early
        when the same syntax error repeats. A timeout is re-issued once. A truncated
        or refused reply is not re-asked.

        Raises:
            LLMJsonParseError: no attempt produced a JSON object.
            LLMConfigurationError: no key or model, or the endpoint refuses the
                configuration.
            LLMRequestTimeoutError: a request timed out twice.
            LLMTruncatedError: the reply stopped at the output token limit.
            LLMRefusalError: the model refused or the provider filtered the reply.
            CacheMissError: a miss while the cache requires hits.
            openai.APIError: any other provider failure, after SDK retries.
        """
        base = list(messages)
        conversation = base
        total = TokenUsage()
        attempts = self.settings.parse_retries + 1
        last_error_msg: str | None = None
        timeout_reissue_available = True

        for attempt in range(attempts):
            try:
                entry, cached = await self._complete_raw(
                    conversation, json_schema, schema_name, backoff_attempt=attempt
                )
            except LLMRequestTimeoutError:
                if not timeout_reissue_available:
                    raise
                timeout_reissue_available = False
                logger.warning("Request timed out; re-issuing once")
                entry, cached = await self._complete_raw(
                    conversation, json_schema, schema_name, backoff_attempt=1
                )
            if entry.usage is not None:
                total = total + entry.usage
            # Re-asking would hit the same limit or the same refusal.
            unusable = _unusable(entry, self.settings.model)
            if unusable is not None:
                self._count(
                    "truncated"
                    if isinstance(unusable, LLMTruncatedError)
                    else "refused"
                )
                raise unusable

            try:
                data, fixes = parse_json_object(entry.content)
            except LLMJsonParseError as error:
                repeated = error.msg == last_error_msg
                last_error_msg = error.msg
                if repeated or attempt == attempts - 1:
                    self._count("parse_abandoned")
                    logger.error(
                        "Giving up after %d attempt(s): %s", attempt + 1, error.msg
                    )
                    raise
                self._count("parse_retries")
                logger.warning(
                    "Reply %d/%d is not a JSON object (%s); re-asking",
                    attempt + 1,
                    attempts,
                    error.msg,
                )
                conversation = [*base, _reask_message(error, entry.content)]
                continue

            if fixes:
                self._count("json_repairs", n=fixes)
            return JsonCompletion(
                data=data,
                content=entry.content,
                mode=entry.mode,
                cached=cached,
                attempts=attempt + 1,
                usage=total,
            )

        raise AssertionError("unreachable")

Attributes

cache = cache if cache is not None else LLMCache() instance-attribute

settings = settings or LLMSettings() instance-attribute

usage = UsageTracker() instance-attribute

Methods:

__init__(settings=None, cache=None, *, http_client=None)

Source code in lpgcast/llm/client.py

def __init__(
    self,
    settings: LLMSettings | None = None,
    cache: LLMCache | None = None,
    *,
    http_client: httpx.AsyncClient | None = None,
) -> None:
    self.settings = settings or LLMSettings()
    self.cache = cache if cache is not None else LLMCache()
    self.usage = UsageTracker()
    self._http_client = http_client
    self._provider_client: openai.AsyncOpenAI | None = None
    self._json_schema_rejected = False

acomplete_json(messages, json_schema, *, schema_name='instance') async

Ask for one JSON object conforming to json_schema.

The reply is parsed, not validated: conformance to the schema is the caller's check. A reply that is not a JSON object is re-asked up to parse_retries times with the error shown, and given up on early when the same syntax error repeats. A timeout is re-issued once. A truncated or refused reply is not re-asked.

Raises:

Type	Description
LLMJsonParseError	no attempt produced a JSON object.
LLMConfigurationError	no key or model, or the endpoint refuses the configuration.
LLMRequestTimeoutError	a request timed out twice.
LLMTruncatedError	the reply stopped at the output token limit.
LLMRefusalError	the model refused or the provider filtered the reply.
CacheMissError	a miss while the cache requires hits.
APIError	any other provider failure, after SDK retries.

Source code in lpgcast/llm/client.py

async def acomplete_json(
    self,
    messages: Sequence[ChatMessage],
    json_schema: dict[str, Any],
    *,
    schema_name: str = "instance",
) -> JsonCompletion:
    """Ask for one JSON object conforming to *json_schema*.

    The reply is parsed, not validated: conformance to the schema is the
    caller's check. A reply that is not a JSON object is re-asked up to
    ``parse_retries`` times with the error shown, and given up on early
    when the same syntax error repeats. A timeout is re-issued once. A truncated
    or refused reply is not re-asked.

    Raises:
        LLMJsonParseError: no attempt produced a JSON object.
        LLMConfigurationError: no key or model, or the endpoint refuses the
            configuration.
        LLMRequestTimeoutError: a request timed out twice.
        LLMTruncatedError: the reply stopped at the output token limit.
        LLMRefusalError: the model refused or the provider filtered the reply.
        CacheMissError: a miss while the cache requires hits.
        openai.APIError: any other provider failure, after SDK retries.
    """
    base = list(messages)
    conversation = base
    total = TokenUsage()
    attempts = self.settings.parse_retries + 1
    last_error_msg: str | None = None
    timeout_reissue_available = True

    for attempt in range(attempts):
        try:
            entry, cached = await self._complete_raw(
                conversation, json_schema, schema_name, backoff_attempt=attempt
            )
        except LLMRequestTimeoutError:
            if not timeout_reissue_available:
                raise
            timeout_reissue_available = False
            logger.warning("Request timed out; re-issuing once")
            entry, cached = await self._complete_raw(
                conversation, json_schema, schema_name, backoff_attempt=1
            )
        if entry.usage is not None:
            total = total + entry.usage
        # Re-asking would hit the same limit or the same refusal.
        unusable = _unusable(entry, self.settings.model)
        if unusable is not None:
            self._count(
                "truncated"
                if isinstance(unusable, LLMTruncatedError)
                else "refused"
            )
            raise unusable

        try:
            data, fixes = parse_json_object(entry.content)
        except LLMJsonParseError as error:
            repeated = error.msg == last_error_msg
            last_error_msg = error.msg
            if repeated or attempt == attempts - 1:
                self._count("parse_abandoned")
                logger.error(
                    "Giving up after %d attempt(s): %s", attempt + 1, error.msg
                )
                raise
            self._count("parse_retries")
            logger.warning(
                "Reply %d/%d is not a JSON object (%s); re-asking",
                attempt + 1,
                attempts,
                error.msg,
            )
            conversation = [*base, _reask_message(error, entry.content)]
            continue

        if fixes:
            self._count("json_repairs", n=fixes)
        return JsonCompletion(
            data=data,
            content=entry.content,
            mode=entry.mode,
            cached=cached,
            attempts=attempt + 1,
            usage=total,
        )

    raise AssertionError("unreachable")

from_settings(settings, *, http_client=None) classmethod

Build a client and its cache from top-level settings.

Source code in lpgcast/llm/client.py

@classmethod
def from_settings(
    cls, settings: LPGCastSettings, *, http_client: httpx.AsyncClient | None = None
) -> LLMClient:
    """Build a client and its cache from top-level settings."""
    return cls(settings.llm, LLMCache(settings.cache), http_client=http_client)

LLMConfigurationError

Bases: RuntimeError

The deployment, not the request, is at fault: no key, no model, a bad key, an unknown model, or a parameter the model refuses. Every other request would fail the same way, so callers should stop rather than skip.

Source code in lpgcast/llm/client.py

class LLMConfigurationError(RuntimeError):
    """The deployment, not the request, is at fault: no key, no model, a bad key,
    an unknown model, or a parameter the model refuses. Every other request
    would fail the same way, so callers should stop rather than skip."""

LLMRefusalError

Bases: LLMReplyError

The model refused, or the provider filtered the reply.

Source code in lpgcast/llm/client.py

class LLMRefusalError(LLMReplyError):
    """The model refused, or the provider filtered the reply."""

LLMReplyError

Bases: RuntimeError

A reply that arrived but cannot be used, and would not improve if re-asked.

Source code in lpgcast/llm/client.py

class LLMReplyError(RuntimeError):
    """A reply that arrived but cannot be used, and would not improve if re-asked."""

    def __init__(self, message: str, content: str) -> None:
        super().__init__(message)
        self.content = content

Attributes

content = content instance-attribute

Methods:

__init__(message, content)

Source code in lpgcast/llm/client.py

def __init__(self, message: str, content: str) -> None:
    super().__init__(message)
    self.content = content

LLMRequestTimeoutError

Bases: RuntimeError

A request exceeded request_timeout_seconds.

Source code in lpgcast/llm/client.py

class LLMRequestTimeoutError(RuntimeError):
    """A request exceeded ``request_timeout_seconds``."""

LLMTruncatedError

Bases: LLMReplyError

The reply stopped at the output token limit (finish_reason == "length").

Source code in lpgcast/llm/client.py

class LLMTruncatedError(LLMReplyError):
    """The reply stopped at the output token limit (``finish_reason == "length"``)."""

Functions: