评估与生产部署:让 RAG 从原型走向生产

评估与生产部署:让 RAG 从原型走向生产

前三篇讲了 RAG 管线的每个环节。这篇讲最后一个问题:怎么知道你的 RAG 系统好不好,以及怎么把它部署到生产环境。

这是 RAG 系列中最容易被忽视、但最决定性的环节。很多人搭了一个 RAG 系统,觉得"能跑"就完事了。但"能跑"和"好用"之间,隔着一套评估和持续优化的体系。

一、为什么 RAG 评估这么难

RAG 评估难,因为它不是单一指标能衡量的。

传统软件有明确的输入输出,测试用例跑完就知道对不对。RAG 不一样:

  • 检索结果好不好? 检索到了相关文档,但排名对不对?数量够不够?
  • 生成回答好不好? 基于检索结果生成的回答,准确吗?完整吗?有没有幻觉?
  • 端到端效果好不好? 用户的问题最终得到满意回答了吗?

这三个问题对应三个不同层面的评估,每个层面都有多个指标。

二、RAG 评估的三个层面

2.1 检索评估

衡量检索器返回的结果质量。

指标 定义 目标
Recall@k top-k 结果中包含相关文档的比例 越高越好,目标 >90%
Precision@k top-k 结果中相关文档的比例 越高越好
MRR(Mean Reciprocal Rank) 正确答案排在第几位的倒数平均值 越接近 1 越好
MAP(Mean Average Precision) 考虑排序精度的平均准确率 越接近 1 越好

怎么算? 需要准备标注数据:每个测试问题标注了"相关文档列表"。

# 标注数据示例
test_cases = [
    {
        "question": "LangChain 的 LCEL 是什么?",
        "relevant_docs": ["lcel-introduction.md", "chain-tutorial.md"]
    },
    {
        "question": "怎么配置 LangSmith?",
        "relevant_docs": ["langsmith-setup.md", "monitoring-guide.md"]
    }
]

# 计算 Recall@5
def recall_at_k(retrieved_docs, relevant_docs, k=5):
    retrieved_top_k = retrieved_docs[:k]
    relevant_count = sum(1 for doc in retrieved_top_k if doc in relevant_docs)
    return relevant_count / len(relevant_docs)

# 计算 MRR
def mrr(retrieved_docs, relevant_docs):
    for i, doc in enumerate(retrieved_docs):
        if doc in relevant_docs:
            return 1.0 / (i + 1)
    return 0.0

2.2 生成评估

衡量模型基于检索结果生成的回答质量。

指标 定义 评估方式
Faithfulness(忠实度) 回答是否基于检索结果,没有幻觉 LLM 判断
Relevance(相关性) 回答是否直接回答了用户问题 LLM 判断
Completeness(完整性) 回答是否覆盖了所有要点 LLM 判断
Correctness(正确性) 回答的事实是否正确 人工标注

LLM 作为评估器:用另一个 LLM 来评估生成回答的质量。

def evaluate_faithfulness(answer, context):
    """评估回答是否忠实于上下文"""
    prompt = f"""评估以下回答是否完全基于提供的上下文。
    如果回答包含了上下文中没有的信息,标记为"不忠实"。

    上下文:{context}

    回答:{answer}

    评估结果(忠实/不忠实)和原因:"""

    result = llm.invoke(prompt)
    return "忠实" in result.content

def evaluate_relevance(answer, question):
    """评估回答是否相关"""
    prompt = f"""评估以下回答是否直接回答了用户问题。

    问题:{question}

    回答:{answer}

    评估结果(相关/不相关)和原因:"""

    result = llm.invoke(prompt)
    return "相关" in result.content

2.3 端到端评估

衡量整个 RAG 管线的最终效果。

指标 定义 说明
Answer Correctness 最终回答的正确性 最核心的指标
Answer Relevancy 最终回答与问题的相关性 是否答非所问
Context Recall 检索结果覆盖了正确答案的程度 检索是否充分
Context Precision 检索结果中相关内容的比例 检索是否精准

三、RAGAS 评估框架

RAGAS(Retrieval Augmented Generation Assessment)是目前最流行的 RAG 评估框架。它提供了标准化的评估流程和指标。

3.1 安装和使用

pip install ragas
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision,
    answer_correctness
)
from datasets import Dataset

# 准备评估数据
data = {
    "question": [
        "LangChain 的 LCEL 是什么?",
        "怎么配置 LangSmith?",
        "RAG 和微调有什么区别?"
    ],
    "answer": [
        "LCEL 是 LangChain 的表达式语言...",
        "配置 LangSmith 需要设置环境变量...",
        "RAG 通过检索外部知识来增强生成..."
    ],
    "contexts": [
        ["LCEL 是 LangChain 的表达式语言,使用管道符号..."],
        ["LangSmith 是 LangChain 的监控平台..."],
        ["RAG 通过检索增强生成..."]
    ],
    "ground_truth": [
        "LCEL(LangChain Expression Language)是...",
        "配置 LangSmith 的步骤:1. 注册账号...",
        "RAG 通过检索外部知识增强模型..."
    ]
}

dataset = Dataset.from_dict(data)

# 运行评估
results = evaluate(
    dataset=dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_recall,
        context_precision,
        answer_correctness
    ],
    llm=ChatOpenAI(model="gpt-4o"),
    embeddings=OpenAIEmbeddings()
)

print(results)
# {
#   'faithfulness': 0.92,
#   'answer_relevancy': 0.88,
#   'context_recall': 0.85,
#   'context_precision': 0.79,
#   'answer_correctness': 0.83
# }

3.2 评估结果解读

指标 分数 说明
faithfulness 0.92 回答很忠实于上下文,幻觉很少
answer_relevancy 0.88 回答和问题高度相关
context_recall 0.85 检索结果覆盖了大部分相关信息
context_precision 0.79 检索结果中有一些不太相关的内容
answer_correctness 0.83 回答基本正确,有小部分不准确

怎么判断整体效果? 没有一个统一的及格线。建议:

  • 先跑一次评估,建立基线
  • 每次改动后重新评估,看指标是否提升
  • 关注趋势,不纠结绝对值

四、评估数据集构建

评估的关键是数据集。好的评估数据集应该覆盖不同类型的查询。

4.1 查询分类

类型 示例 占比
事实查询 "LangChain 的创始人是谁?" 30%
概念理解 "LCEL 和传统 Chain 有什么区别?" 25%
流程查询 "怎么配置 LangSmith?" 20%
对比分析 "RAG 和微调哪个更适合?" 15%
边界测试 "LangChain 能做什么不能做什么?" 10%

4.2 数据集规模

  • 最小可用:20-30 个问题,覆盖主要查询类型
  • 常规评估:100-200 个问题,覆盖各种场景
  • 全面评估:500+ 个问题,覆盖边缘情况

4.3 标注方法

人工标注:最准确,但耗时。每个问题标注:

  • 相关文档列表(检索评估)
  • 标准答案(生成评估)

LLM 辅助标注:用 LLM 生成初步标注,人工审核修正。效率更高。

# LLM 辅助生成标准答案
def generate_ground_truth(question, relevant_docs):
    prompt = f"""基于以下参考文档,回答用户的问题。
    回答要准确、完整、简洁。

    参考文档:{relevant_docs}

    问题:{question}

    标准答案:"""

    return llm.invoke(prompt).content

五、生产部署架构

5.1 基本架构

┌─────────────────────────────────────────────────────┐
│                    RAG 生产系统                      │
│                                                     │
│  ┌──────────┐     ┌──────────────┐                  │
│  │ 用户查询  │────▶│  API Gateway │                  │
│  └──────────┘     └──────┬───────┘                  │
│                          │                          │
│                          ▼                          │
│  ┌──────────────────────────────────────┐           │
│  │          RAG Pipeline                │           │
│  │                                      │           │
│  │  Query改写 → 多路召回 → 去重融合 →    │           │
│  │  Reranking → 上下文压缩 → 生成回答 →  │           │
│  │  幻觉检测                            │           │
│  └──────────────────┬───────────────────┘           │
│                     │                               │
│          ┌──────────┼──────────┐                    │
│          ▼          ▼          ▼                    │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐            │
│  │ 向量数据库 │ │ BM25 索引 │ │ LLM 服务  │            │
│  │(Qdrant)  │ │(Elastic) │ │(API)     │            │
│  └──────────┘ └──────────┘ └──────────┘            │
│                                                     │
│  ┌──────────────────────────────────────┐           │
│  │          监控和日志                    │           │
│  │  检索延迟 / 生成质量 / 用户反馈        │           │
│  └──────────────────────────────────────┘           │
└─────────────────────────────────────────────────────┘

