Skip to content

fix: page-memory fine-hierarchy boundary trim and null-page parent locate #220

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
EricNGOntos merged 5 commits into from
Jul 20, 2026
Merged

fix: page-memory fine-hierarchy boundary trim and null-page parent locate #220

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
834f8d2
refactor: remove residual page location agent and related tools
EricNGOntos Jul 20, 2026
1b05f79
feat: null-page TOC parent locate and VLM-order skeleton sort
EricNGOntos Jul 20, 2026
27ebca2
Merge remote-tracking branch 'origin/main' into develop
EricNGOntos Jul 20, 2026
242b6de
chore: drop stale null-page TODO and duplicate debug import after merge
EricNGOntos Jul 20, 2026
f409c4d
fix: drop fine-hierarchy boundary page when next-title anchor misses
EricNGOntos Jul 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
547 changes: 141 additions & 406 deletions apps/worker/app/services/document_agent/structure/hierarchy_locator.py
View file
Open in desktop

Large diffs are not rendered by default.

242 changes: 8 additions & 234 deletions apps/worker/app/services/document_agent/structure/page_locate_agent.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,31 +1,23 @@
"""Residual page-location agent for hierarchy titles."""
"""VLM page verification for page-memory offset calibration.

This module provides the deterministic-input, VLM-arbitrated helper used by the
page-memory skeleton calibration to confirm which candidate page starts a given
section title. The former residual ReAct sub-agent has been removed; calibration
now drives offset-guided bulk anchoring directly and only needs this verifier.
"""

from __future__ import annotations

import base64
import json
import os
import time
from dataclasses import dataclass, field
from typing import Any, cast

from loguru import logger

from app.services.document_agent.manifest import ToolContext
from app.services.document_agent.structure.hierarchy_locator import (
TitleMatch,
TitleNode,
collect_title_candidate_matches,
iter_leaf_title_nodes,
locate_title_strict_exact,
max_title_depth,
prune_title_nodes_for_emit_depth,
)
from app.services.document_agent.structure.page_locate_subagent import (
PageLocateSubAgent,
SubAgentConfig,
)
from shared.models.schemas.page_memory_config import PageMemoryConfig
from app.services.document_agent.structure.hierarchy_locator import TitleMatch

VLM_CONFIRMED_DEFAULT_CONFIDENCE = 0.75
GREP_ONLY_CONFIDENCE_CAP = 0.62
Expand All @@ -34,224 +26,6 @@
VLM_FAILED_GREP_CONFIDENCE_CAP = 0.54


@dataclass(frozen=True)
class PageLocateConfig:
residual_agent_limit: int = 50
max_emit_depth: int = 5
min_emit_depth: int = 2
vlm_candidate_page_cap: int = 4
full_leaf_sections: bool = False

@classmethod
def from_page_memory_config(
cls,
page_memory_config: PageMemoryConfig,
) -> "PageLocateConfig":
return cls(
residual_agent_limit=page_memory_config.page_locate_residual_agent_limit,
max_emit_depth=page_memory_config.page_locate_max_emit_depth,
min_emit_depth=page_memory_config.page_locate_min_emit_depth,
vlm_candidate_page_cap=(
page_memory_config.page_locate_vlm_candidate_page_cap
),
full_leaf_sections=page_memory_config.page_locate_full_leaf_sections,
)


@dataclass(frozen=True)
class ResidualRequest:
path_titles: tuple[str, ...]
node: TitleNode


@dataclass(frozen=True)
class PageLocatePrepareResult:
nodes: list[TitleNode]
match_overrides: dict[tuple[str, ...], TitleMatch]
summary: dict[str, Any] = field(default_factory=dict)


class PageLocateResidualAgent:
"""Batch residual resolver for C4 title-to-page anchoring."""

def __init__(
self,
*,
ctx: ToolContext | None,
page_texts: dict[int, str],
body_pages: list[int],
page_count: int,
page_offset_hint: int | None,
config: PageLocateConfig | None = None,
) -> None:
self.ctx = ctx
self.page_texts = page_texts
self.body_pages = sorted({page for page in body_pages if 1 <= page <= page_count})
self.page_count = page_count
self.page_offset_hint = page_offset_hint
self.config = config or PageLocateConfig()

def prepare(self, nodes: list[TitleNode]) -> PageLocatePrepareResult:
if not nodes:
return PageLocatePrepareResult(nodes=[], match_overrides={}, summary={})

actual_max_depth = max_title_depth(nodes)
emit_depth = min(actual_max_depth, self.config.max_emit_depth)
emit_depth = max(emit_depth, self.config.min_emit_depth)
direct_matches: dict[tuple[str, ...], TitleMatch] = {}
residuals: list[ResidualRequest] = []

