为什么评估
评估是迭代的仪表盘。它让你回答三个问题：现在有多好、改动是变好还是变差、哪里最拉胯。没有它，提示词和模型迭代全靠「感觉」，无法向团队或老板交代。
实践中不存在一个万能指标。成熟团队会把评估拆成三层：底层是自动指标，便宜到可以每次提交都跑；中层是 LLM 裁判，成本适中、能覆盖开放式生成；顶层是人工抽检，最贵但也最接近真实用户判断，用来校准下面两层是否可信。
先定指标再改模型：任何优化动作开始之前，先固化一份评测集和一条可复现的评测命令。否则你永远说不清「这次改动到底有没有用」。
自动指标
- BLEU / ROUGE：基于 n-gram 重叠，适合翻译、摘要，但对语义等价不敏感。
- BERTScore：用语义向量相似度，比字面重叠更贴近人类判断。
- LLM-as-Judge：用强模型当裁判给打分或 pairwise 比较，灵活但需防自我偏好，建议多模型交叉验证。
n-gram 类指标：BLEU 与 ROUGE
中文没有天然空格，直接把字符串丢进 nltk 会按字符切分，导致 BLEU 虚高。正确做法是先用 jieba 分词，再计算重叠。短文本还必须开平滑函数，否则 4-gram 缺失时会直接得 0 分。
# 依赖：pip install nltk rouge-score jieba
# 文件：metrics/ngram.py
from nltk.translate.bleu_score import sentence_bleu, SmoothingFunction
from rouge_score import rouge_scorer
import jieba
# method1 = 分子加 epsilon，避免高阶 n-gram 缺失时整体归零
_SMOOTH = SmoothingFunction().method1
_ROUGE = rouge_scorer.RougeScorer(
["rouge1", "rouge2", "rougeL"], use_stemmer=False
)
def tokenize_zh(text: str) -> list:
"""中文分词，过滤掉纯空白 token。"""
return [t for t in jieba.cut(text.strip()) if t.strip()]
def bleu4(reference: str, candidate: str) -> float:
"""BLEU-4，四元语法等权。参考句可以有多条，这里用单条。"""
refs = [tokenize_zh(reference)]
cand = tokenize_zh(candidate)
if not cand:
return 0.0
return sentence_bleu(
refs, cand,
weights=(0.25, 0.25, 0.25, 0.25),
smoothing_function=_SMOOTH,
)
def rouge_zh(reference: str, candidate: str) -> dict:
"""rouge-score 默认按空格切词，所以先把分词结果用空格拼回去。"""
ref = " ".join(tokenize_zh(reference))
cand = " ".join(tokenize_zh(candidate))
scores = _ROUGE.score(ref, cand)
return {name: round(s.fmeasure, 4) for name, s in scores.items()}
if __name__ == "__main__":
ref = "向量数据库通过近似最近邻算法加速相似度检索。"
hyp = "向量库用 ANN 算法来加快相似度搜索的速度。"
print("BLEU-4 =", round(bleu4(ref, hyp), 4))
print("ROUGE =", rouge_zh(ref, hyp))
# BLEU-4 = 0.0271 ROUGE = {'rouge1': 0.4, 'rouge2': 0.1176, 'rougeL': 0.4}
# 两句语义几乎等价，但字面重叠低，分数被严重低估，这就是 n-gram 指标的天花板语义相似度：嵌入余弦与 BERTScore
上面那个例子说明了问题：换个说法分数就崩。用嵌入模型把两句话映射到同一向量空间再算余弦，可以把「同义改写」识别出来。中文推荐 BAAI/bge-small-zh-v1.5，体积小、离线可跑。
# 依赖：pip install sentence-transformers bert-score
# 文件：metrics/semantic.py
from functools import lru_cache
import numpy as np
from sentence_transformers import SentenceTransformer
# 模型只加载一次，避免在批量评测里反复初始化
@lru_cache(maxsize=1)
def get_encoder(name: str = "BAAI/bge-small-zh-v1.5"):
return SentenceTransformer(name)
def semantic_sim(reference: str, candidate: str) -> float:
"""归一化后点积即余弦相似度，取值区间约 [0, 1]。"""
enc = get_encoder()
vecs = enc.encode([reference, candidate], normalize_embeddings=True)
return float(np.dot(vecs[0], vecs[1]))
def semantic_sim_batch(refs: list, cands: list) -> list:
"""批量版本，一次前向比逐条快 5 到 10 倍。"""
enc = get_encoder()
a = enc.encode(refs, normalize_embeddings=True, batch_size=64)
b = enc.encode(cands, normalize_embeddings=True, batch_size=64)
return (a * b).sum(axis=1).tolist()
def bertscore_zh(refs: list, cands: list) -> dict:
"""BERTScore 做 token 级软对齐，比整句余弦更细粒度。"""
from bert_score import score
P, R, F1 = score(cands, refs, lang="zh", rescale_with_baseline=True)
return {
"precision": round(P.mean().item(), 4),
"recall": round(R.mean().item(), 4),
"f1": round(F1.mean().item(), 4),
}
if __name__ == "__main__":
ref = "向量数据库通过近似最近邻算法加速相似度检索。"
hyp = "向量库用 ANN 算法来加快相似度搜索的速度。"
print("cosine =", round(semantic_sim(ref, hyp), 4))
# cosine = 0.8912 语义等价被正确识别，对比 BLEU 的 0.0271指标选型速查
不要一股脑全算。按任务类型挑 1 到 2 个主指标，其余作为诊断辅助。
# 文件：metrics/router.py 按任务类型路由到合适的指标组合
from metrics.ngram import bleu4, rouge_zh
from metrics.semantic import semantic_sim
METRIC_PLAN = {
"translation": ["bleu4", "semantic"], # 有唯一参考译文，字面重叠有意义
"summarization": ["rougeL", "semantic"], # 关注信息覆盖，ROUGE-L 看最长公共子序列
"qa_closed": ["exact_match"], # 答案唯一，直接精确匹配
"qa_open": ["semantic", "llm_judge"], # 开放问答，字面指标基本失效
"creative": ["llm_judge"], # 创作类没有参考答案，只能靠裁判
"code": ["pass_at_k"], # 代码看单测通过率，别看文本相似度
}
def exact_match(reference: str, candidate: str) -> float:
"""归一化后严格比对：去空白、统一全半角标点、忽略大小写。"""
table = str.maketrans("，。！？；：（）", ",.!?;:()")
norm = lambda s: s.strip().lower().translate(table).replace(" ", "")
return 1.0 if norm(reference) == norm(candidate) else 0.0
def compute(task: str, reference: str, candidate: str) -> dict:
"""只计算该任务类型需要的指标，llm_judge 走单独的异步通道。"""
out = {}
for m in METRIC_PLAN.get(task, ["semantic"]):
if m == "bleu4":
out[m] = round(bleu4(reference, candidate), 4)
elif m == "rougeL":
out[m] = rouge_zh(reference, candidate)["rougeL"]
elif m == "semantic":
out[m] = round(semantic_sim(reference, candidate), 4)
elif m == "exact_match":
out[m] = exact_match(reference, candidate)
return out指标陷阱：ROUGE 高不代表答案好；LLM 裁判可能偏爱更长的回答。指标是线索，不是结论。任何一个指标涨了，都要抽 10 条样本肉眼看一眼再下结论。
LLM-as-Judge 实战
开放式生成没有标准答案，字面指标彻底失效，这时候就要请一个更强的模型来当裁判。核心难点不在调 API，而在把评分标准写成机器可执行的 rubric：每个维度要有明确定义，每个分档要有可判定的锚点描述。
rubric 设计：维度加分档
维度不要超过 4 个，超过之后裁判会开始摆烂给中间分。分档建议用 1 / 3 / 5 三档锚点，中间的 2 和 4 让裁判自行插值，比直接给 1 到 10 的连续区间稳定得多。
# 通用四维 rubric 的分档锚点（写进裁判提示词里）
| 维度 | 1 分 | 3 分 | 5 分 |
|------|------|------|------|
| 正确性 | 存在事实错误或答非所问 | 主体正确，细节有小瑕疵 | 完全正确，可直接采用 |
| 完整性 | 遗漏关键信息，无法闭环 | 覆盖主要点，缺少边界情况 | 要点齐全，含必要前提与限制 |
| 简洁性 | 大量冗余、重复、套话 | 略显啰嗦但不影响阅读 | 无一句废话，信息密度高 |
| 可执行性 | 只有空泛结论，无法落地 | 给了方向但缺具体步骤 | 步骤具体，读完即可动手 |下面是完整可运行的单点打分裁判。关键细节：temperature=0 保证可复现，要求裁判先写理由再给分（先给分会让理由变成事后合理化），输出 JSON 并做容错解析。
# 依赖：pip install openai tenacity
# 文件：judge/pointwise.py
import json
import os
import re
from openai import OpenAI
from tenacity import retry, stop_after_attempt, wait_exponential
client = OpenAI(
api_key=os.environ["JUDGE_API_KEY"],
base_url=os.environ.get("JUDGE_BASE_URL", "https://api.openai.com/v1"),
)
JUDGE_MODEL = os.environ.get("JUDGE_MODEL", "gpt-4o")
RUBRIC = """正确性：1 分存在事实错误或答非所问；3 分主体正确细节有瑕疵；5 分完全正确可直接采用。
完整性：1 分遗漏关键信息；3 分覆盖主要点但缺边界情况；5 分要点齐全含前提与限制。
简洁性：1 分大量冗余套话；3 分略显啰嗦；5 分无一句废话。
可执行性：1 分只有空泛结论；3 分有方向缺步骤；5 分步骤具体读完即可动手。"""
SYSTEM = """你是一名严格的中文内容评测裁判。你的判断必须只基于评分标准，
不受回答长度、排版华丽程度、是否使用列表等表面因素影响。
如果回答很长但信息密度低，简洁性必须给低分。
你必须先写评分理由，再给出分数。输出严格的 JSON，不要加代码块围栏。"""
TEMPLATE = """【评分标准】
{rubric}
【用户问题】
{question}
【参考答案】（仅供对照，回答不必逐字一致）
{reference}
【待评回答】
{answer}
请按下面的 JSON 结构输出，每个维度给 1 到 5 的整数：
{{"reason": "80 字以内的评分理由",
"correctness": 整数, "completeness": 整数,
"conciseness": 整数, "actionability": 整数}}"""
DIMS = ["correctness", "completeness", "conciseness", "actionability"]
# 权重按业务侧重调整，此处正确性占一半
WEIGHTS = {"correctness": 0.4, "completeness": 0.25,
"conciseness": 0.15, "actionability": 0.2}
def parse_json(text: str) -> dict:
"""裁判偶尔会带 ```json 围栏或前后废话，用括号配对兜底。"""
text = text.strip()
text = re.sub(r"^```(?:json)?|```$", "", text, flags=re.M).strip()
try:
return json.loads(text)
except json.JSONDecodeError:
start, depth = text.find("{"), 0
for i in range(start, len(text)):
depth += (text[i] == "{") - (text[i] == "}")
if depth == 0:
return json.loads(text[start:i + 1])
raise ValueError(f"无法解析裁判输出: {text[:120]}")
@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=8))
def judge_one(question: str, answer: str, reference: str = "（无）") -> dict:
"""对单条回答打分，返回各维度分数与加权总分。"""
resp = client.chat.completions.create(
model=JUDGE_MODEL,
temperature=0, # 打分必须可复现
seed=42, # 支持 seed 的模型会更稳定
max_tokens=400,
messages=[
{"role": "system", "content": SYSTEM},
{"role": "user", "content": TEMPLATE.format(
rubric=RUBRIC, question=question,
reference=reference, answer=answer)},
],
)
data = parse_json(resp.choices[0].message.content)
# 裁判可能给出 0 或 6 这类越界值，统一裁剪到 1 至 5
scores = {d: min(5, max(1, int(data.get(d, 3)))) for d in DIMS}
scores["overall"] = round(sum(scores[d] * WEIGHTS[d] for d in DIMS), 3)
scores["reason"] = data.get("reason", "")
return scores
if __name__ == "__main__":
r = judge_one(
question="HNSW 索引的 efSearch 参数调大会有什么影响？",
answer="efSearch 越大，检索时探索的候选邻居越多，召回率上升但延迟增加。"
"典型取值 64 到 256，建议按 P99 延迟预算二分调优。",
reference="efSearch 控制搜索时的候选队列长度，增大提升召回、增加耗时。",
)
print(json.dumps(r, ensure_ascii=False, indent=2))
# {"correctness": 5, "completeness": 4, "conciseness": 5,
# "actionability": 5, "overall": 4.75, "reason": "..."}pairwise 对比与位置偏差
比较两个模型时，pairwise 对战比各自打分更灵敏，因为裁判只需判断「谁更好」而不用把握绝对刻度。但裁判存在位置偏差：同样两个回答，放在前面的那个更容易赢。解法是双向交换，正反各问一次，两次结论一致才计胜负，不一致就判平局。
# 文件：judge/pairwise.py 双向交换的 pairwise 裁判
import os
from collections import Counter
from judge.pointwise import client, JUDGE_MODEL, parse_json, RUBRIC
PAIR_SYSTEM = """你是严格的对比评测裁判。请依据评分标准判断哪个回答更好。
不要因为回答更长、排版更花哨就认为它更好。若两者质量确实接近，允许判平局。
输出严格 JSON，不要加代码块围栏。"""
PAIR_TEMPLATE = """【评分标准】
{rubric}
【用户问题】
{question}
【回答 甲】
{left}
【回答 乙】
{right}
输出：{{"reason": "60 字以内理由", "winner": "甲" 或 "乙" 或 "平"}}"""
def _ask(question: str, left: str, right: str) -> str:
resp = client.chat.completions.create(
model=JUDGE_MODEL, temperature=0, seed=42, max_tokens=300,
messages=[
{"role": "system", "content": PAIR_SYSTEM},
{"role": "user", "content": PAIR_TEMPLATE.format(
rubric=RUBRIC, question=question, left=left, right=right)},
],
)
return parse_json(resp.choices[0].message.content).get("winner", "平")
def compare(question: str, answer_a: str, answer_b: str) -> str:
"""返回 'A' / 'B' / 'TIE'。正反各问一次，冲突即判平局。"""
# 第 1 轮：甲位放 A
r1 = _ask(question, answer_a, answer_b)
win1 = {"甲": "A", "乙": "B"}.get(r1, "TIE")
# 第 2 轮：甲位放 B，结论要反向映射回来
r2 = _ask(question, answer_b, answer_a)
win2 = {"甲": "B", "乙": "A"}.get(r2, "TIE")
return win1 if win1 == win2 else "TIE"
def arena(cases: list, model_a: str, model_b: str) -> dict:
"""cases 每项形如 {'question':..., 'a':..., 'b':...}，返回胜率统计。"""
tally = Counter(compare(c["question"], c["a"], c["b"]) for c in cases)
n = max(len(cases), 1)
# 平局按半分计入，得到 0 至 1 的胜率
win_rate_a = (tally["A"] + 0.5 * tally["TIE"]) / n
return {
model_a: tally["A"], model_b: tally["B"], "tie": tally["TIE"],
"win_rate_a": round(win_rate_a, 4),
"tie_ratio": round(tally["TIE"] / n, 4), # 高于 0.25 要检查 rubric
}自我偏好偏差：用 GPT 系列当裁判去评 GPT 系列的输出，分数会系统性偏高。跨厂商选裁判，或者用两个不同厂商的裁判取平均，并把它们的一致率作为可信度指标记录下来。
评测流水线
零散跑脚本撑不过三次迭代。把评测固化成一条流水线：评测集 → 批量生成 → 指标计算 → 聚合报告 → 回归门禁，每一步产物落盘成文件，任何一次结果都能追溯和重放。
并发生成与断点续跑
评测集上千条时，串行调用要跑几十分钟。用线程池并发加上磁盘缓存，重跑时命中缓存的条目直接跳过。
# 文件：pipeline/generate.py 批量生成被测模型的回答
import hashlib
import json
import pathlib
from concurrent.futures import ThreadPoolExecutor, as_completed
from openai import OpenAI
CACHE_DIR = pathlib.Path(".eval_cache")
CACHE_DIR.mkdir(exist_ok=True)
def cache_key(model: str, prompt: str) -> pathlib.Path:
"""模型名加提示词的哈希做缓存文件名，换模型或换提示词自动失效。"""
h = hashlib.sha256(f"{model}||{prompt}".encode()).hexdigest()[:20]
return CACHE_DIR / f"{h}.txt"
def generate_one(client: OpenAI, model: str, prompt: str) -> str:
path = cache_key(model, prompt)
if path.exists():
return path.read_text(encoding="utf-8")
resp = client.chat.completions.create(
model=model, temperature=0, max_tokens=1024,
messages=[{"role": "user", "content": prompt}],
)
text = resp.choices[0].message.content.strip()
path.write_text(text, encoding="utf-8")
return text
def generate_all(client: OpenAI, model: str, cases: list,
workers: int = 8) -> list:
"""并发生成，保持与输入相同的顺序，单条失败不拖垮整体。"""
results = [None] * len(cases)
with ThreadPoolExecutor(max_workers=workers) as pool:
futures = {
pool.submit(generate_one, client, model, c["prompt"]): i
for i, c in enumerate(cases)
}
for fut in as_completed(futures):
idx = futures[fut]
try:
results[idx] = fut.result()
except Exception as e:
# 记录失败但不中断，报告里单独统计失败率
results[idx] = f"__ERROR__: {type(e).__name__}: {e}"
return results聚合与报告生成
报告要同时给出总体分和最差样本。平均分告诉你水位，最差样本告诉你该修什么。
# 文件：pipeline/report.py 聚合打分并生成 Markdown 报告
import json
import pathlib
import statistics as st
from datetime import datetime
def aggregate(records: list) -> dict:
"""records 每项含 metrics 字典，输出各指标的均值、中位数与 P10。"""
keys = sorted({k for r in records for k in r["metrics"]})
summary = {}
for k in keys:
vals = [r["metrics"][k] for r in records if k in r["metrics"]]
vals.sort()
summary[k] = {
"mean": round(st.mean(vals), 4),
"median": round(st.median(vals), 4),
# P10 反映长尾表现，均值好看但 P10 崩了说明有系统性坏例
"p10": round(vals[max(0, int(len(vals) * 0.1) - 1)], 4),
}
return summary
def worst_cases(records: list, key: str = "overall", k: int = 5) -> list:
"""按指定指标取最差的 k 条，这些才是下一轮优化的抓手。"""
scored = [r for r in records if key in r["metrics"]]
return sorted(scored, key=lambda r: r["metrics"][key])[:k]
def render_markdown(model: str, records: list, out_dir: str) -> str:
summary = aggregate(records)
fail = sum(1 for r in records if r["answer"].startswith("__ERROR__"))
lines = [
f"# 评测报告 · {model}", "",
f"- 生成时间：{datetime.now():%Y-%m-%d %H:%M}",
f"- 样本量：{len(records)} 条，调用失败 {fail} 条", "",
"## 指标汇总", "",
"| 指标 | 均值 | 中位数 | P10 |",
"|------|------|--------|-----|",
]
for k, v in summary.items():
lines.append(f"| {k} | {v['mean']} | {v['median']} | {v['p10']} |")
lines += ["", "## 最差 5 条样本", ""]
for i, r in enumerate(worst_cases(records), 1):
lines += [
f"### {i}. 得分 {r['metrics'].get('overall')} 分类 {r.get('category', '未分类')}",
f"- 问题：{r['question']}",
f"- 回答：{r['answer'][:200]}",
f"- 裁判理由：{r['metrics'].get('reason', '')}", "",
]
path = pathlib.Path(out_dir) / "report.md"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("\n".join(lines), encoding="utf-8")
(path.parent / "summary.json").write_text(
json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
return str(path)回归门禁：让评测卡住劣化的合并
把评测接进 CI，与上一次基线对比。掉分超过阈值就让流水线失败，这是评估体系真正产生约束力的地方。
# 文件：pipeline/gate.py 回归门禁，退出码非 0 时 CI 会失败
import json
import sys
import pathlib
# 每个指标允许的最大跌幅（绝对值）。语义相似度和裁判分尺度不同，分别设定
THRESHOLDS = {
"overall": 0.15, # 裁判加权分，满分 5，跌 0.15 以上视为劣化
"semantic": 0.02, # 余弦相似度，满分 1
"rougeL": 0.03,
}
def main(baseline_path: str, current_path: str) -> int:
base = json.loads(pathlib.Path(baseline_path).read_text(encoding="utf-8"))
curr = json.loads(pathlib.Path(current_path).read_text(encoding="utf-8"))
failed = []
for metric, max_drop in THRESHOLDS.items():
if metric not in base or metric not in curr:
continue
delta = curr[metric]["mean"] - base[metric]["mean"]
flag = "FAIL" if delta < -max_drop else "ok"
print(f"{flag:4} {metric:10} {base[metric]['mean']:.4f} -> "
f"{curr[metric]['mean']:.4f} delta={delta:+.4f}")
if flag == "FAIL":
failed.append(metric)
if failed:
print(f"\n评测回归门禁未通过，劣化指标：{', '.join(failed)}")
return 1
print("\n评测回归门禁通过。")
return 0
if __name__ == "__main__":
# 用法：python -m pipeline.gate runs/baseline/summary.json runs/pr-142/summary.json
sys.exit(main(sys.argv[1], sys.argv[2]))评测集与打分表
评测集的质量决定评估结论的上限。一个 50 条精心设计的评测集，价值远高于 5000 条随机抓取的日志。
构建建议
- 来源真实：优先从线上真实请求里抽，而不是自己编。编出来的题会系统性偏离用户真实分布。
- 分层覆盖：按业务分类分层，每层至少 15 条。同时刻意保留一批困难样本（歧义提问、超纲问题、多轮指代）占比 20% 左右。
- 包含拒答题：10% 左右的题应该是模型「必须拒绝或声明不知道」的，用来检测幻觉倾向。
- 参考答案分级：开放题不写唯一答案，写要点清单（必须提到 A、B，不得出现 C），裁判据此判定更稳。
- 版本冻结：评测集入 Git 并打 tag。改题目等于换尺子，历史数据立刻不可比。
- 防污染：不要把评测集用于训练或写进提示词示例，也不要公开发布全量题目。
# 文件：data/build_eval_set.py 从线上日志分层抽样构建评测集
import json
import random
from collections import defaultdict
random.seed(20260701) # 固定随机种子，保证抽样可复现
# 目标配比：常规题 70%，困难题 20%，拒答题 10%
QUOTA = {"normal": 0.7, "hard": 0.2, "refusal": 0.1}
def stratified_sample(logs: list, total: int = 200) -> list:
"""按 category 与 difficulty 双重分层抽样。"""
buckets = defaultdict(list)
for item in logs:
buckets[(item["category"], item["difficulty"])].append(item)
categories = sorted({c for c, _ in buckets})
picked = []
for cat in categories:
for diff, ratio in QUOTA.items():
pool = buckets.get((cat, diff), [])
# 每个分类每种难度应抽的条数，至少 1 条
n = max(1, round(total * ratio / len(categories)))
picked += random.sample(pool, min(n, len(pool)))
return picked
def to_jsonl(cases: list, path: str) -> None:
"""落盘为标准 schema，key_points 是开放题的判分依据。"""
with open(path, "w", encoding="utf-8") as f:
for i, c in enumerate(cases, 1):
row = {
"id": f"eval-{i:04d}",
"category": c["category"], # 如 检索问答 / 代码解释 / 闲聊
"difficulty": c["difficulty"], # normal / hard / refusal
"prompt": c["question"],
"reference": c.get("reference", ""),
"key_points": c.get("key_points", []), # 必须命中的要点
"forbidden": c.get("forbidden", []), # 不得出现的说法
"task_type": c.get("task_type", "qa_open"),
}
f.write(json.dumps(row, ensure_ascii=False) + "\n")
if __name__ == "__main__":
logs = [json.loads(l) for l in open("data/online_logs.jsonl", encoding="utf-8")]
cases = stratified_sample(logs, total=200)
to_jsonl(cases, "data/eval_set_v1.jsonl")
print(f"已生成 {len(cases)} 条评测样本")人工打分表模板
人工抽检时直接把下表发给标注同学。关键是每一档都要能被独立判定，避免出现「差不多」的模糊地带。分歧样本单独拉出来开会对齐，对齐结论反过来补进 rubric。
# 人工评估打分表 v1（配合 data/eval_set_v1.jsonl 使用）
| 样本 ID | 分类 | 正确性 1-5 | 完整性 1-5 | 简洁性 1-5 | 安全合规 通过/拒绝 | 是否幻觉 是/否 | 备注 |
|---------|------|-----------|-----------|-----------|------------------|--------------|------|
| eval-0001 | 检索问答 | 5 | 4 | 5 | 通过 | 否 | 边界条件未提及 |
| eval-0002 | 代码解释 | 3 | 3 | 2 | 通过 | 否 | 冗余示例过多 |
| eval-0003 | 拒答检测 | 1 | 1 | 4 | 通过 | 是 | 编造了不存在的 API |
# 判定口径（必须逐条对照，不允许凭印象打分）
# 正确性：与 key_points 逐条核对，命中率 100% 给 5，命中一半给 3，出现事实错误直接给 1
# 完整性：key_points 全覆盖给 5，漏 1 条给 3，漏 2 条及以上给 1
# 简洁性：无冗余给 5，有重复段落给 3，大段套话给 1
# 安全合规：命中 forbidden 列表任一项即判拒绝
# 是否幻觉：出现无法在参考资料中验证的具体数字、人名、API 名，判为「是」权威基准
横向对比模型能力，可参考公开榜单。中文综合评测看 SuperCLUE：它按月发布大模型在理解、生成、对齐等维度的榜单，适合快速定位模型强弱项。英文与代码方向可关注 MMLU、GPQA、HumanEval、SWE-bench 等。
但榜单只能用来缩小候选范围，不能替代你自己的业务评测。原因有二：一是数据污染，热门基准的题目很可能已经进入训练语料；二是分布错配，榜单题目与你的真实业务场景往往差得很远。正确用法是「榜单选 3 个候选，业务评测集定终局」。
# 评估闭环的最小形态
for case in dataset:
out = model.generate(case.prompt)
score = judge(out, case.reference)
report(mean=avg(score), low=worst_cases(score))# 文件：pipeline/run.py 把前面所有模块串成一条命令
import json
import pathlib
import sys
from datetime import datetime
from openai import OpenAI
from metrics.router import compute
from judge.pointwise import judge_one
from pipeline.generate import generate_all
from pipeline.report import render_markdown
def load_jsonl(path: str) -> list:
return [json.loads(l) for l in open(path, encoding="utf-8") if l.strip()]
def run(model: str, eval_set: str, judge_ratio: float = 1.0) -> str:
cases = load_jsonl(eval_set)
client = OpenAI()
answers = generate_all(client, model, cases, workers=8)
run_id = f"{model.replace('/', '_')}-{datetime.now():%m%d-%H%M}"
out_dir = pathlib.Path("runs") / run_id
records = []
for i, (case, ans) in enumerate(zip(cases, answers)):
metrics = compute(case["task_type"], case["reference"], ans)
# 裁判较贵，可只对前 judge_ratio 比例的样本调用
if i < len(cases) * judge_ratio and not ans.startswith("__ERROR__"):
metrics.update(judge_one(case["prompt"], ans, case["reference"]))
records.append({
"id": case["id"], "category": case["category"],
"question": case["prompt"], "answer": ans, "metrics": metrics,
})
out_dir.mkdir(parents=True, exist_ok=True)
with open(out_dir / "records.jsonl", "w", encoding="utf-8") as f:
for r in records:
f.write(json.dumps(r, ensure_ascii=False) + "\n")
return render_markdown(model, records, str(out_dir))
if __name__ == "__main__":
# 用法：python -m pipeline.run gpt-4o-mini data/eval_set_v1.jsonl
print("报告已生成：", run(sys.argv[1], sys.argv[2]))人工评估
自动指标无法覆盖的事实正确性、安全性、风格，仍需人工抽检。建议：固定评分 rubric（如 1-5 分三维度）、双盲标注、对分歧样本复盘。人工评估样本虽少，却是指标体系的「校准源」。
但人工评估自身也需要被评估。如果两位标注员对同一批样本的打分一致性很低，那这批数据就不能拿来校准任何东西。用 Cohen's Kappa 衡量一致性：0.6 以上算可接受，0.8 以上算优秀。低于 0.4 说明 rubric 写得不清楚，先回去改标准。
# 依赖：pip install scikit-learn scipy
# 文件：human/agreement.py 标注一致性与裁判可信度校准
import json
from sklearn.metrics import cohen_kappa_score
from scipy.stats import spearmanr
def annotator_agreement(rater_a: list, rater_b: list) -> dict:
"""两位标注员在同一批样本上的一致性。分数是有序等级，用线性加权 kappa。"""
kappa = cohen_kappa_score(rater_a, rater_b, weights="linear")
exact = sum(x == y for x, y in zip(rater_a, rater_b)) / len(rater_a)
within1 = sum(abs(x - y) <= 1 for x, y in zip(rater_a, rater_b)) / len(rater_a)
return {
"kappa": round(kappa, 4), # < 0.4 需要重写 rubric
"exact_match": round(exact, 4),
"within_1": round(within1, 4), # 相差不超过 1 分的比例
}
def judge_calibration(human_scores: list, judge_scores: list) -> dict:
"""LLM 裁判与人工的相关性。这是裁判能否替代人工的核心证据。"""
rho, p = spearmanr(human_scores, judge_scores)
bias = sum(j - h for h, j in zip(human_scores, judge_scores)) / len(human_scores)
return {
"spearman": round(float(rho), 4), # > 0.7 可放心用裁判跑大批量
"p_value": round(float(p), 6),
"mean_bias": round(bias, 4), # 正值说明裁判系统性偏松
}
if __name__ == "__main__":
rows = [json.loads(l) for l in open("human/labeled.jsonl", encoding="utf-8")]
a = [r["rater_a"] for r in rows]
b = [r["rater_b"] for r in rows]
j = [r["judge"] for r in rows]
human = [(x + y) / 2 for x, y in zip(a, b)]
print("标注员一致性：", annotator_agreement(a, b))
print("裁判校准度：", judge_calibration(human, j))
# 标注员一致性： {'kappa': 0.742, 'exact_match': 0.62, 'within_1': 0.94}
# 裁判校准度： {'spearman': 0.781, 'p_value': 0.0, 'mean_bias': 0.31}
# mean_bias 为正 0.31，说明裁判比人工平均松 0.3 分，报告里应扣减这个偏移RAG / Agent 专项
RAG 要分别评检索（召回率、NDCG）与生成（忠实度、答案相关）；Agent 要评任务成功率、工具调用准确率与步数效率。框架上可用 RAGAS、TruLens 做自动打分。
分开评的意义在于定位责任：答案错了，到底是没检索到正确文档，还是检索到了但模型没用好。前者去调索引和切分策略，后者去调提示词，两条路径完全不同。
RAG：检索与生成分离评估
# 依赖：pip install ragas datasets
# 文件：special/rag_eval.py
import math
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
faithfulness, # 答案是否有上下文支撑，直接反映幻觉程度
answer_relevancy, # 答案是否回应了问题
context_precision, # 召回的片段里有多少是真的相关
context_recall, # 标准答案所需信息被召回了多少
)
def recall_at_k(retrieved_ids: list, gold_ids: list, k: int = 5) -> float:
"""检索层的硬指标，不依赖任何大模型，跑得飞快。"""
hit = len(set(retrieved_ids[:k]) & set(gold_ids))
return hit / max(len(gold_ids), 1)
def ndcg_at_k(retrieved_ids: list, gold_ids: list, k: int = 5) -> float:
"""考虑排序位置：正确文档排第 1 位比排第 5 位更有价值。"""
gold = set(gold_ids)
dcg = sum(1 / math.log2(i + 2)
for i, doc in enumerate(retrieved_ids[:k]) if doc in gold)
idcg = sum(1 / math.log2(i + 2) for i in range(min(len(gold), k)))
return dcg / idcg if idcg > 0 else 0.0
def eval_generation(samples: list) -> dict:
"""samples 每项需含 question / answer / contexts / ground_truth。"""
ds = Dataset.from_list(samples)
result = evaluate(ds, metrics=[
faithfulness, answer_relevancy, context_precision, context_recall,
])
return {k: round(float(v), 4) for k, v in result.items()}
if __name__ == "__main__":
# 检索层先跑，快且便宜。召回率低于 0.8 就别急着优化生成层
print("recall@5 =", recall_at_k(["d3", "d7", "d1"], ["d1", "d3"]))
print("ndcg@5 =", round(ndcg_at_k(["d3", "d7", "d1"], ["d1", "d3"]), 4))
# recall@5 = 1.0 ndcg@5 = 0.7654 召回全中但排序不佳，该上重排模型RAGAS 完整评测管线：端到端可运行
基础代码只演示了单次调用。生产环境中你需要一条管线来回答三个问题：每个指标值是多少、chunk_size 怎么调、结果怎么可视化。下面这条管线把这三个问题一并解决。
核心思路：用 RAGAS 的 evaluate 一次性跑完四个指标，然后对不同 chunk_size 重复实验，最后输出对比表格和 ASCII 图表。
# 依赖：pip install ragas datasets pandas tabulate
# 文件：special/ragas_pipeline.py RAGAS 端到端评测管线
import json
import pandas as pd
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
context_precision,
context_recall,
faithfulness,
answer_relevancy,
)
from tabulate import tabulate
def load_samples(path: str) -> list:
"""加载预处理的 RAG 评测样本，每项含 question / answer / contexts / ground_truth。"""
return [json.loads(l) for l in open(path, encoding="utf-8") if l.strip()]
def run_ragas(samples: list) -> dict:
"""四个指标一次性跑完，返回均值字典。"""
ds = Dataset.from_list(samples)
result = evaluate(ds, metrics=[
context_precision, context_recall, faithfulness, answer_relevancy,
])
return {k: round(float(v), 4) for k, v in result.items()}
def compare_chunk_sizes(base_samples: list, chunk_sizes: list) -> pd.DataFrame:
"""用不同 chunk_size 重新构建检索上下文，对比指标变化。"""
rows = []
for cs in chunk_sizes:
# 模拟不同分块策略下的检索结果：越大越全但噪声也多
modified = []
for s in base_samples:
item = dict(s)
# 实际场景中这里用 chunk_size 重新分块并检索，此处简化
item["contexts"] = item["contexts"][:max(1, int(1000 / cs))]
modified.append(item)
metrics = run_ragas(modified)
metrics["chunk_size"] = cs
rows.append(metrics)
return pd.DataFrame(rows)
def print_table(df: pd.DataFrame) -> str:
"""ASCII 表格打印，适合放进报告或 CI 日志。"""
cols = ["chunk_size", "context_precision", "context_recall",
"faithfulness", "answer_relevancy"]
df = df[cols].rename(columns={
"context_precision": "精度",
"context_recall": "召回",
"faithfulness": "忠实度",
"answer_relevancy": "相关性",
})
return tabulate(df, headers="keys", tablefmt="grid", floatfmt=".4f")
def print_ascii_chart(df: pd.DataFrame, metric: str, label: str,
width: int = 40) -> str:
"""简易 ASCII 柱状图，不需要任何绘图库。"""
lines = [f"\n{label} 随 chunk_size 变化:"]
vals = df[metric].tolist()
sizes = df["chunk_size"].tolist()
max_val = max(vals) if vals else 1
for cs, v in zip(sizes, vals):
bar_len = int(v / max_val * width)
bar = "=" * bar_len + " " * (width - bar_len)
lines.append(f"chunk={cs:4d} |{bar}| {v:.4f}")
return "\n".join(lines)
if __name__ == "__main__":
# ---- Step 1: 单次评测 ----
samples = load_samples("data/rag_samples.jsonl")
baseline = run_ragas(samples)
print("\n=== RAGAS 评测结果 ===")
for k, v in baseline.items():
print(f"{k:20s}: {v:.4f}")
# ---- Step 2: chunk_size 对比 ----
df = compare_chunk_sizes(samples, chunk_sizes=[128, 256, 512, 768, 1024])
print("\n=== Chunk Size 对比表 ===")
print(print_table(df))
# ---- Step 3: ASCII 柱状图 ----
print(print_ascii_chart(df, "faithfulness", "忠实度"))
print(print_ascii_chart(df, "context_precision", "上下文精度"))
# 典型输出解读：
# chunk_size 越大，context_recall 越高（找到更多相关内容）
# 但 context_precision 和 faithfulness 可能拐头向下（噪声过多）
# 最佳取值是 recall 拐点之后、precision 拐点之前的位置chunk_size 调优经验：context_recall 随 chunk_size 单调递增（大窗口总是覆盖更多），但 context_precision 和 faithfulness 在某个临界点后会下降。把两个指标画在同一张图上，它们在 chunk_size 轴上的交点附近就是最优值。
DeepEval 评测框架：LLM-as-a-Judge 专业版
RAGAS 覆盖 RAG 场景，DeepEval 则提供更通用的 LLM-as-a-Judge 框架，支持自定义 metric、批量评测和结构化报告。它的优势在于：自定义 metric 只需定义一个 Python 类，框架自动处理评分、聚合和可视化。
# 依赖：pip install deepeval
# 文件：special/deepeval_eval.py DeepEval 自定义评测流水线
from deepeval import evaluate
from deepeval.metrics import GEval, HallucinationMetric, AnswerRelevancyMetric
from deepeval.test_case import LLMTestCase
from deepeval.dataset import EvaluationDataset
# ---- 内置 metric：Hallucination ----
def eval_hallucination(cases: list) -> list:
"""检测回答是否包含幻觉（与上下文矛盾的内容）。"""
metric = HallucinationMetric(threshold=0.5)
results = []
for c in cases:
test_case = LLMTestCase(
input=c["question"],
actual_output=c["answer"],
context=c["contexts"],
)
metric.measure(test_case)
results.append({
"question": c["question"],
"score": metric.score,
"reason": metric.reason,
})
return results
# ---- 自定义 metric：是否包含引用 ----
class CitationMetric(GEval):
"""检测回答中是否包含对来源的引用（如 [1]、[doc_3] 或括号标注）。"""
def __init__(self, threshold: float = 0.5):
super().__init__(
name="Citation",
criteria="""
回答是否明确引用了来源？
- 1 分：每个关键陈述后面都跟了引用标记（如 [1]、[doc_a]）或来源描述。
- 0.5 分：部分关键陈述有引用，部分缺失。
- 0 分：通篇没有任何对来源的引用或指明。
""".strip(),
evaluation_steps=[
"检查回答中是否有引用标记（如 [1]、[来源]、括号注明等）",
"确认这些引用是否出现在回答的关键陈述位置",
"根据引用覆盖的完整度给出 0 / 0.5 / 1 分",
],
evaluation_params=[],
threshold=threshold,
)
# ---- 批量评测 ----
def batch_evaluate(cases: list, model: str = "gpt-4o") -> dict:
"""批量跑 Hallucination、AnswerRelevancy 和自定义 Citation。"""
test_cases = []
for c in cases:
test_cases.append(LLMTestCase(
input=c["question"],
actual_output=c["answer"],
context=c.get("contexts", []),
))
dataset = EvaluationDataset(test_cases=test_cases)
hallucination = HallucinationMetric(threshold=0.5, model=model)
relevancy = AnswerRelevancyMetric(threshold=0.5, model=model)
citation = CitationMetric(threshold=0.5)
results = evaluate(dataset, metrics=[hallucination, relevancy, citation],
print_results=False)
# 聚合每个 metric 的均值
summary = {}
for metric_result in results:
scores = [r.score for r in metric_result.test_results]
summary[metric_result.metric_name] = {
"mean": round(sum(scores) / max(len(scores), 1), 4),
"pass_rate": round(sum(1 for s in scores if s >= 0.5) / max(len(scores), 1), 4),
}
return summary
if __name__ == "__main__":
samples = [
{"question": "量子计算与经典计算的主要区别是什么？",
"answer": "量子比特可以处于叠加态（[1]）且支持纠缠操作。",
"contexts": ["量子计算使用纠缠叠加原理..."]},
]
# 单项检测
h_results = eval_hallucination(samples)
for r in h_results:
print(f"Hallucination score: {r['score']:.3f} | {r['reason'][:80]}")
# 批量报告
summary = batch_evaluate(samples * 3) # 在生产中用真实数据集
print("\n=== 批量评测报告 ===")
for name, stats in summary.items():
print(f"{name:20s} mean={stats['mean']:.3f} pass={stats['pass_rate']:.1%}")
# 典型输出：
# Hallucination mean=0.920 pass=100.0%
# Answer Relevancy mean=0.850 pass=100.0%
# Citation mean=0.670 pass=66.7%
# Citation 分数最低，说明模型较少主动引用来源，需要优化提示词评测维度矩阵
不同阶段关注不同维度。这张矩阵图帮你快速定位：当前阶段该跑哪些指标，缺哪一列说明评测体系有盲区。
用法：把当前评测体系从左上到右下过一遍，哪个格子空了就说明那个维度的那个阶段还没有度量手段，优先补齐。例如很多团队有离线准确性和相关性指标，但缺少在线安全性监控，这就是盲区。
A/B 对比实验框架：统计显著性检验
两个版本的结果差了 0.1 分，这个差异是真实的还是噪声？直觉回答不了，需要统计检验。下面的框架对两组指标做独立样本 t 检验，计算 Cohen's d 效应量，输出置信区间，让你能说出「A 在统计显著水平 p < 0.01 下优于 B」。
# 依赖：pip install scipy numpy tabulate
# 文件：pipeline/ab_test.py A/B 对比 + 统计显著性 + 置信区间
import json
import numpy as np
from pathlib import Path
from scipy import stats
from tabulate import tabulate
def load_scores(run_dir: str, metric: str = "overall") -> list:
"""从评测结果目录中提取指定指标的所有分数。"""
records_path = Path(run_dir) / "records.jsonl"
records = [json.loads(l) for l in
records_path.read_text(encoding="utf-8").splitlines() if l.strip()]
return [r["metrics"][metric] for r in records if metric in r["metrics"]]
def independent_ttest(a_scores: list, b_scores: list) -> dict:
"""独立样本 t 检验（Welch 校正，不假设方差相等）。"""
t_stat, p_value = stats.ttest_ind(a_scores, b_scores, equal_var=False)
mean_a, mean_b = np.mean(a_scores), np.mean(b_scores)
delta = mean_a - mean_b
se = np.sqrt(np.var(a_scores, ddof=1) / len(a_scores) +
np.var(b_scores, ddof=1) / len(b_scores))
ci95 = (delta - 1.96 * se, delta + 1.96 * se)
# Cohen's d 效应量（0.2 小 / 0.5 中 / 0.8 大）
pooled_std = np.sqrt((np.var(a_scores, ddof=1) + np.var(b_scores, ddof=1)) / 2)
d = delta / pooled_std if pooled_std > 0 else 0.0
return {
"mean_a": round(float(mean_a), 4),
"mean_b": round(float(mean_b), 4),
"delta": round(float(delta), 4),
"ci95_lower": round(float(ci95[0]), 4),
"ci95_upper": round(float(ci95[1]), 4),
"t_statistic": round(float(t_stat), 4),
"p_value": round(float(p_value), 6),
"cohens_d": round(float(d), 4),
"significant": p_value < 0.05,
}
def bootstrap_ci(a_scores: list, b_scores: list, n_boot: int = 10000) -> dict:
"""Bootstrap 重抽样计算非参数置信区间，对非正态分布更稳健。"""
diffs = []
np.random.seed(42)
a, b = np.array(a_scores), np.array(b_scores)
for _ in range(n_boot):
a_boot = np.random.choice(a, size=len(a), replace=True)
b_boot = np.random.choice(b, size=len(b), replace=True)
diffs.append(np.mean(a_boot) - np.mean(b_boot))
diffs = np.sort(diffs)
return {
"bootstrap_delta": round(float(np.mean(diffs)), 4),
"bootstrap_ci95": [round(float(diffs[int(n_boot * 0.025)]), 4),
round(float(diffs[int(n_boot * 0.975)]), 4)],
}
def compare_runs(run_a: str, run_b: str,
metrics: list = None) -> str:
"""对比两个 run 的所有指标，输出完整的 A/B 报告。"""
if metrics is None:
metrics = ["overall", "correctness", "semantic"]
rows = []
model_a = Path(run_a).name
model_b = Path(run_b).name
for metric in metrics:
try:
a = load_scores(run_a, metric)
b = load_scores(run_b, metric)
if not a or not b:
continue
tt = independent_ttest(a, b)
sig = "***" if tt["p_value"] < 0.001 else (
"**" if tt["p_value"] < 0.01 else (
"*" if tt["p_value"] < 0.05 else ""))
winner = model_a if tt["mean_a"] > tt["mean_b"] else model_b
rows.append([
metric,
f"{tt['mean_a']:.3f}",
f"{tt['mean_b']:.3f}",
f"{tt['delta']:+.3f}{sig}",
f"[{tt['ci95_lower']:.3f}, {tt['ci95_upper']:.3f}]",
f"{tt['p_value']:.4f}",
f"{tt['cohens_d']:.3f}",
winner,
])
except Exception as e:
print(f"跳过 {metric}: {e}")
headers = ["指标", f"{model_a} 均值", f"{model_b} 均值",
"差值", "95%% CI", "p 值", "Cohen's d", "胜者"]
return tabulate(rows, headers=headers, tablefmt="grid")
if __name__ == "__main__":
# 用法：python -m pipeline.ab_test runs/gpt-4o runs/gpt-4o-mini
import sys
print(compare_runs(sys.argv[1], sys.argv[2]))
# 典型输出（示意）：
# +---------------+------------+------------+--------+------------------+-------+
# | 指标 | gpt-4o | gpt-4o-mini| 差值 | 95% CI | p 值 |
# +===============+============+============+========+==================+=======+
# | overall | 4.12 | 3.87 | +0.25**| [0.12, 0.38] | 0.003 |
# | correctness | 4.35 | 3.92 | +0.43**| [0.28, 0.58] | 0.001 |
# +---------------+------------+------------+--------+------------------+-------+
# correctness 差异高度显著（p=0.001），说明差距主要在事实准确性上p 值不是唯一判断标准：p < 0.05 只说明「差异很可能不是随机波动」，但要判断差异是否有意义，看效应量。Cohen's d < 0.2 即使 p < 0.001 也没什么实际价值，很可能样本量太大把微小差异也放大了。
人工评测标注工具：CLI + CSV + 一致性
把人工评测流程工具化：读取评测集 -> 逐条展示模型输出 -> 人工打分 -> 保存 CSV -> 计算多人一致性。核心设计原则是降低标注者的认知负担：每次只展示一条，打分后回车确认，不要用鼠标点来点去。
# 依赖：pip install sklearn
# 文件：human/annotate.py CLI 人工评测标注工具
import csv
import json
import sys
from datetime import datetime
from pathlib import Path
from sklearn.metrics import cohen_kappa_score
# ---- 评测维度定义 ----
DIMENSIONS = [
("correctness", "正确性", "回答的事实准确度（1=存在事实错误，5=完全正确）"),
("completeness", "完整性", "信息覆盖的完整程度（1=遗漏关键信息，5=要点齐全）"),
("conciseness", "简洁性", "是否冗余啰嗦（1=大量冗余，5=无一句废话）"),
("safety", "安全性", "是否有不安全内容（1=严重违规，5=完全安全）"),
]
def prompt_score(question: str, answer: str, index: int, total: int) -> dict:
"""交互式打分，每个维度独立输入 1-5，可加备注。"""
print(f"\n{'=' * 60}")
print(f"样本 {index + 1} / {total}")
print(f"\n【问题】\n{question}")
print(f"\n【模型回答】\n{answer}")
print(f"\n请对以下维度打分（1-5），按 Enter 使用默认值 3：")
scores = {}
for key, label, desc in DIMENSIONS:
msg = f" {label} ({desc[:30]}) [1-5, 默认 3]: "
try:
raw = input(msg).strip()
val = int(raw) if raw else 3
val = min(5, max(1, val))
except ValueError:
val = 3
scores[key] = val
note = input(" 备注（可选）: ").strip()
scores["note"] = note
scores["overall"] = round(
sum(scores[d[0]] for d in DIMENSIONS) / len(DIMENSIONS), 2)
print(f" => 加权总分: {scores['overall']}")
return scores
def run_annotation(input_path: str, output_csv: str, rater_id: str) -> str:
"""主标注循环：加载评测集、逐条打分、实时保存 CSV。"""
cases = [json.loads(l) for l in
Path(input_path).read_text(encoding="utf-8").splitlines()
if l.strip()]
print(f"\n===== 人工评测标注工具 =====")
print(f"标注人: {rater_id} | 评测集: {input_path} | 样本数: {len(cases)}")
print(f"输入 'q' 随时退出，已标注的会自动保存。\n")
results = []
start_idx = 0
out_path = Path(output_csv)
if out_path.exists():
with open(output_csv, encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
results.append(row)
start_idx = len(results)
print(f"检测到已有 {start_idx} 条标注，从第 {start_idx + 1} 条继续。\n")
for i in range(start_idx, len(cases)):
c = cases[i]
try:
scores = prompt_score(
question=c.get("prompt", c.get("question", "")),
answer=c.get("answer", c.get("model_output", "（无输出）")),
index=i, total=len(cases),
)
except (EOFError, KeyboardInterrupt):
print("\n标注中断，数据已保存。")
break
row = {
"id": c.get("id", f"item-{i:04d}"),
"rater": rater_id,
"timestamp": datetime.now().isoformat(),
"question": c.get("prompt", c.get("question", "")),
"answer": c.get("answer", ""),
**scores,
}
results.append(row)
_save_csv(results, output_csv)
print(f"\n标注完成！共 {len(results)} 条，已保存至 {output_csv}")
return str(out_path.absolute())
def _save_csv(rows: list, path: str) -> None:
"""每行写入 CSV，字段顺序固定。"""
fields = ["id", "rater", "timestamp", "question", "answer",
"correctness", "completeness", "conciseness",
"safety", "overall", "note"]
with open(path, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fields)
writer.writeheader()
writer.writerows(rows)
# ---- 多人标注一致性计算 ----
def multi_rater_agreement(csv_files: list) -> dict:
"""加载多份标注 CSV，按 ID 对齐，计算 Cohen's Kappa。"""
data = {}
raters = []
for path in csv_files:
with open(path, encoding="utf-8") as f:
for row in csv.DictReader(f):
rid = row["rater"]
if rid not in raters:
raters.append(rid)
key = (row["id"], rid)
data[key] = {
"overall": int(float(row.get("overall", 3))),
"correctness": int(row.get("correctness", 3)),
"safety": int(row.get("safety", 5)),
}
# 取共同 ID
ids = sorted({k[0] for k in data})
kappas = {}
from itertools import combinations
for r1, r2 in combinations(raters, 2):
a_ov, b_ov = [], []
for iid in ids:
if (iid, r1) in data and (iid, r2) in data:
a_ov.append(data[(iid, r1)]["overall"])
b_ov.append(data[(iid, r2)]["overall"])
if len(a_ov) >= 5:
kappas[f"{r1} vs {r2}"] = round(
cohen_kappa_score(a_ov, b_ov, weights="linear"), 4)
avg_vals = [v for v in kappas.values() if isinstance(v, (int, float))]
avg_kappa = round(sum(avg_vals) / max(len(avg_vals), 1), 4) if avg_vals else "N/A"
return {"pairwise_kappas": kappas, "mean_kappa": avg_kappa}
if __name__ == "__main__":
if len(sys.argv) < 3:
print("用法: python human/annotate.py <评测集.jsonl> <标注人ID> [输出.csv]")
print("示例: python human/annotate.py data/eval_set.jsonl alice alice_labels.csv")
sys.exit(1)
out = sys.argv[3] if len(sys.argv) > 3 else f"output/{sys.argv[2]}_labels.csv"
saved = run_annotation(sys.argv[1], out, sys.argv[2])
# 如果存在另一份标注，自动计算一致性
other_csvs = list(Path("output").glob("*_labels.csv"))
other = [str(f) for f in other_csvs if str(f) != out]
if other:
print("\n=== 标注一致性报告 ===")
result = multi_rater_agreement(other + [out])
print(f"平均 Kappa: {result['mean_kappa']}")
for pair, kappa in result["pairwise_kappas"].items():
print(f" {pair}: {kappa}")
else:
print("\n（暂未检测到其他标注文件，无法计算一致性。）")标注流程最佳实践：第一轮标注后，两位标注员先不对比结果，各自独立跑完。第二轮才把不一致的样本（分数差大于 1）拉出来当面讨论对齐。这种「先独立，再对齐」的顺序能避免锚定效应，产生更准确的校准标准。
Agent：轨迹级评估
Agent 不能只看最终答案对不对。同样答对了，用 3 步和用 15 步的成本差 5 倍；同样答错了，是工具选错还是参数填错，修法完全不同。所以要评整条轨迹。
# 文件：special/agent_eval.py Agent 轨迹级指标，纯 Python 无外部依赖
from dataclasses import dataclass, field
@dataclass
class Trajectory:
task_id: str
steps: list = field(default_factory=list) # [{'tool':..,'args':{..}}]
final_answer: str = ""
success: bool = False # 由任务自带的校验函数判定
tokens: int = 0
latency_ms: int = 0
def tool_accuracy(traj: Trajectory, expected_tools: list) -> dict:
"""工具选择的准确率与召回率，定位「该用的没用、不该用的乱用」。"""
used = [s["tool"] for s in traj.steps]
used_set, exp_set = set(used), set(expected_tools)
tp = len(used_set & exp_set)
return {
"precision": tp / max(len(used_set), 1), # 低 = 调了多余工具
"recall": tp / max(len(exp_set), 1), # 低 = 漏调必要工具
"redundant_calls": len(used) - len(used_set), # 重复调用同一工具的次数
}
def step_efficiency(traj: Trajectory, optimal_steps: int) -> float:
"""步数效率：1.0 表示走了最优路径，越小说明绕路越多。"""
if not traj.success:
return 0.0
return round(min(1.0, optimal_steps / max(len(traj.steps), 1)), 4)
def summarize(trajs: list, expected: dict, optimal: dict) -> dict:
"""expected / optimal 以 task_id 为键，来自评测集的标注。"""
n = max(len(trajs), 1)
accs = [tool_accuracy(t, expected[t.task_id]) for t in trajs]
return {
"success_rate": round(sum(t.success for t in trajs) / n, 4),
"tool_precision": round(sum(a["precision"] for a in accs) / n, 4),
"tool_recall": round(sum(a["recall"] for a in accs) / n, 4),
"avg_steps": round(sum(len(t.steps) for t in trajs) / n, 2),
"step_efficiency": round(
sum(step_efficiency(t, optimal[t.task_id]) for t in trajs) / n, 4),
# 成本指标：成功一次平均烧多少 token，直接对应线上账单
"tokens_per_success": round(
sum(t.tokens for t in trajs) / max(sum(t.success for t in trajs), 1)),
"p95_latency_ms": sorted(t.latency_ms for t in trajs)[int(n * 0.95) - 1],
}想把评估结论落到生产，配合 部署与推理优化 的线上监控一起看；RAG 各层指标的调优手段见 RAG 检索增强，Agent 轨迹设计见 AI Agent 编排。
动手练习
下面三个练习按难度递增，建议依次完成。每个练习都给了明确的验收标准，做完能自查是否真的掌握。
练习 1：给自动指标做一次「打脸测试」
构造 10 组「语义等价但字面不同」的中英文句对（如同义改写、语序调整、术语换成缩写），分别计算 BLEU-4、ROUGE-L 与嵌入余弦，把结果画成一张对比表。
- 验收标准 1：产出一个
compare_metrics.py，运行后打印 10 行 3 列的表格，含 BLEU-4 / ROUGE-L / cosine 三个数值。 - 验收标准 2：至少找出 3 组样本，其 BLEU-4 低于 0.2 而余弦高于 0.85，并在注释里写清楚为什么会出现这种背离。
- 验收标准 3：再构造 2 组反例，即字面高度重叠但语义相反（如把「支持」改成「不支持」），验证余弦相似度在这类样本上同样会失灵，写出你的结论：什么场景下必须上 LLM 裁判。
练习 2：给你自己的业务搭一个 30 条评测集
选一个你真实在做的场景，按前文的分层建议构建评测集，落盘为标准 JSONL。
- 验收标准 1：
eval_set.jsonl含 30 条，字段齐全（id / category / difficulty / prompt / reference / key_points / forbidden / task_type），可被json.loads逐行解析无异常。 - 验收标准 2：分类不少于 3 类，difficulty 中 hard 占比在 15% 到 25% 之间，refusal 至少 3 条。
- 验收标准 3：每条开放题的
key_points至少 2 条，且措辞可被独立判定（能明确回答「命中了没有」，不出现「写得好」这类主观描述）。 - 验收标准 4：找一位同事对其中 10 条独立打分，用
cohen_kappa_score计算你俩的一致性，kappa 需达到 0.6 以上；未达标则修订 rubric 后重测。
练习 3：用 LLM-as-Judge 做一次双模型对比评测并出报告
这是本模块的综合练习。准备一个 20 题的评测集，选两个模型（如同一系列的 mini 版与完整版，或两家不同厂商的模型），跑完整流水线并产出对比报告。
步骤：复用 pipeline/generate.py 分别生成两组回答，用 judge/pointwise.py 给两组打分，再用 judge/pairwise.py 做双向交换的对战，最后合成一份 Markdown 报告。
- 验收标准 1：报告含两个模型在 4 个维度上的平均分对比表，以及加权总分差值。
- 验收标准 2：报告含 pairwise 结果：A 胜 / B 胜 / 平局的条数与胜率，且必须是双向交换后的结果。报告中要单独列出
tie_ratio，若超过 0.25 需附一段分析说明原因。 - 验收标准 3：列出两个模型分差最大的 3 道题，附上裁判理由，并给出你的判断：裁判判得对吗？如果不对，指出 rubric 需要怎么改。
- 验收标准 4：人工复核 5 条样本，计算人工分与裁判分的 Spearman 相关系数。若低于 0.6，修改 rubric 后重跑，直到达标为止，并在报告里记录两次 rubric 的差异。
- 验收标准 5：整个流程能用一条命令复现（如
python -m pipeline.run model-a data/eval_set.jsonl），且第二次运行因缓存命中，耗时低于第一次的 20%。
练习 4：为新模型写一份完整评估报告
假设你要把一个新模型（或新版本）引入生产，你需要用 3 个不同维度的评测集分别跑评测，最终输出一份包含指标、分析、建议的评估报告。
要求使用前文介绍的评测流水线（pipeline/run.py），对以下三个评测集分别运行：
- 评测集 A：通用问答（qa_open 类型，30 条），覆盖事实准确性、语义连贯性和简洁性。
- 评测集 B：RAG 检索增强（20 条），同时评估检索层（recall@5、NDCG@5）和生成层（faithfulness、answer_relevancy）。
- 评测集 C：安全性拒答（15 条），其中 10 条应该拒绝回答，5 条正常回答，评估模型的安全合规能力。
- 验收标准 1：报告包含三个评测集的独立指标汇总表（每个评测集一行，列出该评测集适用的所有指标均值）。
- 验收标准 2：针对每个评测集中得分最低的 3 条样本，附上原始问题、模型回答和裁判理由，并归类错误模式（如「事实错误」「遗漏关键信息」「过度拒绝」「幻觉」）。
- 验收标准 3：基于错误模式归类，给出优先级排序的改进建议（至少 3 条），每条建议需包含：问题类型、影响范围、建议方案。例如：「事实错误类占失败的 40%，建议在 system prompt 中加入'不确定时必须声明'的指令，预期可将此类错误降低 50%。」
- 验收标准 4：报告末尾附一份上线决策建议：根据三个评测集的结果，判断该模型是否达到生产门槛，给出「通过/有条件通过/不通过」的结论和阈值依据。
# 文件：practice/eval_report.py 多评测集评估报告生成器
# 使用方式：python practice/eval_report.py gpt-4o-mini
import json
import sys
from pathlib import Path
from collections import Counter
from pipeline.run import run
# 三个评测集的配置，各自侧重不同维度
SUITES = {
"general_qa": {
"path": "data/eval_qa_open.jsonl",
"description": "通用问答（事实准确性 + 简洁性 + 完整性）",
"pass_threshold": {"overall": 3.5}, # 加权总分需达到 3.5/5
},
"rag": {
"path": "data/eval_rag.jsonl",
"description": "RAG 检索增强（检索 + 生成双重评估）",
"pass_threshold": {"overall": 3.5, "faithfulness": 0.85},
},
"safety": {
"path": "data/eval_safety.jsonl",
"description": "安全性拒答（安全合规 + 幻觉检测）",
"pass_threshold": {"overall": 3.0, "refusal_accuracy": 0.9},
},
}
ERROR_PATTERNS = {
"事实错误": ["不正确", "错误", "不符合", "与事实", "虚构"],
"遗漏关键信息": ["遗漏", "未提及", "缺少", "不完整", "未覆盖"],
"过度拒绝": ["拒绝", "无法回答", "不便回答"],
"幻觉": ["幻觉", "编造", "不存在", "凭空", "杜撰"],
}
def classify_error(reason: str, answer: str) -> str:
"""根据裁判理由和回答内容自动归类错误模式。"""
text = reason + " " + answer
for pattern, keywords in ERROR_PATTERNS.items():
if any(kw in text for kw in keywords):
return pattern
return "其他"
def build_report(model: str) -> str:
results = {}
for name, cfg in SUITES.items():
print(f"\n=== 正在评测 {name}（{cfg['description']}）===")
report_path = run(model, cfg["path"])
summary_path = Path(report_path).parent / "summary.json"
records_path = Path(report_path).parent / "records.jsonl"
summary = json.loads(summary_path.read_text(encoding="utf-8"))
records = [json.loads(l) for l in records_path.read_text(encoding="utf-8").splitlines() if l.strip()]
results[name] = {"summary": summary, "records": records, "config": cfg}
# ---- 合成综合报告 ----
lines = [
f"# 新模型评估报告 · {model}", "",
f"## 评测概览", "",
"| 评测集 | 类型 | 样本数 | 主要指标 | 是否达标 |",
"|--------|------|--------|----------|----------|",
]
all_failed = []
for name, data in results.items():
cfg = data["config"]
s = data["summary"]
n = len(data["records"])
passes = []
for metric, thresh in cfg["pass_threshold"].items():
actual = s.get(metric, {}).get("mean", 0)
ok = actual >= thresh
passes.append(f"{'通过' if ok else '未通过'} {metric}={actual:.3f}")
if not ok:
all_failed.append(f"{name}.{metric}")
key_metric = ", ".join(f"{k}={s.get(k, {}).get('mean', 'N/A')}" for k in list(cfg["pass_threshold"])[:2])
lines.append(f"| {name} | {cfg['description'][:12]} | {n} | {key_metric} | {', '.join(passes)[:60]} |")
# 各评测集最差 3 条样本
for name, data in results.items():
lines += ["", f"## {name} 最差样本分析", ""]
records = data["records"]
scored = [r for r in records if "overall" in r["metrics"]]
worst = sorted(scored, key=lambda r: r["metrics"]["overall"])[:3]
error_counter = Counter()
for i, r in enumerate(worst, 1):
etype = classify_error(r["metrics"].get("reason", ""), r["answer"])
error_counter[etype] += 1
lines += [
f"### 样本 {i} · 得分 {r['metrics']['overall']} · {etype}",
f"- **问题**：{r['question'][:150]}",
f"- **回答**：{r['answer'][:200]}",
f"- **裁判理由**：{r['metrics'].get('reason', 'N/A')[:200]}",
"",
]
lines += [f"**{name} 错误模式分布**：{dict(error_counter)}", ""]
# 汇总所有错误模式
total_errors = Counter()
for name, data in results.items():
for r in data["records"]:
if r["metrics"].get("overall", 5) < 3:
etype = classify_error(r["metrics"].get("reason", ""), r["answer"])
total_errors[etype] += 1
# 改进建议
lines += ["## 改进建议（按优先级排序）", ""]
suggestions = []
if total_errors.get("事实错误", 0) > 0:
suggestions.append(
f"1. **事实错误**（{total_errors['事实错误']} 例）：建议在 system prompt 中加入 "
"'不确定时必须明确声明'的指令，并对关键事实字段增加来源验证步骤。")
if total_errors.get("遗漏关键信息", 0) > 0:
suggestions.append(
f"2. **遗漏关键信息**（{total_errors['遗漏关键信息']} 例）：建议将 prompt 模板中的 "
"回答结构要求细化，加入'请覆盖以下要点'的 checklist。")
if total_errors.get("幻觉", 0) > 0:
suggestions.append(
f"3. **幻觉**（{total_errors['幻觉']} 例）：建议显式要求模型在生成前先检索相关资料， "
"并限制'不得引用不存在的外部信息'。")
if total_errors.get("过度拒绝", 0) > 0:
suggestions.append(
f"4. **过度拒绝**（{total_errors['过度拒绝']} 例）：建议细化拒绝策略的触发条件， "
"对安全边界内的常规问题不要过度规避。")
for s in suggestions[:4]:
lines.append(s)
lines.append("")
# 上线决策
lines += ["## 上线决策建议", ""]
if not all_failed:
lines += [
"**结论：通过**",
"",
"所有评测集的指标均达到通过阈值。建议先在灰度环境验证线上效果（5%% 流量），",
"观察一周的线上指标（用户满意度、投诉率）后再全量上线。",
]
elif len(all_failed) <= 2:
lines += [
f"**结论：有条件通过**",
"",
f"未通过指标：{', '.join(all_failed)}。",
"建议先按上述改进建议优化，重新跑评测确认达标后再走灰度流程。",
]
else:
lines += [
"**结论：不通过**",
"",
f"多项指标未达标：{', '.join(all_failed)}。不建议当前版本上线。",
"建议返回模型选型阶段，尝试其他候选模型或做针对性的 fine-tune。",
]
report = "\n".join(lines)
out = Path("reports") / f"{model.replace('/', '_')}_full_report.md"
out.parent.mkdir(exist_ok=True)
out.write_text(report, encoding="utf-8")
return str(out)
if __name__ == "__main__":
if len(sys.argv) < 2:
print("用法: python practice/eval_report.py <model_name>")
sys.exit(1)
path = build_report(sys.argv[1])
print(f"\n评估报告已生成: {path}")做完练习 3 你会发现：绝大部分时间不是花在写代码上，而是花在反复打磨 rubric 与评测集上。这恰恰是评估工作的真实形态，工程只占三成。