133 lines
4.8 KiB
Python
133 lines
4.8 KiB
Python
"""Enhanced ontology criticism agent with SPARQL operations.
|
|
|
|
This module provides enhanced functionality for analyzing and validating ontologies of previous critiques and SPARQL operation support.
|
|
"""
|
|
|
|
import logging
|
|
|
|
from langchain_core.output_parsers import PydanticOutputParser
|
|
from langchain_core.prompts import PromptTemplate
|
|
|
|
from ontocast.agent.common import call_llm_with_retry
|
|
from ontocast.onto.enum import FailureStage, Status, WorkflowNode
|
|
from ontocast.onto.model import OntologyCritiqueReport, Suggestions
|
|
from ontocast.onto.unit_states import UnitOntologyState
|
|
from ontocast.prompt.common import ontology_template, text_template
|
|
from ontocast.prompt.common import system_preamble_ontology as system_preamble
|
|
from ontocast.prompt.criticise_ontology import (
|
|
intro_instruction,
|
|
ontology_criteria,
|
|
template_prompt,
|
|
)
|
|
from ontocast.tool import LLMTool
|
|
from ontocast.tool.atomic import AtomicToolBox
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def criticise_ontology(
|
|
state: UnitOntologyState, tools: AtomicToolBox
|
|
) -> UnitOntologyState:
|
|
"""Enhanced ontology criticism with SPARQL operations.
|
|
|
|
This function performs a critical analysis of the ontology in the current
|
|
state, with SPARQL operation support.
|
|
|
|
Args:
|
|
state: The current unit ontology state containing the ontology to analyze.
|
|
tools: The toolbox instance providing utility functions.
|
|
|
|
Returns:
|
|
UnitOntologyState: Updated state with analysis results.
|
|
"""
|
|
|
|
progress_info = state.get_content_unit_progress_string()
|
|
logger.info(
|
|
f"Ontology Critic for {progress_info}: visit {state.node_visits[WorkflowNode.CRITICISE_ONTOLOGY]}/{state.max_visits_per_node}"
|
|
)
|
|
|
|
if state.content_unit is None:
|
|
state.status = Status.FAILED
|
|
return state
|
|
|
|
current = state.current_ontology or state.ontology_snapshot
|
|
if current.is_null():
|
|
raise ValueError(
|
|
f"Null ontology cannot be criticised: {current.iri} is not a valid ontology"
|
|
)
|
|
|
|
parser = PydanticOutputParser(pydantic_object=OntologyCritiqueReport)
|
|
llm_tool: LLMTool = await tools.get_llm_tool(state.budget_tracker)
|
|
|
|
ontology_ttl = current.graph.serialize(format="turtle")
|
|
|
|
ontology_chapter = ontology_template.format(
|
|
ontology_ttl=ontology_ttl,
|
|
)
|
|
|
|
text_chapter = text_template.format(text=state.content_unit.text)
|
|
|
|
user_instruction = state.ontology_user_instruction
|
|
external_evidence = state.external_evidence_text
|
|
if external_evidence:
|
|
state.mark_external_evidence_used(WorkflowNode.CRITICISE_ONTOLOGY)
|
|
|
|
prompt = PromptTemplate(
|
|
template=template_prompt,
|
|
input_variables=[
|
|
"preamble",
|
|
"intro_instruction",
|
|
"ontology_criteria",
|
|
"user_instruction",
|
|
"ontology_chapter",
|
|
"text_chapter",
|
|
"external_evidence",
|
|
"format_instructions",
|
|
],
|
|
)
|
|
|
|
try:
|
|
critique: OntologyCritiqueReport = await call_llm_with_retry(
|
|
llm_tool=llm_tool,
|
|
prompt=prompt,
|
|
parser=parser,
|
|
prompt_kwargs={
|
|
"preamble": system_preamble,
|
|
"intro_instruction": intro_instruction,
|
|
"ontology_criteria": ontology_criteria,
|
|
"text_chapter": text_chapter,
|
|
"user_instruction": user_instruction,
|
|
"ontology_chapter": ontology_chapter,
|
|
"external_evidence": external_evidence,
|
|
"format_instructions": parser.get_format_instructions(),
|
|
},
|
|
)
|
|
state.set_external_evidence_request(
|
|
WorkflowNode.CRITICISE_ONTOLOGY, critique.external_evidence_request
|
|
)
|
|
logger.info(
|
|
f"Parsed critique report - success: {critique.success}, "
|
|
f"score: {critique.score}, n fixes: {len(critique.actionable_ontology_fixes)}."
|
|
)
|
|
|
|
if critique.success or critique.score > 90:
|
|
state.status = Status.SUCCESS
|
|
state.set_node_status(WorkflowNode.CRITICISE_ONTOLOGY, Status.SUCCESS)
|
|
logger.info("Ontology critique passed")
|
|
else:
|
|
state.status = Status.FAILED
|
|
state.failure_stage = FailureStage.ONTOLOGY_CRITIQUE
|
|
state.set_node_status(WorkflowNode.CRITICISE_ONTOLOGY, Status.FAILED)
|
|
state.suggestions = Suggestions.from_critique_report(critique)
|
|
state.failure_reason = "Ontology Critic suggests improvements"
|
|
logger.info(
|
|
f"Ontology critique failed: {critique.systemic_critique_summary}"
|
|
)
|
|
return state
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to critique ontology: {str(e)}")
|
|
state.set_failure(FailureStage.ONTOLOGY_CRITIQUE, str(e))
|
|
state.set_node_status(WorkflowNode.CRITICISE_ONTOLOGY, Status.FAILED)
|
|
return state
|