""" LLMClient Protocol — the structured-extraction seam for table structure and document heading review. The Protocol declares four calls. Each is async, with a `_sync` sibling for callers without an event loop: - `count_tables(image_path)` — returns the integer count of text-based tables visible in an image. The validation hook (`core.validate`) uses it to surface Camelot-vs-LLM mismatches. The Set-of-Mark split probe (`core/extractors/set_of_mark/split.py`) counts the tables inside a cropped region to decide whether to split it. - `correct_structure(image_path, markdown)` — structural correction step of the Camelot+LLM pipeline (`core/extractors/camelot/llm/pipeline.py`). Takes a Camelot-extracted markdown table and the page image, returns a corrected `LLMTableCorrection`. The LLM is **forbidden from changing numeric values** — that contract is enforced by the system prompt; Camelot remains the source of truth for cell contents. - `vet_structure(image_png, markdown, region_text, page_text)` — grounded structure vetting. The cropped table image is the structural arbiter, and Camelot and the text layer are the value source. Called through `core/extractors/camelot/correspondence/correction.py`, which both the correspondence and the Set-of-Mark extractors use. - `review_headings(nominated)` — judges which repeated document headings are page decoration. Called by `core/fusion/heading_review.py`. Backend selection is via `QUBER_LLM_BACKEND` env var (or `get_llm_client(backend=...)`): - `api` — `PydanticAIClient`. Uses `pydantic-ai` with an Anthropic model. The default. It authenticates with the OAuth subscription token when one is set and with the API key otherwise. - `cli` — `ClaudeCLIClient`. Shells out to the `claude` binary; uses the developer's subscription quota. Used only when selected. - `mock` — `MockLLMClient`. Returns canned responses; for tests. `agents/factory.py` (multi-provider pydantic-ai) is a separate pattern. It builds the agents of the processors in `quber/processors/`, such as table inference with its executive summary. """ from __future__ import annotations import asyncio import json import re import shutil import subprocess from pathlib import Path from typing import Any, Dict, List, Literal, Optional, Protocol, runtime_checkable from loguru import logger from pydantic import BaseModel, Field, ValidationError, field_validator, model_validator from quber.agents.factory import supports_sampling_temperature from quber.agents.langsmith_tracer import usage_metadata_from from quber.settings import DEFAULT_LLM_MODEL, get_settings class CellMerge(BaseModel): """One corrected cell built from one or more source (Camelot) cells — a rejoined split symbol, a flattened multi-row header, a replicated band or spanning label, or a label extended with its footnote marker. Reported by the correction agent so the cell's geometry can be recovered from its source cells, and addressed by its exact position in the corrected table (result text alone is ambiguous when two cells share the same text, e.g. an identical total in the Basic and Diluted EPS rows). """ result: str = Field(description="The combined cell text, exactly as in your corrected markdown.") row: int = Field( ge=0, description=( "0-based row of this cell in YOUR corrected markdown table: the header row is row 0, " "the first body row is row 1." ), ) col: int = Field( ge=0, description="0-based column of this cell in your corrected markdown table, counting from the left.", ) sources: List[str] = Field( description="The Camelot cells you joined to make it, each copied exactly from the Camelot markdown, in reading order." ) source_cells: List[str] = Field( default_factory=list, description=( "The address of each Camelot cell you joined, copied from the tag printed next to " "that cell's text in the Camelot markdown ('[B3] 1,637' -> 'B3'). Copy the printed " "tag; never count positions. Same order as `sources`." ), ) @field_validator("sources", "source_cells") @classmethod def drop_blank_sources(cls, v: List[str]) -> List[str]: """A blank piece names no printed mark, so there is nothing to ground. The prompt asks the agent not to list blank cells, but compliance is not exact; normalizing here makes the record deterministic either way. """ return [s for s in v if s.strip()] # A marker symbol opening a note line: the asterisk family, daggers, # section/pilcrow signs, or '#', followed by whitespace and the note text. # Alphanumeric or parenthesized markers never match — only the bare symbols # that read as prose punctuation when folded into the text. _LEADING_SYMBOL_RE = re.compile(r"^([*†‡§¶#]{1,2})\s+(\S.*)$", re.DOTALL) class FootnoteDef(BaseModel): """One footnote printed below the table: its own marker plus its text. The marker is read off the image, where the footnote line prints it — formatting the image shows (a superscript, a parenthesized digit) decides what is a marker, which downstream text parsing cannot see. An unmarked general note (a basis-of-presentation line with no marker) has an empty marker. """ marker: str = Field( default="", description=( "The footnote's OWN marker exactly as printed at the start of its line: " "'1', '(1)', a letter 'a', an asterisk '*', a dagger. Empty string for an " "unmarked general note that opens with no marker." ), ) text: str = Field(description="The footnote's text, without the leading marker.") @model_validator(mode="after") def lift_leading_symbol(self) -> "FootnoteDef": """A symbol marker folded into the text lifts into the marker field. The prompt asks for the marker and text as separate fields, but compliance is not exact: a symbol legend line ('# Denotes a variance of 100 percent or more') sometimes arrives with an empty marker and the symbol leading the text. The split is unambiguous — a note's text never opens with a bare marker symbol followed by a space — so it normalizes here, the same way stacked marker groups split. """ if not self.marker: m = _LEADING_SYMBOL_RE.match(self.text) if m: self.marker = m.group(1) self.text = m.group(2) return self class FootnoteMark(BaseModel): """One footnote reference marker observed on a table cell, identified by QUOTING the carrying cell — the model never counts positions; it repeats printed text, and a lookup resolves it, the same contract as `CellMerge`. """ marker: str = Field( description=( "ONE reference marker exactly as printed on the cell: '1', '(1)', a letter " "'a', an asterisk '*', a dagger. One entry per cell that carries it. A cell " "printed with SEVERAL stacked markers — '(b)(c)(d)' — carries three markers: " "report three entries quoting the same cell, one per marker, never the " "concatenated string." ) ) cell_text: str = Field( description=( "The CARRYING cell's text, copied exactly from YOUR corrected markdown — " "the whole cell, not just the marker ('Segment EBITDA(3)', '90+ days past " "billing as a % of total(e) - corporate'). Copy the cell; never count or " "describe positions." ) ) kind: Literal["footnote", "section"] = Field( default="footnote", description=( "What the marker points at, judged from the PAGE, not from the marker's " "wording. 'footnote': it points at a footnote printed for this table — a " "note line below the table or nearby that defines it. 'section': it is a " "cross-reference to a named part of the document (a numbered note in the " "financial statements, a schedule, an appendix) whose content lives " "elsewhere, not in a footnote line printed for this table." ), ) # A concatenation of parenthesized marker groups printed as one stack: # '(b)(c)(d)'. Strictly this shape only — each group short and parenthesized — # so a parenthesized negative value can never match. COMPOUND_MARKER_RE = re.compile(r"^(?:\([A-Za-z0-9]{1,3}\)){2,}$") MARKER_GROUP_RE = re.compile(r"\([A-Za-z0-9]{1,3}\)") class LLMTableCorrection(BaseModel): # Field descriptions carry the per-field semantics. Pydantic-AI exposes # them to the LLM via the tool-input schema on the api backend. The CLI # backend sees them through the JSON Schema block that # `ClaudeCLIClient.correct_structure_sync` appends at runtime. # `VET_STRUCTURE_PROMPT` and the CLI backend's appended field guidance # restate some of them in their own words. A change here does not update # those restatements. caption: str = Field( default="", description=( "The sentence that introduces this table, copied exactly as printed from the " "page text, including any bold lead-in words it starts with. Empty if the " "table has none." ), ) title: str = Field( default="", description=( "The table's printed name, copied exactly as printed. A sentence is a caption, " "not a name, and the section heading is not the table's name. Empty if the " "table has no printed name. Never compose one." ), ) markdown: str = Field( description=( "Corrected markdown table. Preserve Camelot's numeric cell " "values exactly; only fix structure (merge multi-row " "headers, drop header-fragment rows, fix column alignment). " "Reproduce cell text verbatim — never add Markdown styling: no " "** (bold) or * (italic) wrappers, and no character that is not in " "the source. A stray * reads downstream as a footnote-reference " "marker, so adding one corrupts footnote handling." ), ) footnotes: List[FootnoteDef] = Field( default_factory=list, description=( "Lines printed below the table that qualify the table (a marker on a header or " "cell is a reference, listed in footnote_refs, never an entry here; no printed " "footnote lines means an empty list): footnote lines that open with " "a printed marker, and short unmarked qualifications that apply to the table as " "a whole (a basis-of-presentation line, 'unaudited', a scale line). One entry " "per line, as a pair of the line's own printed marker (empty for an unmarked " "qualification) and its text without that marker. Narrative prose printed " "below the table is not a footnote and never goes here; it goes in body_text." ), ) body_text: List[str] = Field( default_factory=list, description=( "Narrative prose visible in the image below the table that is neither a " "footnote nor a table qualification: paragraphs discussing the business, " "sentences introducing the next table, any running text of the document. " "One entry per paragraph, its opening words only (the first ten or so). The " "document already holds this text; listing it records that it was seen and " "left out of the table. Empty list if none." ), ) units: str = Field( default="", description=( "Scale/currency unit attribution caption for the table " '(e.g. "(in millions, except percentages and per share data)"). ' "This is the units line, not a footnote and not a data row. " "Exclude it from the markdown body and from footnotes. " "Empty string if the table has no such caption." ), ) header_rows: int = Field( ge=0, description=( "How many leading rows of your corrected markdown are column-header rows, " "counted off the IMAGE. A single header line is 1. A stacked header — a " "spanning period or date band over a row of column labels — counts every " "stacked row ('March 31, 2026' over 'UPB | Carrying Value' is 2). 0 when " "the image shows data starting at the first row: a continuation fragment " "with no column headers of its own. Count rows of YOUR corrected markdown, " "after any flattening you did." ), ) footnote_refs: List[str] = Field( default_factory=list, description=( "Every distinct FOOTNOTE REFERENCE MARKER shown on the table's headers or " "cells, read off the IMAGE (a superscript can be lost in the text). A marker " "is a superscript or parenthetical number, letter, or symbol that points to a " "footnote: e.g. '1', '(1)', a letter 'a', an asterisk '*', a dagger '†' or " "'‡', or a section sign '§'. Write each as it appears. A parenthesized NEGATIVE " "VALUE like '(84)' or a unit like '(%)'/'(£m)' is NOT a marker. A cross-reference " "to a NAMED SECTION of the document — '(Note 16)', 'Schedule II', '(Addendum 3)' " "— is written as printed WITH its word ('Note 16', 'Addendum 3'), never as a " "bare number. Empty list if the table carries no footnote reference markers." ), ) footnote_marks: List[FootnoteMark] = Field( default_factory=list, description=( "WHERE each footnote reference marker sits: one entry per cell that carries a " "marker, addressed by that cell's row/col in your corrected markdown (header " "row is row 0), with the cell's printed Camelot address when it has one. A " "marker printed on several cells gets one entry per carrying cell. Every " "marker listed in footnote_refs should appear here unless you cannot tell " "which cell carries it. Empty list if the table carries no markers." ), ) @field_validator("footnote_refs", mode="after") @classmethod def split_compound_refs(cls, v: List[str]) -> List[str]: """A stacked group reported as one string splits into its markers. The prompt asks for one entry per marker, but compliance is not exact: '(b)(c)(d)' still arrives as one string on some tables. A strict run of short parenthesized groups splits unambiguously — the same compliance normalization as `CellMerge.drop_blank_sources`. """ out: List[str] = [] for marker in v: s = marker.strip() if COMPOUND_MARKER_RE.fullmatch(s): out.extend(MARKER_GROUP_RE.findall(s)) else: out.append(marker) seen: set[str] = set() return [m for m in out if not (m in seen or seen.add(m))] @field_validator("footnote_marks", mode="after") @classmethod def split_compound_marks(cls, v: List[FootnoteMark]) -> List[FootnoteMark]: """One mark carrying a stacked group becomes one mark per marker at the same cell — same normalization as `split_compound_refs`.""" out: List[FootnoteMark] = [] for mark in v: s = mark.marker.strip() if COMPOUND_MARKER_RE.fullmatch(s): out.extend(mark.model_copy(update={"marker": g}) for g in MARKER_GROUP_RE.findall(s)) else: out.append(mark) return out cell_merges: List[CellMerge] = Field( default_factory=list, description=( "Every output cell whose text you did not copy verbatim from a single Camelot " "cell serving the same role: joins ('$' + '666'), flattened header stacks, a " "spanning/band label replicated into several output cells (one entry PER output " "cell, all citing the same source), a row label extended with its footnote " "marker, a label moved to a new position. A single source cell is fine. Give the " "result and the exact source cells, verbatim from the Camelot markdown. List only " "source cells that contain text. Empty if every cell is a verbatim single-cell copy." ), ) class HeadingVerdict(BaseModel): """One nominated heading judged by the document heading review.""" text: str = Field(description="The heading text, copied exactly from the nominated list.") verdict: Literal["section_heading", "running_header", "column_label", "other_non_heading"] = Field( description=( "section_heading: a real heading that starts a section and governs the text " "below it — keep. running_header: a banner the page layout repeats at the top " "of its pages. column_label: a table column label that leaked into the heading " "stream. other_non_heading: repeated text that governs nothing." ) ) reason: str = Field( description="One line: what about the text and its repetition pattern decided the verdict." ) class LLMHeadingReview(BaseModel): """The heading-review agent's answer: one verdict per nominated heading.""" verdicts: List[HeadingVerdict] = Field( description="One verdict per nominated heading, in the order given." ) REVIEW_HEADINGS_PROMPT = """\ You are reviewing the section headings of one parsed document. The parser labeled certain printed lines as section headings. A real section heading is printed once, where its section starts, and governs the text beneath it. Page decoration is not a heading: a running header the layout repeats at the top of many pages governs nothing, and neither does a table column label that leaked into the heading stream. Repetition alone does not make decoration. An author also repeats headings: a sub-heading like "Year 2018 results:" recurs in every section it structures, because each section discusses the same period. That is authored repetition and it stays a `section_heading`. But a text printed as a heading more than once on the same page is never an authored section heading — no author starts the same section twice on one page; that pattern is a table column label or other decoration. You are given only the suspicious headings: each text below appears as a heading on three or more pages, with its page list and where on the page it is printed. Position is the strongest evidence. Layout decoration is stamped at a fixed position at the very top of its pages. An authored repeat sits lower, at varying positions, wherever its section happens to start. Decide what each one is: a heading kept as `section_heading` stays a heading, and any other verdict demotes it to page decoration. When unsure, keep it. A wrongly demoted real heading silently strips section context from every paragraph beneath it, which is far worse than keeping a false one. Return one verdict per heading, in the order given. """ # Role + the one constraint not captured by per-field descriptions. The # schema itself is injected by pydantic-ai's tool-use machinery on the # api backend, and appended at runtime by `ClaudeCLIClient` on # the cli backend (which has no schema-injection channel). CORRECT_STRUCTURE_PROMPT = """\ You are correcting the *structure* of a table extracted from a PDF. You are given two artifacts: 1. A rendered image of the PDF page containing the table. 2. A markdown table extracted by Camelot from the same page. Hard constraints — these are non-negotiable: - **DO NOT change any numeric values** from the Camelot markdown. Camelot is the source of truth for cell contents. - **DO NOT read values from the page image**. The image is reference only: it shows you the layout so you can fix *structure* — merge multi-row headers, drop header-fragment rows, fix column alignment. Never fill cells, add rows, or invent values from what you see in the image. If a value is missing from the Camelot markdown, leave the cell empty — do not reconstruct it from the image. - If the Camelot markdown is empty or has no real data, return it unchanged (or an empty `markdown`). Do not fabricate a table. - Reproduce each cell's text verbatim — never add Markdown styling. Do not wrap cell text in ** (bold) or * (italic), and add no character that is not in the source. You may fix arrangement (column/header placement), never the characters. A stray * is read downstream as a footnote-reference marker, so adding one corrupts footnote handling. `title`, `caption`, and `footnotes` may still be drawn from the image — those are heading/caption metadata, not table cell values. """ VET_STRUCTURE_PROMPT = """\ You vet and fix the STRUCTURE of one table extracted from a PDF, using a cropped image of that table as the arbiter of structure and the provided text as the source of values. You are given four artifacts, all for ONE table: 1. A cropped image of just this table — the ARBITER of column and header structure (how many columns, where they split, multi-row headers). 2. The Camelot markdown — correct values, but the column structure may be wrong (e.g. a spurious empty column that splits a real column, or a header misaligned from its values). Every non-empty cell carries a printed address tag: `[B3] 1,637` means this cell is B3. The tags are REFERENCE ONLY — they let you name a cell by copying the tag printed next to its text. Never copy a tag into your corrected table; cell text is everything after the tag. 3. The table's text layer — the authoritative source for any value. 4. The full text of the page, in reading order — the source for `caption` and `title`. The table's own text layer can cut a line in half where two tables sit side by side; the page text has the whole line. Produce: - `markdown`: the corrected table. Make the column/header structure match the IMAGE — drop spurious empty columns, realign values under the correct headers, merge multi-row / fragmented headers into clear labels. If the image shows rows the Camelot markdown is missing, recover them from the TEXT LAYER. - `caption`: the sentence that introduces this table, copied exactly as printed from the page text, including any bold lead-in words it starts with. Empty if the table has none. - `title`: the table's printed name, copied exactly as printed. A sentence is a caption, not a name, and the section heading is not the table's name. Empty if no name is printed. Never compose, shorten or paraphrase either field. - `footnotes`: the lines printed below the table that QUALIFY the table, read from the image; one entry per line as a PAIR: the line's OWN marker exactly as printed at its start ('1', '(1)', 'a', '*', a dagger), and its text without that marker. The image's formatting decides what is a marker. A short unmarked line that qualifies the whole table (a basis-of-presentation line, "unaudited", a scale line — even one starting with a number, like a year) gets an empty marker. Never put a footnote in the markdown body. Decide for every line below the table what it IS: table data (markdown), a footnote or table qualification (footnotes), or the document's own running text (body_text). A paragraph that discusses the business, states a figure in prose, or introduces the next table is running text even when it sits directly under the footnotes and even when the crop includes it in full. It is NOT a footnote and does not go in `footnotes`. A `footnotes` entry is a LINE PRINTED BELOW THE TABLE. A marker shown on a header or a cell is a reference to a footnote, listed in `footnote_refs`; it is never itself an entry, and an entry is never invented for it. If no footnote lines are printed in the crop, `footnotes` is an empty list, even when headers or cells carry markers. - `body_text`: the running text you excluded, one entry per paragraph, its opening words only. Empty list if the crop shows none. - `units`: the scale/currency attribution caption that applies to the whole table (e.g. "(in millions, except percentages and per share data)" or "($ in thousands)"). Put it here and nowhere else — never in the markdown body and never in `footnotes`. Empty string if the table has no such caption. - `header_rows`: how many leading rows of your corrected markdown are column-header rows, counted off the IMAGE. A single header line is 1. A stacked header — a spanning period or date band over a row of column labels — counts every stacked row ("March 31, 2026" over "UPB | Carrying Value" is 2). A fragment whose first row is already data is 0. Count rows of YOUR corrected markdown, after any flattening you did. - `footnote_refs`: every distinct footnote reference marker shown on a header or cell — a superscript or parenthetical number, letter, or symbol that points to a footnote (e.g. a superscript 1, "(1)", a letter "a", "*", a dagger). A stacked group printed on one cell — "(b)(c)(d)" — is SEVERAL markers, listed separately as "(b)", "(c)", "(d)", never as one concatenated string. Read these off the IMAGE, since a superscript may be absent from the text. A parenthesized negative value like "(84)" or a unit like "(%)" is not a marker. A cross-reference to a NAMED SECTION of the document — "(Note 16)" after a row label, "Schedule II", "(Addendum 3)" — is written as printed WITH its word ("Note 16", "Addendum 3"), never as a bare number: a bare "16" claims a footnote printed below the table, which is not what the page shows. Empty if none. - `footnote_marks`: WHERE each marker sits — one entry per cell that carries a reference marker, read off the IMAGE. Give the marker as printed and the carrying cell's text COPIED EXACTLY from your corrected markdown — the whole cell, never a position or a description. A marker printed on several cells gets one entry per carrying cell: a marker on a column header, a row label, and a data cell are all reported the same way. Every marker in `footnote_refs` should appear here unless you genuinely cannot tell which cell carries it. For each entry, JUDGE what the marker points at and set `kind`: 'footnote' when it points at a note printed for this table, 'section' when it is a cross-reference to a named part of the document whose content lives elsewhere. Judge from what the page shows — is there a footnote line printed for it, or does it name a separate part of the document? - `cell_merges`: report EVERY output cell whose text you did not copy verbatim from a single Camelot cell serving the same role — every cell you joined, flattened, replicated, carried, or extended. That includes, each as a normal case: * a split symbol rejoined with its number ('$' + '666'); * stacked header rows flattened into one label ('Three Months Ended' + 'June 30, 2024'); * a spanning label, section band, or period band you placed into MORE THAN ONE output cell ('GAAP Results' printed once but serving three columns): report EACH output cell as its own entry, all citing the same source; * a row label you extended with its footnote marker from the text layer: cite the label's own cell; * a label you moved or carried to a different position. Header and band rows ANYWHERE in the table count — a period band in the middle of the table is reported exactly like the top header row. A SINGLE source cell is fine — `sources`/`source_cells` may hold one element. `result` is the COMBINED CELL'S OWN TEXT exactly as it appears in your corrected markdown — never the row label or any neighboring cell — and `row`/`col` are THAT cell's position (row 0 is the header row, columns count from 0 on the left). `sources` are the source cells' texts copied verbatim (without the address tag); `source_cells` are their printed addresses (e.g. ['B2', 'B3']) — COPY the tag printed next to that cell's text; never count positions, and never put an address in `sources` or text in `source_cells`. List only source cells that contain text — a blank cell you absorbed has no printed mark, so do not list it. If none of the cells you built a cell from contain text, do not report that cell at all. Camelot emits a FLAT grid with no notion of merged or spanning cells, so the following structural errors are EXPECTED on real tables — treat each as a normal case to fix, not an exception: - SYMBOL SPLIT INTO ITS OWN COLUMN: a currency symbol or sign ($, %, parentheses) that Camelot placed in a separate column belongs WITH the number beside it. Rejoin them into one cell (a "$" column followed by a "666" column becomes the single cell "$ 666"). Never leave a symbol as a standalone column. - MULTI-LEVEL / SPANNING COLUMN HEADER: headers stacked in two or more rows where a top label spans several columns. Flatten them into ONE header row in which every column's label is its full top-to-bottom path — EVERY stacked level from the topmost group label down to the column's own label, none skipped. Specifically: * Combine each column's group label with its own period/date/year, e.g. group "Three Months Ended" + "June 30, 2024" -> "Three Months Ended June 30, 2024". * There may be SEVERAL span groups of UNEQUAL width on the same row (e.g. "Three Months Ended" over three columns AND "Six Months Ended" over two); assign each column to the group it physically sits under. * If a column's sub-label is partial — only a year, or a date left blank because it is shared with the column beside it — carry the missing piece from that column's group or sibling so EVERY column ends up complete (a year-only "2025" under the six-months group becomes "Six Months Ended June 30, 2025"). Use only date/period text that actually appears on the page; never invent a period. * A group label printed ABOVE the span groups is a LEVEL of every column under it — carry it too. "Coverage Data" printed over "Before Management Fees" and "After Management Fees" makes the flattened labels "Coverage Data Before Management Fees" and "Coverage Data After Management Fees". NEVER drop a level of the stack. * Emit exactly ONE header row; every column carries a complete label; no blank header cells and no leftover span-only row. - UNLABELED TOTAL ROW / MISSING HEADER ROW: a row that totals its section but prints no label may be labeled "Total"; a table printed with no header row may receive generic column names ("Item", "Description"). Use ONLY these neutral conventional words — never a specific guess like "Average" or "Net". A label you add this way has no printed source, so never report it in `cell_merges`. - INTERLEAVED EMPTY / SPACER COLUMNS: stream over-segmentation inserts blank columns between real ones. Drop them and bring the real columns together. - HEADER FRAGMENTED ACROSS ROWS: merge the fragments into the single header row described above. After fixing, the table must be RECTANGULAR: one header row, every body row with the same number of columns, and no stray blank columns. Hard constraints — non-negotiable: - Every value in `markdown` must come from the Camelot markdown or the text layer. NEVER read a number off the image, and NEVER invent, compute, or estimate one. - Preserve every Camelot value; only move it to the correct column (rejoining a split symbol to its number is moving, not inventing). - Reproduce each cell's text verbatim — never add Markdown styling. Do not wrap text in ** (bold) or * (italic), and add no character that is not in the source. You fix a cell's ARRANGEMENT — which column or header it sits in — never its CHARACTERS. A stray * is read downstream as a footnote-reference marker, so adding one corrupts footnote handling. - A cell that is BLANK on the page stays blank in your output. Never write a placeholder for an empty cell: no em-dash, hyphen, 'N/A', zero, or any other stand-in. Emit a dash only if that exact dash is printed on the page and present in the Camelot markdown or text layer. """ COUNT_TABLES_PROMPT = """\ You are auditing a page image for table coverage. Return ONLY a single integer: the number of text-based tables visible on the page. Do not count figures, charts, or images. Return just the number. """ @runtime_checkable class LLMClient(Protocol): # Async is the default path; sync siblings live under `_sync` suffix # for callers without an event loop. async def count_tables(self, image_path: Path) -> int: ... async def correct_structure(self, image_path: Path, markdown: str) -> Optional[LLMTableCorrection]: ... async def vet_structure( self, image_png: bytes, markdown: str, region_text: str, page_text: str ) -> Optional[LLMTableCorrection]: ... async def review_headings(self, nominated: str) -> Optional[LLMHeadingReview]: ... def count_tables_sync(self, image_path: Path) -> int: ... def correct_structure_sync(self, image_path: Path, markdown: str) -> Optional[LLMTableCorrection]: ... def vet_structure_sync( self, image_png: bytes, markdown: str, region_text: str, page_text: str ) -> Optional[LLMTableCorrection]: ... def review_headings_sync(self, nominated: str) -> Optional[LLMHeadingReview]: ... class ClaudeCLIClient: """Shells out to `claude -p` for subscription-backed usage. Zero API token cost; depends on the `claude` binary being on PATH.""" def __init__(self, binary: str = "claude", timeout: int = 180) -> None: self.binary = binary self.timeout = timeout def run(self, prompt: str, image_path: Path) -> str: if shutil.which(self.binary) is None: raise RuntimeError( f"`{self.binary}` not found on PATH. Install Claude Code or set " f"QUBER_LLM_BACKEND=api to use the API client." ) cmd = [ self.binary, "-p", prompt + f"\n\nImage: {image_path}", "--output-format", "text", ] result = subprocess.run( cmd, capture_output=True, text=True, timeout=self.timeout, check=False, ) if result.returncode != 0: raise RuntimeError(f"claude CLI failed (exit {result.returncode}): {result.stderr}") return result.stdout.strip() def count_tables_sync(self, image_path: Path) -> int: raw = self.run(COUNT_TABLES_PROMPT, image_path) for token in raw.split(): if token.isdigit(): return int(token) raise ValueError(f"count_tables: could not parse integer from {raw!r}") @staticmethod def extract_json_object(raw: str) -> Optional[Any]: try: return json.loads(raw) except json.JSONDecodeError: start = raw.find("{") end = raw.rfind("}") if start < 0 or end <= start: return None try: return json.loads(raw[start : end + 1]) except json.JSONDecodeError: return None def correct_structure_sync(self, image_path: Path, markdown: str) -> Optional[LLMTableCorrection]: schema_block = json.dumps(LLMTableCorrection.model_json_schema(), indent=2) # `claude -p` has no schema-injection channel and is sensitive to # implicit guidance. Append the schema plus explicit per-field # instructions; the api backend skips this entirely. prompt = ( CORRECT_STRUCTURE_PROMPT + "\n\nReturn a JSON object conforming exactly to this JSON Schema:\n\n" + schema_block + "\n\nField guidance:\n" + "- `markdown` (REQUIRED): the corrected markdown table; include every row " + "from the Camelot extraction with cell values preserved exactly. Reproduce " + "cell text verbatim — no ** (bold) or * (italic) styling, and no character " + "not in the source; a stray * is read downstream as a footnote-reference " + "marker.\n" + '- `title`: the table title from the page image (e.g. "Branded Volume and ' + 'Transactions"); empty string if no title is visible.\n' + '- `subtitle`: any sub-heading below the title (e.g. "For the 3 Months Ended ' + 'September 30, 2025"); empty string if none.\n' + "- `footnotes`: the lines below the table that qualify the table (marker-led " + "footnotes and short unmarked qualifications such as a basis-of-presentation " + "line); one entry per line, as an object pairing its own printed marker (empty " + "string for an unmarked qualification) with its text minus that marker; empty " + "list if none. The document's running text below the table (a paragraph " + "discussing the business, a sentence introducing the next table) is not a " + "footnote: leave it out of `footnotes` and list its opening words in " + "`body_text`.\n" + "- `body_text`: opening words of each running-text paragraph you excluded; " + "empty list if none.\n" + "- `footnote_refs`: every distinct footnote reference marker shown on a header " + "or cell, read off the image (superscript or parenthetical number/letter/symbol, " + "e.g. '1', '(1)', '*', a dagger); a parenthesized negative value or a unit is not " + "a marker; empty list if none.\n" + "\nReturn only the JSON object — no prose, no markdown code fences." + f"\n\n=== Camelot markdown ===\n{markdown}\n" ) for attempt in range(2): raw = self.run(prompt, image_path) data = self.extract_json_object(raw) if data is None: if attempt == 0: prompt += ( "\n\nYour previous response did not contain a valid JSON object. " "Return a single JSON object conforming to the schema above." ) continue logger.error( "ClaudeCLIClient.correct_structure: no JSON object after retry image={image}", image=image_path.name, ) return None try: return LLMTableCorrection.model_validate(data) except ValidationError as exc: if attempt == 0: prompt += ( f"\n\nYour previous response failed validation:\n{exc}\n" "Return a JSON object that conforms exactly to the schema above." ) continue logger.error( "ClaudeCLIClient.correct_structure: validation failed after retry image={image} exc={exc}", image=image_path.name, exc=exc, ) return None return None async def count_tables(self, image_path: Path) -> int: return await asyncio.to_thread(self.count_tables_sync, image_path) async def correct_structure(self, image_path: Path, markdown: str) -> Optional[LLMTableCorrection]: return await asyncio.to_thread(self.correct_structure_sync, image_path, markdown) def vet_structure_sync( self, image_png: bytes, markdown: str, region_text: str, page_text: str ) -> Optional[LLMTableCorrection]: # The correspondence flow's grounded vetting is api-only; the cli backend # has no clean channel for an inline image crop + region text. raise NotImplementedError("vet_structure requires the api backend (QUBER_LLM_BACKEND=api)") async def vet_structure( self, image_png: bytes, markdown: str, region_text: str, page_text: str ) -> Optional[LLMTableCorrection]: raise NotImplementedError("vet_structure requires the api backend (QUBER_LLM_BACKEND=api)") def review_headings_sync(self, nominated: str) -> Optional[LLMHeadingReview]: # No review means no demotion — every heading is kept, which is the # safe direction. The caller logs that the review was skipped. _ = nominated return None async def review_headings(self, nominated: str) -> Optional[LLMHeadingReview]: _ = nominated return None class PydanticAIClient: """Uses pydantic-ai with an Anthropic model. Auth resolution order: 1. `auth_token` arg -> Anthropic OAuth (subscription) bearer token. 2. `ANTHROPIC_AUTH_TOKEN` env var (OAuth subscription). 3. `api_key` arg -> Anthropic API key. 4. `ANTHROPIC_API_KEY` env var (API key). Any OAuth token wins, so an `api_key` arg is ignored while `ANTHROPIC_AUTH_TOKEN` is set. OAuth is preferred so the same subscription that backs `claude -p` also drives the structured-extraction path. The OAuth path uses a wrapper (see `agents/_oauth_gate.py`) that rewrites the system prompt into the 2-block array format the subscription gate requires. That wrapper is a workaround for a pydantic-ai limitation tracked in `issues/pydantic-ai-multi-block-system-prompt.md`. """ # Greedy decoding for the structural tasks: structure vetting, structure # correction, table counting and heading review are deterministic # judgements, so the same PDF must yield the same answer every run and # across machines. Matches the detector and grid-locator agents. Each agent # gets it only when its model accepts a temperature setting # (`supports_sampling_temperature`). Haiku 4.5 does. The vet agent's default # model, claude-sonnet-5, does not, so vetting runs with no temperature # setting by default. 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 # Precedence: constructor arg > QUBER_LLM_MODEL (or ANTHROPIC_MODEL) # env > `quber.settings.DEFAULT_LLM_MODEL`. 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 not resolved_auth and not resolved_key: raise RuntimeError( "PydanticAIClient: neither ANTHROPIC_AUTH_TOKEN nor " "ANTHROPIC_API_KEY is set. Provide one via env or constructor." ) def build_model(model_id: str): if resolved_auth: from quber.agents._oauth_gate import make_oauth_anthropic_model return make_oauth_anthropic_model(model_id, resolved_auth) provider = AnthropicProvider(api_key=resolved_key) return AnthropicModel(model_id, provider=provider) anth_model = build_model(model) # The vet agent runs on its own (stronger) model, the same per-agent # pattern as the grid locator. `QUBER_VET_MODEL` defaults to # claude-sonnet-5. Only an empty value falls back to the project-wide # model. vet_model = llm_settings.vet_model or model vet_anth_model = anth_model if vet_model == model else build_model(vet_model) self.model = model self.vet_model = vet_model def greedy_settings(model_id: str, **extra) -> Optional[ModelSettings]: if supports_sampling_temperature(model_id): return ModelSettings(temperature=self.DEFAULT_TEMPERATURE, **extra) return ModelSettings(**extra) if extra else None model_settings = greedy_settings(model) self.correct_agent = Agent( anth_model, output_type=LLMTableCorrection, system_prompt=CORRECT_STRUCTURE_PROMPT, model_settings=model_settings, ) self.count_agent = Agent( anth_model, output_type=int, system_prompt=COUNT_TABLES_PROMPT, model_settings=model_settings, ) # Grounded structure vetting — cropped table image is the # structural arbiter, Camelot/text-layer are the value source. self.vet_agent = Agent( vet_anth_model, output_type=LLMTableCorrection, system_prompt=VET_STRUCTURE_PROMPT, model_settings=greedy_settings(vet_model), ) # Document heading review: judges the headings nominated as possible # page decoration. A heading is nominated when its text repeats as a # heading on several pages. Its on-page position is evidence for the # agent, not a nomination criterion. Keep is the default; only its # explicit non-heading verdicts demote. Its verdict list echoes every # nominated heading with a reason, so the response grows with the # nomination count and can overrun the default output-token ceiling. # Give the response explicit room. self.heading_agent = Agent( anth_model, output_type=LLMHeadingReview, system_prompt=REVIEW_HEADINGS_PROMPT, model_settings=greedy_settings(model, max_tokens=16384), ) # LangSmith tracer is a no-op if `TRACE_TO_LANGSMITH` is unset. from quber.agents.langsmith_tracer import LangSmithTracer self.tracer = LangSmithTracer(run_name="quber-llm-client") def binary_content(self, image_path: Path): from pydantic_ai import BinaryContent return BinaryContent(data=image_path.read_bytes(), media_type="image/png") def count_inputs(self) -> Dict[str, Any]: return { "messages": [ {"role": "system", "content": COUNT_TABLES_PROMPT}, { "role": "user", "content": [ {"type": "text", "text": "Count tables in this page."}, ], }, ] } def correct_inputs(self, user_text: str) -> Dict[str, Any]: return { "messages": [ {"role": "system", "content": CORRECT_STRUCTURE_PROMPT}, { "role": "user", "content": [{"type": "text", "text": user_text}], }, ] } def count_tables_sync(self, image_path: Path) -> int: inputs = self.count_inputs() with self.tracer.llm_run_sync("count_tables", inputs, model=self.model) as run: result = self.count_agent.run_sync( ["Count tables in this page.", self.binary_content(image_path)] ) count = int(result.output) run.outputs = { "messages": [{"role": "assistant", "content": str(count)}], "count": count, "usage_metadata": usage_metadata_from(result.usage), } return count def correct_structure_sync(self, image_path: Path, markdown: str) -> Optional[LLMTableCorrection]: user_text = f"Camelot markdown:\n{markdown}\n\nCorrect the structure from the page image." inputs = self.correct_inputs(user_text) with self.tracer.llm_run_sync("correct_structure", inputs, model=self.model) as run: result = self.correct_agent.run_sync([user_text, self.binary_content(image_path)]) output: LLMTableCorrection = result.output run.outputs = { "messages": [{"role": "assistant", "content": output.model_dump_json()}], "title": output.title, "caption": output.caption, "markdown": output.markdown, "footnotes": [f.model_dump() for f in output.footnotes], "body_text": list(output.body_text), "usage_metadata": usage_metadata_from(result.usage), } return output async def count_tables(self, image_path: Path) -> int: inputs = self.count_inputs() async with self.tracer.llm_run("count_tables", inputs, model=self.model) as run: result = await self.count_agent.run( ["Count tables in this page.", self.binary_content(image_path)] ) count = int(result.output) run.outputs = { "messages": [{"role": "assistant", "content": str(count)}], "count": count, "usage_metadata": usage_metadata_from(result.usage), } return count async def correct_structure(self, image_path: Path, markdown: str) -> Optional[LLMTableCorrection]: user_text = f"Camelot markdown:\n{markdown}\n\nCorrect the structure from the page image." inputs = self.correct_inputs(user_text) async with self.tracer.llm_run("correct_structure", inputs, model=self.model) as run: result = await self.correct_agent.run([user_text, self.binary_content(image_path)]) output: LLMTableCorrection = result.output run.outputs = { "messages": [{"role": "assistant", "content": output.model_dump_json()}], "title": output.title, "caption": output.caption, "markdown": output.markdown, "footnotes": [f.model_dump() for f in output.footnotes], "body_text": list(output.body_text), "usage_metadata": usage_metadata_from(result.usage), } return output def binary_png(self, image_png: bytes): from pydantic_ai import BinaryContent return BinaryContent(data=image_png, media_type="image/png") def vet_inputs(self, user_text: str) -> Dict[str, Any]: return { "messages": [ {"role": "system", "content": VET_STRUCTURE_PROMPT}, {"role": "user", "content": [{"type": "text", "text": user_text}]}, ] } def vet_structure_sync( self, image_png: bytes, markdown: str, region_text: str, page_text: str ) -> Optional[LLMTableCorrection]: user_text = ( f"=== Camelot markdown ===\n{markdown}\n\n=== Table text layer ===\n{region_text}\n\n" f"=== Page text ===\n{page_text}\n\n" "Fix the structure against the cropped image; values from Camelot or the text layer only; " "copy `caption` and `title` from the page text." ) inputs = self.vet_inputs(user_text) with self.tracer.llm_run_sync("vet_structure", inputs, model=self.vet_model) as run: result = self.vet_agent.run_sync([user_text, self.binary_png(image_png)]) output: LLMTableCorrection = result.output run.outputs = { "messages": [{"role": "assistant", "content": output.model_dump_json()}], "markdown": output.markdown, "units": output.units, "usage_metadata": usage_metadata_from(result.usage), } return output async def vet_structure( self, image_png: bytes, markdown: str, region_text: str, page_text: str ) -> Optional[LLMTableCorrection]: user_text = ( f"=== Camelot markdown ===\n{markdown}\n\n=== Table text layer ===\n{region_text}\n\n" f"=== Page text ===\n{page_text}\n\n" "Fix the structure against the cropped image; values from Camelot or the text layer only; " "copy `caption` and `title` from the page text." ) inputs = self.vet_inputs(user_text) async with self.tracer.llm_run("vet_structure", inputs, model=self.vet_model) as run: result = await self.vet_agent.run([user_text, self.binary_png(image_png)]) output: LLMTableCorrection = result.output run.outputs = { "messages": [{"role": "assistant", "content": output.model_dump_json()}], "markdown": output.markdown, "units": output.units, "usage_metadata": usage_metadata_from(result.usage), } return output def heading_inputs(self, user_text: str) -> Dict[str, Any]: return { "messages": [ {"role": "system", "content": REVIEW_HEADINGS_PROMPT}, {"role": "user", "content": [{"type": "text", "text": user_text}]}, ] } def review_headings_sync(self, nominated: str) -> Optional[LLMHeadingReview]: inputs = self.heading_inputs(nominated) with self.tracer.llm_run_sync("review_headings", inputs, model=self.model) as run: result = self.heading_agent.run_sync(nominated) output: LLMHeadingReview = result.output run.outputs = { "messages": [{"role": "assistant", "content": output.model_dump_json()}], "verdicts": [v.verdict for v in output.verdicts], "usage_metadata": usage_metadata_from(result.usage), } return output async def review_headings(self, nominated: str) -> Optional[LLMHeadingReview]: inputs = self.heading_inputs(nominated) async with self.tracer.llm_run("review_headings", inputs, model=self.model) as run: result = await self.heading_agent.run(nominated) output: LLMHeadingReview = result.output run.outputs = { "messages": [{"role": "assistant", "content": output.model_dump_json()}], "verdicts": [v.verdict for v in output.verdicts], "usage_metadata": usage_metadata_from(result.usage), } return output def _strip_coordinate_frame(markdown: str) -> str: """Remove the inline address tags from an addressed grid render, recovering the plain table exactly as `grid_to_markdown` renders it. Used by the mock client to echo input the way a compliant agent would.""" import re as _re from quber.core.extractors.camelot.acquire import grid_to_markdown rows = [] for line in (markdown or "").splitlines(): stripped = line.strip() if not stripped.startswith("|"): continue cells = [c.strip() for c in stripped.strip("|").split("|")] if cells and any("-" in c for c in cells) and all(set(c) <= {"-", ":", " "} for c in cells): continue rows.append([_re.sub(r"^\[[A-Z]+\d+\]\s*", "", c) for c in cells]) return grid_to_markdown(rows) if rows else "" class MockLLMClient: """Returns canned responses for tests. Does not call any LLM.""" def __init__( self, table_count: int = 1, correction: Optional[LLMTableCorrection] = None, ) -> None: self.table_count = table_count self.correction = correction def count_tables_sync(self, image_path: Path) -> int: _ = image_path return self.table_count def correct_structure_sync(self, image_path: Path, markdown: str) -> Optional[LLMTableCorrection]: _ = image_path if self.correction is not None: return self.correction return LLMTableCorrection(title="", caption="", markdown=markdown, footnotes=[], header_rows=1) async def count_tables(self, image_path: Path) -> int: return self.count_tables_sync(image_path) async def correct_structure(self, image_path: Path, markdown: str) -> Optional[LLMTableCorrection]: return self.correct_structure_sync(image_path, markdown) def vet_structure_sync( self, image_png: bytes, markdown: str, region_text: str, page_text: str ) -> Optional[LLMTableCorrection]: _ = (image_png, region_text, page_text) if self.correction is not None: return self.correction # Echo like a compliant agent: the input arrives inside the printed # coordinate frame, which must never appear in the output. return LLMTableCorrection( title="", caption="", markdown=_strip_coordinate_frame(markdown), footnotes=[], header_rows=1 ) async def vet_structure( self, image_png: bytes, markdown: str, region_text: str, page_text: str ) -> Optional[LLMTableCorrection]: return self.vet_structure_sync(image_png, markdown, region_text, page_text) def review_headings_sync(self, nominated: str) -> Optional[LLMHeadingReview]: # No review means no demotion — every heading is kept. _ = nominated return None async def review_headings(self, nominated: str) -> Optional[LLMHeadingReview]: _ = nominated return None Backend = Literal["cli", "api", "mock"] def get_llm_client(backend: Optional[Backend] = None) -> LLMClient: selected = backend or get_settings().llm.llm_backend if selected == "cli": return ClaudeCLIClient() if selected == "api": return PydanticAIClient() if selected == "mock": return MockLLMClient() raise ValueError(f"Unknown QUBER_LLM_BACKEND: {selected!r}. Expected cli|api|mock.")