ontocast.agent.render_facts

Fact rendering agent for OntoCast.

This module provides functionality for rendering facts from RDF graphs into human-readable formats, making the extracted knowledge more accessible and understandable.

render_facts(state, tools, supplemental_ontologies=None) async

Structured hybrid facts renderer with Turtle/SPARQL decision logic.

This function decides between generating bare Turtle for fresh facts and SPARQL operations for updates based on whether facts exist.

Parameters:

Name	Type	Description	Default
state	UnitFactsState	The current unit facts state	required
tools	AtomicToolBox	The toolbox containing necessary tools	required

Returns:

Name	Type	Description
UnitFactsState	UnitFactsState	Updated state with rendered facts

Source code in ontocast/agent/render_facts.py

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

    This function decides between generating bare Turtle for fresh facts
    and SPARQL operations for updates based on whether facts exist.

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

    Returns:
        UnitFactsState: Updated state with rendered facts
    """

    is_fresh_facts_graph = len(state.content_unit.graph) == 0

    progress_info = state.get_content_unit_progress_string()
    logger.info(f"Render facts for {progress_info}")

    extras = list(supplemental_ontologies or ())
    if is_fresh_facts_graph:
        logger.info("Generating fresh facts as Turtle")
        return await render_facts_fresh(state, tools, supplemental_ontologies=extras)
    else:
        logger.info("Generating facts update")
        return await render_facts_update(state, tools, supplemental_ontologies=extras)

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 (),
    )

    prompt_data = _prepare_prompt_data(state, access, profile)
    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),
                **prompt_data,
            },
            llm_graph_format=state.llm_graph_format,
        )
        state.set_external_evidence_request(
            WorkflowNode.TEXT_TO_FACTS, render_report.external_evidence_request
        )
        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_facts_update(state, tools, supplemental_ontologies=None) async

Render facts updates using SPARQL operations.

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_update(
    state: UnitFactsState,
    tools: AtomicToolBox,
    supplemental_ontologies: Sequence[Ontology] | None = None,
) -> UnitFactsState:
    """Render facts updates using SPARQL operations.

    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 updates for 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=GraphUpdateRenderReport)

    access = ontology_access_for_unit_facts(state)
    prompt_data = _prepare_prompt_data(state, access, profile)
    prompt_data_update = {
        "preamble": preamble,
        "improvement_instruction": render_suggestions_prompt(
            state.suggestions, WorkflowNode.TEXT_TO_FACTS
        ),
        "output_instruction": profile.render_update_output_instruction(),
        "fact_chapter": profile.format_facts_chapter(state.content_unit.graph),
    }
    prompt_data.update(prompt_data_update)
    prompt = _create_prompt_template()
    known_prefixes = build_llm_prefix_map(
        access.ontology_for_prefixes(),
        supplemental_ontologies or (),
    )

    try:
        # Set known prefixes in context before parsing
        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={
                "format_instructions": profile.format_instructions(
                    GraphUpdateRenderReport
                ),
                **prompt_data,
            },
            llm_graph_format=state.llm_graph_format,
        )
        state.set_external_evidence_request(
            WorkflowNode.TEXT_TO_FACTS, render_report.external_evidence_request
        )
        graph_update = render_report.graph_update
        all_rejected = []
        for op in graph_update.triple_operations:
            clean_graph, rejected = finalize_llm_graph(op.graph)
            op.graph = clean_graph
            all_rejected.extend(rejected)
        state.quarantined_literal_triples = all_rejected
        if all_rejected:
            logger.warning(
                "Facts update quarantined %d triple(s) with invalid typed literals",
                len(all_rejected),
            )
        state.facts_updates.append(graph_update)
        state.update_facts()

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

        # Track triples in budget tracker
        state.budget_tracker.add_facts_update(num_operations, num_triples)

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

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