_atransform() — langchain Function Reference

Architecture documentation for the _atransform() function in passthrough.py from the langchain codebase.

Function python LangChainCore Runnables calls 3 called by 1

Entity Profile

LangChainCore→ Runnables→ _atransform() — langchain Function Reference

Dependency Diagram

graph TD
  792ae3d8_8499_60c2_cf08_c723693f4e67["_atransform()"]
  e893b6ae_1c28_41f1_aa8b_a66d0f779312["RunnableAssign"]
  792ae3d8_8499_60c2_cf08_c723693f4e67 -->|defined in| e893b6ae_1c28_41f1_aa8b_a66d0f779312
  07dae0e0_e299_4a3b_4925_f46bdb69bdaf["_atransform()"]
  07dae0e0_e299_4a3b_4925_f46bdb69bdaf -->|calls| 792ae3d8_8499_60c2_cf08_c723693f4e67
  07dae0e0_e299_4a3b_4925_f46bdb69bdaf["_atransform()"]
  792ae3d8_8499_60c2_cf08_c723693f4e67 -->|calls| 07dae0e0_e299_4a3b_4925_f46bdb69bdaf
  933ccc14_844d_c42a_d3ca_33f04cdfba48["assign()"]
  792ae3d8_8499_60c2_cf08_c723693f4e67 -->|calls| 933ccc14_844d_c42a_d3ca_33f04cdfba48
  68f322b4_1349_271a_408d_e2eb04ed643e["atransform()"]
  792ae3d8_8499_60c2_cf08_c723693f4e67 -->|calls| 68f322b4_1349_271a_408d_e2eb04ed643e
  style 792ae3d8_8499_60c2_cf08_c723693f4e67 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/passthrough.py lines 595–634

    async def _atransform(
        self,
        values: AsyncIterator[dict[str, Any]],
        run_manager: AsyncCallbackManagerForChainRun,
        config: RunnableConfig,
        **kwargs: Any,
    ) -> AsyncIterator[dict[str, Any]]:
        # collect mapper keys
        mapper_keys = set(self.mapper.steps__.keys())
        # create two streams, one for the map and one for the passthrough
        for_passthrough, for_map = atee(values, 2, lock=asyncio.Lock())
        # create map output stream
        map_output = self.mapper.atransform(
            for_map,
            patch_config(
                config,
                callbacks=run_manager.get_child(),
            ),
            **kwargs,
        )
        # start map output stream
        first_map_chunk_task: asyncio.Task = asyncio.create_task(
            anext(map_output, None),
        )
        # consume passthrough stream
        async for chunk in for_passthrough:
            if not isinstance(chunk, dict):
                msg = "The input to RunnablePassthrough.assign() must be a dict."
                raise ValueError(msg)  # noqa: TRY004

            # remove mapper keys from passthrough chunk, to be overwritten by map output
            filtered = AddableDict(
                {k: v for k, v in chunk.items() if k not in mapper_keys}
            )
            if filtered:
                yield filtered
        # yield map output
        yield await first_map_chunk_task
        async for chunk in map_output:
            yield chunk