TransformChain Class — langchain Architecture

Architecture documentation for the TransformChain class in transform.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  7144c443_e893_b66a_5e9b_98c6b8dbbaac["TransformChain"]
  f3cef70e_11b0_61c9_7ec0_7308f4b45056["Chain"]
  7144c443_e893_b66a_5e9b_98c6b8dbbaac -->|extends| f3cef70e_11b0_61c9_7ec0_7308f4b45056
  e6c07047_5b21_318e_c458_6d7282d06c1a["transform.py"]
  7144c443_e893_b66a_5e9b_98c6b8dbbaac -->|defined in| e6c07047_5b21_318e_c458_6d7282d06c1a
  4ccfbc6b_6152_f478_6eb4_7b822a2fa40f["_log_once()"]
  7144c443_e893_b66a_5e9b_98c6b8dbbaac -->|method| 4ccfbc6b_6152_f478_6eb4_7b822a2fa40f
  14347cb3_4317_0dde_dd1d_dd89533833be["input_keys()"]
  7144c443_e893_b66a_5e9b_98c6b8dbbaac -->|method| 14347cb3_4317_0dde_dd1d_dd89533833be
  ce421d7b_e328_25f1_6b44_e7cc4a3997e7["output_keys()"]
  7144c443_e893_b66a_5e9b_98c6b8dbbaac -->|method| ce421d7b_e328_25f1_6b44_e7cc4a3997e7
  023ef111_e214_aa65_28ba_021e80a40e68["_call()"]
  7144c443_e893_b66a_5e9b_98c6b8dbbaac -->|method| 023ef111_e214_aa65_28ba_021e80a40e68
  796cc76c_c848_fa6a_171b_5a673a24518d["_acall()"]
  7144c443_e893_b66a_5e9b_98c6b8dbbaac -->|method| 796cc76c_c848_fa6a_171b_5a673a24518d

Relationship Graph

Source Code

libs/langchain/langchain_classic/chains/transform.py lines 20–79

class TransformChain(Chain):
    """Chain that transforms the chain output.

    Example:
        ```python
        from langchain_classic.chains import TransformChain
        transform_chain = TransformChain(input_variables=["text"],
         output_variables["entities"], transform=func())

        ```
    """

    input_variables: list[str]
    """The keys expected by the transform's input dictionary."""
    output_variables: list[str]
    """The keys returned by the transform's output dictionary."""
    transform_cb: Callable[[dict[str, str]], dict[str, str]] = Field(alias="transform")
    """The transform function."""
    atransform_cb: Callable[[dict[str, Any]], Awaitable[dict[str, Any]]] | None = Field(
        None, alias="atransform"
    )
    """The async coroutine transform function."""

    @staticmethod
    @functools.lru_cache
    def _log_once(msg: str) -> None:
        """Log a message once."""
        logger.warning(msg)

    @property
    def input_keys(self) -> list[str]:
        """Expect input keys."""
        return self.input_variables

    @property
    def output_keys(self) -> list[str]:
        """Return output keys."""
        return self.output_variables

    @override
    def _call(
        self,
        inputs: dict[str, str],
        run_manager: CallbackManagerForChainRun | None = None,
    ) -> dict[str, str]:
        return self.transform_cb(inputs)

    @override
    async def _acall(
        self,
        inputs: dict[str, Any],
        run_manager: AsyncCallbackManagerForChainRun | None = None,
    ) -> dict[str, Any]:
        if self.atransform_cb is not None:
            return await self.atransform_cb(inputs)
        self._log_once(
            "TransformChain's atransform is not provided, falling"
            " back to synchronous transform",
        )
        return self.transform_cb(inputs)