Home / Class/ _RunnableLambdaWithRaiseError Class — langchain Architecture

_RunnableLambdaWithRaiseError Class — langchain Architecture

Architecture documentation for the _RunnableLambdaWithRaiseError class in test_history.py from the langchain codebase.

Entity Profile

Dependency Diagram

graph TD
  8699b709_b672_d0ea_67a3_d0d21cb71be9["_RunnableLambdaWithRaiseError"]
  0af0dfde_fe33_f3e5_b5e6_a3d18260dc67["test_history.py"]
  8699b709_b672_d0ea_67a3_d0d21cb71be9 -->|defined in| 0af0dfde_fe33_f3e5_b5e6_a3d18260dc67
  f7d9faa6_6e66_b602_860d_523bc41ed804["with_listeners()"]
  8699b709_b672_d0ea_67a3_d0d21cb71be9 -->|method| f7d9faa6_6e66_b602_860d_523bc41ed804
  d51b4946_e92e_b91a_2ade_66b7ab26b87a["with_alisteners()"]
  8699b709_b672_d0ea_67a3_d0d21cb71be9 -->|method| d51b4946_e92e_b91a_2ade_66b7ab26b87a

Relationship Graph

Source Code

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

class _RunnableLambdaWithRaiseError(RunnableLambda[Input, Output]):
    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],
        )

    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],
        )

Frequently Asked Questions

What is the _RunnableLambdaWithRaiseError class?
_RunnableLambdaWithRaiseError is a class in the langchain codebase, defined in libs/core/tests/unit_tests/runnables/test_history.py.
Where is _RunnableLambdaWithRaiseError defined?
_RunnableLambdaWithRaiseError is defined in libs/core/tests/unit_tests/runnables/test_history.py at line 801.

Analyze Your Own Codebase

Get architecture documentation, dependency graphs, and domain analysis for your codebase in minutes.

Try Supermodel Free