当前位置：首页 > news >正文

AI模型多阶段调用进度追踪系统设计文档

news 来源：原创 2025/9/19 9:30:58

AI模型多阶段调用进度追踪系统设计文档

一、系统概述

为解决AI模型处理大型文件时响应时间长的问题，我们设计并实现了一套异步进度追踪系统。该系统采用Server-Sent Events (SSE) 技术，建立从服务器到客户端的单向实时通信通道，使前端能够实时获取后端文件处理的进度信息。

二、核心功能

异步文件处理：通过异步方式处理上传的文件，避免阻塞HTTP请求线程
实时进度推送：使用SSE技术向前端实时推送处理进度
多阶段处理跟踪：精确追踪每个AI处理阶段的进度
错误处理与恢复：完善的异常处理机制，确保资源得到释放

三、系统架构

系统架构由以下几个关键组件组成：

进度追踪服务：核心服务，管理任务进度和SSE连接
进度追踪控制器：提供REST API，供前端获取进度信息
文件处理服务增强：在现有服务上增加进度报告功能
文件上传控制器适配：修改现有控制器适配异步处理

四、后端实现代码

1. 进度追踪服务接口(IProgressTrackingService.java)

import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;/*** 进度追踪服务接口* 用于追踪长时间运行任务的进度并推送给前端*/
public interface IProgressTrackingService {/*** 创建新的进度追踪会话** @param taskId 任务ID* @return SseEmitter 事件发射器*/SseEmitter createProgressTracker(String taskId);/*** 更新任务进度** @param taskId 任务ID* @param currentStage 当前阶段索引（从0开始）* @param totalStages 总阶段数* @param message 进度消息*/void updateProgress(String taskId, int currentStage, int totalStages, String message);/*** 完成任务进度追踪** @param taskId 任务ID* @param success 是否成功* @param message 完成消息*/void completeProgress(String taskId, boolean success, String message);
}

2. 进度追踪服务实现(ProgressTrackingServiceImpl.java)

import com.fasterxml.jackson.databind.ObjectMapper;
import com.greatech.abnormal_monitoring.service.IProgressTrackingService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;/*** 进度追踪服务实现类* 用于追踪长时间运行任务的进度并推送给前端*/
@Slf4j
@Service
public class ProgressTrackingServiceImpl implements IProgressTrackingService {// 存储任务ID到SSE发射器的映射private final Map<String, SseEmitter> emitterMap = new ConcurrentHashMap<>();// JSON对象映射器private final ObjectMapper objectMapper = new ObjectMapper();/*** 创建新的进度追踪会话** @param taskId 任务ID* @return SseEmitter 事件发射器*/@Overridepublic SseEmitter createProgressTracker(String taskId) {// 创建一个超时时间为30分钟的SSE发射器SseEmitter emitter = new SseEmitter(1800000L);// 设置完成回调emitter.onCompletion(() -> {log.info("进度追踪会话已完成: {}", taskId);emitterMap.remove(taskId);});// 设置超时回调emitter.onTimeout(() -> {log.warn("进度追踪会话超时: {}", taskId);emitterMap.remove(taskId);});// 设置错误回调emitter.onError(ex -> {log.error("进度追踪会话发生错误: {}", taskId, ex);emitterMap.remove(taskId);});// 将发射器存储到映射中emitterMap.put(taskId, emitter);try {// 发送初始化事件Map<String, Object> initialEvent = Map.of("type", "INIT","taskId", taskId,"message", "进度追踪会话已创建","progress", 0);emitter.send(SseEmitter.event().data(objectMapper.writeValueAsString(initialEvent)));} catch (IOException e) {log.error("发送初始化事件失败: {}", taskId, e);emitter.completeWithError(e);}return emitter;}/*** 更新任务进度** @param taskId 任务ID* @param currentStage 当前阶段索引（从0开始）* @param totalStages 总阶段数* @param message 进度消息*/@Overridepublic void updateProgress(String taskId, int currentStage, int totalStages, String message) {SseEmitter emitter = emitterMap.get(taskId);if (emitter == null) {log.warn("尝试更新不存在的进度追踪会话: {}", taskId);return;}try {// 计算进度百分比int progress = (int) Math.round((currentStage * 100.0) / totalStages);// 创建进度事件Map<String, Object> progressEvent = Map.of("type", "PROGRESS","taskId", taskId,"currentStage", currentStage,"totalStages", totalStages,"progress", progress,"message", message);// 发送进度事件emitter.send(SseEmitter.event().data(objectMapper.writeValueAsString(progressEvent)));log.debug("已更新进度: taskId={}, stage={}/{}, progress={}%, message={}", taskId, currentStage, totalStages, progress, message);} catch (IOException e) {log.error("发送进度更新失败: {}", taskId, e);emitter.completeWithError(e);}}/*** 完成任务进度追踪** @param taskId 任务ID* @param success 是否成功* @param message 完成消息*/@Overridepublic void completeProgress(String taskId, boolean success, String message) {SseEmitter emitter = emitterMap.get(taskId);if (emitter == null) {log.warn("尝试完成不存在的进度追踪会话: {}", taskId);return;}try {// 创建完成事件Map<String, Object> completeEvent = Map.of("type", success ? "COMPLETE" : "ERROR","taskId", taskId,"progress", success ? 100 : -1,"message", message,"success", success);// 发送完成事件emitter.send(SseEmitter.event().data(objectMapper.writeValueAsString(completeEvent)));log.info("进度追踪已完成: taskId={}, success={}, message={}", taskId, success, message);// 关闭SSE连接emitter.complete();} catch (IOException e) {log.error("发送完成事件失败: {}", taskId, e);emitter.completeWithError(e);} finally {// 确保从映射中移除emitterMap.remove(taskId);}}
}

