Refine agentic search & orchestration loop (#18057)

## 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

@@ -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,

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

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.
 

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,

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