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.

render_facts_fresh(state, tools, supplemental_ontologies=None) async

Render fresh facts from the current chunk into Turtle format.

Parameters:

Name	Type	Description	Default
state	UnitFactsState	The current unit facts state containing the chunk to render.	required
tools	AtomicToolBox	The toolbox instance providing utility functions.	required

Returns:

Name	Type	Description
UnitFactsState	UnitFactsState	Updated state with rendered facts.

Source code in ontocast/agent/render_facts.py

async def render_facts_fresh(
    state: UnitFactsState,
    tools: AtomicToolBox,
    supplemental_ontologies: Sequence[Ontology] | None = None,
) -> UnitFactsState:
    """Render fresh facts from the current chunk into Turtle format.

    Args:
        state: The current unit facts state containing the chunk to render.
        tools: The toolbox instance providing utility functions.

    Returns:
        UnitFactsState: Updated state with rendered facts.
    """
    logger.info("Rendering fresh facts")
    state.quarantined_literal_triples = []
    llm_tool = await tools.get_llm_tool(state.budget_tracker)
    profile = get_graph_format_profile(state.llm_graph_format)
    parser = PydanticOutputParser(pydantic_object=FactsRenderReport)

    access = ontology_access_for_unit_facts(state)

    known_prefixes = build_llm_prefix_map(
        access.ontology_for_prefixes(),
        supplemental_ontologies or (),
    )

    web_search_enabled = tools.web_grounding_enabled_for_node(
        WorkflowNode.TEXT_TO_FACTS
    )
    prompt_data = _prepare_prompt_data(
        state,
        access,
        profile,
        search_guidelines=search_guidelines_for(
            WorkflowNode.TEXT_TO_FACTS, web_search_enabled
        ),
    )
    prompt_data_fresh = {
        "preamble": preamble,
        "improvement_instruction": "",
        "output_instruction": profile.render_fresh_output_instruction(target="facts"),
    }
    prompt_data.update(prompt_data_fresh)

    prompt = _create_prompt_template()

    try:
        # Set known prefixes in context before parsing
        RDFGraph.set_known_prefixes(known_prefixes if known_prefixes else None)

        render_report: FactsRenderReport = await call_llm_with_retry(
            llm_tool=llm_tool,
            prompt=prompt,
            parser=parser,
            prompt_kwargs={
                "format_instructions": profile.format_instructions(
                    FactsRenderReport,
                    web_search_enabled=web_search_enabled,
                ),
                **prompt_data,
            },
            llm_graph_format=state.llm_graph_format,
        )
        persist_search_request(
            state,
            WorkflowNode.TEXT_TO_FACTS,
            render_report.external_evidence_request,
            web_search_enabled,
        )
        render_report.semantic_graph.sanitize_prefixes_namespaces()
        clean_graph, rejected = finalize_llm_graph(render_report.semantic_graph)
        state.content_unit.graph = clean_graph
        state.quarantined_literal_triples = rejected
        if rejected:
            logger.warning(
                "Fresh facts quarantined %d triple(s) with invalid typed literals",
                len(rejected),
            )

        # Track triples in budget tracker (fresh facts)
        num_triples = len(clean_graph)
        logger.info(f"Fresh facts generated with {num_triples} triple(s).")
        state.budget_tracker.add_facts_update(num_operations=1, num_triples=num_triples)

        state.clear_failure()
        state.set_node_status(WorkflowNode.TEXT_TO_FACTS, Status.SUCCESS)
        return state

    except Exception as e:
        return _handle_rendering_error(state, e, FailureStage.GENERATE_TTL_FOR_FACTS)
    finally:
        # Clear the context after parsing
        RDFGraph.set_known_prefixes(None)

render_ontology_fresh(state, tools, supplemental_ontologies=None) async

Render ontology triples into a human-readable format.

This function takes the triples from the current ontology and renders them into a more accessible format, making the ontological knowledge easier to understand.

Parameters:

Name	Type	Description	Default
state	UnitOntologyState	The current agent state containing the ontology to render.	required
tools	AtomicToolBox	The toolbox instance providing utility functions.	required

Returns:

Name	Type	Description
AgentState	UnitOntologyState	Updated state with rendered triples.

Source code in ontocast/agent/render_ontology.py

