1. 项目概述
在构建基于检索增强生成(RAG)的系统时,追踪哪些文档片段被用于生成最终答案是一个关键需求。这不仅关系到系统的可解释性,也直接影响着调试效率和用户信任度。想象一下,当系统给出一个医学建议时,如果能同时展示这个建议出自哪本权威医学典籍的哪个章节,用户的信任度会大幅提升。
RAG系统通常由两个核心组件构成:检索模块和生成模块。检索模块负责从海量文档中找到与问题相关的片段,生成模块则基于这些片段合成最终答案。在这个过程中,记录和追踪被检索到的文档片段(在LlamaIndex中称为source_nodes)就变得尤为重要。
2. 核心需求解析
2.1 为什么需要追踪文档片段
追踪文档片段主要解决以下几个核心问题:
-
验证答案可靠性:可以检查生成的内容是否确实基于检索到的资料,而不是大模型的"幻觉"产物。这在医疗、法律等专业领域尤为重要。
-
调试系统性能:当答案不准确时,可以快速定位问题是出在检索阶段(没找到正确信息)还是生成阶段(用错了找到的信息)。
-
增强用户信任:向用户展示答案依据的具体文档片段,就像学术论文提供参考文献一样,能显著提升可信度。
-
系统优化依据:积累真实查询中的检索记录,可以分析哪些查询效果不好,为后续优化提供数据支持。
2.2 典型应用场景
以中医知识库为例,假设我们构建了一个包含《伤寒论》《金匮要略》等经典古籍的RAG系统:
-
当用户询问"麻黄汤的组成是什么?"时,系统不仅能给出答案,还能显示这个答案出自《伤寒论》的哪个章节。
-
如果系统给出了一个看似不合理的中药配伍建议,通过查看检索到的片段,我们可以快速判断是古籍原文确实如此,还是系统理解错误。
3. 实现方案详解
3.1 基础方法:直接获取source_nodes
LlamaIndex的Response对象原生支持获取检索到的文档片段,这是最简单的实现方式:
python复制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)
这种方法适合快速验证和简单应用,但缺乏持久化记录的能力。
3.2 进阶方案:自定义回调系统
对于生产环境,我们需要更系统的追踪机制。LlamaIndex提供了灵活的回调系统,可以监听各种事件并记录关键信息。
3.2.1 实现回调处理器
python复制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 = []
self.queries = []
def on_event_start(
self,
event_type: CBEventType,
payload: Dict[str, Any] = None,
event_id: str = "",
**kwargs,
) -> str:
if event_type == CBEventType.QUERY:
# 记录查询问题
self.queries.append(payload.get("query_str", ""))
elif 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)
3.2.2 集成到查询流程
python复制from llama_index.core.callbacks import CallbackManager
# 初始化追踪器
tracker = RetrievalTracker()
callback_manager = CallbackManager([tracker])
# 创建索引时应用回调
index = VectorStoreIndex.from_documents(
documents,
callback_manager=callback_manager
)
# 执行查询
response = query_engine.query("麻黄汤的组成")
# 分析检索结果
print(f"查询问题: {tracker.queries[-1]}")
print(f"检索到 {len(tracker.retrieved_nodes)} 个相关片段")
for node in tracker.retrieved_nodes:
print(f"- 相似度 {node.score:.3f}: {node.text[:100]}...")
3.3 生产级实现:数据库存储
在实际应用中,我们通常需要将检索记录持久化存储以便后续分析。以下是使用SQLite的实现示例:
python复制import sqlite3
import json
from datetime import datetime
class RetrievalLogger:
def __init__(self, db_path="rag_logs.db"):
self.conn = sqlite3.connect(db_path)
self._init_db()
def _init_db(self):
c = self.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,
retrieval_time REAL)''')
self.conn.commit()
def log_query(self, query, response, retrieval_time=None):
source_nodes_json = json.dumps([
{
"text": node.text,
"score": node.score,
"metadata": node.metadata
}
for node in response.source_nodes
], ensure_ascii=False)
c = self.conn.cursor()
c.execute('''INSERT INTO query_logs
(query, answer, source_nodes, timestamp, retrieval_time)
VALUES (?, ?, ?, ?, ?)''',
(str(query), str(response), source_nodes_json,
datetime.now().isoformat(), retrieval_time))
self.conn.commit()
def close(self):
self.conn.close()
# 使用示例
logger = RetrievalLogger()
query = "麻黄汤的组成"
start_time = time.time()
response = query_engine.query(query)
retrieval_time = time.time() - start_time
logger.log_query(query, response, retrieval_time)
logger.close()
4. 高级应用技巧
4.1 记录完整生成上下文
有时我们需要记录最终输入给LLM的完整提示词,这可以通过扩展回调系统实现:
python复制class GenerationContextTracker(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:
context = payload.get("context_str", "")
query = payload.get("query_str", "")
template = payload.get("template", "")
# 这里可以存储到数据库或日志文件
print(f"生成上下文:\n问题: {query}\n")
print(f"模板:\n{template}\n")
print(f"检索上下文:\n{context}")
4.2 中医知识库的特殊处理
在中医古籍应用中,我们可以增强元数据处理,实现更专业的引用展示:
python复制def format_chinese_medicine_citation(node):
source = node.metadata.get("source", "未知典籍")
chapter = node.metadata.get("chapter", "")
section = node.metadata.get("section", "")
citation_parts = [f"《{source}》"]
if chapter:
citation_parts.append(f"第{chapter}章")
if section:
citation_parts.append(f"第{section}节")
return "".join(citation_parts)
# 在展示检索结果时
for node in response.source_nodes:
citation = format_chinese_medicine_citation(node)
print(f"{citation} (相似度: {node.score:.3f})")
print(f"{node.text[:200]}...\n")
5. 性能优化与注意事项
5.1 性能考量
-
异步记录:在高并发场景下,建议使用异步方式写入数据库,避免阻塞主查询流程。
-
采样记录:对于高频查询系统,可以考虑采样记录而非全量记录,减轻存储压力。
-
数据清理:实现定期清理旧日志的机制,防止数据库无限膨胀。
5.2 常见问题排查
-
缺失source_nodes:
- 检查是否使用了支持返回源节点的查询引擎
- 确认没有在响应合成过程中过滤掉这些信息
-
回调不触发:
- 确保正确设置了callback_manager
- 检查事件类型是否匹配
-
元数据不完整:
- 在文档加载阶段确保提取了足够的元数据
- 考虑使用自定义文档处理器
6. 扩展应用:基于检索记录的反馈循环
收集的检索记录可以用于构建反馈循环,持续优化系统:
python复制def analyze_retrieval_patterns(db_path, top_n=100):
conn = sqlite3.connect(db_path)
c = conn.cursor()
# 分析高频低分查询
c.execute('''SELECT query, source_nodes
FROM query_logs
ORDER BY json_extract(source_nodes, '$[0].score') ASC
LIMIT ?''', (top_n,))
poor_retrievals = c.fetchall()
# 分析高频查询
c.execute('''SELECT query, COUNT(*) as cnt
FROM query_logs
GROUP BY query
ORDER BY cnt DESC
LIMIT ?''', (top_n,))
frequent_queries = c.fetchall()
conn.close()
return {
'poor_retrievals': poor_retrievals,
'frequent_queries': frequent_queries
}
这个分析结果可以用于:
- 优化检索模型的参数
- 补充知识库中的薄弱环节
- 调整文档分块策略
7. 实际应用案例
以一个真实的中医问答系统为例,展示完整实现:
python复制# 初始化系统
documents = load_chinese_medicine_documents() # 加载《伤寒论》等古籍
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(similarity_top_k=3)
# 设置追踪系统
logger = RetrievalLogger("chinese_medicine.db")
tracker = RetrievalTracker()
callback_manager = CallbackManager([tracker])
query_engine.callback_manager = callback_manager
# 查询示例
queries = [
"麻黄汤的组成",
"桂枝汤的禁忌人群",
"小柴胡汤的主治症状"
]
for query in queries:
response = query_engine.query(query)
logger.log_query(query, response)
# 展示带引用的答案
print(f"\n问题: {query}")
print(f"答案: {response}")
print("\n依据:")
for i, node in enumerate(response.source_nodes):
citation = format_chinese_medicine_citation(node)
print(f"{i+1}. {citation} (相似度: {node.score:.3f})")
print(f" {node.text[:150]}...")
logger.close()
这个实现不仅回答了用户问题,还提供了权威出处,极大提升了系统的专业性和可信度。
8. 技术选型建议
对于不同规模的应用,可以考虑以下技术方案:
-
小型项目:
- 直接使用Response.source_nodes
- SQLite存储检索记录
- 简单的回调处理
-
中型项目:
- 完整的回调系统
- PostgreSQL/MongoDB存储
- 异步日志记录
- 基础的分析面板
-
大型项目:
- 分布式追踪系统
- 专门的日志分析流水线
- 自动化反馈循环
- 与监控系统集成
9. 避坑指南
在实际实施过程中,有几个常见的坑需要注意:
-
元数据一致性:确保所有文档在加载时都处理了相同的元数据字段,否则后续展示会混乱。
-
分数可比性:不同检索器返回的相似度分数范围可能不同,比较时要注意归一化。
-
回调性能:复杂的回调处理可能影响查询性能,对于延迟敏感的应用要做性能测试。
-
数据安全:如果存储的查询包含用户隐私信息,要做好脱敏处理。
-
版本兼容:LlamaIndex的回调事件类型和payload结构可能随版本变化,升级时要注意检查。
10. 总结与展望
实现RAG系统的检索片段追踪不仅能提升系统透明度,还为持续优化提供了宝贵数据。通过本文介绍的各种方法,开发者可以根据项目需求选择合适的实现方案。
未来可以考虑的方向包括:
- 自动化分析检索记录,主动发现知识库缺口
- 基于用户反馈调整检索策略
- 可视化展示答案的生成路径
- 结合强化学习优化检索参数
追踪文档片段只是RAG可解释性的第一步,但却是构建可靠、可信AI系统的重要基础。
