Utils

zenpyre.utils ¶

Shared helper utilities.

zenpyre.utils.config ¶

Contain configurations.

zenpyre.utils.config.BaseConfig ¶

Bases: ABC

Define the interface shared by all configuration classes.

Concrete subclasses only need to implement :meth:to_kwargs and :meth:get_value; :meth:cache_key is derived from :meth:to_kwargs automatically.

Example

>>> from zenpyre.utils.config import BaseConfig, MISSING
>>> class ChatModelConfig(BaseConfig):
...     def __init__(self, model: str, temperature: float | None = None):
...         self.model = model
...         self.temperature = temperature
...     def get_value(self, name: str, default: Any = MISSING) -> Any:
...         kwargs = self.to_kwargs()
...         if name in kwargs:
...             return kwargs[name]
...         if default is not MISSING:
...             return default
...         raise KeyError(name)
...     def to_kwargs(self) -> dict:
...         return {"model": self.model, "temperature": self.temperature}
...
>>> cfg = ChatModelConfig(model="gpt-4", temperature=0.2)
>>> isinstance(cfg, BaseConfig)
True

zenpyre.utils.config.BaseConfig.cache_key ¶

cache_key(length: int = 64) -> str

Return a stable hash string derived from the current configuration.

Walks the to_kwargs() output and replaces any nested :class:BaseConfig instance with its own cache_key() string, then hashes the result via :func:coola.hashing.hash_object. The substitution step means two structurally-equal-but-distinct nested configs (e.g. two separately constructed ChatModelConfig objects with the same fields) yield the same key, since the nested config contributes its content rather than its object identity.

:func:hash_object canonicalizes its input (e.g. via sorted keys) before hashing, so two configs with identical (post-substitution) to_kwargs() output always produce the same hash regardless of field ordering.

Note that this only covers whatever to_kwargs() returns. A subclass that adds a field must also include it in to_kwargs() for it to affect the cache key.

Useful for cache keys, output filenames, or detecting configuration changes between runs without comparing each field manually.

Parameters:

Name	Type	Description	Default
`length`	`int`	The desired length of the returned hash string.	`64`

Returns:

Type	Description
`str`	A stable hash string, `length` characters long.

Example

>>> from zenpyre.utils.config import BaseConfig, MISSING
>>> class ChatModelConfig(BaseConfig):
...     def __init__(self, model: str, temperature: float | None = None):
...         self.model = model
...         self.temperature = temperature
...     def get_value(self, name: str, default: Any = MISSING) -> Any:
...         kwargs = self.to_kwargs()
...         if name in kwargs:
...             return kwargs[name]
...         if default is not MISSING:
...             return default
...         raise KeyError(name)
...     def to_kwargs(self) -> dict:
...         return {"model": self.model, "temperature": self.temperature}
...
>>> cfg = ChatModelConfig(model="gpt-4", temperature=0.2)
>>> key = cfg.cache_key()
>>> len(key)
64
>>> cfg.cache_key() == cfg.cache_key()
True

zenpyre.utils.config.BaseConfig.get_value `abstractmethod` ¶

get_value(name: str, default: Any = MISSING) -> Any

Get the value of a configuration field by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the field to look up.	required
`default`	`Any`	The value to return if `name` is not present. If omitted, a missing `name` raises `KeyError` instead. Since `None` is a valid field value, omitting `default` is the only way to distinguish "not present" from "present but set to `None`" — passing `default=None` explicitly means a missing field silently returns `None` instead of raising.	`MISSING`

Returns:

Type	Description
`Any`	The value associated with `name`, or `default` if not present and `default` was given.

Raises:

Type	Description
`KeyError`	If `name` is not present in this config and no `default` was given.

Example

>>> from zenpyre.utils.config import BaseConfig, MISSING
>>> class ChatModelConfig(BaseConfig):
...     def __init__(self, model: str, temperature: float | None = None):
...         self.model = model
...         self.temperature = temperature
...     def get_value(self, name: str, default: Any = MISSING) -> Any:
...         kwargs = self.to_kwargs()
...         if name in kwargs:
...             return kwargs[name]
...         if default is not MISSING:
...             return default
...         raise KeyError(name)
...     def to_kwargs(self) -> dict:
...         return {"model": self.model, "temperature": self.temperature}
...
>>> cfg = ChatModelConfig(model="gpt-4", temperature=0.2)
>>> cfg.get_value("model")
'gpt-4'
>>> cfg.get_value("missing_key", default=42)
42

zenpyre.utils.config.BaseConfig.to_kwargs `abstractmethod` ¶

to_kwargs() -> dict[str, Any]

Return the configuration as a flat dict of keyword arguments.

This is the single source of truth for the configuration's content: it is used both to construct/invoke the chat model and, via :meth:cache_key, to derive a stable hash. Any field that should affect caching or equality must be included here.

Returns:

Type	Description
`dict[str, Any]`	A dict of keyword arguments representing this configuration.

Example

>>> from zenpyre.utils.config import BaseConfig, MISSING
>>> class ChatModelConfig(BaseConfig):
...     def __init__(self, model: str, temperature: float | None = None):
...         self.model = model
...         self.temperature = temperature
...     def get_value(self, name: str, default: Any = MISSING) -> Any:
...         kwargs = self.to_kwargs()
...         if name in kwargs:
...             return kwargs[name]
...         if default is not MISSING:
...             return default
...         raise KeyError(name)
...     def to_kwargs(self) -> dict:
...         return {"model": self.model, "temperature": self.temperature}
...
>>> cfg = ChatModelConfig(model="gpt-4", temperature=0.2)
>>> cfg.to_kwargs()
{'model': 'gpt-4', 'temperature': 0.2}

zenpyre.utils.config.Config `dataclass` ¶

Bases: ExtraFieldsConfig

A generic configuration.

zenpyre.utils.config.ExtraFieldsConfig `dataclass` ¶

Bases: BaseConfig

Base for configs that merge arbitrary keyword arguments ("extra") into :meth:to_kwargs, alongside whatever typed fields a subclass declares.

Subclasses just need to be frozen dataclasses with their own typed fields; :meth:to_kwargs, :meth:from_kwargs, the extra/field-name collision check, and __hash__ are all inherited from here and work automatically via introspection (:func:dataclasses.fields) — no per-subclass overriding required.

extra is declared keyword-only (kw_only=True on this class) specifically so that subclasses are free to add non-defaulted, positional-or-keyword fields of their own without hitting dataclass's "non-default argument follows default argument" error, which would otherwise apply because extra (with its default) is defined here in the base class.

Attributes:

Name	Type	Description
`extra`	`dict[str, Any]`	Additional keyword arguments merged into :meth:`to_kwargs`. Must not contain a key that collides with any of this config's own field names (including ones declared by a subclass).

