LLM 评测体系
第 11 章 ragas：RAG 评测的工程范式
ragas 把 RAG 评测收束成
Metric与SingleTurnSample两个抽象。Faithfulness 的两阶段实现——先把答案拆成陈述、再逐条对上下文做 NLI 判定——是理解这套范式的最短路径。
本章要点
- ragas 的核心抽象
Metric与SingleTurnSample/MultiTurnSample数据 schema Faithfulness指标的两阶段实现：LongFormAnswerPrompt + NLIStatementPromptAnswer Relevance的"反向生成 question"奇技Context Recall / Context Precision在 retriever 评测中的角色分工- 36 个内置 metric 类全景，以及 RAG 评测之外的延伸（Agent / SQL / Tool Call）
11.1 仓库一瞥
本章源码引用统一基于 vibrantlabsai/ragas（原 explodinggradients/ragas）的 v0.2.10 tag（本专栏的基准版本）。读者可通过下面命令获取：
git clone https://github.com/vibrantlabsai/ragas.git
cd ragas
git checkout v0.2.10如果你同时在读《RAG 工程与检索系统设计》第 20 章，会看到相反的导入建议，这里先说清楚：那一栏锚的是 ragas v0.4.3，而 ragas.metrics 与 ragas.metrics.collections 这两套 API 正是在 0.2 → 0.4 之间分叉的——v0.2.10 里根本没有 metrics/collections 这个子包，从 ragas.metrics 导入不会有任何 DeprecationWarning，本章后面所有示例都据此写成；到 v0.4.3，旧路径仍能用但开始告警、并预告 v1.0 移除。按你手上装的 ragas 版本选：0.2.x 就照本章写，0.4.x 请读那一栏的迁移说明（collections 下的同名类是另一套单条打分接口，不是改一行 import 就能换）。
仓库整体规模适中——src/ragas/ 下 25 个顶层模块 / 子包，metrics/ 子目录里 __all__ 导出 36 个 metric 类。这个 36 的计数口径要先说清楚，否则你自己数会对不上：src/ragas/metrics/__init__.py 的 __all__ 共 56 项，其中 46 项是大写开头的类名（另外 10 项是 faithfulness / context_recall 这类预实例化的小写单例）；46 个类名里 LLMContextPrecisionWithoutReference 与 LLMSQLEquivalence 各被写了两遍，去重后 44 个；再减去 Metric / MetricType / MetricOutputType / MetricWithLLM / MetricWithEmbeddings / SingleTurnMetric / MultiTurnMetric / DistanceMeasure 这 8 个基础设施类型，剩下的正好 36 个真正的 metric 类。
src/ragas/
├── dataset_schema.py # SingleTurnSample / MultiTurnSample 813 行
├── evaluation.py # 顶层 evaluate() 入口
├── llms/ # LLM 适配器
├── embeddings/ # embedding 适配器
├── metrics/ # 36 个内置 metric 类
│ ├── base.py 715 行
│ ├── _faithfulness.py 322 行
│ ├── _answer_relevance.py 158 行
│ ├── _context_recall.py 240 行
│ ├── _context_precision.py
│ ├── _answer_correctness.py
│ ├── _factual_correctness.py
│ ├── _noise_sensitivity.py
│ ├── _summarization.py
│ ├── _topic_adherence.py
│ ├── _tool_call_accuracy.py # 不只 RAG, 已扩展到 Agent
│ ├── _sql_semantic_equivalence.py # SQL 评测
│ └── ...
└── prompt/ # 提示词系统ragas 的工程定位极清晰：专门为 RAG / Agent 系统的 LLM-as-Judge 评测设计。它不试图替代 lm-eval 做学术 benchmark，也不想做 promptfoo 的 YAML-first 配置，而是把"用 LLM 评测 LLM 系统"这一专门方向做到极致。
11.2 核心抽象：Metric × Sample
flowchart LR Sample["SingleTurnSample<br/>(user_input,<br/>response,<br/>retrieved_contexts,<br/>reference)"] --> Metric[Metric<br/>对单条样本算分] Metric -->|内部| LLM[调用 LLM-judge] Metric -->|或| Embed[调用 embedding 模型] LLM --> Score[float / list] Embed --> Score Score --> Agg[evaluate 聚合] style Sample fill:#dbeafe style Metric fill:#dcfce7 style Agg fill:#fef3c7
SingleTurnSample（dataset_schema.py）是 ragas 数据的最小单元，包含：
user_input：用户问题response：系统回答retrieved_contexts：RAG 检索到的 context（list of str）reference：黄金答案（可选）multi_responses：多候选回答（用于 ensemble）
Metric 是抽象基类（src/ragas/metrics/base.py:73），具体指标再按需叠加两个同样派生自它的基类：
- 单轮指标继承
SingleTurnMetric（src/ragas/metrics/base.py:402），必须实现其抽象方法_single_turn_ascore(sample, callbacks) -> float； - 要调 LLM 的指标同时继承
MetricWithLLM（src/ragas/metrics/base.py:215）。
这套抽象的核心简洁性：每个 metric 只关心"一条样本算出一个分"，聚合由顶层 evaluate() 完成。
11.3 Faithfulness：两阶段 NLI 实现
ragas 的旗舰 metric。来自 ragas 论文（Es et al. 2023, arXiv:2309.15217）。源码在 src/ragas/metrics/_faithfulness.py:165：
@dataclass
class Faithfulness(MetricWithLLM, SingleTurnMetric):
name: str = "faithfulness"
_required_columns: t.Dict[MetricType, t.Set[str]] = field(
default_factory=lambda: {
MetricType.SINGLE_TURN: {
"user_input",
"response",
"retrieved_contexts",
}
}
)
output_type: t.Optional[MetricOutputType] = MetricOutputType.CONTINUOUS
nli_statements_message: PydanticPrompt = field(default_factory=NLIStatementPrompt)
statement_prompt: PydanticPrompt = field(default_factory=LongFormAnswerPrompt)值得注意的工程细节：
_required_columns：明确声明这个 metric 需要哪些字段。如果 sample 缺字段，框架会在 evaluate 阶段提前报错，而不是运行时 KeyError- 两个 PydanticPrompt 字段：
statement_prompt（LongFormAnswerPrompt）+nli_statements_message（NLIStatementPrompt）——揭示了 Faithfulness 的两阶段实现 output_type = CONTINUOUS：声明输出是连续值（0-1）。v0.2.10 里它只在MetricWithLLM.train()做 prompt 优化时用来挑 loss（在train()调用的_optimize_instruction里，src/ragas/metrics/base.py:255-269），evaluate()的聚合不看它，一律nanmean
11.3.1 第一阶段：陈述拆解
LongFormAnswerPrompt（src/ragas/metrics/_faithfulness.py:82）的 instruction（行 83）：
"Given a question, an answer, and sentences from the answer analyze the complexity of each sentence given under 'sentences' and break down each sentence into one or more fully understandable statements while also ensuring no pronouns are used in each statement. Format the outputs in JSON."
特别注意"no pronouns"那一句——这是个工程上极重要的约束。如果 statements 含代词（"He was a physicist"），后续 NLI 判定时 judge 会因代词指代不明而困惑。强制消除代词把每条 statement 变成自包含的命题。
11.3.2 第二阶段：NLI 判定
NLIStatementPrompt（src/ragas/metrics/_faithfulness.py:104）的 instruction（行 105）：
"Your task is to judge the faithfulness of a series of statements based on a given context. For each statement you must return verdict as 1 if the statement can be directly inferred based on the context or 0 if the statement can not be directly inferred based on the context."
每条 statement 输出 (statement, reason, verdict) 三元组（src/ragas/metrics/_faithfulness.py:89-92 的 StatementFaithfulnessAnswer）：
class StatementFaithfulnessAnswer(BaseModel):
statement: str = Field(..., description="the original statement, word-by-word")
reason: str = Field(..., description="the reason of the verdict")
verdict: int = Field(..., description="the verdict(0/1) of the faithfulness.")reason 字段是工程上的关键设计——它强制 judge 给出"为什么是 1 / 为什么是 0"的解释。这是第 6 章 §6.4 讨论的 CoT prompting 落到 ragas 里的具体形态。它同时让判错的样例可以被人工快速复核——这一点比任何"提升多少 pp"的说法都更实在。
11.3.3 聚合公式
_compute_score（src/ragas/metrics/_faithfulness.py:220）的全部代码（略去一行注释）：
def _compute_score(self, answers: NLIStatementOutput):
faithful_statements = sum(
1 if answer.verdict else 0 for answer in answers.statements
)
num_statements = len(answers.statements)
if num_statements:
score = faithful_statements / num_statements
else:
logger.warning("No statements were generated from the answer.")
score = np.nan
return score简洁到 10 行——这就是第 4 章 §4.3.1 的 Faithfulness 公式。注意 np.nan 的处理：当 answer 拆不出任何 statement（极短回答）时不算 0，而是 NaN，避免污染聚合分数。
11.3.4 性能优化版：FaithfulnesswithHHEM
src/ragas/metrics/_faithfulness.py:259 还有一个变体类 FaithfulnesswithHHEM（name = "faithfulness_with_hhem"，默认 device="cpu" / batch_size=10），它把 NLI 判定从 LLM 改成 HuggingFace 上的专用 NLI 模型（vectara/hallucination_evaluation_model）：
self.nli_classifier = AutoModelForSequenceClassification.from_pretrained(
"vectara/hallucination_evaluation_model", trust_remote_code=True
)这个变体的工程意义：LLM 版的 NLI 判定是把全部 statements 打包成一次 LLM 请求（_create_verdicts，src/ragas/metrics/_faithfulness.py:187-199），HHEM 版把这一步换成本地 encoder 按 batch_size 分批前向（src/ragas/metrics/_faithfulness.py:310-319），省掉的是第二次 LLM 调用；第一阶段的陈述拆解仍走 _create_statements，照样要调 LLM（src/ragas/metrics/_faithfulness.py:301）。代价是判定质量取决于这个小模型本身，未必比得上 GPT-4 级 judge。具体省多少、掉多少分，必须在你自己的样本上做一次对照实验——不要照搬任何现成数字。这是工业评测的典型 trade-off——大规模场景用专用模型省钱，敏感场景用 LLM-judge 求精。
11.4 Answer Relevance：反向生成 question 的奇技
src/ragas/metrics/_answer_relevance.py:36 的 ResponseRelevancePrompt instruction（行 39）：
"Generate a question for the given answer and Identify if answer is noncommittal. Give noncommittal as 1 if the answer is noncommittal and 0 if the answer is committal. A noncommittal answer is one that is evasive, vague, or ambiguous. For example, "I don't know" or "I'm not sure" are noncommittal answers"
它的工作原理（第 4 章 §4.3.2 已介绍过，这里看代码细节）：
- 让 LLM 看着
response，反向生成一个可能的question——并发生成strictness份，默认 3（src/ragas/metrics/_answer_relevance.py:95、:140-148） - 把生成的 question 与原始 user_input 算 embedding 余弦相似度，取平均（
src/ragas/metrics/_answer_relevance.py:125-126） - 同时输出
noncommittal标志：任何一份被判为回避，整题分数乘 0（:118、:126）
例子（examples，行 42-61）：
response: "Albert Einstein was born in Germany."
→ generated question: "Where was Albert Einstein born?"
→ 和原始 question 算 cosine 相似度这种"反向生成"的玄妙在于：如果 answer 真的回答了 Q，那 LLM 反推出来的 candidate question 应该和 Q 高度相似。如果 answer 答非所问，反推出来的 question 就会和 Q 偏离。
为什么要单独引入 noncommittal？ 因为对话评测里"我不知道"是一种特殊的"答非所问"——它在语义上和原 Q 接近（同样关于"Einstein 哪里出生"），embedding 相似度可能还不低，但实际上对用户毫无价值。noncommittal 标志专门捕捉这种情况，对应分数直接归 0。
11.5 Context Recall / Context Precision：retriever 的两面镜子
_context_recall.py（240 行）专评 retriever 是否拿全了所需信息。它的核心逻辑：
- 把 reference（黄金答案）和拼接后的 retrieved_contexts 一起交给
ContextRecallClassificationPrompt，一次 LLM 调用里逐句分析 reference（不像 Faithfulness 那样先单独拆陈述） - 对每一句判定"能否归因于 retrieved_contexts"（
attributed取 1/0） - Recall = (可归因的句子数) / (全部句子数)（
src/ragas/metrics/_context_recall.py:117-126）
Context Precision 反过来，对每条 retrieved context 单独发一次请求，判它"对得出 reference 答案有没有用"（ContextPrecisionPrompt，src/ragas/metrics/_context_precision.py:40-43，输入是 question、context 与 reference），得到按检索顺序排列的 0/1 序列 v₁…v_K。但分数不是"有用条数 / 总条数"的朴素比例，而是按排名加权的平均精度（_calculate_average_precision，src/ragas/metrics/_context_precision.py:116-134）：
所以有用的 chunk 排得越靠前分越高：同样是 4 条里 2 条有用，[1,1,0,0] 得 1.0、[1,0,1,0] 得 0.833、[0,1,0,1] 得 0.5、[0,0,1,1] 得 0.417，朴素比例则一律是 0.5（在 ragas 0.2.10 上用桩 LLM 实跑得到这四个值）。这一口径《RAG 工程与检索系统设计》第 20 章锚定的 v0.4.3 也沿用；它衡量的是排序质量，不只是"噪声占比"。
flowchart TB
subgraph 输入
Q[user_input]
R[reference answer]
C[retrieved_contexts]
end
subgraph Recall 路径
R --> S[拆解成 atomic statements]
S --> NLI1[每条与 contexts 做 NLI]
NLI1 --> Recall[Recall = 能命中 / 全部]
end
subgraph Precision 路径
Q --> NLI2[每条 context 逐条判定<br/>对得出 reference 是否有用]
C --> NLI2
R --> NLI2
NLI2 --> Precision[Precision = 按排名加权的平均精度]
end
style Recall fill:#dbeafe
style Precision fill:#dcfce7
两个指标缺一不可：
- 只看 Recall → retriever 可以无脑返回 100 个 chunk 求覆盖，Precision 烂到不行但 Recall 高
- 只看 Precision → retriever 可以只返回 1 个最相关的 chunk，Precision 满分但漏掉关键信息
阈值没有通用标准，要在自家标注集上校准（Context Recall 更看重不漏，门槛通常比 Context Precision 定得高）。第 13 章 RAG 评测会详述 retriever 调参实战。
11.6 内置 Metric 全景
ragas 的 metric 库远不止 Faithfulness / Recall。v0.2.10 的 ragas.metrics.__all__ 导出 36 个 metric 类（下表按类别归拢，类名逐字取自 src/ragas/metrics/__init__.py）：
| 类别 | Metric | 用途 |
|---|---|---|
| RAG 核心 | Faithfulness、FaithfulnesswithHHEM、ResponseRelevancy（旧名 AnswerRelevancy）、ContextRecall、ContextPrecision、ContextUtilization、AnswerCorrectness |
第 4、13 章主用 |
| Context 变体 | LLMContextPrecisionWithReference、LLMContextPrecisionWithoutReference、NonLLMContextPrecisionWithReference、LLMContextRecall、NonLLMContextRecall、ContextEntityRecall |
有无 reference / 用不用 LLM 的四象限 |
| 事实性 | FactualCorrectness、NoiseSensitivity |
抗噪与事实一致性 |
| 摘要 | SummarizationScore |
内容摘要质量 |
| 评判 / rubric | AspectCritic、SimpleCriteriaScore、RubricsScore、InstanceRubrics |
自定义判据 |
| 主题 | TopicAdherenceScore |
是否偏题 |
| Agent | ToolCallAccuracy、AgentGoalAccuracyWithReference、AgentGoalAccuracyWithoutReference |
Agent 评测，详见第 14 章 |
| SQL | LLMSQLEquivalence |
NL2SQL 应用 |
| 多模态 | MultiModalFaithfulness、MultiModalRelevance |
图文混合 RAG |
| 字符串 / 传统 NLP | ExactMatch、StringPresence、NonLLMStringSimilarity、SemanticSimilarity（旧名 AnswerSimilarity）、BleuScore、RougeScore |
无需 LLM 的快速判分 |
| 数据对比 | DataCompyScore |
表格 / 数据集对比 |
注意几个易记错的名字：SQL 那个叫 LLMSQLEquivalence（文件是 _sql_semantic_equivalence.py）；摘要是 SummarizationScore 而不是 Summarization；Agent 目标达成度分成 WithReference / WithoutReference 两个类；ragas 0.2 里没有 ToolCallF1，也没有 ChrFScore。
这套丰富 metric 体系的设计哲学：评测一个 RAG/Agent 系统不是"一个分数"的事，而是十几个独立维度的工程报告。工业实战中通常选 5-8 个关键 metric 同时跑，互相验证。
11.7 ragas 的工程亮点
graph TB A[ragas 工程亮点] --> B[PydanticPrompt 系统<br/>类型安全的 prompt 模板] A --> C[Async 并发判分<br/>大数据集快速跑完] A --> D[multi_responses 字段<br/>为多候选回答预留] A --> E[MetricOutputType 输出类型<br/>prompt 优化时选 loss] A --> F[NaN 处理一致性<br/>避免极端样本污染聚合] style A fill:#fef3c7
每个亮点对应一个具体设计模式：
- PydanticPrompt：每个 prompt 都有
input_model/output_model两个 Pydantic 类——LLM 输出强制结构化解析，避免 LLM-judge 的"格式漂移"问题 - Async 并发：
_single_turn_ascore是 async 的，evaluate 时可以并发跑几十条 LLM 调用 - multi_responses：
SingleTurnSample预留了多候选回答字段（src/ragas/dataset_schema.py:83）；不过 v0.2.10 的内置 metric 都还不读它，要做 best-of-K 得自己写 metric。内置 metric 的"多次采样取多数"走的是另一条路：prompt 的generate_multiple+ensembler.from_discrete（如src/ragas/metrics/_context_precision.py:153-169） - MetricOutputType：
MetricOutputType枚举（src/ragas/metrics/base.py:65-69）定义了 BINARY / DISCRETE / CONTINUOUS / RANKING 四种输出，train()做 prompt 优化时（具体在_optimize_instruction里）据此选 loss（BINARY 用BinaryMetricLoss，CONTINUOUS / DISCRETE 用MSELoss，RANKING 尚未实现，src/ragas/metrics/base.py:255-269） - NaN 处理：极短回答 / 拆不出 statement / API 失败时返回 NaN 而非 0，最终聚合时
safe_nanmean跳过（src/ragas/dataset_schema.py:402、src/ragas/utils.py:42-51）
这五个设计是 ragas 经过近 2 年迭代（2023-05 建仓到 2025-01 的 v0.2.10）沉淀的工程财富，是任何评测框架二开时都该借鉴的模式。
11.8 ragas 的设计取向
11.8.1 ragas 的演化：从 RAG 专用到多场景框架
ragas 仓库 2023 年 5 月建立，最早的 0.0.1 用的还是 ROUGE / BERTScore 这类传统指标；Faithfulness、Answer Relevance、Context Recall、Context Precision 这四个 RAG 指标是 2023 年 6-9 月间陆续加入的（Context Precision 由 context_relevancy 改名而来，2023-09），定位极其垂直："RAG 评测专用框架"。但到本章锚定的 v0.2.10 已有 36 个 metric 类，覆盖 Agent / SQL / 多模态 / 摘要 / 主题一致性等多个场景。
这种演化轨迹反映了一个工程现实：LLM 应用的边界在快速模糊。2023 年大家说"我做 RAG"或"我做 chatbot"——它们是不同应用。但到 2026 年，几乎所有 RAG 系统都在加 Agent 能力（Tool Calling）、几乎所有 chatbot 都接了 RAG。评测工具如果只能服务一种应用，会被市场抛在后面。
ragas 的应对策略是把核心抽象（Metric × Sample）保持不变，新场景靠新 sample 类型扩展：
SingleTurnSample→ 早期 RAG / 单轮问答MultiTurnSample→ 多轮对话 / Agent 工作流- 未来可能出现
MultiModalSample、ToolGraphSample等
这种"扩展点设计"是软件工程的经典——把可变的部分（Sample 类型）和不变的部分（Metric 接口）分开，让框架能演化而不破坏现有 API。第 9 章 OpenAI evals 的 Eval / CompletionFn / Recorder 三抽象走的也是同一思路。
11.8.2 Pydantic-First 的设计趋势
ragas、promptfoo（用 Zod）、langsmith（用 TypeScript types）——这一代评测框架的共同特征是类型系统驱动 prompt 工程。
ragas 的 PydanticPrompt 抽象（贯穿 _faithfulness.py、_answer_relevance.py、_context_recall.py 等）是这个趋势的代表：每个 LLM-judge prompt 都强制定义 input_model 和 output_model 两个 Pydantic 类。LLM 输出会走 Pydantic 解析——格式不对直接报错而非静默错误。
这种设计的工程价值：
- JSON 模式输出受控：output_model 的 JSON Schema 会被写进 prompt（
src/ragas/prompt/pydantic_prompt.py:43-47），解析失败还有修复重试；若再搭配 OpenAI Structured Outputs / Anthropic Tool Use 这类约束解码，格式还能收得更紧 - 类型安全的 prompt 编程：IDE 能在写 prompt 时给提示
- 自动文档化：Pydantic 类的 Field description 自动成为 prompt 的字段说明
- 判分逻辑可重构：output_model 改字段、所有依赖代码自动错——避免"prompt 改了但解析忘改"的隐蔽 bug
这个趋势在 2024-2026 年快速主流化。任何 2024 年之后的新评测框架，如果还在用 text → text 的字符串接口，已经落后了。
11.8.3 ragas 的 Pydantic schema 设计模式
读完 §11.3 的 Faithfulness 源码后，能看到 ragas 用了一套有意思的 Pydantic 设计模式：
# Input model - 喂给 prompt 的东西（src/ragas/metrics/_faithfulness.py:31）
class FaithfulnessStatements(BaseModel):
question: str = Field(description="The question to answer")
answer: str = Field(description="The answer to the question")
sentences: t.Dict[int, str] = Field(
description="A mapping of sentence index to the sentence"
)
# Output model - prompt 期待的输出（src/ragas/metrics/_faithfulness.py:46）
class SentencesSimplified(BaseModel):
sentences: t.List[SentenceComponents] = Field(
description="A list of sentences and their simpler versions"
)
# Prompt class 把两者绑定（src/ragas/metrics/_faithfulness.py:82）
class LongFormAnswerPrompt(
PydanticPrompt[FaithfulnessStatements, SentencesSimplified]
):
instruction = "..."
input_model = FaithfulnessStatements
output_model = SentencesSimplified
examples = [(example_input_1, example_output_1)]这种设计模式有几个工程红利：
- 类型安全：调用 prompt 时编辑器能自动 hint 输入字段
- 自动 schema 注入：PydanticPrompt 把 output_model 自动转成 JSON Schema 注入 prompt（让 LLM 输出严格符合）
- examples 强类型：example 输入输出也是 Pydantic 对象，避免随手写错
- 跨模型兼容：同一份 PydanticPrompt 在 GPT、Claude、Gemini 上都能用——schema 是中性的
工业团队 fork ragas 改造时，最关键的是保留这套模式。直接写 string prompt 的诱惑很大（"5 行就能跑"），但 6 个月后 prompt 演化到 50 处 string interpolation 时维护成本爆炸。
这套"Input model + Output model + PydanticPrompt"是 LLM-judge 工程化的最佳实践之一——本章 §11.8.2 提到的 Pydantic-First 趋势在这里有最完整的体现。
11.8.4 ragas 的 prompt 工程视角：把 LLM-judge 当作可调代码
ragas 的 PydanticPrompt 抽象给读者一个深层启示——LLM-judge 的 prompt 不是文学作品，是软件工程模块。它需要：
- 类型签名：input_model / output_model 强类型
- 测试覆盖：每个 judge prompt 自己也要跑元评测验证
- 版本管理：prompt 进 git，每次改动有 diff
- 依赖管理：prompt 中引用的 examples / rubrics 是独立的"prompt fragment"
这种"prompt as code"的视角让 LLM-judge 可以像普通代码那样被治理。具体实践：
# bad: prompt 写死在判分函数里
def judge(question, answer, context):
prompt = f"""
Given context: {context}
Answer: {answer}
Is it faithful? Output 1 or 0.
"""
return call_llm(prompt)
# good: prompt 是结构化对象
class FaithfulnessPrompt(PydanticPrompt[FaithfulnessInput, FaithfulnessOutput]):
instruction = "Given context and answer, judge if answer is faithful to context."
examples = [...] # 显式 few-shot
output_model = FaithfulnessOutput # 强类型输出ragas 的所有 metric 都用这种"good"风格——这就是它能成为 RAG 评测主流的工程基础。fork 改造时这种风格也要保留。
11.8.5 ragas 给 LLM 评测领域的两个范式贡献
回顾 ragas 项目的整体贡献，有两个范式级别的启发值得专门讨论：
范式 1：把"判分"当作软件工程问题
ragas 之前的 LLM-judge 主要靠 prompt 字符串拼接 + 自由文本输出 + 后处理解析。ragas 用 PydanticPrompt 强制每个 judge 都有：
- 类型化 input / output
- 显式 schema
- 测试覆盖
- 版本管理
把"prompt 工程"从手工艺升级到软件工程——这个范式贡献现在已被 LangChain / promptfoo / langsmith 等工具普遍采纳。
范式 2：把 RAG 评测的术语标准化
Faithfulness / Answer Relevance 出自 ragas 论文（论文第三个指标是 Context Relevance），Context Recall / Context Precision 则是 ragas 库在同一时期补上的。在 ragas 之前 RAG 评测的术语极度混乱，每篇论文 / 每家工具用不同名字说同一件事。
ragas 论文（arXiv:2309.15217）的最大贡献是给行业一套共同语言。现在工程师之间说"我们的 Faithfulness 是 0.85"，对方立即理解什么意思——这种"概念基础设施"的价值远超工具本身。
这两个范式贡献比工具本身更值得带走。
11.9 与同类工具的对照
11.9.1 ragas 与 deepeval / TruLens：评测库三国杀
ragas 不是唯一的"Python 评测库"。它和另外两个流派形成现代 RAG/Agent 评测的"三国杀"：
| 维度 | ragas | deepeval | TruLens |
|---|---|---|---|
| 主创团队 | Exploding Gradients | Confident AI | TruEra（被 Snowflake 收购） |
| 协议 | Apache 2.0 | Apache 2.0 | MIT |
| 集成深度 | LangChain / LlamaIndex / Haystack | pytest-style | TruLens-Eval |
| 默认 metric 风格 | NLI 拆解 + LLM-judge | LLM-judge + 多维 rubric | Feedback Function |
| 适合场景 | RAG 主导 | 单元测试风格 | 与 Snowflake 数据栈集成 |
三家在 metric 名字上有意思的趋同——Faithfulness / Answer Relevance 出自 ragas 论文（arXiv:2309.15217），Context Recall / Context Precision 出自 ragas 库，这套词汇已被同类工具普遍沿用。这是开源生态里"先驱定义术语"的典型现象。
工程团队选哪家：
- 要 RAG 评测最好 → ragas
- 要 pytest 风格、单测无缝集成 → deepeval（其
assert_test()风格） - 已经在 Snowflake 数据栈 → TruLens
但建议只选一家深用——因为 Faithfulness 在三家的具体 prompt 实现略有不同，分数互不可比。混用会让"两家都跑、对比谁的指标更好"变成无意义的练习。
11.9.2 ragas 与 DeepEval / TruLens / Phoenix Evals 的对照
ragas 不是 RAG 评测工具的唯一选择。下表对照 4 个主流 RAG 评测库（依据各仓库 README，定性对比；star 数随时间变化，请以仓库页面为准）：
| 维度 | ragas | DeepEval | TruLens | Phoenix Evals |
|---|---|---|---|---|
| 起源团队 | Exploding Gradients（Shahul ES / Jithin James） | Confident AI | TruEra（已被 Snowflake 收购） | Arize AI |
| 主打理念 | LLM-judge 自动化 + paper-grounded | 测试框架（pytest 风格） | trace-first + 内嵌仪表板 | OpenInference 标准 |
| 核心抽象 | Metric × Sample | Test Case × Metric | Feedback Function | Span Eval |
| Faithfulness | ✅ 论文同名实现 | ✅ G-Eval 通用 | ✅ Groundedness | ✅ Hallucination |
| 自定义 metric 难度 | 低（继承 Metric class） | 低（@metric 装饰器） | 中（feedback 注册） | 中（OTel 心智） |
| 与 LangChain | 强 | 强 | 强 | 中（OTel 优先） |
| 与 LlamaIndex | 强 | 强 | 强 | 强 |
| 与 trace 平台 | 通过 LangSmith / Langfuse 桥接 | 自建 dashboard | 自带 UI | 自建 Phoenix UI |
| 适合场景 | 离线大规模评测 | TDD 风格的回归集 | 已用 Snowflake 的团队 | 已 OTel 化的栈 |
quadrantChart title RAG 评测库选型坐标 x-axis "学习曲线（陡 → 平）" y-axis "独立性（强绑定 → 中立）" quadrant-1 "独立 + 易学" quadrant-2 "独立 + 难学" quadrant-3 "绑定 + 难学" quadrant-4 "绑定 + 易学" ragas: [0.75, 0.85] DeepEval: [0.85, 0.55] TruLens: [0.50, 0.30] Phoenix: [0.55, 0.65]
工业实务的 4 个选型路径：
- 离线大规模评测 + 学术对齐 → ragas（每个指标对应一篇论文，引用方便）
- CI 中跑 RAG 回归集 + pytest 风格 → DeepEval（assert 语法最熟悉）
- 已用 Snowflake → TruLens（同生态）
- 已 OTel + Phoenix → Phoenix Evals（零额外栈）
读者可以把这张表理解为："本章用 ragas 讲透了 RAG 评测的核心抽象，但生产选型未必非 ragas 不可"——理解了 ragas 的 Metric × Sample 模型，再上手同类库会快得多。
11.9.3 ragas 与 LangChain Evaluators 的设计差异
LangChain 自身也提供 evaluator 模块（langchain.evaluation），与 ragas 在功能上有重叠。理解两家差异能避免选型踩坑：
| 维度 | ragas | LangChain Evaluators |
|---|---|---|
| 主要目标 | RAG 评测专用 | 通用 LangChain 工作流评测 |
| Metric 类型 | v0.2.10 的 ragas.metrics.__all__ 导出 36 个 metric 类 |
通过 StringEvaluator / PairwiseStringEvaluator 自定义 |
| Prompt 模板 | PydanticPrompt 强类型 | 字符串模板 |
| 主要受众 | RAG 工程师 | LangChain 用户 |
| 默认是否带 reference | 否（很多 metric 不需要） | 大部分需要 |
LangChain Evaluators 的核心抽象是 "evaluator + criteria"：
from langchain.evaluation import load_evaluator
evaluator = load_evaluator("criteria", criteria="conciseness")
result = evaluator.evaluate_strings(prediction="...", input="...")它的 criteria 列表包括 conciseness / relevance / correctness / coherence 等通用维度，偏通用质量；RAG 专项的拆陈述、逐条归因这类判分流程要靠 ragas 这类专用库。两者在你数据上谁更稳，只能用第 8 章的元评测方法自己测。
工程团队的实操：
- 要 RAG 专用 metric → ragas（无替代）
- 要快速接入 LangSmith dashboard → LangChain Evaluators（与 LangSmith 一体化更紧）
- 混用：用 LangChain Evaluators 做"通用质量"打分（conciseness / coherence），用 ragas 做"RAG 专项"
这种"在 LangChain 工作流里嵌套 ragas"是最常见的混合用法——LangChain 当主流程，ragas 当 RAG 评测插件。
11.9.4 ragas 的局限：什么时候它不够用
诚实告诉读者 ragas 的工程边界：
- 不适合超长 context 评测：context 极长时每次 judge 调用都要塞进整段 context，token 成本和延迟随之线性上涨；response 很长时拆出的 statement 也会很多
- 多模态要谨慎：虽然有
MultiModalFaithfulness，但图文混合的判定可靠性要先在自己的样本上验证，不能默认与纯文本相当 - 不适合实时低延迟评测：单个 metric 每条样例就要 1-3 次串行或并发的 LLM 调用（见 §11.13.1），延迟是秒级以上，难以做同步的实时质量门禁
- 不适合工具调用场景的复杂度：
ToolCallAccuracy（src/ragas/metrics/_tool_call_accuracy.py:17）只做两件事——用is_sequence_aligned判调用顺序是否与reference_tool_calls对齐，再用可插拔的arg_comparison_metric（默认ExactMatch）逐参数比对；它假设工具调用是一条线性序列，复杂 tool graph（DAG 调用、并发分支）要自己写 metric
工程团队遇到这些场景时：
- 长 context → 用滑动窗口 + 抽样判分
- 多模态 → 用专门的 vision-language 评测方案
- 实时低延迟 → 改用规则判分 + Schema validation（第 5 章）
- 复杂 tool graph → 自己实现
MetricWithLLM子类
ragas 的开放扩展架构（继承 MultiTurnMetric / SingleTurnMetric 自定义指标）是它在边界场景仍能用的根本——你不必丢掉框架，只补一个自定义类。
11.9.5 ragas 性能基准（公开数据视角）
网上流传的"ragas 跑 N 题要多少钱多少分钟"表格几乎都没有出处，本专栏不复制。成本和耗时可以自己算准，不需要别人的经验数字——ragas 的开销完全由三个可数的量决定：
总 LLM 调用次数 = 样本数 × Σ(每个 metric 的调用次数)
总成本 = Σ(每次调用的 input tokens × 单价in + output tokens × 单价out)
墙钟耗时 = 总调用次数 ÷ 有效并发（受 provider rate limit 约束）× 单次延迟其中"每个 metric 的调用次数"在 §11.13.1 的表里逐个列了（Faithfulness 2 次、ContextPrecision 随 context 条数线性增长……），单价查你的 provider 定价页，rate limit 查你的账户配额。把这三样代进去，任何规模的预算都是一道算术题。
真要落地，最可靠的做法是先跑一次 20 题的试验批，用 provider 的用量面板读出真实 token 数与耗时，再线性外推——这比抄任何现成表格都准，因为它吃进了你自己的 context 长度、回答长度和账户配额。
外推之后再定频率，常见的档位是：
- 小规模（< 200 题）：成本通常可忽略，每天跑都不肉疼
- 中等规模（500-1000 题）：每周跑 1-2 次合适
- 大规模（5000+）：每月跑 1 次或全集采样跑
这种"按规模匹配频率"是 ragas 投入的工程务实：
- 离线评测：500-1000 题 × 周度
- 在线 1% 采样：每天约 1000 条
- 元评测 calibration：200 题 × 季度
工业团队的实务：根据自家 trace 量算预算，确定能负担多大规模 × 多高频率。这种"计算驱动"的预算决策让 ragas 投入有清晰的 ROI 边界。
11.10 集成：LangChain 生态与可观测性
11.10.1 ragas 与 LangSmith / LangChain 的集成模式
ragas 不是独立运行的——大部分实战使用都是嵌入到 LangChain / LangSmith / LlamaIndex 的工作流里。看一份典型的"LangChain RAG + ragas 评测 + LangSmith trace"端到端集成：
from langsmith import Client
from langchain_core.tracers.context import collect_runs
from langchain_openai import ChatOpenAI
from ragas import evaluate
from ragas.metrics import Faithfulness, ContextRecall, AnswerRelevancy
from datasets import Dataset
# 1. 从 LangSmith 拉一个 dataset
client = Client()
dataset = client.read_dataset(dataset_name="customer-support-v1")
# 2. 跑 LangChain RAG, 收集 (question, answer, contexts)
# 注意：run_id 要自己记，ragas 的 result 里没有这一列
rows, run_ids = [], []
for example in client.list_examples(dataset_id=dataset.id):
question = example.inputs["question"]
with collect_runs() as cb: # langchain_core.tracers.context
chain_result = my_rag_chain.invoke({"question": question})
run_ids.append(cb.traced_runs[0].id)
rows.append({
"user_input": question,
"response": chain_result["answer"],
"retrieved_contexts": chain_result["contexts"],
"reference": example.outputs.get("answer", ""),
})
# 3. ragas 评测
ragas_dataset = Dataset.from_list(rows)
result = evaluate(
dataset=ragas_dataset,
metrics=[Faithfulness(), ContextRecall(), AnswerRelevancy()],
llm=ChatOpenAI(model="gpt-4o", temperature=0),
)
# 4. 把 ragas 分数写回 LangSmith run
# result.to_pandas() = 数据集列 + 分数列拼接（src/ragas/dataset_schema.py:452），
# 里面没有 run_id，所以用上面自己攒的 run_ids 按下标对齐
for run_id, row_score in zip(run_ids, result.scores):
client.create_feedback(
run_id=run_id,
key="faithfulness",
score=row_score["faithfulness"],
)这种集成模式有 3 个工程要点：
- dataset 在 LangSmith / 评测在 ragas / trace 在 LangSmith：每个工具做自己最擅长的事，不重叠
- feedback 双向写回：ragas 分数写回 LangSmith 的 run，让 LangSmith UI 上能看到每条 trace 的 faithfulness 分数
- LLM 客户端可替换：上面用 ChatOpenAI 当 judge，可以无缝换成 ChatAnthropic / ChatGoogle，对应换 judge model
这种"工具组合 + 数据流互通"是 LLM 工程的主流形态。理解每个工具的边界，把它们粘合到自己的 pipeline，是高级工程师的工作。
11.10.2 ragas 与 LangChain / LangGraph 的集成
ragas 与 LangChain 生态的集成在 2024-2025 年快速演化。下面是本专栏锚定的 ragas v0.2.10 所面向的集成形态——它对应的是 LangChain 0.3 分支；LangChain 本身此后已经走到 1.x（《LangChain 设计与实现》《RAG 工程与检索系统设计》两栏锚定的都是 langchain-core 1.2.26），换版本时这段的导入路径要跟着调：
# ragas v0.2.10 + LangChain 0.3 chain 评测
from langchain_core.runnables import RunnablePassthrough
from ragas.integrations.langchain import EvaluatorChain
from ragas.metrics import Faithfulness
# 把 ragas Faithfulness 包装成 LangChain Evaluator
faithfulness_evaluator = EvaluatorChain(metric=Faithfulness())
# 在 LangChain chain 中嵌入评测
# 注意：EvaluatorChain 的 input_keys 走的是 ragas **v1 列名**
#（src/ragas/utils.py:198 get_required_columns_v1 把 user_input/response/
# retrieved_contexts 映射回 question/answer/contexts），
# 传 v2 名字会被 LangChain 判成 "Missing some input keys"
def my_rag_chain(query):
# Faithfulness 要的是 str 与 list[str]，先把 Document / AIMessage 拆成纯文本
contexts = [d.page_content for d in retriever.invoke(query)]
answer = llm.invoke(prompt.format(query=query, context=contexts)).content
score = faithfulness_evaluator.invoke({
"question": query,
"answer": answer,
"contexts": contexts,
})
return {"answer": answer, "faithfulness": score}
# ragas + LangGraph 状态机评测
from langgraph.graph import StateGraph
from ragas.dataset_schema import MultiTurnSample
from ragas.metrics import ToolCallAccuracy
def evaluate_node(state):
# ToolCallAccuracy 是 MultiTurnMetric；Metric.score() 在 0.2 已标 deprecated
# （src/ragas/metrics/base.py:137，0.3 移除），走 multi_turn_score
sample = MultiTurnSample(user_input=state.history,
reference_tool_calls=state.expected_tool_calls)
score = ToolCallAccuracy().multi_turn_score(sample)
state.eval_results.append(score)
return state
graph = StateGraph(...)
graph.add_node("evaluate", evaluate_node)这种"评测嵌入到主流程"的模式让 ragas 不只是离线工具，更是生产 RAG/Agent 系统的实时质量传感器。
工业实务：把 ragas 评测节点嵌入 LangGraph 状态机的关键节点（如 retriever 输出后、generator 输出后），实时检测质量异常。这比"事后跑评测"提前发现问题——从"事后审计"升级到"实时控制"。
11.10.3 ragas 的可观测性集成：与 LangSmith / Langfuse / Phoenix 的对接细节
ragas 评测产生的指标如果只看终端输出，价值有限。它的真正威力在于与可观测性平台一体化。看具体代码：
# 与 LangSmith 集成：v0.2.10 的 ragas.integrations.langsmith 只导出
# upload_dataset() 与 evaluate()（没有 upload_results 这个函数）
from ragas.integrations.langsmith import upload_dataset
from ragas.integrations.langsmith import evaluate as langsmith_evaluate
upload_dataset(testset, dataset_name="customer-v1")
result = langsmith_evaluate(
dataset_name="customer-v1",
llm_or_chain_factory=my_rag_chain,
metrics=[Faithfulness()],
)
# → 评测结果出现在 LangSmith dashboard, 与 trace 关联
# 与 Langfuse 集成：ragas 仓库里**没有** integrations/langfuse 模块
#（v0.2.10 只有 helicone / langchain / langgraph / langsmith /
# llama_index / opik 六个），走 Langfuse 自己的 LangChain callback
from langfuse.langchain import CallbackHandler
callback = CallbackHandler() # 读 LANGFUSE_* 环境变量
result = evaluate(dataset=ds, metrics=[Faithfulness()], callbacks=[callback])
# → 每条评测调用在 Langfuse 里有 trace
# 与 Phoenix 集成
from phoenix.experiments import run_experiment
experiment = run_experiment(
dataset=phoenix_dataset,
task=my_rag_task,
evaluators=[wrap_ragas(Faithfulness())], # wrap_ragas 需自写：ragas 与 phoenix 都不提供这个适配函数
)
# → 跑成 Phoenix 的 experiment, UI 上看每条样例分数这种集成模式让评测从"一次性 CLI 跑"变成"持续生产监控"。工业团队的标准做法：
- CI 时跑 ragas，结果写回 LangSmith/Langfuse 关联到 PR
- 生产时 ragas 在线 grader 实时打分，写到 Phoenix dashboard
- 出现退化时 → 自动 rollback + Slack 通知
ragas 不是孤立工具——它是评测可观测性管线中的一个核心节点。
11.10.4 ragas 与 LangChain Hub 的协同：prompt 资产化
LangChain 在 2023 年推出了 LangChain Hub——把 prompt 当作可共享、可版本化的资产。ragas 与 LangChain Hub 的协同可以让团队的 LLM-judge prompt 资产化：
from langchain import hub
# 从 Hub 拉一个 judge prompt（"<owner>/<prompt-name>" 为占位，换成 Hub 上实际存在的名字）
public_prompt = hub.pull("<owner>/<faithfulness-judge-prompt>")
# 或者 push 自家的 prompt 上去给团队复用
hub.push("acme/customer-support-faithfulness", my_custom_prompt)这种 "prompt as asset" 的工程模式有几个红利：
- 跨项目复用：A 团队的好 prompt 直接被 B 团队用
- 社区共创：开源 ragas / LangChain 社区贡献最佳 prompt（类似 PyPI 包生态）
- 版本演化可见：prompt 的每一次改动都有 commit 记录
- 品质背书：Hub 上的"高 star" prompt 通常经过多团队验证
工业团队的实操：内部建一个 prompt registry（不一定用 LangChain Hub，可以自建），把团队的 judge prompt / 业务 prompt 都集中管理。这种"prompt 管理化"是 2025 年起 LLM 工程的成熟标志。
11.10.5 ragas 与 LangSmith / Langfuse trace 的双向落地——一份完整 wrapper
§11.10.2 提到 ragas 与 LangChain / LangGraph 集成，本节给出更具体的工程粘合层——把 ragas 评测结果双向写进 trace 平台：评测分数挂到原 trace 上、原 trace 的 input/output 直接喂给 ragas。
import asyncio
import json
from dataclasses import dataclass
from typing import Iterable
from datetime import datetime
from ragas import evaluate, EvaluationDataset
from ragas.metrics import Faithfulness, AnswerRelevancy, ContextRecall
@dataclass
class TraceEvalLink:
trace_id: str
run_id: str
metric_scores: dict[str, float]
eval_run_url: str
evaluator: str
evaluated_at: str
class RagasTracePlatformBridge:
"""把 ragas 评测与 LangSmith / Langfuse trace 双向打通"""
def __init__(self, platform: str = "langfuse", client=None):
assert platform in ("langsmith", "langfuse")
self.platform = platform
self.client = client
def _trace_to_sample(self, trace: dict) -> dict:
"""把平台的 trace 拉成 ragas 格式"""
return {
"user_input": trace["input"],
"response": trace["output"],
"retrieved_contexts": trace.get("retrieved_chunks", []),
"reference": trace.get("ground_truth"),
"trace_id": trace["trace_id"],
}
def _build_dataset(self, traces: Iterable[dict]) -> EvaluationDataset:
samples = [self._trace_to_sample(t) for t in traces]
return EvaluationDataset.from_list(samples)
async def evaluate_traces(self, traces: list[dict],
metrics=None) -> list[TraceEvalLink]:
metrics = metrics or [Faithfulness(), AnswerRelevancy(), ContextRecall()]
dataset = self._build_dataset(traces)
result = evaluate(dataset=dataset, metrics=metrics)
return self._link_back(traces, result)
def _link_back(self, traces, result) -> list[TraceEvalLink]:
links = []
for trace, row in zip(traces, result.scores):
link = TraceEvalLink(
trace_id=trace["trace_id"],
run_id=trace.get("run_id", ""),
metric_scores={k: float(v) for k, v in row.items()},
eval_run_url=self._eval_run_url(result),
evaluator="ragas-0.2.x",
evaluated_at=datetime.utcnow().isoformat(),
)
self._upload_score(link)
links.append(link)
return links
def _upload_score(self, link: TraceEvalLink):
if self.platform == "langfuse":
for metric_name, score in link.metric_scores.items():
self.client.score(
trace_id=link.trace_id,
name=f"ragas.{metric_name}",
value=score,
comment=f"From {link.eval_run_url}",
)
else: # langsmith
for metric_name, score in link.metric_scores.items():
self.client.create_feedback(
run_id=link.run_id,
key=f"ragas.{metric_name}",
score=score,
)
def _eval_run_url(self, result) -> str:
return getattr(result, "run_id", "ragas-local")flowchart LR TP[Trace 平台<br/>LangSmith / Langfuse] -->|fetch traces| B[Bridge] B -->|to_sample| DS[ragas EvaluationDataset] DS -->|evaluate| EV[Faithfulness / AR / CR] EV -->|score 矩阵| LK[link_back] LK -->|score upload| TP LK --> RPT[评测报告] style B fill:#e3f2fd style EV fill:#fff3e0 style RPT fill:#e8f5e9
工程实务 4 条集成模式：
- 每天对前 24h trace 1% 采样（约 50-200 条）跑评测——成本按 §11.13.1 的调用次数自己算，这个量级通常不大
- score 命名带
ragas.前缀——dashboard 上能直接区分"ragas 给的分"vs"用户反馈分" - 保留 eval_run_id——后续审计 / debug 时能从 trace 反查到具体评测 run
- 失败 case 自动标 high-priority tag——任何 metric < 0.5 的 trace 被打
low-quality标签
读者部署这套 bridge 后，dashboard 上每条 trace 旁边都会出现"ragas Faithfulness=0.83"的分数 chip——产品经理 / 工程师 / 客服都能直观看见每条对话的客观质量。这是 §17.11.4 "trace 数据语义切片"的具体落地。
研究背景：Langfuse 与 LangSmith 的官方文档都给出了"外部评测分数写回 trace"的接入方式（Langfuse 的 score、LangSmith 的 feedback），本节 bridge 就是按这两个接口写的。
11.10.6 ragas 的"实验追踪"集成——把每次评测变成可复现的 ML 实验
ragas 单跑结果是 jsonl，但生产团队需要"评测 = ML 实验"——每次 PR / 模型变化都自动记录。下面是与 wandb / mlflow 的工程化集成：
import wandb
import mlflow
import json
from datetime import datetime
from dataclasses import asdict
from pathlib import Path
from typing import Iterable
class RagasExperimentTracker:
"""ragas 评测的实验追踪 wrapper"""
def __init__(self, backend: str = "wandb",
project: str = "rag-evals"):
assert backend in ("wandb", "mlflow", "both")
self.backend = backend
self.project = project
def start_run(self, run_name: str, tags: dict, hyperparams: dict):
if self.backend in ("wandb", "both"):
wandb.init(project=self.project, name=run_name,
tags=list(tags.values()),
config=hyperparams)
if self.backend in ("mlflow", "both"):
mlflow.start_run(run_name=run_name)
for k, v in {**tags, **hyperparams}.items():
mlflow.set_tag(k, v) if k in tags else \
mlflow.log_param(k, v)
def log_metrics(self, metrics: dict[str, float], step: int = None):
if self.backend in ("wandb", "both"):
wandb.log(metrics, step=step)
if self.backend in ("mlflow", "both"):
for k, v in metrics.items():
mlflow.log_metric(k, v, step=step)
def log_failure_table(self, failed_rows: list[dict]):
if self.backend in ("wandb", "both"):
cols = list(failed_rows[0].keys()) if failed_rows else []
data = [[r.get(c) for c in cols] for r in failed_rows]
wandb.log({"failures": wandb.Table(columns=cols, data=data)})
if self.backend in ("mlflow", "both"):
tmp = Path("/tmp/failures.jsonl")
tmp.write_text("\n".join(json.dumps(r) for r in failed_rows))
mlflow.log_artifact(str(tmp), "failures")
def log_artifact_dataset(self, dataset_path: Path, name: str):
if self.backend in ("wandb", "both"):
artifact = wandb.Artifact(name, type="dataset")
artifact.add_file(str(dataset_path))
wandb.log_artifact(artifact)
if self.backend in ("mlflow", "both"):
mlflow.log_artifact(str(dataset_path), name)
def end_run(self):
if self.backend in ("wandb", "both"):
wandb.finish()
if self.backend in ("mlflow", "both"):
mlflow.end_run()
# 使用例
async def run_eval_with_tracking(model_name: str, prompt_version: str):
from ragas import evaluate, EvaluationDataset
from ragas.metrics import Faithfulness, AnswerRelevancy
tracker = RagasExperimentTracker(backend="both")
tracker.start_run(
run_name=f"{model_name}_{prompt_version}_{datetime.now():%Y%m%d-%H%M}",
tags={"model": model_name, "prompt_version": prompt_version,
"git_sha": get_git_sha()},
hyperparams={"top_k": 5, "temperature": 0,
"embedder": "text-embedding-3-large"},
)
# ragas evaluate
dataset = EvaluationDataset.from_jsonl("evals/golden.jsonl")
result = evaluate(dataset=dataset,
metrics=[Faithfulness(), AnswerRelevancy()])
# 记录到 tracker
# 注意 EvaluationResult.__getitem__（src/ragas/dataset_schema.py:416）返回的是
# 「每条样本一个分」的 list，不是标量，聚合要自己做
import numpy as np
tracker.log_metrics({
"faithfulness": float(np.nanmean(result["faithfulness"])),
"answer_relevancy": float(np.nanmean(result["answer_relevancy"])),
})
# 失败 case 记录
failed = [row for row in result.scores
if row.get("faithfulness", 1.0) < 0.6]
tracker.log_failure_table(failed)
tracker.log_artifact_dataset(Path("evals/golden.jsonl"), "eval_dataset_v1")
tracker.end_run()flowchart LR PR[PR 触发 CI] --> RUN[ragas evaluate] RUN --> ET[ExperimentTracker] ET --> WB[wandb.log] ET --> ML[mlflow.log_metric] RUN --> FAIL[失败 case] FAIL --> ET WB --> DASH[wandb dashboard<br/>历史曲线] ML --> MLF[MLflow Model Registry] DASH --> CMP[模型 / prompt 横向对比] MLF --> REG[版本注册] style ET fill:#e3f2fd style DASH fill:#e8f5e9
工程实务的 4 个集成模式：
| 团队类型 | 推荐 backend | 理由 |
|---|---|---|
| ML-heavy 团队 | wandb | 模型实验已在用 |
| 平台团队 | mlflow | 与 ML platform 标准对齐 |
| 新创业 | wandb（Free Tier 够用） | 成本可控 |
| 大企业 | both | 不绑定单一供应商 |
具体例子（示意数据，需用你自己的数据复算）：评测集 1000 题 × 4 metrics，每月跑一次、连续 8 个月 = 8 个 run。dashboard 上能看到：
- "Faithfulness 周环比"曲线 → 模型选型决策
- "Answer Relevancy on hard set" → 难度提升后的稳定性
- 失败 case top 10 一直没变 → 黄金集需要补 hard case
11.11 扩展 ragas：自定义 metric 与 LLM
11.11.1 ragas 的 BYOM 模式：自带 Judge LLM
不显式传 llm 时，ragas 用 llm_factory() 造一个 OpenAI 的 gpt-4o-mini 当 judge（src/ragas/llms/base.py:378-379），但它也支持完全 "Bring Your Own Model"：
from ragas.llms import LangchainLLMWrapper
from langchain_anthropic import ChatAnthropic
# 用 Claude 当 judge
custom_llm = LangchainLLMWrapper(ChatAnthropic(model="claude-sonnet-5")) # 示例；按官方模型页选现行型号
# 用本地部署的开源模型
from langchain_community.llms import VLLMOpenAI
local_llm = LangchainLLMWrapper(VLLMOpenAI(
openai_api_base="http://localhost:8000/v1",
model_name="meta-llama/Llama-3.1-70B-Instruct",
))
result = evaluate(dataset=ds, metrics=[Faithfulness()], llm=custom_llm)BYOM 的工程价值：
- 数据合规：用本地部署的开源模型当 judge，敏感数据不出局域网
- 成本控制：大规模评测时，已有 GPU 资源的团队用本地开源模型可以省掉按 token 计费的 API 开销（是否更便宜取决于你的硬件利用率，要自己算）
- 领域专化：fine-tuned 的领域 judge 在专业场景（医疗 / 金融）可能比通用 GPT 更准
但 BYOM 有代价——换成开源模型当 judge 后，与人工标注的一致度可能下降，差多少只能在自家锚点集上测；第 8 章 §8.2.2 引的 JudgeBench 数据也提醒，即便闭源旗舰模型在高难度判分上也远非完美。所以选 BYOM 时必须额外做元评测验证，确保 calibration 不退化太多。
工业实践：日常评测用 BYOM 省钱，关键决策（模型上线 gate）仍用闭源旗舰模型当 judge——双轨制平衡成本与可靠性。
11.11.2 ragas 的"扩展开发"工程模式
工业团队基于 ragas 做扩展开发的常见模式：
模式 1：自定义 Metric
from dataclasses import dataclass, field
from ragas.metrics.base import MetricWithLLM, SingleTurnMetric, MetricType
@dataclass # Metric 是 dataclass，子类不加 @dataclass 的话 name / _required_columns 会被基类 __init__ 覆盖掉
class CustomerSupportPoliteness(MetricWithLLM, SingleTurnMetric):
name: str = "cs_politeness"
_required_columns: dict = field(
default_factory=lambda: {MetricType.SINGLE_TURN: {"user_input", "response"}}
)
async def _single_turn_ascore(self, sample, callbacks):
# 调 LLM 评判 response 是否礼貌
...继承 ragas 抽象类即可——不修改 ragas 源码、自家代码独立维护。
模式 2：自定义 Prompt
# NLI prompt 字段叫 nli_statements_message（src/ragas/metrics/_faithfulness.py:177），传实例即可替换
strict_faithfulness = Faithfulness(nli_statements_message=MyStrictNLIPrompt())模式 3：自定义 LLM
from ragas.llms import LangchainLLMWrapper
custom_llm = LangchainLLMWrapper(MyInternalLLMClient())
result = evaluate(dataset, metrics=[Faithfulness(llm=custom_llm)])模式 4：完整 Pipeline 嵌入
# 把 ragas 评测嵌入到自家 CI / 平台
def my_eval_pipeline(samples):
ragas_result = evaluate(samples, metrics=[Faithfulness()])
# 自定义后处理
custom_metrics = my_custom_eval(samples)
return merge_results(ragas_result, custom_metrics)四种模式从浅到深覆盖了工业扩展的所有场景。ragas 的开放 API 让团队不必 fork 整个仓库——通过继承 / 替换 / 嵌入即可定制。这种"开放扩展点"是开源工具长期可用性的关键。
11.11.3 完整的 ragas 自定义 Metric 实现
整合本章方法学，给一份"自家业务专属 Metric"的完整 ragas 子类实现：
# my_metrics/cs_politeness.py
from __future__ import annotations
import typing as t
from dataclasses import dataclass, field
from pydantic import BaseModel, Field
from ragas.dataset_schema import SingleTurnSample
from ragas.metrics.base import (
MetricOutputType, MetricType, MetricWithLLM, SingleTurnMetric,
)
from ragas.prompt import PydanticPrompt
# 1. 定义 prompt 的 input/output schema
class PolitenessInput(BaseModel):
question: str = Field(description="The user's question")
response: str = Field(description="The chatbot's response")
class PolitenessOutput(BaseModel):
score: int = Field(ge=1, le=5, description="Politeness 1-5")
reason: str = Field(description="Why this score")
has_greeting: bool
has_apology: bool
is_imperative: bool
# 2. 定义 prompt class
class PolitenessPrompt(PydanticPrompt[PolitenessInput, PolitenessOutput]):
instruction = """
评估客服回答的礼貌度（1-5）。
考虑维度:
- 1: 命令式 / 冷淡 / 让人不舒服
- 2: 中性偏冷
- 3: 中性
- 4: 礼貌 / 有问候语 / 必要时道歉
- 5: 极其礼貌 / 体贴 / 主动关心用户感受
"""
input_model = PolitenessInput
output_model = PolitenessOutput
examples = [
(
PolitenessInput(question="退款多久到账?", response="3-5 工作日."),
PolitenessOutput(score=2, reason="过于冷淡, 没问候没解释",
has_greeting=False, has_apology=False, is_imperative=False),
),
(
PolitenessInput(question="你们东西有问题",
response="非常抱歉给您带来不便! 您能描述具体问题吗? 我立即帮您处理."),
PolitenessOutput(score=5, reason="主动道歉 + 询问详情 + 立即解决",
has_greeting=False, has_apology=True, is_imperative=False),
),
]
# 3. 定义 Metric 主类
@dataclass
class CsPoliteness(MetricWithLLM, SingleTurnMetric):
name: str = "cs_politeness"
_required_columns: t.Dict[MetricType, t.Set[str]] = field(
default_factory=lambda: {
MetricType.SINGLE_TURN: {"user_input", "response"}
}
)
output_type: MetricOutputType = MetricOutputType.DISCRETE
politeness_prompt: PydanticPrompt = field(default_factory=PolitenessPrompt)
async def _single_turn_ascore(self, sample: SingleTurnSample, callbacks) -> float:
assert self.llm is not None
result = await self.politeness_prompt.generate(
data=PolitenessInput(
question=sample.user_input,
response=sample.response,
),
llm=self.llm,
callbacks=callbacks,
)
return result.score / 5.0 # normalize to 0-1约 70 行代码完成一个工业级自定义 Metric：
- 类型化 input / output（PydanticPrompt 标准）
- 详细 instruction + 2 个 example（覆盖低分 / 高分）
- 多维度输出（score + reason + 3 个 boolean 标签）
- 与 ragas 主框架无缝集成
- 归一化到 0-1（与其他 metric 可比）
11.11.4 ragas 与"自定义 metric 的工程化"——派生新 metric 的最佳实践
下面给出派生自家 metric 的工程化模板：
import json
from dataclasses import dataclass
import numpy as np
from langchain_core.prompt_values import StringPromptValue
from ragas.metrics.base import MetricType, MetricWithLLM, SingleTurnMetric
from ragas.dataset_schema import SingleTurnSample
@dataclass
class CustomerSatisfactionMetric(MetricWithLLM, SingleTurnMetric): # 要用 self.llm 就得继承 MetricWithLLM
"""评估客服回答的"用户满意度预期"——业务专属 metric"""
name: str = "customer_satisfaction"
_required_columns: dict[MetricType, set[str]] = None
def __post_init__(self):
super().__post_init__()
self._required_columns = {
MetricType.SINGLE_TURN: {"user_input", "response"},
}
# 自定义评估 prompt
self.scoring_prompt = """
评估这条客服回答的用户满意度预期 (1-5):
- 回答专业 + 解决问题: 5
- 回答专业但不完整: 4
- 回答勉强但有诚意: 3
- 回答跑题或冷漠: 2
- 回答错误或不礼貌: 1
Question: {question}
Answer: {answer}
Output: JSON {{"score": int, "reason": str}}
""" # 花括号要写成 {{ }}，否则下面的 str.format 会把它当占位符报 KeyError
async def _single_turn_ascore(self, sample: SingleTurnSample,
callbacks=None) -> float:
"""ragas 0.2.x 派生 metric 的标准接口"""
prompt = self.scoring_prompt.format(
question=sample.user_input,
answer=sample.response,
)
# 调用 ragas 注入的 LLM；BaseRagasLLM 收的是 PromptValue 而不是 str
response = await self.llm.agenerate_text(StringPromptValue(text=prompt))
try:
data = json.loads(response.generations[0][0].text)
return data["score"] / 5.0 # 归一化到 0-1
except (json.JSONDecodeError, KeyError, TypeError):
return np.nan # 解析失败记 NaN（与 ragas 内置 metric 同口径），别记 0 污染均值# 派生 metric 必备的 5 步工程化骨架
class CustomMetricChecklist:
"""派生新 metric 的 5 步检查表"""
STEPS = [
{
"step": "1. 定义需求",
"action": "明确该 metric 与现有 metric 的差异 + 业务理由",
"anti_pattern": "派生现成 metric 的细微变种 = 浪费",
},
{
"step": "2. 写 scoring prompt",
"action": "5 档 rubric + JSON 输出 + reason",
"anti_pattern": "无 rubric 的纯 LLM-judge",
},
{
"step": "3. 实现 _single_turn_ascore",
"action": "继承 SingleTurnMetric + async + 容错",
"anti_pattern": "同步实现（ragas 是 async-first）",
},
{
"step": "4. anchor 200 题验证 κ",
"action": "与人工 anchor 比 κ ≥ 0.6",
"anti_pattern": "上线前不 validate",
},
{
"step": "5. 按 §11.11.6 自己规划中间值复用",
"action": "在派生类里缓存 statements 这类中间产物",
"anti_pattern": "以为 ragas 会自动共享（0.2.10 不会）",
},
]flowchart TB N[需求: 自家 metric] --> S1[Step 1: 定义] S1 --> S2[Step 2: scoring prompt] S2 --> S3[Step 3: 派生 class] S3 --> S4[Step 4: anchor 验证] S4 -->|κ ≥ 0.6| S5[Step 5: 接 dep graph] S4 -->|κ < 0.6| FIX[修 prompt 重测] FIX --> S2 S5 --> PR[提 PR + review] PR --> RELEASE[发布到团队 metric 库] style RELEASE fill:#e8f5e9 style FIX fill:#fff3e0
工程实务的 4 类常见自定义 metric：
| metric | 业务场景 | 估时 |
|---|---|---|
| customer_satisfaction | 客服 chatbot | 1 周 |
| medical_safety_score | 医疗 RAG | 2 周（需领域专家） |
| code_pep8_compliance | 代码生成 | 3 天 |
| chinese_idiomatic_quality | 中文创作 | 1 周 |
具体例子（示意数据，需用你自己的数据复算）：某客服团队派生 customer_satisfaction metric 全程 1 周：
- Day 1：定义 + rubric 设计
- Day 2-3：实现 + 单测
- Day 4-5：anchor 200 题人工标注
- Day 6：跑 κ 计算 → 0.68 ✅
- Day 7：PR review + 接 dep graph
3 类常见派生错误：
| 错误 | 现象 | 修法 |
|---|---|---|
| 命名冲突 | 与内置 metric 同名 | 加业务 prefix（cs_satisfaction） |
| 无 fallback | LLM 返回非 JSON 时 crash | try/except + 返回 NaN |
| 忘记 async | 在 _single_turn_ascore 里用同步阻塞调用，拖垮整批并发 |
LLM 调用用 await 的异步接口 |
11.11.5 ragas 自定义 LLM provider——把 ragas 接到内部 LLM Gateway
ragas 默认用 OpenAI / Anthropic SDK，但工业团队多走自家 LLM Gateway（合规、缓存、限速）。下面给出完整的 custom provider 接入范本——把 ragas 调用全部走自家 gateway。
import asyncio
from typing import Any
from ragas.llms.base import BaseRagasLLM
from ragas.embeddings.base import BaseRagasEmbeddings
from langchain_core.outputs import LLMResult, Generation
from langchain_core.prompt_values import PromptValue
from langchain_core.callbacks import Callbacks
class InternalGatewayLLM(BaseRagasLLM):
"""让 ragas 走内部 LLM Gateway"""
def __init__(self, gateway_url: str, api_key: str,
model: str = "gpt-4o-mini",
timeout: int = 60):
super().__init__()
self.url = gateway_url
self.api_key = api_key
self.model = model
self.timeout = timeout
def _build_payload(self, prompt: PromptValue, n: int = 1) -> dict:
# BaseRagasLLM 的抽象签名收的是 langchain PromptValue，不是 str
# （src/ragas/llms/base.py:72-89），要显式取字符串
return {
"model": self.model,
"messages": [{"role": "user", "content": prompt.to_string()}],
"n": n,
"temperature": 0,
"metadata": {"source": "ragas-eval"}, # 让 Gateway 知道是评测调用
}
async def agenerate_text(self, prompt: PromptValue, n: int = 1,
temperature: float = 0,
stop: list[str] = None,
callbacks: Callbacks = None) -> LLMResult:
import aiohttp
async with aiohttp.ClientSession() as session:
payload = self._build_payload(prompt, n)
async with session.post(
self.url + "/v1/chat/completions",
json=payload,
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=self.timeout,
) as resp:
data = await resp.json()
generations = [[
Generation(text=c["message"]["content"])
for c in data["choices"]
]]
return LLMResult(generations=generations)
def generate_text(self, prompt: PromptValue, n: int = 1,
**kwargs) -> LLMResult:
return asyncio.run(self.agenerate_text(prompt, n, **kwargs))
class InternalGatewayEmbeddings(BaseRagasEmbeddings):
"""让 ragas embedding 调用走自家 Gateway"""
def __init__(self, gateway_url: str, api_key: str,
model: str = "text-embedding-3-large"):
super().__init__()
self.url = gateway_url
self.api_key = api_key
self.model = model
async def aembed_query(self, text: str) -> list[float]:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.post(
self.url + "/v1/embeddings",
json={"model": self.model, "input": text},
headers={"Authorization": f"Bearer {self.api_key}"},
) as resp:
data = await resp.json()
return data["data"][0]["embedding"]
# aembed_query 与 aembed_documents 都是 BaseRagasEmbeddings 的
# @abstractmethod（embeddings/base.py:80-83），少实现一个就实例化不了
async def aembed_documents(self, texts: list[str]) -> list[list[float]]:
return [await self.aembed_query(t) for t in texts]
def embed_query(self, text: str) -> list[float]:
return asyncio.run(self.aembed_query(text))
def embed_documents(self, texts: list[str]) -> list[list[float]]:
return [self.embed_query(t) for t in texts]
# 使用
async def run_ragas_with_gateway():
from ragas import evaluate, EvaluationDataset
from ragas.metrics import Faithfulness, AnswerRelevancy
llm = InternalGatewayLLM(
gateway_url="https://llm-gateway.internal.company.com",
api_key="sk-internal-xxx",
model="gpt-4o-mini",
)
embedder = InternalGatewayEmbeddings(
gateway_url="https://llm-gateway.internal.company.com",
api_key="sk-internal-xxx",
)
dataset = EvaluationDataset.from_jsonl("evals/golden.jsonl")
result = evaluate(
dataset=dataset,
metrics=[Faithfulness(llm=llm), AnswerRelevancy(llm=llm,
embeddings=embedder)],
)
return resultflowchart LR
R[ragas evaluate] --> M[Faithfulness / Answer Relevance]
M --> L[InternalGatewayLLM]
M --> E[InternalGatewayEmbeddings]
L --> GW[LLM Gateway<br/>内部统一入口]
E --> GW
GW --> ROUTE{路由}
ROUTE -->|GPT-4o| OAI[OpenAI API]
ROUTE -->|Claude| ANTH[Anthropic API]
ROUTE -->|内部 vLLM| VLM[自部署模型]
GW --> CACHE[Redis 缓存]
GW --> RL[限速器]
GW --> AUD[审计日志]
style GW fill:#e3f2fd
style CACHE fill:#fff3e0
style AUD fill:#e8f5e9
工程实务的 5 个 Gateway 接入价值：
- 统一审计：所有评测调用都进 audit log，合规审计可追溯
- 缓存命中：同 (prompt, model) 的重复评测命中 Redis，省下的比例取决于重跑频率
- 限速保护：评测大量并发不会打爆 OpenAI rate limit
- 路由灵活：换 model（gpt-4o → claude-opus）改 1 行 config，不改代码
- metadata.source 标识：Gateway 知道是"评测流量"，可与生产流量分账
研究背景：
- ragas 0.2.x 的
BaseRagasLLM只有generate_text/agenerate_text两个抽象方法（src/ragas/llms/base.py:72-89），正适合这种自定义接入
11.11.6 多 metric 同跑的真实成本：ragas 不共享中间结果，要省得自己规划
先把一个常见误解掐掉：v0.2.10 的 evaluate() 并没有任何"metric 依赖图 / 中间结果复用"机制。翻 src/ragas/evaluation.py:255-290 就能看到，它对每条 sample、每个 metric 各自 executor.submit(metric.single_turn_ascore, sample, ...)，metric 之间完全独立、互不知情。后果很具体：Faithfulness 会让 LLM 把 response 拆一遍 statements，AnswerCorrectness（src/ragas/metrics/_answer_correctness.py）为了算 TP/FP/FN 还会再拆一遍同一段 response——这次重复调用没有人帮你省掉。所以同时跑 N 个 metric 的 LLM 成本就是各自成本的加和，不打折。
ragas 在这里真正给你的优化是并发而不是去重：Executor 按 RunConfig.max_workers 把这些互相独立的任务并行发出去，省的是墙钟时间，不是 token 账单。
那"省钱"这件事就得自己做。下面这段不是 ragas 的源码，而是一个可以放在 ragas 外面的编排器骨架——把每个 metric 需要的字段、会产出的中间值、大致的 LLM 调用数显式声明出来，先算清一组 metric 的账，再决定跑哪几个、要不要自己缓存 statements 这类中间产物：
# 自写的编排器骨架（ragas 0.2.10 里没有这个东西）
from dataclasses import dataclass
from typing import Iterable
@dataclass
class MetricRequirement:
"""一个 metric 需要哪些字段才能计算"""
metric_name: str
required_fields: list[str] # 例: ["user_input", "retrieved_contexts"]
intermediate_outputs: list[str] # 计算过程中产生的中间值
cost_units: int # 估算成本（LLM call 数）
class MetricDependencyGraph:
"""自己维护的 metric 依赖表——用于估算共享中间值之后能省多少"""
METRIC_DEPENDENCIES = {
"Faithfulness": MetricRequirement(
"Faithfulness",
required_fields=["user_input", "response", "retrieved_contexts"],
intermediate_outputs=["statements_extracted"],
cost_units=2,
),
"FaithfulnessWithHHEM": MetricRequirement(
"FaithfulnessWithHHEM",
required_fields=["user_input", "response", "retrieved_contexts"],
intermediate_outputs=["statements_extracted"], # 共享！
cost_units=1, # HHEM 用本地模型
),
"AnswerRelevance": MetricRequirement(
"AnswerRelevance",
required_fields=["user_input", "response"],
intermediate_outputs=["reverse_questions"],
cost_units=1,
),
"ContextRecall": MetricRequirement(
"ContextRecall",
required_fields=["user_input", "reference",
"retrieved_contexts"],
intermediate_outputs=["statements_in_reference"],
cost_units=1,
),
"AnswerCorrectness": MetricRequirement(
"AnswerCorrectness",
required_fields=["user_input", "response", "reference"],
intermediate_outputs=["statements_extracted", # 共享 Faithfulness
"statements_in_reference"], # 共享 Recall
cost_units=2,
),
}
def optimal_execution_plan(self,
metrics: list[str]) -> dict:
"""根据 metric 集合算最优执行顺序——共享中间值"""
all_intermediate = set()
total_cost = 0
execution_order = []
shared_count = 0
for m in metrics:
req = self.METRIC_DEPENDENCIES.get(m)
if not req:
continue
shared = set(req.intermediate_outputs) & all_intermediate
shared_count += len(shared)
unique = set(req.intermediate_outputs) - all_intermediate
all_intermediate.update(unique)
# 共享中间值的 metric 减半 cost
effective_cost = req.cost_units * (1 - len(shared) / max(
len(req.intermediate_outputs), 1))
total_cost += effective_cost
execution_order.append({
"metric": m,
"shared_intermediate": list(shared),
"unique_intermediate": list(unique),
"effective_cost": round(effective_cost, 1),
})
return {
"metrics": metrics,
"naive_total_cost": sum(self.METRIC_DEPENDENCIES[m].cost_units
for m in metrics
if m in self.METRIC_DEPENDENCIES),
"optimized_total_cost": round(total_cost, 1),
"savings_pct": round(
(1 - total_cost / max(sum(
self.METRIC_DEPENDENCIES[m].cost_units for m in metrics
if m in self.METRIC_DEPENDENCIES), 1)) * 100, 1),
"execution_order": execution_order,
"shared_intermediate_count": shared_count,
}flowchart LR ROOT[评测请求] --> P[依赖图分析] P --> S1[statements_extracted<br/>共享给 Faithfulness + Correctness] P --> S2[statements_in_reference<br/>共享给 Recall + Correctness] P --> S3[reverse_questions<br/>独占 AnswerRelevance] S1 --> F[Faithfulness 用现成] S1 --> AC[AnswerCorrectness 用现成] S2 --> CR[ContextRecall 用现成] S2 --> AC S3 --> AR[AnswerRelevance 单独跑] F --> R[最终结果] AC --> R CR --> R AR --> R style P fill:#e3f2fd style R fill:#e8f5e9
工程实务的 4 条 metric 编排经验：
- 别指望 ragas 替你 dedupe：
evaluate(metrics=[...])里多写一个 metric，就是实打实多一份 LLM 调用 - 要省就自己缓存中间产物：
statements_extracted这类跨 metric 复用的中间值，只能在你派生的 metric 类里自己存、自己传 - 先用下面这张表算账再决定跑哪几个：省下来的比例完全取决于你选的 metric 组合，不是框架属性
- 调
RunConfig.max_workers省的是时间不是钱：并发拉满不会让 token 变少
具体例子（按上面这个自写编排器的假设参数算出来的示意数据，不是 ragas 的实测行为，需用你自己的数据复算）：跑 4 个 metric 的对比：
| metrics | naive cost units | optimized | 节省 |
|---|---|---|---|
| F + AR | 3 | 3 | 0% |
| F + CR | 3 | 3 | 0% |
| F + AC | 4 | 3 | 25% |
| F + CR + AC | 5 | 3 | 40% |
| F + AR + CR + AC | 6 | 4 | 33% |
洞察：表里那 25-40% 是"如果有人帮你复用中间结果"能省下的上限——而 v0.2.10 的 ragas 一分都没帮你省。这张表的用处是告诉你自己动手值不值：只跑 F + AR 或 F + CR 的组合省不出东西（0%），真正有肉的是 AnswerCorrectness 和 Faithfulness / ContextRecall 同框的场景。
研究背景：
- Apache Spark 的 RDD lazy evaluation、TensorFlow / PyTorch 的 computation graph 都是"中间值复用"的成熟范式——ragas 的 metric 层还没有走到这一步，这正是留给内部 fork 的改造空间（见 §11.14.3）
- v0.2.10 的并发来自
Executor+RunConfig.max_workers，与中间值复用是两件事，别混为一谈
读者拿这张表去评估"要不要在自家 fork 里做中间值缓存"——而不是误以为装上 ragas 就自动享受了这份优化。
11.12 评测集生成与调试
11.12.1 隐藏功能：ragas 的 Synthetic Test Set Generation
ragas 还有一个被严重低估的功能——自动从你的文档库生成评测集：
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from ragas.testset import TestsetGenerator
generator = TestsetGenerator.from_langchain(
llm=ChatOpenAI(model="gpt-4o"),
embedding_model=OpenAIEmbeddings(),
)
# query_distribution 缺省时用 default_query_distribution()：
# SingleHopSpecific / MultiHopAbstract / MultiHopSpecific 三个各占 1/3
testset = generator.generate_with_langchain_docs(
documents=my_documents,
testset_size=200,
)它的工作流程：
- 把你的文档（如知识库）建成
KnowledgeGraph，按长度分布挑一套 transform 切分 - 抽 summary / theme / 实体 / 向量，按相似度在节点间连边
- 各个 Synthesizer 按权重抽 scenario，让 LLM 生成 query 与 grounded reference
- 输出 200 条
TestsetSample（user_input/reference/reference_contexts）
注意 0.1.x 那套 generator_llm / critic_llm / test_size / distributions={"simple": ...} 参数在 0.2 已经全部消失——照旧文档抄会直接 TypeError。
这种"自动生成评测集"对冷启动阶段的团队是救命功能——不需要人工写 200 道题，就能从文档库生成基线评测集。第 3 章 §3.3 的"五步冷启动法"是手动版本，ragas 的 testset generator 是自动化版本。
但注意：自动生成的题 ≠ 真实用户问题。生成的题往往风格趋同、缺少真实用户的口语化 / 错别字 / 模糊表达。所以 ragas 自动生成只能作为冷启动 + 兜底覆盖，长期还要靠 §3.6 hard case mining 从生产挖真题补充。
11.12.2 ragas 的 testset_generator——自动合成评测集
ragas 0.2.x 内置一个常被忽视但极有价值的模块：TestsetGenerator（ragas/testset/synthesizers/）。它能从你的语料库自动合成评测集，省掉人工标注的大头工作量。下面拆解其工程使用方式。
from ragas.testset import TestsetGenerator
from ragas.testset.synthesizers import (
SingleHopSpecificQuerySynthesizer,
MultiHopAbstractQuerySynthesizer,
)
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import DirectoryLoader
class RagasTestsetPipeline:
"""从语料 → 自动合成评测集"""
def __init__(self, llm_model: str = "gpt-4o-mini",
embedder_model: str = "text-embedding-3-large"):
self.llm = ChatOpenAI(model=llm_model)
self.embedder = OpenAIEmbeddings(model=embedder_model)
def build_generator(self):
return TestsetGenerator.from_langchain(
llm=self.llm,
embedding_model=self.embedder,
)
def generate_from_corpus(self, doc_dir: str, n: int = 200):
loader = DirectoryLoader(doc_dir, glob="**/*.md")
documents = loader.load()
gen = self.build_generator()
# 60% 单跳具体题 + 40% 多跳抽象题
# Synthesizer 要的是 BaseRagasLLM：用 gen.llm（from_langchain 已包好），不能直接传 ChatOpenAI
distribution = [
(SingleHopSpecificQuerySynthesizer(llm=gen.llm), 0.6),
(MultiHopAbstractQuerySynthesizer(llm=gen.llm), 0.4),
]
testset = gen.generate_with_langchain_docs(
documents=documents,
testset_size=n,
query_distribution=distribution,
)
return testsetragas 自动合成评测集的核心思想（参考 src/ragas/testset/synthesizers/base.py）：
- 建知识图谱：文档进
KnowledgeGraph，default_transforms（src/ragas/testset/transforms/default.py:31）按语料长度分布动态决定用哪套 transform——长文档占比 ≥ 25% 时才插入HeadlinesExtractor+HeadlineSplitter+SummaryExtractor这条切分链 - 节点属性抽取：
SummaryExtractor/ThemesExtractor/NERExtractor/EmbeddingExtractor给每个节点挂上摘要、主题、实体、向量 - 关系构建：按 embedding 相似度与主题重叠在节点间连边，多跳题就是沿着这些边取样出来的
- query 合成：
query_distribution里的各个 Synthesizer（SingleHopSpecificQuerySynthesizer/MultiHopAbstractQuerySynthesizer/MultiHopSpecificQuerySynthesizer）按权重抽取 scenario，再让 LLM 生成 query 与 grounded reference - persona 调制：
persona_list决定"谁在问"，同一批语料能生出不同视角的问法
flowchart LR D[语料文档] --> CK[chunk 切分] CK --> EMB[embedding] EMB --> KG[knowledge graph] KG --> NE[Node Extractor<br/>entity / theme / temporal] NE --> S1[SingleHop synth<br/>具体题] NE --> S2[MultiHop synth<br/>多跳抽象题] S1 --> A[answer 合成] S2 --> A A --> F[质量过滤] F --> TS[TestSet jsonl] style TS fill:#e8f5e9
工程实务的 5 条使用经验：
- 6:4 是这里手写的比例，不是 ragas 默认值——
default_query_distribution()（src/ragas/testset/synthesizers/__init__.py）给的是 SingleHopSpecific / MultiHopAbstract / MultiHopSpecific 三者各 1/3；要偏向具体题就得像上面这样显式传query_distribution - n=200 是常见起点——题太少时分数的置信区间宽（第 3 章讲样本量），题太多则合成与复核成本线性上涨
- 跑完必须人工 review 5-10%——automatic synth 偶尔出"实质上无法回答的题"，人工抽查能筛掉
- 保留 source_chunk 字段——每个 query 都标注源 chunk_id，后续可用于 retriever 评测的真值标注
- 限制 doc_dir 到"代表性子集"——把整本 RAG 知识库丢进去，cost 会爆炸（每个 doc 至少 2 次 LLM 调用）
具体成本（示意值，单价按 $0.005/call 假设）：200 题 × (1 question gen + 1 answer gen + 0.5 filter) × $0.005/call ≈ $2.5 一次合成（不含建知识图谱时 transform 阶段的调用）。一周更一次新版评测集（语料更新驱动），年成本约 $130。
研究背景：
- Saad-Falcon et al. 2023 "ARES: An Automated Evaluation Framework for Retrieval-Augmented Generation Systems" arXiv:2311.09476 同样用合成数据做 RAG 评测，是同期的代表工作
- ragas 自己的 testset 生成早在 2023-09 就已加入（#150），比 ARES 论文还早；现在这套基于
KnowledgeGraph的 TestsetGenerator 是 2024-10 随 0.2 换上的（#1401，把experimental.testset转正） - LlamaIndex 也有同类的合成评测集工具——思路一致，API 不同；注意它早期的
RagDatasetGenerator所在的llama_index.core.llama_dataset模块在 0.14 的 core 里已经没有了（llamaindex-0.14.24/llama-index-core/llama_index/core/下无llama_dataset/），只剩 docs 示例还在引用，照抄会 ImportError
读者要把"评测集冷启动"成本压到极致 → 先用 TestsetGenerator 合成 200 题 → 人工 review 选 100 题 → 加 50 题手工 hard case = 150 题黄金集起步。这是 §3.10.1 "50 题黄金集冷启动"的另一条路径——更工程化、更省人力。
11.12.3 工程实战：用 ragas 调试 RAG 失败的 5 步法
把第 13 章的方法用 ragas 落地，给一份"RAG 系统出现问题时用 ragas 5 步排查"的 SOP：
import pandas as pd
from ragas import evaluate
from ragas.metrics import (
ContextRecall, ContextPrecision, Faithfulness, AnswerRelevancy, AnswerCorrectness,
)
# Step 1: 跑全套核心指标
result = evaluate(
dataset=ds,
metrics=[
ContextRecall(), ContextPrecision(),
Faithfulness(), AnswerRelevancy(), AnswerCorrectness(),
],
)
df = result.to_pandas()
# Step 2: 找出最差的 N 条
worst = df.nsmallest(20, "faithfulness")
print(worst[["user_input", "faithfulness", "context_recall"]])
# Step 3: 按指标聚合
by_recall_low = df[df["context_recall"] < 0.5]
by_faith_low = df[df["faithfulness"] < 0.5]
overlap = pd.merge(by_recall_low, by_faith_low, on="user_input")
print(f"双低 case: {len(overlap)} (要先修 retriever)")
# Step 4: 调出失败 case 的中间结果
for _, row in worst.iterrows():
print(row["user_input"])
print("Retrieved:", str(row["retrieved_contexts"])[:200]) # 这一列是 list，先转字符串再截断
print("Response:", row["response"][:200])
print("---")
# Step 5: 决定修法
# - retrieved_contexts 缺关键 chunk → 调 retriever (chunk_size, top_k)
# - retrieved 对但 response 不引用 → 调 generator prompt
# - retrieved 无关 → 调 embedding 或 query rewrite这 5 步是工业 RAG 团队遇到指标问题时的标准排查流程。ragas 的 metric 输出 + pandas 集成让这个流程在几分钟内能完成 —— 把"我们 RAG 哪里出问题"从一个开放式问题变成有清晰路径的工程任务。
11.12.4 隐藏陷阱：ragas metric 之间的相关性
工程团队常犯的一个错误：同时报告多个 metric 但没意识到它们之间高度相关。
ragas 的几个核心 metric 在 RAG 场景里有显著相关：
graph LR CR[Context Recall] -->|强正相关| F[Faithfulness] CR -->|强正相关| AC[Answer Correctness] F -->|强正相关| AC CR -.弱负相关.-> AR[Answer Relevance] style CR fill:#dbeafe style F fill:#dcfce7 style AC fill:#fef3c7 style AR fill:#fce7f3
具体含义：
- Context Recall ↑ 通常 Faithfulness ↑：retriever 给出更准确的 context，generator 自然更好做 faithful 回答
- Faithfulness 与 Answer Correctness 相关：一个 faithful 的回答（基于 context）通常 correct（context 本身正确时）
- Recall 与 Relevance 略有取舍：检索 100 个 chunk 提升 Recall，但部分 noise chunk 会让 generator 的回答稍微跑偏（影响 Relevance）
工程含义：单独看任何一个 metric 的变化都可能误导。改 retriever 让 Recall 涨 5pp，连带 Faithfulness 涨 3pp、Answer Correctness 涨 2pp——这看起来像"四个指标都涨了"，实际只是同一个改动的不同投影。
正确报告方式：
- 把强相关的 metric 聚成一组（如 RAG 三件套），主指标只取一个
- 改动归因时只看主指标，连带变化作为辅助证据
- 元评测时检查 metric 之间的相关性矩阵——如果两个 metric 相关 > 0.9，留一个就够
这是评测体系成熟度的标志——团队不只看分数，还看 metric 之间的统计关系。
11.12.5 ragas 的"答案多样性"专项评测——避免 generator 千篇一律
ragas 内置 metric 都聚焦"对不对"维度，而忽视了一个工业重要维度：答案多样性。RAG 系统在 100 个相似问题上重复同一句"很抱歉，我无法帮您"是"对的"——但是用户感觉极差。本节给读者一份 ragas 视角下的"多样性"专项 metric，让 RAG 系统在保持对的同时，告别"机械人格"。
graph LR
A[100 个相似 query] --> B[generator 输出]
B --> C{多样性测度}
C --> D[词汇 distinct-n]
C --> E[语义 embedding cluster]
C --> F[句长 std]
C --> G[句式模板检测]
D & E & F & G --> H[多样性评分]
H --> I{是否过低}
I -->|是| J[生成器在套模板<br/>用户疲劳风险]
I -->|否| K[健康]
4 类多样性维度 + 适用场景：
| 维度 | 度量 | 健康阈值 | 失败征兆 | 适用 |
|---|---|---|---|---|
| 词汇多样性 distinct-n | 不同 n-gram 占比 | distinct-2 ≥ 0.4 | 全部用同样的 50 词 | 所有 RAG |
| 语义多样性 | embedding 聚类数 | ≥ 5 类/100 答 | 答案聚成 1-2 簇 | 高复杂场景 |
| 句长方差 | 长度 std | std ≥ 25 字符 | 全部 80 字 | 客服 / FAQ |
| 模板检测 | 高频前缀 / 套话占比 | ≤ 30% | "感谢您的提问"出现 90%+ | 客服 |
配套实现：ragas 多样性专项 metric：
import statistics
from collections import Counter
from dataclasses import dataclass, field
from typing import Callable
@dataclass
class DiversitySample:
query: str
answer: str
embedding: list[float] | None = None
@dataclass
class AnswerDiversityMetric:
"""ragas-style metric — 给一组答案输出多样性分数"""
high_template_threshold: float = 0.30
def distinct_n(self, answers: list[str], n: int = 2) -> float:
ngrams = []
for ans in answers:
tokens = ans.split()
ngrams.extend([" ".join(tokens[i:i+n]) for i in range(len(tokens)-n+1)])
if not ngrams: return 0.0
return len(set(ngrams)) / len(ngrams)
def length_std(self, answers: list[str]) -> float:
lens = [len(a) for a in answers]
return statistics.stdev(lens) if len(lens) >= 2 else 0.0
def template_dominance(self, answers: list[str], prefix_chars: int = 12) -> dict:
"""检测高频前缀套话"""
prefixes = [a[:prefix_chars].strip() for a in answers]
counter = Counter(prefixes)
most_common, count = counter.most_common(1)[0] if counter else ("", 0)
return {
"top_prefix": most_common,
"occurrence_pct": count / max(len(answers), 1),
"is_template_overused": count / max(len(answers), 1) > self.high_template_threshold,
}
def semantic_cluster_count(self, samples: list[DiversitySample],
cluster_fn: Callable[[list], list[int]] | None = None) -> int:
"""语义聚类（需外部 embedding + clustering），简化为类别数"""
embeddings = [s.embedding for s in samples if s.embedding is not None]
if not embeddings or not cluster_fn:
return 0
return len(set(cluster_fn(embeddings)))
def evaluate(self, samples: list[DiversitySample]) -> dict:
answers = [s.answer for s in samples]
if not answers:
return {"error": "no samples"}
d2 = self.distinct_n(answers, 2)
d3 = self.distinct_n(answers, 3)
len_std = self.length_std(answers)
tmpl = self.template_dominance(answers)
# 综合分数（0-1）
score = (
min(d2 / 0.4, 1.0) * 0.4 +
min(d3 / 0.5, 1.0) * 0.2 +
min(len_std / 25, 1.0) * 0.2 +
(1 - tmpl["occurrence_pct"]) * 0.2
)
return {
"diversity_score": round(score, 3),
"distinct_2": round(d2, 3),
"distinct_3": round(d3, 3),
"length_std": round(len_std, 1),
"template_dominance": tmpl,
"verdict": ("健康" if score >= 0.7
else "中等 - 需 prompt 鼓励多样化" if score >= 0.5
else "差 - 用户感知机械化风险"),
"recommendation": self._recommend(d2, len_std, tmpl),
}
def _recommend(self, d2: float, len_std: float, tmpl: dict) -> list[str]:
recs = []
if d2 < 0.3:
recs.append("提高 temperature 0.2-0.4 / 加 'use varied wording' 指令")
if len_std < 15:
recs.append("prompt 增加 'adapt response length to query complexity'")
if tmpl["is_template_overused"]:
recs.append(f"前缀 '{tmpl['top_prefix']}' 占 {tmpl['occurrence_pct']:.0%}，去除模板化开头")
if not recs:
recs.append("无需改动，多样性健康")
return recs举例（示意数据，需用你自己的数据复算）：某客服 RAG 跑 100 题多样性评测：
- distinct_2 = 0.18（远低于 0.4）/ distinct_3 = 0.25 / length_std = 8 字符（极低）/ 套话 "感谢您的提问，" 占 92%
- score = 0.36，verdict = "差 - 用户感知机械化风险"
- recommendation：3 条具体改动
- 调整后重测：distinct_2 0.45 / distinct_3 0.60 / length_std 32 / 套话 18% / score 0.96
- 同时跑第 4 章的 Answer Relevance + Faithfulness → 几乎不掉（0.85 → 0.83）
配套行业研究背景：
- "Distinct-n metrics" 来自 Li et al. NAACL 2016 "A Diversity-Promoting Objective Function for Neural Conversation Models"
AnswerDiversityMetric 是对一整批答案打分的，不是 ragas 那种逐条样本的 SingleTurnMetric，塞不进 evaluate()；读者可在 evaluate() 跑完后对 result.to_pandas()["response"] 另算一遍——5 分钟看清"答案是否过于机械"，把 RAG 评测从单维度"对不对"扩展到"用户体验维度"。这是 ragas 工程化对"答案体验"的最关键补充 metric。
11.13 规模化与成本
11.13.1 ragas 评测过程的"成本拆解 + 优化策略"
ragas 评测真正能跑起来的瓶颈不是代码，是钱。成本的第一性变量是"每题打多少次 LLM"——这个是可以直接从 src/ragas/metrics/ 的源码数出来的，而 token 量与单价则完全取决于你自己的 context 长度和 provider 定价，本专栏不给编造的数字。下表只列可从源码核实的调用次数：
| metric | 每题 LLM 调用次数（源码依据） | 优化方向 |
|---|---|---|
Faithfulness |
2——_create_statements 一次 + _create_verdicts 一次（_faithfulness.py） |
换 FaithfulnesswithHHEM 把第二次换成本地 NLI 模型 |
ResponseRelevancy |
strictness 次，默认 3（src/ragas/metrics/_answer_relevance.py:95 的 strictness: int = 3，:146 的 for _ in range(self.strictness) 起 strictness 份并发调用） |
strictness=1 先看方差是否可接受 |
ContextRecall |
1 | 改用 small judge（如 gpt-4o-mini） |
ContextPrecision |
M 次，M = retrieved_contexts 条数（src/ragas/metrics/_context_precision.py:151 的 for context in retrieved_contexts 逐条 verify） |
top_k 从 10 降到 5，调用次数直接减半 |
AnswerCorrectness |
3——response、reference 各拆一次陈述（_create_simplified_statements）+ correctness_prompt 判 TP/FP/FN 一次（src/ragas/metrics/_answer_correctness.py:240-269）；另按默认 weights=[0.75, 0.25] 调 embedding 算语义相似度 |
只要事实性时设 weights=[1.0, 0.0]，省掉 embedding 调用 |
AspectCritic |
strictness 次自洽投票，默认 1（src/ragas/metrics/_aspect_critic.py:101；偶数会被自动 +1 保证能多数表决）；每个 aspect 是一个独立的 metric 实例，挂 A 个 aspect 就是 A 份开销 |
只保留真正会看的 aspect |
把"调用次数 × 你自己的平均 token 数 × 你自己的单价"相乘，才是可信的成本估算——先跑 20 题试验批读真实用量，再线性外推。
下面这段是读者自用的剖析工具，不是 ragas 源码——RagasCostProfiler 以及它的 hook / report 方法都是这里现写的，ragas 仓库里没有同名成员。
import time
from dataclasses import dataclass
from typing import Iterable
from collections import defaultdict
@dataclass
class RagasCostReport:
metric: str
samples: int
total_llm_calls: int
avg_latency_ms: int
avg_input_tokens: int
avg_output_tokens: int
cost_usd: float
class RagasCostProfiler:
"""劫持 ragas 评测，记录每个 metric 的真实 LLM 调用数与 token 量"""
def __init__(self, price_in_per_1k: float = 0.005,
price_out_per_1k: float = 0.015):
self.price_in = price_in_per_1k
self.price_out = price_out_per_1k
self.events: list[dict] = []
def hook(self, metric: str, llm_input: str, llm_output: str,
latency_ms: int):
self.events.append({
"metric": metric,
"in_tokens": len(llm_input) // 4,
"out_tokens": len(llm_output) // 4,
"latency_ms": latency_ms,
})
def report(self) -> list[RagasCostReport]:
by_metric = defaultdict(list)
for e in self.events:
by_metric[e["metric"]].append(e)
reports = []
for m, events in by_metric.items():
n = len(events)
avg_in = sum(e["in_tokens"] for e in events) / n
avg_out = sum(e["out_tokens"] for e in events) / n
cost = (sum(e["in_tokens"] for e in events) / 1000 * self.price_in +
sum(e["out_tokens"] for e in events) / 1000 * self.price_out)
reports.append(RagasCostReport(
metric=m, samples=n,
total_llm_calls=n,
avg_latency_ms=int(sum(e["latency_ms"] for e in events) / n),
avg_input_tokens=int(avg_in),
avg_output_tokens=int(avg_out),
cost_usd=round(cost, 4),
))
return sorted(reports, key=lambda r: -r.cost_usd)flowchart LR
R[ragas 评测] --> P[Profiler hook]
P -->|每次 LLM call| E[event log]
E --> AGG[按 metric 聚合]
AGG -->|cost desc 排序| TOP[最高 cost metric]
TOP --> OPT{优化策略?}
OPT -->|多 prompt 合并| O1[Faithfulness 二合一]
OPT -->|降 model tier| O2[ContextRecall 用 mini]
OPT -->|减 N| O3[AnswerRelevance N=1]
OPT -->|降 top_k| O4[ContextPrecision top_k=5]
style TOP fill:#ffebee
style O1 fill:#e8f5e9
工程实务的 5 条降本经验：
- 计数时把重试算进去——ragas 调 LLM 默认带重试（
add_async_retry，src/ragas/llms/base.py:105），限流或解析失败时实际调用数会多于上表的理论值 - judge 降级 + 抽样验证：用 gpt-4o-mini 做 grader，每 100 题用 4o 抽查 5 题验证一致性
- 关掉不需要的 metric：
evaluate()不传metrics时默认跑answer_relevancy/context_precision/faithfulness/context_recall四个（src/ragas/evaluation.py:158-167），按上表单题就是 3 + M + 2 + 1 次 LLM 调用，只留真正要看的 - 分批提交：
evaluate(batch_size=20)是让Executor每次只提交 20 个任务、分批跑完（src/ragas/executor.py:152），用来控并发和内存，不会把多题合进一次 LLM 调用、也不省 token - 缓存 judge 输出：同一份 (question, answer) 重跑时命中本地缓存（hash key）
具体计算（示意值，按每题 $0.012 假设）：3 metric × 1000 样本 × $0.012/题 ≈ $36/次评测。每周一次的话年成本不到 $2k——相对人工逐条复核，这是典型的"花钱买人力"权衡。
11.13.2 ragas 与 LLM-judge 经济学的演进
ragas 上线时（2023），LLM-judge 的成本是工程顾虑——每条 ragas Faithfulness 就要调 2 次 LLM。此后 judge 可用模型的单价明显下降：gpt-4o 发布时（2024-05）输入价 $5 / M token，2024-08 的 gpt-4o-2024-08-06 快照降到 $2.50 / M；同年推出的 gpt-4o-mini（也是 ragas 0.2.10 llm_factory() 的默认 judge）输入价只有 $0.15 / M（三个价格与本专栏 pin 的 promptfoo 0.106.0 内置价格表一致，src/providers/openai/util.ts:48-67）。具体到你的账单，仍按 §11.13.1 的"调用次数 × token × 单价"去算，单价以 provider 官方定价页当天为准。
工程上的影响：
- 采样率可以提高：同样的预算能覆盖更高比例的在线采样
- 每日全集回归变得可行：中小规模评测集每天全量跑一遍，多数团队负担得起
- 元评测频次可以提高：从季度跑向月度跑靠拢
这种"成本曲线"让 ragas（和所有 LLM-judge 评测）的工程价值持续上升。成本顾虑大幅缓解之后，评测体系建设的主要障碍越来越多地落在"组织决心"上。
11.13.3 ragas 的"评测周期 P50 / P95 / P99 报告"——大数据集评测的稳定性可观测
ragas 跑 1000 题评测时，单题延迟分布可能严重偏态：大多数题几秒内完成，少数题因超时重试、超长 context 拖到几十秒甚至数分钟。如果团队只看总耗时或均值，就会忽略长尾这个关键信号。本节给读者一份 ragas 评测延迟分布 + 异常题诊断方案。
graph LR
A[ragas 跑 1000 题] --> B[单题延迟收集]
B --> C{P50}
B --> D{P95}
B --> E{P99}
B --> F{P100 max}
C & D & E & F --> G[延迟分布报告]
G --> H[识别异常题]
H --> I[1. 极长 prompt]
H --> J[2. judge 模型重试]
H --> K[3. token 超限]
H --> L[4. 网络抖动]
I & J & K & L --> M[根因分类 + 优化]
M --> N[超长题切分 / 重试重排 / 限流]
4 类延迟异常 × 信号 × 优化路径：
| 异常类型 | 信号 | 优化路径 | 收益（示意值） |
|---|---|---|---|
| 极长 prompt | latency > P95 + token > 4k | 切分长 context 或截断 | -60% 延迟 |
| judge 重试 | latency > P95 + retry_count ≥ 2 | 加 timeout / 切换模型 | -40% |
| token 超限 | latency > P95 + truncation 触发 | 调小 chunk size | -50% |
| 网络抖动 | latency > P95 + 高 jitter std | 加 backoff retry | -30% |
配套实现：评测周期 P50/P95/P99 报告器：
import statistics
from dataclasses import dataclass, field
from typing import Literal
LatencyAnomalyKind = Literal["long_prompt", "judge_retry", "token_truncation",
"network_jitter", "unknown"]
@dataclass
class SampleEvalLatency:
sample_id: str
latency_ms: int
prompt_token_count: int
judge_retry_count: int = 0
truncated: bool = False
@dataclass
class RagasLatencyReporter:
samples: list[SampleEvalLatency] = field(default_factory=list)
long_prompt_threshold: int = 4000
p95_anomaly_multiplier: float = 2.0 # P95 比 P50 高 2x 视为异常分布
def percentile(self, values: list[int], p: float) -> int:
if not values: return 0
sorted_v = sorted(values)
idx = int(len(sorted_v) * p)
return sorted_v[min(idx, len(sorted_v) - 1)]
def report(self) -> dict:
if not self.samples:
return {"total": 0}
latencies = [s.latency_ms for s in self.samples]
p50 = self.percentile(latencies, 0.50)
p95 = self.percentile(latencies, 0.95)
p99 = self.percentile(latencies, 0.99)
p100 = max(latencies)
return {
"total_samples": len(self.samples),
"p50_ms": p50, "p95_ms": p95, "p99_ms": p99, "p100_ms": p100,
"p95_p50_ratio": round(p95 / max(p50, 1), 2),
"distribution_health": ("severely_skewed" if p95 / max(p50, 1) > 4
else "skewed" if p95 / max(p50, 1) > 2
else "healthy"),
"total_wall_time_seconds": sum(latencies) / 1000,
}
def identify_anomalies(self) -> list[dict]:
if len(self.samples) < 20: return []
latencies = [s.latency_ms for s in self.samples]
p95 = self.percentile(latencies, 0.95)
anomalies = []
for s in self.samples:
if s.latency_ms <= p95: continue
kind: LatencyAnomalyKind = "unknown"
if s.prompt_token_count >= self.long_prompt_threshold:
kind = "long_prompt"
elif s.judge_retry_count >= 2:
kind = "judge_retry"
elif s.truncated:
kind = "token_truncation"
else:
kind = "network_jitter"
anomalies.append({
"sample_id": s.sample_id, "latency_ms": s.latency_ms,
"vs_p95": s.latency_ms - p95,
"kind": kind,
})
return sorted(anomalies, key=lambda a: a["latency_ms"], reverse=True)
def optimization_recommendations(self) -> list[str]:
from collections import Counter
anomalies = self.identify_anomalies()
if not anomalies: return ["延迟分布健康，无需优化"]
kind_counts = Counter(a["kind"] for a in anomalies)
recs = []
for kind, count in kind_counts.most_common():
if kind == "long_prompt":
recs.append(f"{count} 例长 prompt 异常 → 切分 context 或截断 > 4k tokens")
elif kind == "judge_retry":
recs.append(f"{count} 例 judge 重试 → 加 timeout = 30s / 切换备用 judge")
elif kind == "token_truncation":
recs.append(f"{count} 例 token 超限 → 调小 chunk_size 或扩 max_tokens")
elif kind == "network_jitter":
recs.append(f"{count} 例网络抖动 → 加 exponential backoff + retry 3 次")
return recs
def adaptive_concurrency_suggestion(self) -> dict:
report = self.report()
if report["distribution_health"] == "healthy":
return {"recommended_concurrency": 50, "reason": "延迟均匀，可高并发"}
if report["distribution_health"] == "skewed":
return {"recommended_concurrency": 25, "reason": "中度偏态，中等并发避免长尾拖死"}
return {"recommended_concurrency": 10, "reason": "严重偏态，低并发 + queueing"}举例（示意数据，需用你自己的数据复算）：某团队 ragas 跑 1000 题：
- p50 4500ms / p95 20000ms / p99 65000ms / p100 220000ms
- p95/p50 ≈ 4.4 → 严重偏态（代码阈值 > 4）
- identify_anomalies: 50 例异常，其中 18 long_prompt / 12 judge_retry / 10 token_truncation / 10 网络
- recommendations 给出 4 条具体优化
- adaptive_concurrency_suggestion → 建议把并发从 10 提到 25，改完重跑对比
配套行业研究背景：
- "Latency percentile reporting" 来自 Tail at Scale (Dean & Barroso 2013)
- "P99 vs mean for ML systems" 来自 Google SRE Workbook
- "Adaptive concurrency control" 来自 Netflix 的 concurrency-limits 库（2018，把 TCP Vegas 的拥塞控制搬到 RPC 并发度上）——不是 Hystrix：Hystrix 给的是固定大小的线程池隔离与熔断，并发度并不自适应
11.13.4 ragas 与"内部 RAG 平台 multi-tenant"集成——一个评测引擎服务全公司
成熟公司常见架构：内部 RAG 平台被多个业务团队（客服 / HR / 知识库 / 法务）共用。如果每个团队各跑各的 ragas → 资源浪费 + 评测口径不一致 + 跨团队对比无意义。本节给读者一份"multi-tenant ragas 服务化"工程方案——把 ragas 升级为支持租户隔离 / 配额 / 计费 / SLA 的内部评测平台。
graph LR
A[业务团队 1<br/>客服] --> B[ragas 服务网关]
A2[业务团队 2<br/>HR] --> B
A3[业务团队 3<br/>法务] --> B
A4[业务团队 N] --> B
B --> C[认证 + 配额校验]
C --> D[租户隔离]
D --> E[ragas 执行池]
E --> F[结果存储<br/>按 tenant 分区]
E --> G[配额扣减]
F --> H[每团队独立 dashboard]
G --> I[月度账单 chargeback]
F --> J[全公司 metric 一致性看板]
multi-tenant ragas 5 个核心工程关切：
| 关切 | 单租户实现 | multi-tenant 实现 | 必要性 |
|---|---|---|---|
| 配额管理 | 无限制 | 按团队配额 / 月度上限 | 防止单团队烧爆全公司预算 |
| 数据隔离 | 全局共享 | tenant_id 分区 + RBAC | GDPR / 内部合规 |
| metric 版本一致性 | 各自升级 | 中央升级 + 灰度通知 | 跨团队对比可比 |
| 计费 chargeback | 不计费 | token 用量 → 财务系统 | 平台团队成本可分摊 |
| SLA 保障 | 自助维护 | 平台 99.9% / RT < 5s | 业务团队不被评测拖垮 |
配套实现：multi-tenant ragas 服务网关：
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional, Literal
TenantTier = Literal["free", "standard", "enterprise"]
QUOTA_TABLE: dict[TenantTier, dict] = {
"free": {"monthly_evals": 100, "concurrent": 1, "support_sla_hr": 72},
"standard": {"monthly_evals": 5000, "concurrent": 5, "support_sla_hr": 24},
"enterprise": {"monthly_evals": 50_000, "concurrent": 20, "support_sla_hr": 4},
}
@dataclass
class TenantConfig:
tenant_id: str
team_name: str
tier: TenantTier
cost_center: str # 财务部门归属
allowed_metrics: list[str] = field(default_factory=lambda: ["faithfulness", "answer_relevance"])
api_keys: list[str] = field(default_factory=list)
@dataclass
class EvalRequest:
tenant_id: str
api_key: str
dataset_size: int
metrics: list[str]
estimated_tokens: int
@dataclass
class TenantUsageRecord:
tenant_id: str
timestamp: datetime
evals_count: int
tokens_consumed: int
cost_usd: float
success: bool
@dataclass
class MultiTenantRagasGateway:
tenants: dict[str, TenantConfig] = field(default_factory=dict)
usage: list[TenantUsageRecord] = field(default_factory=list)
cost_per_million_tokens: float = 5.0
def authenticate(self, req: EvalRequest) -> Optional[TenantConfig]:
t = self.tenants.get(req.tenant_id)
if not t or req.api_key not in t.api_keys:
return None
return t
def quota_check(self, tenant: TenantConfig, req: EvalRequest) -> dict:
quota = QUOTA_TABLE[tenant.tier]
used_this_month = self._monthly_usage(tenant.tenant_id)
remaining = quota["monthly_evals"] - used_this_month
forbidden = [m for m in req.metrics if m not in tenant.allowed_metrics]
return {
"passed": remaining >= req.dataset_size and not forbidden,
"monthly_remaining": remaining,
"would_consume": req.dataset_size,
"forbidden_metrics": forbidden,
}
def _monthly_usage(self, tenant_id: str) -> int:
cutoff = datetime.now() - timedelta(days=30)
return sum(u.evals_count for u in self.usage
if u.tenant_id == tenant_id and u.timestamp >= cutoff)
def execute(self, req: EvalRequest) -> dict:
tenant = self.authenticate(req)
if not tenant:
return {"status": "401", "reason": "认证失败"}
check = self.quota_check(tenant, req)
if not check["passed"]:
return {"status": "429", "reason": "配额或权限不足", **check}
# 实际执行 ragas...（这里省略）
cost = req.estimated_tokens / 1_000_000 * self.cost_per_million_tokens
self.usage.append(TenantUsageRecord(
tenant_id=req.tenant_id, timestamp=datetime.now(),
evals_count=req.dataset_size, tokens_consumed=req.estimated_tokens,
cost_usd=cost, success=True,
))
return {"status": "200", "cost_usd": cost,
"remaining_quota": check["monthly_remaining"] - req.dataset_size}
def monthly_chargeback_report(self) -> dict:
cutoff = datetime.now() - timedelta(days=30)
by_team: dict[str, dict] = {}
for u in self.usage:
if u.timestamp < cutoff: continue
t = self.tenants.get(u.tenant_id)
if not t: continue
cost_center = t.cost_center
entry = by_team.setdefault(cost_center, {"evals": 0, "cost_usd": 0.0, "teams": []})
entry["evals"] += u.evals_count
entry["cost_usd"] += u.cost_usd
if t.team_name not in entry["teams"]:
entry["teams"].append(t.team_name)
return by_team举例（示意数据，需用你自己的数据复算）：某 1000 人公司接入 multi-tenant ragas 平台 6 个月：
- 12 个业务团队接入，月评测量从 5K → 80K
- 客服团队（enterprise）月用 30K，HR（standard）月用 3K，法务（standard）月用 4.5K
- monthly_chargeback_report 自动出账：客服 cost_center "运营" 收 $1500/月、HR "人资" 收 $150/月
11.13.5 ragas 与"评测结果导出 → BI / 数据仓库"——把评测信号接入公司分析栈
ragas 跑完只输出 JSON / Pandas DataFrame，但企业内部的核心分析栈是 BI（Tableau / Looker / Metabase）/ 数据仓库（Snowflake / BigQuery / ClickHouse）。如果评测信号停在 ragas 内部，BI 仪表盘看不到、产品经理看不到、CFO 月报看不到 — 评测体系再好也只在工程内部循环。本节给读者一份「ragas → DW → BI」的工程化导出方案。
graph LR
A[ragas evaluate] --> B[内部 DataFrame]
B --> C{导出 ETL}
C --> D[Snowflake / BigQuery]
C --> E[ClickHouse]
D & E --> F[标准化表 schema]
F --> G[run_id<br/>commit_sha<br/>metric<br/>score<br/>tenant<br/>ts]
G --> H[BI dashboard]
H --> I[Tableau 评测趋势图]
H --> J[Looker 业务对照]
H --> K[Metabase 周报]
I & J & K --> L[全公司可见的评测信号]
4 类输出目的地 × schema 字段 × 用途：
| 目的地 | 必填字段 | 主要消费者 | 频率 |
|---|---|---|---|
| Snowflake / BigQuery | run_id / commit_sha / metric / score / tenant / ts | 数据团队 | 每次 evaluate |
| ClickHouse | + sample_id / latency_ms | 平台团队 | 每次 evaluate |
| Tableau | 同上 | 产品经理 / 业务老板 | 每周 refresh |
| Metabase 月报 | 聚合 metric | 高管 | 每月 |
配套实现：ragas 结果 → DW 导出器：
import json
from dataclasses import dataclass, field
from datetime import datetime
from typing import Literal
DWBackend = Literal["snowflake", "bigquery", "clickhouse", "csv_export"]
@dataclass
class RagasRunMetadata:
run_id: str
commit_sha: str
branch: str
triggered_by: str
tenant_id: str
yaml_or_dataset_path: str
created_at: datetime
@dataclass
class RagasMetricRow:
"""每条 metric 一行的标准化 schema"""
run_id: str
commit_sha: str
branch: str
tenant_id: str
metric_name: str
score: float
sample_id: str | None
latency_ms: int | None
created_at: datetime
@dataclass
class RagasDWExporter:
backend: DWBackend = "csv_export"
target_table: str = "evals_ragas_results"
target_path: str = "evals_export.csv"
def to_rows(self, meta: RagasRunMetadata,
ragas_result: dict) -> list[RagasMetricRow]:
"""ragas evaluate() 返回 dict, 拆成 per-metric per-sample 行"""
rows = []
# ragas 标准输出有 metrics dict + per_sample list
for metric_name, score in ragas_result.get("aggregate", {}).items():
rows.append(RagasMetricRow(
run_id=meta.run_id, commit_sha=meta.commit_sha,
branch=meta.branch, tenant_id=meta.tenant_id,
metric_name=metric_name, score=score,
sample_id=None, latency_ms=None,
created_at=meta.created_at,
))
for sample in ragas_result.get("per_sample", []):
for metric, score in sample.get("metrics", {}).items():
rows.append(RagasMetricRow(
run_id=meta.run_id, commit_sha=meta.commit_sha,
branch=meta.branch, tenant_id=meta.tenant_id,
metric_name=metric, score=score,
sample_id=sample.get("id"),
latency_ms=sample.get("latency_ms"),
created_at=meta.created_at,
))
return rows
def export(self, meta: RagasRunMetadata, ragas_result: dict) -> dict:
rows = self.to_rows(meta, ragas_result)
if self.backend == "csv_export":
self._to_csv(rows)
elif self.backend == "snowflake":
sql = self._snowflake_insert_sql(rows)
return {"sql_preview": sql[:200], "rows_count": len(rows)}
elif self.backend == "clickhouse":
return self._clickhouse_insert(rows)
return {"rows_inserted": len(rows), "target": self.target_table}
def _to_csv(self, rows: list[RagasMetricRow]):
import csv
from pathlib import Path
path = Path(self.target_path)
write_header = not path.exists()
with path.open("a", newline="") as f:
writer = csv.writer(f)
if write_header:
writer.writerow(["run_id", "commit_sha", "branch", "tenant_id",
"metric_name", "score", "sample_id",
"latency_ms", "created_at"])
for r in rows:
writer.writerow([r.run_id, r.commit_sha, r.branch, r.tenant_id,
r.metric_name, r.score, r.sample_id,
r.latency_ms, r.created_at.isoformat()])
def _snowflake_insert_sql(self, rows: list[RagasMetricRow]) -> str:
values = ",\n".join(
f"('{r.run_id}', '{r.commit_sha}', '{r.branch}', '{r.tenant_id}', "
f"'{r.metric_name}', {r.score}, "
f"{'NULL' if r.sample_id is None else repr(r.sample_id)}, "
f"{'NULL' if r.latency_ms is None else r.latency_ms}, "
f"'{r.created_at.isoformat()}')"
for r in rows
)
return f"INSERT INTO {self.target_table} VALUES\n{values};"
def _clickhouse_insert(self, rows: list[RagasMetricRow]) -> dict:
# 简化：用 HTTP API 形式
return {"backend": "clickhouse", "rows": len(rows),
"table": self.target_table}
@dataclass
class BIDashboardSchemaDoc:
"""生成 BI tool 可消费的 schema 文档"""
@staticmethod
def render() -> str:
return """
| Field | Type | Description | BI Use |
|-------|------|-------------|--------|
| run_id | VARCHAR | 唯一 run 标识 | 时间维度 |
| commit_sha | VARCHAR | git commit | drill-down |
| branch | VARCHAR | branch name | filter |
| tenant_id | VARCHAR | 业务团队 | dimension |
| metric_name | VARCHAR | faithfulness 等 | dimension |
| score | FLOAT | 0-1 | measure |
| sample_id | VARCHAR | 题号 | drill-down |
| latency_ms | INT | 延迟 | measure |
| created_at | TIMESTAMP | run 时间 | time series |
"""举例（示意数据，需用你自己的数据复算）：某团队接入 Snowflake 导出 6 个月：
- 累积 80 万行评测数据
- BI 团队基于 evals_ragas_results 表做了 5 个 dashboard：metric 30 天趋势 / 各业务 tenant 对照 / commit-level drill-down / 异常告警面板 / 月度高管报告
- 产品经理每周看 dashboard，主动发现「上周 Faithfulness 跌 2pp 主要在 tenant=customer-service」
11.14 落地：接入位置与版本迁移
11.14.1 ragas 在工业 RAG 系统的"接入位置"
工业 RAG 系统通常有多个层级，ragas 在其中的"接入位置"决定了评测能力。看一份典型的接入图：
flowchart TB User[用户] --> App[应用层] App --> Cache[缓存层] Cache -->|miss| Retrv[Retriever 层] Retrv --> VDB[(Vector DB)] Retrv --> Reranker[Reranker] Reranker --> Gen[Generator 层] Gen --> Post[后处理 / Citation] Post --> User Eval1[ragas 评测点 A<br/>retriever 输出] -.接入.-> Retrv Eval2[ragas 评测点 B<br/>generator 输入输出] -.接入.-> Gen Eval3[ragas 评测点 C<br/>端到端] -.接入.-> User style Eval1 fill:#dbeafe style Eval2 fill:#dcfce7 style Eval3 fill:#fef3c7
每个评测点的工程意义：
- A 点：评测 retriever 单独表现（Context Recall / Precision），不受 generator 干扰
- B 点：评测 generator 给定 context 时的表现（Faithfulness / Relevance），不受 retriever 错误污染
- C 点：评测端到端用户体验（Answer Correctness / Satisfaction）
工业实务：3 个点都接，互相对照。如果 A 点高分但 C 点低分 → generator 出问题；如果 A 点低分 → retriever 出问题。这种"多点评测 + 对照分析"是 RAG 系统调优的最有效路径。
ragas 的 metric 体系天然支持这种多点接入——ContextRecall / Faithfulness / AnswerCorrectness 各自工作在不同层。理解这种分层接入是把 ragas 用好的关键认知。
11.14.2 ragas 的"配置最佳实践"汇总
把工业团队的常见做法汇总成一份配置模板：
from ragas import evaluate
from ragas.metrics import (
Faithfulness, AnswerRelevancy,
ContextRecall, ContextPrecision,
AnswerCorrectness, NoiseSensitivity,
)
from ragas.llms import LangchainLLMWrapper
from langchain_anthropic import ChatAnthropic
# 1. 选择 judge 模型 — 强烈建议与被测模型不同家族
judge_llm = LangchainLLMWrapper(
ChatAnthropic(
model="claude-sonnet-5", # 示例；按官方模型页选现行型号
temperature=0, # 必须 0
max_retries=3,
)
)
# 2. 配置 metric 集合 — 主指标 + 辅助指标
metrics = [
Faithfulness(llm=judge_llm), # 主指标
AnswerCorrectness(llm=judge_llm), # 主指标
ContextRecall(llm=judge_llm), # retriever 评测
ContextPrecision(llm=judge_llm), # retriever 评测
AnswerRelevancy(llm=judge_llm), # 辅助
NoiseSensitivity(llm=judge_llm), # 鲁棒性
]
# 3. 跑评测 + 错误处理
try:
result = evaluate(
dataset=ds,
metrics=metrics,
raise_exceptions=False, # 单条失败不中断全集
)
except Exception as e:
log.error(f"Eval failed: {e}")
# 4. 多视角分析
df = result.to_pandas()
print("Mean per metric:")
print(df.describe())
print("\nWorst 10 cases:")
print(df.nsmallest(10, "faithfulness"))
print("\nBy category:")
# SingleTurnSample 会丢掉 schema 外的字段，category 要按行号从原始数据自己并回来
df["category"] = categories
print(df.groupby("category").mean(numeric_only=True))这一份模板是 ragas 工业级使用的"50 行起步"。涵盖：
- judge 模型显式指定（不用默认 OpenAI，避免家族 bias）
- temperature=0 强制锁定
- 6 个 metric 同时评（覆盖不同维度）
- 错误处理（单条失败不中断）
- 多视角分析（mean / worst / category）
直接拿这份模板用，替换 ds 和与之按行对齐的 categories 列表即可，省掉从零搭脚手架的一轮。
11.14.3 对工业团队的具体建议：从 ragas 0.x 到内部 fork
ragas 项目本身在 2024-2026 年快速演化（0.1 → 0.2 → 0.3），API 时有 break change。工业团队基于 ragas 上线的应用面临一个工程问题：升级 ragas 还是 fork 锁版？
实战建议：
| 团队规模 | 建议做法 | 原因 |
|---|---|---|
| < 5 人 / 早期 | 跟主线升级 | 节省维护成本 |
| 5-20 人 / 上线 | 锁定 minor 版本 | 减少 break risk |
| 20+ 人 / 规模化 | fork 内部维护 | 完全可控 |
fork 内部维护时的关键工程纪律：
- 保留 ragas 的核心抽象（
Metric,Sample,PydanticPrompt），不要重写 - 自定义 metric 进自家
internal_metrics/目录，与 upstream 隔离 - 定期（季度）从 upstream rebase / cherry-pick bug fix
- 自家修改提 PR 给 ragas 上游（让维护者帮你 review）
这种"fork + 持续 upstream 同步"是开源工程领域的标准模式（Linux 内核、Chromium、TensorFlow 都这么干）。比"完全独立"或"完全跟主线"都更稳健。
11.14.4 ragas 0.2 与 0.1 迁移指南——避免 API 重写陷阱
ragas 在 2024-10（v0.2.0，2024-10-15）从 0.1.x 大跨步到 0.2.x，引入了 EvaluationDataset / SingleTurnSample / MultiTurnSample 三大抽象。许多团队的旧代码在升级时会大面积出错。下面是一份完整迁移指南：
# === ragas 0.1.x 旧写法（v0.2.10 仍兼容：HF Dataset 的旧列名会被自动转成新列名）===
from ragas import evaluate
from datasets import Dataset
from ragas.metrics import faithfulness, answer_relevancy
# 0.1.x 用 HuggingFace Dataset
old_data = Dataset.from_dict({
"question": ["Q1", "Q2"],
"answer": ["A1", "A2"],
"contexts": [["c1"], ["c2"]],
"ground_truth": ["GT1", "GT2"],
})
result = evaluate(old_data, metrics=[faithfulness, answer_relevancy])
# === ragas 0.2.x 新 API（推荐）===
from ragas import evaluate
from ragas.dataset_schema import EvaluationDataset, SingleTurnSample
from ragas.metrics import Faithfulness, AnswerRelevancy
samples = [
SingleTurnSample(
user_input="Q1",
response="A1",
retrieved_contexts=["c1"],
reference="GT1",
),
SingleTurnSample(
user_input="Q2",
response="A2",
retrieved_contexts=["c2"],
reference="GT2",
),
]
new_data = EvaluationDataset(samples=samples)
result = evaluate(new_data, metrics=[Faithfulness(), AnswerRelevancy()])字段重命名映射表：
| 0.1.x 字段 | 0.2.x 字段 | 变更原因 |
|---|---|---|
question |
user_input |
与多轮对话语义统一 |
answer |
response |
区分用户问题 vs 模型响应 |
contexts |
retrieved_contexts |
显式 retrieved（区分 reference） |
ground_truth |
reference |
与 NLP 论文术语对齐 |
用预实例化单例 faithfulness |
推荐 Faithfulness() 自建实例（单例仍导出） |
支持 metric 实例配置 |
| Dataset (HF) | 推荐 EvaluationDataset（HF Dataset 仍可传，evaluate() 入口会先把旧列名转成新列名） |
内置 schema 验证 |
import json
from dataclasses import dataclass
from pathlib import Path
class RagasMigrationHelper:
"""从 0.1.x 数据格式自动迁移到 0.2.x"""
FIELD_MAPPING = {
"question": "user_input",
"answer": "response",
"contexts": "retrieved_contexts",
"ground_truth": "reference",
"ground_truths": "reference", # 处理 plural 别名
}
METRIC_RENAMES = {
"faithfulness": "Faithfulness",
"answer_relevancy": "AnswerRelevancy",
"context_recall": "ContextRecall",
"context_precision": "ContextPrecision",
"answer_correctness": "AnswerCorrectness",
}
def migrate_jsonl(self, old_path: Path, new_path: Path) -> dict:
"""逐条迁移评测集 jsonl"""
migrated_count = 0
skipped = []
with old_path.open() as fin, new_path.open("w") as fout:
for i, line in enumerate(fin):
try:
old_record = json.loads(line)
new_record = self._migrate_record(old_record)
fout.write(json.dumps(new_record, ensure_ascii=False) + "\n")
migrated_count += 1
except (json.JSONDecodeError, KeyError) as e:
skipped.append({"line": i, "error": str(e)})
return {
"migrated": migrated_count,
"skipped": skipped,
"field_mapping_used": self.FIELD_MAPPING,
}
def _migrate_record(self, old: dict) -> dict:
new = {}
for k, v in old.items():
new_key = self.FIELD_MAPPING.get(k, k)
new[new_key] = v
# ground_truths 是 list 类型时取第一个
if "reference" in new and isinstance(new["reference"], list):
new["reference"] = new["reference"][0] if new["reference"] else ""
return new
def migrate_metric_imports(self, code: str) -> str:
"""识别旧 metric 引用，给出修改建议"""
suggestions = []
for old, new in self.METRIC_RENAMES.items():
if f"from ragas.metrics import {old}" in code:
suggestions.append(
f"将 'from ragas.metrics import {old}' 改为 "
f"'from ragas.metrics import {new}'，"
f"调用时改 {old} → {new}()"
)
return "\n".join(suggestions) or "未检测到需迁移的 metric 引用"flowchart LR OLD[ragas 0.1.x 代码] --> M[MigrationHelper] M --> F1[字段重命名<br/>question → user_input] M --> F2[metric 函数 → class] M --> F3[HF Dataset → EvaluationDataset] F1 --> NEW[ragas 0.2.x 代码] F2 --> NEW F3 --> NEW NEW --> CI[CI 验证迁移正确] style NEW fill:#e8f5e9
工程实务的 4 个迁移经验：
- migrate jsonl 评测集与代码分两步：先迁数据 + 跑通 → 再迁代码 + 跑通
- 保留 0.1.x 代码 fallback：用
try / except ImportError保护关键 import - 测试集先小批量验证：100 题验证字段映射正确，再扫全集
- metric 配置随 class 化也变：0.1.x 全局 metric 配，0.2.x 实例配置（temperature 等）
3 类常见迁移坑：
| 坑 | 错误 | 修法 |
|---|---|---|
| ground_truth 是 list | 0.2.x 只接受 str | 取第一个或 \n join |
| 自定义 metric 派生 | 0.1.x 实现的是 _ascore(row, callbacks)，0.2.x 要实现 _single_turn_ascore(sample, callbacks)（旧的 score() 已标 deprecated） |
看新 base class 签名 |
| jsonl 评测集沿用旧字段名 | EvaluationDataset.from_jsonl 不做旧列名转换，question 等字段被丢弃，evaluate() 时报缺列 ValueError |
用本节 helper 转换 |
11.14.5 ragas 的"非 RAG"用法：超出名字的扩展
仓库名叫"RAG Assessment"（ragas），但本章 §11.6 的 metric 表已经显示它早就超出 RAG。看几个工业上常见的"非 RAG"用法：
- 纯 LLM 内容审核：不接 retriever，把 retrieved_contexts 设为"内容政策文档"，跑 Faithfulness 评测——本质成了"内容是否符合政策"的 grader
- 代码生成评测：retrieved_contexts 设为"项目代码片段"，response 设为"模型生成的代码补丁"，Faithfulness 评测"补丁是否符合项目风格"
- SQL 查询验证：用
LLMSQLEquivalencemetric（_sql_semantic_equivalence.py）评测两个 SQL 是否等价 - 多模态 RAG：
MultiModalFaithfulness处理图文混合场景
工程团队的主要 takeaway：ragas 名字虽然窄，但抽象足够通用。不要因为名字限制使用——只要任务能转换成 (question, response, context, reference) 四元组，ragas 都能用上。
这种"通用四元组"是 LLM 评测领域逐渐收敛的标准 schema——OpenAI evals 的 (input, ideal)、promptfoo 的 (vars, output)、lm-eval 的 (doc, response) 在四元组语义下都能对齐。理解这一点，团队选工具时不会被 marketing 标签困住。
11.15 工程角色与长期视角
11.15.1 ragas 在 RAG 工程师"日常工作流"中的位置
ragas 在 RAG 工程师一周工作流里的位置（示意）：
Monday: 早上看周末跑的全集评测报告 (ragas 输出)
改 prompt 应对失败 case
Tuesday: 跑 ragas 验证改动效果
Wednesday: hard case mining (从在线 trace 抓低分 trace)
标注 + 入 ragas 评测集
Thursday: 调 retriever 参数, 跑 ragas Recall 评测
Friday: PR review + 跑 ragas 子集做合并前评测
发布周度评测报告11.15.2 ragas 在 LLM 应用工程"分工"中的位置
回顾 LLM 应用工程的整体分工：
flowchart TB PM[PM<br/>定需求] --> App[应用工程师<br/>实现 RAG] App --> Eval[评测工程师<br/>跑 ragas + 元评测] Eval --> SRE[SRE<br/>监控 + 告警] SRE --> User[用户] User -->|反馈| PM Eval -.发现 hard case 反哺.-> App App -.新功能 trigger.-> Eval style Eval fill:#fef3c7
ragas 在这张图中完全围绕"评测工程师"角色展开——它是评测工程师的核心工具。其他角色与 ragas 的接触：
- PM：看 ragas 输出的 dashboard 决定是否上线
- 应用工程师:跑 ragas 验证自己的改动
- SRE：监控 ragas 在生产采样的指标曲线
11.15.3 ragas 给"评测工程师"的核心工具箱定位
整个评测工程师的工具箱里，ragas 占据"主力 metric 库"的位置。一份角色定位：
工具箱里的工具:
- 黄金集维护: 自家维护 (第 3 章)
- 规则判分: 自家代码 (第 5 章)
- LLM-judge prompt: ragas (第 11 章) + promptfoo (第 12 章)
- Trace 系统: LangSmith / Langfuse (第 17 章)
- CI 集成: promptfoo (第 18 章)
- 红队评测: garak / PyRIT (第 16 章)
- benchmark 跑分: lm-eval (第 10 章)ragas 的位置是"高级 LLM-judge metric"——专门提供 RAG / Agent 等场景的工业级 metric 实现。它不替代其他工具，而是与它们组合使用。
11.15.4 ragas 与 RAG 工程演进的"水涨船高"
最后一个观察——ragas 的发展速度正在被 RAG 工程演进推着走。RAG 系统从 2023 简单"retriever + generator"演化到 2026 复杂的"agentic RAG / GraphRAG / multi-modal RAG"，ragas 的 metric 体系也跟着扩展：
2023: Faithfulness, Answer Relevance, Context Recall, Context Precision, AnswerCorrectness ...
2024: + NoiseSensitivity, ToolCallAccuracy, AgentGoalAccuracy, TopicAdherence, Multi-Modal ...
2025-01 的 v0.2.10: 共 36 个 metric 类这种"工具跟着应用演化"的模式让 ragas 始终保持工业相关性。但也带来工程上的提醒——不要用过时版本。一个 2024 年上半年装的 ragas 0.1.x 就没有 0.2 才有的 Agent / 多模态 metric。
工程实务：
- ragas 升级跟主线（如自托管，每季度评估升级）
- 关注 ragas changelog（GitHub releases）
- 新 metric 出现时评估"我们业务能用上吗"
11.15.5 ragas 的"长期演进史"——3 年 7 次重大版本的工程教训
ragas 从 2023-05 第一次发布到 2026 已 3 年——其版本演进史本身就是 LLM 评测工具发展的缩影。下面给出工程师视角的"演进时间线 + 教训"：
gantt title ragas 3 年演进时间线 dateFormat YYYY-MM section 0.0.x 概念验证 :a1, 2023-05, 270d section 0.1.x HF Dataset API :a2, after a1, 180d 4 核心 metric 稳定 :a3, after a2, 90d section 0.2.x EvaluationDataset 重构 :b1, 2024-10, 60d Multi-turn + Agent 支持 :b2, after b1, 120d section 0.3.x / 0.4.x ragas.experimental 模块 :c1, 2025-07, 90d metrics.collections 新接口 :c2, 2025-10, 120d
| 版本 | 时点 | 核心创新 | 突破 / 教训 |
|---|---|---|---|
| 0.0.x | 2023-Q2 | 4 个 RAG metric | 概念验证，API 不稳 |
| 0.1.0 | 2024-02 | 0.1 系列 API 定型 | 标准化输入接口 |
| 0.1.x | 2024-03 | LangChain / LangSmith integrations（#795） | 与生态深度绑定 |
| 0.2.0 | 2024-10 | EvaluationDataset 重构 + Multi-turn / Agent metric | breaking change，痛但值得 |
| 0.2.10 | 2025-01 | 本专栏 pin 版本 | —— |
| 0.3.0 | 2025-07 | 引入 ragas.experimental |
—— |
| 0.3.7 / 0.4.0 | 2025-10 / 2025-12 | ragas.metrics.collections 新接口（见《RAG 工程与检索系统设计》第 20 章） |
旧路径开始告警 |
| 1.0 | 未发布 | 移除旧 ragas.metrics 路径（v0.4.3 的预告，属预测） |
—— |
from dataclasses import dataclass
from typing import Iterable
@dataclass
class RagasVersionMilestone:
version: str
release_quarter: str
core_innovation: str
breaking_changes: list[str]
lesson_learned: str
class RagasEvolutionLessons:
"""记录 ragas 演进给团队的工程启示"""
MILESTONES = [
RagasVersionMilestone(
"0.0.x", "2023-Q2", "4 个 metric 概念验证",
breaking_changes=[],
lesson_learned="开源工具早期不要套到生产 - API 不稳",
),
RagasVersionMilestone(
"0.1.0", "2024-Q1", "0.1 系列 API 定型",
breaking_changes=["0.0.x 时期接口频繁调整"],
lesson_learned="跟开源生态对齐能 10x 加速",
),
RagasVersionMilestone(
"0.2.0", "2024-Q4", "EvaluationDataset + Sample 重构",
breaking_changes=["几乎所有字段 rename",
"推荐自建 metric 实例"],
lesson_learned="大版本破坏性升级痛但必要 - 用 §11.14.4 helper",
),
RagasVersionMilestone(
"0.2.x", "2024-Q4", "Multi-turn + Agent metric",
breaking_changes=["MultiTurnSample 新抽象"],
lesson_learned="评测工具必须跟模型能力同步演化",
),
]
def lessons_for_team(self) -> dict:
return {
"总教训": "评测工具是流变的 - 团队必须跟版本走",
"关键策略": [
"锁版本但季度评估升级",
"升级走 §11.14.4 migration helper",
"保留 fallback API 6 个月",
"新功能等 .1 / .2 minor 版再上",
],
}工程实务的 4 条 ragas 演进教训：
- 开源工具早期不要赌生产：0.0.x / 0.1.x 都不稳，0.2 才适合生产
- breaking change 来时不要拖：拖到下次 breaking 累积 2x 痛苦
- 跟踪上游 release notes：每月看一次 ragas changelog
- 保留 0.1.x fallback：万一 0.2 严重 bug 能回退
3 类看 ragas 演进的视角：
| 视角 | 启示 |
|---|---|
| 用户视角 | 锁版本 + 季度升级窗口 |
| 贡献者视角 | 给开源工具贡献 PR 时要"提案 + 等待"非"立即 merge" |
| 工程主管视角 | 别赌单一开源依赖 - 准备 fallback |
11.15.6 ragas 与"测试驱动 prompt engineering"——评测信号反向驱动 prompt 改进
一个值得尝试的方向：用评测分数直接驱动 prompt 改进——形成"prompt → ragas → 改 prompt → 再 ragas"的快速闭环。下面给出工程化实现：
import asyncio
from dataclasses import dataclass
from typing import Iterable, Awaitable, Callable
@dataclass
class PromptIterationResult:
iteration: int
prompt_version: str
metric_scores: dict[str, float]
weakest_metric: str
suggested_prompt_change: str
overall_score: float
class PromptDrivenByRagas:
"""ragas 评测 → 自动建议 prompt 改进 → 重测"""
METRIC_TO_PROMPT_FIX = {
"faithfulness": (
"在 prompt 中加：'仅基于给定 context 回答，"
"不要使用 context 之外的信息'"
),
"answer_relevancy": (
"在 prompt 中加：'确保回答直接针对用户问题，"
"不要展开无关话题'"
),
"context_recall": (
"改 retrieval：top_k 增加 / 改 embedding"
),
"context_precision": (
"加 reranker / 减 top_k 提高精度"
),
"answer_correctness": (
"在 prompt 中加：'回答必须事实准确，不确定时说不知道'"
),
}
def __init__(self, ragas_evaluator,
prompt_modifier: Callable[[str, str], str]):
self.evaluator = ragas_evaluator
self.modify = prompt_modifier
async def iterate(self, initial_prompt: str,
max_iterations: int = 5,
target_overall: float = 0.85
) -> list[PromptIterationResult]:
history = []
current_prompt = initial_prompt
for i in range(max_iterations):
scores = await self.evaluator.evaluate_with_prompt(current_prompt)
overall = sum(scores.values()) / max(len(scores), 1)
# 找最弱 metric
weakest = min(scores.items(), key=lambda x: x[1])[0]
suggested = self.METRIC_TO_PROMPT_FIX.get(
weakest, "review prompt manually")
history.append(PromptIterationResult(
iteration=i,
prompt_version=f"v{i}",
metric_scores=scores,
weakest_metric=weakest,
suggested_prompt_change=suggested,
overall_score=round(overall, 3),
))
if overall >= target_overall:
break
# 应用建议改 prompt
current_prompt = self.modify(current_prompt, suggested)
return historyflowchart LR
P0[初始 prompt] --> E1[ragas evaluate]
E1 --> S{overall ≥ 0.85?}
S -->|是| DONE[迭代结束]
S -->|否| W[找最弱 metric]
W --> SUG[映射到 prompt 改建议]
SUG --> M[modify prompt]
M --> E2[ragas evaluate]
E2 --> S
S -->|max_iterations| STOP[到达上限]
style DONE fill:#e8f5e9
style STOP fill:#fff3e0
工程实务的 4 类 metric 驱动改进：
| 弱 metric | 推荐 prompt 改动 | 预期提升（示意值） |
|---|---|---|
| Faithfulness | 加 "仅基于 context" | +5-10pp |
| Answer Relevance | 加 "直接针对问题" | +3-7pp |
| Context Recall | 改 retriever（非 prompt） | +5-15pp |
| Answer Correctness | 加 "不确定时说不知道" | +5-12pp |
具体例子（示意数据，需用你自己的数据复算）：客服 RAG 4 轮迭代：
| 迭代 | prompt 关键改动 | overall |
|---|---|---|
| v0 | 原始 | 0.62 |
| v1 | + 仅基于 context | 0.71 (+9pp) |
| v2 | + 直接针对问题 | 0.76 (+5pp) |
| v3 | + 不确定时说不知道 | 0.83 (+7pp) |
| v4 | reranker 调整 | 0.87 ✅ 达标 |
4 轮迭代从 0.62 → 0.87，3 天工程时间——每一步改动都有分数佐证，而不是凭直觉。
3 类常见迭代陷阱：
| 陷阱 | 现象 | 修法 |
|---|---|---|
| 单维度优化 | Faithfulness 涨但 Relevance 跌 | overall 看综合 |
| 改 prompt 改死 | 加太多约束让 prompt 过长 | 每次只加 1-2 句 |
| 不验证 anchor | 评测分涨但人工感觉变差 | §8.6 元评测 |
研究背景：
- Zhou et al. 2023 "Large Language Models Are Human-Level Prompt Engineers"（APE，arXiv:2211.01910）用评分驱动自动搜索 prompt，是这套思路的学术源头之一
- DSPy 把 prompt optimization 系统化
- "Auto-CoT" 自动改进 chain-of-thought 是同方向
11.15.7 ragas 与"评测分数 → 业务收益"的端到端 attribution——把 RAG 改进真正变成业务语言
工程师向老板汇报"Faithfulness 从 0.78 升到 0.85"时，老板的疑惑往往是："这意味着多少美元？" 本节给读者一份 ragas 评测分数 → 业务收益的 attribution 模型骨架（转换系数必须用你自己的 A/B 实验或历史数据标定，本节数字全是示意值）——把 4 个核心 metric 改进映射到具体业务指标（NPS / 客诉率 / 留存 / 转化），让 RAG 评测投入有"经济学话语"。
graph LR
A[ragas 4 大 metric 改进] --> B[Faithfulness +X]
A --> C[Answer Relevance +Y]
A --> D[Context Recall +Z]
A --> E[Context Precision +W]
B --> F[幻觉率下降]
C --> G[用户满意度上升]
D --> H[问题解决率上升]
E --> I[token 成本下降]
F --> J[客诉率下降<br/>$节省]
G --> K[NPS 上升<br/>留存$]
H --> L[转化率上升<br/>营收$]
I --> M[OPEX 下降<br/>$节省]
J & K & L & M --> N[年度业务影响]
4 大 metric × 业务收益映射表（示意数据，需用你自己的数据复算）：
| Metric | 改进 1pp | 业务影响通道 | 转化系数（示意） | 年度收益估算（百万 MAU 场景，示意） |
|---|---|---|---|---|
| Faithfulness | +1pp | 幻觉率下降 → 客诉率下降 | 1pp ≈ 客诉率 -0.3pp | 客诉处理成本节省 ~$50K/年 |
| Answer Relevance | +1pp | 用户满意度 → NPS → 留存 | 1pp ≈ NPS +0.2 → 留存 +0.05% | 留存价值 ~$120K/年 |
| Context Recall | +1pp | 问题解决率 → 完成率 → 转化 | 1pp ≈ 转化 +0.08% | 营收 ~$200K/年（GMV $1B 场景） |
| Context Precision | +1pp | 减少冗余 context → token 节省 | 1pp ≈ token -2% | OPEX 节省 ~$30K/年 |
配套实现：评测分数 → 业务收益 attribution 引擎：
from dataclasses import dataclass
@dataclass
class MetricBusinessImpactCoefficient:
metric_name: str
direction: str # "lower_is_better" / "higher_is_better"
business_channel: str
annual_revenue_per_pp_usd: float
@dataclass
class RagasROIAttribution:
coefficients: list[MetricBusinessImpactCoefficient]
# 例：[("Faithfulness", "higher_is_better", "客诉率", 50_000), ...]
monthly_active_users: int
revenue_baseline_usd: float
def project_annual_impact(
self, baseline: dict[str, float], candidate: dict[str, float]
) -> dict:
"""每个 metric 改进的业务收益估算"""
per_metric: dict[str, float] = {}
total = 0.0
for c in self.coefficients:
old = baseline.get(c.metric_name)
new = candidate.get(c.metric_name)
if old is None or new is None:
continue
delta_pp = (new - old) * 100
if c.direction == "lower_is_better":
delta_pp = -delta_pp
scale = self.monthly_active_users / 1_000_000
impact = delta_pp * c.annual_revenue_per_pp_usd * scale
per_metric[c.metric_name] = impact
total += impact
roi_pct = total / max(self.revenue_baseline_usd, 1) * 100
return {
"per_metric_usd": per_metric,
"total_annual_impact_usd": total,
"roi_pct_of_baseline_revenue": roi_pct,
}
def explain_to_executives(self, baseline: dict, candidate: dict) -> str:
report = self.project_annual_impact(baseline, candidate)
lines = ["RAG 改进的业务收益估算：\n"]
for m, v in report["per_metric_usd"].items():
lines.append(f" - {m}: ${v:,.0f}/年")
lines.append(f"\n合计年度影响：${report['total_annual_impact_usd']:,.0f}")
lines.append(f"占基线营收 {report['roi_pct_of_baseline_revenue']:.3f}%")
return "\n".join(lines)举例（示意数据，需用你自己的数据复算）：某 1M MAU 客服 SaaS 跑 baseline vs candidate：
- Faithfulness 0.78 → 0.85 (+7pp) → 客诉成本节省 $350K/年
- Answer Relevance 0.72 → 0.80 (+8pp) → 留存价值 $960K/年
- Context Recall 0.65 → 0.78 (+13pp) → 转化营收 $2,600K/年
- Context Precision 0.60 → 0.70 (+10pp) → OPEX 节省 $300K/年
- 合计 ≈ $4.21M/年（示意值）
这个 $4.2M 只是把示意系数代进公式的算术结果，本身不是证据。向老板汇报时要同时交代三件事：系数从哪次实验或哪段历史数据标定，置信区间多宽，各 metric 的收益有没有重复计算（§11.12.4 讲过，Recall、Faithfulness、Correctness 往往是同一次改动的不同投影，直接相加会高估）。系数没标定之前，这张表只能用来讨论"哪条通道值得先做实验验证"，不能当成 ROI 结论。
11.16 跨专栏关联
- 本专栏第 4 章 §4.3 的 Faithfulness / Answer Relevance / Context Recall 定义，全部是本章源码的概念前置
- 本专栏第 13 章 RAG 评测：会用 ragas API 做完整 RAG 系统评测的端到端示例
- 本专栏第 14 章 Agent 评测：会展示 ragas 的 ToolCallAccuracy / GoalAccuracy 在 Agent 评测中的使用
- 《RAG 工程与检索系统设计》专栏第 20 章：RAG 评测的完整落地，retriever 调参的 oracle 用的就是 ragas（注意那一栏锚的是 v0.4.3，API 与本章不同）
- 《LangChain 设计与实现》专栏第 10 章 §10.22：把 ragas 与 TruLens / DeepEval 放在一起比较，讲选哪一个内化进 CI
11.17 本章小结
- ragas 是 RAG 评测的工程化典范：把 Faithfulness / Recall / Relevance / Correctness 等概念变成可调用的 Python class
- Metric × Sample 的最小抽象让 36 个 metric 类共享同一套数据流
- Faithfulness 的两阶段实现（LongFormAnswerPrompt + NLIStatementPrompt）+ "no pronouns" 约束 + reason 字段，是 LLM-judge 工程化的最佳实践范本
- Answer Relevance 的"反向生成 question"巧妙绕过了"绝对分校准难"问题
- 36 个内置 metric 类覆盖 RAG 核心、Agent、SQL、多模态——已超出 RAG 范畴
- 5 个工程亮点（PydanticPrompt、Async、multi_responses、MetricOutputType、NaN 处理）值得任何评测框架借鉴
下一章我们看 promptfoo——生产化评测最易上手的工具，与 ragas / lm-eval 完全不同的 YAML-first 哲学。