"""CellReader Protocol — read named cells off a table image and say what they print. A table printed as an image is read twice: once by the parse, once by the page scan. A cell the scan read is put to this agent with the table's image when the parse read something different there or read no cell there at all. A cell only the parse read, blank in the scan, is never asked about. The agent answers what the page prints, cell by cell. It is given both prior readings. They bracket the answer. A cell one reader gave as `$ 3.000.00` and the other as `S 3,000,00` prints `$ 3,000.00`, which is neither — each side has something the other lost. An agent reading the crop with nothing to check itself against could return `$ 4,010.00` and it would look exactly like a correct reading. It is asked about a whole table at once rather than a cell at a time. A number read out of a bare cell crop has lost its column header and its row label; read in place it sits among values of the same magnitude, where a separator in the wrong place is obvious. Cells are named by the address printed beside them in the grid it is shown, the same convention the structure correction uses, so the agent reads an address off a label rather than counting rows and columns. A failed call changes nothing. Every cell keeps the reading it arrived with, which is the scan's, and the failure is logged. """ from __future__ import annotations import json from pathlib import Path from typing import Any, Dict, List, Literal, Optional, Protocol, 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 CellQuestion(BaseModel): """One cell to read, and what has already been read there.""" address: str = Field(description="The cell's address in the grid, e.g. B3") scan_read: str = Field(description="What the page scan read in this cell") parse_read: Optional[str] = Field( default=None, description="What the parse read in this cell; absent when it read no cell there" ) class CellValue(BaseModel): address: str = Field(description="The address this value belongs to, copied from the question") value: str = Field(description="Exactly what the page prints in that cell, verbatim") class CellValues(BaseModel): cells: List[CellValue] = Field(default_factory=list, description="One entry per address asked about") CELL_READER_PROMPT = f"""\ You are looking at a cropped image of one table from a financial document, and a rendering of that table where every cell carries its address in square brackets. Two tools have read this table. They disagree about some cells, and for other cells only one of them produced a reading at all. You will be given a list of those cells by address, with what each tool read. For each address in the list, report exactly what the page prints in that cell. Rules: - Report the printed characters verbatim: currency symbols, thousands separators, decimal points, parentheses around negatives, percent signs, footnote markers, and minus signs or dashes, exactly as typeset. - The two prior readings are evidence, not options. Where both are wrong, report what the page prints even though it matches neither. Where the cell is legible and one of them is right, report that one. - A common failure in these readings is a thousands separator returned as a decimal point, so `19.543.903` where the page prints `19,543,903`. Look closely at separators and report what is printed. - If the page prints nothing in the cell, return an empty string for it. - Read only the cells you are asked about. Do not report any other cell, do not correct the table's structure, and never include an address tag in a value. - Return one entry per address you were given, using that same address. Return a JSON object conforming exactly to this schema: {json.dumps(CellValues.model_json_schema(), indent=2)} Return only the JSON object — no prose, no markdown code fences. """ @runtime_checkable class CellReader(Protocol): async def read_cells( self, table_image: bytes, addressed_grid: str, questions: List[CellQuestion] ) -> CellValues: ... def question_block(questions: List[CellQuestion]) -> str: """The cells to read, one per line, with the readings already made.""" lines = [] for q in questions: seen = f"scan read {q.scan_read!r}" seen += ( f", parse read {q.parse_read!r}" if q.parse_read is not None else ", the parse read no cell here" ) lines.append(f"- {q.address}: {seen}") return "\n".join(lines) class PydanticAICellReader: """CellReader backed by pydantic-ai with an Anthropic model. Auth resolution mirrors the other agents: `ANTHROPIC_AUTH_TOKEN` (OAuth subscription) preferred over `ANTHROPIC_API_KEY`. """ # The value written into a financial document should not vary between runs # on the same image. Greedy decoding removes the sampling variance. DEFAULT_TEMPERATURE = 0.0 def __init__( self, model: Optional[str] = None, auth_token: Optional[str] = None, api_key: Optional[str] = None, temperature: Optional[float] = DEFAULT_TEMPERATURE, ) -> 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 if temperature is not None and not supports_sampling_temperature(model): temperature = None 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( "PydanticAICellReader: neither ANTHROPIC_AUTH_TOKEN nor " "ANTHROPIC_API_KEY is set. Provide one via env or constructor." ) self.model = model self.temperature = temperature model_settings = None if temperature is None else ModelSettings(temperature=temperature) self.agent = Agent( anth_model, output_type=CellValues, system_prompt=CELL_READER_PROMPT, model_settings=model_settings, ) # LangSmith tracer is a no-op if `TRACE_TO_LANGSMITH` is unset. self.tracer = LangSmithTracer(run_name="quber-cell-reader") def trace_inputs(self, addressed_grid: str, questions: List[CellQuestion]) -> Dict[str, Any]: return { "messages": [ {"role": "system", "content": CELL_READER_PROMPT}, { "role": "user", "content": [ {"type": "text", "text": addressed_grid}, {"type": "text", "text": question_block(questions)}, {"type": "image", "text": "[table crop attached]"}, ], }, ] } async def read_cells( self, table_image: bytes, addressed_grid: str, questions: List[CellQuestion] ) -> CellValues: from pydantic_ai import BinaryContent image = BinaryContent(data=table_image, media_type="image/png") asked = question_block(questions) inputs = self.trace_inputs(addressed_grid, questions) try: async with self.tracer.llm_run("cell_reader", inputs, model=self.model) as run: result = await self.agent.run([addressed_grid, asked, image]) output = result.output run.outputs = { "messages": [{"role": "assistant", "content": output.model_dump_json()}], "cells_read": len(output.cells), "usage_metadata": usage_metadata_from(result.usage), } return output except Exception as exc: logger.error( "cell_reader: LLM call failed; {} disputed cell(s) keep the scan's reading exc={}", len(questions), exc, ) return CellValues(cells=[]) class MockCellReader: """Returns canned values by address. For tests.""" def __init__(self, values: Optional[Dict[str, str]] = None) -> None: self.values = values or {} self.calls: List[List[CellQuestion]] = [] async def read_cells( self, table_image: bytes, addressed_grid: str, questions: List[CellQuestion] ) -> CellValues: _ = table_image, addressed_grid self.calls.append(list(questions)) return CellValues( cells=[ CellValue(address=q.address, value=self.values[q.address]) for q in questions if q.address in self.values ] ) CellReaderBackend = Literal["api", "mock", "off"] def get_cell_reader(backend: Optional[CellReaderBackend] = None) -> Optional[CellReader]: """The cell reader for this run, or None when the check is switched off.""" selected = backend or get_settings().llm.cell_reader_backend if selected == "api": return PydanticAICellReader() if selected == "mock": return MockCellReader() if selected == "off": return None raise ValueError(f"Unknown QUBER_CELL_READER_BACKEND: {selected!r}. Expected api|mock|off.") def read_table_image(path: Path) -> bytes: """The bytes of a rendered table crop, for a caller holding a file.""" return Path(path).read_bytes()