Pathway RAG 评测流水线实战：基于 CUAD 数据集的检索指标与 RAGAS 自动化评估指南

Pathway RAG 评测流水线实战：基于 CUAD 数据集的检索指标与 RAGAS 自动化评估指南

【免费下载链接】pathwayPython ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.项目地址: https://gitcode.com/GitHub_Trending/pa/pathway

本篇技术指南聚焦 Pathway 仓库中integration_tests/rag_evals/目录：一个面向 RAG（检索增强生成）应用的自动化评测与回归测试框架。它通过app.yaml拉起一个基于 Pathway 实时数据框架的 RAG 服务，以 CUAD 合同理解数据集与合成 QA 数据集为基准，从"文档检索质量"与"问答回答质量"两条主线产出 MRR、Hit@k、RAG Accuracy、RAGAS（AnswerCorrectness、Faithfulness）等量化指标，并把应用配置、问答响应与全部指标统一记录到 MLFlow。读完本文，你将掌握如何在该框架上替换数据源、配置评测环境、读懂评测指标的计算方式，以及如何定位"答非所问"等典型故障。

评测框架概览：两大评测类别 + MLFlow 统一日志

按 README 的说明，这套自动化集成测试共覆盖两大评测类别：

• Retrieval metrics（检索指标）：衡量 RAG 应用的索引与召回模块是否能把包含正确答案的文档片段排到靠前位置，核心实现位于 evaluator.py；
• RAGAS 指标：借助ragas库对回答的忠实度（Faithfulness）与答案正确性（AnswerCorrectness）做 LLM-as-a-judge 打分，见 ragas_utils.py。

在评测运行期间，App 配置（app.yaml原文）、每条问答的请求/响应、以及所有评测指标都会被记录到框架内部的 MLFlow server 上。实验入口 experiment.py 中固定使用实验名EXPERIMENT_NAME = "CI RAG Evals"，并为每次运行生成独立的 Run，同时以 Git 分支名（环境变量BRANCH，缺失时从git探测）打上Branch name标签——这意味着它可以同时服务"本地调参实验"与"CI 上的每日回归"，方便对照不同配置、不同分支的评测分数。

数据集准备：CUAD 与合成数据

CUAD 合同问答数据集

评测主要依赖CUAD（Contract Understanding Atticus Dataset）合同理解数据集。框架按"文件—问题—标准答案"三元组的方式组织评测数据，逐行存放在dataset/下的制表符分隔文件（TSV）中，仓库内已有示例文件 dataset/labeled.tsv。

该文件的表头结构反映了 CUAD 的问答形式：每个合同条款类别都有一对列，例如Parties（检索用：存放期望被召回的证据片段列表，形如['BIRCH FIRST GLOBAL INVESTMENTS INC.', 'MA', ...]）与Parties-Answer（回答用：存放标准答案文本，例如Birch First Global Investments Inc. ("Company"); Mount Kowledge Holdings Inc. ("Marketing Affiliate", "MA")）。

真正参与评测的条款类别并不是 CUAD 全量 50 余类，而是从 eval_questions.py 中的EVAL_QUESTIONS选出并展开得到的 28 个问题类型（14 个"检索型 + Answer 型"组合），包括：Parties、Agreement Date、Effective Date、Expiration Date、Governing Law、Competitive Restriction Exception、Exclusivity、Non-Disparagement、Anti-Assignment、Minimum Commitment、License Grant、Non-Transferable License、Audit Rights、Cap On Liability。为便于把紧凑的列名转成适合大模型回答的自然语言问题，同一文件还维护了question_mapper字典（例如把Parties映射为 "Who are the parties in the contract (licensee and licensor)?"）。

使用自定义数据集时，按以下方式接入：

1. 将评测 TSV 下载/导出为Tab Separated Values (.tsv)格式（电子表格工具中通常为 File -> Download -> Tab Separated Values）；
2. 放入仓库的dataset/目录（对应 dataset/labeled.tsv）；
3. 保持列结构与本仓库的labeled.tsv一致，即每行一个合同文件，每列一个条款类别（含-Answer后缀的列存答案）。

合成 QA 数据集

除 CUAD 外，dataset/synthetic_tests/ 下还预置了两份合成 QA 数据，用于在标准答案文本上做更贴近真实问答形态的评估：

