ontocast.agent.render_ontology

Ontology triple rendering agent for OntoCast.

This module provides functionality for rendering RDF triples from ontologies into human-readable formats, making the ontological knowledge more accessible and understandable. The agent decides between generating bare Turtle for fresh ontologies and SPARQL operations for updates.

render_ontology(state, tools, supplemental_ontologies=None) async

Structured hybrid ontology renderer with Turtle/SPARQL decision logic.

This function decides between generating bare Turtle for fresh ontologies and SPARQL operations for updates based on whether the ontology exists.

Parameters:

Name	Type	Description	Default
state	UnitOntologyState	The current unit ontology state	required
tools	AtomicToolBox	The toolbox containing necessary tools	required

Returns:

Name	Type	Description
UnitOntologyState	UnitOntologyState	Updated state with rendered ontology

Source code in ontocast/agent/render_ontology.py

async def render_ontology(
    state: UnitOntologyState,
    tools: AtomicToolBox,
    supplemental_ontologies: Sequence[Ontology] | None = None,
) -> UnitOntologyState:
    """Structured hybrid ontology renderer with Turtle/SPARQL decision logic.

    This function decides between generating bare Turtle for fresh ontologies
    and SPARQL operations for updates based on whether the ontology exists.

    Args:
        state: The current unit ontology state
        tools: The toolbox containing necessary tools

    Returns:
        UnitOntologyState: Updated state with rendered ontology
    """

    progress_info = state.get_content_unit_progress_string()
    logger.info(
        f"Ontology Renderer for {progress_info}: visit {state.node_visits[WorkflowNode.TEXT_TO_ONTOLOGY]}/{state.max_visits_per_node}"
    )
    access = ontology_access_for_unit_ontology(state)
    current = access.effective_ontology_for_prompt()
    # Guardrail for map/reduce flow: if a non-null snapshot exists, stay in update mode.
    has_seed_ontology = access.has_non_null_seed_snapshot()
    has_no_seed_ontology = current.is_null() and not has_seed_ontology

    extras = list(supplemental_ontologies or ())
    if has_no_seed_ontology:
        return await render_ontology_fresh(state, tools, supplemental_ontologies=extras)
    else:
        return await render_ontology_update(
            state, tools, supplemental_ontologies=extras
        )

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)
    (
        general_ontology_instruction_str,
        text_chapter,
        external_evidence,
    ) = _prepare_ontology_common_prompt_layers(state, access)

    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
                ),
            },
            llm_graph_format=state.llm_graph_format,
        )
        state.set_external_evidence_request(
            WorkflowNode.TEXT_TO_ONTOLOGY, render_report.external_evidence_request
        )
        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)

render_ontology_update(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 unit ontology state containing the ontology to render.	required
tools	AtomicToolBox	The toolbox instance providing utility functions.	required

Returns:

Name	Type	Description
UnitOntologyState	UnitOntologyState	Updated state with rendered triples.

Source code in ontocast/agent/render_ontology.py

async def render_ontology_update(
    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 unit ontology state containing the ontology to render.
        tools: The toolbox instance providing utility functions.

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

    profile = get_graph_format_profile(state.llm_graph_format)
    parser = PydanticOutputParser(pydantic_object=GraphUpdateRenderReport)
    access = ontology_access_for_unit_ontology(state)
    current = access.effective_ontology_for_prompt()
    ontology_iri = current.iri
    ontology_desc = current.describe()
    multi_source_note = ""
    if state.ontology_patch_sources:
        joined_sources = ", ".join(state.ontology_patch_sources[:10])
        multi_source_note = (
            "\nThe provided ontology context may combine patches from multiple source "
            f"ontologies: {joined_sources}. Preserve existing IRIs, namespace boundaries, "
            "and avoid collapsing distinct source namespaces unless explicitly justified."
        )
    intro_instruction = intro_instruction_update.format(
        ontology_iri=ontology_iri,
        ontology_desc=ontology_desc,
        multi_source_note=multi_source_note,
    )
    ontology_chapter = profile.format_ontology_chapter(current.graph)
    output_instruction = profile.render_update_output_instruction()
    improvement_instruction_str = render_suggestions_prompt(
        state.suggestions, WorkflowNode.TEXT_TO_ONTOLOGY
    )

    (
        general_ontology_instruction_str,
        text_chapter,
        external_evidence,
    ) = _prepare_ontology_common_prompt_layers(state, access)

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

    try:
        llm_tool = await tools.get_llm_tool(state.budget_tracker)
        RDFGraph.set_known_prefixes(known_prefixes if known_prefixes else None)

        render_report: GraphUpdateRenderReport = 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,
                "improvement_instruction": improvement_instruction_str,
                "ontology_ttl": ontology_chapter,
                "user_instruction": state.ontology_user_instruction,
                "text": text_chapter,
                "external_evidence": external_evidence,
                "format_instructions": profile.format_instructions(
                    GraphUpdateRenderReport
                ),
            },
            llm_graph_format=state.llm_graph_format,
        )
        state.set_external_evidence_request(
            WorkflowNode.TEXT_TO_ONTOLOGY, render_report.external_evidence_request
        )
        graph_update = render_report.graph_update
        state.ontology_updates.append(graph_update)
        state.update_ontology()

        num_operations, num_triples = graph_update.count_total_triples()
        logger.info(
            f"Ontology update has {num_operations} operation(s) "
            f"with {num_triples} total triple(s)."
        )

        # Track triples in budget tracker
        state.budget_tracker.add_ontology_update(num_operations, 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_SPARQL_UPDATE_FOR_ONTOLOGY
        )
    finally:
        # Clear the context after parsing
        RDFGraph.set_known_prefixes(None)