woogi/Conflict_Checker
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: coverage ladder + vision-unverified stamping (classic path)

Browse Source
This commit is contained in:
woogi
2026-08-18 13:35:21 -05:00
parent 23f6d7fe89
commit 48fefa4007
3 changed files with 246 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/pipeline/extractor.py
+112 -15
View File
@@ -21,9 +21,18 @@ from backend.llm import call_json
from backend.prompts import (
EXTRACTOR_SYSTEM_PROMPT,
EXTRACTOR_USER_INSTRUCTION,
TEXT_STRUCTURING_SYSTEM_PROMPT,
TEXT_STRUCTURING_USER_INSTRUCTION,
DISCIPLINE_PREFIXES,
ATTRIBUTE_VOCAB,
)
from backend.text_coverage import (
_norm,
fallback_objects,
merge_objects,
recover_sheet_number,
text_coverage,
)
# prefix (upper) -> discipline, longest-prefix-first for greedy matching
_PREFIX_TO_DISCIPLINE = sorted(
@@ -164,6 +173,8 @@ def _normalize_sheet(parsed: Dict, page_number: int,
clean: List[Dict] = []
dropped = 0
rescued = 0
unverified = 0
page_norm = _norm(page_text) if page_text else ""
for idx, obj in enumerate(raw_objects):
if not isinstance(obj, dict):
@@ -179,9 +190,19 @@ def _normalize_sheet(parsed: Dict, page_number: int,
page_text=page_text):
dropped += 1
continue
grounding = _grounding_stamp(primary_val, source_text, page_text)
if grounding:
# Pre-set stamps (fallback/merge rungs) win; otherwise compute the
# text-layer rescue stamp.
grounding = obj.get("grounding") or _grounding_stamp(
primary_val, source_text, page_text)
if grounding == "text_layer":
rescued += 1
if not grounding and page_text and source_text:
# Vision-unverified: survived the digit guard, but the quoted
# source_text is not present in the deterministic text layer.
# Kept and stamped - the wave-5b verifier prioritizes these.
if _norm(str(source_text)) not in page_norm:
grounding = "vision_unverified"
unverified += 1
# --- location_key: new schema is richer; map to legacy shape + extras ---
lk = obj.get("location_key")
@@ -236,10 +257,11 @@ def _normalize_sheet(parsed: Dict, page_number: int,
**({"grounding": grounding} if grounding else {}),
})
if dropped or rescued:
if dropped or rescued or unverified:
print(f"[Extract] Page {page_number} ({sheet_number}): "
f"dropped {dropped} ungrounded object(s)"
+ (f", rescued {rescued} via text layer" if rescued else ""))
+ (f", rescued {rescued} via text layer" if rescued else "")
+ (f", {unverified} vision-unverified" if unverified else ""))
unresolved = parsed.get("unresolved_items") or []
@@ -270,7 +292,25 @@ def _text_layer_block(page: Dict) -> str:
+ text[:config.TEXT_LAYER_MAX_CHARS])
def _text_structuring_extract(page: Dict, sheet_hint: str = ""):
"""Rung 2 of the extraction ladder: text-only structuring call (no
image). The text layer is authoritative for alphanumeric content - the
model segments it instead of transcribing pixels, so vision misreads
are impossible on this rung."""
instruction = (TEXT_STRUCTURING_USER_INSTRUCTION
.replace("{sheet_hint}", sheet_hint or "")
.replace("{text_layer}",
(page.get("text_layer") or "")
[:config.TEXT_LAYER_MAX_CHARS]))
return call_json(
system_prompt=TEXT_STRUCTURING_SYSTEM_PROMPT,
user_text=instruction,
max_tokens=config.EXTRACT_MAX_TOKENS,
)
def _extract_one(page: Dict, sheet_hint: str = "") -> Dict:
page_text = page.get("text_layer")
user_text = (EXTRACTOR_USER_INSTRUCTION.replace("{sheet_hint}", sheet_hint)
+ _text_layer_block(page))
parsed = call_json(
@@ -280,17 +320,74 @@ def _extract_one(page: Dict, sheet_hint: str = "") -> Dict:
max_tokens=config.EXTRACT_MAX_TOKENS,
)
if not isinstance(parsed, dict):
return {
"page_number": page["page_number"],
"sheet_number": None,
"discipline": "Unknown",
"sheet_title": f"Page {page['page_number']} (extraction failed)",
"level": None,
"scale": None,
"assertions": [],
}
return _normalize_sheet(parsed, page["page_number"],
page_text=page.get("text_layer"))
if not page_text:
# Scanned/raster page: vision-only, keep the legacy failure shape.
return {
"page_number": page["page_number"],
"sheet_number": None,
"discipline": "Unknown",
"sheet_title": f"Page {page['page_number']} (extraction failed)",
"level": None,
"scale": None,
"assertions": [],
}
# Text-bearing page: climb the ladder instead of going dark.
parsed = {"sheet": {}, "objects": []}
sheet = _normalize_sheet(parsed, page["page_number"], page_text=page_text)
cov = text_coverage(page_text or "", sheet["assertions"])
sheet["coverage"] = cov
# Rung 2: text-only structuring when coverage is below floor. MERGE,
# never replace - vision objects (graphical_basis content exists only
# in the image) are kept; the text pass fills what vision missed.
if (page_text and config.EXTRACT_TEXT_RETRY_ENABLED
and cov["ratio"] < config.EXTRACT_COVERAGE_FLOOR):
print(f"[Extract] Page {page['page_number']}: coverage "
f"{cov['ratio']:.0%} < floor - text-only structuring pass")
parsed2 = _text_structuring_extract(page, sheet_hint)
if isinstance(parsed2, dict):
sheet2 = _normalize_sheet(parsed2, page["page_number"],
page_text=page_text)
before = len(sheet["assertions"])
sheet["assertions"] = merge_objects(sheet["assertions"],
sheet2["assertions"])
for key in ("sheet_number", "sheet_title", "discipline",
"level", "scale", "drawing_type"):
if not sheet.get(key) and sheet2.get(key):
sheet[key] = sheet2[key]
cov = text_coverage(page_text, sheet["assertions"])
sheet["coverage"] = cov
print(f"[Extract] Page {page['page_number']}: merged "
f"{len(sheet['assertions']) - before} text-structured "
f"object(s), coverage now {cov['ratio']:.0%}")
# Rung 3: deterministic fallback - a dark text-bearing sheet is
# impossible. Stubs are deduped against earlier rungs.
if (page_text and config.EXTRACT_FALLBACK_ENABLED
and cov["ratio"] < config.EXTRACT_COVERAGE_FLOOR):
stubs = fallback_objects(page_text, page["page_number"],
config.EXTRACT_FALLBACK_MAX_OBJECTS)
stubs = _normalize_sheet({"sheet": {}, "objects": stubs},
page["page_number"],
page_text=page_text)["assertions"]
before = len(sheet["assertions"])
sheet["assertions"] = merge_objects(sheet["assertions"], stubs)
print(f"[Extract] Page {page['page_number']}: fallback merged "
f"{len(sheet['assertions']) - before} text-layer stub(s)")
sheet["coverage"] = text_coverage(page_text, sheet["assertions"])
# Identity recovery: never leave a text-bearing page sheet-less.
if not sheet.get("sheet_number") and page_text:
recovered = recover_sheet_number(page_text)
if recovered:
sheet["sheet_number"] = recovered
sheet["discipline"] = (discipline_from_sheet_number(recovered)
or sheet.get("discipline") or "Unknown")
print(f"[Extract] Page {page['page_number']}: sheet number "
f"recovered from text layer -> {recovered}")
return sheet
def extract_assertions(pages: List[Dict], on_progress=None) -> List[Dict]: