Agents

zenpyre.agents

Contain agents.

zenpyre.agents.AgentChatModel

Bases: Runnable[LanguageModelInput | dict[str, Any], dict[str, Any]]

Wrap a BaseChatModel so it exposes the same input/output shape as an agent (e.g. AgentRunnable / create_agent), without any tool-calling loop.

This is useful for dropping a plain chat model into a workflow that's built to work with agents: callers can treat this object as if it were an agent (accepting the same flexible input shapes and returning the same {"messages": ...} dict shape, with an optional "structured_response" key), while under the hood it's just a single call to the wrapped chat model.

For invoke/ainvoke, the input is coerced to a list of BaseMessage objects (optionally prefixed with a system prompt), passed to the wrapped model, and the resulting AI message is appended to that list. If response_format is set, the model is queried with structured output enabled and a "structured_response" key is added to the result, mirroring create_agent's behavior.

Note

stream/astream do not follow the same {"messages": ...} shape as invoke/ainvoke. They instead yield the raw BaseMessage chunks produced by the wrapped model's own streaming interface, do not accumulate a running message history, and ignore response_format entirely (structured output is not supported in streaming mode). See the docstrings of those methods for details.

Accepted input shapes (for all four methods): - str: treated as a single human message. - list[BaseMessage | str]: used as-is, with any bare strings converted to human messages. - dict with a "messages" key: the value is treated the same way as the list case above (missing key defaults to an empty list).

If a system_prompt was provided at construction time and the resulting message list does not already contain a SystemMessage, a SystemMessage built from system_prompt is prepended.

Attributes:

Name	Type	Description
_model		The chat model to wrap.
_system_prompt		An optional system prompt to prepend to every call, unless the input already contains a system message.
_response_format		An optional schema (e.g. a Pydantic model, a TypedDict, or a JSON schema dict) passed to the wrapped model's with_structured_output. When set, invoke/ ainvoke additionally return a "structured_response" key in the result dict, parsed according to this schema.

Example

>>> from zenpyre.agents import AgentChatModel
>>> agent = AgentChatModel(model=my_chat_model)  # doctest: +SKIP
>>> result = agent.invoke("What is the capital of France?")  # doctest: +SKIP
>>> result["messages"][-1].content  # doctest: +SKIP
'The capital of France is Paris.'

zenpyre.agents.AgentChatModel.__init__

__init__(
    model: BaseChatModel,
    system_prompt: str | None = None,
    response_format: Any | None = None,
) -> None

Initialize the AgentChatModel.

Parameters:

Name	Type	Description	Default
model	BaseChatModel	The chat model to wrap. All calls are delegated to this model; AgentChatModel adds no tool-calling loop or other agent behavior of its own.	required
system_prompt	str | None	An optional system prompt. If set, it is prepended as a SystemMessage to the message list on every call, unless the input already contains a system message. Defaults to None (no system prompt is added).	None
response_format	Any | None	An optional schema describing the desired structured output (e.g. a Pydantic BaseModel subclass, a TypedDict, or a JSON schema dict), with the same semantics as BaseChatModel.with_structured_output. When set, the wrapped model is queried with structured output enabled and the parsed result is exposed under the "structured_response" key of the dict returned by invoke/ainvoke. Defaults to None (no structured output).	None

zenpyre.agents.AgentChatModel.abatch async

abatch(
    inputs: list[LanguageModelInput | dict[str, Any]],
    config: (
        RunnableConfig | list[RunnableConfig] | None
    ) = None,
    **kwargs: Any
) -> list[dict[str, Any]]

Asynchronously invoke the wrapped chat model on a batch of inputs.

This is the async counterpart of :meth:batch; see its docstring for details.

Parameters:

Name	Type	Description	Default
inputs	list[LanguageModelInput | dict[str, Any]]	A list of inputs, each following the shapes described in the class docstring.	required
config	RunnableConfig | list[RunnableConfig] | None	Optional RunnableConfig (or list of one per input) forwarded to the wrapped model's abatch call.	None
**kwargs	Any	Additional keyword arguments forwarded to the wrapped model's abatch call.	{}

