Chat models

zenpyre.chat_models

Contain chat models.

zenpyre.chat_models.CachingChatModel

Bases: BaseChatModel

Wrap a chat model to cache its output, keyed by a hash of the messages, stop words, and call kwargs.

Unlike :class:~zenpyre.runnables.CachingRunnable, this class is a genuine :class:~langchain_core.language_models.BaseChatModel subclass -- it can be used anywhere a chat model is expected, and bind_tools returns another CachingChatModel wrapping the bound inner model, so caching keeps working after tools are bound.

On each call, a cache key is derived from (messages, stop, kwargs) and used to look up a previously cached ChatResult in response_cache. On a cache hit, the cached result is returned without calling the wrapped chat model. On a cache miss, the wrapped chat model is invoked and its result is stored in response_cache before being returned. If response_cache is None, caching is disabled entirely and every call goes straight to the wrapped chat model.

The field is named response_cache rather than cache because :class:~langchain_core.language_models.BaseChatModel already declares a cache: BaseCache | bool | None field for its own, unrelated built-in caching (langchain.cache); reusing that name would silently collide with it and break LangChain's cache-lookup hooks (_generate_with_cache / _agenerate_with_cache), which read self.cache expecting that type.

response_cache must already be open (via :meth:~persista.cache.Cache.open / :meth:~persista.cache.Cache.aopen, or used as a context manager) before it is passed in -- CachingChatModel does not manage its lifecycle, since the same cache instance is typically shared across multiple wrappers or callers.

Parameters:

Name	Type	Description	Default
chat_model		The chat model whose output should be cached.	required
response_cache		The :class:~persista.cache.Cache instance used to store cached results. If None, caching is disabled. The caller configures the cache's backing store and TTL; CachingChatModel has no caching policy of its own beyond what response_cache provides.	required
key_fn		A function that derives a cache key from a (messages, stop, kwargs) tuple. The returned string is used directly as the response_cache key. Defaults to hash_object, which dispatches through coola's hasher registry (e.g. using SerializableHasher for LangChain messages).	required

Example

>>> from langchain_core.language_models import FakeListChatModel
>>> from persista.cache import Cache
>>> from zenpyre.chat_models import CachingChatModel
>>> with Cache() as cache:
...     chat_model = CachingChatModel(
...         chat_model=FakeListChatModel(responses=["hello"]),
...         response_cache=cache,
...     )
...     chat_model.invoke("hi").content
...
'hello'

zenpyre.chat_models.ChatModelConfig dataclass

Bases: ExtraFieldsConfig

A generic chat model configuration.

to_kwargs(), the extra/field-name collision check, and __hash__ are all inherited from :class:~zenpyre.utils.config.ExtraFieldsConfig; this class only needs to declare its own typed field(s).

Parameters:

Name	Type	Description	Default
model	str	The model identifier.	required
extra	dict[str, Any]	Additional keyword arguments merged into :meth:to_kwargs. Must not contain a "model" key; use the model field for that.	dict()

Example

>>> from zenpyre.chat_models import ChatModelConfig
>>> cfg = ChatModelConfig(model="gpt-4", extra={"temperature": 0.2})
>>> cfg.model
'gpt-4'
>>> cfg.to_kwargs()
{'model': 'gpt-4', 'temperature': 0.2}

zenpyre.chat_models.ChatModelConfig.from_kwargs classmethod

from_kwargs(model: str, **kwargs: Any) -> Self

Construct a :class:ChatModelConfig from a model identifier and arbitrary keyword arguments.

A convenience alternative to the regular constructor's extra={...} dict, letting callers pass extra fields directly as keyword arguments instead: ChatModelConfig.from_kwargs("gpt-4", temperature=0.2) is equivalent to ChatModelConfig(model="gpt-4", extra={"temperature": 0.2}).

Parameters:

Name	Type	Description	Default
model	str	The model identifier.	required
**kwargs	Any	Additional keyword arguments, stored as extra.	{}

Returns:

Type	Description
Self	A new :class:ChatModelConfig.

Raises:

Type

Description

TypeError

