基于 MLOps 的 RAG 文档迭代自动评估体系构建与工程化落地实践
大家好，今天我将分享关于基于 MLOps 的 RAG (Retrieval-Augmented Generation) 文档迭代自动评估体系构建与工程化落地实践。RAG 在大型语言模型 (LLM) 的应用中扮演着越来越重要的角色，特别是在知识密集型任务中。然而，RAG 系统的效果很大程度上取决于其知识库的质量。因此，如何持续迭代和优化知识库文档，并对迭代效果进行有效评估，成为了一个关键挑战。
本次分享将围绕以下几个方面展开：
- RAG 系统及文档迭代面临的挑战
- 自动评估体系的设计原则
- 评估指标的选择与实现
- 基于 MLOps 的自动化流程构建
- 工程化落地实践案例
- 未来展望
1. RAG 系统及文档迭代面临的挑战
RAG 系统通过检索外部知识库，增强 LLM 的生成能力，使其能够回答更复杂、更专业的问题。一个典型的 RAG 流程包括：
- 索引 (Indexing): 将文档切分成块 (chunks)，并为每个块生成向量嵌入 (embeddings)。
- 检索 (Retrieval): 接收用户查询，生成查询向量，并在向量数据库中找到最相关的文档块。
- 生成 (Generation): 将检索到的文档块和用户查询一起输入 LLM，生成最终答案。
RAG 系统的性能高度依赖于知识库文档的质量。文档可能存在以下问题：
- 信息过时: 文档包含的信息已经不再准确或有效。
- 信息冗余: 多个文档包含相同或相似的信息。
- 信息缺失: 文档未能覆盖用户可能关心的关键信息。
- 格式不规范: 文档格式不一致，导致检索效果下降。
- 噪音信息: 文档包含大量与主题无关的信息，干扰检索和生成。
为了解决这些问题，我们需要持续迭代和优化知识库文档。然而，手动评估文档迭代的效果非常耗时且主观。因此，构建一个自动评估体系至关重要。
2. 自动评估体系的设计原则
一个有效的自动评估体系应该具备以下几个关键原则：
- 全面性: 评估指标应能覆盖 RAG 系统的各个方面，包括检索质量、生成质量和整体性能。
- 客观性: 评估过程应尽可能减少人为干预，采用自动化指标和流程。
- 可解释性: 评估结果应能提供关于文档质量的洞察，帮助我们找到改进方向。
- 可扩展性: 评估体系应能适应不同类型的文档和 RAG 系统。
- 高效性: 评估过程应尽可能快速，以便及时反馈文档迭代的效果。
基于这些原则，我们可以设计一个包含以下组件的自动评估体系：
- 评估数据集: 包含问题、参考答案 (ground truth) 和可选的上下文信息。
- 评估指标: 用于衡量 RAG 系统在评估数据集上的表现。
- 自动化流程: 用于自动运行评估，生成评估报告。
- 监控和告警: 用于监控评估指标的变化，并在出现异常时发出告警。
3. 评估指标的选择与实现
选择合适的评估指标是构建自动评估体系的关键。以下是一些常用的评估指标，以及它们的实现方法：
3.1 检索质量指标
-
Hit Rate: 衡量检索结果中是否包含正确的文档块。
def calculate_hit_rate(retrieved_docs, relevant_doc_ids): """ 计算 Hit Rate。 Args: retrieved_docs: 检索到的文档 ID 列表。 relevant_doc_ids: 相关的文档 ID 列表。 Returns: Hit Rate 值。 """ hits = 0 for doc_id in retrieved_docs: if doc_id in relevant_doc_ids: hits += 1 return hits / len(relevant_doc_ids) if len(relevant_doc_ids) > 0 else 0 -
Precision@K: 衡量前 K 个检索结果中有多少是相关的。
def calculate_precision_at_k(retrieved_docs, relevant_doc_ids, k): """ 计算 Precision@K。 Args: retrieved_docs: 检索到的文档 ID 列表。 relevant_doc_ids: 相关的文档 ID 列表。 k: 考虑前 K 个结果。 Returns: Precision@K 值。 """ hits = 0 for i in range(min(k, len(retrieved_docs))): if retrieved_docs[i] in relevant_doc_ids: hits += 1 return hits / k if k > 0 else 0 -
Recall@K: 衡量所有相关的文档块中有多少被检索到。
def calculate_recall_at_k(retrieved_docs, relevant_doc_ids, k): """ 计算 Recall@K。 Args: retrieved_docs: 检索到的文档 ID 列表。 relevant_doc_ids: 相关的文档 ID 列表。 k: 考虑前 K 个结果。 Returns: Recall@K 值。 """ hits = 0 for i in range(min(k, len(retrieved_docs))): if retrieved_docs[i] in relevant_doc_ids: hits += 1 return hits / len(relevant_doc_ids) if len(relevant_doc_ids) > 0 else 0 -
Mean Reciprocal Rank (MRR): 衡量第一个相关文档块在检索结果中的排名。
def calculate_mrr(retrieved_docs, relevant_doc_ids): """ 计算 Mean Reciprocal Rank (MRR)。 Args: retrieved_docs: 检索到的文档 ID 列表。 relevant_doc_ids: 相关的文档 ID 列表。 Returns: MRR 值。 """ for i, doc_id in enumerate(retrieved_docs): if doc_id in relevant_doc_ids: return 1 / (i + 1) return 0
3.2 生成质量指标
-
ROUGE (Recall-Oriented Understudy for Gisting Evaluation): 衡量生成答案与参考答案之间的重叠程度。
from rouge import Rouge def calculate_rouge(generated_answer, reference_answer): """ 计算 ROUGE 指标。 Args: generated_answer: 生成的答案。 reference_answer: 参考答案。 Returns: ROUGE 指标字典。 """ rouge = Rouge() scores = rouge.get_scores(generated_answer, reference_answer)[0] return scores -
BLEU (Bilingual Evaluation Understudy): 衡量生成答案与参考答案之间的相似度。
from nltk.translate.bleu_score import sentence_bleu def calculate_bleu(generated_answer, reference_answer): """ 计算 BLEU 指标。 Args: generated_answer: 生成的答案。 reference_answer: 参考答案。 Returns: BLEU 值。 """ reference = [reference_answer.split()] candidate = generated_answer.split() score = sentence_bleu(reference, candidate) return score -
BERTScore: 使用 BERT 模型计算生成答案与参考答案之间的语义相似度。
from bert_score import score def calculate_bertscore(generated_answer, reference_answer): """ 计算 BERTScore 指标。 Args: generated_answer: 生成的答案。 reference_answer: 参考答案。 Returns: BERTScore 指标 (Precision, Recall, F1)。 """ P, R, F1 = score([generated_answer], [reference_answer], lang="en", verbose=False) return {"precision": P.item(), "recall": R.item(), "f1": F1.item()} -
Relevance: 衡量生成答案是否与问题相关。可以使用 LLM 进行判断。
from openai import OpenAI def calculate_relevance(question, generated_answer, openai_api_key): """ 使用 LLM 计算生成答案与问题的相关性。 Args: question: 问题。 generated_answer: 生成的答案。 openai_api_key: OpenAI API Key. Returns: 相关性得分 (0-1)。 """ client = OpenAI(api_key=openai_api_key) prompt = f""" 请判断以下答案是否与问题相关。请输出一个 0 到 1 之间的分数，其中 0 表示完全不相关，1 表示完全相关。 问题：{question} 答案：{generated_answer} 相关性得分： """ try: response = client.completions.create( model="text-davinci-003", prompt=prompt, max_tokens=10, n=1, stop=None, temperature=0.0, ) relevance_score = float(response.choices[0].text.strip()) return relevance_score except Exception as e: print(f"Error calculating relevance: {e}") return 0.0 -
Faithfulness/Groundedness: 衡量生成答案是否忠实于检索到的文档。可以使用 LLM 进行判断。
def calculate_faithfulness(context, generated_answer, openai_api_key): """ 使用 LLM 计算生成答案是否忠实于检索到的文档。 Args: context: 检索到的文档。 generated_answer: 生成的答案。 openai_api_key: OpenAI API Key. Returns: 忠实度得分 (0-1)。 """ client = OpenAI(api_key=openai_api_key) prompt = f""" 请判断以下答案是否与提供的文档一致。请输出一个 0 到 1 之间的分数，其中 0 表示完全不一致，1 表示完全一致。 文档：{context} 答案：{generated_answer} 忠实度得分： """ try: response = client.completions.create( model="text-davinci-003", prompt=prompt, max_tokens=10, n=1, stop=None, temperature=0.0, ) faithfulness_score = float(response.choices[0].text.strip()) return faithfulness_score except Exception as e: print(f"Error calculating faithfulness: {e}") return 0.0
3.3 综合指标
-
RAGAS (RAG Assessment Score): 一个综合评估 RAG 系统性能的指标，包括 Faithfulness, Answer Relevance 和 Context Relevance。
RAGAS 已经有对应的库可以使用，这里不再重复实现。
from ragas import evaluate from ragas.metrics import faithfulness, answer_relevancy, context_relevancy def calculate_ragas(questions, contexts, answers, ground_truths): """ 计算 RAGAS 指标。 Args: questions: 问题列表。 contexts: 检索到的文档列表。 answers: 生成的答案列表。 ground_truths: 参考答案列表。 Returns: RAGAS 评估结果。 """ data = { "question": questions, "contexts": contexts, "answer": answers, "ground_truths": ground_truths, } df = pd.DataFrame(data) results = evaluate( df, metrics=[ faithfulness, answer_relevancy, context_relevancy, ], ) return results
3.4 指标选择建议
选择哪些指标取决于具体的应用场景和关注点。一般来说，建议同时使用多个指标，从不同角度评估 RAG 系统的性能。
- 检索质量: 如果关注检索结果的准确性，可以使用 Hit Rate, Precision@K, Recall@K 和 MRR。
- 生成质量: 如果关注生成答案的质量，可以使用 ROUGE, BLEU, BERTScore, Relevance 和 Faithfulness。
- 综合评估: 如果需要一个综合性的评估指标，可以使用 RAGAS。
4. 基于 MLOps 的自动化流程构建
为了实现自动评估，我们需要构建一个基于 MLOps 的自动化流程。一个典型的流程包括以下几个步骤：
- 数据准备: 从评估数据集中加载问题和参考答案。
- 检索: 使用 RAG 系统检索相关的文档块。
- 生成: 使用 LLM 生成答案。
- 评估: 使用选定的评估指标计算 RAG 系统的性能。
- 报告: 生成评估报告，包含评估指标和可视化结果。
- 监控: 监控评估指标的变化，并在出现异常时发出告警。
可以使用以下工具和技术来实现自动化流程：
- 数据管理: 使用数据库或云存储服务 (例如 AWS S3, Google Cloud Storage) 存储评估数据集。
- 流程编排: 使用 Airflow, Kubeflow Pipelines 或 Prefect 等工具编排自动化流程。
- 模型管理: 使用 MLflow 或 Weights & Biases 等工具管理 RAG 系统和 LLM 的版本。
- 监控告警: 使用 Prometheus 和 Grafana 等工具监控评估指标，并设置告警规则。
以下是一个使用 Airflow 编排自动化流程的示例：
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import pandas as pd
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_relevancy
def load_evaluation_data():
"""
加载评估数据集。
"""
# 从 CSV 文件加载评估数据集
df = pd.read_csv("evaluation_data.csv")
return df
def run_rag_pipeline(question):
"""
运行 RAG 流程。
"""
# TODO: 实现 RAG 流程，包括检索和生成
context = "This is a sample context." # 替换为实际的检索结果
answer = "This is a sample answer." # 替换为实际的生成结果
return context, answer
def evaluate_rag_pipeline(df):
"""
评估 RAG 流程。
"""
questions = df["question"].tolist()
contexts = []
answers = []
ground_truths = df["ground_truth"].tolist()
for question in questions:
context, answer = run_rag_pipeline(question)
contexts.append(context)
answers.append(answer)
data = {
"question": questions,
"contexts": contexts,
"answer": answers,
"ground_truths": ground_truths,
}
df_eval = pd.DataFrame(data)
results = evaluate(
df_eval,
metrics=[
faithfulness,
answer_relevancy,
context_relevancy,
],
)
return results.to_json()
def save_evaluation_report(results_json):
"""
保存评估报告。
"""
with open("evaluation_report.json", "w") as f:
f.write(results_json)
with DAG(
dag_id="rag_evaluation_pipeline",
schedule_interval=None,
start_date=datetime(2023, 1, 1),
catchup=False,
tags=["rag", "evaluation"],
) as dag:
load_data_task = PythonOperator(
task_id="load_evaluation_data",
python_callable=load_evaluation_data,
)
evaluate_task = PythonOperator(
task_id="evaluate_rag_pipeline",
python_callable=evaluate_rag_pipeline,
op_kwargs={"df": load_data_task.output},
)
save_report_task = PythonOperator(
task_id="save_evaluation_report",
python_callable=save_evaluation_report,
op_kwargs={"results_json": evaluate_task.output},
)
load_data_task >> evaluate_task >> save_report_task5. 工程化落地实践案例
假设我们正在构建一个基于 RAG 的智能客服系统，用于回答用户关于公司产品的咨询。
- 数据准备: 收集用户常见的咨询问题和对应的参考答案，构建评估数据集。
- RAG 系统: 使用 FAISS 作为向量数据库，OpenAI 的
text-embedding-ada-002模型生成向量嵌入，OpenAI 的gpt-3.5-turbo模型生成答案。 - 评估指标: 选择 Hit Rate, ROUGE 和 Faithfulness 作为评估指标。
- 自动化流程: 使用 Airflow 编排自动化流程，每天自动运行评估，并将评估报告发送到 Slack 频道。
- 文档迭代: 根据评估报告，识别文档中存在的问题，并进行相应的修改。例如，如果发现某个文档的 Faithfulness 得分较低，说明 LLM 生成的答案与该文档不一致，需要检查文档内容是否准确。
通过以上步骤，我们可以持续迭代和优化知识库文档，提高 RAG 系统的性能，并最终提升用户满意度。
以下是一个简化的数据迭代和评估的示例：
import pandas as pd
from openai import OpenAI
from rouge import Rouge
from bert_score import score
from sklearn.model_selection import train_test_split
# 1. 数据准备
data = {
'question': [
"What is the company's return policy?",
"How do I reset my password?",
"What are the shipping options?",
"What is the warranty on this product?",
"How do I contact customer support?"
],
'context': [
"Our return policy allows returns within 30 days of purchase.",
"To reset your password, click on the 'Forgot Password' link on the login page.",
"We offer standard and express shipping options.",
"The product has a one-year warranty.",
"You can contact customer support via phone, email, or chat."
],
'answer': [
"You can return items within 30 days of purchase.",
"Click 'Forgot Password' on the login page to reset your password.",
"We have standard and express shipping.",
"The product is warranted for one year.",
"Contact customer support via phone, email, or chat."
]
}
df = pd.DataFrame(data)
# 分割数据集为训练集和测试集
train_df, test_df = train_test_split(df, test_size=0.2, random_state=42)
# 2. 模拟文档迭代 (这里简化为直接修改数据)
# 假设我们发现关于退货政策的文档不够清晰，进行了修改
train_df.loc[train_df['question'] == "What is the company's return policy?", 'context'] = "Items can be returned within 30 days for a full refund."
# 3. RAG 系统模拟 (简化)
def rag_pipeline(question, context):
# 模拟检索和生成
# 在实际应用中，这里会涉及向量数据库查询和 LLM 调用
return context
# 4. 评估
def evaluate(df, openai_api_key):
rouge = Rouge()
results = []
for index, row in df.iterrows():
question = row['question']
context = row['context']
reference_answer = row['answer']
# 模拟 RAG 流程
generated_answer = rag_pipeline(question, context)
# 计算 ROUGE
rouge_scores = rouge.get_scores(generated_answer, reference_answer)[0]
# 计算 BERTScore
P, R, F1 = score([generated_answer], [reference_answer], lang="en", verbose=False)
bertscore = {"precision": P.item(), "recall": R.item(), "f1": F1.item()}
# 计算 Faithfulness
client = OpenAI(api_key=openai_api_key)
prompt = f"""
请判断以下答案是否与提供的文档一致。请输出一个 0 到 1 之间的分数，其中 0 表示完全不一致，1 表示完全一致。
文档：{context}
答案：{generated_answer}
忠实度得分：
"""
try:
response = client.completions.create(
model="text-davinci-003",
prompt=prompt,
max_tokens=10,
n=1,
stop=None,
temperature=0.0,
)
faithfulness_score = float(response.choices[0].text.strip())
except Exception as e:
print(f"Error calculating faithfulness: {e}")
faithfulness_score = 0.0
results.append({
'question': question,
'rouge': rouge_scores,
'bertscore': bertscore,
'faithfulness': faithfulness_score
})
return pd.DataFrame(results)
# 5. 运行评估
openai_api_key = "YOUR_OPENAI_API_KEY" # 替换为你的 OpenAI API Key
results_df = evaluate(test_df, openai_api_key)
# 6. 输出结果
print(results_df)
# 可以对结果进行分析，例如计算平均 ROUGE 和 Faithfulness，并与迭代前的数据进行比较这个例子展示了数据准备、文档迭代的模拟、RAG流程的简化以及使用 ROUGE, BERTScore, 和 Faithfulness 进行评估的过程。实际应用中，RAG 流程会更加复杂，包括向量数据库查询和 LLM 调用。
6. 未来展望
RAG 文档迭代自动评估是一个持续发展的领域。未来，我们可以探索以下方向：
- 更智能的评估指标: 使用 LLM 进行更细粒度的评估，例如评估答案的完整性、准确性和流畅性。
- 自动化文档优化: 利用 LLM 自动生成文档摘要、纠正语法错误和提高可读性。
- 主动学习: 根据评估结果，主动选择哪些文档需要更新，以提高迭代效率。
- 多语言支持: 扩展评估体系，支持多种语言的文档和 RAG 系统。
- 更强大的 MLOps 平台: 构建更易于使用和管理的 MLOps 平台，降低自动化评估的门槛。
自动评估是 RAG 系统持续优化的关键
RAG 系统的文档迭代自动评估体系的构建与工程化落地是提升系统性能的关键步骤。通过建立全面的评估指标体系和高效的自动化流程，我们可以持续优化知识库文档，提高 RAG 系统的准确性和可靠性，从而更好地服务于用户。
选择合适的评估指标并持续迭代
自动评估体系的有效性取决于评估指标的选择和自动化流程的构建。选择合适的评估指标，并根据实际应用场景进行调整和优化，是确保评估结果准确性和可靠性的关键。
MLOps 平台支持自动化流程的高效运行
基于 MLOps 的自动化流程可以大大提高评估效率，并为文档迭代提供及时反馈。利用 Airflow, Kubeflow Pipelines 或 Prefect 等工具编排自动化流程，可以实现评估流程的自动化运行、监控和告警，从而确保 RAG 系统的持续优化。