特定领域 RAG中细调嵌入模型能否提升效果？

为特定领域 RAG 细调嵌入模型的作用，探讨这种微调在构建 RAG 流水线时，尤其是在检索步骤以及最终的生成阶段，会带来怎样的不同。不多废话，咱们这就开始吧！

引言：

我们旨在高效地将特定领域的数据摄入到向量存储中，通过利用两个不同的模型来实现：一个微调的嵌入模型和一个预训练的嵌入模型。这两个模型协同工作，处理输入数据，确保有效地捕获并存储相关上下文信息以便检索。这个设置的主要目标是提高语言模型生成响应时的准确性和上下文理解能力，当用户与系统交互时。

为了实现这一点，每个用户查询都会通过微调模型和预训练模型进行处理，这两个模型从查询中提取有意义的表示，并检索出最相关的上下文信息。然后将检索到的上下文传递给一个大型语言模型（LLM），它主要有两个功能。首先，它根据检索到的知识生成响应，确保答案与特定领域的语料库保持一致。其次，LLM 还充当评估器，使用两个关键指标来评估生成响应的质量：“答案相关性得分”和“上下文相关性得分”。这些评估提供了检索到的上下文与查询对齐程度的见解，以及生成的响应是否相关且准确。

然后将评估分数汇总并显示在一个结构化的仪表板上，以便持续监控和优化系统的性能。通过以可视化格式呈现这些分数，决策者可以分析趋势，识别潜在的改进领域，并相应地微调检索和响应生成机制。这种架构不仅提高了响应的质量，还提供了一种评估和优化呈现给用户的信息相关性的结构化方法。

通过这种迭代方法，系统确保特定领域的查询以更高的精度处理，使其特别适用于需要深度上下文理解的应用，如企业搜索解决方案、技术文档检索和知识管理系统。微调嵌入、预训练模型和基于 LLM 的评估的结合创建了一个强大的流水线，增强了检索和响应生成过程，最终导致一个更智能、更有上下文感知能力的问答系统。

实现：

不多耽误时间，咱们马上开始代码实现。下面是项目结构。

.
├── biomedical_rag.py
├── data
│   └── pubmed_knowledge.pdf
├── logger_util.py
├── logs
│   └── app.log
├── pre_processing.py
└── requirements.txt

让我们先从日志工具开始，它会将代码中流动的所有事件记录到日志文件以及控制台中。

import logging
import os
import datetime
import sysclass LoggerUtil:"""一个简单的 Python 应用程序日志工具。"""    def __init__(self, name="app", log_level=logging.INFO, log_file=None, console_output=True):"""初始化日志工具。        参数：name (str): 日志器的名称log_level (int): 日志级别（例如，logging.DEBUG, logging.INFO）log_file (str, 可选）：日志文件的路径。如果为 None，则不进行文件日志记录。console_output (bool): 是否将日志输出到控制台"""self.logger = logging.getLogger(name)self.logger.setLevel(log_level)self.logger.handlers = []  # 清除任何现有的处理器        # 创建格式化器formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s',datefmt='%Y-%m-%d %H:%M:%S')        # 如果请求，则添加控制台处理器if console_output:console_handler = logging.StreamHandler(sys.stdout)console_handler.setFormatter(formatter)self.logger.addHandler(console_handler)        # 如果提供了 log_file，则添加文件处理器if log_file:os.makedirs(os.path.dirname(log_file), exist_ok=True)file_handler = logging.FileHandler(log_file)file_handler.setFormatter(formatter)self.logger.addHandler(file_handler)    def debug(self, message):"""记录一条调试消息。"""self.logger.debug(message)    def info(self, message):"""记录一条信息消息。"""self.logger.info(message)    def warning(self, message):"""记录一条警告消息。"""self.logger.warning(message)    def error(self, message):"""记录一条错误消息。"""self.logger.error(message)    def critical(self, message):"""记录一条严重消息。"""self.logger.critical(message)

上面的 LoggerUtil 类是一个简单又灵活的 Python 应用程序日志工具。它配置了一个具有给定名称、日志级别和可选文件日志记录的记录器，同时确保消息带有时间戳。它支持将日志输出到控制台和指定的日志文件。该类提供了不同严重性级别的日志消息方法，例如调试、信息、警告、错误和严重。这个工具简化了日志设置，并确保了调试和监控时能高效跟踪消息。

