From d236fc4437532cb4b506317059eeda815cb32c13 Mon Sep 17 00:00:00 2001 From: Wang Qi Date: Tue, 4 Aug 2026 10:35:21 +0800 Subject: [PATCH] Let agentic rag search honor metadata filter (#17731) --- api/db/services/dialog_service.py | 103 ++++++++++++------- rag/advanced_rag/agentic_rag.py | 19 +++- rag/advanced_rag/harness/pipeline.py | 9 +- rag/advanced_rag/harness/tools/navigation.py | 17 ++- rag/advanced_rag/harness/tools/search.py | 24 +++-- 5 files changed, 123 insertions(+), 49 deletions(-) diff --git a/api/db/services/dialog_service.py b/api/db/services/dialog_service.py index aa4c3caf50..f385521d69 100644 --- a/api/db/services/dialog_service.py +++ b/api/db/services/dialog_service.py @@ -652,13 +652,24 @@ async def async_chat(dialog, messages, stream=True, **kwargs): attachments_ = "\n\n".join(text_attachments) prompt_config = dialog.prompt_config + if dialog.meta_data_filter: + attachments = await apply_meta_data_filter( + dialog.meta_data_filter, + None, + questions[-1], + chat_mdl, + attachments, + kb_ids=dialog.kb_ids, + metas_loader=lambda: DocMetadataService.get_flatted_meta_by_kbs(dialog.kb_ids), + ) + include_reference_metadata, metadata_fields = _resolve_reference_metadata(prompt_config, request_payload=kwargs) field_map = KnowledgebaseService.get_field_map(dialog.kb_ids) logging.debug(f"field_map retrieved: {field_map}") # try to use sql if field mapping is good to go if field_map: logging.debug("Use SQL to retrieval:{}".format(questions[-1])) - ans = await use_sql(questions[-1], field_map, dialog.tenant_id, chat_mdl, prompt_config.get("quote", True), dialog.kb_ids) + ans = await use_sql(questions[-1], field_map, dialog.tenant_id, chat_mdl, prompt_config.get("quote", True), dialog.kb_ids, doc_ids=attachments) # For aggregate queries (COUNT, SUM, etc.), chunks may be empty but answer is still valid if ans and (ans.get("reference", {}).get("chunks") or ans.get("answer")): if include_reference_metadata and ans.get("reference", {}).get("chunks"): @@ -698,17 +709,6 @@ async def async_chat(dialog, messages, stream=True, **kwargs): if prompt_config.get("cross_languages"): questions = [await cross_languages(dialog.tenant_id, dialog.llm_id, questions[0], prompt_config["cross_languages"])] - if dialog.meta_data_filter: - attachments = await apply_meta_data_filter( - dialog.meta_data_filter, - None, - questions[-1], - chat_mdl, - attachments, - kb_ids=dialog.kb_ids, - metas_loader=lambda: DocMetadataService.get_flatted_meta_by_kbs(dialog.kb_ids), - ) - if prompt_config.get("keyword", False): questions[-1] = questions[-1] + "," + await keyword_extraction(chat_mdl, questions[-1]) refine_question_ts = timer() @@ -989,7 +989,7 @@ async def async_chat(dialog, messages, stream=True, **kwargs): return -async def use_sql(question, field_map, tenant_id, chat_mdl, quota=True, kb_ids=None): +async def use_sql(question, field_map, tenant_id, chat_mdl, quota=True, kb_ids=None, doc_ids=None): """Answer a natural-language question by generating and executing SQL against the document index. Detects the active document engine (Infinity, OceanBase, or Elasticsearch), asks the @@ -1003,6 +1003,7 @@ async def use_sql(question, field_map, tenant_id, chat_mdl, quota=True, kb_ids=N chat_mdl: LLM bundle used to generate SQL from the question. quota: Whether to enforce token-quota checks (default True). kb_ids: Optional list of knowledge-base UUIDs to restrict the query scope. + doc_ids: Optional list of document UUIDs to restrict the query scope. Returns: A dict with keys ``answer`` (formatted response string), ``reference`` @@ -1020,12 +1021,19 @@ async def use_sql(question, field_map, tenant_id, chat_mdl, quota=True, kb_ids=N doc_engine = "es" def _assert_valid_uuid(value: str, label: str = "id") -> None: + if label == "doc_id" and str(value) == "-999": + return try: uuid.UUID(str(value)) except (ValueError, AttributeError, TypeError): logger.warning("SQL injection guard rejected invalid %s value (length=%d)", label, len(str(value))) raise ValueError(f"Invalid {label} format: {value!r}") + if isinstance(doc_ids, str): + doc_ids = [doc_id for doc_id in doc_ids.split(",") if doc_id] + else: + doc_ids = [doc_id for doc_id in doc_ids or [] if doc_id] + # Construct the full table name # For Elasticsearch: ragflow_{tenant_id} (kb_id is in WHERE clause) # For Infinity: ragflow_{tenant_id}_{kb_id} (each KB has its own table) @@ -1068,34 +1076,38 @@ async def use_sql(question, field_map, tenant_id, chat_mdl, quota=True, kb_ids=N return sql.rstrip().rstrip(";").strip() def add_kb_filter(sql): - """Inject a validated kb_id WHERE filter into *sql* for ES/OceanBase engines. + """Inject validated scope filters into *sql*. - Infinity encodes the knowledge-base scope in the table name, so this - function is a no-op for that engine. All kb_id values are validated as - canonical UUIDs before interpolation to prevent SQL injection. + Infinity encodes single-KB scope in the table name, so only document + scope is injected there. All ids are validated before interpolation. """ - # Add kb_id filter for ES/OS only (Infinity already has it in table name) - if doc_engine == "infinity" or not kb_ids: + scope_filters = [] + sql_lower = sql.lower() + if doc_engine != "infinity" and kb_ids and "kb_id =" not in sql_lower and "kb_id=" not in sql_lower: + for kid in kb_ids: + _assert_valid_uuid(kid, "kb_id") + if len(kb_ids) == 1: + scope_filters.append(f"kb_id = '{kb_ids[0]}'") + else: + scope_filters.append("(" + " OR ".join([f"kb_id = '{kid}'" for kid in kb_ids]) + ")") + if doc_ids: + for doc_id in doc_ids: + _assert_valid_uuid(doc_id, "doc_id") + if len(doc_ids) == 1: + scope_filters.append(f"doc_id = '{doc_ids[0]}'") + else: + scope_filters.append("(" + " OR ".join([f"doc_id = '{doc_id}'" for doc_id in doc_ids]) + ")") + if not scope_filters: return sql - # Validate all kb_ids are UUIDs before interpolating into SQL - for kid in kb_ids: - _assert_valid_uuid(kid, "kb_id") + scope_filter = " and ".join(scope_filters) + trailing_clause = re.search(r"\b(group\s+by|having|order\s+by|limit|offset)\b", sql, flags=re.IGNORECASE) + insert_pos = trailing_clause.start() if trailing_clause else len(sql) - # Build kb_filter: single KB or multiple KBs with OR - if len(kb_ids) == 1: - kb_filter = f"kb_id = '{kb_ids[0]}'" + if not re.search(r"\bwhere\b", sql, flags=re.IGNORECASE): + sql = sql[:insert_pos].rstrip() + f" WHERE {scope_filter}" + (" " + sql[insert_pos:] if trailing_clause else "") else: - kb_filter = "(" + " OR ".join([f"kb_id = '{kid}'" for kid in kb_ids]) + ")" - - if "where " not in sql.lower(): - o = sql.lower().split("order by") - if len(o) > 1: - sql = o[0] + f" WHERE {kb_filter} order by " + o[1] - else: - sql += f" WHERE {kb_filter}" - elif "kb_id =" not in sql.lower() and "kb_id=" not in sql.lower(): - sql = re.sub(r"\bwhere\b ", f"where {kb_filter} and ", sql, flags=re.IGNORECASE) + sql = sql[:insert_pos].rstrip() + f" and {scope_filter}" + (" " + sql[insert_pos:] if trailing_clause else "") return sql def is_row_count_question(q: str) -> bool: @@ -1881,12 +1893,33 @@ async def rag_agent(dialog, messages, stream=True, **kwargs): thinking_mode = "medium" gen_conf = dialog.llm_setting or {} + doc_scope = None + if "doc_ids" in kwargs: + if isinstance(kwargs["doc_ids"], str): + doc_scope = [doc_id for doc_id in kwargs["doc_ids"].split(",") if doc_id] + elif isinstance(kwargs["doc_ids"], list): + doc_scope = [doc_id for doc_id in kwargs["doc_ids"] if doc_id] + if "doc_ids" in messages[-1]: + doc_scope = [doc_id for doc_id in messages[-1]["doc_ids"] if doc_id] + if dialog.meta_data_filter: + doc_scope = await apply_meta_data_filter( + dialog.meta_data_filter, + None, + messages[-1].get("content", ""), + chat_mdl, + doc_scope, + kb_ids=dialog.kb_ids, + metas_loader=lambda: DocMetadataService.get_flatted_meta_by_kbs(dialog.kb_ids), + ) + rag_tools = RAGTools( tenant_ids, chat_mdl, embed_mdl=embd_mdl, kb_ids=dialog.kb_ids, tav=Tavily(prompt_config.get("tavily_api_key")) if use_web_search else None, + meta_data_filter=dialog.meta_data_filter, + doc_scope=doc_scope, do_refer=False, thinking_mode=thinking_mode, ) diff --git a/rag/advanced_rag/agentic_rag.py b/rag/advanced_rag/agentic_rag.py index 32d0db729b..04ea22dde5 100644 --- a/rag/advanced_rag/agentic_rag.py +++ b/rag/advanced_rag/agentic_rag.py @@ -77,6 +77,7 @@ class RAGTools: kbs: list[Knowledgebase] | None = None, tav: Tavily | None = None, meta_data_filter: dict | None = None, + doc_scope: List[str] | None = None, user_defined_prompts: dict | None = None, do_refer: bool | None = True, thinking_mode: str = "medium", @@ -107,6 +108,7 @@ class RAGTools: self.tav = tav 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.do_refer = do_refer @@ -140,6 +142,14 @@ class RAGTools: def has_llm(self) -> bool: return self.chat_mdl is not None + def scoped_doc_ids(self, doc_scope: List[str] | None = None) -> List[str] | None: + if self.doc_scope is None: + return doc_scope + if not doc_scope: + return list(self.doc_scope) + allowed = set(self.doc_scope) + return [doc_id for doc_id in doc_scope if doc_id in allowed] + async def _fit_messages(self, system: str, user: str) -> list: """Fit system+user messages into the model's context window.""" from rag.prompts.generator import form_message, message_fit_in @@ -369,6 +379,9 @@ class RAGTools: keywords = ",".join(keywords) logging.info(f"@retrieve: {question}@{keywords}") + doc_scope = self.scoped_doc_ids(doc_scope) + if doc_scope == ["-999"]: + return {"chunks": [], "doc_aggs": []} if doc_scope: candidates = [d for d in doc_scope if isinstance(d, str)] known = await thread_pool_exec(self._filter_known_doc_ids, candidates) @@ -376,6 +389,8 @@ class RAGTools: if valid: doc_scope = valid else: + if self.doc_scope is not None: + return {"chunks": [], "doc_aggs": []} if candidates: logging.warning("retrieve: every supplied doc ID was unknown; falling back to unfiltered retrieval") doc_scope = None @@ -434,7 +449,7 @@ class RAGTools: sql_kb_ids = [kb.id for kb in self.sql_kbs] tenant_id = self.sql_kbs[0].tenant_id try: - ans = await use_sql(question, self.field_map, tenant_id, self.chat_mdl, quota=True, kb_ids=sql_kb_ids) + ans = await use_sql(question, self.field_map, tenant_id, self.chat_mdl, quota=True, kb_ids=sql_kb_ids, doc_ids=self.scoped_doc_ids()) except Exception as e: logging.exception(f"structured_retrieve: use_sql failed: {e}") return {"answer": "", "chunks": [], "doc_aggs": []} @@ -491,6 +506,8 @@ class RAGTools: """Fetch a whole document's chunks in reading order (raw kbinfos).""" if not self.kb_ids: return {"chunks": [], "doc_aggs": []} + if self.doc_scope is not None and doc_id not in self.doc_scope: + return {"chunks": [], "doc_aggs": []} resolved = await thread_pool_exec(self._resolve_doc_tenant, doc_id) if resolved is None: logging.warning(f"fetch_full_document: doc_id {doc_id!r} not in any bound KB — refusing to fetch") diff --git a/rag/advanced_rag/harness/pipeline.py b/rag/advanced_rag/harness/pipeline.py index a8ae567528..302a95e312 100644 --- a/rag/advanced_rag/harness/pipeline.py +++ b/rag/advanced_rag/harness/pipeline.py @@ -13,7 +13,7 @@ _LOG = logging.getLogger(__name__) # (``dataset_navigation_by_tree``) has produced a relevant-document set, these # inherit it as their ``doc_scope`` unless the caller passed one explicitly, so # a follow-up search stays within the routed docs instead of re-scanning the KB. -_DOC_SCOPE_CONSUMERS = {"ontology_navigate", "mindmap_navigate", "graph_explore", "hybrid_search"} +_DOC_SCOPE_CONSUMERS = {"ontology_navigate", "mindmap_navigate", "graph_explore", "hybrid_search", "vector_search", "bm25_search", "structured_query", "dataset_navigation_by_tree"} class Pipeline: @@ -30,7 +30,7 @@ class Pipeline: self.compilation_map = compilation_map or {} self.trace: list[dict] = [] # Latest relevant-document set produced by a routing tool this run. - self._routed_docs: list[str] = [] + self._routed_docs: list[str] = list(getattr(rag_tools, "doc_scope", None) or []) async def execute(self, tool_name: str, **kwargs) -> ToolResult: """Execute a registered tool by name.""" @@ -58,7 +58,10 @@ class Pipeline: # document IDs; remember them so the scope-consuming tools above can # inherit them on later turns. if result.docs: - self._routed_docs = list(result.docs) + if hasattr(self.tools, "scoped_doc_ids"): + self._routed_docs = self.tools.scoped_doc_ids(list(result.docs)) or [] + else: + self._routed_docs = list(result.docs) # Feed the shared citation pool: agent searches go through the # pipeline, so without this their evidence never reaches kbinfos and # the final answer has nothing to cite. diff --git a/rag/advanced_rag/harness/tools/navigation.py b/rag/advanced_rag/harness/tools/navigation.py index f3127452f5..60f480f2a7 100644 --- a/rag/advanced_rag/harness/tools/navigation.py +++ b/rag/advanced_rag/harness/tools/navigation.py @@ -314,6 +314,8 @@ async def ontology_navigate(tools, topic: str, keywords: str = "", doc_scope: li :returns: ``{"answer": "", "chunks": [...], "doc_aggs": [...]}`` """ + if hasattr(tools, "scoped_doc_ids"): + doc_scope = tools.scoped_doc_ids(doc_scope) if not doc_scope: doc_scope = [] _LOG.info(f'[Ontology navigation] Looking through the document catalog for "{topic}" (keywords: {keywords}) in doc: {len(doc_scope)}') @@ -332,6 +334,8 @@ async def mindmap_navigate(tools, topic: str, keywords: str = "", doc_scope: lis :returns: ``{"answer": "", "chunks": [...], "doc_aggs": [...]}`` """ + if hasattr(tools, "scoped_doc_ids"): + doc_scope = tools.scoped_doc_ids(doc_scope) if not doc_scope: doc_scope = [] _LOG.info(f'[Mindmap navigation] Following the concept mindmap for "{topic}" (keywords: {keywords}) in doc: {len(doc_scope)}') @@ -418,7 +422,7 @@ async def _ask_nav_select(tools, query: str, items: list[dict], noun: str, max_i return out -async def _collect_nav_leaves(dataset_api_service, clusters: list[dict]) -> list[dict]: +async def _collect_nav_leaves(dataset_api_service, clusters: list[dict], doc_scope: list[str] | None = None) -> list[dict]: """BFS from the selected clusters down to their document leaves. Each cluster carries ``name`` + ``kb``. A node's children are either document @@ -429,6 +433,7 @@ async def _collect_nav_leaves(dataset_api_service, clusters: list[dict]) -> list seen_docs: set[str] = set() seen_nodes: set[tuple] = set() frontier: list[tuple] = [(c["kb"], c["name"], 0) for c in clusters if c.get("name")] + allowed_docs = set(doc_scope or []) while frontier and len(leaves) < _NAV_TREE_MAX_LEAVES: kb, name, depth = frontier.pop(0) @@ -446,7 +451,7 @@ async def _collect_nav_leaves(dataset_api_service, clusters: list[dict]) -> list for item in data.get("items") or []: if item.get("type") == "doc": did = str(item.get("doc_id") or "").strip() - if did and did not in seen_docs: + if did and (not allowed_docs or did in allowed_docs) and did not in seen_docs: seen_docs.add(did) leaves.append({**item, "kb": kb}) if len(leaves) >= _NAV_TREE_MAX_LEAVES: @@ -481,6 +486,8 @@ async def dataset_navigation_by_tree(tools, topic: str, keywords: str = "", doc_ query = " ".join(part for part in ((topic or "").strip(), (keywords or "").strip()) if part).strip() if not query: return [] + if hasattr(tools, "scoped_doc_ids"): + doc_scope = tools.scoped_doc_ids(doc_scope) _LOG.info('[Dataset navigation] Walking the dataset tree for "%s"', query) @@ -514,7 +521,7 @@ async def dataset_navigation_by_tree(tools, topic: str, keywords: str = "", doc_ _LOG.info("[Dataset navigation] %d/%d cluster(s) selected.", len(selected_clusters), len(clusters)) # 3. Descend the selected clusters to their document leaves. - leaves = await _collect_nav_leaves(dataset_api_service, selected_clusters) + leaves = await _collect_nav_leaves(dataset_api_service, selected_clusters, doc_scope) if not leaves: _LOG.info("[Dataset navigation] no leaf under selected cluster %s.", _nav_cluster_names(selected_clusters)) return [] @@ -566,6 +573,8 @@ async def _kg_scopes(tools, doc_scope: list[str] | None = None): """ from common.misc_utils import thread_pool_exec + if hasattr(tools, "scoped_doc_ids"): + doc_scope = tools.scoped_doc_ids(doc_scope) if doc_scope: by_kb: dict[tuple, list[str]] = {} for doc_id in doc_scope: @@ -759,6 +768,8 @@ async def graph_explore(tools, query: str, keywords: str = "", doc_scope: list[s from rag.advanced_rag.harness.tools.search import _narrow_by_keywords _empty = {"answer": "", "chunks": [], "doc_aggs": []} + if hasattr(tools, "scoped_doc_ids"): + doc_scope = tools.scoped_doc_ids(doc_scope) _LOG.info(f'[Graph exploration] Exploring the knowledge graph for "{query}" (keywords: {keywords})') scopes = await _kg_scopes(tools, doc_scope) diff --git a/rag/advanced_rag/harness/tools/search.py b/rag/advanced_rag/harness/tools/search.py index 822c7812b1..4f26e974ab 100644 --- a/rag/advanced_rag/harness/tools/search.py +++ b/rag/advanced_rag/harness/tools/search.py @@ -184,6 +184,8 @@ async def hybrid_search(tools, query: str, kb_ids: list[str] | None = None, top_ if not tools.kb_ids and not kb_ids: return {"chunks": [], "doc_aggs": []} target_ids = kb_ids or tools.kb_ids + if hasattr(tools, "scoped_doc_ids"): + doc_scope = tools.scoped_doc_ids(doc_scope) _LOG.info(f'[Hybrid search] Searching the knowledge base for "{query}" (keywords: {keywords})') # Query expansion: append the formalized-question keywords + close synonyms @@ -223,13 +225,13 @@ async def hybrid_search(tools, query: str, kb_ids: list[str] | None = None, top_ _LOG.info(f"[Hybrid search] Kept {len(kbinfos['chunks'])} of {length} passage(s) that actually mention the keywords.") 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) + await _expand_with_compiled(tools, query, keywords, kbinfos, doc_scope) if cache is not None: cache[cache_key] = kbinfos return kbinfos -async def vector_search(tools, query: str, kb_ids: list[str] | None = None, top_n: int = 12, keywords: str = "") -> dict: +async def vector_search(tools, query: str, kb_ids: list[str] | None = None, top_n: int = 12, keywords: str = "", doc_scope: list[str] | None = None) -> dict: if not tools.embed_mdl: _LOG.warning("vector_search: no embed_mdl available") return {"chunks": [], "doc_aggs": []} @@ -237,6 +239,8 @@ async def vector_search(tools, query: str, kb_ids: list[str] | None = None, top_ _LOG.info(f'[Vector search] Searching by meaning for "{query}" (keywords: {keywords})') effective_query = f"{query} {keywords}".strip() if keywords else query target_ids = kb_ids or tools.kb_ids + if hasattr(tools, "scoped_doc_ids"): + doc_scope = tools.scoped_doc_ids(doc_scope) kbinfos = await settings.retriever.retrieval( effective_query, tools.embed_mdl, @@ -248,6 +252,7 @@ async def vector_search(tools, query: str, kb_ids: list[str] | None = None, top_ vector_similarity_weight=1.0, aggs=False, highlight=False, + doc_ids=doc_scope, ) kbinfos = _normalize(kbinfos, tools.tenant_ids) if keywords: @@ -257,10 +262,12 @@ async def vector_search(tools, query: str, kb_ids: list[str] | None = None, top_ return kbinfos -async def bm25_search(tools, query: str, kb_ids: list[str] | None = None, top_n: int = 12, keywords: str = "") -> dict: +async def bm25_search(tools, query: str, kb_ids: list[str] | None = None, top_n: int = 12, keywords: str = "", doc_scope: list[str] | None = None) -> dict: _LOG.info(f'[BM25 search] Searching by keyword for "{query}" (keywords: {keywords})') target_ids = kb_ids or tools.kb_ids effective_query = f"{query} {keywords}".strip() if keywords else query + if hasattr(tools, "scoped_doc_ids"): + doc_scope = tools.scoped_doc_ids(doc_scope) kbinfos = await settings.retriever.retrieval( effective_query, None, @@ -272,6 +279,7 @@ async def bm25_search(tools, query: str, kb_ids: list[str] | None = None, top_n: vector_similarity_weight=0, aggs=False, highlight=False, + doc_ids=doc_scope, ) kbinfos = _normalize(kbinfos, tools.tenant_ids) if keywords: @@ -284,7 +292,7 @@ async def bm25_search(tools, query: str, kb_ids: list[str] | None = None, top_n: # ─── Compiled product expansion (zero-LLM, used by hybrid_search with use_compiled=True) ─── -async def _expand_with_compiled(tools, query: str, keywords: str, kbinfos: dict) -> None: +async def _expand_with_compiled(tools, query: str, keywords: str, kbinfos: dict, doc_scope: list[str] | None = None) -> None: """Zero-LLM compiled-product expansion: page_index → tree → KG. For each bound KB, searches compiled entity rows matching the query, @@ -294,7 +302,7 @@ async def _expand_with_compiled(tools, query: str, keywords: str, kbinfos: dict) before = len(kbinfos.get("chunks", [])) seen_ids = {c.get("chunk_id") or c.get("id") for c in kbinfos.get("chunks", [])} - scopes = await _kg_scopes(tools) + scopes = await _kg_scopes(tools, doc_scope) if not scopes: return @@ -764,7 +772,7 @@ async def web_search(tools, query: str, keywords: str = "") -> dict: return {"chunks": [], "doc_aggs": []} -async def structured_query(tools, query: str, keywords: str = "", kb_ids: list[str] | None = None) -> dict: +async def structured_query(tools, query: str, keywords: str = "", kb_ids: list[str] | None = None, doc_scope: list[str] | None = None) -> dict: """Answer from the structured (tabular) KBs by translating the query to SQL. ``keywords`` is accepted for schema conformance but deliberately unused: the @@ -775,12 +783,14 @@ async def structured_query(tools, query: str, keywords: str = "", kb_ids: list[s sql_kbs = [kb for kb in tools.sql_kbs if kb_ids is None or kb.id in kb_ids] if not sql_kbs: return {"answer": "", "chunks": [], "doc_aggs": []} + if hasattr(tools, "scoped_doc_ids"): + doc_scope = tools.scoped_doc_ids(doc_scope) from api.db.services.dialog_service import use_sql tenant_id = sql_kbs[0].tenant_id sql_kb_ids = [kb.id for kb in sql_kbs] try: - ans = await use_sql(query, tools.field_map, tenant_id, tools.chat_mdl, quota=True, kb_ids=sql_kb_ids) + ans = await use_sql(query, tools.field_map, tenant_id, tools.chat_mdl, quota=True, kb_ids=sql_kb_ids, doc_ids=doc_scope) except Exception: _LOG.exception("structured_query failed") return {"answer": "", "chunks": [], "doc_aggs": []}