with_alisteners() — langchain Function Reference

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

Function python LangChainCore Runnables

Entity Profile

LangChainCore→ Runnables→ with_alisteners() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_history.py lines 832–854

    def with_alisteners(
        self,
        *,
        on_start: AsyncListener | None = None,
        on_end: AsyncListener | None = None,
        on_error: AsyncListener | None = None,
    ) -> Runnable[Input, Output]:
        def create_tracer(config: RunnableConfig) -> RunnableConfig:
            tracer = AsyncRootListenersTracer(
                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],
        )