注意：为了创建 pubmed 的知识 pdf，我们将使用来自 huggingface 的“qiaojin/PubMedQA”数据集。

from datasets import load_dataset
from reportlab.lib.pagesizes import letter
from reportlab.pdfgen import canvas
from logger_util import LoggerUtil
import logging# 创建一个记录器，日志同时输出到控制台和文件
log = LoggerUtil(name="pre_processing",log_level=logging.DEBUG,log_file="logs/app.log"
)log.info("正在下载并创建数据框架")
ds = load_dataset("qiaojin/PubMedQA", "pqa_artificial")
df = ds["train"].to_pandas()def extract_contexts(num_of_records: int = 5000):log.info("开始提取上下文")# 限制为前 num_records 条记录df_subset = df.head(num_of_records)    extracted_contexts = []    for i, row in df_subset.iterrows():try:# 根据截图，“context” 似乎已经是一个字典# 其中包含一个键为 “contexts” 的列表，里面是字符串contexts = row['context']['contexts'].tolist()if isinstance(contexts, list):context_text = ' '.join(contexts)extracted_contexts.append(context_text)else:log.info(f"警告：第 {i} 行的 'contexts' 字段不是列表")except (KeyError, TypeError) as e:log.error(f"处理第 {i} 行时出错：{e}")    return extracted_contextsdef _wrap_text(text, canvas, max_width):log.info(f"开始为 PDF 设置文本样式")"""将文本拆分为适合 max_width 的行。"""words = text.split()lines = []current_line = []    for word in words:test_line = ' '.join(current_line + [word])width = canvas.stringWidth(test_line)        if width <= max_width:current_line.append(word)else:# 如果当前行有内容，则添加if current_line:lines.append(' '.join(current_line))current_line = [word]# 如果单个单词太长，则强制它独占一行else:lines.append(word)    # 别忘了最后一行if current_line:lines.append(' '.join(current_line))    return linesdef create_pdf_doc(data, output_filename="data/pubmed_knowledge.pdf"):log.info(f"开始创建 PDF")# 创建一个 letter 大小的画布c = canvas.Canvas(output_filename, pagesize=letter)width, height = letter    # 设置文本区域尺寸margin = 50text_width = width - 2 * margin    # 起始位置和行高x = marginy = height - marginline_height = 14    for string in data:# 创建一个文本对象用于换行文本textobject = c.beginText()textobject.setTextOrigin(x, y)        # 设置换行宽度textobject.setWordSpace(0.1)        # 添加换行后的文本for line in _wrap_text(string, c, text_width):textobject.textLine(line)# 减少可用高度y -= line_height            # 检查是否需要新页面if y < margin:c.drawText(textobject)c.showPage()y = height - margintextobject = c.beginText()textobject.setTextOrigin(x, y)        # 绘制文本对象c.drawText(textobject)y -= line_height  # 段落之间额外的间距    # 保存 PDFc.save()    log.info(f"PDF 创建成功：{output_filename}")# print("开始提取")
# data = extract_contexts()
# print("开始创建 PDF")
# create_pdf_doc(data=data)

这段 “create_pdf_doc”、“_wrap_text” 和 “extract_contexts” 脚本处理并从 “qiaojin/PubMedQA” 数据集提取文本数据，并生成格式化的 PDF 文档。它使用 LoggerUtil 工具记录消息，便于跟踪执行进度和潜在问题。数据集被加载到一个 Pandas DataFrame 中，函数 “extract_contexts” 从前 5000 条记录中检索并连接 “contexts” 字段，同时优雅地处理错误。

为了生成 PDF，“_wrap_text” 确保文本行适合页面宽度，通过将它们拆分为适当大小的段。然后 “create_pdf_doc” 函数将提取的文本组织成可读格式，并使用 ReportLab 将其写入 PDF 文件。它保持边距、处理分页，并确保结构化的文本流动。在整个过程中，日志消息提供了对每个步骤的洞察，使调试和监控变得更加容易。