while True:
selected_nodes = prune_title_nodes_for_emit_depth(nodes, emit_depth=emit_depth)
direct_matches, residuals = self._classify_residuals(selected_nodes)
if (
self.config.full_leaf_sections
or len(residuals) <= self.config.residual_agent_limit
or emit_depth <= self.config.min_emit_depth
):
break
emit_depth -= 1

residual_matches = self._resolve_residuals(residuals)
match_overrides = {**direct_matches, **residual_matches}
summary = {
"agent": "page_locate_residual",
"emit_depth": emit_depth,
"actual_max_depth": actual_max_depth,
"direct_exact_count": len(direct_matches),
"residual_count": len(residuals),
"resolved_residual_count": len(residual_matches),
"residual_limit": self.config.residual_agent_limit,
"vlm_candidate_page_cap": self.config.vlm_candidate_page_cap,
"full_leaf_sections": self.config.full_leaf_sections,
}
logger.info("[page_locate.agent] summary={}", summary)
return PageLocatePrepareResult(
nodes=selected_nodes,
match_overrides=match_overrides,
summary=summary,
)

def _entry_scope(self, node: TitleNode) -> list[int]:
"""Narrow search scope for a node using its printed page + offset hint.

When a node carries a printed_page and we have a page_offset_hint,
restrict grep to a small window around the expected physical page.
The window spans [printed_page, printed_page + offset] (inclusive,
order-independent) to cover the uncertainty between printed numbering
and physical page positions.
"""
if node.printed_page is None or self.page_offset_hint is None:
return self.body_pages
expected = node.printed_page + self.page_offset_hint
lo = min(node.printed_page, expected)
hi = max(node.printed_page, expected)
return [p for p in self.body_pages if lo <= p <= hi]

def _classify_residuals(
self,
nodes: list[TitleNode],
) -> tuple[dict[tuple[str, ...], TitleMatch], list[ResidualRequest]]:
direct: dict[tuple[str, ...], TitleMatch] = {}
residuals: list[ResidualRequest] = []
for path_titles, node in iter_leaf_title_nodes(nodes):
scope = self._entry_scope(node)
match = locate_title_strict_exact(
node.title,
scope_pages=scope,
page_texts=self.page_texts,
)
if match is not None:
direct[path_titles] = TitleMatch(
page=match.page,
confidence=match.confidence,
source=match.source,
matched_line=match.matched_line,
score=match.score,
candidates=match.candidates,
evidence={
**match.evidence,
"page_locate_agent": {
"decision": "direct_strict_exact",
"path_titles": list(path_titles),
},
},
)
else:
residuals.append(ResidualRequest(path_titles=path_titles, node=node))
return direct, residuals

def _resolve_residuals(
self,
residuals: list[ResidualRequest],
) -> dict[tuple[str, ...], TitleMatch]:
matches: dict[tuple[str, ...], TitleMatch] = {}
if not residuals or self.ctx is None:
# No agent runtime (ctx/budget) → leave residuals for physical-hint /
# neighbor-boundary fallback in the resolver. The residual sub-agent
# only runs when a real ToolContext is available.
return matches

sub_config = SubAgentConfig(
candidate_cap=max(self.config.vlm_candidate_page_cap * 2, 4),
verify_page_cap=max(self.config.vlm_candidate_page_cap, 1),
)
for residual in residuals:
scope = self._entry_scope(residual.node)
agent = PageLocateSubAgent(
ctx=self.ctx,
scope_pages=scope,
page_count=self.page_count,
config=sub_config,
page_offset_hint=self.page_offset_hint,
)
result = agent.locate(
title=residual.node.title,
printed_page=residual.node.printed_page,
)
if result.match is None:
logger.warning(
"[page_locate.agent] unresolved title={!r} path_titles={} stop={}",
residual.node.title,
residual.path_titles,
result.stop_reason,
)
continue
base = result.match
agent_evidence = dict(base.evidence.get("page_locate_agent", {}))
agent_evidence["path_titles"] = list(residual.path_titles)
agent_evidence["stop_reason"] = result.stop_reason
matches[residual.path_titles] = TitleMatch(
page=base.page,
confidence=base.confidence,
source=base.source,
matched_line=base.matched_line,
score=base.score,
candidates=base.candidates,
evidence={**base.evidence, "page_locate_agent": agent_evidence},
)
return matches


def grep_title_page_candidates(
*,
title: str,
scope_pages: list[int],
page_texts: dict[int, str],
printed_page: int | None = None,
page_offset_hint: int | None = None,
limit: int | None = None,
) -> list[TitleMatch]:
return collect_title_candidate_matches(
title,
scope_pages=scope_pages,
page_texts=page_texts,
printed_page=printed_page,
page_offset_hint=page_offset_hint,
limit=limit,
)


def verify_section_page_choice(
*,
ctx: ToolContext | None,
Expand Down
Loading
Loading