ontocast.agent

Agent module for OntoCast.

This module provides a collection of agents that handle various aspects of ontology processing, including document conversion, text chunking, fact aggregation, and ontology management. Each agent is designed to perform a specific task in the ontology processing pipeline.

The names re-exported here are the pipeline steps the graph and the unit loop drive. Helper entry points that are only meaningful inside one agent -- the *_fresh render paths, the evidence planner/fetcher -- are reached through their own module rather than from here.

ensure_unit_summary(state, unit_index, tools, budget_tracker=None) async

Summarise one content unit in place, if it is due one and lacks one.

Called from inside the extraction fan-outs rather than from a preceding node. A unit's summary depends only on that unit, so a document-level summarize stage made every unit wait for the slowest summary before any extraction could start, for no dependency.

Idempotent, so the facts fan-out is a no-op when the ontology fan-out already summarised the unit. Failures are logged and leave summary as None: extraction then falls back to the unit's full text.

Parameters:

Name	Type	Description	Default
state	Any	Document state; content_units[unit_index] is mutated.	required
unit_index	int	Index of the unit to summarise.	required
tools	ToolBox	Tool container providing the LLM.	required
budget_tracker	Any	Charged for the call.	None

Source code in ontocast/agent/summarize_chunks.py

async def ensure_unit_summary(
    state: Any,
    unit_index: int,
    tools: ToolBox,
    budget_tracker: Any = None,
) -> None:
    """Summarise one content unit in place, if it is due one and lacks one.

    Called from inside the extraction fan-outs rather than from a preceding
    node. A unit's summary depends only on that unit, so a document-level
    summarize stage made every unit wait for the *slowest* summary before any
    extraction could start, for no dependency.

    Idempotent, so the facts fan-out is a no-op when the ontology fan-out
    already summarised the unit. Failures are logged and leave ``summary`` as
    ``None``: extraction then falls back to the unit's full text.

    Args:
        state: Document state; ``content_units[unit_index]`` is mutated.
        unit_index: Index of the unit to summarise.
        tools: Tool container providing the LLM.
        budget_tracker: Charged for the call.
    """
    unit = state.content_units[unit_index]
    if unit.summary is not None:
        return
    if not state.use_summarization:
        return
    if not should_summarize_unit(unit, state.summarize_sections):
        return
    try:
        unit.summary = await summarize_chunk(
            unit,
            tools,
            max_sentences=state.summary_max_sentences,
            budget_tracker=budget_tracker,
        )
    except Exception as exc:
        logger.warning("Summarization failed for unit %s: %s", unit_index, exc)

normalize_ontology_units(units, tools, base_ontology=None, require_base=False, delete_graph=None)

Merge ontology unit deltas as TripleOps, then apply to base ontology.

Units contain ontology insert delta graphs; delete_graph carries the reconciled delete delta (triples to remove from the base). Deletes execute first, then inserts, as one ordered GraphUpdate — so the applied update list feeds version-bump analysis with true operation types.

Parameters:

Name	Type	Description	Default
units	list[ContentUnit]	ContentUnits with type=ONTOLOGIES and delta graph from each unit.	required
tools	ToolBox	ToolBox instance.	required
base_ontology	Ontology | None	Optional ontology to use as base; merged delta is applied to it.	None
require_base	bool	Whether map/reduce caller expects a base ontology.	False
delete_graph	RDFGraph | None	Optional triples to delete from the base before inserts. Requires a catalog base; ignored (with a warning) otherwise.	None

Returns:

Type	Description
Ontology	Tuple of ( ontology with cleaned graph, list of applied GraphUpdates for versioning, provenance artifact graph stripped from ontology output,
list[GraphUpdate]	).

Source code in ontocast/agent/normalize_ontology.py

