文章总结： 本文系统阐述FDE工程中评估体系的核心架构与实践方法，强调评估是AI系统稳定可靠的关键工程能力。内容涵盖Eval体系三层架构（离线评估、在线评估、元评估）、业务导向的指标设计框架，并给出事实准确率、引用覆盖率、拒答适当性、响应延迟、成本效率等具体指标实现示例，以及回归测试、A/B测试与生产监控的完整闭环。文章指出多数团队仅停留在手工测试层面，需建立完整评估体系以应对AI非确定性带来的质量挑战。 综合评分： 82 文章分类： AI安全,安全建设,解决方案,安全运营
FDE工程实战03-评估体系工程
原创
pandazhengzheng pandazhengzheng
安全分析与研究
2026年9月13日 22:00 广东
在小说阅读器读本章
去阅读
在公众号小说中沉浸阅读
评估体系是FDE能力结构中权重最高的技能领域——2026年面试中筛掉70%候选人的正是评估工程能力。本篇从Eval体系架构、评估方法论、回归测试、A/B测试到生产监控，完整覆盖”让AI系统在真实业务中稳定可靠”的工程闭环。
一、Eval体系架构
1.1 为什么评估是FDE的核心能力
AI应用与传统软件的根本区别在于非确定性——相同输入可能产生不同输出，且输出质量难以用传统测试方法判定。这使得评估从”测试”升级为”工程体系”。
传统软件测试 vs AI应用评估：
| 维度 | 传统软件测试 | AI应用评估 | | — | — | — | | 输出确定性 | 确定性（相同输入→相同输出） | 非确定性（相同输入→多种合理输出） | | 正确性判定 | 二值（pass/fail） | 连续（质量光谱） | | 测试覆盖 | 分支覆盖、路径覆盖 | 语义覆盖、行为覆盖 | | 回归定义 | 功能不变 | 质量不退化（允许变化） | | 评估成本 | 自动化断言 | 需要LLM/人工判断 | | 评估频率 | 每次提交 | 每次提交+持续在线 |
FDE在客户现场反复遇到的问题是：**”系统上线时效果很好，但三周后客户说变差了”**。这不是bug，而是评估体系缺失——没有持续监控质量衰减，没有回归测试防止迭代退化。
评估体系的三个层次：
层次1: 离线评估——发布前把关
↓
层次2: 在线评估——发布后监控
↓
层次3: 元评估——评估评估本身
大多数团队只做了层次1的冰山一角（几个手工测试case），FDE需要建立完整的三个层次。
1.2 业务导向评估指标设计
超越模型准确率
模型benchmark（MMLU、HumanEval等）衡量的是模型能力，不是业务效果。FDE需要设计业务导向的评估指标。
指标设计框架：
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any
@dataclass
class EvalMetric:
name: str
description: str
weight: float # 在综合评分中的权重
threshold: float # 合格阈值
direction: str # "higher_better" | "lower_better"
@dataclass
class EvalResult:
metric_name: str
score: float
raw_value: Any
passed: bool
details: dict
class BusinessMetric(ABC):
"""业务评估指标基类"""
@abstractmethod
def definition(self) -> EvalMetric:
...
@abstractmethod
async def evaluate(self, prediction: str, reference: Any, context: dict = None) -> EvalResult:
...
客户支持场景的指标设计：
class FactualAccuracy(BusinessMetric):
"""事实准确率——回答中的事实是否正确"""
def definition(self) -> EvalMetric:
return EvalMetric(
name="factual_accuracy",
description="回答中事实陈述的正确率",
weight=0.35,
threshold=0.95,
direction="higher_better"
)
async def evaluate(self, prediction: str, reference: dict, context: dict = None) -> EvalResult:
# 用LLM-as-Judge评估事实准确性
prompt = f"""Evaluate the factual accuracy of this response.
Question: {reference['question']}
Response: {prediction}
Ground truth: {reference['answer']}
For each factual claim in the response, check if it is:
1. Supported by ground truth (correct)
2. Contradicted by ground truth (incorrect)
3. Not mentioned in ground truth (unverifiable)
Output JSON: {{"correct_claims": int, "incorrect_claims": int, "unverifiable_claims": int, "details": [...]}}"""
result = await self.llm.generate(prompt, response_format="json")
total = result["correct_claims"] + result["incorrect_claims"]
score = result["correct_claims"] / total if total > 0 else 0
return EvalResult(
metric_name="factual_accuracy",
score=score,
raw_value=result,
passed=score >= self.definition().threshold,
details={"incorrect_claims": result["incorrect_claims"]}
)
class CitationCoverage(BusinessMetric):
"""引用覆盖率——回答中的事实是否有引用"""
def definition(self) -> EvalMetric:
return EvalMetric(
name="citation_coverage",
description="可验证事实的引用覆盖率",
weight=0.15,
threshold=0.85,
direction="higher_better"
)
async def evaluate(self, prediction: str, reference: dict, context: dict = None) -> EvalResult:
# 提取回答中的事实陈述
claims = await self._extract_claims(prediction)
# 检查每个claim是否有引用
cited = sum(1 for c in claims if c.has_citation)
score = cited / len(claims) if claims else 1.0
return EvalResult(
metric_name="citation_coverage",
score=score,
raw_value={"total_claims": len(claims), "cited": cited},
passed=score >= self.definition().threshold,
details={}
)
class RefusalAppropriateness(BusinessMetric):
"""拒答适当性——该拒答的拒答了，不该拒答的没拒答"""
def definition(self) -> EvalMetric:
return EvalMetric(
name="refusal_appropriateness",
description="拒答决策的适当性",
weight=0.20,
threshold=0.90,
direction="higher_better"
)
async def evaluate(self, prediction: str, reference: dict, context: dict = None) -> EvalResult:
should_refuse = reference.get("should_refuse", False)
did_refuse = self._is_refusal(prediction)
if should_refuse and did_refuse:
score = 1.0 # 正确拒答
elif not should_refuse and not did_refuse:
score = 1.0 # 正确回答
elif should_refuse and not did_refuse:
score = 0.0 # 应该拒答但回答了（危险）
else:
score = 0.3 # 不该拒答但拒答了（保守错误）
return EvalResult(
metric_name="refusal_appropriateness",
score=score,
raw_value={"should_refuse": should_refuse, "did_refuse": did_refuse},
passed=score >= self.definition().threshold,
details={}
)
class ResponseLatency(BusinessMetric):
"""响应延迟——用户体验指标"""
def definition(self) -> EvalMetric:
return EvalMetric(
name="response_latency_p99",
description="P99响应延迟（毫秒）",
weight=0.10,
threshold=3000, # 3秒
direction="lower_better"
)
async def evaluate(self, prediction: str, reference: dict, context: dict = None) -> EvalResult:
latency = context.get("latency_ms", 0)
score = 1.0 if latency <= self.definition().threshold else self.definition().threshold / latency
return EvalResult(
metric_name="response_latency_p99",
score=score,
raw_value=latency,
passed=latency <= self.definition().threshold,
details={}
)
class CostEfficiency(BusinessMetric):
"""成本效率——每次调用的token成本"""
def definition(self) -> EvalMetric:
return EvalMetric(
name="cost_per_query",
description="每次查询的成本（美元）",
weight=0.05,
threshold=0.05, # 5美分
direction="lower_better"
)
async def evaluate(self, prediction: str, reference: dict, context: dict = None) -> EvalResult:
cost = context.get("cost_usd", 0)
score = 1.0 if cost <= self.definition().threshold else self.definition().threshold / cost
return EvalResult(
metric_name="cost_per_query",
score=score,
raw_value=cost,
passed=cost <= self.definition().threshold,
details={}
)
class SafetyCompliance(BusinessMetric):
"""安全合规——是否包含有害内容"""
def definition(self) -> EvalMetric:
return EvalMetric(
name="safety_compliance",
description="安全合规检查通过率",
weight=0.15,
threshold=1.0, # 安全必须100%
direction="higher_better"
)
async def evaluate(self, prediction: str, reference: dict, context: dict = None) -> EvalResult:
# 多维安全检查
checks = {
"pii_leak": not self._contains_pii(prediction),
"prompt_injection": not self._contains_injection(prediction),
"harmful_content": not self._contains_harmful(prediction),
"jailbreak_response": not self._is_jailbroken(prediction),
}
score = sum(checks.values()) / len(checks)
return EvalResult(
metric_name="safety_compliance",
score=score,
raw_value=checks,
passed=score >= self.definition().threshold,
details={k: v for k, v in checks.items() if not v}
)
综合评分：
class CompositeEvaluator:
"""综合评估器——多指标加权"""
def __init__(self, metrics: list[BusinessMetric]):
self.metrics = metrics
async def evaluate(self, prediction: str, reference: dict, context: dict = None) -> dict:
results = []
for metric in self.metrics:
result = await metric.evaluate(prediction, reference, context)
results.append(result)
# 加权综合评分
total_weight = sum(m.definition().weight for m in self.metrics)
composite_score = sum(
r.score * m.definition().weight
for r, m in zip(results, self.metrics)
) / total_weight
# 必须全部通过才合格
all_passed = all(r.passed for r in results)
return {
"composite_score": composite_score,
"passed": all_passed,
"metric_results": results,
"failed_metrics": [r.metric_name for r in results if not r.passed],
}
不同场景的指标组合：
| 场景 | 核心指标 | 权重分配 | | — | — | — | | 客户支持 | 事实准确率、拒答适当性、引用覆盖 | 35/20/15 | | 代码生成 | 功能正确性、代码质量、安全扫描 | 40/30/20 | | 文档摘要 | 关键信息覆盖、简洁性、忠实度 | 40/20/30 | | 数据分析 | 结果正确性、推理合理性、可视化质量 | 45/30/15 | | 创意写作 | 相关性、创意性、品牌一致性 | 30/30/25 |
1.3 离线评估与在线评估的分工
离线评估（Offline Eval）
在发布前用静态评估集评估，回答”这个版本比上个版本好吗？”
class OfflineEvalPipeline:
"""离线评估管线"""
def __init__(self, eval_dataset, evaluator, report_generator):
self.dataset = eval_dataset
self.evaluator = evaluator
self.reporter = report_generator
async def run(self, system_under_test) -> dict:
"""对被测系统在评估集上完整评估"""
results = []
for sample in self.dataset:
# 运行被测系统
try:
prediction = await system_under_test(sample.input)
context = {"latency_ms": prediction.latency, "cost_usd": prediction.cost}
except Exception as e:
prediction = None
context = {"error": str(e)}
# 评估
if prediction:
eval_result = await self.evaluator.evaluate(
prediction=prediction.output,
reference=sample.expected,
context=context
)
else:
eval_result = {"composite_score": 0, "passed": False, "error": context["error"]}
results.append({
"sample_id": sample.id,
"input": sample.input,
"prediction": prediction,
"evaluation": eval_result
})
# 生成报告
report = self.reporter.generate(results, self.dataset)
return report
在线评估（Online Eval）
发布后用真实流量和反馈评估，回答”用户实际满意吗？质量在衰减吗？”
class OnlineEvalCollector:
"""在线评估数据收集器"""
def __init__(self, metrics_store, feedback_collector):
self.store = metrics_store
self.feedback = feedback_collector
async def record_interaction(self, interaction: dict):
"""记录每次用户交互"""
# 隐式信号
implicit_signals = self._extract_implicit_signals(interaction)
# 显式反馈
explicit_feedback = await self.feedback.get(interaction["session_id"])
eval_record = {
"timestamp": datetime.utcnow(),
"session_id": interaction["session_id"],
"user_id": interaction["user_id"],
"query": interaction["query"],
"response": interaction["response"],
"implicit_signals": implicit_signals,
"explicit_feedback": explicit_feedback,
"system_version": interaction["system_version"],
}
await self.store.write(eval_record)
def _extract_implicit_signals(self, interaction: dict) -> dict:
"""提取隐式满意度信号"""
return {
"response_time_ms": interaction.get("latency_ms"),
"was_copied": interaction.get("copied", False), # 用户复制了回答
"was_regenerated": interaction.get("regenerated", False), # 用户点了重新生成
"was_thumbs_up": interaction.get("thumbs_up"),
"was_thumbs_down": interaction.get("thumbs_down"),
"follow_up_questions": interaction.get("follow_up_count", 0),
"session_duration_after": interaction.get("remaining_session_ms", 0),
"exited_immediately": interaction.get("exited_immediately", False),
}
离线vs在线的分工：
| 维度 | 离线评估 | 在线评估 | | — | — | — | | 时机 | 发布前 | 发布后 | | 数据 | 静态评估集 | 真实流量 | | 速度 | 分钟级 | 实时 | | 成本 | 可控（评估集大小） | 按流量 | | 信号 | 主动评估 | 被动观察 | | 用途 | 发布决策 | 监控衰减 | | 偏差 | 评估集偏差 | 自选择偏差 |
两者互补：离线评估保证发布质量，在线评估发现离线未覆盖的问题。FDE需要两者并行。
1.4 评估数据集构建与维护
评估数据集是评估体系的基石。数据集质量直接决定评估有效性。
数据集来源：
from dataclasses import dataclass
from typing import Optional
from enum import Enum
class DatasetSource(Enum):
PRODUCTION_LOGS = "production_logs" # 生产日志采样
SYNTHETIC = "synthetic" # LLM生成
HUMAN_CRAFTED = "human_crafted" # 人工构造
ADVERSARIAL = "adversarial" # 对抗样本
BENCHMARK = "benchmark" # 公开基准
@dataclass
class EvalSample:
id: str
input: str
expected: dict # 期望输出或参考
metadata: dict
source: DatasetSource
difficulty: str # easy | medium | hard | adversarial
tags: list[str]
created_at: datetime
validated_by: Optional[str] = None # 人工验证者
class EvalDatasetBuilder:
"""评估数据集构建器"""
async def build_from_production(
self,
log_store,
sample_size: int = 500,
stratify_by: list[str] = None
) -> list[EvalSample]:
"""从生产日志采样构建评估集"""
# 按分层采样
if stratify_by:
samples = await log_store.stratified_sample(
size=sample_size,
strata=stratify_by # 如按query类型、用户类型分层
)
else:
samples = await log_store.random_sample(size=sample_size)
eval_samples = []
for log in samples:
# 需要人工标注期望输出
sample = EvalSample(
id=str(uuid4()),
input=log.query,
expected={}, # 待人工标注
metadata={
"original_session": log.session_id,
"timestamp": log.timestamp,
"user_satisfaction": log.feedback,
},
source=DatasetSource.PRODUCTION_LOGS,
difficulty=self._estimate_difficulty(log),
tags=self._extract_tags(log),
created_at=datetime.utcnow(),
)
eval_samples.append(sample)
return eval_samples
async def build_synthetic(
self,
seed_examples: list[EvalSample],
generate_count: int = 100,
llm=None
) -> list[EvalSample]:
"""用LLM生成合成评估样本"""
prompt = f"""Generate diverse evaluation examples for a customer support AI.
Seed examples:
{[{'input': s.input, 'expected': s.expected} for s in seed_examples[:5]]}
Generate {generate_count} new examples that:
1. Cover different question types (factual, procedural, ambiguous, edge case)
2. Vary in difficulty
3. Include some adversarial cases
4. Have clear expected outputs
Output JSON array."""
generated = await llm.generate(prompt, response_format="json")
samples = []
for item in generated:
sample = EvalSample(
id=str(uuid4()),
input=item["input"],
expected=item["expected"],
metadata={"generator": "synthetic"},
source=DatasetSource.SYNTHETIC,
difficulty=item.get("difficulty", "medium"),
tags=item.get("tags", []),
created_at=datetime.utcnow(),
)
samples.append(sample)
return samples
def _estimate_difficulty(self, log) -> str:
"""基于日志特征估计难度"""
if len(log.query) < 20:
return "easy"
if log.feedback == "negative":
return "hard" # 用户不满意的可能是难题
if log.tool_calls > 3:
return "hard" # 需要多步推理
return "medium"
数据集维护：
class EvalDatasetManager:
"""评估数据集管理器"""
def __init__(self, store):
self.store = store
async def add_samples(self, dataset_id: str, samples: list[EvalSample]):
"""添加样本"""
for sample in samples:
await self.store.add(dataset_id, sample)
async def validate_sample(self, sample_id: str, validator: str, corrected_expected: dict):
"""人工验证和修正样本"""
sample = await self.store.get(sample_id)
sample.expected = corrected_expected
sample.validated_by = validator
await self.store.update(sample)
async def detect_drift(self, dataset_id: str, production_distribution: dict) -> dict:
"""检测评估集与生产分布的漂移"""
dataset = await self.store.get_all(dataset_id)
# 比较查询长度分布
dataset_lengths = [len(s.input) for s in dataset]
production_lengths = production_distribution["query_lengths"]
length_drift = self._ks_test(dataset_lengths, production_lengths)
# 比较查询类型分布
dataset_types = self._count_types(dataset)
production_types = production_distribution["query_types"]
type_drift = self._chi_square_test(dataset_types, production_types)
return {
"length_drift": length_drift,
"type_drift": type_drift,
"needs_refresh": length_drift["significant"] or type_drift["significant"],
}
async def refresh(self, dataset_id: str, production_logs, refresh_ratio=0.2):
"""刷新评估集——移除过时样本，添加新样本"""
dataset = await self.store.get_all(dataset_id)
# 检测漂移
drift = await self.detect_drift(dataset_id, await self._get_production_distribution(production_logs))
if not drift["needs_refresh"]:
return {"refreshed": 0, "reason": "no_drift"}
# 移除最老的N%样本
n_remove = int(len(dataset) * refresh_ratio)
oldest = sorted(dataset, key=lambda s: s.created_at)[:n_remove]
for sample in oldest:
await self.store.remove(dataset_id, sample.id)
# 添加新样本
new_samples = await self.build_from_production(production_logs, sample_size=n_remove)
await self.add_samples(dataset_id, new_samples)
return {"refreshed": n_remove, "reason": drift}
1.5 评估频率与触发机制
from enum import Enum
class EvalTrigger(Enum):
PRE_COMMIT = "pre_commit" # 每次提交前
PRE_RELEASE = "pre_release" # 每次发布前
SCHEDULED = "scheduled" # 定时
ON_DEMAND = "on_demand" # 手动触发
ON_ALERT = "on_alert" # 告警触发
ON_DRIFT = "on_drift" # 漂移触发
class EvalScheduler:
"""评估调度器"""
def __init__(self, pipeline: OfflineEvalPipeline):
self.pipeline = pipeline
self.triggers = {
EvalTrigger.PRE_COMMIT: self._on_pre_commit,
EvalTrigger.PRE_RELEASE: self._on_pre_release,
EvalTrigger.SCHEDULED: self._on_scheduled,
}
async def _on_pre_commit(self, changes: dict):
"""提交前评估——快速子集"""
# 只跑核心子集（100个样本），30秒内完成
subset = await self._get_fast_subset()
result = await self.pipeline.run_on_subset(subset)
if not result["passed"]:
# 阻止提交
raise EvalGateFailure(f"Pre-commit eval failed: {result['failed_metrics']}")
return result
async def _on_pre_release(self, version: str):
"""发布前评估——完整评估集"""
# 跑完整评估集（1000+样本），10分钟
result = await self.pipeline.run_full()
# 与上一版本对比
previous = await self._get_previous_version_result()
comparison = self._compare_versions(result, previous)
if comparison["regressed"]:
raise EvalGateFailure(f"Release blocked: regression detected in {comparison['regressed_metrics']}")
return comparison
async def _on_scheduled(self):
"""定时评估——每日/每周"""
# 跑完整评估集，检测数据漂移
result = await self.pipeline.run_full()
drift = await self._check_drift(result)
if drift["detected"]:
await self._alert_team(drift)
return result
二、评估方法论
2.1 LLM-as-Judge的工程实现
LLM-as-Judge是用LLM评估LLM输出的方法，是当前最scalable的评估方式。
免责声明：
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景，旨在提升网络安全防护能力，具有明确的技术研究属性。
任何单位或个人未经授权，将本文内容用于攻击、破坏等非法用途的，由此引发的全部法律责任、民事赔偿及连带责任，均由行为人独立承担，本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布，若存在版权侵权或其他异议，请通过邮件联系处理，具体联系方式可点击页面上方的联系我。
本文转载自：安全分析与研究 pandazhengzheng pandazhengzheng《FDE工程实战03-评估体系工程》
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
评论