追踪哪些文档片段被用于检索增强生成

在 RAG 系统中，追踪哪些文档片段被用于生成答案是提升系统可解释性、调试能力和可靠性的关键步骤。通过明确记录检索到的文本块（即 source_nodes），你可以：

• 验证答案来源：确保生成的内容确实基于检索到的资料，而非模型幻觉。
• 调试检索效果：分析是检索模块没找到正确信息，还是生成模块用错了信息。
• 增强用户信任：向用户展示答案依据的具体文档片段（例如引用出处）。
• 评估与迭代：积累 bad cases 的数据，指导后续优化。

在 LlamaIndex 中，这一需求可以非常方便地实现。下面我会介绍几种常用的追踪方法，并提供代码示例。

---

一、通过 query_engine 直接获取源节点

最简单的方式是在查询后，从返回的 Response 对象中读取 source_nodes 属性。

query_engine = index.as_query_engine()
response = query_engine.query("麻黄汤的组成是什么？")

# 打印答案
print("答案:", response)

# 追踪检索到的文档片段
print("\n检索到的片段：")
for i, node in enumerate(response.source_nodes):
    print(f"\n--- 片段 {i+1} (相似度: {node.score:.4f}) ---")
    print(node.text)
    # 如果有元数据，也可打印
    print("元数据:", node.metadata)

• response.source_nodes 是一个列表，按相关性降序排列。
• 每个元素包含 node.text（片段内容）、score（相似度分数）、metadata（元数据）等信息。

二、通过自定义回调记录检索日志

如果你希望自动记录所有查询的检索结果（例如存入日志文件或数据库），可以使用 LlamaIndex 的 回调系统。

1. 定义回调处理器

from llama_index.core.callbacks import BaseCallbackHandler, CBEventType
from typing import Any, Dict, List

class RetrievalTracker(BaseCallbackHandler):
    def __init__(self):
        super().__init__()
        self.retrieved_nodes = []

    def on_event_start(
        self,
        event_type: CBEventType,
        payload: Dict[str, Any] = None,
        event_id: str = "",
        **kwargs,
    ) -> str:
        if event_type == CBEventType.RETRIEVE:
            # 检索开始时清空上一次的记录
            self.retrieved_nodes = []
        return event_id

    def on_event_end(
        self,
        event_type: CBEventType,
        payload: Dict[str, Any] = None,
        event_id: str = "",
        **kwargs,
    ):
        if event_type == CBEventType.RETRIEVE and payload:
            # 检索结束时记录结果
            nodes = payload.get("nodes", [])
            self.retrieved_nodes.extend(nodes)

2. 将回调应用到查询引擎

from llama_index.core.callbacks import CallbackManager

tracker = RetrievalTracker()
callback_manager = CallbackManager([tracker])

# 创建索引时传入 callback_manager
index = VectorStoreIndex.from_documents(
    documents,
    callback_manager=callback_manager
)
# 或为查询引擎单独设置
query_engine = index.as_query_engine(callback_manager=callback_manager)

response = query_engine.query("麻黄汤的组成")

# 从 tracker 中获取检索结果
print("追踪到的片段数量:", len(tracker.retrieved_nodes))
for node in tracker.retrieved_nodes:
    print(node.text)

你也可以将 tracker.retrieved_nodes 写入文件或数据库，实现持久化追踪。

三、在自定义检索器中直接暴露源节点

如果你需要更精细的控制，可以单独使用检索器并手动记录。

retriever = index.as_retriever(similarity_top_k=5)
nodes_with_scores = retriever.retrieve("麻黄汤的组成")

# 记录检索结果
for node in nodes_with_scores:
    print(f"分数: {node.score:.4f}")
    print(node.text)
    print(node.metadata)
    print("---")

# 然后手动构建提示词并调用 LLM
from llama_index.core.response_synthesizers import CompactAndRefine
synth = CompactAndRefine()
response = synth.synthesize("麻黄汤的组成", nodes_with_scores)

这种方式将检索和生成解耦，方便记录检索中间结果。

---

四、进阶：记录完整的生成上下文

有时你不仅想记录检索到的片段，还想记录最终输入给 LLM 的完整提示词（包含所有片段和问题）。可以通过自定义响应合成器或回调来实现。

示例：使用回调捕获提示词

from llama_index.core.callbacks import BaseCallbackHandler, CBEventType

class PromptLogger(BaseCallbackHandler):
    def on_event_end(
        self,
        event_type: CBEventType,
        payload: Dict[str, Any] = None,
        event_id: str = "",
        **kwargs,
    ):
        if event_type == CBEventType.TEMPLATING and payload:
            # TEMPLATING 事件发生在生成提示词时
            template = payload.get("template")
            context = payload.get("context")
            query = payload.get("query")
            print("生成的提示词模板:", template)
            print("上下文片段:", context)
            print("原始查询:", query)
            # 这里可以将数据写入日志

callback_manager = CallbackManager([PromptLogger()])
query_engine = index.as_query_engine(callback_manager=callback_manager)
response = query_engine.query("麻黄汤的组成")

注意：具体 payload 内容可能随版本略有变化，建议查看对应版本的文档或源码。

---

五、存储追踪结果到数据库

在生产环境中，你可能希望将每次查询的检索记录存入数据库，便于后续分析和评估。

简单示例（SQLite）

import sqlite3
import json
from datetime import datetime

# 初始化数据库
conn = sqlite3.connect("rag_logs.db")
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS query_logs
             (id INTEGER PRIMARY KEY AUTOINCREMENT,
              query TEXT,
              answer TEXT,
              source_nodes TEXT,
              timestamp TEXT)''')
conn.commit()

# 查询并记录
query = "麻黄汤的组成"
response = query_engine.query(query)

source_nodes_json = json.dumps([
    {
        "text": node.text,
        "score": node.score,
        "metadata": node.metadata
    }
    for node in response.source_nodes
], ensure_ascii=False)

c.execute("INSERT INTO query_logs (query, answer, source_nodes, timestamp) VALUES (?, ?, ?, ?)",
          (query, str(response), source_nodes_json, datetime.now().isoformat()))
conn.commit()

---

六、中医场景的应用示例

假设你有一个包含《伤寒论》《金匮要略》等古籍的知识库，追踪检索片段后，可以在回答中自动标注出处。

response = query_engine.query("桂枝汤的禁忌有哪些？")
print("答案:", response)
print("\n依据来源：")
for node in response.source_nodes:
    source = node.metadata.get("source", "未知")
    chapter = node.metadata.get("chapter", "")
    print(f"- 《{source}》{chapter} (相似度: {node.score:.3f})")
    print(f"  {node.text[:100]}...")

这样既提升了可信度，也便于用户核实。

---

总结

• LlamaIndex 原生支持：Response 对象包含 source_nodes，可直接访问。
• 回调机制：适合自动记录和监控。
• 自定义检索器：提供完全控制权。
• 持久化存储：便于评估和审计。

通过追踪文档片段，你可以将 RAG 系统从一个黑盒变成可解释、可优化的透明系统，在中医等专业领域尤为重要。