使用私有数据和私有基础设施增强 AI 解决方案

使用私有数据和私有基础设施增强 AI 解决方案

多少次你按照建议去优化机器学习模型，结果却发现这些方法根本不适合你的特定场景？答案想必你也清楚——大多数时候，甚至可以说一直都是。归根结底，关键就在于你的数据。你需要不断地测试、碰壁、再测试，直到找到那个最适合你实际情况的方案。这篇文章将深入探讨四种策略，教你如何利用私有数据和私有基础设施，来优化检索增强生成（RAG），从而提升AI解决方案的效果。

“我没有失败，我只是找到了10,000种行不通的方法。” ——

托马斯·爱迪生

在之前的文章中，我们聊过如何通过检索增强生成（RAG）策略，将私有知识注入到像LLaMA 3这样的公共模型中，同时不必担心敏感信息外泄。在自己可控的基础设施上，用RAG来处理私有数据，优势确实很明显。但要想落地，还得在几个关键环节上做足功课。

RAG 回顾

先快速回顾一下RAG的基本流程。它主要由两个过程构成：第一个是“

”，负责从不同来源抓取数据、转换成文本、切分成连贯且语义相关的片段，最后存入向量数据库。第二个是“”，从用户输入的问题出发，利用前一个过程的结果找出相关数据片段，然后丰富模型的上下文，最终生成输出。

具体流程如下图所示：

RAG 过程。图片由作者提供

在上一篇文章中，我们用一家虚构公司ColdF的数据，搭建了“

”和“”这两个环节。接下来，本文将重点解释如何评估并优化这些环节的产出。

改进组件

首先，我们需要明确RAG过程的几个关键发力点：

• 分块方法
  ：优化块大小，确保数据片段既有意义又与上下文相关。

• 嵌入模型
  ：选择并微调模型，以改进语义表示的精确度。

• 向量搜索方法
  ：选好相似度度量标准和搜索参数，提高检索效率。

• 馈送到模型的最终提示
  ：精心设计提示词，提升输出质量。

RAG 管道中的 A/B 测试

确定好要优化的组件后，策略就变成了：对每个组件的不同配置进行对比测试，找出性能更优的那个版本。通过运行两个版本，依据预设的指标来衡量它们的表现。那么问题来了：用什么指标来衡量？为此，我们参考了论文《RAGAS：检索增强生成的自动评估》¹，它提出了三个核心指标：

1. 真实性
   ：检查答案中的信息是否与上下文提供的内容匹配。如果答案中的所有陈述都能直接从上下文中找到或推断出来，那么它就是真实的。举个例子，如果上下文是“在我们5月份访问里斯本期间，我和爱丽丝去了阿尔法玛、拜罗阿尔托、贝伦塔以及许多其他地方。”，而答案是“5月，爱丽丝去了阿尔法玛、拜罗阿尔托、贝伦塔以及许多其他地方。”，那么上下文中支持了所有提取出的陈述，真实性得分为100%。但如果答案是“5月，爱丽丝去了阿尔法玛和圣乔治城堡。”，从答案中提取的两条陈述（“爱丽丝去了阿尔法玛”和“爱丽丝去了圣若热城堡”）只有一条得到上下文支持，那真实性得分就是50%。

2. 答案相关性
   ：检查生成的答案是否完整地、直接地回答了提出的问题。信息正确与否并不重要。例如，如果问题是“葡萄牙的首都是哪里？”，答案是“里斯本是葡萄牙的首都”，那这个答案就是相关的，因为它直接回应了问题。如果答案是“里斯本是一个美丽的城市，有许多景点”，那它可能部分相关，但混入了不直接回答问题的额外信息。这个指标能确保答案始终紧扣主题。

3. 上下文相关性
   ：检查上下文提供的信息对回答问题有多大的帮助。它确保只包含必要且相关的细节，剔除那些对回答问题毫无帮助的多余信息。比如，问题是“5月，爱丽丝在里斯本参观了哪些地方？”，上下文是“在我们5月份访问里斯本期间，爱丽丝去了阿尔法玛、拜罗阿尔托、贝伦塔以及许多其他地方。”，那这个上下文高度相关，因为它只提供了爱丽丝5月去过哪些地方的必要信息。但如果上下文是“在我们5月份访问里斯本期间，爱丽丝遇到了许多有趣的人，吃了美味的食物，并去了许多地方。”，它包含了回答问题不需要的额外细节，那就算不相关。这个指标也叫

   上下文精度
   。

