RunnableEach Class — langchain Architecture

Source Code

libs/core/langchain_core/runnables/base.py lines 5413–5527

class RunnableEach(RunnableEachBase[Input, Output]):
    """RunnableEach class.

    `Runnable` that calls another `Runnable` for each element of the input sequence.

    It allows you to call multiple inputs with the bounded `Runnable`.

    `RunnableEach` makes it easy to run multiple inputs for the `Runnable`.
    In the below example, we associate and run three inputs
    with a `Runnable`:

        ```python
        from langchain_core.runnables.base import RunnableEach
        from langchain_openai import ChatOpenAI
        from langchain_core.prompts import ChatPromptTemplate
        from langchain_core.output_parsers import StrOutputParser
        prompt = ChatPromptTemplate.from_template("Tell me a short joke about
        {topic}")
        model = ChatOpenAI()
        output_parser = StrOutputParser()
        runnable = prompt | model | output_parser
        runnable_each = RunnableEach(bound=runnable)
        output = runnable_each.invoke([{'topic':'Computer Science'},
                                    {'topic':'Art'},
                                    {'topic':'Biology'}])
        print(output)  # noqa: T201

        ```
    """

    @override
    def get_name(self, suffix: str | None = None, *, name: str | None = None) -> str:
        name = name or self.name or f"RunnableEach<{self.bound.get_name()}>"
        return super().get_name(suffix, name=name)

    @override
    def bind(self, **kwargs: Any) -> RunnableEach[Input, Output]:
        return RunnableEach(bound=self.bound.bind(**kwargs))

    @override
    def with_config(
        self, config: RunnableConfig | None = None, **kwargs: Any
    ) -> RunnableEach[Input, Output]:
        return RunnableEach(bound=self.bound.with_config(config, **kwargs))

    @override
    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,
    ) -> RunnableEach[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.

        """
        return RunnableEach(
            bound=self.bound.with_listeners(
                on_start=on_start, on_end=on_end, on_error=on_error
            )