merge_lists() — langchain Function Reference

Architecture documentation for the merge_lists() function in _merge.py from the langchain codebase.

Function python Observability TelemetryUtilities calls 1 called by 2

Entity Profile

Observability→ TelemetryUtilities→ merge_lists() — langchain Function Reference

Dependency Diagram

graph TD
  fb0c3c36_ad00_663d_0e1a_2f6d8fe4af76["merge_lists()"]
  c118f9bd_45f4_d81d_4bb1_d6bae8563551["_merge.py"]
  fb0c3c36_ad00_663d_0e1a_2f6d8fe4af76 -->|defined in| c118f9bd_45f4_d81d_4bb1_d6bae8563551
  ecab3788_6d79_e81f_eeed_6ee38656c1d7["merge_dicts()"]
  ecab3788_6d79_e81f_eeed_6ee38656c1d7 -->|calls| fb0c3c36_ad00_663d_0e1a_2f6d8fe4af76
  8d8ca17c_07f0_a964_a165_8710319a9927["merge_obj()"]
  8d8ca17c_07f0_a964_a165_8710319a9927 -->|calls| fb0c3c36_ad00_663d_0e1a_2f6d8fe4af76
  ecab3788_6d79_e81f_eeed_6ee38656c1d7["merge_dicts()"]
  fb0c3c36_ad00_663d_0e1a_2f6d8fe4af76 -->|calls| ecab3788_6d79_e81f_eeed_6ee38656c1d7
  style fb0c3c36_ad00_663d_0e1a_2f6d8fe4af76 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/utils/_merge.py lines 82–152

def merge_lists(left: list | None, *others: list | None) -> list | None:
    """Add many lists, handling `None`.

    Args:
        left: The first list to merge.
        others: The other lists to merge.

    Returns:
        The merged list.
    """
    merged = left.copy() if left is not None else None
    for other in others:
        if other is None:
            continue
        if merged is None:
            merged = other.copy()
        else:
            for e in other:
                if (
                    isinstance(e, dict)
                    and "index" in e
                    and (
                        isinstance(e["index"], int)
                        or (
                            isinstance(e["index"], str) and e["index"].startswith("lc_")
                        )
                    )
                ):
                    to_merge = [
                        i
                        for i, e_left in enumerate(merged)
                        if "index" in e_left and e_left["index"] == e["index"]
                    ]
                    if to_merge:
                        # TODO: Remove this once merge_dict is updated with special
                        # handling for 'type'.
                        if (left_type := merged[to_merge[0]].get("type")) and (
                            e.get("type") == "non_standard" and "value" in e
                        ):
                            if left_type != "non_standard":
                                # standard + non_standard
                                new_e: dict[str, Any] = {
                                    "extras": {
                                        k: v
                                        for k, v in e["value"].items()
                                        if k != "type"
                                    }
                                }
                            else:
                                # non_standard + non_standard
                                new_e = {
                                    "value": {
                                        k: v
                                        for k, v in e["value"].items()
                                        if k != "type"
                                    }
                                }
                                if "index" in e:
                                    new_e["index"] = e["index"]
                        else:
                            new_e = (
                                {k: v for k, v in e.items() if k != "type"}
                                if "type" in e
                                else e
                            )
                        merged[to_merge[0]] = merge_dicts(merged[to_merge[0]], new_e)
                    else:
                        merged.append(e)
                else:
                    merged.append(e)
    return merged