论文还解释了如何通过提示大型语言模型（LLM）来全自动地测量这些指标。

在本次评估中，我们使用的库

对这些关键指标做了改进，增加了一个新指标：

1. 上下文召回率
   ：这个指标用与

   上下文相关性
   相同的方式衡量上下文与实际答案的一致性，但用的是实际答案而不是生成的答案。要获得这个指标，需要有一个真实答案作为参照。为了评估这些策略的有效性，我们准备了一组10个问题，并基于ColdF的数据给出了实际答案。

和属于，分别用来衡量幻觉情况以及答案与问题的贴合程度。

和属于，分别用来衡量从向量数据库中检索到正确数据块的能力，以及获取所有必要信息的能力。

简单来说：要评估前面提到的四个指标，我们需要准备问题、生成的答案、上下文以及实际答案。

我们将使用

来实现RAG流程。运行代码需要安装Python（版本3.11.9）和以下库：

• ollama==0.2.1

• chromadb==0.5.0

• transformers==4.41.2

• torch==2.3.1

• langchain==0.2.0

• ragas==0.1.9

以下是使用LangChain的代码片段：

# Import necessary libraries and modules
from langchain.embeddings.base import Embeddings
from transformers import BertModel, BertTokenizer, DPRQuestionEncoder, DPRQuestionEncoderTokenizer, RobertaModel, RobertaTokenizer
from langchain.prompts import ChatPromptTemplate
from langchain_text_splitters import MarkdownHeaderTextSplitter
import requests
from langchain_chroma import Chroma
from langchain import hub
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_models import ChatOllama
from operator import itemgetter

# Define a custom embedding class using the DPRQuestionEncoder
class DPRQuestionEncoderEmbeddings(Embeddings):
show_progress: bool = False
"""Whether to show a tqdm progress bar. Must ha ve `tqdm` installed."""

def __init__(self, model_name: str = 'facebook/dpr-question-encoder-single-nq-base'):
# Initialize the tokenizer and model with the specified model name
self.tokenizer = DPRQuestionEncoderTokenizer.from_pretrained(model_name)
self.model = DPRQuestionEncoder.from_pretrained(model_name)

def embed(self, texts):
# Ensure texts is a list
if isinstance(texts, str):
texts = [texts]

embeddings = []
if self.show_progress:
try:
from tqdm import tqdm
iter_ = tqdm(texts, desc="Embeddings")
except ImportError:
logger.warning(
"Unable to show progress bar because tqdm could not be imported. "
"Please install with `pip install tqdm`."
)
iter_ = texts
else:
iter_ = texts

for text in iter_:
# Tokenize the input text
inputs = self.tokenizer(text, return_tensors='pt')
# Generate embeddings using the model
outputs = self.model(**inputs)
# Extract the embedding and convert it to a list
embedding = outputs.pooler_output.detach().numpy()[0]
embeddings.append(embedding.tolist())

return embeddings

def embed_documents(self, documents):
return self.embed(documents)

def embed_query(self, query):
return self.embed([query])[0]

# Define a template for generating prompts
template = """
### CONTEXT
{context}

### QUESTION
Question: {question}

### INSTRUCTIONS
使用上方 CONTEXT markdown 文本回答用户问题。
提供简短的答案。
仅根据 CONTEXT 中的事实回答问题。
如果 CONTEXT 不包含回答问题的必要信息，则返回“NONE”。
"""

# 使用模板创建一个 ChatPromptTemplate 实例
prompt = ChatPromptTemplate.from_template(template)

# 从 URL 获取文本数据
url = "https://raw.githubusercontent.com/cgrodrigues/rag-intro/main/coldf_secret_experiments.txt"
response = requests.get(url)
if response.status_code == 200:
text = response.text
else:
raise Exception(f"Failed to fetch the file: {response.status_code}")

