_seq_input_schema() — langchain Function Reference

Architecture documentation for the _seq_input_schema() function in base.py from the langchain codebase.

Function python CoreAbstractions MessageSchema called by 1

Entity Profile

CoreAbstractions→ MessageSchema→ _seq_input_schema() — langchain Function Reference

Dependency Diagram

graph TD
  8cd00315_4203_1303_e452_46da88c7bb11["_seq_input_schema()"]
  5f3c1f1c_6f8a_e293_7cb5_97c21b4bf214["base.py"]
  8cd00315_4203_1303_e452_46da88c7bb11 -->|defined in| 5f3c1f1c_6f8a_e293_7cb5_97c21b4bf214
  46e37b5b_f6c1_727b_5e9b_8951d46e5dba["get_input_schema()"]
  46e37b5b_f6c1_727b_5e9b_8951d46e5dba -->|calls| 8cd00315_4203_1303_e452_46da88c7bb11
  style 8cd00315_4203_1303_e452_46da88c7bb11 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 2733–2760

def _seq_input_schema(
    steps: list[Runnable[Any, Any]], config: RunnableConfig | None
) -> type[BaseModel]:
    # Import locally to prevent circular import
    from langchain_core.runnables.passthrough import (  # noqa: PLC0415
        RunnableAssign,
        RunnablePick,
    )

    first = steps[0]
    if len(steps) == 1:
        return first.get_input_schema(config)
    if isinstance(first, RunnableAssign):
        next_input_schema = _seq_input_schema(steps[1:], config)
        if not issubclass(next_input_schema, RootModel):
            # it's a dict as expected
            return create_model_v2(
                "RunnableSequenceInput",
                field_definitions={
                    k: (v.annotation, v.default)
                    for k, v in next_input_schema.model_fields.items()
                    if k not in first.mapper.steps__
                },
            )
    elif isinstance(first, RunnablePick):
        return _seq_input_schema(steps[1:], config)

    return first.get_input_schema(config)