From 1b8151a1f6de23c56dd2674a784b9147f0b1f092 Mon Sep 17 00:00:00 2001 From: Yingfeng Date: Thu, 6 Aug 2026 19:42:05 +0800 Subject: [PATCH] Refine agentic search (#17900) --- .../harness/orchestrator/agentic.py | 16 ++++++- .../harness/orchestrator/decompose.py | 32 ++++++++++++-- rag/advanced_rag/harness/planner.py | 12 +++++- rag/advanced_rag/harness/sufficiency.py | 42 ++++++++++++++++--- rag/advanced_rag/harness/tools/gating.py | 11 +++-- rag/advanced_rag/harness/tools/search.py | 35 ++++++++++------ rag/advanced_rag/harness/types.py | 22 +++++++--- 7 files changed, 136 insertions(+), 34 deletions(-) diff --git a/rag/advanced_rag/harness/orchestrator/agentic.py b/rag/advanced_rag/harness/orchestrator/agentic.py index 49c7387023..7d9363d0ec 100644 --- a/rag/advanced_rag/harness/orchestrator/agentic.py +++ b/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) diff --git a/rag/advanced_rag/harness/orchestrator/decompose.py b/rag/advanced_rag/harness/orchestrator/decompose.py index 07c85d5a16..52ebbef6b3 100644 --- a/rag/advanced_rag/harness/orchestrator/decompose.py +++ b/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]] diff --git a/rag/advanced_rag/harness/planner.py b/rag/advanced_rag/harness/planner.py index 9a73df019e..51c40b4a98 100644 --- a/rag/advanced_rag/harness/planner.py +++ b/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): diff --git a/rag/advanced_rag/harness/sufficiency.py b/rag/advanced_rag/harness/sufficiency.py index 9dab9d2947..15b9462a43 100644 --- a/rag/advanced_rag/harness/sufficiency.py +++ b/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"(? 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) diff --git a/rag/advanced_rag/harness/tools/gating.py b/rag/advanced_rag/harness/tools/gating.py index 7b1f2485c9..3b5abbcaa3 100644 --- a/rag/advanced_rag/harness/tools/gating.py +++ b/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 diff --git a/rag/advanced_rag/harness/tools/search.py b/rag/advanced_rag/harness/tools/search.py index 4e2d90b5bc..0b74f42e51 100644 --- a/rag/advanced_rag/harness/tools/search.py +++ b/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 diff --git a/rag/advanced_rag/harness/types.py b/rag/advanced_rag/harness/types.py index 80e5d47715..30b9bee55c 100644 --- a/rag/advanced_rag/harness/types.py +++ b/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