Returns:

Name	Type	Description
	list[dict[str, Any]]	A list of result dicts, one per input, in the same order as
	list[dict[str, Any]]	inputs. Each dict has the same shape as the one returned
by	list[dict[str, Any]]	meth:ainvoke.

zenpyre.agents.AgentChatModel.ainvoke async

ainvoke(
    input: LanguageModelInput | dict[str, Any],
    config: RunnableConfig | None = None,
    **kwargs: Any
) -> dict[str, Any]

Asynchronously invoke the wrapped chat model once and return an agent-shaped result.

This is the async counterpart of :meth:invoke; see its docstring for details on accepted input shapes and the returned dict, including the "structured_response" key when response_format is set.

Parameters:

Name	Type	Description	Default
input	LanguageModelInput | dict[str, Any]	The input to the model. See the class docstring for the accepted shapes (str, list, or dict with a "messages" key).	required
config	RunnableConfig | None	Optional RunnableConfig forwarded to the wrapped model's ainvoke call.	None
**kwargs	Any	Additional keyword arguments forwarded to the wrapped model's ainvoke call.	{}

Returns:

Type	Description
dict[str, Any]	A dict with "messages", and (if
dict[str, Any]	response_format is set) "structured_response".

zenpyre.agents.AgentChatModel.astream async

astream(
    input: LanguageModelInput | dict[str, Any],
    config: RunnableConfig | None = None,
    **kwargs: Any
) -> AsyncIterator[BaseMessage]

Asynchronously stream the wrapped chat model's response chunk by chunk.

As with :meth:stream, this does not return the {"messages": ...} agent shape, does not accumulate message history, and ignores response_format (structured output is not supported in streaming mode).

Parameters:

Name	Type	Description	Default
input	LanguageModelInput | dict[str, Any]	The input to the model. See the class docstring for the accepted shapes (str, list, or dict with a "messages" key).	required
config	RunnableConfig | None	Optional RunnableConfig forwarded to the wrapped model's astream call.	None
**kwargs	Any	Additional keyword arguments forwarded to the wrapped model's astream call.	{}

Yields:

Type	Description
AsyncIterator[BaseMessage]	Successive BaseMessage chunks from the wrapped model.

zenpyre.agents.AgentChatModel.batch

batch(
    inputs: list[LanguageModelInput | dict[str, Any]],
    config: (
        RunnableConfig | list[RunnableConfig] | None
    ) = None,
    **kwargs: Any
) -> list[dict[str, Any]]

Invoke the wrapped chat model on a batch of inputs.

Unlike the default Runnable.batch (which fans out N separate invoke calls across a thread pool), this pushes the batching down to the wrapped model's own batch/ with_structured_output(...).batch method, letting providers that support genuine server-side batching (or internal max_concurrency tuning) take advantage of it directly.

Parameters:

Name	Type	Description	Default
inputs	list[LanguageModelInput | dict[str, Any]]	A list of inputs, each following the shapes described in the class docstring.	required
config	RunnableConfig | list[RunnableConfig] | None	Optional RunnableConfig (or list of one per input) forwarded to the wrapped model's batch call.	None
**kwargs	Any	Additional keyword arguments forwarded to the wrapped model's batch call.	{}

Returns:

Name	Type	Description
	list[dict[str, Any]]	A list of result dicts, one per input, in the same order as
	list[dict[str, Any]]	inputs. Each dict has the same shape as the one returned
by	list[dict[str, Any]]	meth:invoke.

zenpyre.agents.AgentChatModel.invoke

invoke(
    input: LanguageModelInput | dict[str, Any],
    config: RunnableConfig | None = None,
    **kwargs: Any
) -> dict[str, Any]

Invoke the wrapped chat model once and return an agent-shaped result.

Parameters:

Name	Type	Description	Default
input	LanguageModelInput | dict[str, Any]	The input to the model. See the class docstring for the accepted shapes (str, list, or dict with a "messages" key).	required
config	RunnableConfig | None	Optional RunnableConfig forwarded to the wrapped model's invoke call.	None
**kwargs	Any	Additional keyword arguments forwarded to the wrapped model's invoke call.	{}

