From f1641228e2ffa5e2168ed660b87ec7919c45fd42 Mon Sep 17 00:00:00 2001 From: Yingfeng Date: Tue, 11 Aug 2026 13:40:11 +0800 Subject: [PATCH] Refine agentic search & orchestration loop (#18057) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary This PR improves the RAGFlow agentic-search path in three areas: it stops the outer agent from re-looping over the same rag call, lets the medium thinking mode discover and follow new sub-claims mid-loop, and strengthens retrieval by having the LLM emit synonym-rich queries with time/date/number terms boosted. 1. Avoid the outer re-loop — keep all multi-hop cycles inside agentic RAG 2. Dynamic claims in medium mode — keep querying newly discovered sub-questions medium now enables allows_dynamic_claims. During orchestration, when claim analysis discovers a new required sub-question (discovered_claims), the loop spawns it as a new ClaimTarget and continues searching it in subsequent cycles (bounded by the dynamic-claim budget) instead of stopping. Also added: 3. Stronger query strategy — synonym-rich queries + time/date/number weighting LLM-generated synonyms: the claim-analysis prompt now instructs the model to write each next_queries entry as a retrieval-boosted query that actively folds in entity aliases, DATE/TIME synonyms (e.g. 1994 → 1994, 66th Academy Awards), and number/unit variants (e.g. 1.95 m → 6 ft 5 in). Time/date/number boosting: query.py boosts numeric/date tokens to a high weight (_NUM_DATE_TOKEN_RE). --- rag/advanced_rag/harness/config.py | 2 +- .../harness/orchestrator/decompose.py | 140 +++++++++++++++++- .../harness/prompts/report_prompt.py | 20 +++ rag/llm/chat_model.py | 19 +++ rag/nlp/query.py | 25 ++++ 5 files changed, 200 insertions(+), 6 deletions(-) diff --git a/rag/advanced_rag/harness/config.py b/rag/advanced_rag/harness/config.py index 962ab2918e..94e6eee3ce 100644 --- a/rag/advanced_rag/harness/config.py +++ b/rag/advanced_rag/harness/config.py @@ -26,7 +26,7 @@ THINKING_MODES: dict[str, ExecutionStrategy] = { requires_agent_loop=False, requires_sufficiency_judge=True, requires_selective_gen=True, - allows_dynamic_claims=False, + allows_dynamic_claims=True, allows_replan=False, max_orchestrator_cycles=3, max_agent_cycles=0, diff --git a/rag/advanced_rag/harness/orchestrator/decompose.py b/rag/advanced_rag/harness/orchestrator/decompose.py index 33af96f393..a97b0821d5 100644 --- a/rag/advanced_rag/harness/orchestrator/decompose.py +++ b/rag/advanced_rag/harness/orchestrator/decompose.py @@ -19,6 +19,9 @@ _LOG = logging.getLogger(__name__) _MAX_EVIDENCE_SNIPPETS = 6 _MAX_NEXT_QUERIES = 3 +# Upper bound on dynamically-discovered claims per decomposition, to prevent +# open-ended claim expansion from the evidence analysis from blowing up cost. +_MAX_DYNAMIC_CLAIMS = 6 _EVIDENCE_ANALYSIS_SYSTEM = """You are controlling a multi-hop RAG retrieval loop. Judge whether the retrieved passages verify the claim using only the provided evidence. @@ -27,7 +30,16 @@ dates, names, or relationships discovered in the evidence and move closer to the original question. Distinguish final-answer entities from bridge entities. If the passages identify only a clue node in the chain, keep the claim incomplete and search for the -remaining relation needed by the original question. Return JSON only.""" +remaining relation needed by the original question. +Each `next_queries` entry MUST be a retrieval-boosted query: in addition to the core +terms, actively fold in SYNONYMS and variants so a single search recalls more of the +corpus: +- entity synonyms/aliases: "Usain Bolt" -> "Usain Bolt, sprinter, 100 m record holder"; +- DATE/TIME synonyms: "1994" -> "1994, 66th Academy Awards, mid-1990s"; "2011-02-05" -> + "5 February 2011, Feb 5 2011"; +- number/unit variants: "1.95 m" -> "1.95 m, 6 ft 5 in"; "50 m" -> "50 m, 50 metres". +Write each query as a compact, self-contained phrase (terms + synonyms), not a full +sentence. Return JSON only.""" _EVIDENCE_ANALYSIS_USER = """Original question: {question} @@ -50,10 +62,20 @@ Return JSON: "report": "Short evidence-backed finding, or what was learned so far.", "gaps": ["specific missing fact or relationship"], "next_queries": ["standalone follow-up search query"], + "discovered_claims": ["a NEW, previously-unlisted sub-question that the original question requires but the current claim decomposition does not cover, if any; empty if no new sub-question is needed"], "grounded": ["key asserted facts that ARE directly supported by the cited evidence, atomically and verbatim enough to match"], "numbers": ["for numerical/multi-hop answers: each figure used + its source, e.g. '2,161,000 from Wikipedia Demographics of Paris'; list ALL conflicting figures if several sources disagree"] }} -Only list in grounded the facts you actually SAW in the evidence; prior-knowledge guesses go in gaps. If the claim is numerical or multi-hop and the evidence has multiple close-but-different figures, disclose all of them in numbers rather than silently picking one.""" +Only list in grounded the facts you actually SAW in the evidence; prior-knowledge guesses go in gaps. If the claim is numerical or multi-hop and the evidence has multiple close-but-different figures, disclose all of them in numbers rather than silently picking one. discovered_claims must ONLY be genuinely new necessary sub-questions the original question needs (e.g. "which players were on the 1995 Pro Bowl roster"), never re-statements of an existing claim.""" + +# Rewrites the search query right before an ABSTAIN, in a last-ditch attempt to +# surface evidence that the earlier queries missed (e.g. the question constrains a +# year the first pass only returned estimates for, or needs a disambiguating entity). +_QUERY_REWRITE_SYSTEM = """You are rewriting a retrieval query to recover missing evidence. +The previous retrieval came back with no usable evidence. Rewrite the query so a fresh +search is more likely to hit the needed fact: anchor it with explicit years, dates, +full names, role words, and any distinguishing constraints from the question. Do NOT +invent facts. Output a single standalone search query only.""" async def decompose_and_search(state: dict, tools) -> dict: @@ -78,17 +100,29 @@ async def decompose_and_search(state: dict, tools) -> dict: prev_score: float | None = None _STAGNATION_CYCLES = 2 _STAGNATION_GAIN = 0.05 + # Query-rewrite rescue is attempted at most once per decomposition, so a + # mis-anchored query that ABSTAINs cannot trigger an unbounded rewrite loop. + rescue_used = False - for cycle in range(max_cycles): + # Dynamic round budget (Plan B): start at ``max_cycles`` and grant extra + # rounds when a NEW dynamic claim is discovered near the budget boundary, so a + # sub-question uncovered in the final round still gets a chance to be searched + # and verified (otherwise the for-loop would end and the fresh claim would + # never be retrieved). Bounded to ``2 * max_cycles`` to avoid open-ended cost. + cycle = 0 + budget = max_cycles + _MAX_BUDGET = max_cycles * 2 + while cycle < budget: ctx.iteration = cycle unverified = [c for c in ctx.claims if not c.is_verified] if not unverified: break + new_dynamic_this_round = False _LOG.info( "[Decompose search] Round %d of %d: researching %d unresolved claim(s).", cycle + 1, - max_cycles, + budget, len(unverified), ) @@ -148,6 +182,7 @@ async def decompose_and_search(state: dict, tools) -> dict: c.is_verified = analysis["is_verified"] c.confidence = analysis["confidence"] + discovered_claims = analysis.get("discovered_claims", []) c.agent_result = AgentResult( claim_id=c.claim_id, report=analysis["report"], @@ -155,7 +190,7 @@ async def decompose_and_search(state: dict, tools) -> dict: confidence=c.confidence, evidence_ids=evidence_ids, gaps=analysis["gaps"], - discovered_claims=[], + discovered_claims=discovered_claims, grounded=analysis.get("grounded", []), numbers=analysis.get("numbers", []), ) @@ -171,6 +206,27 @@ async def decompose_and_search(state: dict, tools) -> dict: c.claim_id, len(next_queries), ) + + # Dynamic claim discovery: when the evidence analysis reveals a NEW + # necessary sub-question the original question needs but the planner's + # initial decomposition did not cover, add it as a fresh claim so the + # orchestrator retrieves and verifies it in a later round. Mirrors the + # high/ultra agentic orchestrator (agentic.py). Bounded to avoid blow-up. + if mode.allows_dynamic_claims and not c.is_verified: + existing_desc = {cc.description for cc in ctx.claims} + for dc in discovered_claims: + if dc and dc not in existing_desc and len(ctx.claims) < _MAX_DYNAMIC_CLAIMS: + dyn_id = f"c_dyn_{len(ctx.claims)}" + ctx.claims.append(ClaimTarget(claim_id=dyn_id, description=dc)) + attempted_queries[dyn_id] = set() + pending_queries[dyn_id] = [dc] + existing_desc.add(dc) + new_dynamic_this_round = True + _LOG.info( + '[Decompose search] Discovered new sub-question from claim %s: "%s" (queued for next round).', + c.claim_id, + dc, + ) _LOG.info( '[Decompose search] Claim %s after "%s": %s (confidence %.0f%%).', c.claim_id, @@ -274,6 +330,16 @@ async def decompose_and_search(state: dict, tools) -> dict: if action in ("ANSWER", "ANSWER_PARTIAL"): return _finalize(ctx, tools, partial=action == "ANSWER_PARTIAL", loop=completed_cycles) if action == "ABSTAIN": + # Query-rewrite rescue: before refusing, rewrite the search query + # (anchoring years/constraints from the gaps) and re-search once. A + # merely mis-anchored query (e.g. the question pins a year the first + # pass only returned estimates for) otherwise abstains despite the + # data existing. Only attempted once per decomposition. + if not rescue_used and await _rewrite_and_retry(tools, ctx.question, keywords, ctx): + rescue_used = True + prev_score = None + cycle += 1 + continue tools.kbinfos["chunks"] = [] return {"verdict": verdict.__dict__, "abstain": True, "loop": completed_cycles} if action == "FALLBACK_LLM": @@ -281,6 +347,20 @@ async def decompose_and_search(state: dict, tools) -> dict: if not should_continue: break + # Dynamic budget extension (Plan B): if this round discovered a NEW dynamic + # claim near the budget boundary, grant an extra round so it can actually be + # searched (a fresh claim discovered in the final round would otherwise be + # left unretrieved). Stagnation guard still bounds total cost. + if new_dynamic_this_round and cycle + 1 >= budget and budget < _MAX_BUDGET: + budget += 1 + _LOG.info( + "[Decompose] Round %d: discovered new claim(s); extending round budget to %d.", + cycle + 1, + budget, + ) + + cycle += 1 + if not tools.kbinfos.get("chunks"): return {"empty_result": True, "kbinfos": tools.kbinfos, "loop": completed_cycles} @@ -307,6 +387,7 @@ async def _analyze_claim_evidence( "report": "", "gaps": ["no evidence found"], "next_queries": _fallback_queries(question, claim), + "discovered_claims": [], } try: @@ -374,6 +455,7 @@ def _normalize_analysis( report = str(parsed.get("report") or "").strip() or _summarize(result) gaps = _string_list(parsed.get("gaps")) next_queries = _string_list(parsed.get("next_queries"))[:_MAX_NEXT_QUERIES] + discovered_claims = _string_list(parsed.get("discovered_claims"))[:_MAX_NEXT_QUERIES] grounded = _string_list(parsed.get("grounded")) numbers = _string_list(parsed.get("numbers")) @@ -393,6 +475,7 @@ def _normalize_analysis( "report": report, "gaps": gaps, "next_queries": next_queries, + "discovered_claims": discovered_claims, "grounded": grounded, "numbers": numbers, } @@ -415,6 +498,7 @@ def _fallback_analysis( "report": _summarize(result), "gaps": [] if is_verified else ["need more specific evidence"], "next_queries": next_queries, + "discovered_claims": [], "grounded": [], "numbers": [], } @@ -584,3 +668,49 @@ def _summarize(result: dict) -> str: chunks = result.get("chunks", []) texts = [(c.get("content_with_weight") or c.get("content") or c.get("text") or "")[:200] for c in chunks[:3]] return " | ".join(texts) + + +async def _rewrite_and_retry(tools, question: str, keywords: str, ctx: OrchestratorContext) -> bool: + """Last-ditch attempt before an ABSTAIN: rewrite the search query so a fresh + search is anchored on the missing evidence (years, dates, names, constraints) + and retrieve once more. Returns True if fresh evidence was merged; the caller + then continues the decompose loop instead of refusing. At most one rewrite per + decomposition is attempted by the caller (``rescue_used``).""" + unverified = [c for c in ctx.claims if not c.is_verified and c.agent_result] + gaps: list[str] = [] + for c in unverified: + gaps.extend(c.agent_result.gaps or []) + followups = getattr(ctx, "pending_followups", None) or [] + anchors = list(dict.fromkeys([g for g in (gaps + followups) if g and g.strip()]))[:4] + if not anchors: + # Nothing concrete to anchor the rewrite on — avoid a blind re-search. + _LOG.info("[Decompose] ABSTAIN rescue skipped: no gap/followup to anchor a rewrite.") + return False + + user = ( + f"Question: {question}\n\n" + f"Missing evidence / gaps found so far:\n- " + "\n- ".join(anchors) + "\n\n" + "Rewrite ONE standalone search query (anchor it with explicit years, dates, " + "full names, role words and distinguishing constraints from the question) that " + "is most likely to retrieve the missing evidence. Output only the query." + ) + try: + msg = await tools._fit_messages(_QUERY_REWRITE_SYSTEM, user) + ans = await tools.chat_mdl.async_chat(msg[0]["content"], msg[1:], {"temperature": 0.0}) + if isinstance(ans, tuple): + ans = ans[0] + query = (ans or "").strip().strip('"').strip("'").replace("\n", " ") + if not query: + return False + _LOG.info('[Decompose] ABSTAIN rescue: rewritten query = "%s"', query) + result = await hybrid_search(tools, query=query, keywords=keywords, use_compiled=True) + chunks = (result or {}).get("chunks", []) or [] + if not chunks: + _LOG.info("[Decompose] ABSTAIN rescue: rewritten query returned no chunks.") + return False + _merge_kbinfos(tools, result) + _LOG.info("[Decompose] ABSTAIN rescue: recovered %d chunk(s) from rewritten query.", len(chunks)) + return True + except Exception: + _LOG.exception("[Decompose] ABSTAIN rescue query rewrite failed.") + return False diff --git a/rag/advanced_rag/harness/prompts/report_prompt.py b/rag/advanced_rag/harness/prompts/report_prompt.py index 11a2ee0f90..7243aa99d2 100644 --- a/rag/advanced_rag/harness/prompts/report_prompt.py +++ b/rag/advanced_rag/harness/prompts/report_prompt.py @@ -22,6 +22,26 @@ Answer the question's own attribute using the evidence for THAT attribute. If th supports a different attribute, say that you could only find the related (different) attribute and do not present it as the answer to the requested one. +# Reasoning over evidence +Some questions require MULTI-STEP reasoning from the evidence, not verbatim lookup. If the question +does, work through the steps explicitly before giving the final answer: +- NUMERIC: compute the requested value from the evidence's figures (e.g. the delta between two + dates/years, how many pool crossings = (h1+h2) / lane length, an age from a birth date and a + reference date). Show the arithmetic that produces the answer. +- INFERENCE: derive an attribute the evidence implies (e.g. the zodiac sign from a birth date, + which place a person reached from a standings table). +- EXACT-ATTRIBUTE: when the evidence contains the requested entity, extract the precise value asked + for (full name, hometown vs birthplace, first vs last, medal type vs rank), never a + related-but-different one. +- FIRST-NAME: if the question asks for a person's FIRST name, take the first word of their FULL name. + A person is commonly cited by their middle name (e.g. "Sargent Shriver" is the middle + surname of + "Robert Sargent Shriver Jr."), so do not assume the commonly-used given name is the first name — + search the evidence for the complete full name (e.g. written "Shriver, Robert Sargent" or + "Robert Sargent Shriver") and answer with its first word (Robert). +Only reason from facts actually present in the evidence. If the evidence is insufficient to complete +the required step (e.g. a needed intermediate date or number is missing), say so plainly rather than +guessing. For a direct-lookup question, skip straight to the answer. + # Language Answer in the SAME language as the question. Translate retrieved evidence into that language as part of composing the answer; only verbatim quoted snippets may stay in their source language. diff --git a/rag/llm/chat_model.py b/rag/llm/chat_model.py index 5393b449d0..291abd0274 100644 --- a/rag/llm/chat_model.py +++ b/rag/llm/chat_model.py @@ -2101,6 +2101,25 @@ class LiteLLMBase(ABC): logging.info(f"Response tool_calls={message.tool_calls}") results = await asyncio.gather(*[_exec_tool(tc) for tc in message.tool_calls]) + + # Terminal-tool short-circuit: a terminal tool (e.g. ``rag``) + # already composes the final answer itself, so return its + # result directly instead of feeding it back for another LLM + # round. This mirrors the streaming path + # (``async_chat_streamly_with_tools``), which otherwise lets + # the outer model re-invoke ``rag`` on partial answers until + # ``max_rounds`` is exhausted — the source of multi-hop + # timeouts (single request exceeding the client budget). + _terminal = getattr(self, "terminal_tools", None) + if _terminal: + for tc, name, args, result, err in results: + if name in _terminal and not err: + logging.info(f"[Tool loop] The {name} tool produced the final answer — done.") + out = result if isinstance(result, str) else json.dumps(result, ensure_ascii=False) + if out: + ans += out + return self._sanitize_answer(ans), tk_count + history = self._append_history_batch( history, results, diff --git a/rag/nlp/query.py b/rag/nlp/query.py index 081da1325f..7bb2963385 100644 --- a/rag/nlp/query.py +++ b/rag/nlp/query.py @@ -25,6 +25,19 @@ from rag.nlp import rag_tokenizer, term_weight, synonym from rag.utils.redis_conn import REDIS_CONN +# Tokens that are hard time/date/number constraints (a year, a date, a measurement). +# They get a high BM25 weight so chunks carrying the exact value rank above passages +# that merely mention the surrounding entity. Matches: +# pure numbers / decimals: 1994, 2001, 1.95, 3.68 +# dates: 2011-02-05, 1994-06-23, 02/05/2011 +# numbers with a unit: 50m, 1.95m, 6ft5in, 94kg +_NUM_DATE_TOKEN_RE = re.compile( + r"^\d[\d.,/:\-]*$" + r"|^\d+(?:\.\d+)?\s*(?:m|km|cm|mm|kg|g|lb|ft|in|yd|s|ms|min|h|hr|sec|y|yr|yo|yrs|k|m|b|th|nd|rd|st|%)$", + re.IGNORECASE, +) + + class FulltextQueryer(QueryBase): def __init__(self): self.tw = term_weight.Dealer() @@ -64,6 +77,18 @@ class FulltextQueryer(QueryBase): tks_w = [(re.sub(r"[ \\\"'^]", "", tk), w) for tk, w in tks_w] tks_w = [(re.sub(r"^[\+-]", "", tk), w) for tk, w in tks_w if tk] tks_w = [(tk.strip(), w) for tk, w in tks_w if tk.strip()] + # Time/date/number terms are hard constraints in retrieval (e.g. a year, + # a date, a measurement). Boost them so chunks carrying the exact value + # rank well above passages that merely mention the surrounding entity. + # + # NOTE on decimals/dates: ``rag_tokenizer`` splits "1.95" into "1" "95" + # and "2011-02-05" into "2011" "02" "05". Each integer fragment is still + # boosted to 10, because the query_string builder below pairs every two + # adjacent tokens into a bigram phrase (``"1 95"^max*2``). That bigram + # re-connects the fragments and matches the exact value "1.95" in the + # index, so weighting the fragments is exactly what makes the decimal + # match — they are NOT independent numbers in the query. + tks_w = [(tk, 10.0 if _NUM_DATE_TOKEN_RE.match(tk) else w) for tk, w in tks_w] syns = [] for tk, w in tks_w[:256]: # Strip single quotes from synonym terms to avoid Infinity lexer TokenError