Refine agentic search (#17900)

rag/advanced_rag/harness/orchestrator/agentic.py

@@ -148,13 +148,25 @@ async def agentic_research(state: dict, tools) -> dict:
             tools.kbinfos["chunks"] = []
             return {"verdict": verdict.__dict__, "abstain": True}
         if action == "REPLAN":
-            # Ultra: re-plan on low score
+            # Ultra: re-plan on low score. Ground the new plan on the evidence
+            # gathered so far, and carry still-valid verified claims over so a
+            # replan doesn't re-research (and re-bill) work already done.
             from rag.advanced_rag.harness.planner import planner_node
 
             state["feedback"] = verdict.feedback
             state["route"] = route
+            state["seed_chunks"] = list(tools.kbinfos.get("chunks", []) or [])
             new_plan = await planner_node(state, tools)
-            ctx.claims = new_plan.get("claims", ctx.claims)
+            # Keep EVERY verified claim (even ones the new plan omitted — their
+            # evidence is still valid and shouldn't be re-researched), then
+            # append only the new plan's unverified claims.
+            verified = [c for c in ctx.claims if c.is_verified]
+            new_by_desc = {}
+            for c in new_plan.get("claims", ctx.claims):
+                if isinstance(c, ClaimTarget):
+                    new_by_desc.setdefault(c.description, c)
+            seen = {c.description for c in verified}
+            ctx.claims = verified + [c for c in new_by_desc.values() if c.description not in seen]
         if action == "FALLBACK_LLM":
             return _finalize(ctx, tools, partial=True, fallback=True)
 

rag/advanced_rag/harness/orchestrator/decompose.py

@@ -40,6 +40,7 @@ async def decompose_and_search(state: dict, tools) -> dict:
 
         for c, result in zip(unverified, results):
             if result.get("chunks"):
+                _merge_kbinfos(tools, result)
                 c.is_verified = True
                 c.confidence = 0.8
                 c.agent_result = AgentResult(
@@ -47,9 +48,8 @@ async def decompose_and_search(state: dict, tools) -> dict:
                     report=_summarize(result),
                     is_verified=True,
                     confidence=0.8,
-                    evidence_ids=list(range(len(result.get("chunks", [])))),
+                    evidence_ids=_global_evidence_ids(tools, result),
                 )
-                _merge_kbinfos(tools, result)
             else:
                 c.agent_result = AgentResult(
                     claim_id=c.claim_id,
@@ -83,7 +83,15 @@ async def decompose_and_search(state: dict, tools) -> dict:
             tools.kbinfos["chunks"] = []
             return {"verdict": verdict.__dict__, "abstain": True}
 
-    return {"kbinfos": tools.kbinfos}
+    # Cycle exhaustion: flag the answer as partial so the final-answer node
+    # prepends the partial-answer preamble instead of presenting an incomplete
+    # answer as complete. Partial when (a) some claim is still unverified, or
+    # (b) every claim is verified but the final verdict is not SUFFICIENT (e.g.
+    # cross-check flagged conflicts/mismatches) — in both cases an exhaustive
+    # answer was not reached.
+    verdict_status = getattr(verdict, "status", None)
+    partial = (any(not c.is_verified for c in ctx.claims) or (verdict_status is not None and verdict_status != "SUFFICIENT")) and bool(tools.kbinfos.get("chunks"))
+    return {"kbinfos": tools.kbinfos, "partial_answer": partial}
 
 
 def _merge_kbinfos(tools, result: dict):
@@ -108,6 +116,24 @@ def _chunk_key(ck: dict) -> str:
     return ck.get("chunk_id") or ck.get("id") or str(id(ck))
 
 
+def _global_evidence_ids(tools, result: dict) -> list[int]:
+    """Map a search result's chunks to their indices in ``tools.kbinfos``.
+
+    The cross-check resolves evidence IDs against the shared kbinfos pool, so
+    the IDs must be global indices there — not positions within this result.
+    Must be called AFTER ``_merge_kbinfos`` so fresh chunks have indices.
+    """
+    index_by_key: dict[str, int] = {}
+    for idx, ck in enumerate(tools.kbinfos.get("chunks", [])):
+        index_by_key.setdefault(_chunk_key(ck), idx)
+    ids: list[int] = []
+    for ck in result.get("chunks", []):
+        idx = index_by_key.get(_chunk_key(ck))
+        if idx is not None and idx not in ids:
+            ids.append(idx)
+    return ids
+
+
 def _summarize(result: dict) -> str:
     chunks = result.get("chunks", [])
     texts = [c.get("content_with_weight", "")[:200] for c in chunks[:3]]

rag/advanced_rag/harness/planner.py

@@ -39,7 +39,7 @@ async def planner_node(state: dict, tools) -> dict:
         _LOG.warning("planner: no route found, using defaults")
         return _default_plan(state.get("question", ""))
 
-    _LOG.info("[Planner] Working out how to research this %s question: \"%s\"", route.question_type, _snip(route.question))
+    _LOG.info('[Planner] Working out how to research this %s question: "%s"', route.question_type, _snip(route.question))
     if not route.requires_decomposition:
         # Direct mode: single coarse claim
         return _direct_plan(route.question)
@@ -69,6 +69,16 @@ async def planner_node(state: dict, tools) -> dict:
         system, user = prompt.split("Output format", 1)
         system = system.strip()
         user = "Output format" + user
+        # Replanning: the orchestrator sets ``feedback`` from the sufficiency
+        # verdict — the new plan must close those gaps instead of repeating
+        # the previous one.
+        feedback = (state.get("feedback") or "").strip()
+        if feedback:
+            system += (
+                "\n\nA previous research round already ran and left gaps. "
+                "Feedback from the sufficiency check — the new plan MUST address "
+                "these points with different, more targeted claims:\n" + feedback
+            )
         msg = await tools._fit_messages(system, user)
         ans = await tools.chat_mdl.async_chat(msg[0]["content"], msg[1:], {"temperature": 0.2})
         if isinstance(ans, tuple):

rag/advanced_rag/harness/sufficiency.py

@@ -59,17 +59,44 @@ def cross_check_claim(agent_result: AgentResult, all_chunks: dict) -> ClaimCross
         text_lower = text.lower()
 
         for num in numbers:
-            if str(num) not in text_lower:
-                mismatches.append(f"number {num} not found in chunk {eid}")
-            else:
+            # Numbers are extracted as floats ("1976" -> 1976.0) while chunk
+            # text spells them "1976" — match both the raw and integral forms.
+            # Bounded match: a number must not sit adjacent to other digit
+            # characters, so 1976 does not match inside 19760.
+            forms = {str(num), str(int(num))} if float(num).is_integer() else {str(num)}
+            if any(re.search(rf"(?<![\w]){re.escape(f)}(?![\w])", text_lower) for f in forms):
                 matches.append(f"number {num} found in chunk {eid}")
+            else:
+                mismatches.append(f"number {num} not found in chunk {eid}")
 
         for ent in entities:
-            if ent.lower() not in text_lower:
+            # Bounded word/phrase match: Ann must not match Annual (no adjacent
+            # word characters on either side).
+            if re.search(rf"(?<![\w]){re.escape(ent.lower())}(?![\w])", text_lower):
+                matches.append(f"entity '{ent}' found in chunk {eid}")
+            else:
                 mismatches.append(f"entity '{ent}' not found in chunk {eid}")
 
     total = len(matches) + len(mismatches)
-    cross_score = len(matches) / max(total, 1) if total > 0 else 0.0
+    if total == 0:
+        # No evidence was actually examined — fail rather than pass neutrally:
+        # a claim with zero evidence IDs cannot be cross-checked at all.
+        if not agent_result.evidence_ids:
+            return ClaimCrossCheckResult(
+                claim_id=agent_result.claim_id,
+                cross_check_passed=False,
+                cross_check_score=0.0,
+                mismatches=["no evidence"],
+            )
+        # Evidence IDs exist but nothing extractable to verify against (e.g.
+        # Chinese reports yield no capitalized entities and no digits) — the
+        # cross-check cannot falsify the claim, so pass neutrally.
+        return ClaimCrossCheckResult(
+            claim_id=agent_result.claim_id,
+            cross_check_passed=True,
+            cross_check_score=1.0,
+        )
+    cross_score = len(matches) / total
     cross_passed = len(mismatches) < len(matches) * 0.5
 
     return ClaimCrossCheckResult(
@@ -171,7 +198,10 @@ def route_sufficiency_verdict(verdict: SufficiencyVerdict, mode_label: str, cycl
         return ("CONTINUE", False)
 
     if verdict.status == "INSUFFICIENT":
-        if cycle >= max_cycles * 0.8:
+        # ``cycle`` is 0-based, so the last cycle is ``max_cycles - 1`` — the
+        # old ``max_cycles * 0.8`` threshold was never reached for the 3/3/4
+        # cycle budgets, making this branch dead code.
+        if cycle >= max_cycles - 1:
             return ("ANSWER_PARTIAL", False)
         return ("CONTINUE", True)
 

rag/advanced_rag/harness/tools/gating.py

@@ -111,10 +111,13 @@ def get_gated_tools(
         sorted_tools.append(tool_name)
 
     selected = sorted_tools[: phase_config["max_returned"]]
-    defs = [TOOL_REGISTRY[n]["function_schema"] for n in selected if n in TOOL_REGISTRY]
-    for d in defs:
-        d["x_phase"] = phase
-        d["x_phase_hint"] = phase_config["tool_hint"]
+    # Copy the registry schemas before annotating — the registry dicts are
+    # shared process-wide, mutating them would leak phase hints across
+    # concurrent requests.
+    defs = []
+    for n in selected:
+        if n in TOOL_REGISTRY:
+            defs.append({**TOOL_REGISTRY[n]["function_schema"], "x_phase": phase, "x_phase_hint": phase_config["tool_hint"]})
     return defs
 
 

rag/advanced_rag/harness/tools/search.py

@@ -151,6 +151,26 @@ def _narrow_by_keywords(chunks: list[dict], keywords: str) -> list[dict]:
     return out
 
 
+def _narrow_or_keep(chunks: list[dict], keywords: str, label: str) -> list[dict]:
+    """Narrow chunks to keyword sentences, but keep the originals when
+    narrowing would drop everything.
+
+    No keyword overlap does not mean irrelevant — the retriever already ranked
+    these chunks, and a sub-question's wording need not contain the parent
+    question's keywords. Dropping them all produced empty results, unverified
+    claims and pointless retry cycles.
+    """
+    if not keywords or not chunks:
+        return chunks
+    length = len(chunks)
+    narrowed = _narrow_by_keywords(chunks, keywords)
+    if narrowed:
+        _LOG.info(f"[{label}] Kept {len(narrowed)} of {length} passage(s) that actually mention the keywords.")
+        return narrowed
+    _LOG.info(f"[{label}] Keyword narrowing matched nothing — keeping all {length} retrieved passage(s).")
+    return chunks
+
+
 def _search_cache_key(effective_query: str, target_ids, top_n: int, doc_scope) -> tuple:
     """Key a retrieval by what actually determines its result.
 
@@ -220,10 +240,7 @@ async def hybrid_search(tools, query: str, kb_ids: list[str] | None = None, top_
         must_not={"exists": "compile_kwd"},  # plain retrieval = document chunks only; compiled products have their own tools
     )
     kbinfos = _normalize(kbinfos, tools.tenant_ids)
-    if keywords:
-        length = len(kbinfos["chunks"])
-        kbinfos["chunks"] = _narrow_by_keywords(kbinfos.get("chunks", []), keywords)
-        _LOG.info(f"[Hybrid search] Kept {len(kbinfos['chunks'])} of {length} passage(s) that actually mention the keywords.")
+    kbinfos["chunks"] = _narrow_or_keep(kbinfos.get("chunks", []), keywords, "Hybrid search")
     if use_compiled and kbinfos.get("chunks"):
         _LOG.info("[Hybrid search] Compiled expansion enabled — enriching with page_index/tree/KG navigation.")
         await _expand_with_compiled(tools, query, keywords, kbinfos, doc_scope)
@@ -257,10 +274,7 @@ async def vector_search(tools, query: str, kb_ids: list[str] | None = None, top_
         must_not={"exists": "compile_kwd"},
     )
     kbinfos = _normalize(kbinfos, tools.tenant_ids)
-    if keywords:
-        length = len(kbinfos["chunks"])
-        kbinfos["chunks"] = _narrow_by_keywords(kbinfos.get("chunks", []), keywords)
-        _LOG.info(f"[Vector search] Kept {len(kbinfos['chunks'])} of {length} passage(s) that actually mention the keywords.")
+    kbinfos["chunks"] = _narrow_or_keep(kbinfos.get("chunks", []), keywords, "Vector search")
     return kbinfos
 
 
@@ -285,10 +299,7 @@ async def bm25_search(tools, query: str, kb_ids: list[str] | None = None, top_n:
         must_not={"exists": "compile_kwd"},
     )
     kbinfos = _normalize(kbinfos, tools.tenant_ids)
-    if keywords:
-        length = len(kbinfos["chunks"])
-        kbinfos["chunks"] = _narrow_by_keywords(kbinfos.get("chunks", []), keywords)
-        _LOG.info(f"[BM25 search] Kept {len(kbinfos['chunks'])} of {length} passage(s) that actually mention the keywords.")
+    kbinfos["chunks"] = _narrow_or_keep(kbinfos.get("chunks", []), keywords, "BM25 search")
     return kbinfos
 
 

rag/advanced_rag/harness/types.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Any, Literal
+from typing import Literal
 
 
 # ═══════════════════════════════════════════════════════════════
@@ -59,7 +59,7 @@ class ClaimTarget:
     is_verified: bool = False
     confidence: float = 0.0
     suggested_tools: list[str] = field(default_factory=list)
-    agent_result: dict | None = None
+    agent_result: AgentResult | None = None
 
 
 @dataclass
@@ -144,7 +144,6 @@ class OrchestratorContext:
     mode: str
     iteration: int = 0
     current_phase: str = "locate"
-    agent_results: dict[str, Any] = field(default_factory=dict)
     verdict: SufficiencyVerdict | None = None
     history: list[dict] = field(default_factory=list)
     _last_entity: str | None = None
@@ -169,7 +168,18 @@ class OrchestratorContext:
         return unverified[0].description if unverified else None
 
     def has_any_chunks(self) -> bool:
-        return any(r.get("evidence_ids") for r in self.agent_results.values())
+        """True once any claim's research produced evidence passages.
 
-    def record_fallback(self, tool_name: str, fallback_from: str | None = None):
-        pass
+        Reads the claims (whose ``agent_result`` is populated by both
+        orchestrators) — the ``agent_results`` dict is never written to, so
+        reading it would leave the search phase stuck at "locate" forever and
+        gate off every inspector tool.
+        """
+        for c in self.claims:
+            r = c.agent_result
+            if r is None:
+                continue
+            ids = r.get("evidence_ids") if isinstance(r, dict) else r.evidence_ids
+            if ids:
+                return True
+        return False