diff --git a/apps/worker/app/services/document_agent/calibration/phase1.py b/apps/worker/app/services/document_agent/calibration/phase1.py index caff9f87..a651c90d 100644 --- a/apps/worker/app/services/document_agent/calibration/phase1.py +++ b/apps/worker/app/services/document_agent/calibration/phase1.py @@ -167,7 +167,7 @@ def run_calibration_phase1( ) break - inspect_calls = sum(len(scan.rounds) for scan in scans) + inspect_calls = sum(len(scan.scanned_pages) for scan in scans) logger.info( "[calibration.phase1] region={} regimes={} offsets={} inspect_calls={}", region_index, diff --git a/apps/worker/app/services/document_agent/calibration/prompts.py b/apps/worker/app/services/document_agent/calibration/prompts.py index 12391088..a46cc5c7 100644 --- a/apps/worker/app/services/document_agent/calibration/prompts.py +++ b/apps/worker/app/services/document_agent/calibration/prompts.py @@ -4,19 +4,57 @@ from typing import Any +# Multi-page batch verify (null-page / Phase-2 helpers). SECTION_START_ANSWER_KEYS = { - "found": "boolean, true only when the section heading starts on one of these pages", + "found": ( + "boolean, true when that section starts on one of these pages " + "(exact heading or accepted variant)" + ), "found_page": "number|null, the physical page number where it starts", } +# Single-page concurrent scan (calibration Phase-1 forward scan). +SECTION_START_PAGE_ANSWER_KEYS = { + "found": ( + "boolean, true when THIS page's main heading starts the given section " + "(exact or accepted variant)" + ), + "reason": "string, at most 20 Chinese characters explaining the decision", +} + + +def _variant_rules() -> str: + return ( + "Count as a match when either:\n" + "(1) Exact: the page's main heading matches the given title as written; or\n" + "(2) Variant: the page's main heading is clearly the same section, " + "even if a number/letter prefix differs or a document-id/code suffix " + "appears on only one side " + "(e.g. given title '3.2 Foo' ↔ page heading 'Foo'; " + "given title 'Foo (DOC-12)' ↔ page heading 'Foo'; " + "given title 'A.1 Foo Bar' ↔ page heading 'Foo Bar').\n" + "Do NOT count: a contents-list line, a running header/footer, or " + "a passing mention inside ordinary body paragraphs." + ) + def build_section_start_question(title: str) -> str: """Ask whether ``title`` starts as a body heading on the provided pages.""" return ( - f"Does the section titled {title!r} START on one of these pages, as a " - "body heading? A table-of-contents line, a running header or footer, or " - "a passing mention in body text does not count. Report the physical page " - "number printed in the page label above each image." + f"Does the section titled {title!r} START on one of these pages as a " + "body heading (where that section begins)?\n" + f"{_variant_rules()}\n" + "Report the physical page number from the page label above each image." + ) + + +def build_section_start_page_question(title: str) -> str: + """Ask whether ``title`` starts as a body heading on this single page.""" + return ( + f"Does the section titled {title!r} START on THIS page as a " + "body heading (where that section begins)?\n" + f"{_variant_rules()}\n" + "Return found true or false, and a reason of at most 20 Chinese characters." ) @@ -36,3 +74,7 @@ def coerce_found_page(value: Any, *, pages: list[int]) -> int | None: except (TypeError, ValueError): return None return page if page in pages else None + + +def coerce_reason(value: Any) -> str: + return str(value or "").strip().replace("\n", " ") diff --git a/apps/worker/app/services/document_agent/calibration/scan.py b/apps/worker/app/services/document_agent/calibration/scan.py index 5cf0399e..7b773aa1 100644 --- a/apps/worker/app/services/document_agent/calibration/scan.py +++ b/apps/worker/app/services/document_agent/calibration/scan.py @@ -4,20 +4,24 @@ some page after it. The scan walks forward from the candidate with a widening window, feeding each round's cursor into the next one, so a miss never re-opens pages that were already inspected. + +Each round still covers ``window_schedule[i]`` pages, but pages are inspected +one-at-a-time concurrently (never batched into a single VLM call). """ from __future__ import annotations +from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from typing import Any from loguru import logger from app.services.document_agent.calibration.prompts import ( - SECTION_START_ANSWER_KEYS, - build_section_start_question, + SECTION_START_PAGE_ANSWER_KEYS, + build_section_start_page_question, coerce_found, - coerce_found_page, + coerce_reason, ) from app.services.document_agent.manifest import ToolContext from app.services.document_agent.tools.inspect_pages import inspect_pages @@ -46,12 +50,21 @@ def progressive_page_windows( return windows +@dataclass +class PageInspectResult: + page: int + found: bool + reason: str = "" + error: str = "" + + @dataclass class ScanRound: pages: list[int] found: bool found_page: int | None = None error: str = "" + page_results: list[PageInspectResult] = field(default_factory=list) @dataclass @@ -78,12 +91,78 @@ def to_dict(self) -> dict[str, Any]: "found": item.found, "found_page": item.found_page, "error": item.error, + "page_results": [ + { + "page": pr.page, + "found": pr.found, + "reason": pr.reason, + "error": pr.error, + } + for pr in item.page_results + ], } for item in self.rounds ], } +def _inspect_one_page( + *, + ctx: ToolContext, + title: str, + page: int, +) -> PageInspectResult: + result = inspect_pages( + ctx, + { + "pages": [page], + "page_cap": 1, + "question": build_section_start_page_question(title), + "answer_keys": SECTION_START_PAGE_ANSWER_KEYS, + "folder_name": "calibration_scan", + "prefix": "scan", + "usage_task": "calibration.scan_title_forward", + }, + ) + if result.status != "ok": + return PageInspectResult( + page=page, found=False, error=result.error or "inspect.pages failed" + ) + fields = (result.payload or {}).get("fields") or {} + return PageInspectResult( + page=page, + found=coerce_found(fields.get("found")), + reason=coerce_reason(fields.get("reason")), + ) + + +def _inspect_pages_concurrent( + *, + ctx: ToolContext, + title: str, + pages: list[int], +) -> list[PageInspectResult]: + """Inspect each page alone; keep page order in the returned list.""" + if len(pages) == 1: + return [_inspect_one_page(ctx=ctx, title=title, page=pages[0])] + + by_page: dict[int, PageInspectResult] = {} + with ThreadPoolExecutor(max_workers=len(pages)) as pool: + futures = { + pool.submit(_inspect_one_page, ctx=ctx, title=title, page=page): page + for page in pages + } + for future in as_completed(futures): + page = futures[future] + try: + by_page[page] = future.result() + except Exception as exc: # noqa: BLE001 — surface as page error + by_page[page] = PageInspectResult( + page=page, found=False, error=str(exc) + ) + return [by_page[page] for page in pages] + + def scan_title_forward( *, ctx: ToolContext, @@ -94,8 +173,10 @@ def scan_title_forward( ) -> TitleScanResult: """Scan forward from ``start_page`` until the title is found or rounds run out. - Each round opens ``window_schedule[i]`` consecutive pages starting at the - cursor left by the previous round, so no page is inspected twice. + Each round covers ``window_schedule[i]`` consecutive pages starting at the + cursor left by the previous round. Pages inside a round are inspected + concurrently, one page per VLM call; the earliest true page wins. A page + error without a hit is logged and the scan continues to the next window. """ scanned: list[int] = [] rounds: list[ScanRound] = [] @@ -106,36 +187,21 @@ def scan_title_forward( end_page=page_count, window_schedule=window_schedule, ): - result = inspect_pages( - ctx, - { - "pages": pages, - "page_cap": len(pages), - "question": build_section_start_question(title), - "answer_keys": SECTION_START_ANSWER_KEYS, - "folder_name": "calibration_scan", - "prefix": "scan", - "usage_task": "calibration.scan_title_forward", - }, - ) + page_results = _inspect_pages_concurrent(ctx=ctx, title=title, pages=pages) next_start = pages[-1] + 1 scanned.extend(pages) - if result.status != "ok": - rounds.append(ScanRound(pages=pages, found=False, error=result.error or "")) - logger.warning( - "[calibration.scan] title={!r} pages={} inspect failed: {}", - title, - pages, - result.error, + hits = [pr.page for pr in page_results if pr.found] + if hits: + found_page = min(hits) + rounds.append( + ScanRound( + pages=pages, + found=True, + found_page=found_page, + page_results=page_results, + ) ) - break - - fields = (result.payload or {}).get("fields") or {} - found_page = coerce_found_page(fields.get("found_page"), pages=pages) - found = coerce_found(fields.get("found")) and found_page is not None - rounds.append(ScanRound(pages=pages, found=found, found_page=found_page)) - if found: logger.info( "[calibration.scan] title={!r} found on page={} after {} round(s)", title, @@ -151,6 +217,25 @@ def scan_title_forward( rounds=rounds, ) + errors = [pr.error for pr in page_results if pr.error] + if errors: + logger.warning( + "[calibration.scan] title={!r} pages={} inspect failed: {}; " + "continuing to next window", + title, + pages, + errors[0], + ) + rounds.append( + ScanRound( + pages=pages, + found=False, + found_page=None, + error=errors[0] if errors else "", + page_results=page_results, + ) + ) + logger.info( "[calibration.scan] title={!r} not found in pages={}", title, diff --git a/apps/worker/app/services/document_agent/structure/null_page_react.py b/apps/worker/app/services/document_agent/structure/null_page_react.py index 29ed7b16..e4a4fd16 100644 --- a/apps/worker/app/services/document_agent/structure/null_page_react.py +++ b/apps/worker/app/services/document_agent/structure/null_page_react.py @@ -30,56 +30,34 @@ def react_budget() -> int: _REACT_INSTRUCTIONS = """\ -You are the search planner in a small ReAct loop. Propose the next whole-line -text query to find the physical START page of a section. Grep normalizes each -PDF text line independently, collapses whitespace with CJK-aware spacing, and -matches the complete normalized line case-insensitively. It never accepts a -substring inside a longer line. Every candidate page, including a unique one, -must pass visual section-start confirmation. - -The system already grepped the full TOC title once before this loop (see -previous_attempts). Do not repeat that exact full-title query. - -Return one strict json object with action one of (no other keys): -{"action":"grep","query":"..."} -{"action":"strip_header","query":""} -{"action":"strip_footer","query":""} -{"action":"give_up","query":""} - -Do not include a reason field. Use give_up only when no useful untried query -or strip remains. - -Ordered query strategy after the automatic full-title grep (follow this order; -skip a step only if already tried or not applicable to the TOC title / parent -path). Pattern-level only — do not invent document-specific titles: -1. Remove the leading number / letter / punctuation prefix from the TOC title - and grep the remaining title body. -2. When the parent path indicates appendices/annexes (or the TOC label is a +Return exactly one strict json object and no other text. +Fields: action is one of grep, strip_header, strip_footer, give_up; query is a string. + +Ordered query strategy (follow this order; skip a step only if already tried +or not applicable to the given title / parent path). Pattern-level only — +do not invent document-specific titles: +1. Derive the search line from the given title by removing leading number / + letter / punctuation prefixes and trailing metadata qualifiers (document + identifiers/codes, revision labels, and similar). Keep the semantic title + body. Prefer that body over a metadata-only query when both are present. +2. When the parent path indicates appendices/annexes (or the title is a lettered appendix-style entry): grep the structural form - "Appendix " using the letter taken from the TOC label. Prefer this + "Appendix " using the letter taken from the title. Prefer this before inventing other phrases. 3. Only after the above: try other variants such as "Appendix " plus the title body, a shorter distinctive complete-line title variant, or another structural prefix (chapter / part / section / annex) when supported by the title or parent path. -4. Prefer queries specific enough to avoid running headers and passing mentions. - Do not guess page numbers. -Reflection rules (mandatory): -- Read previous_attempts. Reflect on query, hit_page_count, and observation - before answering. +Rules: +- Prefer queries specific enough to avoid running headers and passing mentions. + Do not guess page numbers. - If the last observation is no_line_hits, visual_rejected, empty_normalized_query, grep_tool_error, or duplicate_normalized_query, you - MUST change the query when choosing grep. Emitting the same grep query again - after whitespace/case normalization is invalid for planner-chosen greps. -- Prefer strip_header or strip_footer when candidate pages look like running - headers/footers. Each strip automatically re-greps the last query once for - free. Otherwise advance to the next ordered query strategy. -- strip_header / strip_footer only update a temporary search view; they do not - change stored page text. They do NOT consume react_budget. Call each at most - once per locate. -- Planner greps consume react_budget. The automatic full-title seed grep and - strip auto re-greps are free. + must choose a different normalized query when using grep. +- Use strip_header or strip_footer at most once each when margin text causes + false hits; either action re-runs the last query. +- Use give_up only when no useful untried query or strip action remains. """ @@ -195,7 +173,7 @@ def _propose_react_query( grep_loops_used: int, ) -> tuple[dict[str, Any] | None, dict[str, Any]]: state = { - "toc_title": title, + "title": title, "parent_path": list(parent_titles), "physical_search_scope": [left, right], "grep_loops_remaining": max(0, budget - grep_loops_used), diff --git a/apps/worker/tests/contract/test_calibration_scan_contract.py b/apps/worker/tests/contract/test_calibration_scan_contract.py index d56cfd69..34a5aa4d 100644 --- a/apps/worker/tests/contract/test_calibration_scan_contract.py +++ b/apps/worker/tests/contract/test_calibration_scan_contract.py @@ -54,7 +54,7 @@ def __call__(self, ctx: ToolContext, args: dict[str, Any]) -> ToolResult: "answer": "", "fields": { "found": hit, - "found_page": self.hit_page if hit else None, + "reason": "hit" if hit else "miss", }, }, ) @@ -84,7 +84,8 @@ def test_first_round_opens_the_candidate_page_and_its_successor(patch_inspect) - ctx=_ctx(), title="Chapter 1", start_page=10, page_count=60 ) - assert fake.calls == [[10, 11]] + assert sorted(fake.calls) == [[10], [11]] + assert all(len(call) == 1 for call in fake.calls) assert result.found is True assert result.found_page == 10 @@ -96,12 +97,8 @@ def test_miss_expands_forward_from_the_cursor_without_rescanning(patch_inspect) ctx=_ctx(), title="Chapter 1", start_page=10, page_count=60 ) - assert fake.calls == [ - [10, 11], - [12, 13, 14, 15], - [16, 17, 18, 19, 20, 21], - [22, 23, 24, 25, 26, 27, 28, 29, 30, 31], - ] + assert sorted(fake.calls) == [[page] for page in range(10, 32)] + assert all(len(call) == 1 for call in fake.calls) assert result.found is False assert result.scanned_pages == list(range(10, 32)) assert len(result.scanned_pages) == len(set(result.scanned_pages)) @@ -115,7 +112,7 @@ def test_scan_stops_at_first_hit(patch_inspect) -> None: ctx=_ctx(), title="Chapter 1", start_page=10, page_count=60 ) - assert fake.calls == [[10, 11], [12, 13, 14, 15]] + assert sorted(fake.calls) == [[10], [11], [12], [13], [14], [15]] assert result.found_page == 13 assert result.next_start == 16 @@ -127,7 +124,7 @@ def test_scan_covers_at_most_the_window_schedule(patch_inspect) -> None: ctx=_ctx(), title="Chapter 1", start_page=10, page_count=60 ) - assert len(fake.calls) == len(DEFAULT_WINDOW_SCHEDULE) + assert len(fake.calls) == sum(DEFAULT_WINDOW_SCHEDULE) assert len(result.scanned_pages) == sum(DEFAULT_WINDOW_SCHEDULE) @@ -138,11 +135,11 @@ def test_window_is_clipped_at_the_last_page(patch_inspect) -> None: ctx=_ctx(page_count=13), title="Appendix", start_page=10, page_count=13 ) - assert fake.calls == [[10, 11], [12, 13]] + assert sorted(fake.calls) == [[10], [11], [12], [13]] assert result.next_start is None -def test_each_call_lifts_the_page_cap_to_its_own_window(patch_inspect) -> None: +def test_each_call_is_single_page_with_page_cap_one(patch_inspect) -> None: class _Recording(_FakeInspect): def __init__(self) -> None: super().__init__(hit_page=None) @@ -156,24 +153,41 @@ def __call__(self, ctx: ToolContext, args: dict[str, Any]) -> ToolResult: scan_title_forward(ctx=_ctx(), title="Chapter 1", start_page=10, page_count=60) - assert fake.caps == list(DEFAULT_WINDOW_SCHEDULE) + assert fake.caps == [1] * sum(DEFAULT_WINDOW_SCHEDULE) + assert all(len(call) == 1 for call in fake.calls) -def test_inspect_error_aborts_the_scan(patch_inspect) -> None: - class _Failing(_FakeInspect): +def test_inspect_error_continues_to_later_windows(patch_inspect) -> None: + class _FailThenHit(_FakeInspect): def __call__(self, ctx: ToolContext, args: dict[str, Any]) -> ToolResult: - self.calls.append(list(args.get("pages") or [])) - return ToolResult(status="error", error="calibration visual budget exhausted") + pages = list(args.get("pages") or []) + self.calls.append(pages) + page = pages[0] if pages else None + if page is not None and page <= 11: + return ToolResult( + status="error", error="calibration visual budget exhausted" + ) + hit = page == 13 + return ToolResult( + status="ok", + payload={ + "fields": { + "found": hit, + "reason": "hit" if hit else "miss", + } + }, + ) - fake = patch_inspect(_Failing(hit_page=None)) + fake = patch_inspect(_FailThenHit(hit_page=None)) result = scan_title_forward( ctx=_ctx(), title="Chapter 1", start_page=10, page_count=60 ) - assert fake.calls == [[10, 11]] - assert result.found is False - assert result.rounds[-1].error == "calibration visual budget exhausted" + assert sorted(fake.calls) == [[10], [11], [12], [13], [14], [15]] + assert result.found is True + assert result.found_page == 13 + assert result.rounds[0].error == "calibration visual budget exhausted" def test_string_false_is_not_treated_as_a_hit(patch_inspect) -> None: @@ -186,7 +200,7 @@ def __call__(self, ctx: ToolContext, args: dict[str, Any]) -> ToolResult: payload={ "fields": { "found": "false", - "found_page": pages[0] if pages else None, + "reason": "no", } }, ) @@ -199,24 +213,31 @@ def __call__(self, ctx: ToolContext, args: dict[str, Any]) -> ToolResult: assert result.found is False assert result.found_page is None - assert len(fake.calls) == len(DEFAULT_WINDOW_SCHEDULE) + assert len(fake.calls) == sum(DEFAULT_WINDOW_SCHEDULE) -def test_found_page_outside_the_window_is_rejected(patch_inspect) -> None: - class _Liar(_FakeInspect): +def test_earliest_true_page_wins_within_a_round(patch_inspect) -> None: + class _MultiHit(_FakeInspect): def __call__(self, ctx: ToolContext, args: dict[str, Any]) -> ToolResult: pages = list(args.get("pages") or []) self.calls.append(pages) + page = pages[0] if pages else None + hit = page in {13, 14} return ToolResult( status="ok", - payload={"fields": {"found": True, "found_page": 999}}, + payload={ + "fields": { + "found": hit, + "reason": "hit" if hit else "miss", + } + }, ) - patch_inspect(_Liar(hit_page=None)) + fake = patch_inspect(_MultiHit(hit_page=None)) result = scan_title_forward( ctx=_ctx(), title="Chapter 1", start_page=10, page_count=60 ) - assert result.found is False - assert result.found_page is None + assert result.found_page == 13 + assert sorted(fake.calls) == [[10], [11], [12], [13], [14], [15]]