• 20230203_alphabet_10K.jsonl（映射为AlphabetQA）；
• 20230203_alphabet_10K_tables.jsonl（映射为AlphabetTables）。

从 experiment.py 中的FILE_TO_DATASET_META可见，二者都源自同一份公开文档20230203_alphabet_10K.pdf，通过程序化方式从文档中抽取问答对（user_input+reference_contexts证据片段），生成细节参考了官方发布的技术博客。每条记录均符合ragas.EvaluationDataset.from_jsonl可直接读取的 JSONL 结构。

安装与运行环境准备

依赖安装

首先确保已安装 Pathway 本体及其扩展依赖：

pip install "pathway[all]"

随后安装评测侧所需的额外依赖（ragas、mlflow、evaluate、bert_score、seaborn、langchain-openai等），见 requirements.txt：

pip install -r integration_tests/rag_evals/requirements.txt

环境变量（.env）

评测运行依赖以下环境变量，参考文件为 .example.env：

OPENAI_API_KEY=sk-... RUN_MODE=LOCAL # RUN_MODE=CI MLFLOW_URI="https://..."

复制.example.env为.env并填入真实值即可（框架侧通过load_dotenv()加载）。

理解 RAG 应用配置：app.yaml 逐节拆解

app.yaml 是本次评测所驱动的Pathway Live Data Framework RAG 应用的完整配置，使用 Pathway 的 YAML 配置语法，通过$前缀声明可复用组件，最终导出顶层节点question_answerer。逐节说明如下。

$sources：数据源连接器（运行前必须按需替换）

$sources: - !pw.io.gdrive.read object_id: "1ErwN5WajWsEdIRMBIjyBfncNUmkxRfRy" # large: "..." # small: "..." service_user_credentials_file: /credentials/credentials.json # ./gdrive_indexer.json # /integration_tests/rag_evals/gdrive_indexer.json file_name_pattern: - "*.pdf" - "*.docx" object_size_limit: null with_metadata: true refresh_interval: 30 - !pw.io.gdrive.read object_id: "1GC0jVKLd2_GZb4pJx1umgJwmjOCxzkDW" # synthetic dataset service_user_credentials_file: /credentials/credentials.json file_name_pattern: - "*.pdf" - "*.docx" object_size_limit: null with_metadata: true refresh_interval: 30

默认配置通过pw.io.gdrive.read连接器指向两个 Google Drive 文件夹：一个存放选定版本的 CUAD 合同 PDF/DOCX 原文，另一个存放合成数据集对应的 PDF。关键参数：

• object_id：目标 Drive 文件夹 ID；
• service_user_credentials_file：服务账号凭据 JSON 路径；
• file_name_pattern：按文件名通配符过滤要解析的文件；
• with_metadata: true：保留文件路径等元数据（评测依赖路径做文件级过滤，见下文globmatch过滤）；
• refresh_interval: 30：以秒为单位的轮询刷新周期，体现 Pathway 的实时数据特性。

⚠️ 接入你自己的数据时，需要修改$sources段，将其替换为能访问到你的评测文件的连接器（本地目录、S3、Drive 等均可）。若使用本地文件连接器，建议保持with_metadata并让元数据中包含可标识文件名/路径的字段，否则文件级filter无法生效。

$llm / $embedder：LLM 与嵌入模型

$llm: !pw.xpacks.llm.llms.OpenAIChat model: "gpt-4o-mini" retry_strategy: !pw.udfs.ExponentialBackoffRetryStrategy max_retries: 6 cache_strategy: !pw.udfs.DefaultCache {} temperature: 0 capacity: 8 $embedder: !pw.xpacks.llm.embedders.OpenAIEmbedder model: "text-embedding-ada-002" retry_strategy: !pw.udfs.ExponentialBackoffRetryStrategy max_retries: 6 cache_strategy: !pw.udfs.DefaultCache {}

• temperature: 0保证回答输出的确定性，对可复现评测至关重要；
• ExponentialBackoffRetryStrategy(max_retries=6)与DefaultCache降低 API 抖动与重复调用的成本；
• capacity: 8控制 LLM 侧的并发吞吐上限。

$parser / $splitter：解析与切分

$splitter: !pw.xpacks.llm.splitters.TokenCountSplitter min_tokens: 250 max_tokens: 600 $parser: !pw.xpacks.llm.parsers.DoclingParser {}

