Aggregation utils

Aggregation functions for combining structured outputs from multiple extractions.

Classes:

Name	Description
AggregationError	Raised when aggregation fails due to conflicts or type inconsistencies.

Functions:

Name	Description
aggregate_majority_vote	Aggregate outputs by majority vote per key.
aggregate_unanimous	Aggregate outputs by requiring unanimous agreement per key.
aggregate_single_majority_vote_multi_union	Aggregate outputs with majority vote for single-valued fields and union for list fields.
collect_values_and_type_per_key	Collect values and type per key across a list of outputs.
make_hashable_simple	Recursively convert a value to a hashable representation.

AggregationError

Bases: ValueError

Raised when aggregation of structured outputs fails due to conflicts or inconsistencies.

make_hashable_simple(value)

Takes any pyObj and recursively tries to make it hashable.

list/ set: Are converted to sorted tuples, with their elements processed recursively. tuples: Stay tuples, with their elements processed recursively. dicts: Are converted to sorted tuples of (key, value) tuples, with the values processed recursively. The keys stay unchanged.

Warning

Unhashable types that are not list, set, or dict are not handled and will cause the result to stay unhashable.

Parameters:

Name	Type	Description	Default
value	Any	Obj to make hashable.	required

Returns:

Type	Description
Any	Hashable version of the input value.

Source code in src/kibad_llm/extractors/aggregation_utils.py

def make_hashable_simple(value: Any) -> Any:
    """Takes any pyObj and recursively tries to make it hashable.

    list/ set: Are converted to sorted tuples, with their elements processed recursively.
    tuples: Stay tuples, with their elements processed recursively.
    dicts: Are converted to sorted tuples of (key, value) tuples, with the values processed recursively.
        The keys stay unchanged.

    Warning:
        Unhashable types that are not `list`, `set`, or `dict` are not handled and will cause the result to
        stay unhashable.

    Args:
        value: Obj to make hashable.

    Returns:
        Hashable version of the input value.
    """
    if isinstance(value, (list, set)):
        # sort and remove None values
        return tuple(sorted(make_hashable_simple(v) for v in value if v is not None))
    if isinstance(value, tuple):
        # keep order and None values
        return tuple(make_hashable_simple(v) for v in value)
    if isinstance(value, dict):
        # sort and remove None values
        return tuple(
            sorted((k, make_hashable_simple(v)) for k, v in value.items() if v is not None)
        )
    return value

collect_values_and_type_per_key(structured_outputs, skip_type_mismatches=False)

Collect values and types per key from structured outputs.

Parameters:

Name	Type	Description	Default
structured_outputs	list[dict[str, Any] | None]	list of structured outputs from multiple extractions	required
skip_type_mismatches	bool	If True, skips keys with inconsistent types across extractions instead of raising an error (default: False)	False

Returns:

Type	Description
tuple[dict[str, list[Any]], dict[str, type | None]]	tuple of: - dict mapping keys to list of values - dict mapping keys to their consistent type (or None if all values are None)

Raises: AggregationError: If a key has inconsistent types and skip_type_mismatches is False.

Source code in src/kibad_llm/extractors/aggregation_utils.py

def collect_values_and_type_per_key(
    structured_outputs: list[dict[str, Any] | None], skip_type_mismatches: bool = False
) -> tuple[dict[str, list[Any]], dict[str, type | None]]:
    """Collect values and types per key from structured outputs.

    Args:
        structured_outputs: list of structured outputs from multiple extractions
        skip_type_mismatches: If True, skips keys with inconsistent types across extractions
            instead of raising an error (default: False)

    Returns:
        tuple of:
            - dict mapping keys to list of values
            - dict mapping keys to their consistent type (or None if all values are None)
    Raises:
        AggregationError: If a key has inconsistent types and skip_type_mismatches is False.
    """
    # collect all keys to correctly handle missing entries
    all_keys: set[str] = set()
    for res in structured_outputs:
        if res is not None:
            all_keys.update(res.keys())
    values_per_key = defaultdict(list)
    type_per_key: dict[str, type | None] = dict()
    # get values and type per key
    for res in structured_outputs:
        # skip if complete structured_output is None (LLM query failed)
        if res is not None:
            for key in all_keys:
                value = res.get(key, None)
                values_per_key[key].append(value)
                if value is not None:
                    if key not in type_per_key:
                        type_per_key[key] = type(value)
                    else:
                        if type_per_key[key] != type(value):
                            if not skip_type_mismatches:
                                raise AggregationError(
                                    f"Inconsistent types for key '{key}': "
                                    f"{type_per_key[key]} vs {type(value)}"
                                )
                            else:
                                type_per_key[key] = None
    return values_per_key, type_per_key

aggregate_majority_vote(structured_outputs, skip_type_mismatches=False)

Aggregate structured outputs from multiple extractions.