Returns:

Type	Description
dict[str, Any]	A dict with:
dict[str, Any]	"messages": the full list of BaseMessage objects used for the call (including any prepended system prompt), with the model's response message appended at the end.
dict[str, Any]	"structured_response": present only if response_format was set at construction time. The parsed structured output, as an instance of response_format.

zenpyre.agents.AgentChatModel.stream

stream(
    input: LanguageModelInput | dict[str, Any],
    config: RunnableConfig | None = None,
    **kwargs: Any
) -> Iterator[BaseMessage]

Stream the wrapped chat model's response chunk by chunk.

Unlike :meth:invoke, this does not return the {"messages": ...} agent shape. It is a thin pass-through to the wrapped model's own stream method: the input is coerced to a message list (with the system prompt prepended if applicable) and each BaseMessage chunk produced by the model is yielded as-is. The running message history is not accumulated or returned.

Note

Structured output (response_format) is not supported in streaming mode: most providers/parsers need the complete response before they can validate/parse it against the schema, so this always streams the plain, unstructured model output regardless of whether response_format was set at construction time.

Parameters:

Name	Type	Description	Default
input	LanguageModelInput | dict[str, Any]	The input to the model. See the class docstring for the accepted shapes (str, list, or dict with a "messages" key).	required
config	RunnableConfig | None	Optional RunnableConfig forwarded to the wrapped model's stream call.	None
**kwargs	Any	Additional keyword arguments forwarded to the wrapped model's stream call.	{}

Yields:

Type	Description
BaseMessage	Successive BaseMessage chunks from the wrapped model.

zenpyre.agents.AgentConfig dataclass

Bases: ExtraFieldsConfig, MultilineDisplayMixin

A generic LLM agent configuration.

Subclass this to add provider- or agent-specific parameters as typed fields (e.g. max_tokens for OpenAI, top_k for Anthropic). Since this class is frozen, subclasses must also be frozen dataclasses. Additional fields do not need a default value: extra is declared keyword-only on :class:~zenpyre.utils.config.ExtraFieldsConfig, so it doesn't force subclass fields into a particular ordering.

Fields added by subclasses are picked up automatically by :meth:to_kwargs (and therefore :meth:cache_key) via introspection; you don't need to override either method just to add a field.

One thing subclasses do need to restate: @dataclass(frozen=True) auto-generates a fresh __hash__ for every dataclass-decorated class unless that class's own body defines __hash__ — merely inheriting one does not suppress the override, and the auto-generated version would try to hash the unhashable extra field. Any further subclass should include a delegating method of its own::

def __hash__(self) -> int:
    return AgentConfig.__hash__(self)

(A plain assignment like __hash__ = AgentConfig.__hash__ also suppresses the auto-generation, but static type checkers such as pyright flag it as an "ambiguous base class override" because the inferred self type comes from the parent method rather than the subclass; the delegating-method form above avoids that.)

Attributes:

Name	Type	Description
chat_model	BaseChatModelConfig	The chat model configuration (see :class:~zenpyre.chat_models.BaseChatModelConfig) used by this agent.
system_prompt	str	The system prompt that instructs the LLM on its role and task.
system_prompt_id	str	An identifier for system_prompt. Defaults to hash_string(system_prompt) when constructed via :meth:from_kwargs and left unset, so configs built from the same prompt text get the same id without callers needing to compute it themselves.
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).

Example

>>> from dataclasses import dataclass
>>> from zenpyre.agents import AgentConfig
>>> from zenpyre.chat_models import ChatModelConfig
>>> config = AgentConfig.from_kwargs(
...     chat_model=ChatModelConfig(model="openai:gpt-4o"),
...     system_prompt="You are helpful.",
...     max_tokens=1024,
... )

zenpyre.agents.AgentConfig.from_kwargs classmethod

from_kwargs(
    chat_model: BaseChatModelConfig,
    system_prompt: str,
    system_prompt_id: str | None = None,
    **kwargs: Any
) -> Self

Construct an :class:AgentConfig from a chat model configuration, a system prompt, and arbitrary keyword arguments.

A convenience alternative to the regular constructor's extra={...} dict, letting callers pass extra fields directly as keyword arguments instead: AgentConfig.from_kwargs(chat_model, "You are helpful.", max_retries=3) is equivalent to AgentConfig(chat_model=chat_model, system_prompt="You are helpful.", extra={"max_retries": 3}).

Parameters:

Name	Type	Description	Default
chat_model	BaseChatModelConfig	The chat model configuration used by this agent.	required
system_prompt	str	The system prompt that instructs the LLM on its role and task.	required
system_prompt_id	str | None	An identifier for system_prompt. If None (the default), it is derived automatically via hash_string(system_prompt).	None
**kwargs	Any	Additional keyword arguments, stored as extra.	{}

Returns:

Type	Description
Self	A new :class:AgentConfig.

Raises:

Type

Description

TypeError

If kwargs contains a "chat_model", "system_prompt", or "system_prompt_id" key, since Python's own argument binding intercepts it as a duplicate value for the corresponding explicit parameter before this method's body ever runs. For example, from_kwargs(cm, "prompt", **{"chat_model": other}) raises TypeError: got multiple values for argument 'chat_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.agents import AgentConfig
>>> from zenpyre.chat_models import ChatModelConfig
>>> cfg = AgentConfig.from_kwargs(
...     ChatModelConfig(model="gpt-4"), "You are helpful.", max_retries=3
... )
>>> cfg.to_kwargs()["max_retries"]
3

zenpyre.agents.factory

Contain factories for agents.

zenpyre.agents.factory.AgentChatModelFactory

Bases: BaseAgentFactory, MultilineDisplayMixin

A concrete agent factory that builds a fresh :class:~zenpyre.agents.AgentChatModel on each :meth:make_agent call, wrapping the chat model produced by a :class:~zenpyre.chat_models.factory.base.BaseChatModelFactory.

Each call to :meth:make_agent calls chat_model_factory.make_chat_model() and wraps the resulting chat model in a new :class:~zenpyre.agents.AgentChatModel, together with system_prompt and response_format. This composes with any :class:~zenpyre.chat_models.factory.base.BaseChatModelFactory implementation (e.g. :class:~zenpyre.chat_models.factory.ChatModelFactory, :class:~zenpyre.chat_models.factory.ConfigurableChatModelFactory), keeping chat model creation and agent creation decoupled.

Parameters:

Name	Type	Description	Default
chat_model_factory	BaseChatModelFactory | dict[str, Any] | BaseConfig	The factory used to build the chat model wrapped by the created agent.	required
system_prompt	str | None	An optional system prompt forwarded to :class:~zenpyre.agents.AgentChatModel. See its docstring for details.	None
response_format	Any | None	An optional schema forwarded to :class:~zenpyre.agents.AgentChatModel. See its docstring for details.	None

Example

>>> from langchain_core.language_models import FakeListChatModel
>>> from zenpyre.agents.factory import AgentChatModelFactory
>>> from zenpyre.chat_models.factory import ChatModelFactory
>>> factory = AgentChatModelFactory(
...     chat_model_factory=ChatModelFactory(FakeListChatModel(responses=["hello"])),
...     system_prompt="You are helpful.",
... )
>>> agent = factory.make_agent()

zenpyre.agents.factory.AgentFactory

Bases: BaseAgentFactory, MultilineDisplayMixin

A concrete agent factory that wraps a pre-built agent (a :class:~langchain_core.runnables.Runnable, e.g. :class:~zenpyre.agents.AgentChatModel).

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

Parameters:

Name	Type	Description	Default
agent	Runnable[dict[str, Any], dict[str, Any]]	A fully configured agent (:class:~langchain_core.runnables.Runnable) instance to return from :meth:make_agent.	required

Example

>>> from langchain_core.language_models import FakeListChatModel
>>> from zenpyre.agents import AgentChatModel
>>> from zenpyre.agents.factory import AgentFactory
>>> agent = AgentChatModel(model=FakeListChatModel(responses=["hello"]))
>>> factory = AgentFactory(agent)
>>> agent = factory.make_agent()