5.2 关键组件

API Gateway:接收用户请求,路由到 RAG Pipeline,返回结果。

RAG Pipeline:完整的 RAG 处理流程,包括 Query 改写、检索、Reranking、生成。

向量数据库:存储和检索向量。生产环境推荐 Qdrant 或 Milvus。

BM25 索引:关键词检索。可以用 Elasticsearch 或单独维护。

LLM 服务:生成回答。可以是 OpenAI API、本地模型、或自建服务。

监控系统:记录每次请求的延迟、检索结果、生成回答、用户反馈。

5.3 性能优化

缓存:相同或相似的查询直接返回缓存结果。

from functools import lru_cache
import hashlib

class RAGCache:
    def __init__(self, max_size=10000):
        self.cache = {}
        self.max_size = max_size

    def get_cache_key(self, query):
        return hashlib.md5(query.encode()).hexdigest()

    def get(self, query):
        key = self.get_cache_key(query)
        return self.cache.get(key)

    def set(self, query, result):
        key = self.get_cache_key(query)
        if len(self.cache) >= self.max_size:
            # LRU 淘汰
            oldest_key = next(iter(self.cache))
            del self.cache[oldest_key]
        self.cache[key] = result

异步处理:检索和生成可以部分异步。

import asyncio

async def rag_pipeline(query):
    # 1. Query 改写(异步)
    rewritten_query = await rewrite_query(query)

    # 2. 多路召回(并行)
    vector_task = asyncio.create_task(vector_search(rewritten_query))
    bm25_task = asyncio.create_task(bm25_search(rewritten_query))
    vector_results, bm25_results = await asyncio.gather(vector_task, bm25_task)

    # 3. 去重融合
    merged = merge_results(vector_results, bm25_results)

    # 4. Reranking
    reranked = await rerank(merged, rewritten_query)

    # 5. 生成回答
    answer = await generate_answer(reranked, query)

    return answer

批量处理:多个查询可以批量处理,提高吞吐量。

5.4 降级策略

生产环境需要考虑各种故障情况:

async def rag_with_fallback(query):
    try:
        # 正常流程
        return await rag_pipeline(query)
    except VectorDBError:
        # 向量数据库不可用,降级到纯关键词检索
        logger.warning("Vector DB unavailable, falling back to BM25")
        return await bm25_only_pipeline(query)
    except LLMError:
        # LLM 不可用,返回检索结果摘要
        logger.warning("LLM unavailable, returning retrieved docs")
        return await retrieve_only(query)
    except Exception as e:
        # 其他错误,返回友好提示
        logger.error(f"RAG pipeline error: {e}")
        return "抱歉,系统暂时无法回答这个问题。请稍后再试。"

六、监控和持续优化

6.1 监控指标

指标 说明 告警阈值
检索延迟 从查询到返回检索结果的时间 >500ms
生成延迟 从检索到生成回答的时间 >3s
端到端延迟 从用户提问到收到回答的时间 >5s
缓存命中率 缓存命中的比例 <30%
用户满意度 用户反馈(点赞/点踩) 点踩率 >20%

6.2 日志记录

每次请求都应该记录:

log_entry = {
    "timestamp": "2026-06-05T10:30:00Z",
    "query": "LangChain 的 LCEL 是什么?",
    "rewritten_query": "LangChain LCEL 表达式语言定义和使用方法",
    "retrieved_docs": ["lcel-intro.md", "chain-tutorial.md"],
    "reranked_docs": ["lcel-intro.md"],
    "answer": "LCEL 是 LangChain 的表达式语言...",
    "metrics": {
        "retrieval_latency_ms": 120,
        "generation_latency_ms": 1800,
        "total_latency_ms": 1950,
        "context_precision": 0.85,
        "faithfulness": 0.92
    },
    "user_feedback": "upvote"  # 或 "downvote"
}

