_seq_output_schema() — langchain Function Reference

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

Function python CoreAbstractions MessageSchema called by 1

Entity Profile

CoreAbstractions→ MessageSchema→ _seq_output_schema() — langchain Function Reference

Dependency Diagram

graph TD
  4aed9cf0_9ed2_c8d3_1664_f44d1bf04f3b["_seq_output_schema()"]
  5f3c1f1c_6f8a_e293_7cb5_97c21b4bf214["base.py"]
  4aed9cf0_9ed2_c8d3_1664_f44d1bf04f3b -->|defined in| 5f3c1f1c_6f8a_e293_7cb5_97c21b4bf214
  53ea2a4a_9e29_88d1_5137_c8832858939e["get_output_schema()"]
  53ea2a4a_9e29_88d1_5137_c8832858939e -->|calls| 4aed9cf0_9ed2_c8d3_1664_f44d1bf04f3b
  style 4aed9cf0_9ed2_c8d3_1664_f44d1bf04f3b fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/base.py lines 2763–2811

def _seq_output_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,
    )

    last = steps[-1]
    if len(steps) == 1:
        return last.get_input_schema(config)
    if isinstance(last, RunnableAssign):
        mapper_output_schema = last.mapper.get_output_schema(config)
        prev_output_schema = _seq_output_schema(steps[:-1], config)
        if not issubclass(prev_output_schema, RootModel):
            # it's a dict as expected
            return create_model_v2(
                "RunnableSequenceOutput",
                field_definitions={
                    **{
                        k: (v.annotation, v.default)
                        for k, v in prev_output_schema.model_fields.items()
                    },
                    **{
                        k: (v.annotation, v.default)
                        for k, v in mapper_output_schema.model_fields.items()
                    },
                },
            )
    elif isinstance(last, RunnablePick):
        prev_output_schema = _seq_output_schema(steps[:-1], config)
        if not issubclass(prev_output_schema, RootModel):
            # it's a dict as expected
            if isinstance(last.keys, list):
                return create_model_v2(
                    "RunnableSequenceOutput",
                    field_definitions={
                        k: (v.annotation, v.default)
                        for k, v in prev_output_schema.model_fields.items()
                        if k in last.keys
                    },
                )
            field = prev_output_schema.model_fields[last.keys]
            return create_model_v2(
                "RunnableSequenceOutput", root=(field.annotation, field.default)
            )

    return last.get_output_schema(config)