• 解析器使用DoclingParser，负责把 PDF/DOCX 等原始文档解析为可索引文本；
• 切分器使用按 Token 计数的TokenCountSplitter，将文本切成 250～600 token 的块，兼顾召回粒度与上下文完整度。

混合检索：KNN 向量检索 + BM25 关键词检索

$knn_index: !pw.stdlib.indexing.BruteForceKnnFactory reserved_space: 1000 embedder: $embedder metric: !pw.engine.BruteForceKnnMetricKind.COS $bm25_index: !pw.stdlib.indexing.TantivyBM25Factory {} $retriever_factory: !pw.stdlib.indexing.HybridIndexFactory retriever_factories: - $knn_index - $bm25_index

检索侧采用混合索引：BruteForceKnnFactory基于余弦相似度做稠密向量召回，TantivyBM25Factory（底层为 tantivy 全文检索引擎，仓库library_licenses/中可见其 license 文件）提供稀疏关键词召回，两者通过HybridIndexFactory融合，兼顾语义匹配与字面命中。CUAD 合同场景中大量条款措辞高度专业化，混合检索是保证召回质量的关键设计。

DocumentStore 与问答器

$document_store: !pw.xpacks.llm.document_store.DocumentStore docs: $sources parser: $parser splitter: $splitter retriever_factory: $retriever_factory # https://smith.langchain.com/hub/rlm/rag-prompt $prompt_template: | You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know. Question: {query} Context: {context} Answer: question_answerer: !pw.xpacks.llm.question_answering.BaseRAGQuestionAnswerer llm: $llm indexer: $document_store prompt_template: $prompt_template search_topk: 8 with_cache: true

DocumentStore把"数据源 → 解析 → 切分 → 混合索引"整条链路串起来；BaseRAGQuestionAnswerer在检索时取search_topk: 8个文档块拼入prompt_template交给 LLM。提示词显式要求"不知道就直说不知道"，这一设计与评测侧"无法从文档推断时应回答 I don't know"的判定规则（见下文回答评估）是自洽配合的。配置末尾with_cache: true开启缓存，实际运行时的缓存后端由应用入口（test_eval.py、debug_main.py）以pw.persistence.Backend.filesystem("Cache")落盘。

评测执行入口与 REST 服务形态

评测框架按"先跑服务、再打接口"的集成测试模式工作，核心入口有两个：

1. 应用入口test_eval.py：读取app.yaml构造App（内含question_answerer与host/port），通过QASummaryRestServer启动 HTTP 服务，随后在wait_result_with_checker的检查函数内执行评测；conftest.py会自动下载nltk的punkt数据供分词使用。
2. 评测客户端connector.py：RagConnector封装了与运行中服务通信的两个 HTTP 端点——
   • POST /v2/answer：携带prompt、可选的filters（元数据过滤表达式）、model、return_context_docs，返回生成答案与命中的上下文文档；
   • POST /v2/list_documents：列出当前索引到的文档（可按metadata_filter过滤、按keys裁剪字段）。