3. 进度追踪控制器(ProgressTrackingController.java)

import com.greatech.abnormal_monitoring.service.IProgressTrackingService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.util.UUID;/*** 进度追踪控制器* 提供创建和获取进度追踪会话的接口*/
@Slf4j
@RestController
@RequestMapping("/api/progress")
@RequiredArgsConstructor
@Tag(name = "进度追踪", description = "提供长时间运行任务的进度追踪功能")
public class ProgressTrackingController {private final IProgressTrackingService progressTrackingService;/*** 创建新的进度追踪会话** @return SseEmitter 事件发射器*/@Operation(summary = "创建进度追踪会话", description = "创建一个新的进度追踪会话并返回SSE连接")@GetMapping(value = "/track", produces = MediaType.TEXT_EVENT_STREAM_VALUE)public SseEmitter createProgressTracker(@Parameter(description = "任务ID（可选）") @RequestParam(required = false) String taskId) {// 如果未提供任务ID，则生成一个新IDif (taskId == null || taskId.isEmpty()) {taskId = UUID.randomUUID().toString();}log.info("创建进度追踪会话: {}", taskId);return progressTrackingService.createProgressTracker(taskId);}/*** 获取特定任务的进度追踪会话** @param taskId 任务ID* @return SseEmitter 事件发射器*/@Operation(summary = "获取任务进度", description = "根据任务ID获取进度追踪会话并返回SSE连接")@GetMapping(value = "/track/{taskId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)public SseEmitter getProgressTracker(@Parameter(description = "任务ID", required = true) @PathVariable String taskId) {log.info("获取进度追踪会话: {}", taskId);return progressTrackingService.createProgressTracker(taskId);}
}

4. 修改后的文件上传控制器方法

/*** 上传模板文件并解析数据** @param file       模板文件* @param templateId 模板ID* @return 上传结果*/
@Operation(summary = "上传模板文件并解析", description = "上传模板文件到服务器并解析其内容到数据库")
@PostMapping("/upload-template")
public ServerResponse<Map<String, Object>> uploadAndParseTemplate(@Parameter(description = "模板文件", required = true) @RequestParam("file") MultipartFile file,@RequestParam("templateId") Long templateId) {log.info("开始处理模板文件上传请求，模板ID: {}, 文件名: {}", templateId, file.getOriginalFilename());try {// 处理文件并获取结果（现在包含任务ID）String result = templateService.processTemplateFile(file, templateId);// 从结果中提取任务IDString taskId = "";if (result.contains("任务ID:")) {taskId = result.substring(result.indexOf("任务ID:") + 5).trim();} else {// 对于不支持进度追踪的文件处理（如Excel文件），直接返回处理结果log.info("模板文件处理成功，模板ID: {}, 文件名: {}", templateId, file.getOriginalFilename());Map<String, Object> directResult = new HashMap<>();directResult.put("message", result);directResult.put("completed", true);return ServerResponse.success(directResult);}log.info("模板文件上传成功并开始异步处理，模板ID: {}, 文件名: {}, 任务ID: {}", templateId, file.getOriginalFilename(), taskId);// 构建响应数据Map<String, Object> responseData = new HashMap<>();responseData.put("taskId", taskId);responseData.put("message", "文件上传成功，正在异步处理中");responseData.put("completed", false);return ServerResponse.success(responseData);} catch (Exception e) {log.error("模板文件处理失败，模板ID: {}, 文件名: {}", templateId, file.getOriginalFilename(), e);return ServerResponse.error("文件处理失败: " + e.getMessage());}
}

5. 模板服务处理文本文件方法(TemplateServiceImpl.java中的processTextFile方法)

/*** 处理文本文件** @param file       已保存的文件* @param templateId 模板ID* @return 处理结果*/
public String processTextFile(File file, Long templateId) {log.info("处理文本文件：{}, 使用会话方式调用大模型API", file.getName());// 生成唯一任务ID用于进度追踪String taskId = UUID.randomUUID().toString();log.info("生成进度追踪任务ID: {}", taskId);// 创建新的会话IDString sessionId = aiCommonService.createNewSession();log.info("创建新会话：{}", sessionId);// 使用CompletableFuture异步处理CompletableFuture.runAsync(() -> {try {// 将文件转换为MultipartFile以便上传MultipartFile multipartFile = convertToMultipartFile(file);// 读取提示词模板并按阶段拆分String promptTemplateContent = readPromptTemplate();List<String> promptStages = splitContentIntoStages(promptTemplateContent);log.info("提示词模板已拆分为{}个阶段", promptStages.size());// 初始化进度progressTrackingService.updateProgress(taskId, 0, promptStages.size(), "开始处理文本文件，共" + promptStages.size() + "个阶段");String jsonResponse = null;// 顺序执行每个阶段，只保留最后一个阶段的结果for (int i = 0; i < promptStages.size(); i++) {String stagePrompt = promptStages.get(i);// 更新进度progressTrackingService.updateProgress(taskId, i, promptStages.size(), "正在执行阶段 " + (i+1) + "/" + promptStages.size());log.info("执行阶段 {}/{}", i+1, promptStages.size());// 对于第一个阶段，需要上传文件if (i == 0) {jsonResponse = aiCommonService.callAiModelWithFileContext(stagePrompt, sessionId, multipartFile);} else {// 后续阶段只需要发送提示词jsonResponse = aiCommonService.callAiModelWithFileContext(stagePrompt, sessionId, null);}// 仅记录当前处于哪个阶段，不记录完整响应以避免日志过大log.info("阶段 {}/{} 执行完成", i+1, promptStages.size());// 更新进度progressTrackingService.updateProgress(taskId, i+1, promptStages.size(), "已完成阶段 " + (i+1) + "/" + promptStages.size());}log.info("所有阶段执行完成，获取最终JSON响应");progressTrackingService.updateProgress(taskId, promptStages.size(), promptStages.size(), "所有阶段执行完成，正在处理最终结果");// 将JSON响应转换为Excel文件String excelFilePath = convertJsonToExcelFile(jsonResponse, templateId);// 使用Excel解析逻辑处理生成的Excel文件sheetService.parseExcelFile(new File(excelFilePath), templateId);// 清理会话资源aiCommonService.clearSession(sessionId);log.info("会话已清理：{}", sessionId);// 标记任务完成progressTrackingService.completeProgress(taskId, true, "文本文件已成功处理并解析，原始文件路径：" + file.getAbsolutePath());} catch (Exception e) {// 出错时也要尝试清理会话资源try {aiCommonService.clearSession(sessionId);log.info("会话清理完成：{}", sessionId);} catch (Exception ex) {log.error("清理会话失败：{}", sessionId, ex);}log.error("处理文本文件失败", e);// 标记任务失败progressTrackingService.completeProgress(taskId, false, "处理文件失败: " + e.getMessage());}});// 立即返回，不阻塞请求线程return "文件处理已开始，可通过任务ID: " + taskId + " 追踪进度";
}

6. 提取阶段的灵活实现(splitContentIntoStages方法)

/*** 将文件内容按阶段拆分，使用配置文件中的设置** @param content 文件内容* @return 拆分后的阶段列表*/
private List<String> splitContentIntoStages(String content) {List<String> stages = new ArrayList<>();// 1. 首先尝试使用配置的标识符列表String identifiersConfig = environment.getProperty("template.stage.identifiers");if (identifiersConfig != null && !identifiersConfig.isEmpty()) {return splitByConfiguredIdentifiers(content, identifiersConfig);}// 2. 如果没有配置具体标识符，则使用正则表达式return splitByRegexPattern(content);
}/*** 根据配置的标识符列表拆分内容*/
private List<String> splitByConfiguredIdentifiers(String content, String identifiersConfig) {List<String> stages = new ArrayList<>();String[] identifiers = identifiersConfig.split(",");log.info("使用配置的 {} 个阶段标识符拆分内容", identifiers.length);// 寻找每个阶段的起始和结束位置int currentPos = 0;for (int i = 0; i < identifiers.length; i++) {String identifier = identifiers[i].trim();int stageStart = content.indexOf(identifier, currentPos);if (stageStart != -1) {// 找到阶段起始位置int nextStageStart = -1;// 查找下一个阶段的起始位置（如果存在）if (i < identifiers.length - 1) {nextStageStart = content.indexOf(identifiers[i + 1].trim(), stageStart);}// 提取当前阶段内容String stageContent;if (nextStageStart != -1) {stageContent = content.substring(stageStart, nextStageStart).trim();currentPos = nextStageStart;} else {// 如果是最后一个阶段，则截取到文件末尾stageContent = content.substring(stageStart).trim();}log.info("提取阶段 {}: {} 字符", i+1, stageContent.length());stages.add(stageContent);} else {log.warn("未找到阶段标识符: {}", identifier);}}return stages;
}/*** 使用正则表达式模式拆分内容*/
private List<String> splitByRegexPattern(String content) {List<String> stages = new ArrayList<>();// 获取配置的正则表达式，如果未配置则使用默认值String patternStr = environment.getProperty("template.stage.pattern");if (patternStr == null || patternStr.isEmpty()) {patternStr = "(?:#{0,3}\\s*【?\\s*阶段\\s*(\\d+)\\s*】?)";}log.info("使用正则表达式模式拆分内容: {}", patternStr);// 使用正则表达式匹配阶段标识符Pattern stagePattern = Pattern.compile(patternStr, Pattern.CASE_INSENSITIVE);Matcher matcher = stagePattern.matcher(content);// 存储找到的所有阶段起始位置和编号Map<Integer, Integer> positionToStageNumber = new TreeMap<>();while (matcher.find()) {int stageNumber = Integer.parseInt(matcher.group(1));positionToStageNumber.put(matcher.start(), stageNumber);log.debug("发现阶段 {} 在位置 {}", stageNumber, matcher.start());}// 如果没有找到任何阶段标识，将整个内容作为一个阶段if (positionToStageNumber.isEmpty()) {log.warn("未在内容中找到阶段标识符，将整个内容作为单一阶段处理");stages.add(content);return stages;}// 是否严格按照阶段编号排序boolean strictOrder = Boolean.parseBoolean(environment.getProperty("template.stage.strict-order", "true"));// 获取最大阶段数限制int maxStageCount = Integer.parseInt(environment.getProperty("template.stage.max-count", "10"));// 提取每个阶段的内容List<Integer> positions = new ArrayList<>(positionToStageNumber.keySet());Collections.sort(positions);Map<Integer, String> numberToContent = new HashMap<>();for (int i = 0; i < positions.size(); i++) {int startPos = positions.get(i);int endPos = (i < positions.size() - 1) ? positions.get(i + 1) : content.length();String stageContent = content.substring(startPos, endPos).trim();int stageNumber = positionToStageNumber.get(startPos);if (stageNumber > maxStageCount) {log.warn("阶段编号 {} 超过最大限制 {}，将被忽略", stageNumber, maxStageCount);continue;}log.info("提取阶段 {}: {} 字符", stageNumber, stageContent.length());numberToContent.put(stageNumber, stageContent);}// 按照阶段编号排序if (strictOrder) {List<Integer> sortedNumbers = new ArrayList<>(numberToContent.keySet());Collections.sort(sortedNumbers);for (Integer num : sortedNumbers) {stages.add(numberToContent.get(num));}} else {// 按照在文件中的位置顺序for (int i = 0; i < positions.size(); i++) {int stageNumber = positionToStageNumber.get(positions.get(i));if (stageNumber <= maxStageCount) {stages.add(numberToContent.get(stageNumber));}}}log.info("共拆分出 {} 个阶段", stages.size());return stages;
}

五、前端调用说明

这里提供简要的前端调用说明，以便于前端开发人员理解如何与后端集成。

1. 前端调用流程

前端与后端的交互流程如下：

文件上传：使用标准的HTTP POST请求上传文件到/api/template/upload-template
获取任务ID：从上传响应中提取任务ID
建立SSE连接：使用EventSource接口建立与/api/progress/track/{taskId}的连接
处理进度更新：监听并处理SSE事件，更新UI显示进度
处理完成通知：接收处理完成或失败的通知，更新UI状态

2. 代码示例

以下是前端实现的简化示例：

// 文件上传示例
async function uploadFile(file, templateId) {const formData = new FormData();formData.append('file', file);formData.append('templateId', templateId);try {const response = await fetch('/api/template/upload-template', {method: 'POST',body: formData});const result = await response.json();if (result.code === 200) {if (result.data.taskId) {// 异步处理任务，开始跟踪进度trackProgress(result.data.taskId);} else if (result.data.completed) {// 同步处理已完成showSuccess(result.data.message);}} else {showError(result.message);}} catch (error) {showError('上传失败: ' + error.message);}
}// 进度追踪示例
function trackProgress(taskId) {// 创建EventSource连接const eventSource = new EventSource(`/api/progress/track/${taskId}`);// 处理消息事件eventSource.onmessage = (event) => {const data = JSON.parse(event.data);switch (data.type) {case 'INIT':updateProgressUI(0, data.message);break;case 'PROGRESS':updateProgressUI(data.progress, data.message);break;case 'COMPLETE':updateProgressUI(100, data.message);showSuccess(data.message);eventSource.close();break;case 'ERROR':showError(data.message);eventSource.close();break;}};// 处理错误eventSource.onerror = (error) => {showError('连接中断，无法获取进度更新');eventSource.close();};
}// 更新UI显示进度（由前端实现）
function updateProgressUI(progress, message) {// 更新进度条document.getElementById('progress-bar').value = progress;document.getElementById('progress-percentage').textContent = progress + '%';document.getElementById('progress-message').textContent = message;
}// 显示成功消息（由前端实现）
function showSuccess(message) {// 显示成功提示
}// 显示错误消息（由前端实现）
function showError(message) {// 显示错误提示
}

在Spring Boot应用程序配置中启用异步支持：

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {@Overridepublic Executor getAsyncExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(5);executor.setMaxPoolSize(10);executor.setQueueCapacity(25);executor.setThreadNamePrefix("async-");executor.initialize();return executor;}@Overridepublic AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {return new SimpleAsyncUncaughtExceptionHandler();}
}

确保在应用程序上下文中注册以下组件：

ProgressTrackingServiceImpl
ProgressTrackingController

六、安全性考虑

任务ID安全：生成的任务ID应使用安全的随机数生成器，避免被猜测
资源限制：设置SSE连接和异步处理的超时时间
错误处理：确保异常不会泄露敏感信息
防止资源泄露：正确清理未使用的连接和资源
访问控制：根据需要为进度追踪API添加身份验证

七、性能优化

异步线程池调优：根据服务器性能和预期负载调整线程池参数
超时控制：为长时间运行的任务设置合理的超时时间
内存管理：避免在SSE消息中发送大量数据
数据压缩：考虑对SSE消息进行压缩，减少网络流量

八、故障排除

常见问题及解决方案：

SSE连接中断

原因：网络问题、服务器重启、超时
解决：前端实现自动重连逻辑

任务处理缓慢

原因：AI模型处理速度、服务器资源不足
解决：增加服务器资源、优化AI调用参数

任务进度更新失败

原因：SSE连接已关闭、进度追踪服务异常
解决：检查日志、确保任务ID正确

资源泄露

原因：未正确关闭SSE连接或清理资源
解决：确保所有异常路径都调用资源清理方法

九、扩展方向

系统可以在以下方向进行扩展：

任务持久化：将任务进度存储到数据库，支持服务重启后恢复
任务队列：使用消息队列管理大量任务，避免服务器过载
分布式部署：支持集群环境中的任务进度追踪
管理界面：添加管理界面，监控和管理所有正在运行的任务
推送通知：集成WebSocket或推送通知，提供更实时的进度更新

十、总结

本系统提供了一种高效、可靠的方式来追踪AI模型多阶段调用的处理进度。通过使用Server-Sent Events技术和异步处理，系统能够在长时间运行的任务中向前端实时推送进度更新，大大提高了用户体验。该设计将文件处理与进度报告解耦，使系统更加灵活和可维护。

系统的核心优势在于：

实时进度反馈：通过SSE技术向前端实时推送进度信息
多阶段处理：支持AI模型的多阶段调用，精确追踪每个阶段的进度
灵活配置：支持通过配置文件自定义阶段标识符和处理行为
完善的错误处理：全面的异常处理和资源清理机制
易于集成：前端只需少量代码即可集成该功能

通过这一设计，我们解决了长时间运行任务缺乏进度反馈的问题，提供了更好的用户体验，同时提高了系统的可靠性和可维护性。