前言

在上一篇《Ragas生产环境工程化落地指南》中,我们彻底告别了网上泛滥的Demo教程,解决了Ragas私有化部署、打分不稳定、成本过高、指标失真等核心生产难题。

但很多团队落地后,依然卡在最后一步:手动评测

每次修改分块策略、调整Prompt、更换Embedding模型、升级知识库,都需要开发者手动运行脚本、手动对比分数、手动整理报告。这种模式依然无法适配快速迭代的线上业务,存在三个致命短板:

  • 迭代效率低:每次版本迭代都依赖人工介入,无法做到提交即评测

  • 版本风险高:极易出现隐性效果退化,上线后才发现召回率、忠实度大幅下跌

  • 数据无沉淀:评测数据零散、无版本追溯,无法形成长期效果迭代台账

本文作为Ragas生产落地进阶终篇,带你从零搭建一套企业级全自动RAG评测CI/CD流水线。全程无冗余Demo,所有代码、配置、流程均经过线上业务验证,实现:代码提交自动评测、版本分数自动对比、退化自动告警、评测报告自动归档,彻底实现RAG系统无人值守常态化回归测试

阅读前置:建议先阅读上篇Ragas工程化落地文章,掌握私有化模型适配、指标取舍、分数优化核心知识点。

一、为什么RAG必须做CI/CD自动化评测?

传统软件迭代有单元测试、接口自动化测试,而RAG大模型应用长期缺失标准化回归测试体系,这是绝大多数线上RAG系统越迭代越烂的核心原因。

1.1 RAG迭代的隐形风险

RAG系统的所有改动都是高风险不可见改动

  • 分块大小从512改成1024,肉眼看不出效果差异,可能导致精确率暴跌

  • Prompt微调几句话术,可能引发模型幻觉飙升、忠实度下降

  • 知识库增量更新,可能引入大量噪声文档,污染检索结果

  • 更换Embedding模型,会直接改变全量向量检索逻辑,引发全局效果波动

这些问题,人工测试完全无法覆盖,只有自动化回归评测可以提前拦截。

1.2 自动化流水线核心价值

搭建完成后,可实现RAG项目的标准化工程闭环:

  • 提交即评测:代码/知识库更新提交后,自动触发全量回归测试

  • 退化即拦截:核心指标跌破阈值、版本分数下降超标,直接阻断上线

  • 全程可追溯:所有版本评测数据、低分案例、优化记录全部归档

  • 无人值守迭代:无需人工参与,每日定时巡检、版本迭代自动校验

二、流水线整体架构设计(企业标准)

我们搭建的生产级流水线,摒弃复杂冗余架构,采用轻量、稳定、易落地的四层架构,中小团队可直接复用,无需额外部署重型组件。

2.1 流水线整体流程

线上日志采集 → 测试集更新 → 自动评测执行 → 版本分数对比 → 退化告警拦截 → 报告归档沉淀

2.2 核心组件说明

  • 数据层:线上真实用户Query、RAG检索上下文、模型回答、历史标注数据

  • 评测引擎:Langchain + Ragas + 私有化国产模型(无外网、数据安全)

  • 调度层:Git CI/CD / 定时脚本 / Jenkins(任选其一)

  • 结果层:分数对比、低分案例导出、退化告警、Excel/日志归档

2.3 生产核心设计原则

  • 不依赖公网:全程内网私有化模型推理,杜绝数据泄露

  • 低成本运行:分层评测策略,日常轻量巡检、版本全量评测

  • 稳定可复现:固定裁判模型参数、固化评测Prompt,分数波动可控

  • 可落地拦截:明确阈值红线,直接关联上线流程

三、前置准备工作(环境统一规范)

3.1 依赖安装(生产固定版本,避免兼容问题)

pip install ragas==0.1.10 langchain==0.1.20 langchain-openai datasets python-dotenv pandas

3.2 目录结构(企业标准化目录)

统一项目目录,适配CI自动调度:

ragas_eval/
├── .env                # 密钥与模型配置(不提交代码仓库)
├── eval_config.py      # 评测阈值、红线配置
├── ragas_ci_eval.py    # 核心自动化评测脚本
├── test_sets/          # 沉淀的生产测试集
└── eval_reports/       # 自动归档评测报告

3.3 统一阈值配置(对接上篇生产标准)

新建 eval_config.py,固化团队评测标准,避免人为随意修改:

# 生产阈值配置(通用业务场景)
PROD_THRESHOLD = {
    "faithfulness": 0.85,
    "context_recall": 0.85,
    "context_precision": 0.75,
    "answer_relevancy": 0.80
}

# 版本退化红线:分数下降超过该值判定为版本退化
REGRESSION_DELTA = 0.03

# 分层评测配置
LIGHT_EVAL_SAMPLE_NUM = 20   # 日常轻量评测样本数
FULL_EVAL_SAMPLE_NUM = 200   # 版本迭代全量评测样本数