If kwargs contains a "model" key, since Python's own argument binding intercepts it as a duplicate value for the explicit model parameter before this method's body ever runs. For example, from_kwargs("gpt-4", **{"model": "x"}) raises TypeError: got multiple values for argument 'model'. (This is distinct from the ValueError the regular constructor raises for the same conceptual conflict when extra is passed directly as an already-built dict; that path isn't reachable through this method.)

Example

>>> from zenpyre.chat_models import ChatModelConfig
>>> cfg = ChatModelConfig.from_kwargs("gpt-4", temperature=0.2)
>>> cfg.to_kwargs()
{'model': 'gpt-4', 'temperature': 0.2}

zenpyre.chat_models.ainvoke_structured_llm async

ainvoke_structured_llm(
    *,
    chat_model: BaseChatModel,
    output_type: type[BaseModel],
    system_prompt: str,
    user_content: str,
    timeblock_message: str = "LLM generated answer in {time}"
) -> tuple[Any, dict[str, Any]]

Invoke chat_model for a structured output_type response.

Shared by every agent's single-shot structured LLM call (adaptive query planning, assessment, judging): build the structured-output runnable, invoke it with a system/user message pair, and log token usage.

Parameters:

Name	Type	Description	Default
chat_model	BaseChatModel	LangChain chat model to invoke.	required
output_type	type[BaseModel]	Pydantic model the response must conform to.	required
system_prompt	str	System prompt for the call.	required
user_content	str	User message content for the call.	required
timeblock_message	str	Message template passed to :func:timeblock (e.g. "LLM generated AI risk assessment in {time}").	'LLM generated answer in {time}'

Returns:

Type	Description
Any	A tuple (parsed, raw_response) where parsed is the
dict[str, Any]	structured output (None if the LLM failed to produce it) and
tuple[Any, dict[str, Any]]	raw_response is the raw LangChain response dict, for callers
tuple[Any, dict[str, Any]]	that need to inspect parsing_error or log the raw content on
tuple[Any, dict[str, Any]]	failure.

zenpyre.chat_models.invoke_structured_llm

invoke_structured_llm(
    *,
    chat_model: BaseChatModel,
    output_type: type[BaseModel],
    system_prompt: str,
    user_content: str,
    timeblock_message: str = "LLM generated answer in {time}"
) -> tuple[Any, dict[str, Any]]

Invoke chat_model for a structured output_type response.

Shared by every agent's single-shot structured LLM call (adaptive query planning, assessment, judging): build the structured-output runnable, invoke it with a system/user message pair, and log token usage.

Parameters:

Name	Type	Description	Default
chat_model	BaseChatModel	LangChain chat model to invoke.	required
output_type	type[BaseModel]	Pydantic model the response must conform to.	required
system_prompt	str	System prompt for the call.	required
user_content	str	User message content for the call.	required
timeblock_message	str	Message template passed to :func:timeblock (e.g. "LLM generated AI risk assessment in {time}").	'LLM generated answer in {time}'

Returns:

Type	Description
Any	A tuple (parsed, raw_response) where parsed is the
dict[str, Any]	structured output (None if the LLM failed to produce it) and
tuple[Any, dict[str, Any]]	raw_response is the raw LangChain response dict, for callers
tuple[Any, dict[str, Any]]	that need to inspect parsing_error or log the raw content on
tuple[Any, dict[str, Any]]	failure.

zenpyre.chat_models.resolve_chat_model

resolve_chat_model(
    chat_model: BaseChatModel | dict[str, Any] | BaseConfig,
) -> BaseChatModel

Resolve a LangChain :class:~langchain_core.language_models.BaseChatModel instance from an existing object, a configuration dictionary, or a :class:~zenpyre.utils.config.BaseConfig.

If chat_model is already a :class:~langchain_core.language_models.BaseChatModel instance it is returned as-is. If it is a :class:dict or a :class:~zenpyre.utils.config.BaseConfig, it is treated as an objectory factory configuration and instantiated via :func:objectory.factory. See :func:~zenpyre.utils.resolve.resolve_object for details.

Parameters:

Name	Type	Description	Default
chat_model	BaseChatModel | dict[str, Any] | BaseConfig	Either a fully configured :class:~langchain_core.language_models.BaseChatModel instance, a :class:dict containing an objectory factory specification (must include a "_target_" key pointing to the fully-qualified class name), or a :class:~zenpyre.utils.config.BaseConfig whose to_kwargs() includes a "_target_" key.	required

Returns:

Type	Description
BaseChatModel	A configured
BaseChatModel	class:~langchain_core.language_models.BaseChatModel
BaseChatModel	instance.

Raises:

Type	Description
TypeError	If the resolved object is not a :class:~langchain_core.language_models.BaseChatModel instance.

Example

>>> from langchain_core.language_models import FakeListChatModel
>>> from zenpyre.chat_models import resolve_chat_model
>>> # From an existing instance:
>>> chat_model = resolve_chat_model(FakeListChatModel(responses=["hello"]))
>>> # From a configuration dictionary:
>>> chat_model = resolve_chat_model(  # doctest: +SKIP
...     {
...         "_target_": "langchain_core.language_models.FakeListChatModel",
...         "responses": ["hello"],
...     }
... )

zenpyre.chat_models.factory

Contain factories for chat models.

zenpyre.chat_models.factory.BaseChatModelFactory

Bases: ABC

Abstract base class for LangChain :class:~langchain_core.language_models.BaseChatModel factories.

Subclasses implement :meth:make_chat_model to instantiate and return a configured :class:~langchain_core.language_models.BaseChatModel object. This pattern decouples chat model creation from the rest of the codebase, making it easy to swap chat models (e.g. OpenAI, Anthropic, a fake model for testing) without changing call sites.

Example

>>> from langchain_core.language_models import BaseChatModel, FakeListChatModel
>>> from zenpyre.chat_models.factory import BaseChatModelFactory
>>> class MyChatModelFactory(BaseChatModelFactory):
...     def make_chat_model(self) -> BaseChatModel:
...         return FakeListChatModel(responses=["hello"])
...
>>> factory = MyChatModelFactory()
>>> chat_model = factory.make_chat_model()

zenpyre.chat_models.factory.BaseChatModelFactory.make_chat_model abstractmethod

make_chat_model() -> BaseChatModel

Create and return a configured BaseChatModel instance.

Returns:

Name	Type	Description
A	BaseChatModel	class:~langchain_core.language_models.BaseChatModel
	BaseChatModel	instance ready for use.

zenpyre.chat_models.factory.CachingChatModelFactory

Bases: BaseChatModelFactory, MultilineDisplayMixin

A concrete chat model factory that wraps another chat model factory and caches the resulting model's outputs via :class:~zenpyre.chat_models.CachingChatModel.

The wrapped chat model is built by delegating to the inner factory, then wrapped in a :class:~zenpyre.chat_models.CachingChatModel so every call to the model transparently reads from and writes to cache. Unlike :class:~zenpyre.runnables.CachingRunnable, the returned object is a genuine :class:~langchain_core.language_models.BaseChatModel instance, so bind_tools and other BaseChatModel-specific methods keep working after wrapping.

Parameters:

Name	Type	Description	Default
chat_model_factory	BaseChatModelFactory | dict[str, Any] | BaseConfig	The wrapped chat model factory, an objectory configuration, or a :class:~BaseConfig that resolves to a :class:~BaseChatModelFactory.	required
cache	Cache | None	The :class:~persista.cache.Cache instance used to store cached results. If None, caching is disabled.	None
key_fn	Callable[[Any], str] | None	A function that derives a cache key from an input. See :class:~zenpyre.chat_models.CachingChatModel for details.	None

Example

>>> from langchain_core.language_models import FakeListChatModel
>>> from persista.cache import Cache
>>> from zenpyre.chat_models.factory import CachingChatModelFactory, ChatModelFactory
>>> factory = CachingChatModelFactory(
...     chat_model_factory=ChatModelFactory(FakeListChatModel(responses=["hello"])),
...     cache=Cache(),
... )
>>> chat_model = factory.make_chat_model()

zenpyre.chat_models.factory.ChatModelFactory

Bases: BaseChatModelFactory, MultilineDisplayMixin

A concrete BaseChatModel factory that wraps a pre-built :class:~langchain_core.language_models.BaseChatModel instance.

Use this when the chat model is already instantiated and you simply want to wrap it in the :class:~BaseChatModelFactory interface — for example, when injecting a fixed chat model into a component that expects a factory.

Parameters:

Name	Type	Description	Default
chat_model	BaseChatModel	A fully configured :class:~langchain_core.language_models.BaseChatModel instance to return from :meth:make_chat_model.	required

Example

>>> from langchain_core.language_models import FakeListChatModel
>>> from zenpyre.chat_models.factory import ChatModelFactory
>>> factory = ChatModelFactory(FakeListChatModel(responses=["hello"]))
>>> chat_model = factory.make_chat_model()

zenpyre.chat_models.factory.ConfigurableChatModelFactory

Bases: BaseChatModelFactory, MultilineDisplayMixin

A concrete BaseChatModel factory that accepts either a pre-built :class:~langchain_core.language_models.BaseChatModel instance or a configuration dictionary.

When a dict is provided it is resolved at each :meth:make_chat_model call via :func:~zenpyre.chat_models.resolve.resolve_chat_model, which uses objectory to instantiate the configured class. When an instance is provided it is returned as-is.

Parameters:

Name	Type	Description	Default
chat_model	BaseChatModel | dict[str, Any]	A fully configured :class:~langchain_core.language_models.BaseChatModel instance, or a :class:dict containing an objectory factory specification (must include a "_target_" key pointing to the fully-qualified class name).	required

Example

>>> from langchain_core.language_models import FakeListChatModel
>>> from zenpyre.chat_models.factory import ConfigurableChatModelFactory
>>> factory = ConfigurableChatModelFactory(FakeListChatModel(responses=["hello"]))
>>> chat_model = factory.make_chat_model()

zenpyre.chat_models.factory.InitChatModelFactory

Bases: BaseChatModelFactory, MultilineDisplayMixin

A concrete BaseChatModel factory that wraps langchain.chat_models.init_chat_model, building a fresh chat model on each :meth:make_chat_model call.

Each call to :meth:make_chat_model forwards model and any additional keyword arguments to init_chat_model, which resolves the appropriate provider integration (e.g. from a "provider:model" string or model_provider) and instantiates it.

Parameters:

Name	Type	Description	Default
model	str | None	The model name, optionally prefixed with its provider (e.g. "openai:gpt-4o"). Forwarded as-is to init_chat_model.	None
**kwargs	Any	Additional keyword arguments forwarded as-is to init_chat_model (e.g. model_provider, configurable_fields, config_prefix, or any provider-specific parameter such as temperature). See init_chat_model's own documentation for the full list of accepted arguments.	{}

Example

>>> from zenpyre.chat_models.factory import InitChatModelFactory
>>> factory = InitChatModelFactory(model="openai:gpt-4o-mini", api_key="sk-...")
>>> chat_model = factory.make_chat_model()