Union

UnionExtractor for multi-pass extraction with per-pass parameter overrides.

Classes:

Name	Description
UnionExtractor	Runs extraction once per override set on the same text and aggregates the results.

UnionExtractor(overrides, aggregator, return_as_list=None, **kwargs)

Extractor that repeats extraction multiple times and aggregates results per key.

This extractor calls the base extraction function multiple times (for each entry in overrides) on the same input text and aggregates the structured outputs.

Attributes:

Name	Type	Description
aggregator		Aggregator function to use for aggregating results
overrides		A list of dictionaries containing parameter overrides for each extraction.
return_as_list		List of field names to return as lists of all extracted values (default: None)
default_kwargs		Additional keyword arguments passed to the base extraction function.

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

def __init__(
    self,
    overrides: list[dict] | dict[str, dict],
    aggregator: Aggregator,
    return_as_list: list[str] | None = None,
    **kwargs,
):
    if len(overrides) < 1:
        raise ValueError("overrides must contain at least one set of parameters")
    if isinstance(overrides, list):
        overrides = {str(i): override for i, override in enumerate(overrides)}
    self.overrides = overrides
    self.aggregator = aggregator
    self.return_as_list = return_as_list or []
    self.default_kwargs = kwargs

__call__(*args, **kwargs)

Process singular text in multiple passes with different overrides.

Parameters:

Name	Type	Description	Default
*args	Any	Are forwarded unchanged: extract_from_text_lenient	()

Other Parameters:

Name	Type	Description
*	Any	Refer to extract_from_text_lenient

Returns:

Type	Description
dict[str, Any]	Dict with the key structured that holds the aggregated structured outputs.
dict[str, Any]	Additionally, there can be lists for fields at the keys "{field}_list".

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

def __call__(self, *args, **kwargs) -> dict[str, Any]:
    """Process singular text in multiple passes with different overrides.

    Args:
        *args (Any): Are forwarded unchanged:
            [`extract_from_text_lenient`][kibad_llm.extractors.base.extract_from_text_lenient]

    Keyword Args:
        * (Any): Refer to [`extract_from_text_lenient`][kibad_llm.extractors.base.extract_from_text_lenient]

    Returns:
        Dict with the key `structured` that holds the aggregated structured outputs.
        Additionally, there can be lists for fields at the keys `"{field}_list"`.
    """
    combined_kwargs = {**self.default_kwargs, **kwargs}
    results = []
    for override_name, override_params in self.overrides.items():
        current_kwargs = {**combined_kwargs, **override_params}
        current_result = extract_from_text_lenient(*args, **current_kwargs)
        results.append(current_result)

    structured_outputs = [v.get("structured", None) for v in results]
    aggregated_structured = self.aggregator(structured_outputs)

    result: dict[str, Any] = {
        "structured": aggregated_structured,
    }
    for field in self.return_as_list:
        result[f"{field}_list"] = [v.get(field, None) for v in results]
    return result