_transform() — langchain Function Reference

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

Function python LangChainCore Runnables calls 3 called by 1

Entity Profile

LangChainCore→ Runnables→ _transform() — langchain Function Reference

Dependency Diagram

graph TD
  a51fc650_dd24_4986_7118_5682f1932c8b["_transform()"]
  e893b6ae_1c28_41f1_aa8b_a66d0f779312["RunnableAssign"]
  a51fc650_dd24_4986_7118_5682f1932c8b -->|defined in| e893b6ae_1c28_41f1_aa8b_a66d0f779312
  ce4d1e34_3bb5_37f8_6976_fc0c99c23892["_transform()"]
  ce4d1e34_3bb5_37f8_6976_fc0c99c23892 -->|calls| a51fc650_dd24_4986_7118_5682f1932c8b
  ce4d1e34_3bb5_37f8_6976_fc0c99c23892["_transform()"]
  a51fc650_dd24_4986_7118_5682f1932c8b -->|calls| ce4d1e34_3bb5_37f8_6976_fc0c99c23892
  933ccc14_844d_c42a_d3ca_33f04cdfba48["assign()"]
  a51fc650_dd24_4986_7118_5682f1932c8b -->|calls| 933ccc14_844d_c42a_d3ca_33f04cdfba48
  6b634a0f_8373_1838_7f85_ec4585108419["transform()"]
  a51fc650_dd24_4986_7118_5682f1932c8b -->|calls| 6b634a0f_8373_1838_7f85_ec4585108419
  style a51fc650_dd24_4986_7118_5682f1932c8b fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/passthrough.py lines 538–582

    def _transform(
        self,
        values: Iterator[dict[str, Any]],
        run_manager: CallbackManagerForChainRun,
        config: RunnableConfig,
        **kwargs: Any,
    ) -> Iterator[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 = safetee(values, 2, lock=threading.Lock())

        # create map output stream
        map_output = self.mapper.transform(
            for_map,
            patch_config(
                config,
                callbacks=run_manager.get_child(),
            ),
            **kwargs,
        )

        # get executor to start map output stream in background
        with get_executor_for_config(config) as executor:
            # start map output stream
            first_map_chunk_future = executor.submit(
                next,
                map_output,
                None,
            )
            # consume passthrough stream
            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
                filtered = AddableDict(
                    {k: v for k, v in chunk.items() if k not in mapper_keys}
                )
                if filtered:
                    yield filtered
            # yield map output
            yield cast("dict[str, Any]", first_map_chunk_future.result())
            for chunk in map_output:
                yield chunk