SimpleSequentialChain Class — langchain Architecture

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

Class python

Entity Profile

Dependency Diagram

graph TD
  d01877d8_1123_f550_8a2f_0981eecde24e["SimpleSequentialChain"]
  f3cef70e_11b0_61c9_7ec0_7308f4b45056["Chain"]
  d01877d8_1123_f550_8a2f_0981eecde24e -->|extends| f3cef70e_11b0_61c9_7ec0_7308f4b45056
  ad5b5c43_91a3_7293_1f45_d2577f9abe35["sequential.py"]
  d01877d8_1123_f550_8a2f_0981eecde24e -->|defined in| ad5b5c43_91a3_7293_1f45_d2577f9abe35
  a68813a1_d4b2_53bb_bb98_e3e9bbb69e30["input_keys()"]
  d01877d8_1123_f550_8a2f_0981eecde24e -->|method| a68813a1_d4b2_53bb_bb98_e3e9bbb69e30
  8a2729ba_855b_eb10_d62b_d296307bb7ef["output_keys()"]
  d01877d8_1123_f550_8a2f_0981eecde24e -->|method| 8a2729ba_855b_eb10_d62b_d296307bb7ef
  dbb29170_cc29_b773_e4de_50231f248825["validate_chains()"]
  d01877d8_1123_f550_8a2f_0981eecde24e -->|method| dbb29170_cc29_b773_e4de_50231f248825
  36348bf6_4ec4_a0d7_3590_010c39d13c55["_call()"]
  d01877d8_1123_f550_8a2f_0981eecde24e -->|method| 36348bf6_4ec4_a0d7_3590_010c39d13c55
  c19f490d_78d9_035d_46c9_7f74bcf787a7["_acall()"]
  d01877d8_1123_f550_8a2f_0981eecde24e -->|method| c19f490d_78d9_035d_46c9_7f74bcf787a7

Relationship Graph

Source Code

libs/langchain/langchain_classic/chains/sequential.py lines 123–208

class SimpleSequentialChain(Chain):
    """Simple chain where the outputs of one step feed directly into next."""

    chains: list[Chain]
    strip_outputs: bool = False
    input_key: str = "input"
    output_key: str = "output"

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

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

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

    @model_validator(mode="after")
    def validate_chains(self) -> Self:
        """Validate that chains are all single input/output."""
        for chain in self.chains:
            if len(chain.input_keys) != 1:
                msg = (
                    "Chains used in SimplePipeline should all have one input, got "
                    f"{chain} with {len(chain.input_keys)} inputs."
                )
                raise ValueError(msg)
            if len(chain.output_keys) != 1:
                msg = (
                    "Chains used in SimplePipeline should all have one output, got "
                    f"{chain} with {len(chain.output_keys)} outputs."
                )
                raise ValueError(msg)
        return self

    def _call(
        self,
        inputs: dict[str, str],
        run_manager: CallbackManagerForChainRun | None = None,
    ) -> dict[str, str]:
        _run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
        _input = inputs[self.input_key]
        color_mapping = get_color_mapping([str(i) for i in range(len(self.chains))])
        for i, chain in enumerate(self.chains):
            _input = chain.run(
                _input,
                callbacks=_run_manager.get_child(f"step_{i + 1}"),
            )
            if self.strip_outputs:
                _input = _input.strip()
            _run_manager.on_text(
                _input,
                color=color_mapping[str(i)],
                end="\n",
                verbose=self.verbose,
            )
        return {self.output_key: _input}

    async def _acall(
        self,
        inputs: dict[str, Any],
        run_manager: AsyncCallbackManagerForChainRun | None = None,
    ) -> dict[str, Any]:
        _run_manager = run_manager or AsyncCallbackManagerForChainRun.get_noop_manager()
        _input = inputs[self.input_key]
        color_mapping = get_color_mapping([str(i) for i in range(len(self.chains))])
        for i, chain in enumerate(self.chains):
            _input = await chain.arun(
                _input,
                callbacks=_run_manager.get_child(f"step_{i + 1}"),
            )
            if self.strip_outputs:
                _input = _input.strip()
            await _run_manager.on_text(
                _input,