Runnables

zenpyre.runnables

Contain runnables.

zenpyre.runnables.CachingRunnable

Bases: Runnable[Input, Output], MultilineDisplayMixin

Wrap a Runnable to cache its output, keyed by a hash of the input.

On each call, key_fn(input) is used to look up a previously cached result in cache. On a cache hit, the cached result is returned without calling the wrapped runnable. On a cache miss, the wrapped runnable is invoked and its result is stored in cache before being returned. If cache is None, caching is disabled entirely and every call goes straight to the wrapped runnable.

batch/abatch look up each input's cache entry individually, then call the wrapped runnable's own batch/abatch for only the inputs that missed -- so a partially-cached batch still benefits from the wrapped runnable's batching, rather than falling back to one call per miss.

Unlike subclassing a caching base class, this wrapper works with any Runnable — including third-party ones you don't control — since caching is composed around the runnable rather than baked into its class hierarchy.

Parameters:

Name	Type	Description	Default
runnable	Runnable[Input, Output]	The runnable whose output should be cached.	required
cache	Cache | None	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; CachingRunnable has no caching policy of its own beyond what cache provides. 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 -- CachingRunnable does not manage its lifecycle.	None
key_fn	Callable[[Input], str] | None	A function that derives a cache key from an input. The returned string is used directly as the cache key. Defaults to hash_object, which dispatches through coola's hasher registry (e.g. using DocumentHasher for Document inputs). Override this if you need input-specific hashing behavior that differs from the registered default.	None

Example

>>> from langchain_core.runnables import RunnableLambda
>>> from persista.cache import Cache
>>> from zenpyre.runnables import CachingRunnable
>>> runnable = RunnableLambda(lambda x: x.upper())
>>> with Cache() as cache:
...     cached = CachingRunnable(runnable=runnable, cache=cache)
...     cached.invoke("hello")
...
'HELLO'

zenpyre.runnables.InputOutputPair dataclass

Bases: Generic[Input, Output]

A frozen pair holding a runnable's input alongside the output it produced.

Parameters:

Name	Type	Description	Default
input	Input	The input value passed to the wrapped runnable.	required
output	Output	The output value produced by the wrapped runnable for input.	required

zenpyre.runnables.InputOutputRunnable

Bases: Runnable[Input, InputOutputPair[Input, Output]], Generic[Input, Output]

Wrap a runnable so invoking it returns an :class:InputOutputPair of its input and output, instead of just the output.

This is useful whenever downstream code needs to know which input produced a given output -- e.g. logging, evaluation harnesses, or building a dataset of (input, output) examples -- without having to thread the input through the wrapped runnable itself or zip inputs and outputs back together by hand afterwards.

.batch()/.abatch() delegate to the wrapped runnable's own batch/abatch implementation (rather than falling back to the default per-item invoke loop that :class:Runnable provides), so any batching optimizations the inner runnable implements (e.g. batched LLM calls) are preserved. When return_exceptions=True and a given input's call fails, the corresponding entry in the returned list is the raw exception, not an :class:InputOutputPair -- exactly as :meth:Runnable.batch behaves for the wrapped runnable itself.

Parameters:

Name	Type	Description	Default
runnable	Runnable[Input, Output]	The runnable to wrap.	required

Example

>>> from langchain_core.runnables import RunnableLambda
>>> from zenpyre.runnables import InputOutputRunnable
>>> inner = RunnableLambda(lambda x: x.upper())
>>> wrapped = InputOutputRunnable(inner)
>>> result = wrapped.invoke("hello")
>>> result.input
'hello'
>>> result.output
'HELLO'
>>> [pair.output for pair in wrapped.batch(["a", "b"])]
['A', 'B']

zenpyre.runnables.RecordingRunnable

Bases: Runnable[Input, Output], MultilineDisplayMixin, Generic[Input, Output]

Wrap a Runnable to record the input and output of each invocation to a record store.

