Bind lifecycle listeners to a Runnable, returning a new Runnable.
The Run object contains information about the run, including its id,
type, input, output, error, start_time, end_time, and
any tags or metadata added to the run.
with_listeners(
self,
*,
on_start: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None,
on_end: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None,
on_error: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None
) -> Runnable[Input, Output]Example:
from langchain_core.runnables import RunnableLambda
from langchain_core.tracers.schemas import Run
import time
def test_runnable(time_to_sleep: int):
time.sleep(time_to_sleep)
def fn_start(run_obj: Run):
print("start_time:", run_obj.start_time)
def fn_end(run_obj: Run):
print("end_time:", run_obj.end_time)
chain = RunnableLambda(test_runnable).with_listeners(
on_start=fn_start, on_end=fn_end
)
chain.invoke(2)| Name | Type | Description |
|---|---|---|
on_start | Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None | Default: NoneCalled before the |
on_end | Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None | Default: NoneCalled after the |
on_error | Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None | Default: NoneCalled if the |