async def render_ontology_fresh(
    state: UnitOntologyState,
    tools: AtomicToolBox,
    supplemental_ontologies: Sequence[Ontology] | None = None,
) -> UnitOntologyState:
    """Render ontology triples into a human-readable format.

    This function takes the triples from the current ontology and renders them
    into a more accessible format, making the ontological knowledge easier to
    understand.

    Args:
        state: The current agent state containing the ontology to render.
        tools: The toolbox instance providing utility functions.

    Returns:
        AgentState: Updated state with rendered triples.
    """

    profile = get_graph_format_profile(state.llm_graph_format)
    parser = PydanticOutputParser(pydantic_object=OntologyRenderReport)
    logger.info("Rendering fresh ontology")
    intro_instruction = intro_instruction_fresh.format(
        current_domain=state.current_domain
    )
    output_instruction = profile.render_fresh_output_instruction(target="ontology")
    ontology_ttl = ""
    improvement_instruction_str = ""
    access = ontology_access_for_unit_ontology(state)
    web_search_enabled = tools.web_grounding_enabled_for_node(
        WorkflowNode.TEXT_TO_ONTOLOGY
    )
    (
        general_ontology_instruction_str,
        text_chapter,
        external_evidence,
    ) = _prepare_ontology_common_prompt_layers(
        state,
        access,
        search_guidelines=search_guidelines_for(
            WorkflowNode.TEXT_TO_ONTOLOGY, web_search_enabled
        ),
    )

    prompt = _create_ontology_render_prompt_template()
    known_prefixes = build_llm_prefix_map(
        access.ontology_for_prefixes(),
        supplemental_ontologies or (),
    )

    try:
        RDFGraph.set_known_prefixes(known_prefixes if known_prefixes else None)
        llm_tool = await tools.get_llm_tool(state.budget_tracker)
        render_report: OntologyRenderReport = await call_llm_with_retry(
            llm_tool=llm_tool,
            prompt=prompt,
            parser=parser,
            prompt_kwargs={
                "preamble": system_preamble,
                "intro_instruction": intro_instruction,
                "ontology_instruction": general_ontology_instruction_str,
                "output_instruction": output_instruction,
                "ontology_ttl": ontology_ttl,
                "user_instruction": state.ontology_user_instruction,
                "improvement_instruction": improvement_instruction_str,
                "text": text_chapter,
                "external_evidence": external_evidence,
                "format_instructions": profile.format_instructions(
                    OntologyRenderReport,
                    web_search_enabled=web_search_enabled,
                ),
            },
            llm_graph_format=state.llm_graph_format,
        )
        persist_search_request(
            state,
            WorkflowNode.TEXT_TO_ONTOLOGY,
            render_report.external_evidence_request,
            web_search_enabled,
        )
        state.current_ontology = render_report.ontology
        state.current_ontology.graph.sanitize_prefixes_namespaces()

        num_triples = len(state.current_ontology.graph)
        logger.info(f"New ontology created with {num_triples} triple(s).")

        # Track triples in budget tracker (fresh ontology)
        state.budget_tracker.add_ontology_update(
            num_operations=1, num_triples=num_triples
        )

        state.clear_failure()
        state.set_node_status(WorkflowNode.TEXT_TO_ONTOLOGY, Status.SUCCESS)
        return state

    except Exception as e:
        return _handle_ontology_render_error(
            state, e, FailureStage.GENERATE_TTL_FOR_ONTOLOGY
        )
    finally:
        RDFGraph.set_known_prefixes(None)

summarize_chunk(unit, tools, *, max_sentences) async

Compress a content unit for downstream extraction.

Source code in ontocast/agent/summarize_chunks.py

async def summarize_chunk(
    unit: ContentUnit,
    tools: ToolBox,
    *,
    max_sentences: int,
) -> str:
    """Compress a content unit for downstream extraction."""
    section_label = unit.section_label or "unclassified"
    prompt = _SUMMARIZE_PROMPT.format_prompt(
        max_sentences=max_sentences,
        section_label=section_label,
        text=unit.text,
    )
    response = await tools.llm(prompt)
    summary = (response.content or "").strip()
    if not summary:
        raise ValueError("Summarization returned empty text")
    logger.debug(
        "Summarized unit %s (%s): %s -> %s chars",
        unit.index,
        section_label,
        len(unit.text),
        len(summary),
    )
    return summary