Entries with the same key are aggregated based on their value types: - Primitive types (str, int, float, bool): majority vote - Dict types: majority vote - List types: majority vote per item

This is meant to aggregate outputs from repeated queries with the same schema, where each extraction may produce slightly different results due to LLM variability. The majority vote ensures that only values consistently appearing across extractions are kept, reducing noise and improving reliability.

Parameters:

Name	Type	Description	Default
structured_outputs	InputType	list of structured outputs from multiple extractions	required
skip_type_mismatches	bool	If True, skips keys with inconsistent types across extractions instead of raising an error (default: False)	False

Returns:

Type	Description
OutputType	aggregated structured output or None if all entries are None

Raises:

Type	Description
NotImplementedError	If a value type is encountered that has not been listed above.

Source code in src/kibad_llm/extractors/aggregation_utils.py

def aggregate_majority_vote(
    structured_outputs: InputType, skip_type_mismatches: bool = False
) -> OutputType:
    """Aggregate structured outputs from multiple extractions.

    Entries with the same key are aggregated based on their value types:
    - Primitive types (str, int, float, bool): majority vote
    - Dict types: majority vote
    - List types: majority vote per item

    This is meant to aggregate outputs from repeated queries with the same schema,
    where each extraction may produce slightly different results due to LLM variability.
    The majority vote ensures that only values consistently appearing across extractions
    are kept, reducing noise and improving reliability.

    Args:
        structured_outputs: list of structured outputs from multiple extractions
        skip_type_mismatches: If True, skips keys with inconsistent types across extractions
            instead of raising an error (default: False)

    Returns:
        aggregated structured output or None if all entries are None

    Raises:
        NotImplementedError: If a value type is encountered that has not been listed above.
    """
    if all(res is None for res in structured_outputs):
        return None

    values_per_key, type_per_key = collect_values_and_type_per_key(
        structured_outputs, skip_type_mismatches=skip_type_mismatches
    )

    aggregated: dict[str, Any] = dict()
    for key, values in values_per_key.items():
        value_type = type_per_key.get(key, None)
        if value_type is None:
            # if all values are None
            aggregated[key] = None
        else:
            # Aggregate based on type
            # Note: None values participate in voting intentionally. For repeated queries
            # on the same input, frequent None results indicate genuine extraction difficulty,
            # and the aggregated result should reflect this uncertainty.
            if issubclass(value_type, (str, int, float, bool)):
                # single-value: majority vote for primitive types
                aggregated[key] = _majority_vote(values, exclude_none=False)
            elif issubclass(value_type, dict):
                # single-value: majority vote for dicts
                values_hashable = [
                    make_hashable_simple(v) if v is not None else None for v in values
                ]
                majority_hashable = _majority_vote(values_hashable, exclude_none=False)
                # convert back to dict
                mapping = dict(zip(values_hashable, values))
                aggregated[key] = (
                    mapping[majority_hashable] if majority_hashable is not None else None
                )
            elif issubclass(value_type, list):
                # multi-value: majority vote per item for list types
                # explicitly pass the number of structured outputs since some values may
                # be None and thus not in current values
                aggregated[key] = _multi_entry_majority_vote(values)
            else:
                raise NotImplementedError(f"Unsupported value type for aggregation: {value_type}")

    return aggregated

aggregate_unanimous(structured_outputs, skip_type_mismatches=False)

Aggregate structured outputs with non-overlapping keys.

Combines results from multiple extractions where each extraction is expected to populate different keys (e.g., when a complex schema is split into multiple simpler queries). Each key should appear with a non-None value in at most one extraction.

Parameters:

Name	Type	Description	Default
structured_outputs	InputType	list of structured outputs from multiple extractions	required
skip_type_mismatches	bool	If True, skips keys with inconsistent types across extractions instead of raising an error (default: False)	False

Returns:

Type	Description
OutputType	aggregated structured output or None if all entries are None

Raises:

Type	Description
AggregationError	If the same key has non-None values in multiple extractions. (Through _aggregate_unanimous)
NotImplementedError	If the encountered value is not in (str, int, float, bool, dict, list).

Source code in src/kibad_llm/extractors/aggregation_utils.py

def aggregate_unanimous(
    structured_outputs: InputType, skip_type_mismatches: bool = False
) -> OutputType:
    """Aggregate structured outputs with non-overlapping keys.

    Combines results from multiple extractions where each extraction is expected to
    populate different keys (e.g., when a complex schema is split into multiple simpler
    queries). Each key should appear with a non-None value in at most one extraction.

    Args:
        structured_outputs: list of structured outputs from multiple extractions
        skip_type_mismatches: If True, skips keys with inconsistent types across extractions
            instead of raising an error (default: False)

    Returns:
        aggregated structured output or None if all entries are None

    Raises:
        AggregationError: If the same key has non-None values in multiple extractions. (Through `_aggregate_unanimous`)
        NotImplementedError: If the encountered value is not in (str, int, float, bool, dict, list).
    """

    if all(res is None for res in structured_outputs):
        return None

    values_per_key, type_per_key = collect_values_and_type_per_key(
        structured_outputs, skip_type_mismatches=skip_type_mismatches
    )

    aggregated: dict[str, Any] = dict()
    for key, values in values_per_key.items():
        value_type = type_per_key.get(key, None)
        if value_type is None:
            # if all values are None
            aggregated[key] = None
        else:
            # Aggregate based on type: all non-None values must be identical, otherwise raise error
            if issubclass(value_type, (str, int, float, bool)):
                aggregated[key] = _aggregate_unanimous(values)
            elif issubclass(value_type, dict):
                # make dicts hashable for comparison
                values_hashable = [
                    make_hashable_simple(v) if v is not None else None for v in values
                ]
                majority_hashable = _aggregate_unanimous(values_hashable)
                # convert back to dict
                mapping = dict(zip(values_hashable, values))
                aggregated[key] = (
                    mapping[majority_hashable] if majority_hashable is not None else None
                )
            elif issubclass(value_type, list):
                values_hashable = [
                    make_hashable_simple(v) if v is not None else None for v in values
                ]
                majority_hashable = _aggregate_unanimous(values_hashable)
                # convert back to dict
                mapping = dict(zip(values_hashable, values))
                aggregated[key] = (
                    mapping[majority_hashable] if majority_hashable is not None else None
                )
            else:
                raise NotImplementedError(f"Unsupported value type for aggregation: {value_type}")

    return aggregated

aggregate_single_majority_vote_multi_union(structured_outputs, skip_type_mismatches=False)

Aggregate structured outputs from multiple extractions.

Entries with the same key are aggregated based on their value types: - Primitive types (str, int, float, bool): majority vote - Dict types: majority vote - List types: union of all items across extractions

This is meant to aggregate outputs from queries over different document chunks, where single-valued fields (primitives, dicts) should converge to a consistent value via majority vote, while multi-valued fields (lists) may contain different valid items from each chunk that should all be collected.

Parameters:

Name	Type	Description	Default
structured_outputs	InputType	list of structured outputs from multiple extractions	required
skip_type_mismatches	bool	If True, skips keys with inconsistent types across extractions instead of raising an error (default: False)	False

Returns:

Type	Description
OutputType	aggregated structured output or None if all entries are None

Raises:

Type	Description
NotImplementedError	If the encountered value type is not in (str, int, float, bool, dict, list).

Source code in src/kibad_llm/extractors/aggregation_utils.py

def aggregate_single_majority_vote_multi_union(
    structured_outputs: InputType, skip_type_mismatches: bool = False
) -> OutputType:
    """Aggregate structured outputs from multiple extractions.

    Entries with the same key are aggregated based on their value types:
    - Primitive types (str, int, float, bool): majority vote
    - Dict types: majority vote
    - List types: union of all items across extractions

    This is meant to aggregate outputs from queries over different document chunks,
    where single-valued fields (primitives, dicts) should converge to a consistent
    value via majority vote, while multi-valued fields (lists) may contain different
    valid items from each chunk that should all be collected.

    Args:
        structured_outputs: list of structured outputs from multiple extractions
        skip_type_mismatches: If True, skips keys with inconsistent types across extractions
            instead of raising an error (default: False)

    Returns:
        aggregated structured output or None if all entries are None

    Raises:
        NotImplementedError: If the encountered value type is not in (str, int, float, bool, dict, list).
    """

    if all(res is None for res in structured_outputs):
        return None

    values_per_key, type_per_key = collect_values_and_type_per_key(
        structured_outputs, skip_type_mismatches=skip_type_mismatches
    )

    aggregated: dict[str, Any] = dict()
    for key, values in values_per_key.items():
        value_type = type_per_key.get(key, None)
        if value_type is None:
            # if all values are None
            aggregated[key] = None
        else:
            # Aggregate based on type
            if issubclass(value_type, (str, int, float, bool)):
                # single-value: this should be identical across all outputs, raises AggregationError if not
                aggregated[key] = _majority_vote(values, exclude_none=True)
            elif issubclass(value_type, dict):
                # single-value: this should be identical across all outputs, raises AggregationError if not
                # make dicts hashable for comparison
                values_hashable = [
                    make_hashable_simple(v) if v is not None else None for v in values
                ]
                majority_hashable = _majority_vote(values_hashable, exclude_none=True)
                # convert back to dict
                mapping = dict(zip(values_hashable, values))
                aggregated[key] = (
                    mapping[majority_hashable] if majority_hashable is not None else None
                )
            elif issubclass(value_type, list):
                # multi-value: union per item for list types
                aggregated[key] = _multi_entry_union(values)
            else:
                raise NotImplementedError(f"Unsupported value type for aggregation: {value_type}")

    return aggregated