# 定义用于分割 markdown 文本的标题
headers_to_split_on = [
("#", "Header 1")
]

# 使用指定的标题创建一个 MarkdownHeaderTextSplitter 实例
markdown_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on, strip_headers=False
)

# 使用 markdown 分割器分割文本
docs_splits = markdown_splitter.split_text(text)

# 初始化一个聊天模型
llm = ChatOllama(model="llama3")

# 使用自定义嵌入从文档中创建一个 Chroma 向量存储
vectorstore = Chroma.from_documents(documents=docs_splits, embedding=DPRQuestionEncoderEmbeddings())

# 从向量存储中创建一个检索器
retriever = vectorstore.as_retriever()

# 定义一个格式化文档以供显示的函数
def format_docs(docs):
return "nn".join(doc.page_content for doc in docs)

# 创建一个检索增强生成 (RAG) 链
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| RunnablePassthrough.assign(context=itemgetter("context"))
| {"answer": prompt | llm | StrOutputParser(), 
 "context": itemgetter("context")}
)

# 使用问题调用 RAG 链
result = rag_chain.invoke("Who led the Experiment 1?")
print(result)

代码末尾定义了一个RAG链，可以用以下代码来评估各项指标：

# 导入必要的库和模块
import pandas as pd
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
context_precision,
faithfulness,
answer_relevancy,
context_recall
)
from langchain_community.chat_models import ChatOllama

def get_questions_answers_contexts(rag_chain):
""" 读取问题和答案列表并返回一个 
用于评估的 ragas 数据集 """
# 文件的 URL
url = 'https://raw.githubusercontent.com/cgrodrigues/rag-intro/main/coldf_question_and_answer.psv'

# 从 URL 获取文件
response = requests.get(url)
data = response.text
 
# 将数据按行分割
lines = data.split('n')

# 通过管道符号分割每一行并创建元组
rag_dataset = []

for line in lines[1:10]: # 仅限前 10 个问题
if line.strip():# 确保该行不为空
question, reference_answer = line.split('|')
result = rag_chain.invoke(question)
generated_answer = result['answer']
contexts = result['context']

rag_dataset.append({
"question": question,
"answer": generated_answer, 
"contexts": [contexts], 
"ground_truth": reference_answer
})


rag_df = pd.DataFrame(rag_dataset)
rag_eval_datset = Dataset.from_pandas(rag_df)

# 返回 lragas 数据集
return rag_eval_datset

def get_metrics(rag_dataset):
""" 对于 RAG 数据集，计算指标的真实性、
答案相关性、上下文精度和上下文召回率 """
# 我们要评估的指标列表
metrics = [
faithfulness,
answer_relevancy,
context_precision,
context_recall
]

# 我们将使用带有 LLaMA 3 模型的本地 ollama
langchain_llm =ChatOllama(model="llama3")
langchain_embeddings = DPRQuestionEncoderEmbeddings('facebook/dpr-question_encoder-single-nq-base')

# 返回指标
results = evaluate(rag_dataset, metrics=metrics, llm=langchain_llm, embeddings=langchain_embeddings)
return results

# 获取 RAG 数据集 
rag_dataset = get_questions_answers_contexts(rag_chain)

# 计算指标
results = get_metrics(rag_dataset)
print(results)

这段代码的输出示例如下：

{
'faithfulness': 0.8611, 
'answer_relevancy': 0.8653, 
'context_precision': 0.7778, 
'context_recall': 0.8889
}

如前所述，前两个指标（真实性和答案相关性）与生成过程挂钩。想要改进它们，就得调整语言模型或修改提供给模型的提示词。而后两个指标（上下文精度和上下文召回率）则与检索过程相关，改进它们需要研究文档的存储、索引和选择方式。

分块方法

分块方法的核心是确保数据被切割成最优的片段，方便后续检索。一般思路是尝试不同的块大小，在块太小（丢失上下文）和块太大（压垮检索系统）之间找到一个平衡点。在基线方案中，我们是按每个实验进行分块的；这样一来，实验的某些部分可能会被稀释，最终没能体现在嵌入向量中。解决这个问题的一个思路是使用父文档检索器。这个方法不仅能检索相关的文档片段或段落，还能顺带拉回它的父文档，从而保留相关片段周围的上下文语境。下面是测试这个方法的代码：

