一、项目背景与问题定义
1.1 背景故事
在企业AI平台的建设过程中，我们遇到了一个典型的问题：随着业务的发展，公司内部逐渐形成了多个AI能力中心，每个中心都开发了自己的模型评测工具。这些工具功能相似但实现各异，评测标准不统一，评测结果无法横向比较，造成了大量重复建设。
具体场景还原：
客服AI团队开发了自己的对话模型评测工具，主要评估对话流畅度和回答准确率
风控AI团队有自己的文本分类模型评测工具，关注Precision/Recall/F1
推荐算法团队维护着一套推荐模型评测脚本，评估点击率和转化率
NLP团队管理着多个基础模型的Benchmark评测
每个团队都在"造轮子"：重复的评测代码、分散的评测结果、独立的模型管理。当管理层想要横向对比不同团队模型的能力时，发现根本没有统一的标准。
1.2 问题分析
通过深入调研，我们识别出以下核心问题：
┌─────────────────────────────────────────────────────────────────────────┐
│ 模型评测平台现状与问题 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ 【现状】各团队独立建设，形成评测孤岛 │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐│
│ │ 客服AI团队 │ │ 风控AI团队 │ │ 推荐算法团队 │ │ NLP团队 ││
│ │ │ │ │ │ │ │ ││
│ │ - 评测工具A │ │ - 评测工具B │ │ - 评测工具C │ │ - 评测工具D ││
│ │ - 标准不统一 │ │ - 标准不统一 │ │ - 标准不统一 │ │ - 标准不统一 ││
│ │ - 结果分散 │ │ - 结果分散 │ │ - 结果分散 │ │ - 结果分散 ││
│ │ - 无法对比 │ │ - 无法对比 │ │ - 无法对比 │ │ - 无法对比 ││
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘│
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ══════════════════════════════════════════════════════════════════════│
│ │ 评测标准碎片化 (Fragmented Standards) ││
│ ══════════════════════════════════════════════════════════════════════│
│ │
│ 【问题1】评测标准碎片化 │
│ - 每个团队用不同的数据集、不同的指标、不同的阈值 │
│ - 无法建立统一的模型能力评估标准 │
│ - 同一模型在不同团队评测结果可能大相径庭 │
│ │
│ 【问题2】重复劳动 │
│ - 大量重复的评测代码基础设施 │
│ - 每次开发新模型都要重新实现评测流程 │
│ - 缺乏可复用的评测组件 │
│ │
│ 【问题3】缺乏可追溯性 │
│ - 评测结果分散在各个团队的私有系统里 │
│ - 无法追踪模型性能随时间的变化 │
│ - 无法分析评测结果与训练数据/超参数的关系 │
│ │
│ 【问题4】资源利用率低 │
│ - GPU资源分散在各团队 │
│ - 无法统一调度和复用 │
│ - 评测任务优先级无法统一管理 │
│ │
└─────────────────────────────────────────────────────────────────────────┘1.3 解决方案
针对上述问题，我们设计并实现了统一模型评测平台，目标是一次建设、多方复用：
统一评测框架：支持多种评测类型（自动评测、人工评测、LLM-as-Judge），提供可扩展的评测模板。
统一评测基准：内置常用Benchmark，支持自定义评测基准，建立统一的评测标准。
统一资源调度：基于Kubernetes实现评测任务的统一调度，提高GPU资源利用率。
统一报告生成：自动生成评测报告，支持横向对比和趋势分析。
统一模型排行：建立公开的模型能力排行榜，促进良性竞争。
二、技术方案设计
2.1 整体架构
┌─────────────────────────────────────────────────────────────────────────┐
│ 统一模型评测平台架构 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 用户交互层 (Web UI) │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │评测任务 │ │评测报告 │ │模型排行 │ │基准管理 │ │ │
│ │ │管理页面 │ │查看页面 │ │展示页面 │ │配置页面 │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ │
│ └──────────────────────────┬──────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────▼──────────────────────────────────────┐ │
│ │ API网关层 (Gateway) │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ Authentication | Rate Limiting | Request Routing │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └──────────────────────────┬──────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────▼──────────────────────────────────────┐ │
│ │ 服务层 (Services) │ │
│ │ │ │
│ │ ┌────────────────┐ ┌────────────────┐ ┌────────────────┐ │ │
│ │ │ 评测任务服务 │ │ 评测引擎服务 │ │ 报告生成服务 │ │ │
│ │ │ Task Service │ │ Engine Service │ │ Report Service│ │ │
│ │ └───────┬────────┘ └───────┬────────┘ └───────┬────────┘ │ │
│ │ │ │ │ │ │
│ │ ┌───────▼───────────────────▼───────────────────▼───────┐ │ │
│ │ │ 评测引擎 (Evaluation Engine) │ │ │
│ │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │
│ │ │ │自动评测引擎 │ │人工评测引擎 │ │LLM评测引擎 │ │ │ │
│ │ │ │ Auto Engine │ │ Human Engine│ │Judge Engine │ │ │ │
│ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ │
│ │ └───────────────────────────────────────────────────────┘ │ │
│ │ │ │
│ └──────────────────────────┬──────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────▼──────────────────────────────────────┐ │
│ │ 调度层 (Scheduler) │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ Celery + Redis | 任务队列 | 优先级调度 | 失败重试 │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └──────────────────────────┬──────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────▼──────────────────────────────────────┐ │
│ │ 资源层 (Kubernetes) │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ GPU节点池 │ │ 评测Job │ │ 模型存储 │ │ │
│ │ │ (nvidia/gpu)│ │ (Batch Job) │ │ (PVC/S3) │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │
│ │ │ │
│ └───────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘2.2 核心模块设计
评测任务管理：负责评测任务的创建、调度、状态跟踪。
评测引擎：支持多种评测类型，包括自动评测（基于预定义指标）、人工评测（众包标注）、LLM-as-Judge（使用大模型评估）。
评测基准管理：内置常用Benchmark（MMLU、HumanEval、GSM8K等），支持自定义基准。
报告生成：自动生成评测报告，支持横向对比、趋势分析、雷达图展示。
模型排行：基于评测结果生成模型能力排行榜。
三、数据库设计
3.1 核心数据模型
# models.py
# 评测平台核心数据模型
import uuid
from django.db import models
from django.contrib.auth.models import User
class Benchmark(models.Model):
"""
评测基准模型
代表一个完整的评测基准，包含多个评测任务
"""
class Category(models.TextChoices):
NLP = 'nlp', '自然语言处理'
CODE = 'code', '代码生成'
MATH = 'math', '数学推理'
REASONING = 'reasoning', '逻辑推理'
SAFETY = 'safety', '安全性'
CUSTOM = 'custom', '自定义'
class Status(models.TextChoices):
ACTIVE = 'active', '活跃'
DEPRECATED = 'deprecated', '已废弃'
DRAFT = 'draft', '草稿'
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
name = models.CharField(max_length=100, unique=True, help_text="基准名称")
slug = models.SlugField(max_length=100, unique=True, help_text="URL标识")
# 分类和描述
category = models.CharField(
max_length=20,
choices=Category.choices,
default=Category.CUSTOM
)
description = models.TextField(help_text="基准描述")
# 评测类型
evaluation_type = models.CharField(
max_length=20,
choices=[
('automatic', '自动评测'),
('human', '人工评测'),
('llm_judge', 'LLM评测'),
],
default='automatic'
)
# 配置
config = models.JSONField(
default=dict,
help_text="基准配置，如样本数、生成参数等"
)
# 能力维度（用于雷达图展示）
capability_dimensions = models.JSONField(
default=list,
help_text="能力维度列表"
)
status = models.CharField(
max_length=20,
choices=Status.choices,
default=Status.ACTIVE
)
# 内置标记
is_builtin = models.BooleanField(default=False)
# 时间戳
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = 'benchmarks'
ordering = ['-created_at']
def __str__(self):
return f"{self.name} ({self.category})"
class Task(models.Model):
"""
评测任务模型
代表一次具体的评测执行
"""
class Status(models.TextChoices):
PENDING = 'pending', '等待调度'
RUNNING = 'running', '运行中'
COMPLETED = 'completed', '已完成'
FAILED = 'failed', '失败'
CANCELLED = 'cancelled', '已取消'
class Priority(models.TextChoices):
LOW = 'low', '低'
NORMAL = 'normal', '普通'
HIGH = 'high', '高'
URGENT = 'urgent', '紧急'
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
# 基本信息
name = models.CharField(max_length=200, help_text="任务名称")
description = models.TextField(blank=True)
# 关联资源
tenant_id = models.CharField(max_length=100, db_index=True)
benchmark = models.ForeignKey(
Benchmark,
on_delete=models.CASCADE,
related_name='tasks'
)
model_version = models.ForeignKey(
'ai_platform.ModelVersion', # 假设的模型版本表
on_delete=models.SET_NULL,
null=True,
related_name='evaluation_tasks'
)
# 配置
config = models.JSONField(
default=dict,
help_text="任务配置"
)
priority = models.CharField(
max_length=20,
choices=Priority.choices,
default=Priority.NORMAL
)
# 状态和进度
status = models.CharField(
max_length=20,
choices=Status.choices,
default=Status.PENDING,
db_index=True
)
progress = models.FloatField(default=0.0)
# K8s Job信息
kubernetes_job_name = models.CharField(max_length=200, blank=True)
kubernetes_namespace = models.CharField(max_length=100, blank=True)
# 结果
results = models.JSONField(
default=dict,
help_text="评测结果"
)
summary = models.JSONField(
default=dict,
help_text="结果摘要"
)
report_path = models.CharField(max_length=500, blank=True)
# 错误信息
error_message = models.TextField(blank=True)
retry_count = models.PositiveIntegerField(default=0)
# 成本统计
gpu_hours = models.FloatField(default=0.0)
estimated_cost = models.FloatField(default=0.0)
# 时间戳
created_at = models.DateTimeField(auto_now_add=True)
started_at = models.DateTimeField(null=True, blank=True)
completed_at = models.DateTimeField(null=True, blank=True)
created_by = models.ForeignKey(
User,
on_delete=models.SET_NULL,
null=True,
related_name='evaluation_tasks'
)
class Meta:
db_table = 'evaluation_tasks'
ordering = ['-created_at']
indexes = [
models.Index(fields=['tenant_id', 'status']),
models.Index(fields=['benchmark', 'status']),
models.Index(fields=['created_at']),
]
def __str__(self):
return f"{self.name} ({self.status})"
class EvaluationResult(models.Model):
"""
单条评测结果模型
存储每个样本的评测结果
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
task = models.ForeignKey(
Task,
on_delete=models.CASCADE,
related_name='results'
)
# 样本信息
sample_id = models.CharField(max_length=200, db_index=True)
sample_input = models.TextField()
sample_expected = models.TextField(blank=True)
sample_reference = models.TextField(blank=True)
# 模型输出
model_output = models.TextField()
# 评测结果
is_correct = models.BooleanField(null=True)
score = models.FloatField(null=True)
metrics = models.JSONField(default=dict)
# 详细评分
reasoning = models.TextField(blank=True, help_text="评分理由")
aspect_scores = models.JSONField(
default=dict,
help_text="各维度得分"
)
# 元数据
latency_ms = models.FloatField(null=True)
token_count = models.PositiveIntegerField(null=True)
# 时间戳
evaluated_at = models.DateTimeField(auto_now_add=True)
class Meta:
db_table = 'evaluation_results'
indexes = [
models.Index(fields=['task', 'sample_id']),
models.Index(fields=['task', 'is_correct']),
]
class Leaderboard(models.Model):
"""
排行榜模型
存储模型的排名信息
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
benchmark = models.ForeignKey(
Benchmark,
on_delete=models.CASCADE,
related_name='leaderboards'
)
model_version = models.ForeignKey(
'ai_platform.ModelVersion',
on_delete=models.CASCADE,
related_name='rankings'
)
# 排名信息
rank = models.PositiveIntegerField()
score = models.FloatField()
score_type = models.CharField(max_length=50)
# 详细指标
metrics = models.JSONField(default=dict)
# 统计信息
evaluation_count = models.PositiveIntegerField(default=1)
last_evaluation_at = models.DateTimeField()
# 时间戳
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = 'leaderboards'
unique_together = ['benchmark', 'model_version', 'score_type']
ordering = ['rank']
indexes = [
models.Index(fields=['benchmark', 'rank']),
models.Index(fields=['score']),
]四、评测引擎实现
4.1 评测引擎架构
┌─────────────────────────────────────────────────────────────────────────┐
│ 评测引擎架构 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 评测引擎工厂 (EngineFactory) │ │
│ │ │ │
│ │ ┌────────────────┬────────────────┬────────────────┐ │ │
│ │ │ │ │ │ │ │
│ │ ▼ ▼ ▼ ▼ │ │
│ │ ┌────────────┐ ┌────────────┐ ┌────────────┐ ┌────────────┐ │ │
│ │ │ 自动评测引擎 │ │ 人工评测引擎 │ │LLM评测引擎 │ │ 混合评测引擎 │ │ │
│ │ │ AutoEngine │ │HumanEngine │ │JudgeEngine │ │HybridEngine│ │ │
│ │ └──────┬─────┘ └──────┬─────┘ └──────┬─────┘ └──────┬─────┘ │ │
│ │ │ │ │ │ │ │
│ └─────────┼────────────────┼────────────────┼────────────────┼──────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ 评测执行层 (Evaluation Runner) │ │
│ │ │ │
│ │ 样本输入 ──→ 分发 ──→ 执行评测 ──→ 收集结果 ──→ 聚合计算 │ │
│ │ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌──────────┐ │ │
│ │ │ 回调写入 │ │ │
│ │ └──────────┘ │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘4.2 评测引擎核心实现
# engines/base.py
# 评测引擎基类和公共接口
from __future__ import annotations
import asyncio
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional, Dict, Any, AsyncGenerator
from enum import Enum
logger = logging.getLogger(__name__)
class EvaluationType(Enum):
"""评测类型枚举"""
AUTOMATIC = "automatic"
HUMAN = "human"
LLM_JUDGE = "llm_judge"
HYBRID = "hybrid"
@dataclass
class Sample:
"""
评测样本
包含输入、期望输出、参考答案等
"""
id: str
input: str
expected: Optional[str] = None
reference: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
return {
'id': self.id,
'input': self.input,
'expected': self.expected,
'reference': self.reference,
'metadata': self.metadata
}
@dataclass
class SampleResult:
"""
样本评测结果
"""
sample_id: str
model_output: str
# 评测结果
is_correct: Optional[bool] = None
score: Optional[float] = None
metrics: Dict[str, Any] = field(default_factory=dict)
# 详细评分（用于LLM-as-Judge）
reasoning: str = ""
aspect_scores: Dict[str, float] = field(default_factory=dict)
# 性能指标
latency_ms: float = 0.0
token_count: int = 0
# 错误信息
error: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
return {
'sample_id': self.sample_id,
'model_output': self.model_output,
'is_correct': self.is_correct,
'score': self.score,
'metrics': self.metrics,
'reasoning': self.reasoning,
'aspect_scores': self.aspect_scores,
'latency_ms': self.latency_ms,
'token_count': self.token_count,
'error': self.error
}
@dataclass
class EvaluationConfig:
"""评测配置"""
max_tokens: int = 2048
temperature: float = 0.0
top_p: float = 1.0
batch_size: int = 16
parallel_workers: int = 4
# 评测参数
pass_threshold: float = 0.7
scoring_method: str = "default" # default, strict, relaxed
# LLM-Judge参数
judge_model: Optional[str] = None
judge_prompt_template: Optional[str] = None
# 人工评测参数
annotator_count: int = 3
quality_threshold: float = 0.8
def to_dict(self) -> Dict[str, Any]:
return {
'max_tokens': self.max_tokens,
'temperature': self.temperature,
'top_p': self.top_p,
'batch_size': self.batch_size,
'parallel_workers': self.parallel_workers,
'pass_threshold': self.pass_threshold,
'scoring_method': self.scoring_method,
'judge_model': self.judge_model,
'annotator_count': self.annotator_count,
'quality_threshold': self.quality_threshold
}
@dataclass
class EvaluationResult:
"""完整评测结果"""
task_id: str
benchmark_name: str
# 聚合指标
total_samples: int = 0
passed_samples: int = 0
failed_samples: int = 0
# 综合得分
overall_score: float = 0.0
# 各维度得分（用于雷达图）
dimension_scores: Dict[str, float] = field(default_factory=dict)
# 详细指标
metrics: Dict[str, float] = field(default_factory=dict)
# 样本结果
sample_results: List[SampleResult] = field(default_factory=list)
# 统计信息
avg_latency_ms: float = 0.0
total_tokens: int = 0
total_time_seconds: float = 0.0
# 错误统计
error_count: int = 0
errors: List[str] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
return {
'task_id': self.task_id,
'benchmark_name': self.benchmark_name,
'total_samples': self.total_samples,
'passed_samples': self.passed_samples,
'failed_samples': self.failed_samples,
'pass_rate': self.passed_samples / self.total_samples if self.total_samples > 0 else 0,
'overall_score': self.overall_score,
'dimension_scores': self.dimension_scores,
'metrics': self.metrics,
'sample_results': [r.to_dict() for r in self.sample_results],
'avg_latency_ms': self.avg_latency_ms,
'total_tokens': self.total_tokens,
'total_time_seconds': self.total_time_seconds,
'error_count': self.error_count,
'errors': self.errors
}
class EvaluationEngine(ABC):
"""
评测引擎基类
所有评测引擎都必须实现此接口
"""
def __init__(self, config: EvaluationConfig):
self.config = config
self.logger = logging.getLogger(self.__class__.__name__)
@property
@abstractmethod
def evaluation_type(self) -> EvaluationType:
"""返回评测类型"""
pass
@abstractmethod
async def evaluate_sample(
self,
sample: Sample,
model: Any
) -> SampleResult:
"""
评测单个样本
Args:
sample: 评测样本
model: 模型接口
Returns:
样本评测结果
"""
pass
async def evaluate_batch(
self,
samples: List[Sample],
model: Any,
progress_callback: Optional[callable] = None
) -> List[SampleResult]:
"""
批量评测
Args:
samples: 样本列表
model: 模型接口
progress_callback: 进度回调函数
Returns:
评测结果列表
"""
results = []
total = len(samples)
# 根据并发数分组处理
batch_size = self.config.parallel_workers
for i in range(0, total, batch_size):
batch = samples[i:i + batch_size]
# 并行处理当前批次
batch_results = await asyncio.gather(
*[self.evaluate_sample(sample, model) for sample in batch],
return_exceptions=True
)
# 处理结果
for result in batch_results:
if isinstance(result, Exception):
# 处理异常
error_result = SampleResult(
sample_id=batch[0].id, # 简化处理
model_output="",
error=str(result)
)
results.append(error_result)
else:
results.append(result)
# 报告进度
if progress_callback:
progress = (i + len(batch)) / total
progress_callback(progress, i + len(batch), total)
return results
async def evaluate(
self,
task_id: str,
benchmark_name: str,
samples: List[Sample],
model: Any,
progress_callback: Optional[callable] = None
) -> EvaluationResult:
"""
执行完整评测
Args:
task_id: 任务ID
benchmark_name: 基准名称
samples: 样本列表
model: 模型接口
progress_callback: 进度回调
Returns:
完整评测结果
"""
start_time = datetime.now()
# 执行批量评测
sample_results = await self.evaluate_batch(
samples, model, progress_callback
)
# 计算聚合结果
return self._aggregate_results(
task_id=task_id,
benchmark_name=benchmark_name,
sample_results=sample_results,
start_time=start_time
)
def _aggregate_results(
self,
task_id: str,
benchmark_name: str,
sample_results: List[SampleResult],
start_time: datetime
) -> EvaluationResult:
"""聚合评测结果"""
end_time = datetime.now()
total_time = (end_time - start_time).total_seconds()
# 统计通过/失败
passed = sum(1 for r in sample_results if r.is_correct is True)
failed = sum(1 for r in sample_results if r.is_correct is False)
errors = sum(1 for r in sample_results if r.error is not None)
# 计算平均得分
valid_scores = [r.score for r in sample_results if r.score is not None]
overall_score = sum(valid_scores) / len(valid_scores) if valid_scores else 0
# 聚合维度得分
dimension_scores: Dict[str, List[float]] = {}
for result in sample_results:
for dim, score in result.aspect_scores.items():
if dim not in dimension_scores:
dimension_scores[dim] = []
dimension_scores[dim].append(score)
aggregated_dimensions = {
dim: sum(scores) / len(scores)
for dim, scores in dimension_scores.items()
}
# 计算性能统计
latencies = [r.latency_ms for r in sample_results if r.latency_ms > 0]
avg_latency = sum(latencies) / len(latencies) if latencies else 0
total_tokens = sum(r.token_count for r in sample_results)
# 收集错误信息
error_messages = [r.error for r in sample_results if r.error]
return EvaluationResult(
task_id=task_id,
benchmark_name=benchmark_name,
total_samples=len(sample_results),
passed_samples=passed,
failed_samples=failed,
overall_score=overall_score,
dimension_scores=aggregated_dimensions,
sample_results=sample_results,
avg_latency_ms=avg_latency,
total_tokens=total_tokens,
total_time_seconds=total_time,
error_count=errors,
errors=error_messages
)4.3 自动评测引擎
# engines/automatic.py
# 自动评测引擎实现
import re
import time
import asyncio
from typing import List, Dict, Any, Optional, Set
from difflib import SequenceMatcher
from engines.base import (
EvaluationEngine,
EvaluationType,
Sample,
SampleResult,
EvaluationConfig,
)
class AutomaticEvaluationEngine(EvaluationEngine):
"""
自动评测引擎
支持多种自动评测指标：
- 精确匹配 (Exact Match)
- 包含匹配 (Contains)
- 模糊匹配 (Fuzzy Match)
- 正则匹配 (Regex)
- 选择题匹配 (Choice Match)
"""
def __init__(self, config: EvaluationConfig):
super().__init__(config)
self._metrics_calculators = {
'exact_match': self._exact_match_score,
'contains': self._contains_score,
'fuzzy_match': self._fuzzy_match_score,
'choice_match': self._choice_match_score,
'rouge': self._rouge_score,
'bleu': self._bleu_score,
}
@property
def evaluation_type(self) -> EvaluationType:
return EvaluationType.AUTOMATIC
async def evaluate_sample(
self,
sample: Sample,
model: Any
) -> SampleResult:
"""评测单个样本"""
start_time = time.time()
try:
# 获取模型输出
model_output = await self._get_model_output(
model,
sample.input,
sample.metadata
)
# 计算延迟
latency_ms = (time.time() - start_time) * 1000
# 获取期望输出和评测指标
expected = sample.expected
metrics_spec = sample.metadata.get('metrics', ['exact_match'])
# 计算各项指标
scores = {}
is_correct = None
main_score = 0.0
for metric_name in metrics_spec:
if metric_name in self._metrics_calculators:
score = self._metrics_calculators[metric_name](
model_output,
expected,
sample.metadata
)
scores[metric_name] = score
# 主指标决定是否正确
if metric_name == metrics_spec[0]:
main_score = score
is_correct = score >= self.config.pass_threshold
return SampleResult(
sample_id=sample.id,
model_output=model_output,
is_correct=is_correct,
score=main_score,
metrics=scores,
latency_ms=latency_ms,
token_count=len(model_output) // 4 # 估算
)
except Exception as e:
self.logger.error(f"评测样本失败: {sample.id}, error={e}")
return SampleResult(
sample_id=sample.id,
model_output="",
error=str(e),
latency_ms=(time.time() - start_time) * 1000
)
async def _get_model_output(
self,
model: Any,
input_text: str,
metadata: Dict[str, Any]
) -> str:
"""调用模型获取输出"""
# 构建prompt
prompt = self._build_prompt(input_text, metadata)
# 调用模型
response = await model.predict(
prompt=prompt,
max_tokens=self.config.max_tokens,
temperature=self.config.temperature,
stop_sequences=metadata.get('stop_sequences', [])
)
return response.get('text', '').strip()
def _build_prompt(
self,
input_text: str,
metadata: Dict[str, Any]
) -> str:
"""构建prompt"""
system_prompt = metadata.get('system_prompt', '')
prompt_template = metadata.get('prompt_template', '{input}')
return f"{system_prompt}\n\n{prompt_template.format(input=input_text)}"
def _exact_match_score(
self,
output: str,
expected: str,
metadata: Dict[str, Any]
) -> float:
"""精确匹配评分"""
# 标准化处理
output = output.strip().lower()
expected = expected.strip().lower()
# 可选：忽略大小写和空白
ignore_case = metadata.get('ignore_case', True)
normalize_whitespace = metadata.get('normalize_whitespace', True)
if normalize_whitespace:
output = ' '.join(output.split())
expected = ' '.join(expected.split())
return 1.0 if output == expected else 0.0
def _contains_score(
self,
output: str,
expected: str,
metadata: Dict[str, Any]
) -> float:
"""包含匹配评分"""
output = output.lower()
expected = expected.lower()
if expected in output:
return 1.0
# 部分匹配
words = expected.split()
matched = sum(1 for w in words if w in output)
return matched / len(words) if words else 0.0
def _fuzzy_match_score(
self,
output: str,
expected: str,
metadata: Dict[str, Any]
) -> float:
"""模糊匹配评分（基于编辑距离）"""
similarity = SequenceMatcher(None, output, expected).ratio()
# 应用阈值
threshold = metadata.get('fuzzy_threshold', 0.8)
return similarity if similarity >= threshold else 0.0
def _choice_match_score(
self,
output: str,
expected: str,
metadata: Dict[str, Any]
) -> float:
"""选择题匹配评分"""
# 提取选项字母
choice_pattern = metadata.get('choice_pattern', r'([A-Z])')
output_choices = re.findall(choice_pattern, output.upper())
expected_choices = re.findall(choice_pattern, expected.upper())
if not expected_choices:
return 0.0
# 检查第一个选项是否匹配
if output_choices and output_choices[0] == expected_choices[0]:
return 1.0
return 0.0
def _rouge_score(
self,
output: str,
expected: str,
metadata: Dict[str, Any]
) -> float:
"""
ROUGE评分
注意：简化实现，实际应使用rouge库
"""
output_words = set(output.lower().split())
expected_words = set(expected.lower().split())
if not expected_words:
return 0.0
intersection = output_words & expected_words
return len(intersection) / len(expected_words)
def _bleu_score(
self,
output: str,
expected: str,
metadata: Dict[str, Any]
) -> float:
"""
BLEU评分
注意：简化实现，实际应使用nltk或sacrebleu库
"""
output_ngrams = self._get_ngrams(output.lower().split(), 2)
expected_ngrams = self._get_ngrams(expected.lower().split(), 2)
if not expected_ngrams:
return 0.0
intersection = output_ngrams & expected_ngrams
precision = len(intersection) / len(output_ngrams) if output_ngrams else 0
return precision
def _get_ngrams(self, tokens: List[str], n: int) -> Set[str]:
"""获取n-gram集合"""
return set(' '.join(tokens[i:i+n]) for i in range(len(tokens) - n + 1))4.4 LLM-as-Judge评测引擎
# engines/llm_judge.py
# LLM-as-Judge评测引擎
import json
import time
from typing import List, Dict, Any, Optional
from engines.base import (
EvaluationEngine,
EvaluationType,
Sample,
SampleResult,
EvaluationConfig,
)
class LLMJudgeEvaluationEngine(EvaluationEngine):
"""
LLM-as-Judge评测引擎
使用大模型作为裁判评估其他模型的输出
适用于开放式问题的评估
"""
# 默认评分prompt模板
DEFAULT_JUDGE_PROMPT = """
你是一个专业的AI模型评估专家。请评估以下模型输出在各个维度的表现。
## 评测样本
输入: {input}
期望输出: {expected}
模型输出: {output}
## 评测维度
{dimensions}
## 评分标准
每个维度0-10分：
- 9-10: 优秀，几乎完美
- 7-8: 良好，满足要求
- 5-6: 一般，有改进空间
- 3-4: 较差，存在明显问题
- 1-2: 很差，严重偏离要求
- 0: 完全错误
## 输出格式
请以JSON格式输出：
{{
"scores": {{
"dimension_name": score,
...
}},
"reasoning": "详细的评分理由",
"overall": 综合得分(0-10)
}}
"""
def __init__(self, config: EvaluationConfig):
super().__init__(config)
# 默认能力维度
self.default_dimensions = [
{"name": "准确性", "description": "回答是否正确回答了问题"},
{"name": "相关性", "description": "回答是否与问题相关"},
{"name": "完整性", "description": "回答是否完整覆盖了问题的各个方面"},
{"name": "清晰度", "description": "回答是否清晰易懂"},
{"name": "安全性", "description": "回答是否安全无害"},
]
@property
def evaluation_type(self) -> EvaluationType:
return EvaluationType.LLM_JUDGE
async def evaluate_sample(
self,
sample: Sample,
model: Any
) -> SampleResult:
"""使用LLM评测单个样本"""
start_time = time.time()
try:
# 获取模型输出
model_output = await self._get_model_output(model, sample)
# 使用Judge模型评分
judge_result = await self._judge_with_llm(
sample=sample,
model_output=model_output,
judge_model=model
)
# 计算延迟
latency_ms = (time.time() - start_time) * 1000
return SampleResult(
sample_id=sample.id,
model_output=model_output,
score=judge_result['overall'] / 10.0, # 转换为0-1
aspect_scores={
dim['name']: judge_result['scores'].get(dim['name'], 0) / 10.0
for dim in self.default_dimensions
},
reasoning=judge_result.get('reasoning', ''),
latency_ms=latency_ms
)
except Exception as e:
self.logger.error(f"LLM评测失败: {sample.id}, error={e}")
return SampleResult(
sample_id=sample.id,
model_output="",
error=str(e),
latency_ms=(time.time() - start_time) * 1000
)
async def _get_model_output(
self,
model: Any,
sample: Sample
) -> str:
"""获取被评测模型的输出"""
response = await model.predict(
prompt=sample.input,
max_tokens=self.config.max_tokens,
temperature=self.config.temperature,
**sample.metadata.get('model_params', {})
)
return response.get('text', '').strip()
async def _judge_with_llm(
self,
sample: Sample,
model_output: str,
judge_model: Any
) -> Dict[str, Any]:
"""使用LLM进行评判"""
# 构建评测prompt
prompt = self._build_judge_prompt(
input_text=sample.input,
expected=sample.expected or "",
output=model_output,
dimensions=sample.metadata.get(
'judge_dimensions',
self.default_dimensions
)
)
# 调用Judge模型
response = await judge_model.predict(
prompt=prompt,
max_tokens=1024,
temperature=0.3, # 较低温度保证稳定性
)
# 解析Judge结果
return self._parse_judge_response(response.get('text', '{}'))
def _build_judge_prompt(
self,
input_text: str,
expected: str,
output: str,
dimensions: List[Dict[str, str]]
) -> str:
"""构建评判prompt"""
# 使用自定义模板或默认模板
template = self.config.judge_prompt_template or self.DEFAULT_JUDGE_PROMPT
# 格式化维度描述
dim_descriptions = "\n".join(
f"- {d['name']}: {d.get('description', '')}"
for d in dimensions
)
return template.format(
input=input_text,
expected=expected,
output=output,
dimensions=dim_descriptions
)
def _parse_judge_response(self, response_text: str) -> Dict[str, Any]:
"""解析Judge模型的响应"""
try:
# 尝试提取JSON
json_str = self._extract_json(response_text)
result = json.loads(json_str)
# 验证和标准化
return {
'scores': result.get('scores', {}),
'reasoning': result.get('reasoning', ''),
'overall': result.get('overall', 5.0)
}
except json.JSONDecodeError:
self.logger.warning(f"解析Judge响应失败: {response_text[:100]}...")
# 返回默认值
return {
'scores': {},
'reasoning': '解析失败',
'overall': 5.0
}
def _extract_json(self, text: str) -> str:
"""从文本中提取JSON"""
import re
# 尝试提取```json...```代码块
match = re.search(r'```(?:json)?\s*([\s\S]*?)\s*```', text)
if match:
return match.group(1)
# 尝试直接解析整个文本
return text五、任务调度实现
5.1 Celery任务定义
# tasks/evaluation_tasks.py
# 评测任务Celery实现
import os
import json
import logging
from datetime import datetime
from typing import Dict, Any, List, Optional
from celery import Celery
from celery.signals import task_success, task_failure
from django.conf import settings
logger = logging.getLogger(__name__)
# 创建Celery应用
app = Celery('evaluation')
# 配置
app.conf.update(
broker_url=os.getenv('CELERY_BROKER_URL', 'redis://localhost:6379/0'),
result_backend=os.getenv('CELERY_RESULT_BACKEND', 'redis://localhost:6379/1'),
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='Asia/Shanghai',
enable_utc=True,
task_routes={
'tasks.evaluation_tasks.*': {'queue': 'evaluation'},
'tasks.report_tasks.*': {'queue': 'report'},
},
task_annotations={
'tasks.evaluation_tasks.run_evaluation': {
'rate_limit': '10/m',
'time_limit': 7200, # 2小时超时
}
}
)
@app.task(
bind=True,
name='tasks.evaluation_tasks.run_evaluation',
max_retries=3,
default_retry_delay=60,
)
def run_evaluation(self, task_id: str) -> Dict[str, Any]:
"""
执行评测任务
这是主要的评测任务入口
"""
from evaluations.models import Task, EvaluationResult
from evaluations.engine_factory import EngineFactory
from evaluations.progress_tracker import ProgressTracker
logger.info(f"开始执行评测任务: {task_id}")
try:
# 1. 获取任务信息
task = Task.objects.get(id=task_id)
# 2. 更新任务状态
task.status = Task.Status.RUNNING
task.started_at = datetime.now()
task.save()
# 3. 加载评测配置
benchmark = task.benchmark
config = task.config
# 4. 创建进度追踪器
progress_tracker = ProgressTracker(task_id)
# 5. 加载评测样本
samples = load_benchmark_samples(benchmark, config)
# 6. 创建评测引擎
engine = EngineFactory.create_engine(
evaluation_type=benchmark.evaluation_type,
config=config
)
# 7. 加载模型
model = load_model(task.model_version)
# 8. 执行评测
async def run():
result = await engine.evaluate(
task_id=str(task.id),
benchmark_name=benchmark.name,
samples=samples,
model=model,
progress_callback=progress_tracker.update
)
return result
# 由于Celery不支持async，需要使用事件循环
import asyncio
result = asyncio.get_event_loop().run_until_complete(run())
# 9. 保存结果
save_evaluation_results(task, result)
# 10. 更新任务状态
task.status = Task.Status.COMPLETED
task.completed_at = datetime.now()
task.progress = 1.0
task.summary = {
'overall_score': result.overall_score,
'passed': result.passed_samples,
'failed': result.failed_samples,
'total_time': result.total_time_seconds,
'avg_latency': result.avg_latency_ms,
}
task.save()
# 11. 触发后续任务（生成报告、更新排行）
generate_report.delay(str(task.id))
update_leaderboard.delay(str(task.id))
logger.info(
f"评测任务完成: {task_id}, score={result.overall_score:.4f}"
)
return {
'status': 'success',
'task_id': task_id,
'score': result.overall_score
}
except Task.DoesNotExist:
logger.error(f"任务不存在: {task_id}")
return {'status': 'error', 'message': 'Task not found'}
except Exception as e:
logger.error(f"评测任务失败: {task_id}, error={e}")
# 更新任务状态
try:
task = Task.objects.get(id=task_id)
task.status = Task.Status.FAILED
task.error_message = str(e)
task.retry_count = self.request.retries
task.save()
except Exception:
pass
# 重试
if self.request.retries < self.max_retries:
raise self.retry(exc=e)
return {'status': 'error', 'message': str(e)}
@app.task(
bind=True,
name='tasks.evaluation_tasks.generate_report',
max_retries=2,
)
def generate_report(self, task_id: str) -> Dict[str, Any]:
"""
生成评测报告
"""
from evaluations.models import Task
from evaluations.report_generator import ReportGenerator
logger.info(f"生成评测报告: {task_id}")
try:
task = Task.objects.get(id=task_id)
generator = ReportGenerator()
report_path = generator.generate(
task_id=task_id,
output_format='html'
)
task.report_path = report_path
task.save()
return {
'status': 'success',
'report_path': report_path
}
except Exception as e:
logger.error(f"报告生成失败: {task_id}, error={e}")
if self.request.retries < self.max_retries:
raise self.retry(exc=e)
return {'status': 'error', 'message': str(e)}
@app.task(
bind=True,
name='tasks.evaluation_tasks.update_leaderboard',
max_retries=2,
)
def update_leaderboard(self, task_id: str) -> Dict[str, Any]:
"""
更新模型排行榜
"""
from evaluations.models import Task, Leaderboard
logger.info(f"更新排行榜: {task_id}")
try:
task = Task.objects.get(id=task_id)
if task.model_version and task.status == Task.Status.COMPLETED:
# 获取当前排名
score = task.summary.get('overall_score', 0)
# 查找或创建排行榜记录
leaderboard, created = Leaderboard.objects.get_or_create(
benchmark=task.benchmark,
model_version=task.model_version,
score_type='overall',
defaults={
'rank': 0,
'score': score,
'metrics': task.summary,
'evaluation_count': 1,
'last_evaluation_at': task.completed_at
}
)
if not created:
# 更新已有记录
leaderboard.score = score
leaderboard.metrics = task.summary
leaderboard.evaluation_count += 1
leaderboard.last_evaluation_at = task.completed_at
leaderboard.save()
# 重新计算排名
recalculate_rankings(task.benchmark.id)
return {'status': 'success'}
except Exception as e:
logger.error(f"排行榜更新失败: {task_id}, error={e}")
return {'status': 'error', 'message': str(e)}
def recalculate_rankings(benchmark_id: str) -> None:
"""
重新计算排行榜排名
"""
from evaluations.models import Leaderboard
# 按分数降序排列
rankings = Leaderboard.objects.filter(
benchmark_id=benchmark_id
).order_by('-score')
# 更新排名
for rank, entry in enumerate(rankings, 1):
entry.rank = rank
entry.save()
@app.task(name='tasks.evaluation_tasks.schedule_batch_evaluation')
def schedule_batch_evaluation(
benchmark_id: str,
model_version_ids: List[str],
config: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
批量调度评测任务
"""
from evaluations.models import Task, Benchmark
from django.contrib.auth.models import User
logger.info(
f"批量调度评测: benchmark={benchmark_id}, "
f"models={len(model_version_ids)}"
)
benchmark = Benchmark.objects.get(id=benchmark_id)
# 获取默认配置
if config is None:
config = benchmark.config
task_ids = []
for model_version_id in model_version_ids:
task = Task.objects.create(
name=f"{benchmark.name} - {model_version_id[:8]}",
benchmark=benchmark,
model_version_id=model_version_id,
config=config,
priority=Task.Priority.NORMAL,
)
# 入队执行
run_evaluation.delay(str(task.id))
task_ids.append(str(task.id))
return {
'status': 'success',
'scheduled_count': len(task_ids),
'task_ids': task_ids
}
# ==================== 辅助函数 ====================
def load_benchmark_samples(benchmark, config: Dict[str, Any]) -> List:
"""加载评测样本"""
from evaluations.engine_base import Sample
# 从数据库或文件加载样本
# 这里简化实现
samples_data = benchmark.config.get('samples', [])
return [
Sample(
id=s.get('id', str(i)),
input=s['input'],
expected=s.get('expected'),
reference=s.get('reference'),
metadata=s.get('metadata', {})
)
for i, s in enumerate(samples_data)
]
def load_model(model_version_id: str) -> Any:
"""加载模型"""
# 实际实现应从模型服务获取
class MockModel:
async def predict(self, prompt: str, **kwargs):
return {'text': 'Mock response'}
return MockModel()
def save_evaluation_results(task, result) -> None:
"""保存评测结果"""
from evaluations.models import EvaluationResult
for sample_result in result.sample_results:
EvaluationResult.objects.create(
task=task,
sample_id=sample_result.sample_id,
sample_input="", # 可从context获取
model_output=sample_result.model_output,
is_correct=sample_result.is_correct,
score=sample_result.score,
metrics=sample_result.metrics,
reasoning=sample_result.reasoning,
aspect_scores=sample_result.aspect_scores,
latency_ms=sample_result.latency_ms,
token_count=sample_result.token_count,
)5.2 Kubernetes部署配置
# k8s/evaluation-platform.yaml
# 评测平台Kubernetes部署配置
apiVersion: v1
kind: Namespace
metadata:
name: evaluation-platform
labels:
name: evaluation-platform
---
# ConfigMap - Django配置
apiVersion: v1
kind: ConfigMap
metadata:
name: evaluation-django-config
namespace: evaluation-platform
data:
DEBUG: "0"
ALLOWED_HOSTS: "*"
DATABASE_URL: "postgres://user:pass@postgres:5432/evaluation"
REDIS_URL: "redis://redis:6379/0"
CELERY_BROKER_URL: "redis://redis:6379/0"
CELERY_RESULT_BACKEND: "redis://redis:6379/1"
OBJECT_STORAGE_ENDPOINT: "http://minio:9000"
OBJECT_STORAGE_BUCKET: "evaluation-reports"
---
# ConfigMap - Benchmark配置
apiVersion: v1
kind: ConfigMap
metadata:
name: benchmark-config
namespace: evaluation-platform
data:
benchmarks.yaml: |
builtin:
- name: "MMLU"
slug: "mmlu"
category: "nlp"
description: "Massive Multitask Language Understanding"
evaluation_type: "automatic"
dimensions:
- name: "语言理解"
- name: "知识推理"
- name: "数学能力"
- name: "HumanEval"
slug: "humaneval"
category: "code"
description: "Python代码生成评测"
evaluation_type: "automatic"
dimensions:
- name: "代码正确性"
- name: "代码风格"
- name: "GSM8K"
slug: "gsm8k"
category: "math"
description: "小学数学应用题"
evaluation_type: "automatic"
dimensions:
- name: "数学推理"
- name: "计算准确性"
---
# Deployment - Django API服务
apiVersion: apps/v1
kind: Deployment
metadata:
name: evaluation-api
namespace: evaluation-platform
spec:
replicas: 2
selector:
matchLabels:
app: evaluation-api
template:
metadata:
labels:
app: evaluation-api
spec:
containers:
- name: api
image: evaluation-platform/api:latest
ports:
- containerPort: 8000
envFrom:
- configMapRef:
name: evaluation-django-config
resources:
requests:
cpu: "500m"
memory: "512Mi"
limits:
cpu: "2"
memory: "2Gi"
livenessProbe:
httpGet:
path: /health/
port: 8000
initialDelaySeconds: 30
periodSeconds: 10
readinessProbe:
httpGet:
path: /ready/
port: 8000
initialDelaySeconds: 5
periodSeconds: 5
nodeSelector:
node-type: compute
---
# Service - Django API服务
apiVersion: v1
kind: Service
metadata:
name: evaluation-api
namespace: evaluation-platform
spec:
type: ClusterIP
ports:
- port: 80
targetPort: 8000
selector:
app: evaluation-api
---
# Deployment - Celery Worker
apiVersion: apps/v1
kind: Deployment
metadata:
name: evaluation-celery-worker
namespace: evaluation-platform
spec:
replicas: 3
selector:
matchLabels:
app: evaluation-celery-worker
template:
metadata:
labels:
app: evaluation-celery-worker
spec:
containers:
- name: worker
image: evaluation-platform/worker:latest
command: ["celery", "-A", "tasks", "worker", "-Q", "evaluation,report", "--concurrency=4"]
envFrom:
- configMapRef:
name: evaluation-django-config
resources:
requests:
cpu: "1"
memory: "1Gi"
limits:
cpu: "4"
memory: "8Gi"
# GPU资源
nvidia.com/gpu: "1"
volumeMounts:
- name: model-cache
mountPath: /models
volumes:
- name: model-cache
persistentVolumeClaim:
claimName: model-cache-pvc
nodeSelector:
node-type: gpu
tolerations:
- key: "nvidia.com/gpu"
operator: "Exists"
effect: "NoSchedule"
---
# Deployment - Celery Beat (定时任务调度器)
apiVersion: apps/v1
kind: Deployment
metadata:
name: evaluation-celery-beat
namespace: evaluation-platform
spec:
replicas: 1
selector:
matchLabels:
app: evaluation-celery-beat
template:
metadata:
labels:
app: evaluation-celery-beat
spec:
containers:
- name: beat
image: evaluation-platform/worker:latest
command: ["celery", "-A", "tasks", "beat", "--scheduler", "celery_beats schedulers.DatabaseScheduler"]
envFrom:
- configMapRef:
name: evaluation-django-config
---
# Deployment - Redis
apiVersion: apps/v1
kind: Deployment
metadata:
name: evaluation-redis
namespace: evaluation-platform
spec:
replicas: 1
selector:
matchLabels:
app: evaluation-redis
template:
metadata:
labels:
app: evaluation-redis
spec:
containers:
- name: redis
image: redis:7-alpine
ports:
- containerPort: 6379
resources:
requests:
memory: "256Mi"
limits:
memory: "1Gi"
---
# Service - Redis
apiVersion: v1
kind: Service
metadata:
name: evaluation-redis
namespace: evaluation-platform
spec:
type: ClusterIP
ports:
- port: 6379
selector:
app: evaluation-redis
---
# HorizontalPodAutoscaler - Celery Worker
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: evaluation-celery-worker-hpa
namespace: evaluation-platform
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: evaluation-celery-worker
minReplicas: 2
maxReplicas: 10
metrics:
- type: External
external:
metric:
name: celery_task_queue_length
selector:
matchLabels:
queue: evaluation
target:
type: AverageValue
averageValue: "10"六、报告生成实现
6.1 报告生成服务
# report/generator.py
# 评测报告生成器
import os
from datetime import datetime
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
import logging
from django.conf import settings
from django.template import Template, Context
logger = logging.getLogger(__name__)
@dataclass
class ReportData:
"""报告数据"""
task_id: str
task_name: str
benchmark_name: str
model_name: str
# 基本指标
overall_score: float
pass_rate: float
total_samples: int
# 维度得分（用于雷达图）
dimension_scores: Dict[str, float]
# 详细指标
metrics: Dict[str, float]
# 样本结果统计
sample_stats: Dict[str, Any]
# 性能指标
performance: Dict[str, float]
# 错误分析
error_analysis: Dict[str, Any]
# 时间信息
created_at: str
duration_seconds: float
class ReportGenerator:
"""
评测报告生成器
支持生成HTML、PDF、JSON格式的报告
"""
def __init__(self):
self.template_dir = os.path.join(
settings.BASE_DIR,
'templates',
'evaluation_reports'
)
self.output_dir = os.path.join(
settings.MEDIA_ROOT,
'reports'
)
os.makedirs(self.output_dir, exist_ok=True)
def generate(
self,
task_id: str,
output_format: str = 'html'
) -> str:
"""
生成评测报告
Args:
task_id: 任务ID
output_format: 输出格式 (html/pdf/json)
Returns:
报告文件路径
"""
# 1. 收集报告数据
report_data = self._collect_report_data(task_id)
# 2. 根据格式生成报告
if output_format == 'html':
return self._generate_html(report_data)
elif output_format == 'pdf':
return self._generate_pdf(report_data)
elif output_format == 'json':
return self._generate_json(report_data)
else:
raise ValueError(f"不支持的格式: {output_format}")
def _collect_report_data(self, task_id: str) -> ReportData:
"""收集报告所需数据"""
from evaluations.models import Task, EvaluationResult
task = Task.objects.select_related(
'benchmark', 'model_version'
).get(id=task_id)
# 获取样本结果
results = EvaluationResult.objects.filter(task=task)
# 计算维度得分
dimension_scores = self._calculate_dimension_scores(results)
# 计算详细指标
metrics = self._calculate_metrics(results)
# 样本统计
sample_stats = self._calculate_sample_stats(results)
# 性能统计
performance = self._calculate_performance(results, task)
# 错误分析
error_analysis = self._analyze_errors(results)
return ReportData(
task_id=str(task.id),
task_name=task.name,
benchmark_name=task.benchmark.name,
model_name=task.model_version.name if task.model_version else "Unknown",
overall_score=task.summary.get('overall_score', 0),
pass_rate=task.summary.get('pass_rate', 0),
total_samples=task.summary.get('total_samples', 0),
dimension_scores=dimension_scores,
metrics=metrics,
sample_stats=sample_stats,
performance=performance,
error_analysis=error_analysis,
created_at=task.completed_at.isoformat() if task.completed_at else "",
duration_seconds=task.summary.get('total_time', 0)
)
def _generate_html(self, data: ReportData) -> str:
"""生成HTML报告"""
template_path = os.path.join(self.template_dir, 'report.html')
with open(template_path, 'r') as f:
template = Template(f.read())
context = Context({
'report': data,
'generated_at': datetime.now().isoformat(),
'radar_chart_data': self._prepare_radar_chart_data(data),
'line_chart_data': self._prepare_line_chart_data(data),
})
html_content = template.render(context)
# 保存文件
filename = f"report_{data.task_id}.html"
output_path = os.path.join(self.output_dir, filename)
with open(output_path, 'w', encoding='utf-8') as f:
f.write(html_content)
return output_path
def _generate_json(self, data: ReportData) -> str:
"""生成JSON报告"""
import json
report_dict = {
'task_id': data.task_id,
'task_name': data.task_name,
'benchmark_name': data.benchmark_name,
'model_name': data.model_name,
'summary': {
'overall_score': data.overall_score,
'pass_rate': data.pass_rate,
'total_samples': data.total_samples,
},
'dimension_scores': data.dimension_scores,
'metrics': data.metrics,
'sample_stats': data.sample_stats,
'performance': data.performance,
'error_analysis': data.error_analysis,
'metadata': {
'created_at': data.created_at,
'duration_seconds': data.duration_seconds,
'generated_at': datetime.now().isoformat(),
}
}
filename = f"report_{data.task_id}.json"
output_path = os.path.join(self.output_dir, filename)
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(report_dict, f, ensure_ascii=False, indent=2)
return output_path
def _generate_pdf(self, data: ReportData) -> str:
"""生成PDF报告（需要weasyprint或其他PDF库）"""
# 先生成HTML
html_path = self._generate_html(data)
# 转换为PDF
# 这里简化实现，实际应使用weasyprint或playwright
# from weasyprint import HTML
# HTML(filename=html_path).write_pdf(pdf_path)
return html_path.replace('.html', '.pdf')
def _calculate_dimension_scores(self, results) -> Dict[str, float]:
"""计算各维度得分"""
dimension_totals: Dict[str, List[float]] = {}
for result in results:
for dim, score in (result.aspect_scores or {}).items():
if dim not in dimension_totals:
dimension_totals[dim] = []
dimension_totals[dim].append(score)
return {
dim: sum(scores) / len(scores)
for dim, scores in dimension_totals.items()
}
def _calculate_metrics(self, results) -> Dict[str, float]:
"""计算详细指标"""
total = results.count()
if total == 0:
return {}
passed = results.filter(is_correct=True).count()
return {
'pass_rate': passed / total,
'fail_rate': (total - passed) / total,
'average_score': sum(r.score or 0 for r in results) / total,
}
def _calculate_sample_stats(self, results) -> Dict[str, Any]:
"""计算样本统计"""
total = results.count()
correct = results.filter(is_correct=True).count()
incorrect = results.filter(is_correct=False).count()
errors = results.exclude(error='').count()
return {
'total': total,
'correct': correct,
'incorrect': incorrect,
'errors': errors,
'success_rate': correct / total if total > 0 else 0,
}
def _calculate_performance(self, results, task) -> Dict[str, float]:
"""计算性能指标"""
return {
'total_duration_seconds': task.summary.get('total_time', 0),
'average_latency_ms': task.summary.get('avg_latency', 0),
'total_tokens': sum(r.token_count or 0 for r in results),
'gpu_hours': task.gpu_hours,
}
def _analyze_errors(self, results) -> Dict[str, Any]:
"""分析错误"""
error_results = results.exclude(error='')
error_types: Dict[str, int] = {}
for result in error_results:
error_msg = result.error or 'Unknown'
error_types[error_msg] = error_types.get(error_msg, 0) + 1
return {
'total_errors': error_results.count(),
'error_types': error_types,
'error_rate': error_results.count() / results.count() if results.count() > 0 else 0,
}
def _prepare_radar_chart_data(self, data: ReportData) -> Dict[str, Any]:
"""准备雷达图数据"""
return {
'labels': list(data.dimension_scores.keys()),
'datasets': [{
'label': data.model_name,
'data': list(data.dimension_scores.values()),
}]
}
def _prepare_line_chart_data(self, data: ReportData) -> Dict[str, Any]:
"""准备折线图数据（如果有历史数据）"""
# 简化实现
return {
'labels': ['Step 1', 'Step 2', 'Step 3'],
'datasets': [{
'label': 'Score',
'data': [0.5, 0.7, data.overall_score],
}]
}七、API接口实现
# api/views.py
# 评测平台API视图
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from django_filters.rest_framework import DjangoFilterBackend
from evaluations.models import Benchmark, Task, EvaluationResult, Leaderboard
from evaluations.serializers import (
BenchmarkSerializer,
TaskSerializer,
TaskCreateSerializer,
EvaluationResultSerializer,
LeaderboardSerializer,
)
from tasks.evaluation_tasks import run_evaluation, schedule_batch_evaluation
class BenchmarkViewSet(viewsets.ModelViewSet):
"""
评测基准API
支持列表、详情、内置基准管理
"""
queryset = Benchmark.objects.all()
serializer_class = BenchmarkSerializer
permission_classes = [IsAuthenticated]
filter_backends = [DjangoFilterBackend]
filterset_fields = ['category', 'status', 'is_builtin']
@action(detail=True, methods=['get'])
def samples(self, request, pk=None):
"""获取基准的评测样本"""
benchmark = self.get_object()
samples = benchmark.config.get('samples', [])
return Response({
'benchmark': benchmark.name,
'sample_count': len(samples),
'samples': samples[:10] # 只返回前10个样本作为预览
})
class TaskViewSet(viewsets.ModelViewSet):
"""
评测任务API
支持创建、执行、查询评测任务
"""
queryset = Task.objects.all()
permission_classes = [IsAuthenticated]
filter_backends = [DjangoFilterBackend]
filterset_fields = ['status', 'benchmark', 'priority']
def get_serializer_class(self):
if self.action == 'create':
return TaskCreateSerializer
return TaskSerializer
def perform_create(self, serializer):
"""创建任务"""
task = serializer.save(created_by=self.request.user)
# 入队执行
run_evaluation.delay(str(task.id))
@action(detail=True, methods=['post'])
def cancel(self, request, pk=None):
"""取消任务"""
task = self.get_object()
if task.status in [Task.Status.PENDING, Task.Status.RUNNING]:
task.status = Task.Status.CANCELLED
task.save()
return Response({'status': 'cancelled'})
return Response(
{'error': f'无法取消状态为 {task.status} 的任务'},
status=status.HTTP_400_BAD_REQUEST
)
@action(detail=True, methods=['post'])
def retry(self, request, pk=None):
"""重试失败的任务"""
task = self.get_object()
if task.status == Task.Status.FAILED:
task.status = Task.Status.PENDING
task.error_message = ''
task.save()
run_evaluation.delay(str(task.id))
return Response({'status': 'retrying'})
return Response(
{'error': '只能重试失败的任务'},
status=status.HTTP_400_BAD_REQUEST
)
@action(detail=False, methods=['post'])
def batch_create(self, request):
"""批量创建评测任务"""
benchmark_id = request.data.get('benchmark_id')
model_version_ids = request.data.get('model_version_ids', [])
config = request.data.get('config')
if not benchmark_id or not model_version_ids:
return Response(
{'error': '需要指定 benchmark_id 和 model_version_ids'},
status=status.HTTP_400_BAD_REQUEST
)
result = schedule_batch_evaluation.delay(
benchmark_id=benchmark_id,
model_version_ids=model_version_ids,
config=config
)
return Response({
'status': 'scheduled',
'task_ids': result.get()
})
@action(detail=True, methods=['get'])
def results(self, request, pk=None):
"""获取任务的所有评测结果"""
task = self.get_object()
results = EvaluationResult.objects.filter(task=task)
page = self.paginate_queryset(results)
if page is not None:
serializer = EvaluationResultSerializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = EvaluationResultSerializer(results, many=True)
return Response(serializer.data)
class LeaderboardViewSet(viewsets.ReadOnlyModelViewSet):
"""
排行榜API
只读API，支持按基准查询排行榜
"""
queryset = Leaderboard.objects.all()
serializer_class = LeaderboardSerializer
permission_classes = [IsAuthenticated]
filter_backends = [DjangoFilterBackend]
filterset_fields = ['benchmark']
ordering_fields = ['rank', 'score', 'evaluation_count']
@action(detail=False, methods=['get'])
def top(self, request):
"""获取排行榜前三名"""
benchmark_slug = request.query_params.get('benchmark')
queryset = self.get_queryset()
if benchmark_slug:
queryset = queryset.filter(benchmark__slug=benchmark_slug)
top_entries = queryset.order_by('rank')[:3]
serializer = self.get_serializer(top_entries, many=True)
return Response(serializer.data)
@action(detail=False, methods=['get'])
def comparison(self, request):
"""
模型对比
返回指定模型的对比数据
"""
model_ids = request.query_params.getlist('model_id')
benchmark_slug = request.query_params.get('benchmark')
if not model_ids or not benchmark_slug:
return Response(
{'error': '需要指定 model_id 和 benchmark'},
status=status.HTTP_400_BAD_REQUEST
)
entries = Leaderboard.objects.filter(
benchmark__slug=benchmark_slug,
model_version_id__in=model_ids
)
serializer = self.get_serializer(entries, many=True)
return Response(serializer.data)八、优缺点分析与后续优化
8.1 方案优点
统一标准化：
建立了统一的评测标准，不同团队的模型可以在同一基准下比较
内置多种Benchmark，覆盖NLP、代码、数学、推理等多个领域
评测结果具有可追溯性，方便问题排查和性能分析
架构灵活性：
支持多种评测类型（自动、人工、LLM-as-Judge），适应不同场景需求
可扩展的评测引擎设计，方便添加新的评测方法
基于Celery的任务调度支持水平扩展
资源效率：
基于Kubernetes统一调度GPU资源，提高利用率
支持批量评测和定时评测，减少人工操作
任务优先级机制确保紧急评测优先执行
数据驱动：
完整的评测数据存储，支持历史趋势分析
自动生成可视化报告，直观展示评测结果
排行榜机制促进团队间的良性竞争
8.2 当前局限
评测基准覆盖不足：
内置Benchmark数量有限，覆盖场景不够全面
缺乏针对特定业务场景的专业评测基准
需要持续建设和维护高质量的评测数据集
LLM-as-Judge成本较高：
使用大模型作为裁判会产生较高的API调用成本
评判结果的稳定性有待提升
评判标准的校准需要大量人工工作
模型支持有限：
当前主要支持文本模型的评测
对多模态模型（图像、视频、语音）的评测支持不足
模型调用接口需要适配不同的推理服务
8.3 后续优化方向
Benchmark生态建设：
引入更多业界标准Benchmark
支持团队自定义Benchmark和评测模板
建立Benchmark质量评估和筛选机制
评测能力增强：
增加对抗性评测、鲁棒性评测等高级评测能力
支持评测结果的自动化分析和建议
引入主动学习，自动识别需要重点评测的样本
多模态扩展：
扩展图像、视频、语音模型的评测支持
引入跨模态一致性评测
支持多模态融合效果的评估
智能化运维：
引入智能调度，根据任务特征自动选择最优资源
实现预测性扩缩容，提前准备评测资源
建立评测成本的实时监控和优化
统一模型评测平台的建设是一个持续迭代的过程。本文介绍的是v1.0版本的核心能力，后续我们将根据用户反馈和业务需求持续优化，为企业AI平台的能力评估提供更加专业、高效的支持。
评论区