zenpyre.agents.factory.BaseAgentFactory

Bases: ABC

Abstract base class for agent factories.

Subclasses implement :meth:make_agent to instantiate and return a configured agent — a :class:~langchain_core.runnables.Runnable that accepts the same flexible input shapes as :class:~zenpyre.agents.AgentChatModel (a str, a list of messages, or a dict with a "messages" key) and returns a {"messages": ...} dict shape. This pattern decouples agent creation from the rest of the codebase, making it easy to swap agents (e.g. a plain chat model wrapped in :class:~zenpyre.agents.AgentChatModel, or a tool-calling agent built with create_agent) without changing call sites.

Example

>>> from typing import Any
>>> from langchain_core.language_models import FakeListChatModel
>>> from langchain_core.runnables import Runnable
>>> from zenpyre.agents import AgentChatModel
>>> from zenpyre.agents.factory import BaseAgentFactory
>>> class MyAgentFactory(BaseAgentFactory):
...     def make_agent(self) -> Runnable[dict[str, Any], dict[str, Any]]:
...         return AgentChatModel(model=FakeListChatModel(responses=["hello"]))
...
>>> factory = MyAgentFactory()
>>> agent = factory.make_agent()

zenpyre.agents.factory.BaseAgentFactory.make_agent abstractmethod

make_agent() -> Runnable[dict[str, Any], dict[str, Any]]

Create and return a configured agent instance.

Returns:

Name	Type	Description
A	Runnable[dict[str, Any], dict[str, Any]]	class:~langchain_core.runnables.Runnable instance
	Runnable[dict[str, Any], dict[str, Any]]	ready for use as an agent.

zenpyre.agents.factory.CachingAgentFactory

Bases: BaseAgentFactory, MultilineDisplayMixin

A concrete agent factory that wraps another agent factory and caches the resulting agent's outputs via :class:~zenpyre.runnables.CachingRunnable.

Each call to :meth:make_agent builds a fresh agent from agent_factory and wraps it in a :class:~zenpyre.runnables.CachingRunnable, so repeated calls to the wrapped agent with the same (or equivalent, per key_fn) input are served from cache instead of re-invoking the underlying agent.

Parameters:

Name	Type	Description	Default
agent_factory	BaseAgentFactory | dict[str, Any] | BaseConfig	The factory used to build the underlying agent to cache.	required
cache	Cache | None	The :class:~persista.cache.Cache instance used to store cached results. If None, caching is disabled.	None
key_fn	Callable[[dict[str, Any]], str] | None	An optional function used to compute a cache key from the agent's input. If None, :class:~zenpyre.runnables.CachingRunnable's default key-computation strategy is used.	None

Example

>>> from langchain_core.language_models import FakeListChatModel
>>> from persista.cache import Cache
>>> from zenpyre.agents import AgentChatModel
>>> from zenpyre.agents.factory import AgentFactory, CachingAgentFactory
>>> inner_agent = AgentChatModel(model=FakeListChatModel(responses=["hello"]))
>>> factory = CachingAgentFactory(
...     agent_factory=AgentFactory(inner_agent),
...     cache=Cache(),
... )
>>> agent = factory.make_agent()  # doctest: +SKIP

zenpyre.agents.factory.ConfigurableAgentFactory

Bases: BaseAgentFactory, MultilineDisplayMixin

A concrete agent factory that accepts either a pre-built agent (a :class:~langchain_core.runnables.Runnable, e.g. :class:~zenpyre.agents.AgentChatModel) or a configuration dictionary.

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

Parameters:

Name	Type	Description	Default
agent	Runnable[dict[str, Any], dict[str, Any]] | dict[str, Any]	A fully configured agent (:class:~langchain_core.runnables.Runnable) 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.agents import AgentChatModel
>>> from zenpyre.agents.factory import ConfigurableAgentFactory
>>> agent = AgentChatModel(model=FakeListChatModel(responses=["hello"]))
>>> factory = ConfigurableAgentFactory(agent)
>>> agent = factory.make_agent()

zenpyre.agents.factory.CreateAgentFactory