This is a transparent passthrough wrapper: calling invoke, ainvoke, batch, or abatch behaves exactly like calling the wrapped runnable directly (same return value, same exceptions), with the side effect of writing one :class:~zenpyre.records.Record per invocation to record_store via :meth:~zenpyre.record_stores.base.BaseRecordStore.add_records. Each record gets a fresh, randomly generated ID (not derived from its content), so that two calls with identical input/output/extra are still both recorded rather than one silently overwriting the other via the store's upsert semantics.

Note

If the wrapped runnable raises during invoke/ ainvoke, the exception propagates immediately and no record is written for that call -- unlike batch/abatch with return_exceptions=True, which does record failed items (see "error" below). If you want failed single calls recorded too for a fully consistent audit trail, wrap the self._runnable.invoke(...)/ainvoke(...) calls in invoke/ainvoke in a try/except that builds an error record before re-raising, mirroring :meth:_record_batch's handling.

Each record's metadata is assembled as a plain dict with the following keys, then passed through serializer (see below) as a whole before being stored:

• "input" / "output": the invocation's raw input and output.
• "timestamp": an ISO 8601 UTC timestamp of when the call completed.
• "run_id": the run_id from the call's RunnableConfig, if the caller supplied one explicitly; otherwise None. This is not LangChain's internally auto-generated run ID (that isn't accessible from a plain wrapper like this one), only one explicitly passed in by the caller.
• "error": None on success. On a batch item that failed with return_exceptions=True, this holds str(exception) and "output" is None.
• Any additional keys from extra (fixed for this wrapper's lifetime, e.g. an experiment ID) and/or from the call's config["metadata"] (varies per invocation, e.g. a session or user ID). If the same key appears in both, the per-call config["metadata"] value wins. Neither may use one of the reserved keys above; doing so raises :exc:ValueError.

stream/astream are supported on a best-effort basis: each chunk is yielded to the caller immediately (true streaming isn't delayed), while chunks are accumulated internally (via +, as LangChain message chunks support) to reconstruct a final output for recording once the stream is exhausted. If chunks don't support +, or the stream raises before completing, no record is written for that call (a warning is logged) rather than raising an error into the caller's stream.

Parameters:

Name	Type	Description	Default
runnable	Runnable[Input, Output]	The inner Runnable to wrap.	required
record_store	BaseRecordStore	The store to write input/output records 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). For metadata that varies per call, pass it via config={"metadata": {...}} on the individual invoke/batch/... call instead; it takes precedence over extra on key collision. Must not use a reserved key (see above).	None
serializer	Callable[[dict[str, Any]], dict[str, Any]] | None	A function applied to the whole assembled metadata dict before it's stored, to make it JSON-friendly (or to apply any other custom transform). Defaults to :func:default_serialize, applied to the whole dict (so it naturally recurses into "input", "output", and any extra/config["metadata"] values in one pass).	None

Example

>>> from zenpyre.record_stores import DuckDBRecordStore
>>> from zenpyre.runnables import RecordingRunnable
>>> store = DuckDBRecordStore(":memory:")
>>> recorded = RecordingRunnable(
...     chat_model, store, extra={"experiment_id": "exp-42"}
... )  # doctest: +SKIP
>>> recorded.invoke("Hello!", config={"metadata": {"session_id": "s-1"}})  # doctest: +SKIP
AIMessage(content='Hi there!')
>>> store.all()[0].metadata["experiment_id"], store.all()[0].metadata[
...     "session_id"
... ]  # doctest: +SKIP
('exp-42', 's-1')

zenpyre.runnables.RecordingRunnable.reserved_metadata_keys property

reserved_metadata_keys: frozenset[str]

The metadata keys reserved for this class's own use.

Neither extra nor a call's config["metadata"] may use one of these; doing so raises :exc:ValueError. A subclass may override this property to change the reserved set.

zenpyre.runnables.RecordingRunnable.__init__

__init__(
    runnable: Runnable[Input, Output],
    record_store: BaseRecordStore,
    *,
    extra: dict[str, Any] | None = None,
    serializer: (
        Callable[[dict[str, Any]], dict[str, Any]] | None
    ) = None
) -> None

Initialize the wrapper.

Parameters:

Name	Type	Description	Default
runnable	Runnable[Input, Output]	The inner Runnable to wrap and record calls of.	required
record_store	BaseRecordStore	The store to write input/output records 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). Must not contain a key from :attr:reserved_metadata_keys.	None
serializer	Callable[[dict[str, Any]], dict[str, Any]] | None	A function applied to the whole assembled metadata dict before it's stored. Defaults to :func:default_serialize when None.	None

Raises:

Type	Description
ValueError	If extra contains a key from :attr:reserved_metadata_keys.

zenpyre.runnables.RecordingRunnable.abatch async

abatch(
    inputs: list[Input],
    config: (
        RunnableConfig | list[RunnableConfig] | None
    ) = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any
) -> list[Output]

Asynchronously invoke the wrapped runnable on a list of inputs and record each call.

The async counterpart of :meth:batch: calls self.runnable.abatch(inputs, config, return_exceptions, **kwargs), then writes one :class:~zenpyre.records.Record per (input, result) pair to the record store in a single :meth:~zenpyre.record_stores.base.BaseRecordStore.add_records call, before returning the results unchanged. Unlike :meth:ainvoke, a failed item (when return_exceptions=True) is still recorded, with "error" set to str(exception) and "output" set to None.

Parameters:

Name	Type	Description	Default
inputs	list[Input]	The list of inputs to pass to the wrapped runnable.	required
config	RunnableConfig | list[RunnableConfig] | None	Optional run configuration, either a single config shared by every item or a list with one config per item (matching inputs in length and order). If a config has a "metadata" key, those entries are merged into that item's stored record.	None
return_exceptions	bool	If True, exceptions raised by individual items are returned in results instead of propagating, and are still recorded (see above). If False, an exception from any item propagates immediately from self.runnable.abatch itself, and no records are written for this call at all.	False
**kwargs	Any	Additional keyword arguments forwarded to the wrapped runnable's abatch.	{}

Returns:

Type	Description
list[Output]	One result per input, in the same order as inputs, unchanged from what the wrapped runnable returned (which may include exception instances when return_exceptions=True).

zenpyre.runnables.RecordingRunnable.ainvoke async

ainvoke(
    input: Input,
    config: RunnableConfig | None = None,
    **kwargs: Any
) -> Output

Asynchronously invoke the wrapped runnable and record the call.

The async counterpart of :meth:invoke: calls self.runnable.ainvoke(input, config, **kwargs), writes one :class:~zenpyre.records.Record to the record store capturing input, the returned output, a timestamp, and any run_id/extra metadata, then returns that output unchanged.

Parameters:

Name	Type	Description	Default
input	Input	The input to pass to the wrapped runnable.	required
config	RunnableConfig | None	Optional run configuration. If it has a "metadata" key, those entries are merged into the stored record (see the class docstring).	None
**kwargs	Any	Additional keyword arguments forwarded to the wrapped runnable's ainvoke.	{}

Returns:

Type	Description
Output	The wrapped runnable's output, unchanged.

Raises:

Type	Description
Exception	Whatever the wrapped runnable itself raises. In that case, this method does not catch it, so no record is written for the failed call (see the class docstring's Note).

zenpyre.runnables.RecordingRunnable.astream async

astream(
    input: Input,
    config: RunnableConfig | None = None,
    **kwargs: Any
) -> AsyncIterator[Output]

Asynchronously stream the wrapped runnable's output chunks, recording an accumulated final result once the stream completes.

The async counterpart of :meth:stream: each chunk from self.runnable.astream(input, config, **kwargs) is yielded to the caller immediately, with no added latency. In parallel, chunks are accumulated via + (see :func:_try_add) to reconstruct a final output. Once the stream is exhausted (in a finally block, so this also runs if the caller stops iterating early or the stream raises), one :class:~zenpyre.records.Record is written for the accumulated result, unless no chunk could be accumulated at all (in which case a warning is logged and nothing is recorded for this call).

Parameters:

Name	Type	Description	Default
input	Input	The input to pass to the wrapped runnable.	required
config	RunnableConfig | None	Optional run configuration. If it has a "metadata" key, those entries are merged into the stored record (see the class docstring).	None
**kwargs	Any	Additional keyword arguments forwarded to the wrapped runnable's astream.	{}

Yields:

Type	Description
AsyncIterator[Output]	Each output chunk from the wrapped runnable, unchanged and in the same order, as soon as it's produced.

zenpyre.runnables.RecordingRunnable.batch

batch(
    inputs: list[Input],
    config: (
        RunnableConfig | list[RunnableConfig] | None
    ) = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any
) -> list[Output]

Invoke the wrapped runnable on a list of inputs and record each call.

Calls self.runnable.batch(inputs, config, return_exceptions, **kwargs), then writes one :class:~zenpyre.records.Record per (input, result) pair to the record store in a single :meth:~zenpyre.record_stores.base.BaseRecordStore.add_records call, before returning the results unchanged. Unlike :meth:invoke, a failed item (when return_exceptions=True) is still recorded, with "error" set to str(exception) and "output" set to None.

Parameters:

Name	Type	Description	Default
inputs	list[Input]	The list of inputs to pass to the wrapped runnable.	required
config	RunnableConfig | list[RunnableConfig] | None	Optional run configuration, either a single config shared by every item or a list with one config per item (matching inputs in length and order). If a config has a "metadata" key, those entries are merged into that item's stored record.	None
return_exceptions	bool	If True, exceptions raised by individual items are returned in results instead of propagating, and are still recorded (see above). If False, an exception from any item propagates immediately from self.runnable.batch itself, and no records are written for this call at all.	False
**kwargs	Any	Additional keyword arguments forwarded to the wrapped runnable's batch.	{}

Returns:

Type	Description
list[Output]	One result per input, in the same order as inputs, unchanged from what the wrapped runnable returned (which may include exception instances when return_exceptions=True).

zenpyre.runnables.RecordingRunnable.invoke

invoke(
    input: Input,
    config: RunnableConfig | None = None,
    **kwargs: Any
) -> Output

Invoke the wrapped runnable and record the call.

Calls self.runnable.invoke(input, config, **kwargs), writes one :class:~zenpyre.records.Record to the record store capturing input, the returned output, a timestamp, and any run_id/extra metadata, then returns that output unchanged.

Parameters:

Name	Type	Description	Default
input	Input	The input to pass to the wrapped runnable.	required
config	RunnableConfig | None	Optional run configuration. If it has a "metadata" key, those entries are merged into the stored record (see the class docstring).	None
**kwargs	Any	Additional keyword arguments forwarded to the wrapped runnable's invoke.	{}

Returns:

Type	Description
Output	The wrapped runnable's output, unchanged.

Raises:

Type	Description
Exception	Whatever the wrapped runnable itself raises. In that case, this method does not catch it, so no record is written for the failed call (see the class docstring's Note).

zenpyre.runnables.RecordingRunnable.stream

stream(
    input: Input,
    config: RunnableConfig | None = None,
    **kwargs: Any
) -> Iterator[Output]

Stream the wrapped runnable's output chunks, recording an accumulated final result once the stream completes.

Each chunk from self.runnable.stream(input, config, **kwargs) is yielded to the caller immediately, with no added latency. In parallel, chunks are accumulated via + (see :func:_try_add) to reconstruct a final output. Once the stream is exhausted (in a finally block, so this also runs if the caller stops iterating early or the stream raises), one :class:~zenpyre.records.Record is written for the accumulated result, unless no chunk could be accumulated at all (in which case a warning is logged and nothing is recorded for this call).

Parameters:

Name	Type	Description	Default
input	Input	The input to pass to the wrapped runnable.	required
config	RunnableConfig | None	Optional run configuration. If it has a "metadata" key, those entries are merged into the stored record (see the class docstring).	None
**kwargs	Any	Additional keyword arguments forwarded to the wrapped runnable's stream.	{}

Yields:

Type	Description
Output	Each output chunk from the wrapped runnable, unchanged and in the same order, as soon as it's produced.

zenpyre.runnables.resolve_runnable

resolve_runnable(
    runnable: (
        Runnable[Input, Output]
        | dict[str, Any]
        | BaseConfig
    ),
) -> Runnable[Input, Output]

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

If runnable is already a :class:~langchain_core.runnables.Runnable 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
runnable	Runnable[Input, Output] | dict[str, Any] | BaseConfig	Either a fully configured :class:~langchain_core.runnables.Runnable 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
Runnable[Input, Output]	A configured :class:~langchain_core.runnables.Runnable
Runnable[Input, Output]	instance.

Raises:

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

Example

>>> from langchain_core.runnables import Runnable
>>> from zenpyre.runnables import resolve_runnable
>>> class MyRunnable(Runnable):
...     def invoke(self, input: Any, config: Any = None, **kwargs: Any) -> Any:
...         return input
...
>>> # From an existing instance:
>>> runnable = resolve_runnable(MyRunnable())
>>> # From a configuration dictionary:
>>> runnable = resolve_runnable(
...     {"_target_": "langchain_core.runnables.RunnablePassthrough"}
... )

zenpyre.runnables.structured_output_runnable

structured_output_runnable(
    chat_model: BaseChatModel,
    output_type: type[T],
    *,
    include_raw: Literal[False] = False,
    **kwargs: Any
) -> Runnable[LanguageModelInput, T]

structured_output_runnable(
    chat_model: BaseChatModel,
    output_type: type[T],
    *,
    include_raw: Literal[True],
    **kwargs: Any
) -> Runnable[LanguageModelInput, dict[str, Any]]

structured_output_runnable(
    chat_model: BaseChatModel,
    output_type: type[T],
    *,
    include_raw: bool = False,
    **kwargs: Any
) -> (
    Runnable[LanguageModelInput, T]
    | Runnable[LanguageModelInput, dict[str, Any]]
)

Build a Runnable that returns validated, structured output, with a JSON-parsing fallback.

This composes chat_model.with_structured_output(output_type, include_raw=True) -- itself a :class:~langchain_core.runnables.Runnable returning {"raw": AIMessage, "parsed": T | None, "parsing_error": Exception | None} -- with a small unwrapping step piped after it via |.

If the chat model's native structured-output parsing fails (e.g. the model didn't emit a proper tool call, which is common with small or local models), the unwrap step falls back to manually parsing the raw message content as JSON, without making a second LLM call.

include_raw controls both the output shape and the failure behavior, mirroring with_structured_output's own contract:

• include_raw=False (default): invoking returns T directly. If both native parsing and the JSON fallback fail, this raises :class:StructuredOutputError.
• include_raw=True: invoking returns a dict with the same "raw"/"parsed"/"parsing_error" keys as with_structured_output(..., include_raw=True), plus a "used_fallback": bool key. This mode never raises on parse failure, matching the underlying method's own fail-open contract: "parsed" is populated whenever native parsing or the JSON fallback succeeds, and "parsing_error" is only set if both fail.

Because the result is a plain |-composed :class:~langchain_core.runnables.RunnableSequence, it already implements invoke, ainvoke, batch, abatch, stream, astream, and config propagation -- nothing here reimplements the Runnable interface. batch/abatch in particular delegate to each step's own batch implementation (so the chat model's native batching is preserved), and correctly skip re-processing items that already failed when return_exceptions=True.

Parameters:

Name	Type	Description	Default
chat_model	BaseChatModel	The chat model to wrap.	required
output_type	type[T]	The type (e.g. a Pydantic model) that the LLM output should be parsed into.	required
include_raw	bool	If False (default), invoking returns output_type directly and raises :class:StructuredOutputError on total parse failure. If True, invoking returns a {"raw", "parsed", "parsing_error", "used_fallback"} dict and never raises on parse failure.	False
**kwargs	Any	Additional keyword arguments forwarded to chat_model.with_structured_output (e.g. method, strict), letting callers tune the native structured-output behavior without bypassing this wrapper's JSON-parsing fallback.	{}

Returns:

Type	Description
Runnable[LanguageModelInput, T] | Runnable[LanguageModelInput, dict[str, Any]]	A Runnable[LanguageModelInput, T] if include_raw=False,
Runnable[LanguageModelInput, T] | Runnable[LanguageModelInput, dict[str, Any]]	or a Runnable[LanguageModelInput, dict[str, Any]] if
Runnable[LanguageModelInput, T] | Runnable[LanguageModelInput, dict[str, Any]]	include_raw=True.

Example

>>> from pydantic import BaseModel
>>> class Answer(BaseModel):
...     value: int
...
>>> chain = structured_output_runnable(chat_model, Answer)  # doctest: +SKIP
>>> chain.invoke("What is 2+2?")  # doctest: +SKIP
Answer(value=4)

zenpyre.runnables.factory

Contain factories for runnables.

zenpyre.runnables.factory.BaseRunnableFactory

Bases: ABC, Generic[Input, Output]

Abstract base class for Runnable factories.

Subclasses implement :meth:make_runnable to instantiate and return a configured :class:~langchain_core.runnables.Runnable object. This pattern decouples Runnable creation from the rest of the codebase, making it easy to swap implementations without changing call sites.

Example

>>> from typing import Any
>>> from langchain_core.runnables import Runnable
>>> from zenpyre.runnables.factory import BaseRunnableFactory
>>> class MyRunnableFactory(BaseRunnableFactory):
...     def make_runnable(self) -> Runnable[Any, Any]:
...         class MyRunnable(Runnable[Any, Any]):
...             def invoke(self, input: Any, config: Any = None, **kwargs: Any) -> Any:
...                 return input
...         return MyRunnable()
...
>>> factory = MyRunnableFactory()
>>> runnable = factory.make_runnable()

zenpyre.runnables.factory.BaseRunnableFactory.make_runnable abstractmethod

make_runnable() -> Runnable[Input, Output]

Create and return a configured Runnable instance.

Returns:

Name	Type	Description
A	Runnable[Input, Output]	class:~langchain_core.runnables.Runnable
	Runnable[Input, Output]	instance ready for use.

zenpyre.runnables.factory.ConfigurableRunnableFactory

Bases: BaseRunnableFactory[Input, Output], MultilineDisplayMixin

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

When a dict is provided it is resolved at each :meth:make_runnable 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
runnable	Runnable[Input, Output] | dict[str, Any]	A fully configured :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 typing import Any
>>> from langchain_core.runnables import Runnable
>>> from zenpyre.runnables.factory import ConfigurableRunnableFactory
>>> class MyRunnable(Runnable[Any, Any]):
...     def invoke(self, input: Any, config: Any = None, **kwargs: Any) -> Any:
...         return input
...
>>> factory = ConfigurableRunnableFactory(MyRunnable())
>>> runnable = factory.make_runnable()

zenpyre.runnables.factory.RunnableFactory

Bases: BaseRunnableFactory[Input, Output], MultilineDisplayMixin

A concrete Runnable factory that wraps a pre-built :class:~langchain_core.runnables.Runnable instance.

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

Parameters:

Name	Type	Description	Default
runnable	Runnable[Input, Output]	A fully configured :class:~langchain_core.runnables.Runnable instance to return from :meth:make_runnable.	required

Example

>>> from typing import Any
>>> from langchain_core.runnables import Runnable
>>> from zenpyre.runnables.factory import RunnableFactory
>>> class MyRunnable(Runnable[Any, Any]):
...     def invoke(self, input: Any, config: Any = None, **kwargs: Any) -> Any:
...         return input
...
>>> factory = RunnableFactory(MyRunnable())
>>> runnable = factory.make_runnable()