评估与生产部署:让 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 篇文章。回顾整个系列:
- RAG 系统全景——管线全貌、组件选型、真实系统分析、常见坑
- 文档切分——五种切分策略、不同文档类型实践、参数选择
- 向量检索——Embedding 模型、向量数据库、混合检索、Reranking
- 检索后优化——Query 改写、HyDE、多路召回、上下文压缩、幻觉抑制
- 评估与生产部署——评估指标、RAGAS、生产架构、监控、持续优化
系列: