Source code for langchain.chains.sequential
"""Chain pipeline where the outputs of one step feed directly into next."""
from typing import Any, Dict, List, Optional
from langchain.callbacks.manager import (
    AsyncCallbackManagerForChainRun,
    CallbackManagerForChainRun,
)
from langchain.chains.base import Chain
from langchain.pydantic_v1 import Extra, root_validator
from langchain.utils.input import get_color_mapping
[docs]class SequentialChain(Chain):
    """Chain where the outputs of one chain feed directly into next."""
    chains: List[Chain]
    input_variables: List[str]
    output_variables: List[str]  #: :meta private:
    return_all: bool = False
    class Config:
        """Configuration for this pydantic object."""
        extra = Extra.forbid
        arbitrary_types_allowed = True
    @property
    def input_keys(self) -> List[str]:
        """Return expected input keys to the chain.
        :meta private:
        """
        return self.input_variables
    @property
    def output_keys(self) -> List[str]:
        """Return output key.
        :meta private:
        """
        return self.output_variables
    @root_validator(pre=True)
    def validate_chains(cls, values: Dict) -> Dict:
        """Validate that the correct inputs exist for all chains."""
        chains = values["chains"]
        input_variables = values["input_variables"]
        memory_keys = list()
        if "memory" in values and values["memory"] is not None:
            """Validate that prompt input variables are consistent."""
            memory_keys = values["memory"].memory_variables
            if set(input_variables).intersection(set(memory_keys)):
                overlapping_keys = set(input_variables) & set(memory_keys)
                raise ValueError(