async def criticise_facts(
state: UnitFactsState, tools: AtomicToolBox
) -> UnitFactsState:
"""Critically analyze facts in the current content unit.
Args:
state: The current unit facts state containing the chunk to analyze.
tools: The toolbox instance providing utility functions.
Returns:
UnitFactsState: Updated state with analysis results.
"""
if not state.content_unit:
logger.warning("No current content unit to analyze")
return state
progress_info = state.get_content_unit_progress_string()
logger.info(
f"Facts critic for {progress_info}: visit {state.node_visits[WorkflowNode.CRITICISE_FACTS]}/{state.max_visits_per_node}"
)
llm_tool = await tools.get_llm_tool(state.budget_tracker)
profile = get_graph_format_profile(state.llm_graph_format)
parser = PydanticOutputParser(pydantic_object=FactsCritiqueReport)
ctx = ontology_access_for_unit_facts(state).effective_ontology_for_prompt()
# Same chapter the renderer gets, index appendix included. Building it
# without the suffix left the critic reading opaque IRIs while guideline 6a
# told the renderer to resolve them through the TERM INDEX -- so the critic
# judged term choices it could not read. Also memoised on the shared
# snapshot, so this stops re-serialising the ontology on every visit.
ontology_chapter = ctx.prompt_chapter(
profile, max_triples=state.ontology_context_max_triples
)
facts_chapter = profile.format_facts_chapter(
state.content_unit.graph
) + _build_quarantine_chapter(state)
text_chapter = text_template.format(text=state.content_unit.extraction_text)
user_instruction = (
user_template.format(user_instruction=state.facts_user_instruction)
if state.facts_user_instruction
else ""
)
prompt = PromptTemplate(
template=template_prompt,
input_variables=[
"preamble",
"evaluation_instruction",
"user_instruction",
"ontology_chapter",
"facts_chapter",
"text_chapter",
"graph_format_instruction",
"format_instructions",
],
)
graph_format_instruction = profile.critique_graph_instruction()
web_search_enabled = tools.web_grounding_enabled_for_node(
WorkflowNode.CRITICISE_FACTS
)
search_guidelines = search_guidelines_for(
WorkflowNode.CRITICISE_FACTS, web_search_enabled
)
evaluation_instruction_str = evaluation_instruction
if search_guidelines:
evaluation_instruction_str = f"{evaluation_instruction}\n\n{search_guidelines}"
prompt_data = {
"preamble": preamble,
"evaluation_instruction": evaluation_instruction_str,
"user_instruction": user_instruction,
"ontology_chapter": ontology_chapter,
"facts_chapter": facts_chapter,
"text_chapter": text_chapter,
"graph_format_instruction": graph_format_instruction,
"format_instructions": profile.format_instructions(
FactsCritiqueReport,
web_search_enabled=web_search_enabled,
),
}
try:
critique: FactsCritiqueReport = await call_llm_with_retry(
llm_tool=llm_tool,
prompt=prompt,
parser=parser,
prompt_kwargs=prompt_data,
llm_graph_format=state.llm_graph_format,
)
persist_search_request(
state,
WorkflowNode.CRITICISE_FACTS,
critique.external_evidence_request,
web_search_enabled,
)
logger.debug(
f"Parsed critique report - success: {critique.success}, "
f"score: {critique.score}"
)
# Acceptance is decided from defects that can be pointed at: the
# deterministic findings already collected against this graph, plus the
# critic's own fixes at the configured severity. `score` and `success`
# are recorded and no longer consulted -- see acceptance.py for what the
# score gate measured and why it could not be calibrated.
defects = material_defects(
state.deterministic_findings,
critique.actionable_triple_fixes,
tools.acceptance_policy,
)
reason = accept_reason(defects)
state.attempt_log.append(
LoopAttempt(
render_attempt=state.node_visits[WorkflowNode.TEXT_TO_FACTS],
critic_attempt=state.node_visits[WorkflowNode.CRITICISE_FACTS],
kind="critic",
score=critique.score,
success=not defects,
accept_reason=reason,
n_actionable_fixes=len(critique.actionable_triple_fixes),
severity_counts=Counter(
fix.severity for fix in critique.actionable_triple_fixes
),
n_deterministic_findings=len(state.deterministic_findings),
n_mandatory_findings=sum(
1 for finding in state.deterministic_findings if finding.mandatory
),
triple_count=len(state.content_unit.graph),
)
)
if not defects:
state.status = Status.SUCCESS
state.set_node_status(WorkflowNode.CRITICISE_FACTS, Status.SUCCESS)
# An accepting critic has no outstanding requests. Clearing here is
# not redundant with the reset in render_facts_update: the loop can
# accept on a *later* critic attempt of the same render (after an
# external-evidence search), with no render in between to consume
# the suggestions the earlier, rejecting attempt left behind. The
# finding-driven repair then runs next, and must see only findings.
state.suggestions = Suggestions()
logger.info(
"Facts critique passed (score %s, no material defect)",
critique.score,
)
else:
state.status = Status.FAILED
state.set_node_status(WorkflowNode.CRITICISE_FACTS, Status.FAILED)
state.failure_stage = FailureStage.FACTS_CRITIQUE
state.suggestions = Suggestions.from_critique_report(critique)
state.failure_reason = f"Facts unit has {len(defects)} material defect(s)"
logger.info(
"Facts critique rejected on %s: %s (score %s)",
reason,
"; ".join(defect.message for defect in defects[:3]),
critique.score,
)
return state
except Exception as e:
logger.error(f"Failed to criticize facts: {str(e)}")
state.set_failure(FailureStage.FACTS_CRITIQUE, str(e))
state.set_node_status(WorkflowNode.CRITICISE_FACTS, Status.FAILED)
return state