6.3 持续优化循环

部署上线
    │
    ▼
收集用户查询和反馈
    │
    ▼
定期跑评估(每周/每月)
    │
    ▼
分析薄弱环节
    │
    ▼
针对性优化(切分/检索/生成)
    │
    ▼
A/B 测试验证
    │
    ▼
上线新版本
    │
    └──→ 回到第一步

6.4 用户反馈的利用

用户反馈是 RAG 优化的金矿:

  • 点踩的回答:分析是检索问题(没检索到相关文档)还是生成问题(检索到了但回答不好)
  • 高频查询:优化这些查询的检索效果,收益最大
  • 无结果的查询:说明知识库有缺口,需要补充文档
# 分析点踩的原因
def analyze_downvote(query, retrieved_docs, answer, context):
    if not retrieved_docs:
        return "no_results"  # 知识库没有相关文档

    relevance = evaluate_relevance(answer, query)
    if relevance < 0.5:
        return "irrelevant_answer"  # 检索到了但回答不相关

    faithfulness = evaluate_faithfulness(answer, context)
    if faithfulness < 0.5:
        return "hallucination"  # 有幻觉

    return "other"  # 其他原因

七、LangSmith 集成

LangSmith 是 LangChain 官方的监控平台,可以追踪 RAG 管线的每一步。

7.1 基本配置

import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-rag-project"

# 之后正常调用 LangChain 代码,所有调用自动追踪

7.2 在 LangSmith 中查看 RAG 链路

LangSmith 会记录每次 RAG 请求的完整链路:

用户查询 → Query 改写 → 向量检索 → BM25 检索 → 去重融合 → Reranking → 生成回答
  │           │           │          │          │          │          │
  │           │           │          │          │          │          └─ 输出 + Token 数
  │           │           │          │          │          └─ 重排序后结果
  │           │           │          │          └─ 合并去重结果
  │           │           │          └─ BM25 返回结果
  │           │           └─ 向量检索返回结果
  │           └─ 改写后的查询
  └─ 原始查询

每一步的延迟、输入输出、Token 消耗都清晰可见。

7.3 用 LangSmith 做评估

LangSmith 内置了评估功能,可以直接在平台上跑评估:

from langsmith import Client

client = Client()

# 创建评估数据集
dataset = client.create_dataset(
    name="rag-evaluation",
    description="RAG 系统评估数据集"
)

# 添加测试用例
client.create_examples(
    dataset_id=dataset.id,
    inputs=[
        {"question": "LangChain 的 LCEL 是什么?"},
        {"question": "怎么配置 LangSmith?"},
    ],
    outputs=[
        {"answer": "LCEL 是 LangChain 的表达式语言..."},
        {"answer": "配置 LangSmith 需要..."},
    ]
)

# 运行评估
from langsmith.evaluation import evaluate

results = evaluate(
    target=rag_pipeline,
    data=dataset,
    evaluators=[faithfulness, answer_relevancy, context_recall],
    experiment_prefix="rag-v1"
)

八、总结

RAG 系统的评估和持续优化是一个工程体系,不是一次性的任务。

核心要点:

  • 三个层面的评估:检索评估、生成评估、端到端评估
  • RAGAS 是最常用的评估框架,提供标准化指标
  • 生产部署需要考虑:缓存、异步、降级、监控
  • 持续优化是核心:收集反馈 → 定期评估 → 针对性优化 → A/B 测试 → 上线
  • LangSmith 是 RAG 监控的标配,追踪每一步的输入输出和延迟

这完成了 RAG 系列全部 5 篇文章。回顾整个系列:

  1. RAG 系统全景——管线全貌、组件选型、真实系统分析、常见坑
  2. 文档切分——五种切分策略、不同文档类型实践、参数选择
  3. 向量检索——Embedding 模型、向量数据库、混合检索、Reranking
  4. 检索后优化——Query 改写、HyDE、多路召回、上下文压缩、幻觉抑制
  5. 评估与生产部署——评估指标、RAGAS、生产架构、监控、持续优化

系列