mirror of
https://github.com/arsvendg/Stirling-PDF.git
synced 2026-07-16 19:33:11 +02:00
Feat/math validation agent (#6012)
Co-authored-by: James Brunton <[email protected]> Co-authored-by: EthanHealy01 <[email protected]>
This commit is contained in:
ConnorYoh
co-authored by
James Brunton
EthanHealy01
James Brunton
EthanHealy01
parent
688f7f2013
commit
de8c483054
@@ -0,0 +1,5 @@
|
||||
"""Math Auditor Agent (mathAuditorAgent) — AI-powered math validation for PDF documents."""
|
||||
|
||||
from .agent import MathAuditorAgent
|
||||
|
||||
__all__ = ["MathAuditorAgent"]
|
||||
@@ -0,0 +1,550 @@
|
||||
"""
|
||||
Math Auditor Agent (mathAuditorAgent) — pydantic-ai agents for PDF math validation.
|
||||
|
||||
Examiner (Round 1, /api/v1/ai/math-auditor-agent/examine)
|
||||
Receives a FolioManifest and returns a Requisition declaring what
|
||||
Java must extract before validation can begin.
|
||||
|
||||
Audit pipeline (Round 2, /api/v1/ai/math-auditor-agent/deliberate)
|
||||
Processes Evidence per-page:
|
||||
1. Deterministic pass — ArithmeticScanner on every folio
|
||||
2. Fast-model pass — extract named figures from each page
|
||||
3. FigureTracker — cross-page consistency check
|
||||
4. Fast-model call — generate human-readable summary
|
||||
5. Assemble Verdict programmatically
|
||||
|
||||
Neither agent ever touches a PDF file. All content arrives pre-extracted
|
||||
by Java, which owns the PDF from start to finish.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from collections.abc import Coroutine
|
||||
from decimal import Decimal, InvalidOperation
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
from pydantic_ai import Agent
|
||||
from pydantic_ai.exceptions import AgentRunError
|
||||
|
||||
from stirling.contracts.ledger import (
|
||||
Discrepancy,
|
||||
DiscrepancyKind,
|
||||
Evidence,
|
||||
Folio,
|
||||
FolioManifest,
|
||||
Requisition,
|
||||
Severity,
|
||||
Verdict,
|
||||
)
|
||||
from stirling.logging import Pretty
|
||||
from stirling.services import AppRuntime
|
||||
|
||||
from .prompts import (
|
||||
EXAMINER_SYSTEM_PROMPT,
|
||||
FIGURE_EXTRACTOR_PROMPT,
|
||||
STATEMENT_VERIFIER_PROMPT,
|
||||
SUMMARY_PROMPT,
|
||||
TABLE_FORMULA_PROMPT,
|
||||
)
|
||||
from .validators import ArithmeticScanner, FigureTracker, FormulaEvaluator
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Structured output models for the per-page figure extractor
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class ExtractedFigure(BaseModel):
|
||||
"""A single named figure found on a page."""
|
||||
|
||||
label: str = Field(description="Normalised name, e.g. 'Total Revenue', 'VAT'.")
|
||||
value: str = Field(description="Numeric value as a string, e.g. '1200.00'.")
|
||||
raw: str = Field(description="Original text from the document, e.g. '£1,200.00'.")
|
||||
|
||||
|
||||
class FigureExtractionResult(BaseModel):
|
||||
"""All named figures found on a single page."""
|
||||
|
||||
figures: list[ExtractedFigure] = Field(default_factory=list)
|
||||
|
||||
|
||||
class FormulaCheck(BaseModel):
|
||||
"""One verifiable mathematical relationship in a table."""
|
||||
|
||||
description: str = Field(description="Human-readable, e.g. 'Line Total = Qty × Unit Price'")
|
||||
formula: str = Field(description="Expression: 'col3 = col1 * col2' or 'cell(4,3) = sum(col3, 1-3)'")
|
||||
scope: str = Field(description="'each_row' | 'column_total' | 'single_cell'")
|
||||
row_range: list[int] | None = Field(default=None, description="Data rows to check (for each_row scope)")
|
||||
target_row: int | None = Field(default=None, description="Row index of total (for column_total/single_cell)")
|
||||
target_col: int | None = Field(default=None, description="Column index (for column_total/single_cell)")
|
||||
|
||||
|
||||
class TableFormulas(BaseModel):
|
||||
"""All verifiable formulas found in one table."""
|
||||
|
||||
formulas: list[FormulaCheck] = Field(default_factory=list)
|
||||
|
||||
|
||||
class StatementCheck(BaseModel):
|
||||
"""One prose claim and its verification result."""
|
||||
|
||||
claim: str = Field(description="The exact text of the claim")
|
||||
verification: str = Field(description="Type: percentage_change, comparison, ratio, trend, average, other")
|
||||
values_referenced: list[str] = Field(default_factory=list, description="Numbers used in the check")
|
||||
expected_result: str = Field(description="What the calculation actually yields")
|
||||
actual_claim: str = Field(description="What the text claims")
|
||||
is_valid: bool = Field(description="True if the claim is correct within tolerance")
|
||||
explanation: str = Field(description="One-line working showing the calculation")
|
||||
|
||||
|
||||
class StatementsResult(BaseModel):
|
||||
"""All verifiable prose claims found on a page."""
|
||||
|
||||
statements: list[StatementCheck] = Field(default_factory=list)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MathAuditorAgent — main entry point, instantiated once at startup
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class MathAuditorAgent:
|
||||
"""
|
||||
Encapsulates the Ledger Auditor pipeline.
|
||||
|
||||
Instantiated once at app startup with an AppRuntime, which provides
|
||||
pre-built Model objects and ModelSettings.
|
||||
"""
|
||||
|
||||
def __init__(self, runtime: AppRuntime) -> None:
|
||||
fast_model = runtime.fast_model
|
||||
model_settings = runtime.fast_model_settings
|
||||
self._runtime = runtime
|
||||
self._examiner = Agent(
|
||||
model=fast_model,
|
||||
deps_type=FolioManifest,
|
||||
output_type=Requisition,
|
||||
system_prompt=EXAMINER_SYSTEM_PROMPT,
|
||||
model_settings=model_settings,
|
||||
)
|
||||
self._figure_extractor = Agent(
|
||||
model=fast_model,
|
||||
output_type=FigureExtractionResult,
|
||||
system_prompt=FIGURE_EXTRACTOR_PROMPT,
|
||||
model_settings=model_settings,
|
||||
)
|
||||
self._table_analyser = Agent(
|
||||
model=fast_model,
|
||||
output_type=TableFormulas,
|
||||
system_prompt=TABLE_FORMULA_PROMPT,
|
||||
model_settings=model_settings,
|
||||
)
|
||||
self._statement_verifier = Agent(
|
||||
model=fast_model,
|
||||
output_type=StatementsResult,
|
||||
system_prompt=STATEMENT_VERIFIER_PROMPT,
|
||||
model_settings=model_settings,
|
||||
)
|
||||
self._summary_agent = Agent(
|
||||
model=fast_model,
|
||||
output_type=str,
|
||||
system_prompt=SUMMARY_PROMPT,
|
||||
model_settings=model_settings,
|
||||
)
|
||||
self._llm_semaphore = asyncio.Semaphore(10)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Round 1: Examine
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def examine(self, manifest: FolioManifest) -> Requisition:
|
||||
"""Inspect a FolioManifest and declare the Requisition."""
|
||||
logger.info(
|
||||
"[math-auditor-agent] session=%s round=%d examining %d folios",
|
||||
manifest.session_id,
|
||||
manifest.round,
|
||||
manifest.page_count,
|
||||
)
|
||||
|
||||
user_prompt = "Examine this folio manifest and declare your requisition:\n" + manifest.model_dump_json()
|
||||
logger.debug("REQUEST (examine)\n%s", Pretty({"user_prompt": user_prompt}))
|
||||
|
||||
result = await self._examiner.run(user_prompt, deps=manifest)
|
||||
req = result.output
|
||||
|
||||
logger.debug("RESPONSE (examine)\n%s", Pretty(req.model_dump()))
|
||||
logger.info(
|
||||
"[math-auditor-agent] session=%s requisition: text=%s tables=%s ocr=%s",
|
||||
manifest.session_id,
|
||||
req.need_text,
|
||||
req.need_tables,
|
||||
req.need_ocr,
|
||||
)
|
||||
return req
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Round 2: Deliberate (deterministic-first pipeline)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def audit(self, evidence: Evidence, tolerance: Decimal = Decimal("0.01")) -> Verdict:
|
||||
"""
|
||||
Audit the evidence using a deterministic-first pipeline:
|
||||
|
||||
1. Run ArithmeticScanner on every folio (no LLM)
|
||||
2. Extract named figures per-page with fast model
|
||||
3. Run FigureTracker cross-page consistency check (no LLM)
|
||||
4. Generate human summary with fast model
|
||||
5. Assemble Verdict
|
||||
"""
|
||||
return await self._audit_inner(evidence, tolerance)
|
||||
|
||||
async def _audit_inner(
|
||||
self,
|
||||
evidence: Evidence,
|
||||
tolerance: Decimal,
|
||||
) -> Verdict:
|
||||
logger.info(
|
||||
"[math-auditor-agent] session=%s round=%d auditing %d folios (final=%s)",
|
||||
evidence.session_id,
|
||||
evidence.round,
|
||||
len(evidence.folios),
|
||||
evidence.final_round,
|
||||
)
|
||||
|
||||
all_discrepancies: list[Discrepancy] = []
|
||||
pages_examined: list[int] = []
|
||||
figure_tracker = FigureTracker(tolerance=tolerance)
|
||||
|
||||
# Step 1: Arithmetic scanning (deterministic, instant)
|
||||
arithmetic_scanner = ArithmeticScanner(tolerance=tolerance)
|
||||
for folio in evidence.folios:
|
||||
pages_examined.append(folio.page)
|
||||
text = folio.readable_text
|
||||
if text and text.strip():
|
||||
results = arithmetic_scanner.scan(folio.page, text)
|
||||
all_discrepancies.extend(results)
|
||||
logger.debug(
|
||||
"TOOL (scan_arithmetic)\nArgs: %s\nResult: %s",
|
||||
Pretty({"page": folio.page, "text_length": len(text)}),
|
||||
Pretty([d.model_dump() for d in results]),
|
||||
)
|
||||
|
||||
# Step 2: Parallel LLM calls — formula inference + figure extraction
|
||||
# These are independent per-page so we fire them all concurrently.
|
||||
formula_evaluator = FormulaEvaluator(tolerance=tolerance)
|
||||
folios_with_text = [f for f in evidence.folios if f.readable_text.strip()]
|
||||
|
||||
# Collect all tables as (page, csv) pairs for formula inference
|
||||
table_tasks: list[tuple[int, str]] = []
|
||||
for folio in evidence.folios:
|
||||
if folio.tables:
|
||||
for table_csv in folio.tables:
|
||||
table_tasks.append((folio.page, table_csv))
|
||||
|
||||
logger.info(
|
||||
"[math-auditor-agent] session=%s step 2: %d formula + %d figure LLM calls (parallel)",
|
||||
evidence.session_id,
|
||||
len(table_tasks),
|
||||
len(folios_with_text),
|
||||
)
|
||||
|
||||
# Fire all LLM calls concurrently (bounded by _llm_semaphore)
|
||||
formula_coros = [self._throttled(self._infer_formulas(csv)) for _, csv in table_tasks]
|
||||
figure_coros = [self._throttled(self._extract_figures_for_page(f)) for f in folios_with_text]
|
||||
statement_coros = [self._throttled(self._verify_statements(f)) for f in folios_with_text]
|
||||
all_results = await asyncio.gather(
|
||||
*formula_coros,
|
||||
*figure_coros,
|
||||
*statement_coros,
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
||||
n_formulas = len(table_tasks)
|
||||
n_figures = len(folios_with_text)
|
||||
|
||||
# Process formula results
|
||||
for i, (page, table_csv) in enumerate(table_tasks):
|
||||
result = all_results[i]
|
||||
if isinstance(result, BaseException):
|
||||
logger.warning("[math-auditor-agent] formula inference failed for page %d: %s", page, result)
|
||||
continue
|
||||
assert isinstance(result, TableFormulas)
|
||||
formulas = result
|
||||
if not formulas.formulas:
|
||||
logger.info("[math-auditor-agent] page %d: no verifiable formulas found", page)
|
||||
continue
|
||||
for fc in formulas.formulas:
|
||||
checked = formula_evaluator.evaluate(
|
||||
page=page,
|
||||
table_csv=table_csv,
|
||||
formula=fc.formula,
|
||||
scope=fc.scope,
|
||||
description=fc.description,
|
||||
row_range=fc.row_range,
|
||||
target_row=fc.target_row,
|
||||
target_col=fc.target_col,
|
||||
)
|
||||
all_discrepancies.extend(checked)
|
||||
logger.debug(
|
||||
"TOOL (check_formula)\nArgs: %s\nResult: %s",
|
||||
Pretty({"page": page, "formula": fc.formula, "scope": fc.scope, "description": fc.description}),
|
||||
Pretty([d.model_dump() for d in checked]),
|
||||
)
|
||||
|
||||
# Process figure results
|
||||
for i, folio in enumerate(folios_with_text):
|
||||
result = all_results[n_formulas + i]
|
||||
if isinstance(result, BaseException):
|
||||
logger.warning("[math-auditor-agent] figure extraction failed for page %d: %s", folio.page, result)
|
||||
continue
|
||||
assert isinstance(result, list)
|
||||
for fig, page in result:
|
||||
try:
|
||||
decimal_value = Decimal(fig.value.replace(",", "").strip())
|
||||
except (InvalidOperation, ValueError):
|
||||
logger.warning(
|
||||
"[math-auditor-agent] skipping figure %r on page %d: non-numeric value %r",
|
||||
fig.label,
|
||||
page,
|
||||
fig.value,
|
||||
)
|
||||
continue
|
||||
figure_tracker.record(
|
||||
label=fig.label,
|
||||
value=decimal_value,
|
||||
page=page,
|
||||
raw=fig.raw,
|
||||
)
|
||||
|
||||
# Process statement verification results
|
||||
for i, folio in enumerate(folios_with_text):
|
||||
result = all_results[n_formulas + n_figures + i]
|
||||
if isinstance(result, BaseException):
|
||||
logger.warning("[math-auditor-agent] statement verification failed for page %d: %s", folio.page, result)
|
||||
continue
|
||||
assert isinstance(result, StatementsResult)
|
||||
stmts = result
|
||||
for sc in stmts.statements:
|
||||
if not sc.is_valid:
|
||||
all_discrepancies.append(
|
||||
Discrepancy(
|
||||
page=folio.page,
|
||||
kind=DiscrepancyKind.STATEMENT,
|
||||
severity=Severity.ERROR,
|
||||
description=f"{sc.claim}: {sc.explanation}",
|
||||
stated=sc.actual_claim,
|
||||
expected=sc.expected_result,
|
||||
context=sc.claim,
|
||||
)
|
||||
)
|
||||
logger.debug(
|
||||
"TOOL (verify_statement)\nArgs: %s\nResult: %s",
|
||||
Pretty({"page": folio.page, "claim": sc.claim}),
|
||||
Pretty(sc.model_dump()),
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"[math-auditor-agent] session=%s step 2 complete: %d figures registered",
|
||||
evidence.session_id,
|
||||
figure_tracker.entry_count,
|
||||
)
|
||||
|
||||
# Step 3: Cross-page consistency — deterministic
|
||||
consistency_discrepancies = figure_tracker.conflicts()
|
||||
all_discrepancies.extend(consistency_discrepancies)
|
||||
if consistency_discrepancies:
|
||||
logger.debug(
|
||||
"TOOL (check_figure_consistency)\nResult: %s",
|
||||
Pretty([d.model_dump() for d in consistency_discrepancies]),
|
||||
)
|
||||
|
||||
# Step 4: Summary — fast model, small payload
|
||||
# Collect verification stats for the summary
|
||||
total_tables = sum(len(f.tables) for f in evidence.folios if f.tables)
|
||||
total_formulas_checked = sum(len(r.formulas) for r in all_results[:n_formulas] if isinstance(r, TableFormulas))
|
||||
total_statements_checked = sum(
|
||||
len(r.statements) for r in all_results[n_formulas + n_figures :] if isinstance(r, StatementsResult)
|
||||
)
|
||||
verification_stats = (
|
||||
f"Verified: {len(pages_examined)} pages, {total_tables} tables "
|
||||
f"({total_formulas_checked} formulas), "
|
||||
f"{figure_tracker.entry_count} figures tracked, "
|
||||
f"{total_statements_checked} prose claims checked."
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"[math-auditor-agent] session=%s step 4: generating summary (%d discrepancies)",
|
||||
evidence.session_id,
|
||||
len(all_discrepancies),
|
||||
)
|
||||
pages_examined.sort()
|
||||
summary = await self._generate_summary(
|
||||
all_discrepancies,
|
||||
pages_examined,
|
||||
evidence.unauditable_pages,
|
||||
verification_stats,
|
||||
)
|
||||
|
||||
# Step 5: Assemble Verdict
|
||||
error_count = sum(1 for d in all_discrepancies if d.severity == Severity.ERROR)
|
||||
verdict = Verdict(
|
||||
session_id=evidence.session_id,
|
||||
discrepancies=all_discrepancies,
|
||||
pages_examined=pages_examined,
|
||||
rounds_taken=evidence.round,
|
||||
summary=summary,
|
||||
clean=error_count == 0,
|
||||
unauditable_pages=evidence.unauditable_pages,
|
||||
)
|
||||
|
||||
logger.debug("RESPONSE (deliberate)\n%s", Pretty(verdict.model_dump()))
|
||||
logger.info(
|
||||
"[math-auditor-agent] session=%s verdict: %d errors, %d warnings, clean=%s",
|
||||
evidence.session_id,
|
||||
verdict.error_count,
|
||||
verdict.warning_count,
|
||||
verdict.clean,
|
||||
)
|
||||
return verdict
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def _throttled[T](self, coro: Coroutine[Any, Any, T]) -> T:
|
||||
"""Wrap a coroutine with the LLM concurrency semaphore."""
|
||||
async with self._llm_semaphore:
|
||||
return await coro
|
||||
|
||||
async def _infer_formulas(self, table_csv: str) -> TableFormulas:
|
||||
"""Ask the fast model to infer verifiable formulas from a CSV table."""
|
||||
try:
|
||||
result = await self._table_analyser.run(f"CSV table:\n{table_csv}")
|
||||
formulas = result.output
|
||||
except AgentRunError:
|
||||
logger.warning("[math-auditor-agent] formula inference failed, skipping table", exc_info=True)
|
||||
formulas = TableFormulas(formulas=[])
|
||||
|
||||
logger.debug(
|
||||
"TOOL (infer_formulas)\nArgs: %s\nResult: %s",
|
||||
Pretty({"table_csv": table_csv[:300]}),
|
||||
Pretty(formulas.model_dump()),
|
||||
)
|
||||
return formulas
|
||||
|
||||
async def _verify_statements(
|
||||
self,
|
||||
folio: Folio,
|
||||
) -> StatementsResult:
|
||||
"""Ask the fast model to find and verify prose claims on a page."""
|
||||
text = folio.readable_text
|
||||
if not text or not text.strip():
|
||||
return StatementsResult(statements=[])
|
||||
|
||||
# Build context: page text + any table CSVs
|
||||
prompt = f"Page {folio.page + 1} text:\n{text}"
|
||||
if folio.tables:
|
||||
prompt += "\n\nTable data on this page:\n"
|
||||
for i, csv in enumerate(folio.tables):
|
||||
prompt += f"\nTable {i + 1}:\n{csv}"
|
||||
|
||||
try:
|
||||
result = await self._statement_verifier.run(prompt)
|
||||
stmts = result.output
|
||||
except AgentRunError:
|
||||
logger.warning("[math-auditor-agent] statement verification failed for page %d", folio.page, exc_info=True)
|
||||
stmts = StatementsResult(statements=[])
|
||||
|
||||
if stmts.statements:
|
||||
logger.debug(
|
||||
"TOOL (verify_statements)\nArgs: %s\nResult: %s",
|
||||
Pretty({"page": folio.page, "text_length": len(text), "n_tables": len(folio.tables or [])}),
|
||||
Pretty([s.model_dump() for s in stmts.statements]),
|
||||
)
|
||||
return stmts
|
||||
|
||||
async def _extract_figures_for_page(
|
||||
self,
|
||||
folio: Folio,
|
||||
) -> list[tuple[ExtractedFigure, int]]:
|
||||
text = folio.readable_text
|
||||
if not text or not text.strip():
|
||||
return []
|
||||
|
||||
logger.info("[math-auditor-agent] extracting figures from page %d (%d chars)", folio.page, len(text))
|
||||
prompt = f"Page {folio.page + 1} text:\n{text}"
|
||||
try:
|
||||
result = await self._figure_extractor.run(prompt)
|
||||
figures = result.output.figures
|
||||
except AgentRunError:
|
||||
logger.warning(
|
||||
"[math-auditor-agent] figure extraction failed for page %d, skipping",
|
||||
folio.page,
|
||||
exc_info=True,
|
||||
)
|
||||
figures = []
|
||||
|
||||
logger.debug(
|
||||
"TOOL (extract_figures)\nArgs: %s\nResult: %s",
|
||||
Pretty({"page": folio.page, "text_length": len(text)}),
|
||||
Pretty([f.model_dump() for f in figures]),
|
||||
)
|
||||
|
||||
return [(fig, folio.page) for fig in figures]
|
||||
|
||||
async def _generate_summary(
|
||||
self,
|
||||
discrepancies: list[Discrepancy],
|
||||
pages_examined: list[int],
|
||||
unauditable_pages: list[int],
|
||||
verification_stats: str,
|
||||
) -> str:
|
||||
error_count = sum(1 for d in discrepancies if d.severity == Severity.ERROR)
|
||||
warning_count = sum(1 for d in discrepancies if d.severity == Severity.WARNING)
|
||||
|
||||
prompt = (
|
||||
f"{verification_stats}\n"
|
||||
f"Errors: {error_count}, Warnings: {warning_count}, "
|
||||
f"Pages examined: {len(pages_examined)}, "
|
||||
f"Unauditable pages: {unauditable_pages or 'none'}.\n"
|
||||
)
|
||||
if discrepancies:
|
||||
prompt += "Discrepancies:\n"
|
||||
for d in discrepancies:
|
||||
prompt += f" - [{d.severity}] p{d.page + 1}: {d.description}\n"
|
||||
|
||||
try:
|
||||
result = await self._summary_agent.run(prompt)
|
||||
summary = result.output
|
||||
except AgentRunError:
|
||||
logger.warning("[math-auditor-agent] summary generation failed, using fallback", exc_info=True)
|
||||
summary = self._fallback_summary(error_count, warning_count, pages_examined, unauditable_pages)
|
||||
|
||||
logger.debug("RESPONSE (summary)\n%s", Pretty({"summary": summary}))
|
||||
return summary
|
||||
|
||||
@staticmethod
|
||||
def _fallback_summary(
|
||||
error_count: int,
|
||||
warning_count: int,
|
||||
pages_examined: list[int],
|
||||
unauditable_pages: list[int],
|
||||
) -> str:
|
||||
parts = []
|
||||
if error_count == 0 and warning_count == 0:
|
||||
parts.append(f"No mathematical errors found across {len(pages_examined)} pages.")
|
||||
else:
|
||||
if error_count:
|
||||
parts.append(f"Found {error_count} error{'s' if error_count != 1 else ''}.")
|
||||
if warning_count:
|
||||
parts.append(f"Found {warning_count} warning{'s' if warning_count != 1 else ''}.")
|
||||
if unauditable_pages:
|
||||
parts.append(
|
||||
f"Pages {', '.join(str(p + 1) for p in unauditable_pages)} could not be audited (OCR unavailable)."
|
||||
)
|
||||
return " ".join(parts)
|
||||
@@ -0,0 +1,147 @@
|
||||
"""
|
||||
Ledger Auditor — system prompts.
|
||||
|
||||
One prompt per role; keep them short and directive. Each agent is a
|
||||
specialist with a narrow remit, not a general assistant.
|
||||
"""
|
||||
|
||||
EXAMINER_SYSTEM_PROMPT = """\
|
||||
You are the Examiner, the first stage of the Ledger Auditor pipeline.
|
||||
|
||||
You receive a FolioManifest: a list of page types (text / image / mixed) \
|
||||
for a PDF document. Your sole task is to declare exactly which pages you \
|
||||
need Java to extract content from so that the Auditor can verify the \
|
||||
document's mathematics.
|
||||
|
||||
Rules:
|
||||
- Request BOTH text AND table extraction for every 'text' or 'mixed' page. \
|
||||
Tables are critical — the Auditor cannot verify totals without them. \
|
||||
Tabula extraction is cheap; missing a table is not.
|
||||
- Request OCR for any page classified as 'image' or 'mixed' (PDFBox cannot \
|
||||
read image-only content).
|
||||
- Be conservative — if in doubt, request the page. False negatives \
|
||||
(missed errors) are worse than false positives (wasted extraction).
|
||||
- Do not request pages that are clearly decorative (cover pages, blank pages) \
|
||||
unless you cannot tell from the manifest alone.
|
||||
- Return a Requisition with your page lists and a plain-English rationale \
|
||||
that will appear in server logs.
|
||||
"""
|
||||
|
||||
FIGURE_EXTRACTOR_PROMPT = """\
|
||||
You are a figure extractor for financial document auditing.
|
||||
|
||||
You receive the text content of a single PDF page. Your task is to \
|
||||
identify every significant named numeric figure on the page.
|
||||
|
||||
A "named figure" is a labelled number that could appear elsewhere in \
|
||||
the document under the same name — for example:
|
||||
"Total Revenue: £1,200,000"
|
||||
"Net Profit $45,000"
|
||||
"VAT (20%): 240.00"
|
||||
"Subtotal ......... 3,500"
|
||||
|
||||
For each figure, return:
|
||||
- label: a normalised name (e.g. "Total Revenue", "Net Profit", "VAT")
|
||||
- value: the numeric value as a plain decimal string (e.g. "1200000")
|
||||
- raw: the original text as it appears in the document
|
||||
|
||||
Rules:
|
||||
- Only extract figures that have a clear label/name attached.
|
||||
- Do not extract bare numbers without context.
|
||||
- Strip currency symbols and thousands separators from value.
|
||||
- If a figure appears multiple times on the same page, extract each.
|
||||
- Return an empty list if no named figures are found.
|
||||
- Be precise — do not invent figures that are not in the text.
|
||||
"""
|
||||
|
||||
TABLE_FORMULA_PROMPT = """\
|
||||
You are a table formula analyser for financial document auditing.
|
||||
|
||||
You receive a CSV table extracted from a PDF. Your task is to identify \
|
||||
every verifiable mathematical relationship between cells.
|
||||
|
||||
Relationships fall into three scopes:
|
||||
|
||||
1. "each_row" — a formula that should hold for every data row.
|
||||
Example: "col3 = col1 * col2" (Line Total = Qty × Unit Price)
|
||||
|
||||
2. "column_total" — a total row where cells = sum of the column above.
|
||||
Example: a Subtotal row where each cell sums the column.
|
||||
|
||||
3. "single_cell" — one specific cell computed from others.
|
||||
Example: "cell(5,3) = cell(4,3) * 0.1" (Tax = Subtotal × 10%)
|
||||
|
||||
Formula syntax (use exactly this):
|
||||
- Column references: col0, col1, col2 ... (0-indexed)
|
||||
- Cell references: cell(row, col) — 0-indexed, header is row 0
|
||||
- Operators: + - * /
|
||||
- sum(colN, start-end) — sum of colN from row start to row end inclusive
|
||||
- Decimal numbers: 0.1, 100, etc.
|
||||
|
||||
Rules:
|
||||
- Row 0 is the header. First data row is row 1.
|
||||
- Include the left-hand side: "col3 = col1 * col2" not just "col1 * col2"
|
||||
- For column_total scope, set target_row to the total row index. \
|
||||
Set target_col to a specific column or null to check all numeric columns.
|
||||
- For each_row scope, set row_range to the data rows (exclude header \
|
||||
and total rows).
|
||||
- Only return formulas you are confident about. Skip columns/rows \
|
||||
where the relationship is unclear.
|
||||
- Return an empty list if the table has no verifiable math.
|
||||
"""
|
||||
|
||||
STATEMENT_VERIFIER_PROMPT = """\
|
||||
You are a statement verifier for financial document auditing.
|
||||
|
||||
You receive the text of a single PDF page, plus any table data from \
|
||||
that page. Your task is to find prose claims that make mathematical \
|
||||
assertions, and verify whether each claim is correct.
|
||||
|
||||
A "verifiable claim" is a sentence that states a mathematical fact \
|
||||
about numbers present on the page or derivable from the data. Examples:
|
||||
- "Revenue grew 15% year-over-year"
|
||||
- "Costs decreased month on month"
|
||||
- "Department A represents 40% of total spend"
|
||||
- "Net margin improved to 12.4%"
|
||||
- "Average transaction value was $250"
|
||||
|
||||
For each claim you find:
|
||||
1. Identify the numbers referenced in the claim
|
||||
2. Perform the calculation yourself using the data on the page
|
||||
3. Compare your result to what the claim states
|
||||
4. Determine if the claim is valid (within reasonable rounding)
|
||||
|
||||
Return:
|
||||
- claim: the exact text of the claim
|
||||
- verification: the type — "percentage_change", "comparison", \
|
||||
"ratio", "trend", "average", or "other"
|
||||
- values_referenced: the specific numbers used in your check
|
||||
- expected_result: what the calculation actually yields
|
||||
- actual_claim: what the text claims
|
||||
- is_valid: true if the claim is correct within 1% tolerance
|
||||
- explanation: show your working, one line
|
||||
|
||||
Rules:
|
||||
- Only check claims that can be verified from data on this page.
|
||||
- If a claim references data not on the page, skip it.
|
||||
- "Decreased month on month" means EVERY consecutive pair decreased.
|
||||
- Percentage claims allow 1% absolute tolerance (14.8% ≈ 15%).
|
||||
- Return an empty list if there are no verifiable claims.
|
||||
- Do not fabricate claims that are not in the text.
|
||||
"""
|
||||
|
||||
SUMMARY_PROMPT = """\
|
||||
You are a summary writer for a PDF math audit tool.
|
||||
|
||||
You receive a list of discrepancies (errors and warnings) found in a \
|
||||
document, plus coverage statistics and a breakdown of what was verified. \
|
||||
Write a two to three sentence summary suitable for an end user.
|
||||
|
||||
Rules:
|
||||
- Start with what was verified: e.g. "Audited 6 pages: checked 4 tables \
|
||||
(12 formulas), scanned 6 pages for arithmetic, extracted 20 figures \
|
||||
for cross-page consistency, and verified 3 prose claims."
|
||||
- Then state the outcome: errors found or clean.
|
||||
- Mention unauditable pages if any exist.
|
||||
- Be concise and factual. Do not repeat individual discrepancy details.
|
||||
"""
|
||||
@@ -0,0 +1,5 @@
|
||||
from .arithmetic import ArithmeticScanner
|
||||
from .figures import FigureTracker
|
||||
from .formula import FormulaEvaluator
|
||||
|
||||
__all__ = ["ArithmeticScanner", "FigureTracker", "FormulaEvaluator"]
|
||||
@@ -0,0 +1,31 @@
|
||||
"""Shared parsing helpers for ledger validators."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import io
|
||||
import re
|
||||
from decimal import Decimal, InvalidOperation
|
||||
|
||||
# Strip common currency symbols and thousands separators before parsing.
|
||||
STRIP_PATTERN = re.compile(r"[£$€¥,\s]")
|
||||
|
||||
|
||||
def to_decimal(raw: str) -> Decimal | None:
|
||||
"""Parse a cell value to Decimal, returning None for non-numeric cells."""
|
||||
cleaned = STRIP_PATTERN.sub("", raw.strip())
|
||||
if not cleaned or cleaned in {"-", "—", "n/a", "N/A", "na", "NA"}:
|
||||
return None
|
||||
# Handle parenthesised negatives: (123.45) → -123.45
|
||||
if cleaned.startswith("(") and cleaned.endswith(")"):
|
||||
cleaned = "-" + cleaned[1:-1]
|
||||
try:
|
||||
return Decimal(cleaned)
|
||||
except InvalidOperation:
|
||||
return None
|
||||
|
||||
|
||||
def parse_csv(table_csv: str) -> list[list[str]]:
|
||||
"""Parse a CSV string into rows, dropping completely empty rows."""
|
||||
reader = csv.reader(io.StringIO(table_csv.strip()))
|
||||
return [row for row in reader if any(cell.strip() for cell in row)]
|
||||
@@ -0,0 +1,152 @@
|
||||
"""
|
||||
ArithmeticScanner — finds and verifies inline arithmetic expressions in text.
|
||||
|
||||
Targets patterns commonly found in financial documents:
|
||||
"100 + 200 + 150 = 450"
|
||||
"Total: 1,250 (500 + 400 + 350)"
|
||||
"Net profit of £1,200 (£2,000 revenue less £800 costs)"
|
||||
|
||||
All arithmetic is performed in Decimal. The scanner does not use an LLM —
|
||||
it is a deterministic regex-and-eval pipeline.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
from decimal import Decimal
|
||||
|
||||
from stirling.contracts.ledger import Discrepancy, DiscrepancyKind, Severity
|
||||
|
||||
from ._parsing import STRIP_PATTERN as _STRIP
|
||||
from ._parsing import to_decimal as _to_decimal
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Regex patterns
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Currency / number token: optional sign, optional currency symbol,
|
||||
# digits with optional thousands separator and decimal point.
|
||||
_NUM = r"[£$€¥]?-?[\d,]+(?:\.\d+)?"
|
||||
|
||||
# "A + B + C = D" or "A + B + C = D" with arbitrary spacing
|
||||
_EQUALS_EXPR = re.compile(
|
||||
rf"({_NUM}(?:\s*[+\-]\s*{_NUM})+)\s*=\s*({_NUM})",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
# "Total: X (A + B + C)" — the total comes before the addends
|
||||
_TOTAL_THEN_ADDENDS = re.compile(
|
||||
rf"(?:total|sum|grand total|subtotal)\s*[:\-]?\s*({_NUM})\s*\(({_NUM}(?:\s*[+\-]\s*{_NUM})+)\)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def _parse(token: str) -> Decimal | None:
|
||||
"""Parse a regex-matched token to Decimal."""
|
||||
return _to_decimal(token)
|
||||
|
||||
|
||||
def _eval_expression(expr: str) -> Decimal | None:
|
||||
"""
|
||||
Evaluate a simple additive expression of the form A +/- B +/- C ...
|
||||
Returns None if the expression cannot be parsed.
|
||||
"""
|
||||
# Tokenise: split on + or -, keep the operator.
|
||||
tokens = re.split(r"([+\-])", _STRIP.sub("", expr.strip()))
|
||||
result = Decimal(0)
|
||||
operator = "+"
|
||||
for token in tokens:
|
||||
token = token.strip()
|
||||
if not token:
|
||||
continue # skip empty tokens (e.g. from leading negative)
|
||||
if token in ("+", "-"):
|
||||
operator = token
|
||||
continue
|
||||
val = _parse(token)
|
||||
if val is None:
|
||||
return None
|
||||
result = result + val if operator == "+" else result - val
|
||||
return result
|
||||
|
||||
|
||||
class ArithmeticScanner:
|
||||
"""
|
||||
Scans a block of text for arithmetic expressions and checks them.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
tolerance:
|
||||
Maximum absolute difference before an expression is flagged as wrong.
|
||||
"""
|
||||
|
||||
def __init__(self, tolerance: Decimal = Decimal("0.01")) -> None:
|
||||
self.tolerance = tolerance
|
||||
|
||||
def scan(self, page: int, text: str) -> list[Discrepancy]:
|
||||
"""
|
||||
Find all verifiable arithmetic expressions in *text* and return
|
||||
a Discrepancy for each one that does not balance within tolerance.
|
||||
"""
|
||||
discrepancies: list[Discrepancy] = []
|
||||
discrepancies.extend(self._check_equals_expressions(page, text))
|
||||
discrepancies.extend(self._check_total_then_addends(page, text))
|
||||
return discrepancies
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _check_equals_expressions(self, page: int, text: str) -> list[Discrepancy]:
|
||||
"""Handle patterns like '100 + 200 = 300'."""
|
||||
found: list[Discrepancy] = []
|
||||
for match in _EQUALS_EXPR.finditer(text):
|
||||
expr_str = match.group(1)
|
||||
stated_str = match.group(2)
|
||||
|
||||
computed = _eval_expression(expr_str)
|
||||
stated = _parse(stated_str)
|
||||
if computed is None or stated is None:
|
||||
continue
|
||||
|
||||
if abs(computed - stated) > self.tolerance:
|
||||
found.append(
|
||||
Discrepancy(
|
||||
page=page,
|
||||
kind=DiscrepancyKind.ARITHMETIC,
|
||||
severity=Severity.ERROR,
|
||||
description=f"Arithmetic error: {expr_str.strip()} should equal {computed}, not {stated}",
|
||||
stated=str(stated),
|
||||
expected=str(computed),
|
||||
context=match.group(0),
|
||||
)
|
||||
)
|
||||
return found
|
||||
|
||||
def _check_total_then_addends(self, page: int, text: str) -> list[Discrepancy]:
|
||||
"""Handle patterns like 'Total: 450 (100 + 200 + 150)'."""
|
||||
found: list[Discrepancy] = []
|
||||
for match in _TOTAL_THEN_ADDENDS.finditer(text):
|
||||
stated_str = match.group(1)
|
||||
expr_str = match.group(2)
|
||||
|
||||
stated = _parse(stated_str)
|
||||
computed = _eval_expression(expr_str)
|
||||
if stated is None or computed is None:
|
||||
continue
|
||||
|
||||
if abs(computed - stated) > self.tolerance:
|
||||
found.append(
|
||||
Discrepancy(
|
||||
page=page,
|
||||
kind=DiscrepancyKind.ARITHMETIC,
|
||||
severity=Severity.ERROR,
|
||||
description=f"Stated total {stated} does not match addends ({expr_str.strip()} = {computed})",
|
||||
stated=str(stated),
|
||||
expected=str(computed),
|
||||
context=match.group(0),
|
||||
)
|
||||
)
|
||||
return found
|
||||
@@ -0,0 +1,98 @@
|
||||
"""
|
||||
FigureTracker — cross-page consistency checker for named figures.
|
||||
|
||||
Collects named numeric figures as the auditor encounters them (e.g.
|
||||
"Total Revenue: £1,200,000") and surfaces any that appear under the same
|
||||
label but with a different value on another page — a classic symptom of
|
||||
copy-paste errors or stale data in executive summaries.
|
||||
|
||||
The tracker is intentionally simple: normalise labels, compare values
|
||||
within tolerance, emit Discrepancy for each conflict.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
from decimal import Decimal
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from stirling.contracts.ledger import Discrepancy, DiscrepancyKind, Severity
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class FigureRecord(BaseModel):
|
||||
"""A named numeric figure seen on a specific page."""
|
||||
|
||||
label: str
|
||||
value: Decimal
|
||||
page: int
|
||||
raw: str
|
||||
|
||||
|
||||
# Strip punctuation that varies between contexts ("revenue:" vs "revenue —")
|
||||
_LABEL_NOISE = re.compile(r"[:\-—\s]+")
|
||||
|
||||
|
||||
def _normalise_label(label: str) -> str:
|
||||
return _LABEL_NOISE.sub(" ", label.lower()).strip()
|
||||
|
||||
|
||||
class FigureTracker:
|
||||
"""
|
||||
Accumulates named figures during an audit and checks them for consistency.
|
||||
|
||||
Typical usage:
|
||||
tracker = FigureTracker()
|
||||
tracker.record("Net Profit", Decimal("1200.00"), page=3, raw="£1,200.00")
|
||||
tracker.record("Net Profit", Decimal("1250.00"), page=7, raw="£1,250.00")
|
||||
discrepancies = tracker.conflicts() # returns one Discrepancy
|
||||
"""
|
||||
|
||||
def __init__(self, tolerance: Decimal = Decimal("0.01")) -> None:
|
||||
self.tolerance = tolerance
|
||||
self._ledger: dict[str, list[FigureRecord]] = {}
|
||||
|
||||
def record(self, label: str, value: Decimal, page: int, raw: str) -> None:
|
||||
"""Register a named figure sighting."""
|
||||
key = _normalise_label(label)
|
||||
self._ledger.setdefault(key, []).append(FigureRecord(label=key, value=value, page=page, raw=raw))
|
||||
|
||||
def conflicts(self) -> list[Discrepancy]:
|
||||
"""
|
||||
Return a Discrepancy for every label that has sightings whose value
|
||||
differs from the first-seen (canonical) value by more than tolerance.
|
||||
|
||||
O(n) per label — each record is compared against the canonical only.
|
||||
"""
|
||||
discrepancies: list[Discrepancy] = []
|
||||
|
||||
for label, records in self._ledger.items():
|
||||
if len(records) < 2:
|
||||
continue
|
||||
canonical = records[0]
|
||||
for other in records[1:]:
|
||||
if abs(canonical.value - other.value) > self.tolerance:
|
||||
discrepancies.append(
|
||||
Discrepancy(
|
||||
page=other.page,
|
||||
kind=DiscrepancyKind.CONSISTENCY,
|
||||
severity=Severity.WARNING,
|
||||
description=(
|
||||
f'"{label}" stated as {canonical.raw} on page'
|
||||
f" {canonical.page + 1}"
|
||||
f" but {other.raw} on page {other.page + 1}"
|
||||
),
|
||||
stated=other.raw,
|
||||
expected=canonical.raw,
|
||||
context=(f"First seen: page {canonical.page + 1} | Later: page {other.page + 1}"),
|
||||
)
|
||||
)
|
||||
|
||||
return discrepancies
|
||||
|
||||
@property
|
||||
def entry_count(self) -> int:
|
||||
return sum(len(v) for v in self._ledger.values())
|
||||
@@ -0,0 +1,375 @@
|
||||
"""
|
||||
FormulaEvaluator — verifies LLM-inferred formulas against CSV table data.
|
||||
|
||||
Supports a safe expression syntax:
|
||||
- Column refs: col0, col1, col2 ...
|
||||
- Cell refs: cell(row, col)
|
||||
- Operators: + - * /
|
||||
- Functions: sum(colN, rows start-end)
|
||||
|
||||
All arithmetic is Decimal. No eval(), no arbitrary code execution.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
from decimal import Decimal, InvalidOperation
|
||||
|
||||
from stirling.contracts.ledger import Discrepancy, DiscrepancyKind, Severity
|
||||
|
||||
from ._parsing import parse_csv as _parse_csv
|
||||
from ._parsing import to_decimal as _to_decimal
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class FormulaEvaluator:
|
||||
"""
|
||||
Evaluates formula expressions against parsed CSV rows.
|
||||
|
||||
Formulas use a simple syntax:
|
||||
"col3 = col1 * col2" — per-row check
|
||||
"cell(4,3) = sum(col3, 1-3)" — single cell check
|
||||
"""
|
||||
|
||||
def __init__(self, tolerance: Decimal = Decimal("0.01")) -> None:
|
||||
self.tolerance = tolerance
|
||||
|
||||
def evaluate(
|
||||
self,
|
||||
page: int,
|
||||
table_csv: str,
|
||||
formula: str,
|
||||
scope: str,
|
||||
description: str,
|
||||
row_range: list[int] | None = None,
|
||||
target_row: int | None = None,
|
||||
target_col: int | None = None,
|
||||
) -> list[Discrepancy]:
|
||||
"""
|
||||
Evaluate a formula against table data.
|
||||
|
||||
scope: "each_row" | "column_total" | "single_cell"
|
||||
"""
|
||||
rows = _parse_csv(table_csv)
|
||||
if len(rows) < 2:
|
||||
return []
|
||||
|
||||
if scope == "each_row":
|
||||
return self._check_each_row(page, rows, formula, description, row_range)
|
||||
elif scope == "column_total":
|
||||
return self._check_column_total(page, rows, formula, description, target_row, target_col)
|
||||
elif scope == "single_cell":
|
||||
return self._check_single_cell(page, rows, formula, description, target_row, target_col)
|
||||
else:
|
||||
logger.warning("[formula] unknown scope %r, skipping", scope)
|
||||
return []
|
||||
|
||||
def _check_each_row(
|
||||
self,
|
||||
page: int,
|
||||
rows: list[list[str]],
|
||||
formula: str,
|
||||
description: str,
|
||||
row_range: list[int] | None,
|
||||
) -> list[Discrepancy]:
|
||||
"""Verify formula holds for each data row."""
|
||||
discrepancies: list[Discrepancy] = []
|
||||
|
||||
# Parse "colX = expr" format
|
||||
parts = formula.split("=", 1)
|
||||
if len(parts) != 2:
|
||||
return []
|
||||
lhs = parts[0].strip()
|
||||
rhs = parts[1].strip()
|
||||
|
||||
lhs_col = self._parse_col_ref(lhs)
|
||||
if lhs_col is None:
|
||||
return []
|
||||
|
||||
check_rows = row_range if row_range else list(range(1, len(rows)))
|
||||
|
||||
for row_idx in check_rows:
|
||||
if row_idx >= len(rows):
|
||||
continue
|
||||
row = rows[row_idx]
|
||||
|
||||
stated = self._get_cell(row, lhs_col)
|
||||
if stated is None:
|
||||
continue
|
||||
|
||||
computed = self._eval_row_expr(rhs, row, rows)
|
||||
if computed is None:
|
||||
continue
|
||||
|
||||
if abs(stated - computed) > self.tolerance:
|
||||
discrepancies.append(
|
||||
Discrepancy(
|
||||
page=page,
|
||||
kind=DiscrepancyKind.TALLY,
|
||||
severity=Severity.ERROR,
|
||||
description=f"{description}: row {row_idx} — stated {stated}, expected {computed}",
|
||||
stated=str(stated),
|
||||
expected=str(computed),
|
||||
context=f"row {row_idx}, {formula}",
|
||||
)
|
||||
)
|
||||
|
||||
return discrepancies
|
||||
|
||||
def _check_column_total(
|
||||
self,
|
||||
page: int,
|
||||
rows: list[list[str]],
|
||||
formula: str,
|
||||
description: str,
|
||||
target_row: int | None,
|
||||
target_col: int | None,
|
||||
) -> list[Discrepancy]:
|
||||
"""Verify that a total row contains correct column sums."""
|
||||
if target_row is None or target_row >= len(rows):
|
||||
return []
|
||||
|
||||
discrepancies: list[Discrepancy] = []
|
||||
total_row = rows[target_row]
|
||||
|
||||
# Determine which columns to check
|
||||
cols_to_check: list[int] = []
|
||||
if target_col is not None:
|
||||
cols_to_check = [target_col]
|
||||
else:
|
||||
# Check all numeric columns in the total row
|
||||
cols_to_check = list(range(len(total_row)))
|
||||
|
||||
# Determine addend rows (all rows between header and total row)
|
||||
addend_rows = list(range(1, target_row))
|
||||
|
||||
for col in cols_to_check:
|
||||
stated = self._get_cell(total_row, col)
|
||||
if stated is None:
|
||||
continue
|
||||
|
||||
computed = Decimal(0)
|
||||
has_addends = False
|
||||
for r_idx in addend_rows:
|
||||
if r_idx >= len(rows):
|
||||
continue
|
||||
val = self._get_cell(rows[r_idx], col)
|
||||
if val is not None:
|
||||
computed += val
|
||||
has_addends = True
|
||||
|
||||
if not has_addends:
|
||||
continue
|
||||
|
||||
if abs(stated - computed) > self.tolerance:
|
||||
discrepancies.append(
|
||||
Discrepancy(
|
||||
page=page,
|
||||
kind=DiscrepancyKind.TALLY,
|
||||
severity=Severity.ERROR,
|
||||
description=f"{description}: column {col} — stated {stated}, expected {computed}",
|
||||
stated=str(stated),
|
||||
expected=str(computed),
|
||||
context=f"column {col}, total row {target_row}",
|
||||
)
|
||||
)
|
||||
|
||||
return discrepancies
|
||||
|
||||
def _check_single_cell(
|
||||
self,
|
||||
page: int,
|
||||
rows: list[list[str]],
|
||||
formula: str,
|
||||
description: str,
|
||||
target_row: int | None,
|
||||
target_col: int | None,
|
||||
) -> list[Discrepancy]:
|
||||
"""Verify a single cell formula (e.g. Grand Total = Subtotal + Tax)."""
|
||||
parts = formula.split("=", 1)
|
||||
if len(parts) != 2:
|
||||
return []
|
||||
|
||||
# Parse target from LHS cell(r,c) if not provided explicitly
|
||||
if target_row is None or target_col is None:
|
||||
lhs_match = re.match(r"cell\(\s*(\d+)\s*,\s*(\d+)\s*\)", parts[0].strip())
|
||||
if lhs_match:
|
||||
target_row = int(lhs_match.group(1))
|
||||
target_col = int(lhs_match.group(2))
|
||||
else:
|
||||
return []
|
||||
|
||||
if target_row >= len(rows):
|
||||
return []
|
||||
|
||||
rhs = parts[1].strip()
|
||||
|
||||
stated = self._get_cell(rows[target_row], target_col)
|
||||
if stated is None:
|
||||
return []
|
||||
|
||||
computed = self._eval_row_expr(rhs, rows[target_row], rows)
|
||||
if computed is None:
|
||||
return []
|
||||
|
||||
if abs(stated - computed) > self.tolerance:
|
||||
return [
|
||||
Discrepancy(
|
||||
page=page,
|
||||
kind=DiscrepancyKind.TALLY,
|
||||
severity=Severity.ERROR,
|
||||
description=f"{description}: stated {stated}, expected {computed}",
|
||||
stated=str(stated),
|
||||
expected=str(computed),
|
||||
context=f"cell({target_row},{target_col}), {formula}",
|
||||
)
|
||||
]
|
||||
return []
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Expression evaluation
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _eval_row_expr(self, expr: str, row: list[str], all_rows: list[list[str]]) -> Decimal | None:
|
||||
"""
|
||||
Evaluate an expression in the context of a specific row.
|
||||
Supports: colN refs, cell(r,c) refs, +, -, *, /
|
||||
Also supports: sum(colN, start-end)
|
||||
"""
|
||||
# Handle sum() function first
|
||||
sum_pattern = re.compile(r"sum\(\s*col(\d+)\s*,\s*(\d+)\s*-\s*(\d+)\s*\)")
|
||||
resolved = expr
|
||||
for match in sum_pattern.finditer(expr):
|
||||
col = int(match.group(1))
|
||||
start = int(match.group(2))
|
||||
end = int(match.group(3))
|
||||
total = Decimal(0)
|
||||
for r_idx in range(start, end + 1):
|
||||
if r_idx < len(all_rows):
|
||||
val = self._get_cell(all_rows[r_idx], col)
|
||||
if val is not None:
|
||||
total += val
|
||||
resolved = resolved.replace(match.group(0), str(total))
|
||||
|
||||
# Handle cell(r, c) references
|
||||
cell_pattern = re.compile(r"cell\(\s*(\d+)\s*,\s*(\d+)\s*\)")
|
||||
for match in cell_pattern.finditer(resolved):
|
||||
r = int(match.group(1))
|
||||
c = int(match.group(2))
|
||||
if r < len(all_rows):
|
||||
val = self._get_cell(all_rows[r], c)
|
||||
if val is not None:
|
||||
resolved = resolved.replace(match.group(0), str(val))
|
||||
else:
|
||||
return None
|
||||
else:
|
||||
return None
|
||||
|
||||
# Replace colN references with values from the current row.
|
||||
# Use re.sub with word boundaries to avoid col1 corrupting col12.
|
||||
_failed = False
|
||||
|
||||
def _col_replacer(m: re.Match[str]) -> str:
|
||||
nonlocal _failed
|
||||
col_idx = int(m.group(1))
|
||||
val = self._get_cell(row, col_idx)
|
||||
if val is None:
|
||||
_failed = True
|
||||
return m.group(0)
|
||||
return str(val)
|
||||
|
||||
resolved = re.sub(r"\bcol(\d+)\b", _col_replacer, resolved)
|
||||
if _failed:
|
||||
return None
|
||||
|
||||
# Evaluate the resulting arithmetic expression safely
|
||||
return self._safe_eval(resolved)
|
||||
|
||||
def _safe_eval(self, expr: str) -> Decimal | None:
|
||||
"""
|
||||
Evaluate a simple arithmetic expression containing only
|
||||
numbers and +, -, *, / operators. Respects standard operator
|
||||
precedence (* and / bind tighter than + and -). No eval().
|
||||
"""
|
||||
try:
|
||||
raw = re.findall(r"\d+(?:\.\d+)?|[+\-*/]", expr.strip())
|
||||
if not raw:
|
||||
return None
|
||||
|
||||
# Build (values, ops) lists, merging a leading '-' or an
|
||||
# operator-adjacent '-' into the next number token.
|
||||
values: list[Decimal] = []
|
||||
ops: list[str] = []
|
||||
i = 0
|
||||
while i < len(raw):
|
||||
tok = raw[i]
|
||||
if tok in "+-*/" and not values and tok == "-":
|
||||
# Leading negative: merge with next number
|
||||
i += 1
|
||||
if i >= len(raw):
|
||||
return None
|
||||
values.append(Decimal("-" + raw[i]))
|
||||
elif tok in "+-*/":
|
||||
# Operator followed by '-' → negative operand
|
||||
if (
|
||||
tok in "+-*/"
|
||||
and i + 1 < len(raw)
|
||||
and raw[i + 1] == "-"
|
||||
and i + 2 < len(raw)
|
||||
and raw[i + 2] not in "+-*/"
|
||||
):
|
||||
ops.append(tok)
|
||||
values.append(Decimal("-" + raw[i + 2]))
|
||||
i += 2 # skip the '-' and the number
|
||||
else:
|
||||
ops.append(tok)
|
||||
else:
|
||||
values.append(Decimal(tok))
|
||||
i += 1
|
||||
|
||||
if not values:
|
||||
return None
|
||||
|
||||
# Pass 1: evaluate * and /
|
||||
j = 0
|
||||
while j < len(ops):
|
||||
if ops[j] in ("*", "/"):
|
||||
if ops[j] == "*":
|
||||
values[j] = values[j] * values[j + 1]
|
||||
else:
|
||||
if values[j + 1] == 0:
|
||||
return None
|
||||
values[j] = values[j] / values[j + 1]
|
||||
values.pop(j + 1)
|
||||
ops.pop(j)
|
||||
else:
|
||||
j += 1
|
||||
|
||||
# Pass 2: evaluate + and -
|
||||
result = values[0]
|
||||
for j, op in enumerate(ops):
|
||||
if op == "+":
|
||||
result += values[j + 1]
|
||||
elif op == "-":
|
||||
result -= values[j + 1]
|
||||
|
||||
return result
|
||||
except (InvalidOperation, IndexError, ValueError):
|
||||
return None
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@staticmethod
|
||||
def _parse_col_ref(ref: str) -> int | None:
|
||||
match = re.match(r"col(\d+)", ref.strip())
|
||||
return int(match.group(1)) if match else None
|
||||
|
||||
@staticmethod
|
||||
def _get_cell(row: list[str], col: int) -> Decimal | None:
|
||||
if col >= len(row):
|
||||
return None
|
||||
return _to_decimal(row[col])
|
||||
@@ -21,8 +21,11 @@ from stirling.contracts import (
|
||||
PdfQuestionRequest,
|
||||
PdfQuestionResponse,
|
||||
SupportedCapability,
|
||||
ToolOperationStep,
|
||||
UnsupportedCapabilityResponse,
|
||||
)
|
||||
from stirling.contracts.pdf_edit import EditPlanResponse
|
||||
from stirling.models.agent_tool_models import AgentToolId, MathAuditorAgentParams
|
||||
from stirling.services import AppRuntime
|
||||
|
||||
|
||||
@@ -53,6 +56,14 @@ class OrchestratorAgent:
|
||||
name="delegate_user_spec",
|
||||
description="Delegate requests to create or revise a user agent spec and return the draft result.",
|
||||
),
|
||||
ToolOutput(
|
||||
self.math_auditor_agent,
|
||||
name="math_auditor_agent",
|
||||
description=(
|
||||
"Delegate requests to check arithmetic, validate table totals, "
|
||||
"audit financial calculations, or verify mathematical accuracy in PDFs."
|
||||
),
|
||||
),
|
||||
ToolOutput(
|
||||
self.unsupported_capability,
|
||||
name="unsupported_capability",
|
||||
@@ -66,6 +77,8 @@ class OrchestratorAgent:
|
||||
"Use delegate_pdf_edit for requested PDF modifications. "
|
||||
"Use delegate_pdf_question for questions about PDF contents. "
|
||||
"Use delegate_user_spec for requests to create or define an agent spec. "
|
||||
"Use math_auditor_agent for requests to check arithmetic, validate "
|
||||
"table totals, audit financial calculations, or verify math in PDFs. "
|
||||
"Use unsupported_capability only when none of the other outputs fit."
|
||||
),
|
||||
model_settings=runtime.fast_model_settings,
|
||||
@@ -93,6 +106,7 @@ class OrchestratorAgent:
|
||||
SupportedCapability.ORCHESTRATE
|
||||
| SupportedCapability.AGENT_REVISE
|
||||
| SupportedCapability.AGENT_NEXT_ACTION
|
||||
| SupportedCapability.MATH_AUDITOR_AGENT
|
||||
):
|
||||
raise ValueError(f"Cannot resume orchestrator with capability: {capability}")
|
||||
case _ as unreachable:
|
||||
@@ -123,6 +137,17 @@ class OrchestratorAgent:
|
||||
async def _run_agent_draft(self, request: OrchestratorRequest) -> AgentDraftWorkflowResponse:
|
||||
return await UserSpecAgent(self.runtime).draft(AgentDraftRequest(user_message=request.user_message))
|
||||
|
||||
async def math_auditor_agent(self, ctx: RunContext[OrchestratorDeps]) -> EditPlanResponse:
|
||||
return EditPlanResponse(
|
||||
summary="Validate mathematical calculations in the document.",
|
||||
steps=[
|
||||
ToolOperationStep(
|
||||
tool=AgentToolId.MATH_AUDITOR_AGENT,
|
||||
parameters=MathAuditorAgentParams(),
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
async def unsupported_capability(
|
||||
self,
|
||||
ctx: RunContext[OrchestratorDeps],
|
||||
|
||||
Reference in New Issue
Block a user