Spotify 工程团队最近分享了一个关于 LLM Eval 实验设计的核心观点:把 LLM 评估当成漏斗,而不是一刀切的闸门。 这句话看似简单,但背后指向的是很多团队在用 LLM 做自动评估时踩的坑——把 eval 当成 binary pass/fail 的裁判,结果要么放过了坏输出,要么误杀了好输出。
一刀切的困境
很多团队拿到 LLM eval 的第一反应是:写一个 prompt,让模型给输出打 "good" 或 "bad",然后拿这个标签做 gate。问题立刻浮现:
- 单点判断太脆弱。 一个 prompt 的措辞变化、模型版本切换,都可能让通过率从 80% 暴跌到 30%。
- 误杀成本高。 在推荐系统、搜索排序这类场景,误杀一条好结果意味着用户永远看不到它,而漏过一条坏结果只是暂时体验下降。两者不对称。
- 无法迭代。 你只知道 "通过率变了",但不知道是哪一类输出被误判了,无从调优。
Spotify 的核心建议是:不要让 eval 在一个点上做 fork(分流、二选一),而是让它像漏斗一样,层层过滤、逐步收紧。
漏斗模型的三层结构
一个实用的 LLM eval 漏斗通常包含三层:
| 层级 | 目标 | 典型做法 |
|---|---|---|
| 广口层 | 快速剔除明显劣质输出 | 轻量规则 + 简短 LLM 判断 |
| 细筛层 | 对存疑输出做精细评估 | 多维度评分、多模型交叉验证 |
| 精检层 | 人工审核剩余少量样本 | 专家 review + 标注反馈回路 |
关键数字:广口层可能过滤掉 60-70% 的输出,细筛层再过滤 15-20%,最终只有 5-10% 进入人工审核。这样人工成本可控,而自动评估的误判被后续层级兜底。
实践:搭一个三层漏斗
下面用一个 Python 示例演示如何实现这个漏斗。场景是评估 LLM 生成的音乐推荐理由(和 Spotify 的业务贴近)。你可以直接改造用于自己的输出评估。
安装依赖
pip install openai pydantic
漏斗实现
import openai
import json
from pydantic import BaseModel
from typing import List
client = openai.OpenAI() # 需设置 OPENAI_API_KEY 环境变量
# ---- Pydantic 模型定义,约束 LLM 输出格式 ----
class QuickJudgeResult(BaseModel):
"""广口层:快速判断"""
obviously_bad: bool # 是否明显劣质
reason: str
class DetailedScoreResult(BaseModel):
"""细筛层:多维度评分"""
relevance: float # 0-1,与用户兴趣的相关性
coherence: float # 0-1,逻辑连贯性
specificity: float # 0-1,是否具体而非泛泛而谈
overall_pass: bool # 综合是否通过
notes: str
# ---- 第一层:广口过滤 ----
def funnel_layer_quick(candidate: str, user_context: str) -> bool:
"""返回 True 表示通过(保留),False 表示明显劣质(剔除)"""
prompt = f"""你是一个快速质检员。判断以下音乐推荐理由是否【明显劣质】。
明显劣质的标准:包含事实错误、完全无关、或逻辑混乱到无法理解。
如果只是"不够精彩"但不算明显劣质,请判为 NOT obviously_bad。
用户兴趣:{user_context}
推荐理由:{candidate}
请严格按 JSON 格式输出:{{"obviously_bad": bool, "reason": str}}"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0,
)
result = QuickJudgeResult(**json.loads(response.choices[0].message.content))
print(f"[广口层] obviously_bad={result.obviously_bad} | {result.reason}")
return not result.obviously_bad
# ---- 第二层:细筛评分 ----
def funnel_layer_detailed(candidate: str, user_context: str) -> bool:
"""多维度评分,返回 True 表示通过"""
prompt = f"""你是一个资深音乐编辑,对推荐理由做精细评估。
从三个维度打分(0到1):
- relevance:与用户兴趣的相关程度
- coherence:逻辑连贯、表述清晰的程度
- specificity:是否给出了具体理由而非泛泛而谈
综合三个维度,overall_pass 为 True 表示可以上线,False 表示需要改进。
用户兴趣:{user_context}
推荐理由:{candidate}
请严格按 JSON 格式输出:
{{"relevance": float, "coherence": float, "specificity": float, "overall_pass": bool, "notes": str}}"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0,
)
result = DetailedScoreResult(**json.loads(response.choices[0].message.content))
print(f"[细筛层] rel={result.relevance:.2f} coh={result.coherence:.2f} "
f"spec={result.specificity:.2f} pass={result.overall_pass} | {result.notes}")
return result.overall_pass
# ---- 漏斗主流程 ----
def eval_funnel(candidates: List[str], user_context: str) -> dict:
"""三层漏斗:广口 → 细筛 → 人工候选"""
layer1_pass = []
layer1_fail = []
for c in candidates:
if funnel_layer_quick(c, user_context):
layer1_pass.append(c)
else:
layer1_fail.append(c)
layer2_pass = []
layer2_fail = []
for c in layer1_pass:
if funnel_layer_detailed(c, user_context):
layer2_pass.append(c)
else:
layer2_fail.append(c)
return {
"total": len(candidates),
"layer1_pass": len(layer1_pass),
"layer1_fail": len(layer1_fail),
"layer2_pass": len(layer2_pass),
"layer2_fail": len(layer2_fail),
"human_review_needed": layer2_pass, # 这些进入人工精检
}
# ---- 运行示例 ----
if __name__ == "__main__":
user_interest = "喜欢独立摇滚和后朋克,最近在听 The National 和 Interpol"
candidates = [
"推荐 Arcade Fire——他们的编曲层次感和 The National 的沉稳叙事有相似之处,适合你从后朋克过渡到更丰富的独立摇滚。",
"推荐周杰伦——他的音乐很受欢迎,你一定会喜欢。", # 明显不相关
"推荐 Joy Division——如果你喜欢 Interpol,那 Joy Division 是必须听的,Interpol 的声音深受他们影响。", # 相关且具体
"推荐一些摇滚音乐——摇滚很棒,你应该试试。", # 相关但极其泛泛
"推荐 Radiohead——他们的实验性风格可能让你耳目一新,尤其是 OK Computer 这张专辑的氛围构建。", # 相关且具体
]
result = eval_funnel(candidates, user_interest)
print("\n===== 漏斗结果 =====")
print(f"总输入: {result['total']}")
print(f"广口层通过: {result['layer1_pass']} / 剔除: {result['layer1_fail']}")
print(f"细筛层通过: {result['layer2_pass']} / 剔除: {result['layer2_fail']}")
print(f"需人工审核: {len(result['human_review_needed'])} 条")
for i, c in enumerate(result["human_review_needed"], 1):
print(f" {i}. {c}")
运行前设置 API Key:
export OPENAI_API_KEY="sk-..."
python eval_funnel.py
预期输出类似:
[广口层] obviously_bad=False | 与用户兴趣相关,理由具体
[广口层] obviously_bad=True | 与用户兴趣完全无关
[广口层] obviously_bad=False | 与后朋克兴趣相关
[广口层] obviously_bad=False | 虽然泛泛但不算明显劣质
[广口层] obviously_bad=False | 相关且具体
===== 漏斗结果 =====
总输入: 5
广口层通过: 4 / 剔除: 1
细筛层通过: 3 / 剔除: 1
需人工审核: 3 条
注意泛泛的那条"推荐一些摇滚音乐"在广口层没被剔除(不算明显劣质),但在细筛层因为 specificity 低而被过滤——这正是漏斗的价值。
为什么漏斗比一刀切更适合实验
Spotify 强调的"funnel, not a fork"不只是工程技巧,更是实验方法论:
1. 降低单点 eval 的风险压力。 任何一个 LLM judge prompt 都有误判率。当你把所有压力压在一个判断点上,误判率直接等于系统故障率。漏斗把误判分散到多层,每一层的误判被下一层兜住。
2. 让数据可分析。 一刀切只给你一个通过率数字。漏斗给你每一层的通过率、剔除原因分布、维度评分分布。你可以回答"最近通过率下降是因为广口层太严还是细筛层标准变了"这类问题。
3. 成本可控。 广口层用便宜的小模型(gpt-4o-mini),细筛层用贵但精准的大模型(gpt-4o),人工层只处理极少量样本。总成本远低于"全部用大模型评估"或"全部人工审核"。
4. 支持渐进式收紧。 新功能上线时,漏斗可以先放宽标准收集数据,再根据人工审核反馈逐步收紧细筛层的阈值。这比"一开始就设严标准、上线受阻"更务实。
落地时的几个坑
| 常见问题 | 建议 |
|---|---|
| 广口层太严,大量好输出被误杀 | 广口层只剔除"明显劣质",宁可多放过 |
| 细筛层评分不稳定,同一输入分数波动大 | 用 temperature=0,多跑几次取平均,或用两个模型交叉验证 |
| 人工层样本太少,反馈不够统计显著 | 定期从细筛层通过的样本中随机抽样 5-10% 做人工审核,不只审被剔除的 |
| 漏斗层数太多,延迟叠加 | 广口层可以批量并行,细筛层只处理少量样本,总延迟通常可控 |
上手清单
- 先定义"明显劣质"的标准。 这是广口层的核心。标准越清晰,广口层越稳定。
- 选两个模型分工。 小模型做广口层,大模型做细筛层。成本和质量的平衡点就在这里。
- 用 Pydantic 或 JSON Schema 约束输出格式。 LLM eval 最怕格式漂移,结构化输出是稳定性的基础。
- 保留每一层的原始输出。 不要只存最终 pass/fail,存每一层的评分和理由,后续分析才有数据。
- 设一个随机抽样通道。 不管自动评估多自信,始终随机抽一小比例送人工审核,这是校准自动 eval 的唯一可靠方式。
漏斗思维的本质是:承认 LLM eval 会犯错,然后用多层结构让错误可以被后续纠正,而不是让一个错误变成最终判决。 这比追求"完美的 eval prompt"更现实,也比"干脆全靠人工"更高效。