def normalize_ontology_units(
    units: list[ContentUnit],
    tools: ToolBox,
    base_ontology: Ontology | None = None,
    require_base: bool = False,
    delete_graph: RDFGraph | None = None,
) -> tuple[Ontology, list[GraphUpdate], RDFGraph]:
    """Merge ontology unit deltas as TripleOps, then apply to base ontology.

    Units contain ontology insert delta graphs; ``delete_graph`` carries the
    reconciled delete delta (triples to remove from the base). Deletes execute
    first, then inserts, as one ordered GraphUpdate — so the applied update
    list feeds version-bump analysis with true operation types.

    Args:
        units: ContentUnits with type=ONTOLOGIES and delta graph from each unit.
        tools: ToolBox instance.
        base_ontology: Optional ontology to use as base; merged delta is applied to it.
        require_base: Whether map/reduce caller expects a base ontology.
        delete_graph: Optional triples to delete from the base before inserts.
            Requires a catalog base; ignored (with a warning) otherwise.

    Returns:
        Tuple of (
            ontology with cleaned graph,
            list of applied GraphUpdates for versioning,
            provenance artifact graph stripped from ontology output,
        ).
    """
    has_deletes = delete_graph is not None and len(delete_graph) > 0
    if not units and not has_deletes:
        if base_ontology is not None:
            return base_ontology, [], RDFGraph()
        return Ontology(graph=RDFGraph()), [], RDFGraph()

    for unit in units:
        unit.sanitize()
    _ = tools

    if require_base and (base_ontology is None or base_ontology.is_null()):
        logger.warning(
            "normalize_ontology_units expected a base ontology but none was available; "
            "continuing with merged aggregated ontology output."
        )

    operations: list[TripleOp] = []
    if has_deletes:
        if base_ontology is None or base_ontology.is_null():
            logger.warning(
                "normalize_ontology_units received %d delete triple(s) without a "
                "catalog base; deletes are dropped (nothing to delete from).",
                len(delete_graph) if delete_graph is not None else 0,
            )
        else:
            operations.append(TripleOp(type="delete", graph=delete_graph))
    operations.extend(
        TripleOp(type="insert", graph=unit.graph)
        for unit in units
        if len(unit.graph) > 0
    )
    merged_update = GraphUpdate(triple_operations=operations)
    if not merged_update.triple_operations:
        merged_update = None

    if base_ontology is not None and not base_ontology.is_null():
        base_graph = base_ontology.graph
        if merged_update is not None:
            updated_graph, _ = AgentState.render_updated_graph(
                base_graph, [merged_update], max_triples=None
            )
            graph_changed = set(updated_graph) != set(base_graph)
            if graph_changed:
                result = base_ontology.derive_updated_version(updated_graph)
            else:
                result = base_ontology.model_copy(deep=True)
                result.graph = updated_graph
        else:
            result = base_ontology.model_copy(deep=True)
        result.sync_properties_to_graph()
        cleaned_graph, provenance_graph = split_ontology_and_provenance_graph(
            result.graph
        )
        result.graph = cleaned_graph
        result.sync_properties_to_graph()
        applied = [merged_update] if merged_update else []
        return result, applied, provenance_graph

    aggregated_delta = RDFGraph()
    bindings: dict[str, str] = {}
    for unit in units:
        for triple in unit.graph:
            aggregated_delta.add(triple)
        incoming = {
            prefix: str(namespace)
            for prefix, namespace in unit.graph.namespaces()
            if prefix
        }
        bindings = merge_namespace_bindings(bindings, incoming)
    for prefix, namespace in bindings.items():
        aggregated_delta.bind(prefix, namespace)

    cleaned_graph, provenance_graph = split_ontology_and_provenance_graph(
        aggregated_delta
    )
    if base_ontology is not None and not base_ontology.is_null():
        result = Ontology(
            graph=cleaned_graph,
            ontology_id=base_ontology.ontology_id,
            title=base_ontology.title,
            description=base_ontology.description,
            iri=base_ontology.iri,
        )
    else:
        # No catalog base: provisional Ontology from first domain namespace stem.
        anchor = _working_anchor_from_graph(cleaned_graph) or NULL_ONTOLOGY.iri
        result = Ontology(
            graph=cleaned_graph,
            ontology_id=None,
            iri=anchor if anchor != NULL_ONTOLOGY.iri else NULL_ONTOLOGY.iri,
            title=None,
            description=None,
            skip_graph_identity_sync=True,
        )
    applied = [merged_update] if merged_update else []
    return result, applied, provenance_graph

select_catalog_ontology_for_excerpt(ontology_manager, llm_tool, excerpt, ontology_selection_user_instruction='') async

Use the LLM to select one catalog ontology, or :data:NULL_ONTOLOGY if none fit.

The excerpt is usually a content unit's text. Empty excerpt or an empty catalog yields NULL_ONTOLOGY without calling the model.

Source code in ontocast/agent/select_ontology_catalog.py

async def select_catalog_ontology_for_excerpt(
    ontology_manager: OntologyManager,
    llm_tool: LLMTool,
    excerpt: str,
    ontology_selection_user_instruction: str = "",
) -> Ontology:
    """Use the LLM to select one catalog ontology, or :data:`NULL_ONTOLOGY` if none fit.

    The excerpt is usually a content unit's text. Empty excerpt or an empty
    catalog yields ``NULL_ONTOLOGY`` without calling the model.
    """
    text = excerpt.strip()
    if not text or not ontology_manager.has_ontologies:
        return NULL_ONTOLOGY

    ontologies = ontology_manager.ontologies
    num_ontologies = len(ontologies)
    if num_ontologies == 0:
        return NULL_ONTOLOGY

    lines: list[str] = []
    for i, o in enumerate(ontologies, start=1):
        lines.append(f"{i}. {o.describe()}")
    ontologies_list = "\n\n".join(lines)
    none_index = num_ontologies + 1

    model_cls = create_ontology_selector_report_model(num_ontologies)
    parser = PydanticOutputParser(pydantic_object=model_cls)
    prompt = PromptTemplate(
        template=template_prompt,
        input_variables=[
            "excerpt",
            "ontologies_list",
            "num_ontologies",
            "none_index",
            "ontology_selection_user_instruction",
            "format_instructions",
        ],
    )

    selector = await call_llm_with_retry(
        llm_tool=llm_tool,
        prompt=prompt,
        parser=parser,
        prompt_kwargs={
            "excerpt": text,
            "ontologies_list": ontologies_list,
            "num_ontologies": num_ontologies,
            "none_index": none_index,
            "ontology_selection_user_instruction": ontology_selection_user_instruction.strip(),
            "format_instructions": parser.get_format_instructions(),
        },
    )

    idx = selector.answer_index
    if idx == none_index:
        logger.debug("LLM selected: no suitable catalog ontology (none index)")
        return NULL_ONTOLOGY
    if 1 <= idx <= num_ontologies:
        return ontologies[idx - 1]
    logger.warning("Invalid answer_index %s from selector; using NULL_ONTOLOGY", idx)
    return NULL_ONTOLOGY