lpgcast.llm

The model client, its response cache, the tolerant JSON parser and usage accounting.

Modules:

Name	Description
cache	On-disk cache of model responses, keyed by the logical request.
client	The model client: one JSON object per request, from any OpenAI-compatible endpoint.
jsonparse	Tolerant parsing of a model's JSON reply.
usage	Token and call accounting.

Attributes

__all__ = ['CacheMissError', 'CachedResponse', 'ChatMessage', 'JsonCompletion', 'LLMCache', 'LLMClient', 'LLMConfigurationError', 'LLMJsonParseError', 'LLMRefusalError', 'LLMReplyError', 'LLMRequestTimeoutError', 'LLMTruncatedError', 'TokenUsage', 'UsageTracker', 'cache_key', 'parse_json_object', 'track_usage'] module-attribute

Classes

CacheMissError

Bases: RuntimeError

A request is not cached and the cache is set to require hits.

Source code in lpgcast/llm/cache.py

class CacheMissError(RuntimeError):
    """A request is not cached and the cache is set to require hits."""

CachedResponse

Bases: BaseModel

One stored response.

Source code in lpgcast/llm/cache.py

class CachedResponse(BaseModel):
    """One stored response."""

    content: str = Field(description="Response text as received.")
    mode: Literal["json_schema", "json_object"] = Field(
        description="Response format the request was sent with."
    )
    model: str | None = Field(default=None, description="Model the endpoint reported.")
    finish_reason: str | None = Field(default=None, description="Provider stop reason.")
    refusal: str | None = Field(
        default=None, description="The model's refusal message, when it refused."
    )
    usage: TokenUsage | None = Field(
        default=None, description="Token counts, replayed on a hit."
    )

Attributes

content = Field(description='Response text as received.') class-attribute instance-attribute

finish_reason = Field(default=None, description='Provider stop reason.') class-attribute instance-attribute

mode = Field(description='Response format the request was sent with.') class-attribute instance-attribute

model = Field(default=None, description='Model the endpoint reported.') class-attribute instance-attribute

refusal = Field(default=None, description="The model's refusal message, when it refused.") class-attribute instance-attribute

usage = Field(default=None, description='Token counts, replayed on a hit.') class-attribute instance-attribute

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

LLMCache

Response cache under <dir>/llm/<key>.json.

Source code in lpgcast/llm/cache.py

class LLMCache:
    """Response cache under ``<dir>/llm/<key>.json``."""

    def __init__(self, settings: CacheSettings | None = None) -> None:
        self.settings = settings or CacheSettings()
        base = self.settings.dir or default_cache_dir()
        self.dir = Path(base).expanduser() / LLM_CACHE_SUBDIR
        if self.settings.enabled:
            self.dir.mkdir(parents=True, exist_ok=True)

    @property
    def enabled(self) -> bool:
        return self.settings.enabled

    @property
    def require_hit(self) -> bool:
        return self.settings.require_hit

    @property
    def writable(self) -> bool:
        return self.settings.enabled and not self.settings.read_only

    def path(self, key: str) -> Path:
        return self.dir / f"{key}.json"

    def get(self, key: str) -> CachedResponse | None:
        """The cached response for *key*, or None on a miss or unreadable entry."""
        if not self.enabled:
            return None
        path = self.path(key)
        if not path.exists():
            return None
        try:
            with open(path, encoding="utf-8") as handle:
                return CachedResponse.model_validate(json.load(handle))
        except (OSError, ValueError) as error:
            logger.warning("Unreadable cache entry %s: %s", path, error)
            return None

    def set(self, key: str, entry: CachedResponse) -> None:
        """Store *entry* atomically; a no-op when the cache is not writable."""
        if not self.writable:
            return
        path = self.path(key)
        handle, tmp_name = tempfile.mkstemp(
            dir=path.parent, prefix=f"{path.stem}.", suffix=".tmp"
        )
        tmp = Path(tmp_name)
        try:
            with os.fdopen(handle, "w", encoding="utf-8") as out:
                json.dump(entry.model_dump(), out, indent=2, ensure_ascii=False)
            os.replace(tmp, path)
        except (OSError, TypeError, ValueError) as error:
            logger.warning("Could not write cache entry %s: %s", path, error)
            tmp.unlink(missing_ok=True)

    async def aget(self, key: str) -> CachedResponse | None:
        """:meth:`get` off the event loop."""
        return await asyncio.to_thread(self.get, key)

    async def aset(self, key: str, entry: CachedResponse) -> None:
        """:meth:`set` off the event loop."""
        await asyncio.to_thread(self.set, key, entry)

    def clear(self) -> int:
        """Delete every entry; returns how many were removed."""
        removed = 0
        if self.dir.exists():
            for path in self.dir.glob("*.json"):
                path.unlink(missing_ok=True)
                removed += 1
        return removed

Attributes

dir = Path(base).expanduser() / LLM_CACHE_SUBDIR instance-attribute

enabled property

require_hit property

settings = settings or CacheSettings() instance-attribute

writable property

Methods:

__init__(settings=None)

Source code in lpgcast/llm/cache.py

def __init__(self, settings: CacheSettings | None = None) -> None:
    self.settings = settings or CacheSettings()
    base = self.settings.dir or default_cache_dir()
    self.dir = Path(base).expanduser() / LLM_CACHE_SUBDIR
    if self.settings.enabled:
        self.dir.mkdir(parents=True, exist_ok=True)

aget(key) async

:meth:get off the event loop.

Source code in lpgcast/llm/cache.py

async def aget(self, key: str) -> CachedResponse | None:
    """:meth:`get` off the event loop."""
    return await asyncio.to_thread(self.get, key)

aset(key, entry) async

:meth:set off the event loop.

Source code in lpgcast/llm/cache.py

async def aset(self, key: str, entry: CachedResponse) -> None:
    """:meth:`set` off the event loop."""
    await asyncio.to_thread(self.set, key, entry)

clear()

Delete every entry; returns how many were removed.

Source code in lpgcast/llm/cache.py

def clear(self) -> int:
    """Delete every entry; returns how many were removed."""
    removed = 0
    if self.dir.exists():
        for path in self.dir.glob("*.json"):
            path.unlink(missing_ok=True)
            removed += 1
    return removed

get(key)

The cached response for key, or None on a miss or unreadable entry.

Source code in lpgcast/llm/cache.py

def get(self, key: str) -> CachedResponse | None:
    """The cached response for *key*, or None on a miss or unreadable entry."""
    if not self.enabled:
        return None
    path = self.path(key)
    if not path.exists():
        return None
    try:
        with open(path, encoding="utf-8") as handle:
            return CachedResponse.model_validate(json.load(handle))
    except (OSError, ValueError) as error:
        logger.warning("Unreadable cache entry %s: %s", path, error)
        return None

path(key)

Source code in lpgcast/llm/cache.py

def path(self, key: str) -> Path:
    return self.dir / f"{key}.json"

set(key, entry)

Store entry atomically; a no-op when the cache is not writable.

Source code in lpgcast/llm/cache.py

def set(self, key: str, entry: CachedResponse) -> None:
    """Store *entry* atomically; a no-op when the cache is not writable."""
    if not self.writable:
        return
    path = self.path(key)
    handle, tmp_name = tempfile.mkstemp(
        dir=path.parent, prefix=f"{path.stem}.", suffix=".tmp"
    )
    tmp = Path(tmp_name)
    try:
        with os.fdopen(handle, "w", encoding="utf-8") as out:
            json.dump(entry.model_dump(), out, indent=2, ensure_ascii=False)
        os.replace(tmp, path)
    except (OSError, TypeError, ValueError) as error:
        logger.warning("Could not write cache entry %s: %s", path, error)
        tmp.unlink(missing_ok=True)

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."""

LLMJsonParseError

Bases: ValueError

A reply that is not a JSON object, with the decoder's message and position.

Source code in lpgcast/llm/jsonparse.py

class LLMJsonParseError(ValueError):
    """A reply that is not a JSON object, with the decoder's message and position."""

    def __init__(self, message: str, *, msg: str, pos: int, doc: str) -> None:
        super().__init__(message)
        self.msg = msg
        self.pos = pos
        self.doc = doc

Attributes

doc = doc instance-attribute

msg = msg instance-attribute

pos = pos instance-attribute

Methods:

__init__(message, *, msg, pos, doc)

Source code in lpgcast/llm/jsonparse.py

def __init__(self, message: str, *, msg: str, pos: int, doc: str) -> None:
    super().__init__(message)
    self.msg = msg
    self.pos = pos
    self.doc = doc

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"``)."""

TokenUsage

Bases: BaseModel

Token counts for one call, as far as the provider reports them.

None means not reported, which is not zero.

Source code in lpgcast/llm/usage.py

