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 1

Entity Profile

LangChainCore→ Runnables→ with_listeners() — langchain Function Reference

Dependency Diagram

graph TD
  b764001c_07b7_5cd0_0622_626aeebd081f["with_listeners()"]
  1ad7446c_5ef5_eeb3_2fc5_22a110a58b17["RunnableBinding"]
  b764001c_07b7_5cd0_0622_626aeebd081f -->|defined in| 1ad7446c_5ef5_eeb3_2fc5_22a110a58b17
  c95b18f5_8485_9405_07af_f9e9f27e3fc6["with_listeners()"]
  c95b18f5_8485_9405_07af_f9e9f27e3fc6 -->|calls| b764001c_07b7_5cd0_0622_626aeebd081f
  c95b18f5_8485_9405_07af_f9e9f27e3fc6["with_listeners()"]
  b764001c_07b7_5cd0_0622_626aeebd081f -->|calls| c95b18f5_8485_9405_07af_f9e9f27e3fc6
  style b764001c_07b7_5cd0_0622_626aeebd081f fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 6022–6072

    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.
        """

        def listener_config_factory(config: RunnableConfig) -> RunnableConfig:
            return {
                "callbacks": [
                    RootListenersTracer(
                        config=config,
                        on_start=on_start,
                        on_end=on_end,
                        on_error=on_error,
                    )
                ],
            }

        return self.__class__(
            bound=self.bound,
            kwargs=self.kwargs,
            config=self.config,
            config_factories=[listener_config_factory, *self.config_factories],
            custom_input_type=self.custom_input_type,
            custom_output_type=self.custom_output_type,
        )