import osfrom llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings, StorageContext
from llama_index.core.query_engine import FLAREInstructQueryEngine
from llama_index.llms.anthropic import Anthropic
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.vector_stores.qdrant import QdrantVectorStore
from qdrant_client import QdrantClient, AsyncQdrantClient
from qdrant_client.http.models import VectorParams, Distance
from dotenv import load_dotenv, find_dotenv
from deepeval.integrations.llama_index import (DeepEvalAnswerRelevancyEvaluator,DeepEvalContextualRelevancyEvaluator
)from pre_processing import df
import nest_asyncio
from logger_util import LoggerUtil
import logging# 创建一个记录器，日志同时输出到控制台和文件
log = LoggerUtil(name="biomedical_rag",log_level=logging.DEBUG,log_file="logs/app.log"
)# 初始化 nest_asyncio
nest_asyncio.apply()load_dotenv(find_dotenv())
log.info("开始设置全局参数")COLLECTION_NAME = os.environ.get("QDRANT_COLLECTION_NAME")Settings.chunk_size = int(os.environ.get("CHUNK_SIZE"))
Settings.chunk_overlap = int(os.environ.get("CHUNK_OVERLAP"))Settings.embed_model = HuggingFaceEmbedding(model_name=os.environ.get("EMBEDDING_MODEL_ID"), device="mps",trust_remote_code=True)
Settings.llm = Anthropic(model=os.environ.get("ANTHROPIC_MODEL_ID"), api_key=os.environ.get("ANTHROPIC_API_KEY"))q_client = QdrantClient(url=os.environ.get("QDRANT_URL"), api_key=os.environ.get("QDRANT_API_KEY"))
async_q_client = AsyncQdrantClient(url=os.environ.get("QDRANT_URL"), api_key=os.environ.get("QDRANT_API_KEY"))vector_store = QdrantVectorStore(collection_name=COLLECTION_NAME, client=q_client, aclient=async_q_client,dense_config=VectorParams(size=int(os.environ.get("EMBEDDING_DIM")), distance=Distance.COSINE))storage_ctx = StorageContext.from_defaults(vector_store=vector_store)
vector_store_index: VectorStoreIndex
docstore = Noneif q_client.collection_exists(collection_name=COLLECTION_NAME):log.info("从现有集合创建向量索引")vector_store_index = VectorStoreIndex.from_vector_store(vector_store=vector_store)
else:log.info("集合不存在，因此创建并索引数据。")docs = SimpleDirectoryReader(input_dir="data", required_exts=[".pdf"]).load_data(show_progress=True)    vector_store_index = VectorStoreIndex.from_documents(documents=docs,# 我们的密集嵌入模型embed_model=Settings.embed_model,storage_context=storage_ctx,)evals_questions = []
def compute_question_answer_evals():    for i, row in df[:10].iterrows():ctxs = []question = row["question"]# answer = row["long_answer"]# evals_questions.append((question, answer))        log.info(f"第 {i} 个问题用于评估：{question}")query_engine = vector_store_index.as_query_engine(similarity_top_k=50, response_mode='tree_summarize')        # flare_query_engine = FLAREInstructQueryEngine(#     query_engine=query_engine,#     max_iterations=5,#     verbose=True# )response = query_engine.query(str_or_query_bundle=question)        for node in response.source_nodes:ctxs.append(node.text)        ans_rel_evaluator = DeepEvalAnswerRelevancyEvaluator(threshold=0.7, model=os.environ.get("OPENAI_MODEL_ID"))ctx_rel_evaluator = DeepEvalContextualRelevancyEvaluator(threshold=0.7, model=os.environ.get("OPENAI_MODEL_ID"))ans_rel_result = ans_rel_evaluator.evaluate(response=response.response, query=question, contexts=ctxs)ctx_rel_result = ctx_rel_evaluator.evaluate(response=response.response, query=question, contexts=ctxs)        log.info(f"响应：{response}")log.info(f"答案相关性：{ans_rel_result}")log.info(f"上下文相关性：{ctx_rel_result}")log.info("="*200)compute_question_answer_evals()

这个脚本实现了一个使用 llama_index、Qdrant、Anthropic 和 gpt-4o 的生物医学检索增强型生成（RAG）系统，用于问答和评估。它从配置日志开始，使用 LoggerUtil 跟踪执行过程。加载环境变量以设置参数，例如分块大小、嵌入模型和 Qdrant 集合名称。脚本初始化 nest_asyncio 以允许嵌套事件循环，从而启用操作的异步执行。

系统的内核涉及使用 Qdrant 创建或加载向量存储。如果指定了集合，则从它构建 VectorStoreIndex；否则，脚本从 “data” 目录读取 PDF 文档，使用 Hugging Face 模型进行嵌入，并将它们索引到 Qdrant 中。这个向量索引允许高效检索问答所需的相关信息。

函数 compute_question_answer_evals 处理数据集的前十个记录，提取问题并使用基于相似性的检索方法查询向量存储索引。检索到的上下文随后使用 DeepEvalAnswerRelevancyEvaluator 和 DeepEvalContextualRelevancyEvaluator 进行评估，以评估答案和检索到的上下文与给定问题的相关性。结果，包括检索到的响应、答案相关性得分和上下文相关性得分，被记录下来以便进一步分析。这个设置提供了一个自动化流水线，用于评估生物医学知识检索和响应生成的有效性。

指标可视化：

最后，将前面过程的日志发送给 LLM，以获取所需的可视化数据以及数据的关键亮点。将数据发送给 LLM 并根据我们的需求获取数据以及关键亮点的过程将留在第三部分介绍。现在，让我们看看指标是如何可视化的，以及仪表板会是什么样子。

仪表板显示每个选项卡下的 5 个实验。

实验和指标研究：

实验在三个方面有所不同：

1. 分块大小：使用了两个值——128 和 512。
2. 分块重叠：使用了两个值——50 和 100。
3. 嵌入模型：测试了三种不同的嵌入模型：

• pavanamantha/distilroberta-pubmed-embeddings
• sentence-transformers/all-MiniLM-L6-v2
• pavanamantha/bge-base-en-biomed

实验的关键发现

分块大小的影响（128 对比 512）

• 将分块大小从 128 增加到 512（实验 1 → 2 和 3 → 4）导致：
• 上下文相关性降低（实验 1 和 2 中从 0.20 降至 0.08，实验 3 和 4 中从 0.19 降至 0.08）。
• 实验 2 中 相关性更高（0.61），但实验 4 中出现了 负相关（-0.74），表明分块的影响因嵌入模型而异。
• 实验 2 中答案相关性略有提升，但实验 4 中有所下降。

分块重叠的影响（50 对比 100）

• 较高的重叠度（实验 2 和 4）并不一定 能改善检索效果。
• 实验 4 中的 负相关（-0.74） 表明，在某些情况下，更多的重叠可能会损害检索效果。

嵌入模型的影响

DistilRoberta-PubMed（实验 1 和 2）

• 产生了 较高的答案相关性（约 0.83–0.85），但 上下文相关性较低。
• 当分块大小增加时，相关性更高（0.61）。

MiniLM-L6-v2（实验 3 和 4）

• 显示出 较高的答案相关性（约 0.78–0.81），但 在较大的分块大小下失败（实验 4，相关性为 -0.74）。

BGE-Base-En-Biomed（实验 5）

• 在两者之间取得了平衡，保持了 较好的答案相关性（0.84）。
• 相关性（0.33）表明与其它模型相比，上下文稍微更有用一些。

结论

我们的实验表明，微调嵌入（DistilRoberta-PubMed 和 BGE-Base-En-Biomed）始终能够产生较高的答案相关性（约 0.83–0.85），尽管上下文相关性较低。在它们之间，DistilRoberta-PubMed 在较大的分块（实验 2）中实现了最高的相关性（0.61），表明增加分块大小有时可以提高检索效果。

相比之下，预训练模型（MiniLM-L6-v2）在分块方面表现不佳，实验 4 中的 负相关（-0.74） 表明，增加更多的上下文实际上降低了答案质量。

总体而言，在生物医学 RAG 中，微调嵌入优于预训练模型，提供了 对检索质量不佳的更好鲁棒性。然而，分块策略起着关键作用——较大的分块可能对某些嵌入有帮助，但对其他嵌入可能有害。关键要点？选择正确的嵌入模型和分块策略对于优化检索增强型生成性能至关重要。