Feat/configurable metadata display (#13464)

### What problem does this PR solve?

Currently, RAGFlow's Search and Chat interfaces display only raw
vectorized text chunks during retrieval, without contextual information
about their source documents. Users cannot see document titles, page
numbers, upload dates, or custom metadata fields that would help them
understand and trust the retrieved results.

This PR introduces an **optional metadata display feature** that
enriches retrieved chunks with document-level metadata in both the
Search tab and Chatbot interface.

**Key improvements:**
- **Search results**: Display document metadata as styled badges beneath
chunk snippets
- **Chat citations**: Show metadata in citation popovers and reference
lists for better source context
- **LLM context**: Metadata is injected into the LLM prompt to enable
more accurate, citation-aware responses
- **External API support**: Applications using RAGFlow's SDK retrieval
endpoints (`/v1/retrieval`, `/v1/searchbots/retrieval_test`) can opt-in
via request parameters
- **User control**: Multi-select dropdown UI allows users to choose
which metadata fields to display

**Implementation approach:**
- ✅ Reuses existing `DocMetadataService` infrastructure (no new database
tables or indices)
- ✅ Settings stored in existing JSON configuration fields
(`search_config.reference_metadata`, `prompt_config.reference_metadata`)
- ✅ No database migrations required
- ✅ Disabled by default (fully opt-in and backward-compatible)
- ✅ Dynamic metadata field selection populated from actual document
metadata keys
- ✅ Fixed critical bug where Python's builtin `set()` was shadowed by a
route handler function

**Modified endpoints (all backward-compatible):**
- `POST /v1/retrieval` (Public SDK)
- `POST /v1/searchbots/retrieval_test` (Searchbots)
- `POST /v1/chunk/retrieval_test` (UI/Internal)
- Chat completions endpoints (via `extra_body.reference_metadata` or
`prompt_config`)

### Type of change

- [x] New Feature (non-breaking change which adds functionality)


###Images
-
<img width="879" height="1275" alt="image"
src="https://github.com/user-attachments/assets/95b2d731-31ae-45a1-b081-bf5893f52aeb"
/>
<br><br>
<br><br>

<img width="1532" height="362" alt="image"
src="https://github.com/user-attachments/assets/9cebc65b-b7a7-459f-b25e-3b13fa9b638e"
/>
<br><br>
<br><br>

<img width="2586" height="1320" alt="image"
src="https://github.com/user-attachments/assets/2153d493-d899-461f-a7a9-041391e07776"
/>

---------

Co-authored-by: Cursor Agent <cursoragent@cursor.com>
Co-authored-by: Attili-sys <Attili-sys@users.noreply.github.com>
Co-authored-by: Ahmad Intisar <ahmadintisar@Ahmads-MacBook-M4-Pro.local>

api/apps/restful_apis/openai_api.py

@@ -48,44 +48,35 @@ def _validate_llm_id(llm_id, tenant_id, llm_setting=None):
     return None
 
 
+import logging
+from api.utils.reference_metadata_utils import enrich_chunks_with_document_metadata
+
 def _build_reference_chunks(reference, include_metadata=False, metadata_fields=None):
     chunks = chunks_format(reference)
     if not include_metadata:
+        logging.debug("Skipping document metadata enrichment (include_metadata=False)")
         return chunks
 
-    doc_ids_by_kb = {}
-    for chunk in chunks:
-        kb_id = chunk.get("dataset_id")
-        doc_id = chunk.get("document_id")
-        if not kb_id or not doc_id:
-            continue
-        doc_ids_by_kb.setdefault(kb_id, set()).add(doc_id)
-
-    if not doc_ids_by_kb:
-        return chunks
-
-    meta_by_doc = {}
-    for kb_id, doc_ids in doc_ids_by_kb.items():
-        meta_map = DocMetadataService.get_metadata_for_documents(list(doc_ids), kb_id)
-        if meta_map:
-            meta_by_doc.update(meta_map)
-
+    normalized_fields = None
     if metadata_fields is not None:
-        metadata_fields = {f for f in metadata_fields if isinstance(f, str)}
-        if not metadata_fields:
+        if not isinstance(metadata_fields, list):
+            return chunks
+        normalized_fields = {f for f in metadata_fields if isinstance(f, str)}
+        if not normalized_fields:
             return chunks
 
-    for chunk in chunks:
-        doc_id = chunk.get("document_id")
-        if not doc_id:
-            continue
-        meta = meta_by_doc.get(doc_id)
-        if not meta:
-            continue
-        if metadata_fields is not None:
-            meta = {k: v for k, v in meta.items() if k in metadata_fields}
-        if meta:
-            chunk["document_metadata"] = meta
+    logging.debug(
+        "Enriching %d chunks with document metadata (fields: %s)",
+        len(chunks),
+        "ALL" if normalized_fields is None else list(normalized_fields),
+    )
+
+    enrich_chunks_with_document_metadata(
+        chunks,
+        normalized_fields,
+        kb_field="dataset_id",
+        doc_field="document_id",
+    )
 
     return chunks
 

api/apps/sdk/doc.py

@@ -13,6 +13,7 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 #
+import logging
 from io import BytesIO
 
 from quart import request, send_file
@@ -37,6 +38,18 @@ from rag.prompts.generator import cross_languages, keyword_extraction
 MAXIMUM_OF_UPLOADING_FILES = 256
 
 
+from api.utils.reference_metadata_utils import (
+    enrich_chunks_with_document_metadata,
+    resolve_reference_metadata_preferences,
+)
+
+def _resolve_reference_metadata(req: dict, search_config: dict | None = None):
+    return resolve_reference_metadata_preferences(req, search_config)
+
+def _enrich_chunks_with_document_metadata(chunks: list[dict], metadata_fields=None) -> None:
+    enrich_chunks_with_document_metadata(chunks, metadata_fields)
+
+
 @manager.route("/datasets/<dataset_id>/documents/<document_id>", methods=["GET"])  # noqa: F821
 @token_required
 async def download(tenant_id, dataset_id, document_id):
@@ -450,6 +463,7 @@ async def retrieval_test(tenant_id):
             return get_error_data_result("`highlight` should be a boolean")
     else:
         return get_error_data_result("`highlight` should be a boolean")
+    include_metadata, metadata_fields = _resolve_reference_metadata(req)
     try:
         tenant_ids = list(set([kb.tenant_id for kb in kbs]))
         e, kb = KnowledgebaseService.get_by_id(kb_ids[0])
@@ -508,6 +522,15 @@ async def retrieval_test(tenant_id):
         for c in ranks["chunks"]:
             c.pop("vector", None)
 
+        if include_metadata:
+            logging.info(
+                "sdk.retrieval reference_metadata enabled dataset_ids=%s fields=%s chunks=%s",
+                kb_ids,
+                sorted(metadata_fields) if metadata_fields else None,
+                len(ranks["chunks"]),
+            )
+            enrich_chunks_with_document_metadata(ranks["chunks"], metadata_fields)
+
         ##rename keys
         renamed_chunks = []
         for chunk in ranks["chunks"]:

api/apps/sdk/session.py

@@ -44,6 +44,10 @@ from rag.prompts.template import load_prompt
 from rag.prompts.generator import cross_languages, keyword_extraction
 from common.constants import RetCode, LLMType
 from common import settings
+from api.utils.reference_metadata_utils import (
+    enrich_chunks_with_document_metadata,
+    resolve_reference_metadata_preferences,
+)
 
 
 @token_required
@@ -327,6 +331,7 @@ async def retrieval_test_embedded():
     tenant_id = objs[0].tenant_id
     if not tenant_id:
         return get_error_data_result(message="permission denined.")
+    search_config = {}
 
     async def _retrieval():
         nonlocal similarity_threshold, vector_similarity_weight, top, rerank_id
@@ -337,8 +342,11 @@ async def retrieval_test_embedded():
         meta_data_filter = {}
         chat_mdl = None
         if req.get("search_id", ""):
-            search_config = SearchService.get_detail(req.get("search_id", "")).get("search_config", {})
-            meta_data_filter = search_config.get("meta_data_filter", {})
+            nonlocal search_config
+            detail = SearchService.get_detail(req.get("search_id", ""))
+            if detail:
+                search_config = detail.get("search_config", {})
+                meta_data_filter = search_config.get("meta_data_filter", {})
             if meta_data_filter.get("method") in ["auto", "semi_auto"]:
                 chat_id = search_config.get("chat_id", "")
                 if chat_id:
@@ -414,6 +422,11 @@ async def retrieval_test_embedded():
 
         for c in ranks["chunks"]:
             c.pop("vector", None)
+
+        include_metadata, metadata_fields = _resolve_reference_metadata(req, search_config)
+        if include_metadata:
+            enrich_chunks_with_document_metadata(ranks["chunks"], metadata_fields)
+
         ranks["labels"] = labels
 
         return get_json_result(data=ranks)
@@ -529,3 +542,6 @@ async def mindmap():
         return server_error_response(Exception(mind_map["error"]))
     return get_json_result(data=mind_map)
 
+
+def _resolve_reference_metadata(req, search_config=None):
+    return resolve_reference_metadata_preferences(req, search_config)

api/db/services/dialog_service.py

@@ -33,6 +33,10 @@ from api.db.services.knowledgebase_service import KnowledgebaseService
 from api.db.services.langfuse_service import TenantLangfuseService
 from api.db.services.llm_service import LLMBundle
 from common.metadata_utils import apply_meta_data_filter
+from api.utils.reference_metadata_utils import (
+    enrich_chunks_with_document_metadata,
+    resolve_reference_metadata_preferences,
+)
 from api.db.services.tenant_llm_service import TenantLLMService
 from api.db.joint_services.tenant_model_service import get_model_config_by_id, get_model_config_by_type_and_name, get_tenant_default_model_by_type
 from common.time_utils import current_timestamp, datetime_format
@@ -48,6 +52,16 @@ from rag.utils.tavily_conn import Tavily
 from common.string_utils import remove_redundant_spaces
 from common import settings
 
+def _resolve_reference_metadata(request_payload=None, config=None):
+    return resolve_reference_metadata_preferences(request_payload or {}, config)
+
+def _enrich_chunks_with_document_metadata(chunks, metadata_fields=None):
+    enrich_chunks_with_document_metadata(chunks, metadata_fields)
+
+def _chunk_kb_id_for_doc(row_dict, kb_ids, doc_id):
+    if len(kb_ids or []) == 1:
+        return kb_ids[0]
+    return row_dict.get("kb_id") or row_dict.get("kb_id_kwd")
 
 def _normalize_internet_flag(value):
     if isinstance(value, bool):
@@ -70,6 +84,15 @@ def _should_use_web_search(prompt_config, internet=None):
     return normalized is True
 
 
+def _resolve_reference_metadata(config, request_payload=None):
+    return resolve_reference_metadata_preferences(request_payload or {}, config)
+
+
+def _enrich_chunks_with_document_metadata(chunks, metadata_fields=None):
+    enrich_chunks_with_document_metadata(chunks, metadata_fields)
+
+
+
 class DialogService(CommonService):
     model = Dialog
 
@@ -547,6 +570,7 @@ async def async_chat(dialog, messages, stream=True, **kwargs):
         attachments_ = "\n\n".join(text_attachments)
 
     prompt_config = dialog.prompt_config
+    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
@@ -555,6 +579,14 @@ async def async_chat(dialog, messages, stream=True, **kwargs):
         ans = await use_sql(questions[-1], field_map, dialog.tenant_id, chat_mdl, prompt_config.get("quote", True), dialog.kb_ids)
         # 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"):
+                if len(dialog.kb_ids) != 1 and any(not c.get("kb_id") for c in ans["reference"]["chunks"]):
+                    logging.warning(
+                        "Skipping some _enrich_chunks_with_document_metadata results because "
+                        "dialog.kb_ids has %d entries and use_sql returned chunks without kb_id.",
+                        len(dialog.kb_ids),
+                    )
+                _enrich_chunks_with_document_metadata(ans["reference"]["chunks"], metadata_fields)
             yield ans
             return
         else:
@@ -675,6 +707,14 @@ async def async_chat(dialog, messages, stream=True, **kwargs):
                 if ck["content_with_weight"]:
                     kbinfos["chunks"].insert(0, ck)
 
+    if include_reference_metadata:
+        logging.debug(
+            "reference_metadata enrichment enabled for async_chat: chunk_count=%d metadata_fields=%s",
+            len(kbinfos.get("chunks", [])),
+            metadata_fields,
+        )
+        _enrich_chunks_with_document_metadata(kbinfos.get("chunks", []), metadata_fields)
+
     knowledges = kb_prompt(kbinfos, max_tokens)
     logging.debug("{}->{}".format(" ".join(questions), "\n->".join(knowledges)))
 
@@ -1121,11 +1161,12 @@ Please correct the error and write SQL again using json_extract_string(chunk_dat
 
     docid_idx = set([ii for ii, c in enumerate(tbl["columns"]) if c["name"].lower() == "doc_id"])
     doc_name_idx = set([ii for ii, c in enumerate(tbl["columns"]) if c["name"].lower() in ["docnm_kwd", "docnm"]])
+    kb_id_idx = set([ii for ii, c in enumerate(tbl["columns"]) if c["name"].lower() in ["kb_id", "kb_id_kwd"]])
 
     logging.debug(f"use_sql: All columns: {[(i, c['name']) for i, c in enumerate(tbl['columns'])]}")
-    logging.debug(f"use_sql: docid_idx={docid_idx}, doc_name_idx={doc_name_idx}")
+    logging.debug(f"use_sql: docid_idx={docid_idx}, doc_name_idx={doc_name_idx}, kb_id_idx={kb_id_idx}")
 
-    column_idx = [ii for ii in range(len(tbl["columns"])) if ii not in (docid_idx | doc_name_idx)]
+    column_idx = [ii for ii in range(len(tbl["columns"])) if ii not in (docid_idx | doc_name_idx | kb_id_idx)]
 
     logging.debug(f"use_sql: column_idx={column_idx}")
     logging.debug(f"use_sql: field_map={field_map}")
@@ -1221,8 +1262,11 @@ Please correct the error and write SQL again using json_extract_string(chunk_dat
             where_match = re.search(r"\bwhere\b(.+?)(?:\bgroup by\b|\border by\b|\blimit\b|$)", sql, re.IGNORECASE)
             if where_match:
                 where_clause = where_match.group(1).strip()
-                # Build a query to get doc_id and docnm_kwd with the same WHERE clause
-                chunks_sql = f"select doc_id, docnm_kwd from {table_name} where {where_clause}"
+                # Build a query to get source fields with the same WHERE clause.
+                # Single-KB queries can derive kb_id from the dialog, while multi-KB
+                # ES/OS queries need the row value for metadata enrichment.
+                chunks_kb_column = ", kb_id" if not (kb_ids and len(kb_ids) == 1) else ""
+                chunks_sql = f"select doc_id, {expected_doc_name_column}{chunks_kb_column} from {table_name} where {where_clause}"
                 # Add LIMIT to avoid fetching too many chunks
                 if "limit" not in chunks_sql.lower():
                     chunks_sql += " limit 20"
@@ -1233,8 +1277,18 @@ Please correct the error and write SQL again using json_extract_string(chunk_dat
                         # Build chunks reference - use case-insensitive matching
                         chunks_did_idx = next((i for i, c in enumerate(chunks_tbl["columns"]) if c["name"].lower() == "doc_id"), None)
                         chunks_dn_idx = next((i for i, c in enumerate(chunks_tbl["columns"]) if c["name"].lower() in ["docnm_kwd", "docnm"]), None)
+                        chunks_kb_idx = next((i for i, c in enumerate(chunks_tbl["columns"]) if c["name"].lower() in ["kb_id", "kb_id_kwd"]), None)
                         if chunks_did_idx is not None and chunks_dn_idx is not None:
-                            chunks = [{"doc_id": r[chunks_did_idx], "docnm_kwd": r[chunks_dn_idx]} for r in chunks_tbl["rows"]]
+                            chunks = []
+                            for r in chunks_tbl["rows"]:
+                                chunk = {"doc_id": r[chunks_did_idx], "docnm_kwd": r[chunks_dn_idx]}
+                                row_dict = {chunks_tbl["columns"][i]["name"]: r[i] for i in range(len(chunks_tbl["columns"])) if i < len(r)}
+                                kb_id = _chunk_kb_id_for_doc(row_dict, kb_ids, chunk["doc_id"])
+                                if kb_id:
+                                    chunk["kb_id"] = kb_id
+                                elif chunks_kb_idx is not None:
+                                    chunk["kb_id"] = r[chunks_kb_idx]
+                                chunks.append(chunk)
                             # Build doc_aggs
                             doc_aggs = {}
                             for r in chunks_tbl["rows"]:
@@ -1264,7 +1318,22 @@ Please correct the error and write SQL again using json_extract_string(chunk_dat
     result = {
         "answer": "\n".join([columns, line, rows]),
         "reference": {
-            "chunks": [{"doc_id": r[docid_idx], "docnm_kwd": r[doc_name_idx]} for r in tbl["rows"]],
+            "chunks": [
+                {
+                    key: value
+                    for key, value in {
+                        "doc_id": r[docid_idx],
+                        "docnm_kwd": r[doc_name_idx],
+                        "kb_id": _chunk_kb_id_for_doc(
+                            {tbl["columns"][i]["name"]: r[i] for i in range(len(tbl["columns"])) if i < len(r)},
+                            kb_ids,
+                            r[docid_idx],
+                        ),
+                    }.items()
+                    if value
+                }
+                for r in tbl["rows"]
+            ],
             "doc_aggs": [{"doc_id": did, "doc_name": d["doc_name"], "count": d["count"]} for did, d in doc_aggs.items()],
         },
         "prompt": sys_prompt,
@@ -1414,6 +1483,7 @@ async def async_ask(question, kb_ids, tenant_id, chat_llm_name=None, search_conf
     chat_llm_name = search_config.get("chat_id", chat_llm_name)
     rerank_id = search_config.get("rerank_id", "")
     meta_data_filter = search_config.get("meta_data_filter")
+    include_reference_metadata, metadata_fields = _resolve_reference_metadata(search_config)
 
     kbs = KnowledgebaseService.get_by_ids(kb_ids)
     embedding_list = list(set([kb.embd_id for kb in kbs]))
@@ -1450,6 +1520,13 @@ async def async_ask(question, kb_ids, tenant_id, chat_llm_name=None, search_conf
         rerank_mdl=rerank_mdl,
         rank_feature=label_question(question, kbs)
     )
+    if include_reference_metadata:
+        logging.debug(
+            "reference_metadata enrichment enabled for async_ask: chunk_count=%d metadata_fields=%s",
+            len(kbinfos.get("chunks", [])),
+            metadata_fields,
+        )
+        _enrich_chunks_with_document_metadata(kbinfos.get("chunks", []), metadata_fields)
 
     knowledges = kb_prompt(kbinfos, max_tokens)
     sys_prompt = PROMPT_JINJA_ENV.from_string(ASK_SUMMARY).render(knowledge="\n".join(knowledges))

api/db/services/doc_metadata_service.py

@@ -772,6 +772,36 @@ class DocMetadataService:
             logging.error(f"Error getting flattened metadata for KBs {kb_ids}: {e}")
             return {}
 
+    @classmethod
+    def get_metadata_keys_by_kbs(cls, kb_ids: List[str]) -> List[str]:
+        """
+        Get unique metadata field names across multiple knowledge bases.
+
+        Args:
+            kb_ids: List of knowledge base IDs
+
+        Returns:
+            Sorted list of unique metadata field names
+        """
+        if not kb_ids:
+            return []
+
+        logging.debug(f"get_metadata_keys_by_kbs start: n_kbs={len(kb_ids)}")
+        keys: set[str] = set()
+        try:
+            for kb_id in kb_ids:
+                results = cls._search_metadata(kb_id, condition={"kb_id": kb_id})
+                for _doc_id, doc in cls._iter_search_results(results):
+                    doc_meta = cls._extract_metadata(doc)
+                    if not isinstance(doc_meta, dict):
+                        continue
+                    keys.update(str(k) for k in doc_meta.keys())
+            logging.debug(f"get_metadata_keys_by_kbs end: n_keys={len(keys)}, kb_ids={kb_ids}")
+            return sorted(keys)
+        except Exception as e:
+            logging.error(f"Error getting metadata keys for KBs {kb_ids}: {e}")
+            return []
+
     @classmethod
     def get_metadata_for_documents(cls, doc_ids: Optional[List[str]], kb_id: str) -> Dict[str, Dict]:
         """

api/utils/reference_metadata_utils.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
+
+logger = logging.getLogger(__name__)
+
+
+def resolve_reference_metadata_preferences(
+    request_payload: dict | None = None,
+    config_payload: dict | None = None,
+) -> tuple[bool, set[str] | None]:
+    """
+    Resolve metadata include/fields from request and optional config.
+    Request values take precedence over config values.
+    Supports legacy request keys: include_metadata / metadata_fields.
+    """
+    request_payload = request_payload or {}
+    config_payload = config_payload or {}
+
+    config_ref = config_payload.get("reference_metadata", {})
+    request_ref = request_payload.get("reference_metadata", {})
+
+    resolved: dict = {}
+    if isinstance(config_ref, dict):
+        resolved.update(config_ref)
+    if isinstance(request_ref, dict):
+        resolved.update(request_ref)
+
+    if "include_metadata" in request_payload:
+        resolved["include"] = bool(request_payload.get("include_metadata"))
+    if "metadata_fields" in request_payload:
+        resolved["fields"] = request_payload.get("metadata_fields")
+
+    include_metadata = bool(resolved.get("include", False))
+    fields = resolved.get("fields")
+    if fields is None:
+        return include_metadata, None
+    if not isinstance(fields, list):
+        logger.warning(
+            "reference_metadata.fields is not a list; include_metadata=%s fields=%r type=%s resolved=%r. "
+            "enrich_chunks_with_document_metadata will skip enrichment.",
+            include_metadata,
+            fields,
+            type(fields).__name__,
+            resolved,
+        )
+        return include_metadata, set()
+    return include_metadata, {f for f in fields if isinstance(f, str)}
+
+
+def enrich_chunks_with_document_metadata(
+    chunks: list[dict],
+    metadata_fields: set[str] | None = None,
+    *,
+    kb_field: str = "kb_id",
+    doc_field: str = "doc_id",
+    output_field: str = "document_metadata",
+) -> None:
+    """
+    Mutates chunk payloads in-place by attaching `document_metadata`.
+    Field names can be customized for different chunk schemas.
+    """
+    if metadata_fields is not None and not metadata_fields:
+        return
+
+    doc_ids_by_kb: dict[str, set[str]] = {}
+    for chunk in chunks:
+        kb_ids = chunk.get(kb_field)
+        doc_id = chunk.get(doc_field)
+        if not kb_ids or not doc_id:
+            continue
+        if isinstance(kb_ids, (list, tuple)):
+            for kid in kb_ids:
+                if kid:
+                    doc_ids_by_kb.setdefault(kid, set()).add(doc_id)
+        else:
+            doc_ids_by_kb.setdefault(kb_ids, set()).add(doc_id)
+
+    if not doc_ids_by_kb:
+        return
+
+    # Resolve service lazily so callers/tests that swap service modules at runtime
+    # (e.g. via monkeypatch) don't get stuck with a stale class reference.
+    from api.db.services.doc_metadata_service import DocMetadataService
+    metadata_getter = getattr(DocMetadataService, "get_metadata_for_documents", None)
+    if not callable(metadata_getter):
+        logging.warning(
+            "DocMetadataService.get_metadata_for_documents is unavailable; "
+            "skipping metadata enrichment."
+        )
+        return
+
+    meta_by_doc: dict[str, dict] = {}
+    for kb_id, doc_ids in doc_ids_by_kb.items():
+        meta_map = metadata_getter(list(doc_ids), kb_id)
+        if meta_map:
+            meta_by_doc.update(meta_map)
+            logging.debug("Fetched metadata for %d docs in kb_id=%s", len(meta_map), kb_id)
+
+    for chunk in chunks:
+        doc_id = chunk.get(doc_field)
+        if not doc_id:
+            continue
+        meta = meta_by_doc.get(doc_id)
+        if not meta:
+            continue
+        if metadata_fields is not None:
+            meta = {k: v for k, v in meta.items() if k in metadata_fields}
+        if meta:
+            chunk[output_field] = meta
+            logging.debug("Enriched chunk for doc_id=%s with %d metadata fields: %s", doc_id, len(meta), list(meta.keys()))