class TokenUsage(BaseModel):
    """Token counts for one call, as far as the provider reports them.

    ``None`` means *not reported*, which is not zero.
    """

    input_tokens: int | None = Field(default=None, description="Prompt tokens.")
    output_tokens: int | None = Field(default=None, description="Completion tokens.")
    reasoning_tokens: int | None = Field(
        default=None, description="Thinking tokens, counted inside output_tokens."
    )
    cache_read_input_tokens: int | None = Field(
        default=None,
        description=(
            "Prompt tokens served from the provider's prompt cache, counted "
            "inside input_tokens. Unrelated to LPGCast's response cache."
        ),
    )

    def is_empty(self) -> bool:
        """True when the provider reported nothing."""
        return all(value is None for value in self.model_dump().values())

    def __add__(self, other: TokenUsage) -> TokenUsage:
        def _sum(a: int | None, b: int | None) -> int | None:
            return None if a is None and b is None else (a or 0) + (b or 0)

        return TokenUsage(
            **{
                name: _sum(getattr(self, name), getattr(other, name))
                for name in TokenUsage.model_fields
            }
        )

Attributes

cache_read_input_tokens = Field(default=None, description="Prompt tokens served from the provider's prompt cache, counted inside input_tokens. Unrelated to LPGCast's response cache.") class-attribute instance-attribute

input_tokens = Field(default=None, description='Prompt tokens.') class-attribute instance-attribute

output_tokens = Field(default=None, description='Completion tokens.') class-attribute instance-attribute

reasoning_tokens = Field(default=None, description='Thinking tokens, counted inside output_tokens.') class-attribute instance-attribute

Methods:

__add__(other)

Source code in lpgcast/llm/usage.py

def __add__(self, other: TokenUsage) -> TokenUsage:
    def _sum(a: int | None, b: int | None) -> int | None:
        return None if a is None and b is None else (a or 0) + (b or 0)

    return TokenUsage(
        **{
            name: _sum(getattr(self, name), getattr(other, name))
            for name in TokenUsage.model_fields
        }
    )

is_empty()

True when the provider reported nothing.

Source code in lpgcast/llm/usage.py

def is_empty(self) -> bool:
    """True when the provider reported nothing."""
    return all(value is None for value in self.model_dump().values())

UsageTracker

Bases: BaseModel

Calls, events and tokens for a run or a scope within it.

Source code in lpgcast/llm/usage.py

class UsageTracker(BaseModel):
    """Calls, events and tokens for a run or a scope within it."""

    calls: int = Field(default=0, description="Provider calls that returned.")
    cache_hits: int = Field(default=0, description="Responses served from cache.")
    calls_failed: int = Field(default=0, description="Provider calls that raised.")
    timeouts: int = Field(default=0, description="Subset of calls_failed.")
    rate_limited: int = Field(default=0, description="Subset of calls_failed.")
    rejected: int = Field(
        default=0, description="Subset of calls_failed: configuration faults."
    )
    truncated: int = Field(
        default=0, description="Replies cut off at the output token limit."
    )
    refused: int = Field(default=0, description="Replies the model refused.")
    parse_retries: int = Field(default=0, description="Re-asks after bad JSON.")
    parse_abandoned: int = Field(default=0, description="Requests given up on.")
    json_repairs: int = Field(default=0, description="Bracket-kind repairs.")
    json_schema_fallbacks: int = Field(
        default=0, description="Switches from json_schema to json_object."
    )
    chars_sent: int = Field(default=0, description="Prompt characters sent.")
    chars_received: int = Field(default=0, description="Response characters.")
    billed: TokenUsage = Field(
        default_factory=TokenUsage, description="Tokens from live provider calls."
    )
    replayed: TokenUsage = Field(
        default_factory=TokenUsage, description="Tokens of responses served from cache."
    )

    def incr(self, counter: str, n: int = 1) -> None:
        """Add *n* to an integer counter."""
        setattr(self, counter, getattr(self, counter) + n)

    def add_call(self, chars_sent: int, chars_received: int, usage: TokenUsage) -> None:
        """Charge one live provider call."""
        self.calls += 1
        self.chars_sent += chars_sent
        self.chars_received += chars_received
        self.billed = self.billed + usage

    def add_hit(self, usage: TokenUsage | None) -> None:
        """Charge one cache hit."""
        self.cache_hits += 1
        if usage is not None:
            self.replayed = self.replayed + usage

Attributes

billed = Field(default_factory=TokenUsage, description='Tokens from live provider calls.') class-attribute instance-attribute

cache_hits = Field(default=0, description='Responses served from cache.') class-attribute instance-attribute

calls = Field(default=0, description='Provider calls that returned.') class-attribute instance-attribute

calls_failed = Field(default=0, description='Provider calls that raised.') class-attribute instance-attribute

chars_received = Field(default=0, description='Response characters.') class-attribute instance-attribute

chars_sent = Field(default=0, description='Prompt characters sent.') class-attribute instance-attribute

json_repairs = Field(default=0, description='Bracket-kind repairs.') class-attribute instance-attribute

json_schema_fallbacks = Field(default=0, description='Switches from json_schema to json_object.') class-attribute instance-attribute

