SequentialChain Class — langchain Architecture

Architecture documentation for the SequentialChain class in sequential.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  68ecb97b_8f93_4457_513e_9002d34f708b["SequentialChain"]
  f3cef70e_11b0_61c9_7ec0_7308f4b45056["Chain"]
  68ecb97b_8f93_4457_513e_9002d34f708b -->|extends| f3cef70e_11b0_61c9_7ec0_7308f4b45056
  ad5b5c43_91a3_7293_1f45_d2577f9abe35["sequential.py"]
  68ecb97b_8f93_4457_513e_9002d34f708b -->|defined in| ad5b5c43_91a3_7293_1f45_d2577f9abe35
  8b89c4ab_85d6_4fa4_c77e_0bb2839cb37e["input_keys()"]
  68ecb97b_8f93_4457_513e_9002d34f708b -->|method| 8b89c4ab_85d6_4fa4_c77e_0bb2839cb37e
  55a71161_c780_2529_e19a_f9d14914ecef["output_keys()"]
  68ecb97b_8f93_4457_513e_9002d34f708b -->|method| 55a71161_c780_2529_e19a_f9d14914ecef
  9a79e356_f008_9ba8_a047_1413b54e13c6["validate_chains()"]
  68ecb97b_8f93_4457_513e_9002d34f708b -->|method| 9a79e356_f008_9ba8_a047_1413b54e13c6
  b66566fe_04a3_9200_5ef2_c270b021325d["_call()"]
  68ecb97b_8f93_4457_513e_9002d34f708b -->|method| b66566fe_04a3_9200_5ef2_c270b021325d
  6da6d25b_9115_3894_df2c_74202eb2c030["_acall()"]
  68ecb97b_8f93_4457_513e_9002d34f708b -->|method| 6da6d25b_9115_3894_df2c_74202eb2c030

Relationship Graph

Source Code

libs/langchain/langchain_classic/chains/sequential.py lines 16–120

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]
    return_all: bool = False

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        extra="forbid",
    )

    @property
    def input_keys(self) -> list[str]:
        """Return expected input keys to the chain."""
        return self.input_variables

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

    @model_validator(mode="before")
    @classmethod
    def validate_chains(cls, values: dict) -> Any:
        """Validate that the correct inputs exist for all chains."""
        chains = values["chains"]
        input_variables = values["input_variables"]
        memory_keys = []
        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)
                msg = (
                    f"The input key(s) {''.join(overlapping_keys)} are found "
                    f"in the Memory keys ({memory_keys}) - please use input and "
                    f"memory keys that don't overlap."
                )
                raise ValueError(msg)

        known_variables = set(input_variables + memory_keys)

        for chain in chains:
            missing_vars = set(chain.input_keys).difference(known_variables)
            if chain.memory:
                missing_vars = missing_vars.difference(chain.memory.memory_variables)

            if missing_vars:
                msg = (
                    f"Missing required input keys: {missing_vars}, "
                    f"only had {known_variables}"
                )
                raise ValueError(msg)
            overlapping_keys = known_variables.intersection(chain.output_keys)
            if overlapping_keys:
                msg = f"Chain returned keys that already exist: {overlapping_keys}"
                raise ValueError(msg)

            known_variables |= set(chain.output_keys)

        if "output_variables" not in values:
            if values.get("return_all", False):
                output_keys = known_variables.difference(input_variables)
            else:
                output_keys = chains[-1].output_keys
            values["output_variables"] = output_keys
        else:
            missing_vars = set(values["output_variables"]).difference(known_variables)
            if missing_vars:
                msg = f"Expected output variables that were not found: {missing_vars}."
                raise ValueError(msg)

        return values

    def _call(
        self,
        inputs: dict[str, str],
        run_manager: CallbackManagerForChainRun | None = None,
    ) -> dict[str, str]: