get_unique_config_specs() — langchain Function Reference

Architecture documentation for the get_unique_config_specs() function in utils.py from the langchain codebase.

Function python CoreAbstractions RunnableInterface

Entity Profile

CoreAbstractions→ RunnableInterface→ get_unique_config_specs() — langchain Function Reference

Dependency Diagram

graph TD
  602139c4_b1d2_a84f_ebc8_177d4d52aa81["get_unique_config_specs()"]
  ca66092c_447c_d201_0d3c_cfa6ca2cc9d3["utils.py"]
  602139c4_b1d2_a84f_ebc8_177d4d52aa81 -->|defined in| ca66092c_447c_d201_0d3c_cfa6ca2cc9d3
  style 602139c4_b1d2_a84f_ebc8_177d4d52aa81 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/runnables/utils.py lines 641–670

def get_unique_config_specs(
    specs: Iterable[ConfigurableFieldSpec],
) -> list[ConfigurableFieldSpec]:
    """Get the unique config specs from a sequence of config specs.

    Args:
        specs: The config specs.

    Returns:
        The unique config specs.

    Raises:
        ValueError: If the runnable sequence contains conflicting config specs.
    """
    grouped = groupby(
        sorted(specs, key=lambda s: (s.id, *(s.dependencies or []))), lambda s: s.id
    )
    unique: list[ConfigurableFieldSpec] = []
    for spec_id, dupes in grouped:
        first = next(dupes)
        others = list(dupes)
        if len(others) == 0 or all(o == first for o in others):
            unique.append(first)
        else:
            msg = (
                "RunnableSequence contains conflicting config specs"
                f"for {spec_id}: {[first, *others]}"
            )
            raise ValueError(msg)
    return unique