zenpyre.utils.config.ExtraFieldsConfig.from_kwargs `classmethod` ¶

from_kwargs(**kwargs: Any) -> Self

Construct a config from arbitrary keyword arguments, routing each one to a real field if it matches one, or to extra otherwise.

A convenience alternative to the regular constructor's extra={...} dict, letting callers pass extra fields directly as keyword arguments instead of building the dict themselves. Which keys count as "real fields" is determined by introspection (:func:dataclasses.fields) on cls, so this works automatically for any subclass without needing its own override — including subclasses that add their own typed fields.

Parameters:

Name	Type	Description	Default
`**kwargs`	`Any`	Keyword arguments to route: keys matching one of this class's own dataclass field names (other than `extra`) are passed through as that field's value; any remaining keys are collected into `extra`.	`{}`

Returns:

Type	Description
`Self`	A new instance of `cls`.

Raises:

Type Description

ValueError

If, after routing, a key intended for extra collides with one of this config's own field names — this can't actually happen through normal use of this method (a key either matches a field name and is routed there, or it doesn't and goes to extra, so a collision would require the field-name set to change between routing and construction). Documented for completeness rather than as a realistic failure mode.

zenpyre.utils.config.ExtraFieldsConfig.get_value ¶

get_value(name: str, default: Any = MISSING) -> Any

Get the value of a dataclass field (or extra entry) by name.

Checks this config's own dataclass fields first (via getattr, without needing to build the full :meth:to_kwargs dict just for a single lookup), then falls back to extra for keys that only live there.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the field (or `extra` key) to look up.	required
`default`	`Any`	The value to return if `name` is not present. If omitted, a missing `name` raises `KeyError` instead. Since `None` is a valid field value, omitting `default` is the only way to distinguish "not present" from "present but set to `None`" — passing `default=None` explicitly means a missing field silently returns `None` instead of raising.	`MISSING`

Returns:

Type	Description
`Any`	The value associated with `name`, or `default` if not present and `default` was given.

Raises:

Type	Description
`KeyError`	If `name` is not present in this config and no `default` was given.

zenpyre.utils.config.ExtraFieldsConfig.to_kwargs ¶

to_kwargs() -> dict[str, Any]

Return every dataclass field (including ones declared by a subclass) merged with extra, as a flat dict.

Fields are collected via introspection (:func:dataclasses.fields) rather than hardcoded by name, so a subclass that adds a new typed field gets it included here automatically, with no need to override this method.

Returns:

Type	Description
`dict[str, Any]`	A dict of keyword arguments representing this configuration.

zenpyre.utils.dataclass ¶

Contain dataclass utilities.

zenpyre.utils.dataclass.dataclasses_to_dataframe ¶

dataclasses_to_dataframe(items: list[Any]) -> DataFrame

Convert a list of dataclass instances into a Polars DataFrame.

Serialises each item to a dict via :func:dataclasses.asdict and builds a :class:polars.DataFrame from the resulting list of dicts. Works with any dataclass type, including frozen dataclasses, as long as all field values are compatible with Polars' type inference.

Parameters:

Name	Type	Description	Default
`items`	`list[Any]`	The list of dataclass instances to convert.	required

Returns:

Type	Description
`DataFrame`	A `polars.DataFrame` with one row per item and one column
`DataFrame`	per field.

Raises:

Type	Description
`TypeError`	If any item in `items` is not a dataclass instance.

Example

>>> from dataclasses import dataclass
>>> from zenpyre.utils.dataclass import dataclasses_to_dataframe
>>> @dataclass(frozen=True)
... class Point:
...     x: int
...     y: int
...
>>> frame = dataclasses_to_dataframe([Point(1, 2), Point(3, 4)])
>>> frame
shape: (2, 2)
┌─────┬─────┐
│ x   ┆ y   │
│ --- ┆ --- │
│ i64 ┆ i64 │
╞═════╪═════╡
│ 1   ┆ 2   │
│ 3   ┆ 4   │
└─────┴─────┘

zenpyre.utils.dataclass.load_dataclasses ¶

load_dataclasses(
    path: Path | str, cls: type[T]
) -> list[T]

Load a list of dataclass instances from a JSON file.

Reads a JSON array of objects from path and converts each one into an instance of cls by unpacking its keys as keyword arguments.

Parameters:

Name	Type	Description	Default
`path`	`Path \| str`	The source file path, as previously written by :func:`save_dataclasses`.	required
`cls`	`type[T]`	The dataclass type to reconstruct each entry as.	required

Returns:

Type	Description
`list[T]`	A list of `cls` instances loaded from `path`.

Raises:

Type	Description
`FileNotFoundError`	If `path` does not exist.
`TypeError`	If the JSON content is not a list.
`ValueError`	If an entry in the list is missing a required field or has an unexpected field for `cls`.

Example

>>> from dataclasses import dataclass
>>> from zenpyre.utils.dataclass import load_dataclasses
>>> @dataclass(frozen=True)
... class Point:
...     x: int
...     y: int
...
>>> points = load_dataclasses("points.json", Point)  # doctest: +SKIP

zenpyre.utils.dataclass.save_dataclasses ¶

save_dataclasses(
    items: list[Any],
    path: Path | str,
    *,
    exist_ok: bool = False
) -> None

Save a list of dataclass instances to a JSON file.

Serialises each item to a dict via :func:dataclasses.asdict and writes the resulting list to path as a JSON array. Works with any dataclass type, including frozen dataclasses, as long as all field values are JSON-serialisable.

Parameters:

Name	Type	Description	Default
`items`	`list[Any]`	The list of dataclass instances to save.	required
`path`	`Path \| str`	The destination file path.	required
`exist_ok`	`bool`	If `exist_ok` is `False` (the default), `FileExistsError` is raised if the target file already exists. If `exist_ok` is `True`, `FileExistsError` will not be raised unless the given path already exists in the file system and is not a file.	`False`

Raises:

Type	Description
`TypeError`	If any item in `items` is not a dataclass instance.
`FileExistsError`	If `path` already exists and `exist_ok=False`.
`OSError`	If `path` cannot be written to, or if `path` exists as a directory (regardless of `exist_ok`).

Example

>>> from dataclasses import dataclass
>>> from zenpyre.utils.dataclass import save_dataclasses
>>> @dataclass(frozen=True)
... class Point:
...     x: int
...     y: int
...
>>> points = save_dataclasses([Point(1, 2), Point(3, 4)])  # doctest: +SKIP

zenpyre.utils.duckdb ¶

Contain DuckDB utility functions.

zenpyre.utils.duckdb.prepare_duckdb_path ¶

prepare_duckdb_path(path: Path | str) -> Path | str

Prepare a path for use with duckdb.connect.

If path is the special in-memory sentinel (":memory:"), it is returned unchanged. Otherwise, path is sanitized and its parent directory is created if it does not already exist, so that DuckDB can create the database file without failing on a missing directory.

Parameters:

Name	Type	Description	Default
`path`	`Path \| str`	The DuckDB connection target -- either the in-memory sentinel `":memory:"`, or a filesystem path to a database file.	required

Returns:

Type	Description
`Path \| str`	`":memory:"` unchanged, or the sanitized, absolute `Path`
`Path \| str`	whose parent directory is guaranteed to exist.

zenpyre.utils.fallback ¶

Fallback helpers used when optional dependencies are unavailable.

zenpyre.utils.hashing ¶

Provide UUID hashing utilities for Python dictionaries.

zenpyre.utils.hashing.hash_dict_uuid ¶

hash_dict_uuid(data: dict[str, Any]) -> str

Compute a stable, reproducible UUID for a Python dictionary.

Serialises data via :func:json.dumps with sort_keys=True to guarantee a consistent ordering regardless of dict insertion order, then derives a deterministic UUID using :func:uuid.uuid5 with a fixed project-specific namespace.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	The dictionary to hash. All values must be JSON-serialisable.	required

Returns:

Type	Description
`str`	A lowercase UUID string of the form
`str`	`'xxxxxxxx-xxxx-5xxx-xxxx-xxxxxxxxxxxx'`.

Raises:

Type	Description
`TypeError`	If any value in `data` is not JSON-serialisable.

Example

>>> from zenpyre.utils.hashing import hash_dict_uuid
>>> hash_dict_uuid({"source": "cats.txt", "page": 1})  # doctest: +ELLIPSIS
'...'
>>> hash_dict_uuid({"page": 1, "source": "cats.txt"}) == hash_dict_uuid(
...     {"source": "cats.txt", "page": 1}
... )
True

zenpyre.utils.imports ¶

Helpers to detect and validate optional dependencies.

zenpyre.utils.imports.check_duckdb ¶

check_duckdb() -> None

Check if the duckdb package is installed.

Raises:

Type	Description
`RuntimeError`	if the `duckdb` package is not installed.

Example

>>> from zenpyre.utils.imports import check_duckdb
>>> check_duckdb()

zenpyre.utils.imports.check_faker ¶

check_faker() -> None

Check if the faker package is installed.

Raises:

Type	Description
`RuntimeError`	if the `faker` package is not installed.

Example

>>> from zenpyre.utils.imports import check_faker
>>> check_faker()

zenpyre.utils.imports.check_langchain_anthropic ¶

check_langchain_anthropic() -> None

Check if the langchain_anthropic package is installed.

Raises:

Type	Description
`RuntimeError`	if the `langchain_anthropic` package is not installed.

Example

>>> from zenpyre.utils.imports import check_langchain_anthropic
>>> check_langchain_anthropic()

zenpyre.utils.imports.check_langchain_chroma ¶

check_langchain_chroma() -> None

Check if the langchain_chroma package is installed.

Raises:

Type	Description
`RuntimeError`	if the `langchain_chroma` package is not installed.

Example

>>> from zenpyre.utils.imports import check_langchain_chroma
>>> check_langchain_chroma()

zenpyre.utils.imports.check_langchain_google_genai ¶

check_langchain_google_genai() -> None

Check if the langchain_google_genai package is installed.

Raises:

Type	Description
`RuntimeError`	if the `langchain_google_genai` package is not installed.

Example

>>> from zenpyre.utils.imports import check_langchain_google_genai
>>> check_langchain_google_genai()

zenpyre.utils.imports.check_langchain_huggingface ¶

check_langchain_huggingface() -> None

Check if the langchain_huggingface package is installed.

Raises:

Type	Description
`RuntimeError`	if the `langchain_huggingface` package is not installed.

Example

>>> from zenpyre.utils.imports import check_langchain_huggingface
>>> check_langchain_huggingface()

zenpyre.utils.imports.check_langchain_ollama ¶

check_langchain_ollama() -> None

Check if the langchain_ollama package is installed.

Raises:

Type	Description
`RuntimeError`	if the `langchain_ollama` package is not installed.

Example

>>> from zenpyre.utils.imports import check_langchain_ollama
>>> check_langchain_ollama()

zenpyre.utils.imports.check_langchain_openai ¶

check_langchain_openai() -> None

Check if the langchain_openai package is installed.

Raises:

Type	Description
`RuntimeError`	if the `langchain_openai` package is not installed.

Example

>>> from zenpyre.utils.imports import check_langchain_openai
>>> check_langchain_openai()

zenpyre.utils.imports.check_langchain_text_splitters ¶

check_langchain_text_splitters() -> None

Check if the langchain_text_splitters package is installed.

Raises:

Type	Description
`RuntimeError`	if the `langchain_text_splitters` package is not installed.

Example

>>> from zenpyre.utils.imports import check_langchain_text_splitters
>>> check_langchain_text_splitters()

zenpyre.utils.imports.check_persista ¶

check_persista() -> None

Check if the persista package is installed.

Raises:

Type	Description
`RuntimeError`	if the `persista` package is not installed.

Example

>>> from zenpyre.utils.imports import check_persista
>>> check_persista()

zenpyre.utils.imports.duckdb_available ¶

duckdb_available(fn: F) -> F

Implement a decorator to execute a function only if duckdb package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `duckdb` package is installed, otherwise `None`.

Example

>>> from zenpyre.utils.imports import duckdb_available
>>> @duckdb_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

zenpyre.utils.imports.faker_available ¶

faker_available(fn: F) -> F

Implement a decorator to execute a function only if faker package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `faker` package is installed, otherwise `None`.

Example

>>> from zenpyre.utils.imports import faker_available
>>> @faker_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

zenpyre.utils.imports.is_duckdb_available `cached` ¶

is_duckdb_available() -> bool

Indicate if the duckdb package is installed or not.

Returns:

Type	Description
`bool`	`True` if `duckdb` is available otherwise `False`.

Example

>>> from zenpyre.utils.imports import is_duckdb_available
>>> is_duckdb_available()

zenpyre.utils.imports.is_faker_available `cached` ¶

is_faker_available() -> bool

Indicate if the faker package is installed or not.

Returns:

Type	Description
`bool`	`True` if `faker` is available otherwise `False`.

Example

>>> from zenpyre.utils.imports import is_faker_available
>>> is_faker_available()

zenpyre.utils.imports.is_langchain_anthropic_available `cached` ¶

is_langchain_anthropic_available() -> bool

Indicate if the langchain_anthropic package is installed or not.

Returns:

Type	Description
`bool`	`True` if `langchain_anthropic` is available otherwise `False`.

Example

>>> from zenpyre.utils.imports import is_langchain_anthropic_available
>>> is_langchain_anthropic_available()

zenpyre.utils.imports.is_langchain_chroma_available `cached` ¶

is_langchain_chroma_available() -> bool

Indicate if the langchain_chroma package is installed or not.

Returns:

Type	Description
`bool`	`True` if `langchain_chroma` is available otherwise `False`.

Example

>>> from zenpyre.utils.imports import is_langchain_chroma_available
>>> is_langchain_chroma_available()

zenpyre.utils.imports.is_langchain_google_genai_available `cached` ¶

is_langchain_google_genai_available() -> bool

Indicate if the langchain_google_genai package is installed or not.

Returns:

Type	Description
`bool`	`True` if `langchain_google_genai` is available otherwise `False`.

Example

>>> from zenpyre.utils.imports import is_langchain_google_genai_available
>>> is_langchain_google_genai_available()

zenpyre.utils.imports.is_langchain_huggingface_available `cached` ¶

is_langchain_huggingface_available() -> bool

Indicate if the langchain_huggingface package is installed or not.

Returns:

Type	Description
`bool`	`True` if `langchain_huggingface` is available otherwise `False`.

Example

>>> from zenpyre.utils.imports import is_langchain_huggingface_available
>>> is_langchain_huggingface_available()

zenpyre.utils.imports.is_langchain_ollama_available `cached` ¶

is_langchain_ollama_available() -> bool

Indicate if the langchain_ollama package is installed or not.

Returns:

Type	Description
`bool`	`True` if `langchain_ollama` is available otherwise `False`.

Example

>>> from zenpyre.utils.imports import is_langchain_ollama_available
>>> is_langchain_ollama_available()

zenpyre.utils.imports.is_langchain_openai_available `cached` ¶

is_langchain_openai_available() -> bool

Indicate if the langchain_openai package is installed or not.

Returns:

Type	Description
`bool`	`True` if `langchain_openai` is available otherwise `False`.

Example

>>> from zenpyre.utils.imports import is_langchain_openai_available
>>> is_langchain_openai_available()

zenpyre.utils.imports.is_langchain_text_splitters_available `cached` ¶

is_langchain_text_splitters_available() -> bool

Indicate if the langchain_text_splitters package is installed or not.

Returns:

Type	Description
`bool`	`True` if `langchain_text_splitters` is available otherwise `False`.

Example

>>> from zenpyre.utils.imports import is_langchain_text_splitters_available
>>> is_langchain_text_splitters_available()

zenpyre.utils.imports.is_persista_available `cached` ¶

is_persista_available() -> bool

Indicate if the persista package is installed or not.

Returns:

Type	Description
`bool`	`True` if `persista` is available otherwise `False`.

Example

>>> from zenpyre.utils.imports import is_persista_available
>>> is_persista_available()

zenpyre.utils.imports.langchain_anthropic_available ¶

langchain_anthropic_available(fn: F) -> F

Implement a decorator to execute a function only if langchain_anthropic package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `langchain_anthropic` package is installed, otherwise `None`.

Example

>>> from zenpyre.utils.imports import langchain_anthropic_available
>>> @langchain_anthropic_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

zenpyre.utils.imports.langchain_chroma_available ¶

langchain_chroma_available(fn: F) -> F

Implement a decorator to execute a function only if langchain_chroma package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `langchain_chroma` package is installed, otherwise `None`.

Example

>>> from zenpyre.utils.imports import langchain_chroma_available
>>> @langchain_chroma_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

zenpyre.utils.imports.langchain_google_genai_available ¶

langchain_google_genai_available(fn: F) -> F

Implement a decorator to execute a function only if langchain_google_genai package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `langchain_google_genai` package is installed, otherwise `None`.

Example

>>> from zenpyre.utils.imports import langchain_google_genai_available
>>> @langchain_google_genai_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

zenpyre.utils.imports.langchain_huggingface_available ¶

langchain_huggingface_available(fn: F) -> F

Implement a decorator to execute a function only if langchain_huggingface package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `langchain_huggingface` package is installed, otherwise `None`.

Example

>>> from zenpyre.utils.imports import langchain_huggingface_available
>>> @langchain_huggingface_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

zenpyre.utils.imports.langchain_ollama_available ¶

langchain_ollama_available(fn: F) -> F

Implement a decorator to execute a function only if langchain_ollama package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `langchain_ollama` package is installed, otherwise `None`.

Example

>>> from zenpyre.utils.imports import langchain_ollama_available
>>> @langchain_ollama_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

zenpyre.utils.imports.langchain_openai_available ¶

langchain_openai_available(fn: F) -> F

Implement a decorator to execute a function only if langchain_openai package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `langchain_openai` package is installed, otherwise `None`.

Example

>>> from zenpyre.utils.imports import langchain_openai_available
>>> @langchain_openai_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

zenpyre.utils.imports.langchain_text_splitters_available ¶

langchain_text_splitters_available(fn: F) -> F

Implement a decorator to execute a function only if langchain_text_splitters package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `langchain_text_splitters` package is installed, otherwise `None`.

Example

>>> from zenpyre.utils.imports import langchain_text_splitters_available
>>> @langchain_text_splitters_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

zenpyre.utils.imports.persista_available ¶

persista_available(fn: F) -> F

Implement a decorator to execute a function only if persista package is installed.

Parameters:

Name	Type	Description	Default
`fn`	`F`	The function to execute.	required

Returns:

Type	Description
`F`	A wrapper around `fn` if `persista` package is installed, otherwise `None`.

Example

>>> from zenpyre.utils.imports import persista_available
>>> @persista_available
... def my_function(n: int = 0) -> int:
...     return 42 + n
...
>>> my_function()

zenpyre.utils.imports.raise_duckdb_missing_error ¶

raise_duckdb_missing_error() -> NoReturn

Raise a RuntimeError to indicate the duckdb package is missing.

zenpyre.utils.imports.raise_faker_missing_error ¶

raise_faker_missing_error() -> NoReturn

Raise a RuntimeError to indicate the faker package is missing.

zenpyre.utils.imports.raise_langchain_anthropic_missing_error ¶

raise_langchain_anthropic_missing_error() -> NoReturn

Raise a RuntimeError to indicate the langchain_anthropic package is missing.

zenpyre.utils.imports.raise_langchain_chroma_missing_error ¶

raise_langchain_chroma_missing_error() -> NoReturn

Raise a RuntimeError to indicate the langchain_chroma package is missing.

zenpyre.utils.imports.raise_langchain_google_genai_missing_error ¶

raise_langchain_google_genai_missing_error() -> NoReturn

Raise a RuntimeError to indicate the langchain_google_genai package is missing.

zenpyre.utils.imports.raise_langchain_huggingface_missing_error ¶

raise_langchain_huggingface_missing_error() -> NoReturn

Raise a RuntimeError to indicate the langchain_huggingface package is missing.

zenpyre.utils.imports.raise_langchain_ollama_missing_error ¶

raise_langchain_ollama_missing_error() -> NoReturn

Raise a RuntimeError to indicate the langchain_ollama package is missing.

zenpyre.utils.imports.raise_langchain_openai_missing_error ¶

raise_langchain_openai_missing_error() -> NoReturn

Raise a RuntimeError to indicate the langchain_openai package is missing.

zenpyre.utils.imports.raise_langchain_text_splitters_missing_error ¶

raise_langchain_text_splitters_missing_error() -> NoReturn

Raise a RuntimeError to indicate the langchain_text_splitters package is missing.

zenpyre.utils.imports.raise_persista_missing_error ¶

raise_persista_missing_error() -> NoReturn

Raise a RuntimeError to indicate the persista package is missing.

zenpyre.utils.json_to_structured ¶

Utilities for parsing raw JSON text emitted by a language model into a validated Pydantic model.

This module exists to work around language models (particularly small or local models, e.g. via Ollama) that do not reliably support native structured-output or tool-calling enforcement. Instead of relying on a framework's structured-output mechanism (which typically requires the model to emit a tool call), these utilities take the model's raw text response -- which may contain a JSON object wrapped in markdown code fences, surrounded by commentary, or with minor syntactic issues like trailing commas -- and attempt to extract, clean, parse, and validate it against a target Pydantic schema.

Typical usage:

from pydantic import BaseModel, Field

class WeatherReport(BaseModel):
    city: str = Field(min_length=1)
    temperature_celsius: float
    condition: str

result = agent.invoke({"messages": [{"role": "user", "content": "..."}]})
raw_content = result["messages"][-1].content

report = parse_json_to_structured(raw_content, WeatherReport)

zenpyre.utils.json_to_structured.JsonStructuredOutputParseError ¶

Bases: Exception

Raised when a model's raw text output cannot be converted into the target schema.

This error covers two distinct failure modes, both surfaced through the same exception type for simplicity: 1. No valid JSON object could be extracted or decoded from the raw content. 2. A JSON object was successfully decoded, but it does not satisfy the target Pydantic schema (missing fields, wrong types, failed validators).

Attributes:

Name	Type	Description
`raw_content`		The original, unmodified text that failed to parse or validate. Callers can use this for logging, debugging, or triggering a retry with a modified prompt.

zenpyre.utils.json_to_structured.parse_json_to_structured ¶

parse_json_to_structured(
    content: str, schema: type[T]
) -> T

Parse a model's raw text output into a validated instance of a Pydantic schema.

This is the core conversion function: given the free-form text content of a language model's response, it attempts to locate a JSON object within that text, decode it, and validate it against schema.

Parsing proceeds through the following stages, stopping at the first stage that succeeds: 1. Strip any surrounding markdown code fence, then attempt json.loads directly on the result. 2. Apply common JSON syntax fixes (e.g. trailing comma removal) to the fence-stripped text, then attempt json.loads again. 3. Fall back to brace-matched extraction of the first top-level JSON object anywhere in the original content, apply the same syntax fixes, and attempt json.loads on the extracted substring.

If a JSON object is successfully decoded at any stage, it is then validated against schema via schema.model_validate.

Parameters:

Name	Type	Description	Default
`content`	`str`	The raw text content of a model's response. May be plain JSON, JSON wrapped in a markdown code fence, or JSON embedded alongside other commentary text.	required
`schema`	`type[T]`	The Pydantic model class to validate the decoded JSON against.	required

Returns:

Type	Description
`T`	An instance of `schema` populated and validated from the parsed JSON.

Raises:

Type	Description
`JsonStructuredOutputParseError`	If no valid JSON object can be decoded from `content` at all, or if a JSON object is decoded but fails validation against `schema`. In both cases, `content` is attached to the exception via its `raw_content` attribute.

zenpyre.utils.json_to_structured.parse_json_to_structured_with_retry ¶

parse_json_to_structured_with_retry(
    invoke_fn: Callable[[], str],
    schema: type[T],
    max_attempts: int = 3,
) -> T

Repeatedly invoke a model and parse its output until it matches a schema.

This provides a bounded retry loop around parse_json_to_structured for cases where a single generation may occasionally produce invalid or unparseable JSON (e.g. a non-deterministic small local model). Each attempt performs a fresh model invocation via invoke_fn -- this function does not reuse or repair a previous failed response, since re-generating is typically more reliable than patching malformed output.

Parameters:

Name	Type	Description	Default
`invoke_fn`	`Callable[[], str]`	A zero-argument callable that performs one full model or agent invocation and returns the raw text content of the response (e.g. `lambda: agent.invoke(...)["messages"][-1].content`). This function is called once per attempt, up to `max_attempts` times.	required
`schema`	`type[T]`	The Pydantic model class each attempt's output is validated against.	required
`max_attempts`	`int`	The maximum number of invoke-and-parse attempts before giving up. Must be at least 1.	`3`

Returns:

Type	Description
`T`	An instance of `schema` from the first attempt that parses and
`T`	validates successfully.

Raises:

Type	Description
`JsonStructuredOutputParseError`	If every attempt up to `max_attempts` fails to produce parseable, schema-valid JSON. The raised exception is the one from the final failed attempt, with its `raw_content` reflecting that last attempt's output.

zenpyre.utils.rich ¶

Common utilities for rich.

zenpyre.utils.rich.configure_rich_logging ¶

configure_rich_logging(
    level: int = INFO,
    *,
    fmt: str = "%(message)s",
    datefmt: str = "[%Y-%m-%d %H:%M:%S]",
    rich_tracebacks: bool = True,
    markup: bool = True,
    force: bool = False,
    **kwargs: Any
) -> None

Configure the root logger to use :class:rich.logging.RichHandler.

Sets up Python's standard :mod:logging module with a single :class:~rich.logging.RichHandler so that all log output is rendered through Rich's console, with coloured levels, timestamps, and optional pretty-printed tracebacks.

After calling this function, obtain a logger for the current module with the standard idiom::

logger = logging.getLogger(__name__)

Using :func:logging.getLogger with __name__ gives the logger a dotted name matching its import path (e.g. langchain_metrics.console), which slots it correctly into the logging hierarchy and allows callers to tune or silence it by name.

Note

:func:logging.basicConfig is a no-op if the root logger already has handlers configured (e.g. if another library called basicConfig first, or if you call this function more than once). Pass force=True to remove any existing handlers and reconfigure unconditionally.

Parameters:

Name	Type	Description	Default
`level`	`int`	Minimum log level for the root logger. Accepts any constant from :mod:`logging` (e.g. `logging.DEBUG`, `logging.WARNING`). Defaults to `logging.INFO`.	`INFO`
`fmt`	`str`	Log record format string passed to :class:`logging.Formatter`. Because Rich renders level, timestamp, and logger name itself, the default `"%(message)s"` avoids duplicating those fields. Change this if you need additional fields such as `%(name)s` in the formatted output.	`'%(message)s'`
`datefmt`	`str`	Date/time format string for the `%(asctime)s` field, following :func:`time.strftime` conventions. Defaults to `"[%Y-%m-%d %H:%M:%S]"`.	`'[%Y-%m-%d %H:%M:%S]'`
`rich_tracebacks`	`bool`	When `True`, exceptions are rendered by Rich as syntax-highlighted, multi-frame tracebacks instead of the standard Python traceback format. Defaults to `True`.	`True`
`markup`	`bool`	When `True`, Rich markup tags (e.g. `[bold red]`) in log messages are interpreted and rendered. Set to `False` if log messages may contain literal square brackets that should not be treated as markup. Defaults to `True`.	`True`
`force`	`bool`	When `True`, any existing handlers on the root logger are removed before applying the new configuration, ensuring this call always takes effect. Equivalent to passing `force=True` to :func:`logging.basicConfig` (requires Python 3.8+). Defaults to `False`.	`False`
`**kwargs`	`Any`	Additional keyword arguments forwarded verbatim to :class:`~rich.logging.RichHandler`. For example:: `configure_rich_logging(show_path=False, tracebacks_show_locals=True)` See the `Rich documentation <https://rich.readthedocs.io/en/latest/logging.html>`_ for the full list of supported parameters.	`{}`

Raises:

Type	Description
`TypeError`	If an unrecognised keyword argument is passed that :class:`~rich.logging.RichHandler` does not accept.

Example

>>> import logging
>>> from zenpyre.utils.rich import configure_rich_logging
>>> configure_rich_logging()
>>> logger = logging.getLogger(__name__)
>>> logger.info("Rich logging is active")

zenpyre.utils.rich.make_progressbar ¶

make_progressbar(*, transient: bool = False) -> Progress

Create a standardised Rich progress bar for use across the codebase.

Builds a Progress instance with a consistent column layout: - A spinner indicating the task is active. - Description text. - A bar showing percentage complete. - The percentage as a number (e.g. '42%'). - An M-of-N counter (e.g. '42/100'). - Elapsed time since the task started. - Estimated time remaining.

Using this factory ensures all progress bars in the codebase share the same appearance. Use as a context manager to start and stop rendering.

Parameters:

Name	Type	Description	Default
`transient`	`bool`	If True, the progress bar is cleared from the terminal when the context manager exits. Useful for short-lived tasks where the bar would clutter the output. Defaults to False.	`False`

Returns:

Type	Description
`Progress`	A configured `Progress` instance, ready to use as a context manager.

Example

>>> from zenpyre.utils.rich import make_progressbar
>>> rows = list(range(10))
>>> with make_progressbar() as progress:
...     task = progress.add_task("Processing papers", total=len(rows))
...     for row in rows:
...         print(row)
...         progress.advance(task)
...

zenpyre.utils.rich.make_spinner ¶

make_spinner(*, transient: bool = True) -> Progress

Create a Rich spinner for tasks where the total number of items is unknown.

Builds a :class:~rich.progress.Progress instance with a column layout suited for indeterminate tasks:

A spinner indicating the task is active.
Description text.
An M-of-N counter showing items processed so far (e.g. 42/?).
Processing speed (e.g. 12.3/s).
Elapsed time since the task started.

Parameters:

Name	Type	Description	Default
`transient`	`bool`	If `True`, the spinner is cleared from the terminal when the context manager exits. Useful for short-lived tasks where the spinner would clutter the output. Defaults to `True`.	`True`

Returns:

Type	Description
`Progress`	A configured :class:`~rich.progress.Progress` instance ready to
`Progress`	use as a context manager.

Example

>>> from zenpyre.utils.rich import make_spinner
>>> items = list(range(10))
>>> with make_spinner() as progress:
...     task = progress.add_task("Processing items...", total=None)
...     for item in items:
...         progress.advance(task)
...

zenpyre.utils.rich.print_document ¶

print_document(
    doc: Document,
    max_length: int = 500,
    console: Console | None = None,
    compact_metadata: bool = False,
) -> None

Pretty-print a LangChain document to the terminal using rich.

Renders the document as a bordered panel titled with its id, containing two nested panels: a content panel with the document's page_content (truncated on a word boundary and annotated with the omitted character count if it exceeds max_length, with the total/truncated character count shown in its subtitle), and, if metadata is non-empty, a metadata panel listing entries sorted by key.

Parameters:

Name	Type	Description	Default
`doc`	`Document`	The document to display.	required
`max_length`	`int`	Maximum number of content characters to display before truncating. Defaults to `500`.	`500`
`console`	`Console \| None`	An optional rich :class:`~rich.console.Console` to print to. If `None`, the current active console (as returned by :func:`rich.get_console`) is used.	`None`
`compact_metadata`	`bool`	If `True`, render metadata entries as a single dimmed inline line (`key: value · key: value`) instead of one per line. Useful when scanning many documents in a row and one metadata line per key would take up too much vertical space. Defaults to `False`.	`False`

zenpyre.utils.rich.print_documents_metadata ¶

print_documents_metadata(
    documents: Sequence[Document],
    separator: str = "•",
    console: Console | None = None,
) -> None

Pretty-print metadata for a sequence of documents, one line each.

Renders a bordered panel containing one line per document: the document's id (if present) followed by its metadata entries, sorted by key, rendered inline as key: value {separator} key: value. Documents with no metadata show a dimmed placeholder instead.

Parameters:

Name	Type	Description	Default
`documents`	`Sequence[Document]`	The documents whose metadata to display.	required
`separator`	`str`	The string used to separate metadata entries on each line. Defaults to `"·"`.	`'•'`
`console`	`Console \| None`	An optional rich :class:`~rich.console.Console` to print to. If `None`, the current active console (as returned by :func:`rich.get_console`) is used.	`None`

zenpyre.utils.rich.print_markdown ¶

print_markdown(
    msg: str,
    *,
    title: str | None = None,
    title_align: AlignMethod = "left",
    box: bool = True,
    panel: bool = True,
    max_length: int | None = None,
    console: Console | None = None
) -> None

Render a Markdown string to the console, optionally inside a Rich panel.

Prints msg as rendered Markdown, wrapped in a :class:~rich.panel.Panel unless panel is False. Uses the provided console if given, otherwise falls back to the shared instance.

Parameters:

Name	Type	Description	Default
`msg`	`str`	The Markdown content to render. May contain any Markdown syntax supported by :class:`~rich.markdown.Markdown`. An empty string renders a blank panel.	required
`title`	`str \| None`	Optional title displayed on the panel border. Pass `None` (the default) to render a panel with no title. Ignored if `panel` is `False`.	`None`
`title_align`	`AlignMethod`	Horizontal alignment of `title` along the panel border (`"left"`, `"center"`, or `"right"`). Defaults to `"left"`. Ignored if `panel` is `False`.	`'left'`
`box`	`bool`	If `True` (the default), render the panel with a visible rounded border. If `False`, render with a minimal (near-invisible) border. Ignored if `panel` is `False`.	`True`
`panel`	`bool`	If `True` (the default), wrap the rendered Markdown in a :class:`~rich.panel.Panel`. If `False`, print it directly and ignore `title`, `title_align`, and `box`.	`True`
`max_length`	`int \| None`	Maximum number of characters of `msg` to render before truncating on a word boundary, with the omitted character count appended. Pass `None` (the default) to render `msg` in full.	`None`
`console`	`Console \| None`	Optional :class:`~rich.console.Console` instance to use for this call. Overrides the shared default for this invocation only. Pass `None` (the default) to use the shared instance.	`None`

Example

>>> from zenpyre.utils.rich import print_markdown
>>> print_markdown("**hello**", title="Demo")
>>> print_markdown("**hello**", title="Demo", box=False)
>>> print_markdown("**hello**", panel=False)

zenpyre.utils.rich.print_pretty ¶

print_pretty(
    data: Any,
    *,
    title: str | None = None,
    title_align: AlignMethod = "left",
    box: bool = True,
    panel: bool = True,
    console: Console | None = None
) -> None

Render an arbitrary object to the console in a pretty format, optionally inside a Rich panel.

Prints data using :class:~rich.pretty.Pretty, wrapped in a :class:~rich.panel.Panel unless panel is False. Uses the provided console if given, otherwise falls back to the shared instance. Works with any Python object — dicts, lists, dataclasses, Pydantic models, and so on.

Parameters:

Name	Type	Description	Default
`data`	`Any`	The object to render. Passed directly to :class:`~rich.pretty.Pretty`, which handles formatting.	required
`title`	`str \| None`	Optional title displayed on the panel border. Pass `None` (the default) to render a panel with no title. Ignored if `panel` is `False`.	`None`
`title_align`	`AlignMethod`	Horizontal alignment of `title` along the panel border (`"left"`, `"center"`, or `"right"`). Defaults to `"left"`. Ignored if `panel` is `False`.	`'left'`
`box`	`bool`	If `True` (the default), render the panel with a visible rounded border. If `False`, render with a minimal (near-invisible) border. Ignored if `panel` is `False`.	`True`
`panel`	`bool`	If `True` (the default), wrap the rendered object in a :class:`~rich.panel.Panel`. If `False`, print it directly and ignore `title`, `title_align`, and `box`.	`True`
`console`	`Console \| None`	Optional :class:`~rich.console.Console` instance to use for this call. Overrides the shared default for this invocation only. Pass `None` (the default) to use the shared instance.	`None`

Example

>>> from zenpyre.utils.rich import print_pretty
>>> print_pretty({"key": "value"}, title="Demo")
>>> print_pretty({"key": "value"}, title="Demo", box=False)
>>> print_pretty({"key": "value"}, panel=False)

zenpyre.utils.token_usage ¶

Contain utilities to compute token usage.

zenpyre.utils.token_usage.accumulate_token_usage ¶

accumulate_token_usage(
    usage: dict[str, int], result: Any
) -> None

Add the token usage found in result to a running usage total, in place.

This is a convenience wrapper around :func:get_token_usage for callers that need to keep a running total across several calls (e.g. one entry per iteration of a loop), instead of summing a single batch upfront. result can be any shape accepted by :func:get_token_usage, such as a single BaseMessage, the dict returned by agent.invoke(...), or the list returned by agent.batch(...).

Parameters:

Name	Type	Description	Default
`usage`	`dict[str, int]`	The running token usage totals to update, in place. Missing keys are treated as `0`. Typically built up by repeated calls to this function, e.g. starting from `{}`.	required
`result`	`Any`	Any object potentially containing `AIMessage` instances, as accepted by :func:`get_token_usage`.	required

Example

>>> from langchain_core.messages import AIMessage, UsageMetadata
>>> from zenpyre.utils.token_usage import accumulate_token_usage
>>> usage = {}
>>> accumulate_token_usage(
...     usage,
...     AIMessage(
...         content="hi",
...         usage_metadata=UsageMetadata(input_tokens=10, output_tokens=5, total_tokens=15),
...     ),
... )
>>> usage
{'input_tokens': 10, 'output_tokens': 5, 'total_tokens': 15}

zenpyre.utils.token_usage.format_token_usage ¶

format_token_usage(usage: UsageMetadata) -> str

Format token usage as a human-readable string for terminal display.

Parameters:

Name	Type	Description	Default
`usage`	`UsageMetadata`	The token usage to format, as returned by :func:`~zenpyre.utils.tokens.get_invoke_token_usage` or :func:`~zenpyre.utils.tokens.get_batch_token_usage`.	required

Returns:

Type	Description
`str`	A single-line string summarizing input, output, and total
`str`	token counts, suitable for printing to a terminal.

Example

>>> from langchain_core.messages import UsageMetadata
>>> from zenpyre.utils.token_usage import format_token_usage
>>> usage = UsageMetadata(input_tokens=1234, output_tokens=567, total_tokens=1801)
>>> print(format_token_usage(usage))
[tokens] in=1,234 | out=567 | total=1,801

zenpyre.utils.token_usage.get_token_usage ¶

get_token_usage(result: Any) -> UsageMetadata

Sum token usage across every AI message found anywhere within result.

Unlike :func:get_invoke_token_usage and :func:get_batch_token_usage, this function does not branch on the shape of result. Instead it walks result depth-first (via coola's dfs_iterate) and collects every AIMessage instance found, regardless of how deeply it is nested. This makes it a convenient, shape-agnostic entry point that works for a single BaseMessage, the dict returned by agent.invoke(...), the list returned by agent.batch(...), or any other structure that contains AIMessage objects.

If result contains no AIMessage instances (including cases where result is of an unrecognized type, e.g. an int or a plain str), this function returns all-zero usage rather than raising an error.

Parameters:

Name	Type	Description	Default
`result`	`Any`	Any object potentially containing `AIMessage` instances, such as a `BaseMessage`, the dict returned by `agent.invoke(...)`, or the list returned by `agent.batch(...)`.	required

Returns:

Type	Description
`UsageMetadata`	A `UsageMetadata` dict with token counts summed across every `AIMessage` found within `result`.

zenpyre.utils.token_usage.log_token_usage ¶

log_token_usage(
    result: Any, *, only_if_nonzero: bool = True
) -> None

Log the token usage for a single invocation or a batch of them.

Parameters:

Name	Type	Description	Default
`result`	`Any`	The dict returned by `agent.invoke(...)`, or the list of dicts returned by `agent.batch(...)`.	required
`only_if_nonzero`	`bool`	If `True` (the default), nothing is logged when `result` has no `total_tokens` (i.e. `total_tokens == 0`), which is typically the case when `result` contains no `AIMessage` with usage metadata. Set to `False` to always log, even when `total_tokens` is zero.	`True`

Utils

zenpyre.utils ¶

zenpyre.utils.config ¶

zenpyre.utils.config.BaseConfig ¶

zenpyre.utils.config.BaseConfig.cache_key ¶

zenpyre.utils.config.BaseConfig.get_value abstractmethod ¶

zenpyre.utils.config.BaseConfig.to_kwargs abstractmethod ¶

zenpyre.utils.config.Config dataclass ¶

zenpyre.utils.config.ExtraFieldsConfig dataclass ¶

zenpyre.utils.config.ExtraFieldsConfig.from_kwargs classmethod ¶

zenpyre.utils.config.ExtraFieldsConfig.get_value ¶

zenpyre.utils.config.ExtraFieldsConfig.to_kwargs ¶

zenpyre.utils.dataclass ¶

zenpyre.utils.dataclass.dataclasses_to_dataframe ¶

zenpyre.utils.dataclass.load_dataclasses ¶

zenpyre.utils.dataclass.save_dataclasses ¶

zenpyre.utils.duckdb ¶

zenpyre.utils.duckdb.prepare_duckdb_path ¶

zenpyre.utils.fallback ¶

zenpyre.utils.hashing ¶

zenpyre.utils.hashing.hash_dict_uuid ¶

zenpyre.utils.imports ¶

zenpyre.utils.imports.check_duckdb ¶

zenpyre.utils.imports.check_faker ¶

zenpyre.utils.imports.check_langchain_anthropic ¶

zenpyre.utils.imports.check_langchain_chroma ¶

zenpyre.utils.imports.check_langchain_google_genai ¶

zenpyre.utils.imports.check_langchain_huggingface ¶

zenpyre.utils.imports.check_langchain_ollama ¶

zenpyre.utils.imports.check_langchain_openai ¶

zenpyre.utils.imports.check_langchain_text_splitters ¶

zenpyre.utils.imports.check_persista ¶

zenpyre.utils.imports.duckdb_available ¶

zenpyre.utils.imports.faker_available ¶

zenpyre.utils.imports.is_duckdb_available cached ¶

zenpyre.utils.imports.is_faker_available cached ¶

zenpyre.utils.imports.is_langchain_anthropic_available cached ¶

zenpyre.utils.imports.is_langchain_chroma_available cached ¶

zenpyre.utils.imports.is_langchain_google_genai_available cached ¶

zenpyre.utils.imports.is_langchain_huggingface_available cached ¶

zenpyre.utils.imports.is_langchain_ollama_available cached ¶

zenpyre.utils.imports.is_langchain_openai_available cached ¶

zenpyre.utils.imports.is_langchain_text_splitters_available cached ¶

zenpyre.utils.imports.is_persista_available cached ¶

zenpyre.utils.imports.langchain_anthropic_available ¶

zenpyre.utils.imports.langchain_chroma_available ¶

zenpyre.utils.imports.langchain_google_genai_available ¶

zenpyre.utils.imports.langchain_huggingface_available ¶

zenpyre.utils.imports.langchain_ollama_available ¶

zenpyre.utils.imports.langchain_openai_available ¶

zenpyre.utils.imports.langchain_text_splitters_available ¶

zenpyre.utils.imports.persista_available ¶

zenpyre.utils.imports.raise_duckdb_missing_error ¶

zenpyre.utils.imports.raise_faker_missing_error ¶

zenpyre.utils.imports.raise_langchain_anthropic_missing_error ¶

zenpyre.utils.imports.raise_langchain_chroma_missing_error ¶

zenpyre.utils.imports.raise_langchain_google_genai_missing_error ¶

zenpyre.utils.imports.raise_langchain_huggingface_missing_error ¶

zenpyre.utils.imports.raise_langchain_ollama_missing_error ¶

zenpyre.utils.imports.raise_langchain_openai_missing_error ¶

zenpyre.utils.imports.raise_langchain_text_splitters_missing_error ¶

zenpyre.utils.imports.raise_persista_missing_error ¶

zenpyre.utils.json_to_structured ¶

zenpyre.utils.json_to_structured.JsonStructuredOutputParseError ¶

zenpyre.utils.json_to_structured.parse_json_to_structured ¶

zenpyre.utils.json_to_structured.parse_json_to_structured_with_retry ¶

zenpyre.utils.rich ¶

zenpyre.utils.rich.configure_rich_logging ¶

zenpyre.utils.rich.make_progressbar ¶

zenpyre.utils.rich.make_spinner ¶

zenpyre.utils.rich.print_document ¶

zenpyre.utils.rich.print_documents_metadata ¶

zenpyre.utils.rich.print_markdown ¶

zenpyre.utils.rich.print_pretty ¶

zenpyre.utils.token_usage ¶

zenpyre.utils.token_usage.accumulate_token_usage ¶

zenpyre.utils.token_usage.format_token_usage ¶

zenpyre.utils.token_usage.get_token_usage ¶

zenpyre.utils.token_usage.log_token_usage ¶

zenpyre.utils.config.BaseConfig.get_value `abstractmethod` ¶

zenpyre.utils.config.BaseConfig.to_kwargs `abstractmethod` ¶

zenpyre.utils.config.Config `dataclass` ¶

zenpyre.utils.config.ExtraFieldsConfig `dataclass` ¶

zenpyre.utils.config.ExtraFieldsConfig.from_kwargs `classmethod` ¶

zenpyre.utils.imports.is_duckdb_available `cached` ¶

zenpyre.utils.imports.is_faker_available `cached` ¶

zenpyre.utils.imports.is_langchain_anthropic_available `cached` ¶

zenpyre.utils.imports.is_langchain_chroma_available `cached` ¶

zenpyre.utils.imports.is_langchain_google_genai_available `cached` ¶

zenpyre.utils.imports.is_langchain_huggingface_available `cached` ¶

zenpyre.utils.imports.is_langchain_ollama_available `cached` ¶

zenpyre.utils.imports.is_langchain_openai_available `cached` ¶

zenpyre.utils.imports.is_langchain_text_splitters_available `cached` ¶

zenpyre.utils.imports.is_persista_available `cached` ¶