feat(chat): add Querit web search provider (#17813)

rag/advanced_rag/agentic_rag.py

@@ -55,7 +55,7 @@ from rag.prompts.generator import (
     sufficiency_select,
 )
 from api.db.db_models import Document, Knowledgebase
-from rag.utils.tavily_conn import Tavily
+from rag.utils.web_search_conn import WebSearchProvider
 
 
 # Tokens held back from the model's context when fitting retrieved evidence
@@ -75,7 +75,7 @@ class RAGTools:
         embed_mdl: LLMBundle | None = None,
         kb_ids: List[str] | None = None,
         kbs: list[Knowledgebase] | None = None,
-        tav: Tavily | None = None,
+        web_search: WebSearchProvider | None = None,
         meta_data_filter: dict | None = None,
         doc_scope: List[str] | None = None,
         user_defined_prompts: dict | None = None,
@@ -106,7 +106,7 @@ class RAGTools:
             for kb in kbs:
                 _exclude_sql_kb(kb)
 
-        self.tav = tav
+        self.web_search = web_search
         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 {}
@@ -137,7 +137,7 @@ class RAGTools:
         return bool(self.sql_kbs and self.field_map)
 
     def has_web(self) -> bool:
-        return self.tav is not None
+        return self.web_search is not None
 
     def has_llm(self) -> bool:
         return self.chat_mdl is not None
@@ -423,15 +423,15 @@ class RAGTools:
         return {"chunks": kbinfos.get("chunks", []), "doc_aggs": kbinfos.get("doc_aggs", [])}
 
     async def web_retrieve(self, query: str) -> dict[str, list]:
-        """Retrieve chunks from the public web (Tavily). Raw kbinfos shape."""
-        if self.tav is None:
+        """Retrieve chunks from the public web. Raw kbinfos shape."""
+        if self.web_search is None:
             return {"chunks": [], "doc_aggs": []}
         try:
-            tav_res = await thread_pool_exec(self.tav.retrieve_chunks, query)
+            web_res = await thread_pool_exec(self.web_search.retrieve_chunks, query)
         except Exception:
             logging.exception("web_retrieve failed")
             return {"chunks": [], "doc_aggs": []}
-        return {"chunks": tav_res.get("chunks", []), "doc_aggs": tav_res.get("doc_aggs", [])}
+        return {"chunks": web_res.get("chunks", []), "doc_aggs": web_res.get("doc_aggs", [])}
 
     async def structured_retrieve(self, question: str) -> dict[str, Any]:
         """Query the structured (tabular) KBs by translating to SQL.

rag/advanced_rag/harness/tools/search.py

@@ -768,8 +768,8 @@ async def web_search(tools, query: str, keywords: str = "") -> dict:
         from common.misc_utils import thread_pool_exec
 
         effective_query = f"{query} {keywords}".strip() if keywords else query
-        tav_res = await thread_pool_exec(tools.tav.retrieve_chunks, effective_query)
-        return {"chunks": tav_res.get("chunks", []), "doc_aggs": tav_res.get("doc_aggs", [])}
+        web_res = await thread_pool_exec(tools.web_search.retrieve_chunks, effective_query)
+        return {"chunks": web_res.get("chunks", []), "doc_aggs": web_res.get("doc_aggs", [])}
     except Exception:
         _LOG.exception("web_search failed")
         return {"chunks": [], "doc_aggs": []}

rag/advanced_rag/tree_structured_query_decomposition_retrieval.py

@@ -19,7 +19,7 @@ from functools import partial
 from api.db.services.llm_service import LLMBundle
 from rag.prompts import kb_prompt
 from rag.prompts.generator import sufficiency_check, multi_queries_gen
-from rag.utils.tavily_conn import Tavily
+from rag.utils.web_search_conn import create_web_search_provider
 from timeit import default_timer as timer
 
 
@@ -49,13 +49,13 @@ class TreeStructuredQueryDecompositionRetrieval:
         except Exception as e:
             logging.error(f"Knowledge base retrieval error: {e}")
 
-        # 2. Web retrieval (if Tavily API is configured)
+        # 2. Web retrieval (if a web search provider is configured)
         try:
-            if self.internet_enabled and self.prompt_config.get("tavily_api_key"):
-                tav = Tavily(self.prompt_config["tavily_api_key"])
-                tav_res = tav.retrieve_chunks(search_query)
-                kbinfos["chunks"].extend(tav_res["chunks"])
-                kbinfos["doc_aggs"].extend(tav_res["doc_aggs"])
+            web_search = create_web_search_provider(self.prompt_config) if self.internet_enabled else None
+            if web_search:
+                web_res = web_search.retrieve_chunks(search_query)
+                kbinfos["chunks"].extend(web_res["chunks"])
+                kbinfos["doc_aggs"].extend(web_res["doc_aggs"])
         except Exception as e:
             logging.error(f"Web retrieval error: {e}")
 

rag/utils/querit_conn.py

@@ -0,0 +1,125 @@
+#
+#  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+#
+
+import logging
+from typing import Any
+
+import requests
+
+from common.http_client import DEFAULT_TIMEOUT
+from common.misc_utils import get_uuid
+from rag.nlp import rag_tokenizer
+
+logger = logging.getLogger(__name__)
+
+QUERIT_SEARCH_URL = "https://api.querit.ai/v1/search"
+
+
+class Querit:
+    def __init__(self, api_key: str):
+        self.api_key = api_key
+
+    def search(self, query: str) -> list[dict[str, Any]]:
+        try:
+            response = requests.post(
+                QUERIT_SEARCH_URL,
+                headers={
+                    "Accept": "application/json",
+                    "Authorization": f"Bearer {self.api_key}",
+                    "Content-Type": "application/json",
+                },
+                json={
+                    "query": query,
+                    "count": 6,
+                    "chunksPerDoc": 1,
+                },
+                timeout=DEFAULT_TIMEOUT,
+            )
+            response.raise_for_status()
+            response_data = response.json()
+            if not isinstance(response_data, dict):
+                raise TypeError("Querit API response must be a JSON object.")
+
+            results_container = response_data.get("results", {})
+            if not isinstance(results_container, dict):
+                raise TypeError("Querit API response field results must be an object.")
+            results = results_container.get("result", [])
+            if not isinstance(results, list):
+                raise TypeError("Querit API response field results.result must be an array.")
+
+            normalized_results = []
+            for result in results:
+                if not isinstance(result, dict):
+                    continue
+                content = _querit_text(result.get("snippet"))
+                if not content:
+                    continue
+                normalized_results.append(
+                    {
+                        "url": _querit_text(result.get("url")),
+                        "title": _querit_text(result.get("title")),
+                        "content": content,
+                        "score": 1.0,
+                    }
+                )
+            return normalized_results
+        except (requests.RequestException, TypeError, ValueError) as error:
+            logger.error("Querit search failed: %s", _safe_error_message(error, self.api_key))
+            return []
+
+    def retrieve_chunks(self, question: str) -> dict[str, list]:
+        chunks = []
+        doc_aggs = []
+        logger.info("[Querit]Q: %s", question)
+        for result in self.search(question):
+            chunk_id = get_uuid()
+            chunks.append(
+                {
+                    "chunk_id": chunk_id,
+                    "content_ltks": rag_tokenizer.tokenize(result["content"]),
+                    "content_with_weight": result["content"],
+                    "doc_id": chunk_id,
+                    "docnm_kwd": result["title"],
+                    "kb_id": [],
+                    "important_kwd": [],
+                    "image_id": "",
+                    "similarity": result["score"],
+                    "vector_similarity": 1.0,
+                    "term_similarity": 0,
+                    "vector": [],
+                    "positions": [],
+                    "url": result["url"],
+                }
+            )
+            doc_aggs.append(
+                {
+                    "doc_name": result["title"],
+                    "doc_id": chunk_id,
+                    "count": 1,
+                    "url": result["url"],
+                }
+            )
+            logger.info("[Querit]R: %s...", result["content"][:128])
+        return {"chunks": chunks, "doc_aggs": doc_aggs}
+
+
+def _querit_text(value: Any) -> str:
+    return "" if value is None else str(value)
+
+
+def _safe_error_message(error: Exception, api_key: str) -> str:
+    message = str(error) or error.__class__.__name__
+    return message.replace(api_key, "[REDACTED]") if api_key else message

rag/utils/web_search_conn.py

@@ -0,0 +1,66 @@
+#
+#  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+#
+
+import logging
+from typing import Protocol
+
+from rag.utils.querit_conn import Querit
+from rag.utils.tavily_conn import Tavily
+
+WEB_SEARCH_PROVIDER_TAVILY = "tavily"
+WEB_SEARCH_PROVIDER_QUERIT = "querit"
+
+logger = logging.getLogger(__name__)
+
+
+class WebSearchProvider(Protocol):
+    def retrieve_chunks(self, question: str) -> dict[str, list]:
+        """Return web results in RAGFlow's chunk and document aggregate shape."""
+
+
+def _get_api_key(prompt_config: dict, field: str) -> str:
+    api_key = prompt_config.get(field)
+    return api_key.strip() if isinstance(api_key, str) else ""
+
+
+def has_web_search_provider(prompt_config: dict | None) -> bool:
+    if not prompt_config:
+        return False
+    provider = prompt_config.get("web_search_provider", WEB_SEARCH_PROVIDER_TAVILY)
+    if provider == WEB_SEARCH_PROVIDER_TAVILY:
+        return bool(_get_api_key(prompt_config, "tavily_api_key"))
+    if provider == WEB_SEARCH_PROVIDER_QUERIT:
+        return bool(_get_api_key(prompt_config, "querit_api_key"))
+    return False
+
+
+def create_web_search_provider(prompt_config: dict | None) -> WebSearchProvider | None:
+    if not prompt_config:
+        logger.debug("Web search provider resolution: provider=none status=disabled")
+        return None
+
+    provider = prompt_config.get("web_search_provider", WEB_SEARCH_PROVIDER_TAVILY)
+    if provider not in (WEB_SEARCH_PROVIDER_TAVILY, WEB_SEARCH_PROVIDER_QUERIT):
+        logger.debug("Web search provider resolution: provider=%s status=invalid", provider)
+        return None
+    if not has_web_search_provider(prompt_config):
+        logger.debug("Web search provider resolution: provider=%s status=disabled", provider)
+        return None
+
+    logger.debug("Web search provider resolution: provider=%s status=resolved", provider)
+    if provider == WEB_SEARCH_PROVIDER_QUERIT:
+        return Querit(_get_api_key(prompt_config, "querit_api_key"))
+    return Tavily(_get_api_key(prompt_config, "tavily_api_key"))