Class RunnableEach

libs/core/langchain_core/runnables/base.py:4329–4437

Runnable that delegates calls to another Runnable with 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

class RunnableEach(RunnableEachBase[Input, Output]):
    """Runnable that delegates calls to another Runnable
    with 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:

        .. code-block:: 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
    """

    @classmethod
    def get_lc_namespace(cls) -> List[str]:
        """Get the namespace of the langchain object."""
        return ["langchain", "schema", "runnable"]

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

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

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

    def with_listeners(
        self,
        *,
        on_start: Optional[
            Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]
        ] = None,
        on_end: Optional[
            Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]
        ] = None,
        on_error: Optional[
            Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]