with_listeners() — langchain Function Reference

Architecture documentation for the with_listeners() function in base.py from the langchain codebase.

Function python LangChainCore Runnables calls 1 called by 2

Entity Profile

LangChainCore→ Runnables→ with_listeners() — langchain Function Reference

Dependency Diagram

graph TD
  c95b18f5_8485_9405_07af_f9e9f27e3fc6["with_listeners()"]
  4a62481c_02cb_a5de_1833_50669d5351a6["Runnable"]
  c95b18f5_8485_9405_07af_f9e9f27e3fc6 -->|defined in| 4a62481c_02cb_a5de_1833_50669d5351a6
  6606ea6b_af21_fc53_21ca_cf61d28c7999["with_listeners()"]
  6606ea6b_af21_fc53_21ca_cf61d28c7999 -->|calls| c95b18f5_8485_9405_07af_f9e9f27e3fc6
  b764001c_07b7_5cd0_0622_626aeebd081f["with_listeners()"]
  b764001c_07b7_5cd0_0622_626aeebd081f -->|calls| c95b18f5_8485_9405_07af_f9e9f27e3fc6
  b764001c_07b7_5cd0_0622_626aeebd081f["with_listeners()"]
  c95b18f5_8485_9405_07af_f9e9f27e3fc6 -->|calls| b764001c_07b7_5cd0_0622_626aeebd081f
  style c95b18f5_8485_9405_07af_f9e9f27e3fc6 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 1669–1739

    def 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]:
        """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.

        Args:
            on_start: Called before the `Runnable` starts running, with the `Run`
                object.
            on_end: Called after the `Runnable` finishes running, with the `Run`
                object.
            on_error: Called if the `Runnable` throws an error, with the `Run`
                object.

        Returns:
            A new `Runnable` with the listeners bound.

        Example:
            ```python
            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)
            ```
        """
        return RunnableBinding(
            bound=self,
            config_factories=[
                lambda config: {
                    "callbacks": [
                        RootListenersTracer(
                            config=config,
                            on_start=on_start,
                            on_end=on_end,
                            on_error=on_error,
                        )
                    ],
                }
            ],
        )