# Import necessary libraries and modules
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain.text_splitter import RecursiveCharacterTextSplitter


# Create the parent document retriever
parent_document_retriever = ParentDocumentRetriever(
vectorstore = Chroma(collection_name="parents", 
 embedding_function=DPRQuestionEncoderEmbeddings('facebook/dpr-question_encoder-single-nq-base')),
docstore = InMemoryStore(),
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200),
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1500),
)

parent_document_retriever.add_documents(docs_splits)


# Create a retrieval-augmented generation (RAG) chain
rag_chain_pr = (
{"context": parent_document_retriever | format_docs, "question": RunnablePassthrough()}
| RunnablePassthrough.assign(context=itemgetter("context"))
| {"answer": prompt | llm | StrOutputParser(), 
 "context": itemgetter("context")}
)

# Get the RAG dataset 
rag_dataset = get_questions_answers_contexts(rag_chain_pr)

# Calculate the metrics
results = get_metrics(rag_dataset)
print(results)

结果如下：

{
'faithfulness': 0.6667, 
'answer_relevancy': 0.4867, 
'context_precision': 0.7778, 
'context_recall': 0.6574
}

结果显示，这种变化对提升性能并没有什么帮助。上下文召回率的下降说明检索过程没正常工作，上下文里缺少完整的信息。忠实度和答案相关性指标的变化，也是因为上下文质量不佳导致的。这种情况下，可以尝试评估其他分块和检索方法。

嵌入模型

嵌入模型负责把文本块转换成稠密向量表示。不同的模型可能针对不同主题进行过训练，有时能带来更好的嵌入效果。选择嵌入方法时，需要在计算效率和嵌入质量之间做个权衡。

我们比较了几种不同的嵌入模型，比如密集段落检索（"facebook/dpr-question_encoder-single-nq-base"）、Sentence-BERT（"paraphrase-MiniLM-L6-v2"），以及Chroma的默认模型（"all-MiniLM-L6-v2"）。每个模型都有自己的优势，在特定领域的数据上评估一下，就能找出哪个模型能提供最准确的语义表示。

要更换嵌入模型，需要定义一个名为 "SentenceBertEncoderEmbeddings" 的新类，实现Sentence-BERT模型。然后，用这个新类替换掉之前实现了密集段落检索模型的 "DPRQuestionEncoderEmbeddings"。下面是使用Sentence-BERT模型进行测试的代码：

# Import necessary libraries and modules
import pandas as pd
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
context_precision,
faithfulness,
answer_relevancy,
context_recall
)
from langchain_community.chat_models import ChatOllama
from sentence_transformers import SentenceTransformer


# Define a custom embedding class using the DPRQuestionEncoder
class SentenceBertEncoderEmbeddings(Embeddings):
show_progress: bool = False
"""Whether to show a tqdm progress bar. Must ha ve `tqdm` installed."""

def __init__(self, model_name: str = 'paraphrase-MiniLM-L6-v2'):
# Initialize the tokenizer and model with the specified model name
self.model = SentenceTransformer(model_name)

def embed(self, texts):
# Ensure texts is a list
if isinstance(texts, str):
texts = [texts]

embeddings = []
if self.show_progress:
try:
from tqdm import tqdm
iter_ = tqdm(texts, desc="Embeddings")
except ImportError:
logger.warning(
"Unable to show progress bar because tqdm could not be imported. "
"Please install with `pip install tqdm`."
)
iter_ = texts
else:
iter_ = texts

for text in iter_:
embeddings.append(self.model.encode(text).tolist())

return embeddings

def embed_documents(self, documents):
return self.embed(documents)

def embed_query(self, query):
return self.embed([query])[0]


# Create a Chroma vector store from the documents using the custom embeddings
vectorstore = Chroma.from_documents(documents=docs_splits, embedding=SentenceBertEncoderEmbeddings())

# Create a retriever from the vector store
retriever = vectorstore.as_retriever()

