async def render_ontology_update(
state: UnitOntologyState,
tools: AtomicToolBox,
supplemental_ontologies: Sequence[Ontology] | None = None,
) -> UnitOntologyState:
"""Complement an existing snapshot via GraphUpdate inserts."""
state.quarantined_literal_triples = []
profile = get_graph_format_profile(state.llm_graph_format)
parser = PydanticOutputParser(pydantic_object=GraphUpdateRenderReport)
access = ontology_access_for_unit_ontology(state)
intro_instruction = _build_update_intro(state, access)
ontology_chapter = profile.format_ontology_chapter(
access.effective_graph_for_prompt(),
max_triples=state.ontology_context_max_triples,
)
output_instruction = profile.render_update_output_instruction()
improvement_instruction_str = render_suggestions_prompt(
state.suggestions, WorkflowNode.TEXT_TO_ONTOLOGY
)
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_graph_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,
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,
)
# No insert_hook: the facts repairs are instance-level (literal retyping,
# unit-code resolution) and have no ontology counterpart. The ontology
# side's deterministic validator runs in the loop instead, against the
# net delta -- see stategraph/atomic.py::_collect_ontology_findings.
graph_update, rejected = finalize_update_report(render_report)
state.quarantined_literal_triples = rejected
log_quarantine("Ontology", rejected)
state.ontology_updates.append(graph_update)
applied = state.update_ontology()
if not applied:
# The ONTOLOGY_MAX_TRIPLES backstop discarded the whole update, so
# this billed render changed nothing. Returning SUCCESS with an
# unchanged graph is not a lie the run can see, and a validator run
# afterwards would inspect the *previous* graph and report it clean.
# The status stays SUCCESS -- the pre-update graph is intact and a
# re-render would hit the same ceiling -- but the discard is counted.
logger.warning(
"Ontology update discarded: applying it would exceed "
"ONTOLOGY_MAX_TRIPLES=%s. The render was billed and had no "
"effect on the working graph.",
state.ontology_max_triples,
)
state.budget_tracker.incr("ontology/update_rejected_over_budget")
# Suggestions are consumed by exactly the render they were raised
# against. Leaving them set carried them into every later render of the
# unit -- the leak that put two contradictory contracts in one facts
# prompt (see CHANGELOG [Unreleased]); the ontology path had the same
# defect and no repair pass to notice it.
state.suggestions = Suggestions()
# Findings were consumed by this render; the loop re-collects fresh.
state.deterministic_findings = []
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)."
)
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_GRAPH_UPDATE_FOR_ONTOLOGY
)
finally:
RDFGraph.set_known_prefixes(None)