简介这是一份基于Java和Python开发的AI模型评估平台后端设计源码适合需要搭建模型测试与评估服务后台的开发者与研究人员。项目以Java构建核心后端框架辅以Python脚本处理算法与数据任务结合两者优势适用于模型评估、性能对比和指标管理等场景。压缩包共76个文件大小143KB以66个Java源文件为主包含2个Python脚本以及Docker容器化配置、Maven工程文件、XML/YAML定义、说明文档和Git忽略规则等辅助文件。目录结构清晰便于快速搭建环境和二次开发。当前已有496人学习下载。通过这份源码读者能掌握AI模型评估后端的功能模块划分、依赖管理及容器化部署流程获得一套完整可借鉴的工程实现可在其基础上扩展评估指标、接口服务或集成更多模型算法提升模型评估工作效率。1. 为什么AI模型评估平台要把Java和Python拼在一起写想象一个场景算法团队交付了十几个模型版本业务方追问哪个能上线。评估平台就是把这些模型丢到统一数据集上跑指标出结论。真正动手做你会发现评测流程一半是工程一半是算法任务调度、权限、进度、报告归Java模型推理、指标计算、数据清洗归Python。网上能找到的基于Java和Python开发的AI模型评估平台后端设计源码大多就是这个套路——Java Spring Boot撑起Web端和任务编排Python FastAPI或Flask做推理计算。适合谁适合既要给后端同事交差、又要跟算法对接的测试开发或后端开发。做之前先信一件事这不是多此一举而是让两种语言各干擅长的事。2. 后端总体架构与进程通信Java负责管Python负责算2.1 技术选型边界为什么不是Java死磕到底或Python单飞一个AI模型评估平台核心要解决三件事管理模型和数据集、按配置执行评测、输出量化指标。第一和第三件里管理是工程第三件里计算是算法。Java在工程面的优势不必多说Spring Boot的依赖注入、事务、安全框架做Web后端几乎是标准答案Python在算法面的优势也很直接onnx、torch、sklearn、pandas这些库补齐了评测计算链。混合架构表面上多了一次进程通信其实换来了更大的选择空间模型推理框架想换就换只要暴露统一接口Java侧保持稳定算法迭代不影响主流程。常见做法是Java侧Spring Boot开两个端口8080给前端/外部内部保留一个管理端口Python侧FastAPI监听8000只接受Java侧调用或MQ消息。为什么不是Python单飞因为评测系统一旦接入SSO、多租户、审计日志、工单审批Python要绕很多路。为什么不是Java死磕到底因为模型推理和指标计算如果都靠Java重写你会陷进造轮子。这个边界是我做了多个评估项目后觉得最省心的Java只做流程和APIPython只做计算。前后端分离项目实战里前端不直接碰Python服务所有请求都过Java这样权限和跨域也能统一收口。2.2 三种进程通信方式HTTP回调、消息队列、共享文件系统的取舍评估平台里Java和Python之间的通信我见过三种主流做法它们不是随便选的直接决定你后面会不会半夜被叫起来。通信方式适用场景成本主要坑同步HTTP POST评测耗时10秒比如单模型小样本最低只要一个RestTemplate长时间评测会占住连接前端转圈消息队列RabbitMQ/Kafka评测分钟级到小时级需要额外部署中间件重复消费、消息顺序、死信处理共享文件系统数据库轮询离线批量评测没有中间件最低依赖同一块存储任务状态更新延迟容易重复扫描我的选型标准很简单看一眼评测任务单条的P99耗时。如果10秒内能返回结果直接同步HTTP不要为了架构好看上消息队列如果评测要跑几分钟甚至几小时同步HTTP就是给自己埋雷连接一断Java侧拿到异常Python侧还在傻算结果是黑匣子。这时候应该走消息队列或异步回调。共享文件系统适合内网离线环境比如每天凌晨批量跑一批模型Java把任务JSON写进目录Python worker扫到就开工结果写数据库。但轮询间隔至少要30秒不然文件还没写完就被读了。2.3 后端源码目录结构与最小骨架我一般会这样组织一个混合后端工程Java和Python独立成两个项目避免互相污染依赖eval-platform/ ├── java-backend/ │ ├── pom.xml │ └── src/main/java/com/eval/ │ ├── controller/ # REST接口 │ ├── service/ # 业务逻辑 │ ├── mapper/ # MyBatis数据访问 │ ├── task/ # 线程池/任务调度 │ └── config/ # Redis、RestTemplate配置 ├── python-eval/ │ ├── requirements.txt │ ├── app.py # FastAPI入口 │ ├── metrics/ # 指标计算实现 │ └── workers/ # 消息队列消费者 └── deploy/ ├── docker-compose.yml └── nginx.conf环境准备这一步别偷懒。Java侧用8或11都可以Python侧推荐3.8到3.10python安装时顺便把venv建好别把依赖装到全局。Python侧有个特别疼的点numpy、torch、scikit-learn版本必须锁死不然两台机器跑出来的指标不一样。Java侧依赖相对简单Spring Boot就用稳定版加上MySQL驱动、MyBatis、Redis客户端。如果评测需要GPUPython服务的机器和Java后端机器分开部署因为GPU驱动的异常不应该拖垮整个平台。3. 数据模型与任务状态机一次评测从提交到出报告的完整轨迹3.1 核心表设计模型、数据集、评测任务、指标结果评测平台的数据模型比普通项目多一层任务概念。所有业务围绕四个表模型表、数据集表、评测任务表、指标结果表。我第一次设计时漏了指标结果表直接把结果塞在任务表里的JSON字段结果一个任务跑几十个指标查询和对比都痛苦。下面是核心表结构CREATE TABLE model_info ( id bigint NOT NULL AUTO_INCREMENT, model_name varchar(128) NOT NULL, model_version varchar(32) NOT NULL, framework varchar(32) NOT NULL COMMENT pytorch/tensorflow/onnx, model_path varchar(512) NOT NULL, status tinyint NOT NULL DEFAULT 0, created_by varchar(64) NOT NULL, create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk_model_version (model_name,model_version) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT模型注册表; CREATE TABLE dataset_info ( id bigint NOT NULL AUTO_INCREMENT, dataset_name varchar(128) NOT NULL, dataset_path varchar(512) NOT NULL, sample_count int NOT NULL, labels json DEFAULT NULL, create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT评测数据集表; CREATE TABLE eval_task ( id bigint NOT NULL AUTO_INCREMENT, task_no varchar(32) NOT NULL, model_id bigint NOT NULL, dataset_id bigint NOT NULL, status tinyint NOT NULL DEFAULT 0 COMMENT 0待执行,1运行中,2成功,3失败,4取消, priority tinyint NOT NULL DEFAULT 5, config_json json DEFAULT NULL, result_summary json DEFAULT NULL, create_by varchar(64) NOT NULL, create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, finish_time datetime DEFAULT NULL, PRIMARY KEY (id), UNIQUE KEY uk_task_no (task_no), KEY idx_model_dataset (model_id,dataset_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT评测任务表; CREATE TABLE eval_metric_result ( id bigint NOT NULL AUTO_INCREMENT, task_id bigint NOT NULL, metric_name varchar(64) NOT NULL, metric_value double NOT NULL, sample_count int NOT NULL, extra_info json DEFAULT NULL, create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id), KEY idx_task (task_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT指标结果表;注意几个细节模型表上加了唯一键防止同一个模型版本被重复注册算法团队经常手滑传两遍数据集表的labels字段存标签列表因为分类任务评测需要知道类别名任务表的task_no是业务单号对外展示用不能暴露自增id。指标结果表单独一张是因为一个任务会产出多个指标存JSON虽然省事但想按准确率排序模型版本时就傻眼了。3.2 任务状态机与幂等设计评测任务的常见状态只有五个待执行、运行中、成功、失败、取消。状态流转必须严格限定不然会出现成功后又变失败这种玄学问题。我的规则是只有待执行可以变运行中或取消运行中只能变成功或失败失败可以被重试重新拉回待执行。不允许运行中直接取消因为Python侧可能已经在跑模型强行改状态会造成结果回写时覆盖新状态。Java侧创建任务时必须做幂等。同一份配置连续提交两次第二次应该提示任务已存在而不是再造一个新任务。最简单的做法是用task_no做唯一键插入前查一次高级一点用Redis的setNX做短时间窗口内的幂等控制。给一个Spring Boot创建任务的最小实现public String createTask(CreateTaskRequest req) { String taskNo EV System.currentTimeMillis() RandomUtil.randomNumbers(4); // 业务校验模型和数据集必须存在否则直接抛异常 ModelInfo model modelMapper.selectById(req.getModelId()); DatasetInfo dataset datasetMapper.selectById(req.getDatasetId()); if (model null || dataset null) { throw new BizException(模型或数据集不存在); } // 幂等检查同一模型数据集配置摘要短时间内不重复创建 String configKey req.getModelId() _ req.getDatasetId() _ req.getConfig().hashCode(); if (redisTemplate.hasKey(eval:dup: configKey)) { throw new BizException(重复提交请稍后再试); } redisTemplate.opsForValue().set(eval:dup: configKey, taskNo, Duration.ofMinutes(10)); EvalTask task new EvalTask(); task.setTaskNo(taskNo); task.setModelId(req.getModelId()); task.setDatasetId(req.getDatasetId()); task.setStatus(0); task.setConfigJson(req.getConfig().toJSONString()); taskMapper.insert(task); // 发送到MQ或直接调用Python这里先占位 taskDispatchService.dispatch(task); return taskNo; }这段代码里有一点容易被忽略配置的hashCode只能做粗粒度判断如果两个任务配置只是batchSize不同其他都一样理论上允许并行。但实际业务中同一模型同一数据集同时跑两个不同batchSize的评测没有意义所以锁住是划算的。Redis过期时间设10分钟足够覆盖人工重复点击的窗口。3.3 状态机踩坑点超时、重试和取消的逻辑闭环第一超时处理。很多任务卡在运行中是因为Python进程崩溃但Java侧没收到回调。我的做法是任务表加一个planned_finish_time字段待执行时根据历史评测耗时估算出一个预计完成时间定时任务每分钟扫一次超过预计时间仍停留在运行中的直接标记为失败并带上timeout错误信息。这个兜底必须有否则用户会看到一个永远转圈的记录。第二重试次数。失败任务不能无限重试否则一个坏数据集会让整个任务队列堵住。每个任务允许重试3次每次创建成功时从history表里查重试次数超过就置为最终失败。还有重试时要把旧的失败记录保留方便对比是哪次开始翻车。第三取消操作。取消任务只对待执行状态有效对运行中任务只能标记取消中等Python worker计算完再回调确认。这样设计是为了保证数据一致性虽然取消不够即时但至少不会出现任务结束时回写一个已取消的状态。4. Java后端实现从API到调度再到Python调用的完整链路4.1 创建评测任务的接口与参数校验Java后端是评测平台的入口接口设计要让你能放心地暴露给前端。参数校验我用Spring Boot自带的Valid 自定义注解避免在业务代码里写一堆if。下面是一个Controller示例RestController RequestMapping(/api/eval) public class EvalTaskController { Autowired private EvalTaskService evalTaskService; PostMapping(/task) public ResultString createTask(RequestBody Valid CreateTaskRequest req) { // Valid 触发参数校验比如modelId不能为空、数据集不能为空 return Result.ok(evalTaskService.createTask(req)); } GetMapping(/task/{taskNo}) public ResultEvalTaskVO queryTask(PathVariable(taskNo) String taskNo) { return Result.ok(evalTaskService.query(taskNo)); } PostMapping(/task/cancel) public ResultVoid cancelTask(RequestBody CancelTaskRequest req) { evalTaskService.cancel(req.getTaskNo()); return Result.ok(); } }CreateTaskRequest里的字段至少包括model_id、dataset_id、config_json。config_json是灵活配置项比如batch_size、是否使用GPU、需要计算哪些指标。校验时要检查config里的指标名是不是在白名单里防止有人传一个total_loss之类根本不存在的指标。前后端分离项目实战中这里通常会遇到后端跨域问题。如果你有前端页面建议在Java侧统一配置CorsFilter允许的Origin用配置项管理别直接打星号因为评估平台可能涉及模型文件下载权限校验不能省。4.2 任务调度独立线程池还是消息队列评测任务的调度我分两级第一级是Java内部线程池负责把任务交给Python并监听结果第二级是消息队列负责在评测时间长时做缓冲。先说线程池配置Bean(evalThreadPool) public ThreadPoolTaskExecutor evalThreadPool() { ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); executor.setCorePoolSize(4); executor.setMaxPoolSize(8); executor.setQueueCapacity(1000); executor.setKeepAliveSeconds(60); executor.setThreadNamePrefix(eval-task-); // CallerRunsPolicy任务满时由调用线程执行保证不丢任务 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.initialize(); return executor; }这几个参数不是随便拍的。评测任务的耗时大头在Python侧推理和图片解码Java侧只是发起HTTP请求和更新数据库所以核心线程数4就够开太多反而浪费。queueCapacity设1000是控制积压可见性如果任务超过1000后续任务直接走拒绝策略这样你能在监控指标里看到被拒绝的数量而不是让任务在内存里积压到OOM。CallerRunsPolicy的意思是线程池都满了的时候由提交任务的调用线程自己去做分发虽然会短暂阻塞接口但不会丢任务。对于评测这种可以接受等待的任务这个策略比丢弃好得多。如果评测任务要跑10分钟以上我不建议用Java线程池直接调因为一次重启就可能丢几十个任务。这时候我会引入RabbitMQJava创建任务后发一条消息Python worker消费并执行执行完回调Java接口更新状态。RabbitMQ的ack机制能保证任务不被吞掉。这个改动不影响前面的表设计只是把dispatch方法换成MQ生产者的写法。4.3 调用Python服务的HTTP客户端与超时设置Java调用Python服务最直接的是用RestTemplate关键在超时设置。我见过太多人放着默认超时不管结果一次模型推理慢一点前端直接报网络错误。推荐自定义RestTemplateBean public RestTemplate evalRestTemplate() { SimpleClientHttpRequestFactory factory new SimpleClientHttpRequestFactory(); factory.setConnectTimeout(3000); // 连接超时3秒Python服务没起来会快速失败 factory.setReadTimeout(30000); // 读取超时30秒同步评测的最长等待 RestTemplate restTemplate new RestTemplate(factory); return restTemplate; }调用代码public String callPythonEval(EvalTask task) { String url http://python-eval:8000/eval/run; JSONObject req new JSONObject(); req.put(taskNo, task.getTaskNo()); req.put(modelPath, modelService.getById(task.getModelId()).getModelPath()); req.put(datasetPath, datasetService.getById(task.getDatasetId()).getDatasetPath()); req.put(batchSize, task.getConfigJson().getIntValue(batchSize)); req.put(device, task.getConfigJson().getString(device)); ResponseEntityString resp evalRestTemplate().postForEntity(url, req.toJSONString(), String.class); if (resp.getStatusCode().is2xxSuccessful()) { return resp.getBody(); } throw new BizException(Python评测服务调用失败); }注意这个同步调用只适合短任务30秒是上限。如果评测平均要2分钟你设30秒必然超时。这时候必须改成异步Java收到创建任务请求后立即返回后台线程把任务发给PythonPython跑完再回调/api/eval/callback更新状态。这样readTimeout可以放宽到60秒但更稳的做法还是让Python立刻返回受理成功再异步回传结果。5. Python评估服务实现细节与避坑指标都算出来了结果还是不放心5.1 用FastAPI封装评测接口路由与请求体Python侧我选FastAPI它是目前评测服务里最省心的选择自带OpenAPI文档pydantic做参数校验async支持也很干净。一个最简评测接口如下from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel from typing import Optional from metrics.classification import compute_classification_metrics app FastAPI(titlemodel-eval-service) class EvalRequest(BaseModel): taskNo: str modelPath: str datasetPath: str batchSize: int 32 device: str cpu # 支持传入需要计算的指标列表默认只算基础指标 metricNames: Optional[list] [accuracy, precision, recall, f1] app.post(/eval/run) def run_eval(req: EvalRequest, background_tasks: BackgroundTasks): # 立刻返回实际评测放到后台任务 background_tasks.add_task(do_eval, req) return {code: 0, message: accepted, taskNo: req.taskNo} def do_eval(req: EvalRequest): results compute_classification_metrics(req) # 计算结果后回调Java后端 callback_to_java(req.taskNo, results)这里有一个参数细节metricNames不要硬编码所有指标有的任务只需要跑精确率有的需要AUC。把指标列表暴露在请求里Python侧用字典做指标名到函数的映射可以避免写一堆if-else。do_eval里如果评测时间很长我会把它改成发送到消息队列让独立worker消费而不是用FastAPI的BackgroundTasks因为后台任务在服务重启时会丢。5.2 指标计算与多进程资源共享的几个参数分类任务最常见的指标就是accuracy、precision、recall、f1。sklearn直接提供全套函数但要注意average参数。宏平均适合各类别样本量不均的情况微平均则受大类别主导不加分辨就乱用评测结论容易被算法同学质疑。计算函数如下import numpy as np from sklearn.metrics import accuracy_score, precision_recall_fscore_support def compute_classification_metrics(req: EvalRequest): # 假设你已经用torch或其他框架加载了模型并推理得到preds import torch model load_model(req.modelPath, req.device) loader load_dataset(req.datasetPath, batch_sizereq.batchSize) all_preds [] all_labels [] with torch.no_grad(): for batch in loader: logits model(batch[inputs].to(req.device)) preds np.argmax(logits.cpu().numpy(), axis1) all_preds.extend(preds.tolist()) all_labels.extend(batch[labels].tolist()) all_preds np.array(all_preds) all_labels np.array(all_labels) acc accuracy_score(all_labels, all_preds) precision, recall, f1, _ precision_recall_fscore_support( all_labels, all_preds, averagemacro, zero_division0) return { taskNo: req.taskNo, accuracy: float(acc), precision: float(precision), recall: float(recall), f1: float(f1), sampleCount: int(len(all_labels)), }多进程评测时不要在每个进程里都加载一遍模型那会直接把内存打爆。常见做法是用multiprocessing的pool初始化函数只加载一次各进程共享只读的模型权重和数据集索引。参数上batchSize控制在模型可推理的最优数值一般二分类任务用32或64图像任务如果单张图片很大就降到8或16。device参数在CPU和GPU之间切换GPU评测时千万不要把所有任务都扔到cuda:0不然显存溢出会连累其他服务。5.3 评测服务避坑四个常见翻车点与排查思路第一指标不一样Java侧和Python侧误差0.001。现象同一个模型同一份数据接口返回的准确率和入库后查询的准确率差了千分之一。原因Python推理时用的是float32argmax后类别都一样但sklearn在计算F1时内部做了浮点累加不同库版本会引入微小误差。解决统一用logits而非概率做预测指标计算全部用float64并且固定numpy和sklearn版本。这不是一次性的升级依赖后必须回归一遍指标。第二中文路径下的图片读不到。现象数据集放在评测数据/分类/车辆目录下Python报FileNotFoundError。原因Windows默认编码不是UTF-8FastAPI接收JSON没问题但文件系统访问用到了本地字符集。解决代码里统一用pathlib和utf-8字符串路径全部改为完整绝对路径尽量避免中文但至少要能兼容。第三同步HTTP评测超时。现象Java侧报ReadTimeoutExceptionPython侧日志显示任务还在跑。原因readTimeout设了30秒但模型推理时间超过30秒。解决把同步调改成后台任务回调或者走消息队列。这个改动的痛苦程度至少值两天工时所以早做比晚做好。第四Python依赖一团乱换了台机器就跑不起来。现象本机能跑部署到服务器报ImportError两个机器numpy版本不同导致ndarray的方法签名都不一样。原因没有用requirements和虚拟环境隔离。解决固定requirements.txt并且把torch、numpy这类重依赖用Docker镜像固化下来。我一般会在deploy目录下放一个Python服务的Dockerfile这样算法同事换机器只接触代码包不接触环境。6. 验证方法与进阶先跑通最小闭环再谈分布式6.1 单机验证最简流程启动顺序别搞乱。先启动MySQL初始化表结构再启动Java后端确认Swagger能看到接口最后启动Python服务。用curl直接调Java创建任务接口curl -X POST http://localhost:8080/api/eval/task \ -H Content-Type: application/json \ -d {modelId:1,datasetId:1,config:{batchSize:16,device:cpu,metricNames:[accuracy,f1]}}如果返回taskNo说明Java侧通过等几秒再查任务状态看到status2就说明Python侧和回调都通了。这一步能跑通整个后端骨架就是成立的剩下的都是填功能。6.2 从单机到分布式的演进要点先别一上来就上K8s。单机验证通过后把Python worker单独部署到一台机器Java通过MQ发消息这样Java重启不会中断正在跑的评测。然后给监控系统的关键指标加三个任务状态流转成功率保证没有卡死的运行中任务、Python服务响应P99、消息队列积压数量。这三个指标能暴露九成问题。Java侧再做一次API平滑重启让存量任务等两分钟再关闭线程池这是最便宜的高可用改造。6.3 一个我踩过的坑评测结果的可复现性最后一次提醒也是我血泪经验评测任务必须固定随机种子。之前我负责的平台上同一个模型连续跑两次F1差了0.005算法团队一口咬定平台有Bug。排查到最后才发现是数据加载时的shuffle随机性以及多进程并行时对样本顺序的依赖。从那以后任务配置里强制带上seed字段并行评测worker数固定数据集的每个样本还额外存了索引最后结果表里记录seed。现在再有人质疑结果直接把任务号和seed丢给他让他自己复现。希望这篇文章的架构和避坑细节能帮到你少走这些弯路。本文还有配套的精品资源点击获取