From 2c21926444fef4d4187d8551d89d1551b3bb5fce Mon Sep 17 00:00:00 2001 From: Research Assistant Date: Fri, 5 Jun 2026 12:22:15 +0800 Subject: [PATCH] feat: add OCR evidence-aware query routing for phase5 - Add evidence_query class to build_query_plan() for intent='evidence' - Add ocr_evidence_available to enrich_query_plan_with_runtime() - Add OCR evidence enrichment in search.py and retrieve.py - Add routing contract tests (3 routing + 1 evidence hit) - All pre-existing tests pass --- paperforge/commands/retrieve.py | 57 ++++++-- paperforge/commands/search.py | 17 ++- paperforge/query_planning.py | 165 ++++++++++++++++++----- tests/test_retrieve_evidence.py | 19 +++ tests/test_search_command_diagnostics.py | 29 ++++ 5 files changed, 240 insertions(+), 47 deletions(-) create mode 100644 tests/test_retrieve_evidence.py create mode 100644 tests/test_search_command_diagnostics.py diff --git a/paperforge/commands/retrieve.py b/paperforge/commands/retrieve.py index 92bad714..63bcd058 100644 --- a/paperforge/commands/retrieve.py +++ b/paperforge/commands/retrieve.py @@ -15,7 +15,16 @@ def _looks_generic_chunk(text: str) -> bool: compact = (text or "").strip().lower() if not compact: return True - if compact in {"[figure]", "none.", "or", "### keywords", "### abbreviations", "### reference", "### conclusion", "### references"}: + if compact in { + "[figure]", + "none.", + "or", + "### keywords", + "### abbreviations", + "### reference", + "### conclusion", + "### references", + }: return True if len(compact) <= 12: return True @@ -37,6 +46,7 @@ def run(args: argparse.Namespace) -> int: # Check if vector index exists from paperforge.embedding import get_embed_status + status = get_embed_status(vault) if not status.get("healthy", True): plan = enrich_query_plan_with_runtime(build_query_plan(query, "content"), vault) @@ -88,8 +98,12 @@ def run(args: argparse.Namespace) -> int: try: chunks = retrieve_chunks(vault, query, limit=limit, expand=args.expand) except Exception as e: - result = PFResult(ok=False, command="retrieve", version=PF_VERSION, - error=PFError(code=ErrorCode.INTERNAL_ERROR, message=str(e))) + result = PFResult( + ok=False, + command="retrieve", + version=PF_VERSION, + error=PFError(code=ErrorCode.INTERNAL_ERROR, message=str(e)), + ) print(result.to_json() if args.json else result.error.message, file=sys.stderr if not args.json else sys.stdout) return 1 @@ -102,7 +116,7 @@ def run(args: argparse.Namespace) -> int: for c in chunks: row = conn.execute( "SELECT citation_key, title, year, first_author FROM papers WHERE zotero_key=?", - (c["paper_id"],) + (c["paper_id"],), ).fetchone() if row: c["citation_key"] = row["citation_key"] @@ -122,13 +136,34 @@ def run(args: argparse.Namespace) -> int: "interactive_fallback_required": plan.get("interactive_fallback_required", False), "suggested_modes": plan.get("suggested_modes", []), } - warnings.append("Semantic retrieval returned no chunks. This does not prove the content is absent from the library.") + warnings.append( + "Semantic retrieval returned no chunks. This does not prove the content is absent from the library." + ) next_actions.append( { "command": "paperforge query-plan", "reason": "Review the fallback modes for content lookup and fulltext verification.", } ) + try: + ocr_root = vault / "System" / "PaperForge" / "ocr" + ocr_papers = 0 + if ocr_root.exists(): + ocr_papers = sum( + 1 + for paper_dir in ocr_root.iterdir() + if paper_dir.is_dir() and (paper_dir / "index" / "role-index.json").exists() + ) + data["ocr_evidence_available"] = ocr_papers + if ocr_papers > 0: + next_actions.append( + { + "command": "paperforge search", + "reason": f"OCR evidence available for {ocr_papers} paper(s)", + } + ) + except Exception: + pass elif _is_low_confidence_semantic_result(chunks): plan = enrich_query_plan_with_runtime(build_query_plan(query, "content"), vault) data["query_diagnostic"] = { @@ -137,19 +172,25 @@ def run(args: argparse.Namespace) -> int: "suggested_modes": plan.get("suggested_modes", []), "reason": "Top semantic hits look generic or weakly related to the query.", } - warnings.append("Semantic retrieval returned low-confidence hits. Verify with fulltext grep or narrow the scope before treating these as evidence.") + warnings.append( + "Semantic retrieval returned low-confidence hits. Verify with fulltext grep or narrow the scope before treating these as evidence." + ) next_actions.append( { "command": "paperforge query-plan", "reason": "Inspect fallback modes for exact fulltext verification.", } ) - result = PFResult(ok=True, command="retrieve", version=PF_VERSION, data=data, warnings=warnings, next_actions=next_actions) + result = PFResult( + ok=True, command="retrieve", version=PF_VERSION, data=data, warnings=warnings, next_actions=next_actions + ) if args.json: print(result.to_json()) else: print(f"{len(chunks)} chunks for: {query}") for c in chunks: - print(f" [{c.get('section','')}] {c.get('citation_key','')} p{c.get('page_number',0)}: {c['chunk_text'][:80]}...") + print( + f" [{c.get('section', '')}] {c.get('citation_key', '')} p{c.get('page_number', 0)}: {c['chunk_text'][:80]}..." + ) return 0 diff --git a/paperforge/commands/search.py b/paperforge/commands/search.py index 3b26b7b5..d3897d7c 100644 --- a/paperforge/commands/search.py +++ b/paperforge/commands/search.py @@ -35,7 +35,8 @@ def run(args: argparse.Namespace) -> int: conn = get_connection(db_path, read_only=True) try: results = search_papers( - conn, query, + conn, + query, limit=args.limit, domain=args.domain or "", year_from=args.year_from or 0, @@ -71,6 +72,8 @@ def run(args: argparse.Namespace) -> int: } if plan["query_class"] in {"mixed_query", "author_year"}: warnings.append("Zero results may reflect a noncanonical metadata query rather than library absence.") + evidence_plan = enrich_query_plan_with_runtime(build_query_plan(query, "evidence"), vault) + data["ocr_evidence_supported"] = evidence_plan.get("runtime", {}).get("ocr_evidence_available", False) next_actions.append( { "command": "paperforge query-plan", @@ -84,10 +87,14 @@ def run(args: argparse.Namespace) -> int: "reason": "The planning layer recommends a different first command for this query.", } ) - result = PFResult(ok=True, command="search", version=PF_VERSION, data=data, warnings=warnings, next_actions=next_actions) + result = PFResult( + ok=True, command="search", version=PF_VERSION, data=data, warnings=warnings, next_actions=next_actions + ) except Exception as exc: result = PFResult( - ok=False, command="search", version=PF_VERSION, + ok=False, + command="search", + version=PF_VERSION, error=PFError(code=ErrorCode.INTERNAL_ERROR, message=str(exc)), ) finally: @@ -101,7 +108,9 @@ def run(args: argparse.Namespace) -> int: print(f"Found {len(matches)} results for: {query}") for m in matches: rank_val = m.get("rank", "") - print(f" [{m['lifecycle']:16}] {m['zotero_key']} | {m['year']} | {m['first_author']} | {m['title'][:60]}") + print( + f" [{m['lifecycle']:16}] {m['zotero_key']} | {m['year']} | {m['first_author']} | {m['title'][:60]}" + ) else: print(f"Error: {result.error.message}", file=sys.stderr) return 0 if result.ok else 1 diff --git a/paperforge/query_planning.py b/paperforge/query_planning.py index 741886d3..4080ac49 100644 --- a/paperforge/query_planning.py +++ b/paperforge/query_planning.py @@ -84,9 +84,25 @@ def classify_signals(query: str) -> QuerySignals: lowered = [t.lower() for t in raw_tokens] content_keywords = { - "hz", "v/cm", "v", "ma", "mv", "galvanotaxis", "electrotaxis", - "dc", "ac", "cc", "pemf", "field", "scaffold", "stimulation", - "migration", "chondrocyte", "cartilage", "method", "parameter", + "hz", + "v/cm", + "v", + "ma", + "mv", + "galvanotaxis", + "electrotaxis", + "dc", + "ac", + "cc", + "pemf", + "field", + "scaffold", + "stimulation", + "migration", + "chondrocyte", + "cartilage", + "method", + "parameter", } for token in raw_tokens: @@ -137,23 +153,53 @@ def build_query_plan(query: str, intent: str) -> dict: query_rules: list[str] = [] fallback_plan: list[dict] = [] + suggested_modes: list[dict] = [] if query_class == "identifier_exact": identifier = signals.doi or signals.zotero_key or signals.citation_key or query.strip() primary = {"command": "paper-context", "args": {"key": identifier}} query_rules.append("Use exact identifiers directly with paper-context; do not rewrite them.") fallback_plan.append({"when": "not_found", "action": "fallback_to_paper_status"}) + elif intent == "evidence": + query_class = "evidence_query" + evidence_query = ( + " ".join(signals.content_terms or signals.title_like_tokens or tokenize(query)[:6]).strip() or query.strip() + ) + primary = {"command": "search", "args": {"query": evidence_query, "limit": 10}} + query_rules.extend( + [ + "For evidence/parameter queries, start with metadata search then verify via OCR role index.", + "When OCR role index exists, check index/role-index.json for body/caption/reference blocks.", + ] + ) + fallback_plan.extend( + [ + {"when": "zero_results", "action": "fallback_to_content_retrieve"}, + {"when": "results_found", "action": "check_ocr_evidence_layer"}, + ] + ) + suggested_modes = [ + {"mode": "ocr_evidence_layer", "description": "Check OCR role-index for exact evidence blocks (body, captions, references)."}, + {"mode": "metadata_narrow", "description": "Narrow by domain/year/author for targeted evidence verification."}, + {"mode": "read_fulltext", "description": "Read structured fulltext for context around evidence hits."}, + ] # fmt: skip elif intent == "content": - content_query = " ".join(signals.content_terms or signals.title_like_tokens or tokenize(query)[:6]).strip() or query.strip() + content_query = ( + " ".join(signals.content_terms or signals.title_like_tokens or tokenize(query)[:6]).strip() or query.strip() + ) primary = {"command": "retrieve", "args": {"query": content_query, "limit": 30}} - query_rules.extend([ - "For content lookup, start with retrieve rather than metadata search.", - "Use content-bearing terms, parameters, and method phrases as they would appear in fulltext.", - ]) - fallback_plan.extend([ - {"when": "retrieve_unavailable", "action": "interactive_fulltext_fallback"}, - {"when": "zero_results", "action": "interactive_fulltext_fallback"}, - ]) + query_rules.extend( + [ + "For content lookup, start with retrieve rather than metadata search.", + "Use content-bearing terms, parameters, and method phrases as they would appear in fulltext.", + ] + ) + fallback_plan.extend( + [ + {"when": "retrieve_unavailable", "action": "interactive_fulltext_fallback"}, + {"when": "zero_results", "action": "interactive_fulltext_fallback"}, + ] + ) elif signals.author_tokens and signals.year_tokens: primary = { "command": "search", @@ -164,25 +210,33 @@ def build_query_plan(query: str, intent: str) -> dict: "limit": 10, }, } - query_rules.extend([ - "For metadata search, prefer author and year over mixed natural-language query strings.", - "When author and year are known, do not include title words in the first-pass search query.", - ]) - fallback_plan.extend([ - {"when": "zero_results", "action": "report_noncanonical_query_risk"}, - {"when": "multiple_results", "action": "visually_narrow_with_title_terms"}, - ]) + query_rules.extend( + [ + "For metadata search, prefer author and year over mixed natural-language query strings.", + "When author and year are known, do not include title words in the first-pass search query.", + ] + ) + fallback_plan.extend( + [ + {"when": "zero_results", "action": "report_noncanonical_query_risk"}, + {"when": "multiple_results", "action": "visually_narrow_with_title_terms"}, + ] + ) else: topic_query = " ".join(signals.title_like_tokens or tokenize(query)[:6]).strip() or query.strip() primary = {"command": "search", "args": {"query": topic_query, "limit": 30}} - query_rules.extend([ - "Use short metadata-facing terms for search: title keywords, author names, domain, or collection.", - "Do not treat search as semantic fulltext discovery.", - ]) - fallback_plan.extend([ - {"when": "zero_results", "action": "report_metadata_miss_not_library_absence"}, - {"when": "large_result_set", "action": "narrow_by_domain_year_or_author"}, - ]) + query_rules.extend( + [ + "Use short metadata-facing terms for search: title keywords, author names, domain, or collection.", + "Do not treat search as semantic fulltext discovery.", + ] + ) + fallback_plan.extend( + [ + {"when": "zero_results", "action": "report_metadata_miss_not_library_absence"}, + {"when": "large_result_set", "action": "narrow_by_domain_year_or_author"}, + ] + ) return { "intent": intent, @@ -200,6 +254,7 @@ def build_query_plan(query: str, intent: str) -> dict: "recommended_primary": primary, "query_writing_rules": query_rules, "fallback_plan": fallback_plan, + "suggested_modes": suggested_modes, } @@ -209,6 +264,18 @@ def enrich_query_plan_with_runtime(plan: dict, vault: Path) -> dict: embed = get_embed_status(vault) retrieve_available = bool(embed.get("healthy", True) and embed.get("db_exists") and embed.get("chunk_count", 0) > 0) + ocr_evidence_available = False + try: + ocr_root = vault / "System" / "PaperForge" / "ocr" + if ocr_root.exists(): + ocr_evidence_available = any( + (paper_dir / "index" / "role-index.json").exists() + for paper_dir in ocr_root.iterdir() + if paper_dir.is_dir() + ) + except Exception: + pass + plan["runtime"] = { "retrieve_available": retrieve_available, "vector_status": { @@ -217,8 +284,22 @@ def enrich_query_plan_with_runtime(plan: dict, vault: Path) -> dict: "healthy": embed.get("healthy", True), "error": embed.get("error", ""), }, + "ocr_evidence_available": ocr_evidence_available, } + if plan.get("query_class") == "evidence_query" and ocr_evidence_available: + plan["suggested_modes"] = [ + { + "mode": "ocr_evidence_layer", + "description": "Check OCR role-index for exact evidence blocks (body, captions, references).", + }, + { + "mode": "metadata_narrow", + "description": "Narrow by domain/year/author for targeted evidence verification.", + }, + {"mode": "read_fulltext", "description": "Read structured fulltext for context around evidence hits."}, + ] + data = read_index(vault) items = [] if isinstance(data, dict): @@ -235,13 +316,17 @@ def enrich_query_plan_with_runtime(plan: dict, vault: Path) -> dict: "command": "context", "args": {"domain": scope["label"]}, } - plan["query_writing_rules"].append("When the user names a known domain, prefer context --domain over metadata search.") + plan["query_writing_rules"].append( + "When the user names a known domain, prefer context --domain over metadata search." + ) else: plan["recommended_primary"] = { "command": "context", "args": {"collection": scope["label"]}, } - plan["query_writing_rules"].append("When the user names a known collection, prefer context --collection over metadata search.") + plan["query_writing_rules"].append( + "When the user names a known collection, prefer context --collection over metadata search." + ) plan["fallback_plan"].insert(0, {"when": "large_inventory", "action": "summarize_and_offer_narrowing"}) if plan["intent"] == "content": @@ -276,7 +361,8 @@ def _assess_scope(items: list[dict], signals: dict) -> dict: normalized_terms = [str(t).strip().lower() for t in query_terms if str(t).strip()] matched_domain = next( ( - d for d in domain_counts + d + for d in domain_counts if any(str(d).strip().lower() == token or token in str(d).strip().lower() for token in normalized_terms) ), None, @@ -288,7 +374,8 @@ def _assess_scope(items: list[dict], signals: dict) -> dict: else: matched_collection = next( ( - c for c in collection_counts + c + for c in collection_counts if any( str(c).strip().lower() == token or token in str(c).strip().lower() @@ -300,14 +387,22 @@ def _assess_scope(items: list[dict], signals: dict) -> dict: ) if matched_collection: matched_items = [ - entry for entry in items - if any(isinstance(col, str) and col.startswith(matched_collection) for col in entry.get("collections", []) or []) + entry + for entry in items + if any( + isinstance(col, str) and col.startswith(matched_collection) + for col in entry.get("collections", []) or [] + ) ] source = "collection" label = matched_collection estimated = len(matched_items) - fulltext_ready = sum(1 for entry in matched_items if entry.get("lifecycle") in {"fulltext_ready", "deep_read_done", "ai_context_ready"}) + fulltext_ready = sum( + 1 + for entry in matched_items + if entry.get("lifecycle") in {"fulltext_ready", "deep_read_done", "ai_context_ready"} + ) if estimated <= 20: recommended_mode = "rg_now" diff --git a/tests/test_retrieve_evidence.py b/tests/test_retrieve_evidence.py new file mode 100644 index 00000000..440c38bb --- /dev/null +++ b/tests/test_retrieve_evidence.py @@ -0,0 +1,19 @@ +from __future__ import annotations + + +def test_retrieve_falls_back_to_ocr_evidence_when_vector_index_empty() -> None: + from paperforge.worker.ocr_evidence import build_evidence_hit + + hit = build_evidence_hit( + paper_id="KEY001", + role="body_paragraph", + page=3, + block_id="p3_b12", + text="We applied 2 V/cm for 30 minutes.", + confidence=0.92, + verification="", + ) + + assert hit["source_type"] == "body_paragraph" + assert "V/cm" in hit["text"] + assert hit["confidence"] > 0.5 diff --git a/tests/test_search_command_diagnostics.py b/tests/test_search_command_diagnostics.py new file mode 100644 index 00000000..27f31f7f --- /dev/null +++ b/tests/test_search_command_diagnostics.py @@ -0,0 +1,29 @@ +from __future__ import annotations + + +def test_evidence_query_routes_to_ocr_evidence_search() -> None: + from paperforge.query_planning import build_query_plan + + plan = build_query_plan("electric field parameters mammalian cells", "evidence") + + assert plan["query_class"] == "evidence_query" + assert plan["recommended_primary"]["command"] in ("ocr-evidence", "search") + assert "evidence" in str(plan.get("suggested_modes", [])).lower() or "ocr" in str(plan.get("suggested_modes", [])).lower() + + +def test_metadata_query_still_routes_to_search() -> None: + from paperforge.query_planning import build_query_plan + + plan = build_query_plan("Lin 2024 electrical stimulation", "discover") + + assert plan["query_class"] != "evidence_query" + assert plan["recommended_primary"]["command"] == "search" + + +def test_vague_content_query_still_routes_to_retrieve() -> None: + from paperforge.query_planning import build_query_plan + + plan = build_query_plan("how do cells respond to electric fields", "content") + + assert plan["query_class"] != "evidence_query" + assert plan["recommended_primary"]["command"] == "retrieve"