with_listeners() — langchain Function Reference

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

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ with_listeners() — langchain Function Reference

Dependency Diagram

graph TD
  f7d9faa6_6e66_b602_860d_523bc41ed804["with_listeners()"]
  8699b709_b672_d0ea_67a3_d0d21cb71be9["_RunnableLambdaWithRaiseError"]
  f7d9faa6_6e66_b602_860d_523bc41ed804 -->|defined in| 8699b709_b672_d0ea_67a3_d0d21cb71be9
  style f7d9faa6_6e66_b602_860d_523bc41ed804 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_history.py lines 802–830

    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]:
        def create_tracer(config: RunnableConfig) -> RunnableConfig:
            tracer = RootListenersTracer(
                config=config,
                on_start=on_start,
                on_end=on_end,
                on_error=on_error,
            )
            tracer.raise_error = True
            return {
                "callbacks": [tracer],
            }

        return RunnableBinding(
            bound=self,
            config_factories=[create_tracer],
        )