题解评测自动化流水线：批量跑分、回归测试与持续验证
一、手动测一道题 5 分钟，测 100 道呢
刷完一道算法题后，提交到 LeetCode 等在线评测平台，等几秒就能看到结果。这是单题的评测流程。但如果你在搭建一个 AI 题解生成系统，每天要产出几十篇题解，每篇都包含不同语言的实现，手动逐一提交验证就完全不现实了。
你需要一套自动化评测流水线——把代码、测试用例、评测环境串联起来，批量执行，自动汇总结果。这篇文章讨论的就是如何设计这样一条管道。
二、流水线的整体架构
flowchart LR
A[代码仓库] –> B[触发评测]
B –> C{语言路由}
C –>|Python| D[Python 沙箱]
C –>|Java| E[Java 沙箱]
C –>|C++| F[C++ 沙箱]
C –>|Go| G[Go 沙箱]
D –> H[结果收集器]
E –> H
F –> H
G –> H
H –> I[评测报告生成]
I –> J{评审策略}
J –>|全部通过| K[标记为已通过]
J –>|部分失败| L[生成错误报告]
J –>|全部失败| M[触发人工审查]
L –> N[通知相关开发者]
M –> N
流水线分为五个阶段：
三、核心实现
import subprocess
import json
import time
import tempfile
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
from concurrent.futures import ThreadPoolExecutor, as_completed
@dataclass
class TestCase:
"""单个测试用例"""
input_data: str # 标准输入内容
expected_output: str # 期望的标准输出
description: str = "" # 用例描述
@dataclass
class EvalResult:
"""单次评测结果"""
problem_id: int
language: str
passed: bool
time_ms: float
memory_mb: float
error_msg: str = ""
test_case_index: int = 0
class SandboxExecutor:
"""沙箱执行器：在受限环境中运行用户代码
安全设计：
1. 每次执行在独立的临时目录中
2. 设置严格的超时和内存限制
3. 禁止网络访问和文件系统写入（除临时目录外）
"""
# 各语言超时限制（毫秒）
TIME_LIMITS = {"python": 2000, "java": 3000, "cpp": 2000, "go": 2000}
def execute(
self,
code: str,
language: str,
test_case: TestCase,
work_dir: str,
) -> EvalResult:
"""在沙箱中执行代码并返回结果"""
# 1. 写入代码文件
ext = {"python": "py", "java": "java", "cpp": "cpp", "go": "go"}
code_file = Path(work_dir) / f"solution.{ext.get(language, 'py')}"
code_file.write_text(code, encoding="utf-8")
# 2. 根据语言构造执行命令
cmd = self._build_command(language, code_file)
# 3. 执行并捕获输出
try:
start = time.perf_counter()
proc = subprocess.run(
cmd,
input=test_case.input_data,
capture_output=True,
text=True,
timeout=self.TIME_LIMITS.get(language, 2000) / 1000,
cwd=work_dir,
)
elapsed = (time.perf_counter() – start) * 1000 # 转为毫秒
actual_output = proc.stdout.strip()
expected = test_case.expected_output.strip()
return EvalResult(
problem_id=0, # 由调用方填充
language=language,
passed=(actual_output == expected),
time_ms=round(elapsed, 2),
memory_mb=0, # 精确内存数据需依赖 cgroup 或 psutil
error_msg="" if actual_output == expected else
f"期望: {expected[:100]}\\n实际: {actual_output[:100]}",
)
except subprocess.TimeoutExpired:
return EvalResult(
problem_id=0,
language=language,
passed=False,
time_ms=self.TIME_LIMITS.get(language, 2000),
memory_mb=0,
error_msg="执行超时",
)
except Exception as e:
return EvalResult(
problem_id=0,
language=language,
passed=False,
time_ms=0,
memory_mb=0,
error_msg=f"运行异常: {str(e)}",
)
def _build_command(self, language: str, code_file: Path) -> list[str]:
"""根据语言构造执行命令"""
commands = {
"python": ["python3", str(code_file)],
"java": ["java", str(code_file.with_suffix(""))],
"cpp": [str(code_file.with_suffix(""))],
"go": ["go", "run", str(code_file)],
}
return commands.get(language, ["python3", str(code_file)])
class EvalPipeline:
"""评测流水线：批量执行评测任务"""
def __init__(self, max_workers: int = 4):
self.executor = SandboxExecutor()
self.max_workers = max_workers
def run_batch(
self,
problems: list[dict], # 每项包含 code, language, test_cases
) -> dict:
"""批量评测，返回汇总报告"""
all_results: list[EvalResult] = []
problem_reports: dict[int, dict] = {}
with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
futures = []
for prob in problems:
for tc_idx, tc in enumerate(prob["test_cases"]):
with tempfile.TemporaryDirectory() as work_dir:
future = pool.submit(
self._eval_single,
prob["problem_id"],
prob["code"],
prob["language"],
tc,
tc_idx,
work_dir,
)
futures.append(future)
for future in as_completed(futures):
result = future.result()
all_results.append(result)
# 按题号汇总
for result in all_results:
pid = result.problem_id
if pid not in problem_reports:
problem_reports[pid] = {
"total": 0,
"passed": 0,
"avg_time_ms": 0,
"errors": [],
}
report = problem_reports[pid]
report["total"] += 1
if result.passed:
report["passed"] += 1
else:
report["errors"].append(result.error_msg)
# 计算平均耗时
for pid, report in problem_reports.items():
times = [r.time_ms for r in all_results if r.problem_id == pid]
report["avg_time_ms"] = round(sum(times) / len(times), 2) if times else 0
return {
"summary": {
"total_problems": len(problem_reports),
"total_cases": len(all_results),
"total_passed": sum(1 for r in all_results if r.passed),
"pass_rate": (
sum(1 for r in all_results if r.passed) / len(all_results)
if all_results else 0
),
},
"details": problem_reports,
}
def _eval_single(
self,
problem_id: int,
code: str,
language: str,
test_case: TestCase,
tc_idx: int,
work_dir: str,
) -> EvalResult:
"""执行单个评测任务"""
result = self.executor.execute(code, language, test_case, work_dir)
result.problem_id = problem_id
result.test_case_index = tc_idx
return result
四、边界分析与权衡
4.1 沙箱安全性
当前使用 subprocess + 超时限制做基础隔离。但在平台级部署时，必须考虑恶意代码的风险——如 rm -rf /、fork 炸弹、网络外连等。真正的生产环境应该使用 Docker 容器或 gVisor 做进程级沙箱。
4.2 评测精度
Python 的 subprocess 只能捕获标准输出，无法精确测量内存使用。内存评测通常依赖 Linux cgroup 或专门的评测内核模块（如 cgroup memory controller）。
4.3 大规模评测的调度
当题目数量达到数千级别时，简单的线程池不够。需要引入分布式任务队列（如 Celery + Redis），将评测任务分发到多台机器并行执行。
4.4 浮点误差处理
如果题目的期望输出是浮点数，直接字符串比较会因为精度问题导致误判。需要用相对误差或绝对误差做近似匹配，而非严格相等。
五、总结
自动化评测流水线的核心挑战不在评测本身——单个代码运行测试是简单的事。挑战在于：在保证安全性的前提下，让评测规模化、可观测、可追溯。随着 AI 题解生成能力的提升，这条流水线会更加重要——它不是一次性的工具，而是持续保障题解质量的基础设施。
