atransform() — langchain Function Reference

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

Function python LangChainCore Runnables calls 2 called by 1

Entity Profile

LangChainCore→ Runnables→ atransform() — langchain Function Reference

Dependency Diagram

graph TD
  3673596e_21a0_2e75_b2b6_4e0b42a59cee["atransform()"]
  4a62481c_02cb_a5de_1833_50669d5351a6["Runnable"]
  3673596e_21a0_2e75_b2b6_4e0b42a59cee -->|defined in| 4a62481c_02cb_a5de_1833_50669d5351a6
  9fd2d82a_5f60_914f_a756_f78c796456f8["atransform()"]
  9fd2d82a_5f60_914f_a756_f78c796456f8 -->|calls| 3673596e_21a0_2e75_b2b6_4e0b42a59cee
  d04ee734_b887_f11d_7551_064b5b024357["astream()"]
  3673596e_21a0_2e75_b2b6_4e0b42a59cee -->|calls| d04ee734_b887_f11d_7551_064b5b024357
  6b0af8a6_40de_913f_d715_e591ebb530c1["atransform()"]
  3673596e_21a0_2e75_b2b6_4e0b42a59cee -->|calls| 6b0af8a6_40de_913f_d715_e591ebb530c1
  style 3673596e_21a0_2e75_b2b6_4e0b42a59cee fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 1564–1608

    async def atransform(
        self,
        input: AsyncIterator[Input],
        config: RunnableConfig | None = None,
        **kwargs: Any | None,
    ) -> AsyncIterator[Output]:
        """Transform inputs to outputs.

        Default implementation of atransform, which buffers input and calls `astream`.

        Subclasses must override this method if they can start producing output while
        input is still being generated.

        Args:
            input: An async iterator of inputs to the `Runnable`.
            config: The config to use for the `Runnable`.
            **kwargs: Additional keyword arguments to pass to the `Runnable`.

        Yields:
            The output of the `Runnable`.

        """
        final: Input
        got_first_val = False

        async for ichunk in input:
            # The default implementation of transform is to buffer input and
            # then call stream.
            # It'll attempt to gather all input into a single chunk using
            # the `+` operator.
            # If the input is not addable, then we'll assume that we can
            # only operate on the last chunk,
            # and 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 got_first_val:
            async for output in self.astream(final, config, **kwargs):
                yield output