Bases: BaseAgentFactory, MultilineDisplayMixin

A concrete agent factory that wraps langchain.agents.create_agent, building a fresh tool-calling agent graph on each :meth:make_agent call.

Each call to :meth:make_agent calls chat_model_factory.make_chat_model() and forwards the resulting chat model, together with any additional keyword arguments (e.g. tools, system_prompt, response_format, middleware), to create_agent. This composes with any :class:~zenpyre.chat_models.factory.base.BaseChatModelFactory implementation (e.g. :class:~zenpyre.chat_models.factory.ChatModelFactory, :class:~zenpyre.chat_models.factory.ConfigurableChatModelFactory), keeping chat model creation and agent creation decoupled.

Unlike :class:~zenpyre.agents.factory.AgentChatModelFactory (which builds a tool-free :class:~zenpyre.agents.AgentChatModel), the agent produced here can call tools in a loop before returning a final answer, as implemented by create_agent.

Parameters:

Name	Type	Description	Default
chat_model_factory	BaseChatModelFactory | dict[str, Any] | BaseConfig	The factory used to build the chat model wrapped by the created agent.	required
**kwargs	Any	Additional keyword arguments forwarded as-is to create_agent (e.g. tools, system_prompt, response_format, middleware, checkpointer, store, name). See create_agent's own documentation for the full list of accepted arguments.	{}

Example

>>> from langchain_core.language_models import FakeListChatModel
>>> from zenpyre.agents.factory import CreateAgentFactory
>>> from zenpyre.chat_models.factory import ChatModelFactory
>>> factory = CreateAgentFactory(
...     chat_model_factory=ChatModelFactory(FakeListChatModel(responses=["hello"])),
...     system_prompt="You are helpful.",
... )
>>> agent = factory.make_agent()

zenpyre.agents.factory.RecordingAgentFactory

Bases: BaseAgentFactory, MultilineDisplayMixin

A concrete agent factory that wraps another agent factory and records the resulting agent's calls via :class:~zenpyre.runnables.RecordingRunnable.

Each call to :meth:make_agent builds a fresh agent from agent_factory, calls record_store_factory.make_record_store(), and wraps the agent in a :class:~zenpyre.runnables.RecordingRunnable, so every call to the wrapped agent writes a record of its input and output to the resulting record store. This composes with any :class:~zenpyre.record_stores.factory.base.BaseRecordStoreFactory implementation (e.g. :class:~zenpyre.record_stores.factory.RecordStoreFactory, :class:~zenpyre.record_stores.factory.ConfigurableRecordStoreFactory), keeping record store creation and agent creation decoupled.

Parameters:

Name	Type	Description	Default
agent_factory	BaseAgentFactory | dict[str, Any] | BaseConfig	The factory used to build the underlying agent to record.	required
record_store_factory	BaseRecordStoreFactory | dict[str, Any] | BaseConfig	The factory used to build the store input/output records are written to.	required
extra	dict[str, Any] | None	Additional metadata merged into every record written by this wrapper, fixed for its lifetime (e.g. an experiment ID). See :class:~zenpyre.runnables.RecordingRunnable for the set of reserved keys this must not contain.	None
serializer	Callable[[dict[str, Any]], dict[str, Any]] | None	A function applied to the whole assembled metadata dict before it's stored. If None, :class:~zenpyre.runnables.RecordingRunnable's default (:func:~zenpyre.runnables.recording.default_serialize) is used.	None

Example

>>> from langchain_core.language_models import FakeListChatModel
>>> from zenpyre.agents import AgentChatModel
>>> from zenpyre.agents.factory import AgentFactory, RecordingAgentFactory
>>> from zenpyre.record_stores import InMemoryRecordStore
>>> from zenpyre.record_stores.factory import RecordStoreFactory
>>> inner_agent = AgentChatModel(model=FakeListChatModel(responses=["hello"]))
>>> factory = RecordingAgentFactory(
...     agent_factory=AgentFactory(inner_agent),
...     record_store_factory=RecordStoreFactory(InMemoryRecordStore()),
... )
>>> agent = factory.make_agent()