Source code for langchain.chains.transform
"""Chain that runs an arbitrary python function."""
import functools
import logging
from typing import Any, Awaitable, Callable, Dict, List, Optional
from langchain.callbacks.manager import (
    AsyncCallbackManagerForChainRun,
    CallbackManagerForChainRun,
)
from langchain.chains.base import Chain
from langchain.pydantic_v1 import Field
logger = logging.getLogger(__name__)
[docs]class TransformChain(Chain):
    """Chain that transforms the chain output.
    Example:
        .. code-block:: python
            from langchain.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: Optional[
        Callable[[Dict[str, Any]], Awaitable[Dict[str, Any]]]
    ] = Field(None, alias="atransform")
    """The async coroutine transform function."""
    @staticmethod
    @functools.lru_cache
    def _log_once(msg: str) -> None:
        """Log a message once.
        :meta private:
        """
        logger.warning(msg)
    @property
    def input_keys(self) -> List[str]:
        """Expect input keys.
        :meta private:
        """
        return self.input_variables
    @property
    def output_keys(self) -> List[str]:
        """Return output keys.
        :meta private:
        """