diff --git a/api/db/services/dialog_service.py b/api/db/services/dialog_service.py index bde5d2858d..a17a01a5b3 100644 --- a/api/db/services/dialog_service.py +++ b/api/db/services/dialog_service.py @@ -1920,6 +1920,7 @@ async def rag_agent(dialog, messages, stream=True, **kwargs): web_search=create_web_search_provider(prompt_config) if use_web_search else None, meta_data_filter=dialog.meta_data_filter, doc_scope=doc_scope, + empty_response=prompt_config.get("empty_response", ""), do_refer=False, thinking_mode=thinking_mode, ) @@ -1983,8 +1984,9 @@ async def rag_agent(dialog, messages, stream=True, **kwargs): if getattr(chat_mdl, "mdl", None) is not None: chat_mdl.mdl.terminal_tools = {"rag"} if stream: - # Surface the agentic pipeline's bracket-tagged progress logs to the - # client as content, interleaved with the real token stream. + # Surface the outer model's reasoning, agent progress logs, and the + # final-answer model's reasoning as one continuous think block. The + # final answer itself is emitted from the inner graph's own deltas. from rag.advanced_rag.think_log import install_think_log_handler, set_think_log_sink, reset_think_log_sink install_think_log_handler() @@ -1997,6 +1999,16 @@ async def rag_agent(dialog, messages, stream=True, **kwargs): except RuntimeError: pass + def _answer_sink(delta, is_think=False): + if delta: + event_queue.put_nowait(("inner_think" if is_think else "answer", delta)) + + def _tool_started_sink(): + event_queue.put_nowait(("tool_started",)) + + rag_tools.answer_sink = _answer_sink + rag_tools.tool_started_sink = _tool_started_sink + async def _drive_stream(): try: stream_iter = chat_mdl.async_chat_streamly_delta(rag_tools.sys_prompt(), messages, gen_conf) @@ -2005,38 +2017,82 @@ async def rag_agent(dialog, messages, stream=True, **kwargs): except Exception: logging.exception("rag_agent: agentic stream failed") finally: - event_queue.put_nowait(("stream_done",)) + loop.call_soon_threadsafe(event_queue.put_nowait, ("stream_done",)) token = set_think_log_sink(_log_sink) drive = asyncio.create_task(_drive_stream()) - last_state = None - log_think_open = False + answer_deltas = [] + answer_started = False + think_closed = False + outer_tool_started = False + + async def _close_think_and_flush_answer(): + nonlocal answer_started, think_closed + if not think_closed: + yield {"answer": "", "reference": {}, "audio_binary": None, "final": False, "end_to_think": True} + think_closed = True + if not answer_started: + answer_started = True + for delta in answer_deltas: + yield {"answer": delta, "reference": {}, "audio_binary": tts(tts_mdl, delta), "final": False} + try: + # The outer model emits this as a synthetic token while it + # invokes the terminal tool. Make it part of the single progress + # block instead of forwarding its marker separately. + yield {"answer": "", "reference": {}, "audio_binary": None, "final": False, "start_to_think": True} while True: item = await event_queue.get() if item[0] == "log": - if not log_think_open: - yield {"answer": "", "reference": {}, "audio_binary": None, "final": False, "start_to_think": True} - log_think_open = True + if think_closed: + continue yield {"answer": item[1] + "\n", "reference": {}, "audio_binary": None, "final": False} continue + if item[0] == "tool_started": + outer_tool_started = True + continue + if item[0] == "answer": + if not answer_started: + async for output in _close_think_and_flush_answer(): + yield output + answer_deltas.append(item[1]) + yield {"answer": item[1], "reference": {}, "audio_binary": tts(tts_mdl, item[1]), "final": False} + continue + if item[0] == "inner_think": + if think_closed: + continue + value = re.sub(r"", "", item[1]) + if value: + yield {"answer": value, "reference": {}, "audio_binary": None, "final": False} + continue if item[0] == "stream_done": break _, kind, value, state = item - if state is not None: - last_state = state - # A real stream event follows the logs -> close the log think block. - if log_think_open: - yield {"answer": "", "reference": {}, "audio_binary": None, "final": False, "end_to_think": True} - log_think_open = False - if kind == "marker": - flags = {"start_to_think": True} if value == "" else {"end_to_think": True} - yield {"answer": "", "reference": {}, "audio_binary": None, "final": False, **flags} + if kind != "text" or not value: + # The outer model's think markers are folded into the one + # block opened above; they must not create extra markers. continue - yield {"answer": value, "reference": {}, "audio_binary": tts(tts_mdl, value), "final": False} - if log_think_open: - yield {"answer": "", "reference": {}, "audio_binary": None, "final": False, "end_to_think": True} - log_think_open = False + + # Forward outer-model thinking text, including any tail that + # arrives after the research-complete log. The state tells us + # whether this is still inside the model's think section. + # Once that section is closed and the terminal tool has + # started, subsequent text is the aggregate tool result and is + # intentionally ignored. + if state is not None and state.in_think: + value = re.sub(r"", "", value) + if value: + yield {"answer": value, "reference": {}, "audio_binary": None, "final": False} + elif not outer_tool_started: + # Some providers omit explicit reasoning metadata and + # emit plain text before the tool call. Preserve it as + # outer thinking for compatibility with async_chat. + value = re.sub(r"", "", value) + if value: + yield {"answer": value, "reference": {}, "audio_binary": None, "final": False} + if not think_closed: + async for output in _close_think_and_flush_answer(): + yield output finally: reset_think_log_sink(token) if not drive.done(): @@ -2048,12 +2104,12 @@ async def rag_agent(dialog, messages, stream=True, **kwargs): except Exception: logging.exception("rag_agent: drive task error") - full_answer = last_state.full_text if last_state else "" - if full_answer: - final = await decorate_answer(_extract_visible_answer(full_answer)) - final["final"] = True - final["audio_binary"] = None - yield final + answer_text = "".join(answer_deltas) + final = await decorate_answer(answer_text) + final["final"] = True + final["answer"] = "" + final["audio_binary"] = None + yield final else: answer = await chat_mdl.async_chat(rag_tools.sys_prompt(), messages, gen_conf) user_content = messages[-1].get("content", "[content not available]") diff --git a/rag/advanced_rag/agentic_rag.py b/rag/advanced_rag/agentic_rag.py index 2c7a0e61c1..ca438b7a66 100644 --- a/rag/advanced_rag/agentic_rag.py +++ b/rag/advanced_rag/agentic_rag.py @@ -32,6 +32,7 @@ the fast non-tool-calling path. import logging import re +from collections.abc import Callable from typing import Any, List import json_repair @@ -42,7 +43,7 @@ from api.db.services.llm_service import LLMBundle from common import settings from common.misc_utils import thread_pool_exec from common.token_utils import num_tokens_from_string -from rag.advanced_rag.agentic_rag_graph import _strip_think_stream +from rag.advanced_rag.agentic_rag_graph import _split_think_stream from rag.app.tag import label_question from rag.llm.tool_decorator import tool from rag.prompts.generator import ( @@ -79,6 +80,7 @@ class RAGTools: meta_data_filter: dict | None = None, doc_scope: List[str] | None = None, user_defined_prompts: dict | None = None, + empty_response: str = "", do_refer: bool | None = True, thinking_mode: str = "medium", ): @@ -110,8 +112,14 @@ class RAGTools: self.meta_data_filter = meta_data_filter self.doc_scope = list(dict.fromkeys(doc_scope)) if doc_scope is not None else None self.user_defined_prompts = user_defined_prompts or {} - self.kbinfos = {"chunks": [], "doc_aggs": []} + self.empty_response = empty_response self.do_refer = do_refer + # Optional sink used by the outer agent stream to preserve the final + # answer deltas produced by the inner research graph. The tool API + # still returns the complete string to the caller, but the stream + # endpoint can forward the original deltas instead of that aggregate. + self.answer_sink: Callable[[str, bool], None] | None = None + self.tool_started_sink: Callable[[], None] | None = None # Citation pool shared with the final-answer node: the graph publishes # the chunks it actually used here (in the SAME order the answer's # ``[ID:n]`` markers index), so the caller can resolve references. @@ -562,11 +570,15 @@ class RAGTools: """ from rag.advanced_rag.agentic_rag_graph import run_agentic_rag + if self.tool_started_sink is not None: + self.tool_started_sink() messages = [{"role": "user", "content": question}] if question else [] final = "" - async for delta in _strip_think_stream(run_agentic_rag(self, messages)): - if isinstance(delta, str): + async for kind, delta in _split_think_stream(run_agentic_rag(self, messages)): + if kind == "answer": final += delta + if self.answer_sink is not None: + self.answer_sink(delta, kind == "think") for p, r in [(r"\(\**(ID:\d)\**\)", "[\1]")]: final = re.sub(p, r, final) return final diff --git a/rag/advanced_rag/agentic_rag_graph.py b/rag/advanced_rag/agentic_rag_graph.py index 036b432103..86ce6f4468 100644 --- a/rag/advanced_rag/agentic_rag_graph.py +++ b/rag/advanced_rag/agentic_rag_graph.py @@ -33,6 +33,7 @@ from __future__ import annotations import asyncio import json +import re import logging from typing import Any, TypedDict @@ -85,45 +86,65 @@ def _partial_tag_tail(s: str, tag: str) -> int: return 0 -async def _strip_think_stream(stream): - """Strip ... spans from a token stream.""" +async def _split_think_stream(stream): + """Split model deltas into ``think`` and ``answer`` text. + + Besides ordinary ``...`` streams, some providers emit the + opening tag only on the first reasoning delta and append ```` to + every subsequent delta. An unmatched closing tag therefore still marks + the text before it as reasoning. + """ buf = "" in_think = False + async for token in stream: if not isinstance(token, str): - yield token + _LOG.warning("Ignoring non-string agentic RAG stream item of type %s", type(token).__name__) continue buf += token - out = [] + while buf: - if not in_think: - idx = buf.find(_THINK_OPEN) - if idx == -1: - hold = _partial_tag_tail(buf, _THINK_OPEN) - if hold: - out.append(buf[: len(buf) - hold]) - buf = buf[len(buf) - hold :] - else: - out.append(buf) - buf = "" - break - out.append(buf[:idx]) - buf = buf[idx + len(_THINK_OPEN) :] - in_think = True - else: - idx = buf.find(_THINK_CLOSE) - if idx != -1: - buf = buf[idx + len(_THINK_CLOSE) :] + if in_think: + close_idx = buf.find(_THINK_CLOSE) + if close_idx >= 0: + if close_idx: + yield "think", buf[:close_idx] + buf = buf[close_idx + len(_THINK_CLOSE) :] in_think = False continue + hold = _partial_tag_tail(buf, _THINK_CLOSE) + safe = buf[: len(buf) - hold] if hold else buf + if safe: + yield "think", safe buf = buf[len(buf) - hold :] if hold else "" break - piece = "".join(out) - if piece: - yield piece - if buf and not in_think: - yield buf + + open_idx = buf.find(_THINK_OPEN) + close_idx = buf.find(_THINK_CLOSE) + + if close_idx >= 0 and (open_idx < 0 or close_idx < open_idx): + if close_idx: + yield "think", buf[:close_idx] + buf = buf[close_idx + len(_THINK_CLOSE) :] + continue + + if open_idx >= 0: + if open_idx: + yield "answer", buf[:open_idx] + buf = buf[open_idx + len(_THINK_OPEN) :] + in_think = True + continue + + hold = max(_partial_tag_tail(buf, _THINK_OPEN), _partial_tag_tail(buf, _THINK_CLOSE)) + safe = buf[: len(buf) - hold] if hold else buf + if safe: + yield "answer", safe + buf = buf[len(buf) - hold :] if hold else "" + break + + if buf: + yield ("think" if in_think else "answer"), re.sub(r"", "", buf) # ── Graph construction ── @@ -235,22 +256,19 @@ def build_agentic_graph(tools, token_queue: asyncio.Queue, gen_conf: dict | None tools.kbinfos = kbinfos - # Abstain - if abstain: - msg = "I cannot answer this question based on the available information." - token_queue.put_nowait(msg) - return {"final_answer": msg} - - # Empty result - if empty_result or not kbinfos["chunks"]: - msg = "I don't have enough information based on the available sources." - token_queue.put_nowait(msg) - return {"final_answer": msg} + no_evidence = abstain or empty_result or not kbinfos["chunks"] + if no_evidence and tools.empty_response: + _LOG.info("[Composing the answer] No supporting evidence was found; returning the configured empty response without calling the answer model.") + token_queue.put_nowait(tools.empty_response) + return {"final_answer": tools.empty_response} # Build evidence evidence = kb_prompt(kbinfos, tools.chat_mdl.max_length) parts = [f"Question:\n{question}\n"] + if no_evidence: + parts.append("No supporting evidence was retrieved. State clearly that the available sources are insufficient, and do not answer from general knowledge.\n") + # Include pre_summary from agent results if available pre_summary = kbinfos.get("pre_summary") if pre_summary: