""" StatusInspector — verifies a proposed gap condition against the table image. The grounding stage classifies every corrected cell that lacks a measured box by deterministic text evidence: printed-but-unlocated (header or label row), one-character, an authorized conventional label, or unverifiable. All but the one-character code are HYPOTHESES — the text test cannot see the page. The one-character code is definitional and is never inspected. Whether an unlocated header really is a printed band, or an added 'Total' really sits on a totals row whose label position is blank, is visible only on the image. This agent looks. It gets the cropped table image, the corrected markdown, and the flagged cells each with its proposed condition spelled out, and for each cell answers whether the image supports that condition, with one line of evidence. The agent only judges. The deterministic gate, `inspect_gap_cells` in `core/extractors/set_of_mark/inspection.py`, holds authority and applies the three-way verdict. `holds` keeps the proposed status and records the evidence. `contradicted` moves the cell to `defect`. `cannot_tell`, a missing finding, or a failed call moves it to `unverified`, the user-inspection class. No verdict upgrades a status, so a wrong answer can only send a cell to a human, never away from one. """ from __future__ import annotations import json from typing import Any, Dict, List, Literal, Optional, Protocol, Sequence, runtime_checkable from loguru import logger from pydantic import BaseModel, Field from quber.agents.factory import supports_sampling_temperature from quber.agents.langsmith_tracer import LangSmithTracer, usage_metadata_from from quber.settings import DEFAULT_LLM_MODEL, get_settings class CellFinding(BaseModel): """The inspector's finding for one flagged cell. Field order is deliberate: `evidence` comes before `holds` so the model describes what the image shows BEFORE committing to a verdict — observation first, categorization second. """ row: int = Field(ge=0, description="The flagged cell's row, copied from the input list.") col: int = Field(ge=0, description="The flagged cell's column, copied from the input list.") evidence: str = Field( description=( "FIRST, before judging: one short sentence describing what the image " "actually shows at that position — what is printed there, or that " "nothing is." ) ) verdict: Literal["holds", "contradicted", "cannot_tell"] = Field( description=( "THEN the verdict, from what you just described. 'holds': the image " "positively shows the stated condition. 'contradicted': the image " "positively shows something ELSE — a different word printed at the " "position, or a structure the condition misdescribes. 'cannot_tell': " "the image does not settle it either way. Never force the nearest " "option — when none fits, say cannot_tell." ) ) class InspectionReport(BaseModel): """One finding per flagged cell, in the order given.""" findings: List[CellFinding] = Field(description="A finding for EVERY flagged cell.") INSPECT_PROMPT = """\ You verify conditions that a table extraction has proposed for cells it could not anchor to the page, by LOOKING at the table image. You get: 1. A cropped image of one table — the only authority. 2. The extraction's corrected markdown for that table (row 0 is the header row; the separator line does not count). 3. A list of flagged cells: row, col, the cell's text, and the CONDITION the extraction proposed for it, spelled out in words. For EVERY flagged cell, look FIRST and judge SECOND: describe in one sentence what the image actually shows at that cell's position (`evidence`), and only then give a three-way verdict on the stated condition — `holds` when your observation positively supports it, `contradicted` when the image positively shows something else (a different printed word, a structure the condition misdescribes), `cannot_tell` when the image settles nothing. Never force the nearest option; when none fits, say cannot_tell. Condition guidance: - A condition claiming the text is PRINTED in a header or band position holds only if you can see that text printed there (it may wrap across lines or share a line with a neighboring column's text). - A condition claiming the text is a PRINTED row label holds only if you can see it printed at that row (typically wrapped across two lines). - A condition claiming the extraction ADDED a conventional label is judged differently: for this condition the text is EXPECTED to be absent from the page — its absence is what the condition asserts, so "the word is not printed there" is evidence FOR the condition, never against it. It holds when the position is blank on the page AND the convention fits what the image shows — 'Total' on a row that visibly totals its section, generic column names on a table printed with no header row. It is contradicted when the page DOES print something at that position (a different word the extraction should have used), or the visible structure contradicts the convention (the row is an average, not a total). - A condition claiming the text has NO KNOWN PRINTED SOURCE holds when the page prints nothing at the position — absence is expected there, exactly as for an added conventional label. It is contradicted ONLY when the page prints a DIFFERENT word at that position that the extraction should have used instead. Data values sitting where a header row was inserted do not count as a different word. And if you can see the EXACT same text printed at the position, answer cannot_tell and say where — the extraction is right and only the record's sourcing is open, which is not a defect. Be strict: `holds` only on positive visual evidence, `contradicted` only on positive contrary evidence. Give one finding per flagged cell, no more, no fewer. """ @runtime_checkable class StatusInspector(Protocol): """Verifies proposed gap conditions against the table image.""" async def inspect( self, crop_png: bytes, markdown: str, flagged: Sequence[Dict[str, Any]], page: int ) -> Optional[InspectionReport]: ... class PydanticAIStatusInspector: """StatusInspector backed by pydantic-ai with an Anthropic vision model. Mirrors `PydanticAICaptureAdvisor`: same auth resolution (`ANTHROPIC_AUTH_TOKEN` preferred over `ANTHROPIC_API_KEY`), greedy decoding so the same table yields the same findings run to run. A failed call returns None — the caller keeps the proposed statuses unconfirmed, which the gate downgrades to `unverified`; inspection must never take a table down with it. """ DEFAULT_TEMPERATURE = 0.0 def __init__( self, model: Optional[str] = None, auth_token: Optional[str] = None, api_key: Optional[str] = None, ) -> None: from pydantic_ai import Agent from pydantic_ai.models.anthropic import AnthropicModel from pydantic_ai.providers.anthropic import AnthropicProvider from pydantic_ai.settings import ModelSettings llm_settings = get_settings().llm model = model or llm_settings.model or DEFAULT_LLM_MODEL resolved_auth = auth_token or llm_settings.anthropic_auth_token resolved_key = api_key or llm_settings.anthropic_api_key if resolved_auth: from quber.agents._oauth_gate import make_oauth_anthropic_model anth_model = make_oauth_anthropic_model(model, resolved_auth) elif resolved_key: provider = AnthropicProvider(api_key=resolved_key) anth_model = AnthropicModel(model, provider=provider) else: raise RuntimeError( "PydanticAIStatusInspector: neither ANTHROPIC_AUTH_TOKEN nor " "ANTHROPIC_API_KEY is set. Provide one via env or constructor." ) self.model = model self.agent = Agent( anth_model, output_type=InspectionReport, system_prompt=INSPECT_PROMPT, model_settings=ModelSettings(temperature=self.DEFAULT_TEMPERATURE) if supports_sampling_temperature(model) else None, ) self.tracer = LangSmithTracer(run_name="quber-status-inspector") def trace_inputs(self, user_text: str) -> Dict[str, Any]: return { "messages": [ {"role": "system", "content": INSPECT_PROMPT}, {"role": "user", "content": [{"type": "text", "text": user_text}]}, ] } async def inspect( self, crop_png: bytes, markdown: str, flagged: Sequence[Dict[str, Any]], page: int ) -> Optional[InspectionReport]: from pydantic_ai import BinaryContent user_text = ( f"=== Corrected markdown ===\n{markdown}\n\n" "=== Flagged cells (verify each against the image) ===\n" + json.dumps(list(flagged), indent=1) ) inputs = self.trace_inputs(user_text) try: async with self.tracer.llm_run("inspect_status", inputs, model=self.model) as run: result = await self.agent.run( [user_text, BinaryContent(data=crop_png, media_type="image/png")] ) report: InspectionReport = result.output run.outputs = { "messages": [{"role": "assistant", "content": report.model_dump_json()}], "usage_metadata": usage_metadata_from(result.usage), } return report except Exception as exc: logger.warning( "page {}: the status inspection call failed; the table's flagged cells " "are recorded as unverified in the review flags: {}", page, exc, ) return None class MockStatusInspector: """Returns canned findings and records what it was asked. For tests.""" def __init__(self, findings: Optional[List[CellFinding]] = None) -> None: self.findings = findings or [] self.calls: List[Dict[str, Any]] = [] async def inspect( self, crop_png: bytes, markdown: str, flagged: Sequence[Dict[str, Any]], page: int ) -> Optional[InspectionReport]: _ = crop_png self.calls.append({"markdown": markdown, "flagged": list(flagged), "page": page}) return InspectionReport(findings=self.findings) def get_status_inspector(backend: Optional[str] = None) -> Optional[StatusInspector]: """Resolve the status inspector, or None when disabled (`off`).""" selected = backend or get_settings().llm.status_inspector_backend if selected == "api": return PydanticAIStatusInspector() if selected == "mock": return MockStatusInspector() if selected == "off": return None raise ValueError(f"Unknown QUBER_STATUS_INSPECTOR_BACKEND: {selected!r}. Expected api|mock|off.")