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) 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	AgentState	The current agent state	required
tools	ToolBox	The toolbox containing necessary tools	required

Returns:

Name	Type	Description
AgentState	AgentState	Updated state with rendered facts

Source code in ontocast/agent/render_facts.py

async def render_facts(state: AgentState, tools: ToolBox) -> AgentState:
    """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 agent state
        tools: The toolbox containing necessary tools

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

    is_first_visit = len(state.current_chunk.graph) == 0

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

    if is_first_visit:
        logger.info("Generating fresh facts as Turtle")
        return await render_facts_fresh(state, tools)
    else:
        logger.info("Generating facts update")
        return await render_facts_update(state, tools)

render_facts_fresh(state, tools) async

Render fresh facts from the current chunk into Turtle format.

Parameters:

Name	Type	Description	Default
state	AgentState	The current agent state containing the chunk to render.	required
tools	ToolBox	The toolbox instance providing utility functions.	required

Returns:

Name	Type	Description
AgentState	AgentState	Updated state with rendered facts.

Source code in ontocast/agent/render_facts.py

async def render_facts_fresh(state: AgentState, tools: ToolBox) -> AgentState:
    """Render fresh facts from the current chunk into Turtle format.

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

    Returns:
        AgentState: Updated state with rendered facts.
    """
    logger.info("Rendering fresh facts")
    llm_tool = tools.llm
    parser = PydanticOutputParser(pydantic_object=SemanticTriplesFactsReport)

    # Extract prefixes from the ontology graph to help with parsing LLM responses
    # that may use these prefixes without declaring them
    known_prefixes = {}
    if state.current_ontology and state.current_ontology.graph:
        for prefix, namespace_uri in state.current_ontology.graph.namespaces():
            if prefix:  # Skip empty prefixes
                known_prefixes[prefix] = str(namespace_uri)

    # Also add the ontology prefix explicitly if we know it
    if state.current_ontology.prefix and state.current_ontology.namespace:
        known_prefixes[state.current_ontology.prefix] = state.current_ontology.namespace

    prompt_data = _prepare_prompt_data(state)
    prompt_data_fresh = {
        "preamble": preamble,
        "improvement_instruction": "",
        "output_instruction": output_instruction_empty,
    }
    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)

        proj = await call_llm_with_retry(
            llm_tool=llm_tool,
            prompt=prompt,
            parser=parser,
            prompt_kwargs={
                "format_instructions": parser.get_format_instructions(),
                **prompt_data,
            },
        )
        proj.semantic_graph.sanitize_prefixes_namespaces()
        state.current_chunk.graph = proj.semantic_graph

        # Track triples in budget tracker (fresh facts)
        num_triples = len(proj.semantic_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()
        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) async

Render facts updates using SPARQL operations.

Parameters:

Name	Type	Description	Default
state	AgentState	The current agent state containing the chunk to render.	required
tools	ToolBox	The toolbox instance providing utility functions.	required

Returns:

Name	Type	Description
AgentState	AgentState	Updated state with rendered facts.

Source code in ontocast/agent/render_facts.py

async def render_facts_update(state: AgentState, tools: ToolBox) -> AgentState:
    """Render facts updates using SPARQL operations.

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

    Returns:
        AgentState: Updated state with rendered facts.
    """
    logger.info("Rendering updates for facts")
    llm_tool = tools.llm
    parser = PydanticOutputParser(pydantic_object=GraphUpdate)

    prompt_data = _prepare_prompt_data(state)
    prompt_data_update = {
        "preamble": preamble,
        "improvement_instruction": render_suggestions_prompt(
            state.suggestions, WorkflowNode.TEXT_TO_FACTS
        ),
        "output_instruction": output_instruction_sparql,
        "fact_chapter": facts_template.format(
            facts_ttl=state.current_chunk.graph.serialize(format="turtle")
        ),
    }
    prompt_data.update(prompt_data_update)
    prompt = _create_prompt_template()

    try:
        graph_update = await call_llm_with_retry(
            llm_tool=llm_tool,
            prompt=prompt,
            parser=parser,
            prompt_kwargs={
                "format_instructions": parser.get_format_instructions(),
                **prompt_data,
            },
        )
        state.facts_updates.append(graph_update)

        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
        )