四、核心代码:可直接上线的CI自动化评测脚本

基于上篇私有化模型代码升级改造,新增版本对比、退化检测、自动归档、结果校验、退出码拦截能力,完全适配CI/CD流水线调度。

新建 ragas_ci_eval.py

import os
import sys
import pandas as pd
from datetime import datetime
from dotenv import load_dotenv
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import faithfulness, context_precision, context_recall, answer_relevancy
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from ragas.llms import LangchainLLMWrapper
from ragas.embeddings import LangchainEmbeddingsWrapper

# 加载自定义配置与环境变量
from eval_config import PROD_THRESHOLD, REGRESSION_DELTA
load_dotenv()

# ===================== 初始化私有化裁判模型(稳定版) =====================
def init_ragas_engine():
    base_url = os.getenv("PRIVATE_MODEL_BASE_URL")
    api_key = os.getenv("PRIVATE_MODEL_API_KEY")
    model_name = os.getenv("JUDGE_MODEL_NAME")
    embed_model_name = os.getenv("EMBED_MODEL_NAME")

    # 固定参数,杜绝打分波动
    llm = ChatOpenAI(
        base_url=base_url,
        api_key=api_key,
        model=model_name,
        temperature=0.0,
        max_tokens=2048,
        timeout=30,
        streaming=False
    )
    embed = OpenAIEmbeddings(
        base_url=base_url,
        api_key=api_key,
        model=embed_model_name
    )
    return LangchainLLMWrapper(llm), LangchainEmbeddingsWrapper(embed)

# ===================== 加载生产测试集(可对接线上日志) =====================
def load_prod_test_set():
    # 实际生产可替换为:读取线上日志解析后的数据集/本地沉淀测试集
    test_data = {
        "question": [
            "企业员工社保补缴需要哪些材料?",
            "公司试用期员工是否享有年终奖?",
            "员工年假可以累计到次年吗?",
            "公积金缴存基数多久调整一次?"
        ],
        "contexts": [
            ["员工社保补缴需提供身份证复印件、劳动合同、补缴申请表、工资流水证明,提交至人事部门审核。"],
            ["公司规章制度规定,试用期员工未转正不参与年度年终奖评选,转正员工可正常享受年终奖福利。"],
            ["公司年假原则上当年休完,因工作原因无法休假的,可申请累计至次年一季度。"],
            ["公积金缴存基数每年7月统一调整一次,依据上一年度月均工资核算。"]
        ],
        "answer": [
            "社保补缴需准备身份证复印件、劳动合同、补缴申请表、工资流水证明,提交人事审核。",
            "试用期员工不参与年终奖评选,无年终奖权益。",
            "年假原则当年休完,特殊工作原因可累计至次年一季度。",
            "公积金基数每年7月统一调整,根据上年度月均工资计算。"
        ],
        "ground_truth": [
            "社保补缴材料:身份证复印件、劳动合同、补缴申请表、工资流水证明。",
            "试用期员工无年终奖,仅转正员工享有。",
            "年假可因工作原因累计至次年一季度。",
            "公积金缴存基数每年7月调整一次。"
        ]
    }
    return Dataset.from_dict(test_data)