parse_abandoned = Field(default=0, description='Requests given up on.') class-attribute instance-attribute

parse_retries = Field(default=0, description='Re-asks after bad JSON.') class-attribute instance-attribute

rate_limited = Field(default=0, description='Subset of calls_failed.') class-attribute instance-attribute

refused = Field(default=0, description='Replies the model refused.') class-attribute instance-attribute

rejected = Field(default=0, description='Subset of calls_failed: configuration faults.') class-attribute instance-attribute

replayed = Field(default_factory=TokenUsage, description='Tokens of responses served from cache.') class-attribute instance-attribute

timeouts = Field(default=0, description='Subset of calls_failed.') class-attribute instance-attribute

truncated = Field(default=0, description='Replies cut off at the output token limit.') class-attribute instance-attribute

Methods:

add_call(chars_sent, chars_received, usage)

Charge one live provider call.

Source code in lpgcast/llm/usage.py

def add_call(self, chars_sent: int, chars_received: int, usage: TokenUsage) -> None:
    """Charge one live provider call."""
    self.calls += 1
    self.chars_sent += chars_sent
    self.chars_received += chars_received
    self.billed = self.billed + usage

add_hit(usage)

Charge one cache hit.

Source code in lpgcast/llm/usage.py

def add_hit(self, usage: TokenUsage | None) -> None:
    """Charge one cache hit."""
    self.cache_hits += 1
    if usage is not None:
        self.replayed = self.replayed + usage

incr(counter, n=1)

Add n to an integer counter.

Source code in lpgcast/llm/usage.py

def incr(self, counter: str, n: int = 1) -> None:
    """Add *n* to an integer counter."""
    setattr(self, counter, getattr(self, counter) + n)

Functions:

cache_key(messages, json_schema, settings)

SHA-256 of the logical request.

Keyed on the messages the caller passed, not on the wire request: a json_object fallback adds a schema message, and a replay must hit whichever format served the original run.

Source code in lpgcast/llm/cache.py

def cache_key(
    messages: Sequence[Mapping[str, Any]],
    json_schema: dict[str, Any],
    settings: LLMSettings,
) -> str:
    """SHA-256 of the logical request.

    Keyed on the messages the caller passed, not on the wire request: a
    ``json_object`` fallback adds a schema message, and a replay must hit
    whichever format served the original run.
    """
    payload = {
        "messages": [dict(message) for message in messages],
        "json_schema": json_schema,
        "config": cache_key_config(settings),
    }
    canonical = json.dumps(
        payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

parse_json_object(text)

Parse a reply as one JSON object.

Strict parse first; then the first fenced block; then the bracket-kind repair. Raw control characters inside strings are tolerated.

Returns:

Type	Description
tuple[dict[str, Any], int]	The object, and the number of brackets repaired (0 if none).

Raises:

Type	Description
LLMJsonParseError	with the strict error's position and context.

Source code in lpgcast/llm/jsonparse.py

def parse_json_object(text: str) -> tuple[dict[str, Any], int]:
    """Parse a reply as one JSON object.

    Strict parse first; then the first fenced block; then the bracket-kind
    repair. Raw control characters inside strings are tolerated.

    Returns:
        The object, and the number of brackets repaired (0 if none).

    Raises:
        LLMJsonParseError: with the strict error's position and context.
    """
    text = sanitize_json_text(text)
    try:
        return _loads_object(text), 0
    except json.JSONDecodeError as strict_error:
        reported = strict_error
        match = _JSON_FENCE_RE.search(text)
        if match and match.group(1):
            fenced = match.group(1)
            try:
                return _loads_object(fenced), 0
            except json.JSONDecodeError as fence_error:
                text, reported = fenced, fence_error

        repaired, fixes = repair_bracket_kinds(text)
        if fixes:
            try:
                return _loads_object(repaired), fixes
            except json.JSONDecodeError:
                pass

        raise LLMJsonParseError(
            _format_json_error(text, reported),
            msg=reported.msg,
            pos=reported.pos,
            doc=text,
        ) from strict_error

track_usage(tracker=None)

Charge calls made in this block (and tasks it spawns) to tracker too.

Task-local: concurrent episodes each charge their own tracker.

Source code in lpgcast/llm/usage.py

@contextmanager
def track_usage(tracker: UsageTracker | None = None) -> Generator[UsageTracker]:
    """Charge calls made in this block (and tasks it spawns) to *tracker* too.

    Task-local: concurrent episodes each charge their own tracker.
    """
    tracker = tracker if tracker is not None else UsageTracker()
    token = _active_tracker.set(tracker)
    try:
        yield tracker
    finally:
        _active_tracker.reset(token)