_transform() — langchain Function Reference

Architecture documentation for the _transform() function in base.py from the langchain codebase.

Function python LangChainCore Runnables calls 1

Entity Profile

LangChainCore→ Runnables→ _transform() — langchain Function Reference

Dependency Diagram

graph TD
  b67c5f0f_1c9e_c943_1c09_d8d8f8f5b97f["_transform()"]
  c6a370e4_a02a_c56a_38eb_7ca734dcbfea["RunnableLambda"]
  b67c5f0f_1c9e_c943_1c09_d8d8f8f5b97f -->|defined in| c6a370e4_a02a_c56a_38eb_7ca734dcbfea
  87d3d0f0_cf29_e59f_63c9_a28e3dff7138["_transform()"]
  b67c5f0f_1c9e_c943_1c09_d8d8f8f5b97f -->|calls| 87d3d0f0_cf29_e59f_63c9_a28e3dff7138
  style b67c5f0f_1c9e_c943_1c09_d8d8f8f5b97f fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 5052–5112

    def _transform(
        self,
        chunks: Iterator[Input],
        run_manager: CallbackManagerForChainRun,
        config: RunnableConfig,
        **kwargs: Any,
    ) -> Iterator[Output]:
        final: Input
        got_first_val = False
        for ichunk in chunks:
            # By definitions, RunnableLambdas consume all input before emitting output.
            # If the input is not addable, then we'll assume that we can
            # only operate on the last chunk.
            # So we'll iterate until we get to the last chunk!
            if not got_first_val:
                final = ichunk
                got_first_val = True
            else:
                try:
                    final = final + ichunk  # type: ignore[operator]
                except TypeError:
                    final = ichunk

        if inspect.isgeneratorfunction(self.func):
            output: Output | None = None
            for chunk in call_func_with_variable_args(
                self.func, final, config, run_manager, **kwargs
            ):
                yield chunk
                if output is None:
                    output = chunk
                else:
                    try:
                        output = output + chunk
                    except TypeError:
                        output = chunk
        else:
            output = call_func_with_variable_args(
                self.func, final, config, run_manager, **kwargs
            )

        # If the output is a Runnable, use its stream output
        if isinstance(output, Runnable):
            recursion_limit = config["recursion_limit"]
            if recursion_limit <= 0:
                msg = (
                    f"Recursion limit reached when invoking {self} with input {final}."
                )
                raise RecursionError(msg)
            for chunk in output.stream(
                final,
                patch_config(
                    config,
                    callbacks=run_manager.get_child(),
                    recursion_limit=recursion_limit - 1,
                ),
            ):
                yield chunk
        elif not inspect.isgeneratorfunction(self.func):
            # Otherwise, just yield it
            yield cast("Output", output)