# ===================== 核心评测与退化检测 =====================
def run_eval():
    # 初始化引擎
    ragas_llm, ragas_embed = init_ragas_engine()
    # 加载数据集
    eval_ds = load_prod_test_set()
    # 执行评测
    result = evaluate(
        dataset=eval_ds,
        metrics=[faithfulness, context_recall, context_precision, answer_relevancy],
        llm=ragas_llm,
        embeddings=ragas_embed,
        batch_size=4
    )
    df = result.to_pandas()

    # 1. 计算当前版本平均分
    curr_score = {
        "faithfulness": df["faithfulness"].mean(),
        "context_recall": df["context_recall"].mean(),
        "context_precision": df["context_precision"].mean(),
        "answer_relevancy": df["answer_relevancy"].mean()
    }

    # 2. 校验是否低于生产阈值
    fail_threshold = False
    for k, v in curr_score.items():
        if v < PROD_THRESHOLD[k]:
            print(f"[阈值告警] {k} 分数不达标:{v:.4f},最低阈值:{PROD_THRESHOLD[k]}")
            fail_threshold = True

    # 3. 对比历史版本(简易版本回溯,可扩展数据库)
    history_file = "eval_reports/history_score.csv"
    if os.path.exists(history_file):
        history_df = pd.read_csv(history_file)
        last_score = history_df.iloc[-1]
        for k in curr_score.keys():
            delta = curr_score[k] - last_score[k]
            if delta < -REGRESSION_DELTA:
                print(f"[版本退化] {k} 下降 {abs(delta):.4f},超过红线阈值{REGRESSION_DELTA}")
                fail_threshold = True

    # 4. 自动归档报告
    os.makedirs("eval_reports", exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    report_path = f"eval_reports/ragas_eval_report_{timestamp}.csv"
    df.to_csv(report_path, index=False, encoding="utf-8-sig")

    # 5. 更新历史分数记录
    score_line = pd.DataFrame([curr_score])
    score_line["time"] = timestamp
    if os.path.exists(history_file):
        score_line.to_csv(history_file, mode="a", header=False, index=False)
    else:
        score_line.to_csv(history_file, index=False)

    # 6. CI拦截:不达标直接退出,阻断上线
    if fail_threshold:
        print("[ERROR] RAG评测不通过,存在指标不达标或版本退化,禁止上线!")
        sys.exit(1)
    else:
        print("[SUCCESS] RAG自动化评测通过,所有指标达标!")
        print("当前版本分数:", curr_score)
        sys.exit(0)

if __name__ == "__main__":
    run_eval()

五、流水线核心能力解析(生产核心亮点)

5.1 自动版本退化检测

脚本会自动读取上一版本的评测分数,对比当前版本,只要任意核心指标下降超过0.03,直接判定版本退化,触发拦截。彻底解决“迭代优化越改越差、无人发现”的问题。

5.2 生产阈值强校验

严格对齐上篇文章企业标准阈值,忠实度、召回率、精确率、相关性任一指标不达标,直接评测失败,阻断上线流程,守住业务底线。

5.3 全量数据自动归档

每次评测自动生成带时间戳的详细报告,保存每一条样本的打分明细,同时记录各版本平均分,形成长期迭代台账,方便复盘优化、效果追溯。

5.4 CI退出码拦截机制

评测不通过返回非0退出码,可被Git、Jenkins等CI工具识别,直接阻断代码合并、版本发布,实现评测不通过、绝不上线的工程规范。

六、CI/CD流水线接入方式(3种落地方案)

根据团队基建能力,任选一种方案接入,从轻量到重度全覆盖。

6.1 方案一:定时巡检(零成本、最易落地)

通过服务器Crontab配置每日定时评测,实现每日自动巡检,监控RAG系统长期效果波动:

# 每天凌晨2点自动执行评测
0 2 * * * /usr/bin/python3 /你的项目路径/ragas_ci_eval.py >> /你的项目路径/eval_log.log 2>&1

6.2 方案二:代码提交触发(适配Git/GitLab CI)

新增 .gitlab-ci.yml 配置,代码提交合并后自动触发回归测试:

stages:
  - ragas_eval

ragas_auto_eval:
  stage: ragas_eval
  script:
    - python ragas_ci_eval.py
  only:
    - main
    - develop

6.3 方案三:Jenkins流水线(企业重度方案)

在Jenkins发布流水线中前置RAG评测步骤,评测通过才执行打包发布,完全闭环管控,适合大型企业标准化发布流程。

七、流水线生产优化进阶方案

7.1 对接线上真实日志,替代静态测试集

脚本中load_prod_test_set函数可二次开发,对接线上MongoDB/MySQL日志,自动抽取每日真实用户Query、问答对,动态更新测试集,让评测永远贴合真实线上场景。

7.2 低分案例自动复盘

归档报告中自动汇总低分样本,可二次开发脚本,自动输出问题清单:幻觉案例、漏召回案例、答非所问案例,精准指导迭代优化。

7.3 企业微信/钉钉告警

在脚本退化检测逻辑后,增加webhook推送,评测失败、版本退化自动推送告警消息,无需人工盯守。

7.4 分层评测落地

日常定时任务执行轻量评测(20样本),快速监控;版本发布前执行全量评测(200+样本),严控上线质量,兼顾成本与效果。

八、常见流水线落地坑点&解决方案

  • CI环境密钥失效:统一配置服务器环境变量,禁止代码携带密钥,符合安全规范

  • 多次评测分数波动:确认裁判模型temperature=0,固化模型版本、关闭随机参数

  • 评测耗时过长:开启批量请求、截断超长上下文、分层评测、缓存重复推理结果

  • 历史版本对比异常:清理脏数据,固定测试集样本,版本迭代统一基准

  • 国产模型打分偏严导致拦截误报:基于自身模型校准专属阈值,不直接套用OpenAI标准

九、总结

如果说上篇文章解决了Ragas能不能生产用的问题,那本文就彻底解决了RAG能不能标准化、自动化、工程化迭代的问题。

RAG系统的工程化成熟,不在于模型多先进、Prompt多精巧,而在于有没有一套可持续回归、可量化、可拦截的评测体系

这套Ragas+LangChain CI/CD流水线,彻底终结了RAG“凭感觉调参、靠运气上线、出问题复盘”的野蛮迭代模式,让每一次优化、每一次版本更新都有数据背书、有质量兜底、有风险拦截。

最终落地闭环:私有化稳定评测 → 自动化定时巡检 → 版本迭代回归拦截 → 低分问题复盘优化 → 持续迭代提质

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