# Create a retrieval-augmented generation (RAG) chain
rag_chain_ce = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| RunnablePassthrough.assign(context=itemgetter("context"))
| {"answer": prompt | llm | StrOutputParser(), 
 "context": itemgetter("context")})

# Get the RAG dataset 
rag_dataset = get_questions_answers_contexts(rag_chain_ce)

# Calculate the metrics
results = get_metrics(rag_dataset)
print(results)

结果如下：

{
'faithfulness': 0.5278, 
'answer_relevancy': 0.5306, 
'context_precision': 0.5556, 
'context_recall': 0.7997
}

这种情况下，编码器的变化导致了指标性能下降。这并不意外，因为DPR的检索精度高于Sentence-BERT，在我们这个需要精确文档检索的场景下更合适。切换到Sentence-BERT后，'faithfulness' 和 'answer relevancy' 指标的显著下降，恰恰凸显了为高精度检索任务选择合适的嵌入模型是多么重要。

向量搜索方法

向量搜索方法根据相似度度量来检索最相关的块。常见的方法包括欧氏距离（L2）、余弦相似度等。调整搜索方法有望提升最终输出的质量。

代码如下：

# Import necessary libraries and modules
import pandas as pd
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
context_precision,
faithfulness,
answer_relevancy,
context_recall
)
from langchain_community.chat_models import ChatOllama

# Create a Chroma vector store from the documents 
# using the custom embeddings and also changing to 
# cosine similarity search
vectorstore = Chroma.from_documents(collection_name="dist", 
documents=docs_splits, 
embedding=DPRQuestionEncoderEmbeddings(), 
collection_metadata={"hnsw:space": "cosine"})

# Create a retriever from the vector store
retriever = vectorstore.as_retriever()

# Create a retrieval-augmented generation (RAG) chain
rag_chain_dist = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| RunnablePassthrough.assign(context=itemgetter("context"))
| {"answer": prompt | llm | StrOutputParser(), 
 "context": itemgetter("context")})

# Get the RAG dataset 
rag_dataset = get_questions_answers_contexts(rag_chain_dist)

# Calculate the metrics
results = get_metrics(rag_dataset)
print(results)

结果如下：

{
'faithfulness': 0.9444, 
'answer_relevancy': 0.8504, 
'context_precision': 0.6667, 
'context_recall': 0.8889
}

'faithfulness' 的提升表明，尽管 'context precision' 有所下降，但使用余弦相似度进行向量搜索，确实增强了检索到的文档与查询之间的一致性。总体来看，较高的 'faithfulness' 和 'context recall' 说明，在当前场景下，余弦相似度是一种更有效的向量搜索方法，这也印证了选择合适的搜索方法对于优化检索性能的重要性。

用于馈送模型的最终提示

最终提示的构建，本质上是把检索到的数据整合进模型查询中。提示词里的细微改动都可能对结果产生显著影响，这往往是个需要反复试验的过程。在提示词中给出一些示例，可以引导模型生成更准确、更相关的输出。

结论

优化检索增强生成（RAG）管道是一个迭代过程，很大程度上取决于应用程序的具体数据和上下文。本文探讨了四种关键策略：改进分块方法、选择并微调嵌入模型、挑选有效的向量搜索方法，以及设计精确的提示词。每一个环节在提升RAG系统性能方面都扮演着重要角色。

结果也再次证明，

。例如，在我们的场景下，密集段落检索（DPR）优于Sentence-BERT，但换一个数据集或任务要求，结果可能就大不相同。同样，切换到余弦相似度进行向量搜索，带来了更好的置信度和上下文回忆率，这说明哪怕检索过程中的一个小调整，也能举足轻重。

优化RAG管道的旅程，就是持续测试、从失败中学习、并进行明智调整的过程。通过拥抱这种迭代方法，你可以让自己的AI解决方案更有效地满足特定需求。记住，成功的关键在于深刻理解你的数据，大胆尝试不同策略，并不断完善你的流程。

参考文献

[1]Es, S., James, J., Espinosa-Anke, L., & Schockaert, S. (2023). RAGAS: Automated Evaluation of Retrieval Augmented Generation. Exploding Gradients, CardiffNLP, Cardiff University, AMPLYFI.