文件过滤表达式由 utils.py 的create_file_filter生成，形式为globmatch(\<file_name>`, path)`，即只允许系统回答"来自指定合同文件"的问题，避免跨文件泄题污染评测结果。

CI 语义上，test_eval.py 的wait_for_start会先轮询/v2/list_documents，直到索引文档数达到预期（代码注释中的 CUAD 文件数 23 + 合成数据 1），服务就绪后才开始评测；随后以MIN_ACCURACY（CI 下 0.5，debug_main.py 中的独立冒烟为 0.6）作为整体准确率下限判定通过与否。

本地运行步骤

本地联调时并不需要手动起服务——评测脚本本身就是"起服务 + 打接口"的合体，直接按 README 与 run_locally.sh 执行即可。run_locally.sh首先做一件必要的事：用sed把app.yaml中指向 CI 环境的凭据路径/integration_tests/rag_evals/gdrive_indexer.json改写为本地相对路径./gdrive_indexer.json，然后运行pytest test_eval.py：

file_path="./app.yaml" sed -i 's|service_user_credentials_file: /integration_tests/rag_evals/gdrive_indexer.json|service_user_credentials_file: ./gdrive_indexer.json|' "$file_path" pytest test_eval.py

即本地运行的最小操作序列为：

cd integration_tests/rag_evals pip install -r requirements.txt # 1. 复制 .example.env 为 .env 并填入 OPENAI_API_KEY、RUN_MODE=LOCAL、MLFLOW_URI # 2. 按需修改 app.yaml 的 $sources bash run_locally.sh

MLFlow：评测日志中心

如 README 所述，评测的前提是有一个可用的 MLFlow Tracking Server来接收应用的配置、预测与指标。需按 MLFlow 官方 quickstart 启动 Tracking Server 后，将地址写入.env的MLFLOW_URI。

experiment.py 的run_eval_experiment展示了完整的日志契约：

• 校验MLFLOW_URI，mlflow.set_tracking_uri(...)；
• Run 名默认取当前时间戳，实验名为固定值CI RAG Evals；
• 记录标签Branch name，并把app.yaml作为 artifact 归档，保证"哪个配置跑出哪个分数"可追溯；
• 接着顺序执行CUAD 评测（run_cuad_eval）与合成数据评测（run_alphabet_eval）。

CUAD 评测写入 MLFlow 的内容包括：

若在mlflow.log_table时遇到Request Entity Too Large for url报错，说明提交的表体积超出了反向代理上限，需要提高 nginx 侧的proxy-body-size（见下文"故障排查"）。

评测指标是如何算出来的：源码级拆解

整体流程与数据集拆分

CUAD 评测的核心循环在 experiment.pyrun_cuad_eval与 evaluator.pyRAGEvaluator中：

1. CuadDataset.from_tsv读入labeled.tsv，prepare_dataset按 28 个问题类型把每一行展开为Data(file, question, label, reworded_question, question_category, ...)列表；
2. RAGEvaluator.apredict_dataset()用asyncio.gather对整份数据集并发调用/v2/answer（每个问题都带上create_file_filter(file)把检索范围限定到对应合同文件），收集response（回答）与context_docs（命中文档），写入predicted_dataset；
3. 用"-Answer"是否出现在question_category中把预测集拆成检索子集与回答子集（见CuadDatasetUtils.split_retrieval_answer_datasets）：非 Answer 型（检索型）用于文档召回评测，Answer 型用于最终答案正确性评测；
4. 检索子集经prepare_retrieval_dataset清洗——剔除标签为空/[]/nan的行，并把形如"['a', 'b', 'c']"的字符串化标签解析为字符串列表（额外拼接整句以便 top-k 命中），随后进入检索指标计算；
5. 回答子集经prepare_answer_dataset处理——空标签统一替换为CuadMissingData.ANSWER_LABEL = "no information found"（constants.py），表示"该问题从合同无法推断出答案"，再逐行做答案正确性判定并取均值作为总体准确率。

Retrieval metrics：命中判定与 MRR / Hit@k

evaluator.py 的_calculate_dataset_retrieval_metrics定义了如下判定：

• 对每条样本，取预测返回的docs文本列表与 ground-truth 标签（证据片段列表）做词集合交并比：len(set(label) ∩ set(pred)) / len(set(label))；
• 若某一 rank 位置的文档与标签的交并比 ≥STRDIFF_MIN_SIMILARITY（0.65），则记该位置为命中位次hit_k；get_hit_index假定每条 ground truth 为单一短语/文本。

基于命中位次列表聚合出五项指标（实现见get_mrr/get_hit_k）：

回答正确性判定：字符串、日期、语义与 LLM Judge 的组合

回答子集逐行调用BaseAnswerEvaluator.evaluate(pred, label) -> bool，仓库中提供了三种可插拔实现：

1. StringSimEvaluator：规则型兜底。先判断若预测含"no information"/"don't know"/"do not know"，仅当标签恰好是缺失占位（no information found或[]）时判对；若标签是MM/DD/YY格式日期则先做日期规范化后精确比较；否则对去除非字母数字字符后的字符串做SequenceMatcher相似度，阈值SEQUENCE_MATCH_MIN_THRESHOLD = 0.4；
2. BertScoreEvaluator：用evaluate.load("bertscore")计算预测与标签的 BERTScore F1，超过BERT_SIMILARITY_CUTOFF = 0.6判对；
3. LLMAnswerEvaluator（默认采用）：把pred与label交给一个结构化输出的 GPT-4o-mini 评判器（structured_llm.py 中基于OpenAIChat扩展的OpenAIStructuredChat，通过response_format返回 pydantic 模型AnswerCorrectness(is_correct: bool)）。评判器的 system 提示词明确放行"措辞、日期格式、详细程度不同但语义正确"的回答，例如参考值为01/01/2021、回答写成...date is 1st of January 2021.也判 True；同时规定若标签为空/nan/no information而系统回复 "I don't know." 也判 True。

experiment.py 对 CUAD 回答子集使用的是LLMAnswerEvaluator，逐行得到布尔值sim后求均值即Total RAG Accuracy。

RAGAS 评测

RAGAS 部分运行在合成数据集（AlphabetQA/AlphabetTables 全部样本）与 CUAD 的 Answer 型子集之上：

• ragas_utils.py 把每条预测组装成SingleTurnSample(user_input=问题, retrieved_contexts=命中文档, response=模型回答, reference=标准答案, reference_contexts=证据片段)，构造EvaluationDataset；
• 评估器 LLM 用LangchainLLMWrapper(ChatOpenAI(model="gpt-4o-mini", temperature=0.0))；
• 使用两类指标：
  • AnswerCorrectness：weights=[1.0, 0.0]、beta=1.5（略微偏向 recall），并在提示词中追加"允许答案比标准答案更冗长/更简洁，日期格式与详细程度不同也算对"的宽容规则；
  • Faithfulness：检验生成回答中的每个论断是否都能由检索到的上下文支撑。
• 各指标按样本求平均后，以{dataset}-{metric}-Ragas命名mlflow.log_metric，并把逐样本 RAGAS 明细表log_table存档。

典型故障排查

症状：App 一直回答 "I don't know"

按 README 的说明，这几乎总是某个数据摄入（ingestion）环节出问题的信号，而不是模型本身的问题。排查顺序建议如下：

1. 看日志定位失败模块；
2. 没有看到任何文件被解析：优先检查连接器（$sources）配置是否正确——凭证路径、object_id、文件通配符是否匹配；
3. 文件确实进入流水线后仍然答不上来，问题可能出在parser（解析）、splitter（切分）或 embedder（嵌入）上；
4. 为隔离问题，可以把 RAG 应用当作独立程序单独运行做手工排查：python integration_tests/rag_evals/debug_main.py（对应仓库中的 debug_main.py，它会读取同一份app.yaml在 8092 端口拉起服务，便于用 curl/客户端逐条验证索引与回答）。

另外从判定逻辑反推一个易混淆点：若文档里确实找不到答案，模型回答 "I don't know" 本身是正确行为且会被评判器判对；只有文档明明已入库却仍检索不到、导致模型无上下文可用时，才是需要排查的摄入问题。

症状：MLFlow 报 "Request Entity Too Large for url"

当 RAGAS/预测明细表等大体积内容通过mlflow.log_table上传时，若体量超过反向代理限制会出现该错误（experiment.py 中在该调用旁也留有对应注释）。解决方法是提高 nginx 配置中的proxy-body-size，使代理允许转发更大的请求体。

把评测跑成你自己的调参循环

integration_tests/rag_evals/从设计上就是一个可复用的 RAG 评测基座，README 亦明确指出当前app.yaml的取值是"合理的默认值"，鼓励针对自己的场景调参寻找最优配置。建议的自定义路径为：

1. 换数据：把 dataset/labeled.tsv 换成你自己的"文件—问题—标准答案"三元组（或自备合成 QA JSONL），并保证待检索文档可被$sources连接器访问；
2. 换配置：调整 app.yaml 中的切分窗口（TokenCountSplitter的 min/max tokens）、search_topk、检索索引组合、提示词模板、LLM 模型与温度等，每次运行都会作为独立 MLFlow Run 留下app.yamlartifact 与全套指标，便于横向对比；
3. 当门槛：把 CI 下的MIN_ACCURACY与分支绑定，让每次合入前自动回归，任何让检索或回答质量回退的改动都会在指标上显形——这正是本目录作为集成测试存在的意义。

通过"检索指标 + 回答正确性 + RAGAS 三类指标 × 真实合同与合成文档两类数据 × MLFlow 全量留痕"的组合，该框架为 Pathway 上构建的 RAG 应用提供了一条可量化、可复现、可持续迭代的质量保障闭环，你可以直接复用它来验证自己的 RAG 管线与调参效果。

【免费下载链接】pathwayPython ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.项目地址: https://gitcode.com/GitHub_Trending/pa/pathway