1. 项目概述从“空白焦虑”到“优雅等待”做AI Skill技能开发的朋友估计都遇到过这个让人头疼的场景用户满怀期待地触发了一个需要复杂推理、调用大模型或者处理大量数据的技能然后呢屏幕就卡住了转圈圈或者干脆一片空白。用户心里开始打鼓“是卡死了吗”“我网不好”“这玩意儿是不是坏了”几秒之后耐心耗尽直接关掉或刷新。一次糟糕的体验可能就让用户对这个技能乃至背后的服务失去信心。这就是典型的“长耗时执行”带来的用户体验灾难。我们开发的AI Skill本质上是一个响应式服务它接收用户输入语音、文本、点击经过后端处理再返回结果。当后端处理逻辑复杂比如需要调用一个生成数百字文案的LLM大语言模型、执行一个多步的数据分析流程或者等待一个外部API的慢速响应时这个“处理时间”很容易就超过了几秒钟。在Web或移动端交互中超过2-3秒的等待而没有反馈就足以引发用户的“空白焦虑”。所以这个项目的核心目标非常明确消灭用户面前的“空白等待”将长耗时操作从阻塞式的“同步等待”转变为流畅的“异步交互”。这不是简单地加个Loading动画虽然那也有用而是一套从前端到后端涵盖状态管理、超时控制、错误重试和结果推送的完整技术方案。它关乎的不仅是技术实现更是对用户心理的把握和产品体验的打磨。无论你是用Java Spring Boot、Python FastAPI还是Node.js来构建Skill后端这套设计思路都是相通的。2. 核心设计思路异步化与状态驱动面对长耗时任务最朴素的想法是“让请求等在那里直到完成”。这就是同步Synchronous处理。在Skill的HTTP接口中表现为一个请求线程被占用直到所有处理逻辑结束才返回响应。这种做法有几个致命伤连接超时HTTP服务器、负载均衡器或客户端通常都有连接超时设置如30秒、60秒。任务一旦超过这个时间连接会被强行断开用户收到一个冷冰冰的“超时错误”。资源耗尽每个等待的请求都占用一个服务器线程/连接。大量用户同时触发长任务会导致线程池迅速耗尽新的请求无法被处理服务瘫痪。用户体验割裂用户无法得知进度只能干等且无法进行其他操作。因此我们的设计必须转向异步Asynchronous。核心思路是“快速响应后台处理主动通知”。2.1 方案选型任务队列与WebSocket/SSE实现异步主流有两种技术路径路径一任务队列 轮询这是最经典、最稳健的方案。用户触发请求后后端立即创建一个任务Job将其放入消息队列如RabbitMQ、Redis Streams、Kafka并返回一个唯一的task_id。后端有独立的Worker进程消费队列中的任务并执行。前端则定期例如每秒带着这个task_id向一个查询接口发起请求询问任务状态“处理中”、“成功”、“失败”和结果。优点解耦彻底可扩展性强能应对极高的并发和非常耗时的任务。任务状态持久化即使服务重启也不会丢失。缺点前端需要实现轮询逻辑实时性有延迟取决于轮询间隔且会产生大量“无意义”的查询请求。路径二任务队列 WebSocket / Server-Sent Events这是更现代的“主动推送”方案。创建任务后后端同样返回task_id。但同时前端与后端建立一个长连接WebSocket或订阅一个事件流SSE。当后台Worker完成任务后通过消息中间件或直接调用将结果推送到对应的连接上前端实时接收并展示。优点实时性极高用户体验流畅网络开销小无需轮询。缺点实现复杂度较高需要维护长连接状态在服务器重启或网络波动时连接会中断需要额外的重连和状态恢复机制。如何选择对于大多数AI Skill场景我推荐采用“混合方案”核心流程使用“任务队列轮询”因为它足够简单、可靠、兼容性好无论是Web、App还是小程序都能轻松实现。在需要极强实时性的子场景如逐字输出的AI生成中补充WebSocket/SSE。例如LLM生成文本时可以先用轮询查询任务是否开始一旦开始则建立WebSocket连接接收流式输出的token。本项目将主要围绕“任务队列Redis作为Broker 轮询状态查询”这一稳健路径展开并会详细探讨如何优化轮询体验以及如何无缝集成推送能力作为升级选项。2.2 状态机设计清晰定义任务生命周期异步任务的核心是状态管理。我们必须明确定义一个任务从生到死的所有状态这通常用一个状态机来实现PENDING等待中 - PROCESSING处理中 - SUCCESS成功 / FAILED失败 | | - CANCELLED已取消PENDING任务已创建并进入队列等待Worker领取。这是接口立即返回时的状态。PROCESSINGWorker已领取任务正在执行。前端收到此状态时可以显示具体的进度条如果有进度信息。SUCCESS任务执行成功结果数据可用。FAILED任务执行失败。关键点必须提供失败的error_code和error_message方便前端展示友好错误也便于排查。CANCELLED任务被用户或系统主动取消。在数据库中我们需要一张async_tasks表来持久化这些状态字段至少包括task_id,user_id,skill_type,input_params,status,progress,result_data,error_info,created_at,updated_at。实操心得progress字段0-100的整数非常有用。对于可预测进度的任务如处理100个文件Worker可以定期更新进度。前端根据进度更新UI能极大缓解用户的等待焦虑。即使进度是估算的也比没有强。3. 后端实现详解从创建到回调我们以Python FastAPI Celery使用Redis作为Broker为例拆解后端实现。这套组合在Python生态中非常成熟。如果你用的是Java可以用Spring Boot Async / RabbitMQ如果是Node.js可以用Bull库。3.1 任务创建与快速响应当用户请求到达Skill的入口接口如/api/skill/ask时我们不做耗时操作只做三件事参数校验。生成唯一task_id并创建任务记录状态为PENDING。将任务信息发送到Celery消息队列。立即返回task_id和状态。# app/api/endpoints/skill.py from celery import current_app from fastapi import APIRouter, HTTPException from pydantic import BaseModel import uuid router APIRouter() class SkillRequest(BaseModel): query: str # 其他参数... router.post(/ask) async def create_skill_task(request: SkillRequest): # 1. 参数校验 (略) # 2. 生成任务ID和记录 task_id str(uuid.uuid4()) task_record { task_id: task_id, user_id: get_current_user_id(), # 从token获取 skill_type: complex_ai_analysis, input_params: request.dict(), status: PENDING, progress: 0, created_at: datetime.utcnow() } # 存入数据库 (伪代码) await db.tasks.insert_one(task_record) # 3. 发起异步任务 # process_skill_task 是Celery task的函数名 current_app.send_task(worker.process_skill_task, args[task_id], kwargsrequest.dict()) # 4. 立即返回 return { code: 200, msg: 任务已提交正在处理中, data: { task_id: task_id, status: PENDING, check_url: f/api/tasks/{task_id}/status # 告知前端查询地址 } }注意事项生成task_id务必使用全局唯一的算法如UUID。不要用简单的自增ID以防被猜测和遍历。返回的响应中最好直接给出状态查询的URL这对前端开发者非常友好。3.2 Worker设计与任务执行Worker是执行耗时任务的苦力。一个健壮的Worker需要处理好以下几点1. 任务函数与状态更新# app/worker/tasks.py from celery import Celery from app.core.celery_app import celery_app from app.db.mongodb import get_db # 假设用MongoDB import time from your_ai_service import heavy_ai_processing # 你的核心AI处理函数 celery_app.task(bindTrue, nameworker.process_skill_task) def process_skill_task(self, task_id, **kwargs): db get_db() # 1. 更新状态为 PROCESSING db.tasks.update_one({task_id: task_id}, {$set: {status: PROCESSING, updated_at: datetime.utcnow()}}) try: # 2. 执行核心AI处理逻辑 # 这里模拟一个长耗时操作 result heavy_ai_processing(kwargs[query]) # 假设heavy_ai_processing可以接受一个回调来更新进度 # for i in range(100): # do_partial_work() # db.tasks.update_one({task_id: task_id}, {$set: {progress: i1}}) # 3. 成功更新状态和结果 db.tasks.update_one( {task_id: task_id}, {$set: { status: SUCCESS, progress: 100, result_data: result, updated_at: datetime.utcnow() }} ) return result except Exception as e: # 4. 失败更新状态和错误信息 db.tasks.update_one( {task_id: task_id}, {$set: { status: FAILED, error_info: {code: INTERNAL_ERROR, message: str(e)}, updated_at: datetime.utcnow() }} ) # 重要重新抛出异常让Celery知道任务失败了可用于重试 raise self.retry(exce, countdown60) if not_max_retries else None2. 超时与重试机制这是稳定性的关键。在Celery中可以直接在task装饰器中配置。celery_app.task(bindTrue, name“worker.process_skill_task”, max_retries3, soft_time_limit300, time_limit330) def process_skill_task(self, task_id, **kwargs): # ...max_retries3任务失败后自动重试最多3次。soft_time_limit300任务执行300秒5分钟后会收到一个SoftTimeLimitExceeded异常可以捕获并做清理工作然后优雅退出。time_limit330硬超时超过330秒5.5分钟Worker进程会被强制终止。在except块中调用self.retry(countdown60)可以实现指数退避重试比如60秒后重试。避坑技巧soft_time_limit应略小于time_limit给任务一个自我清理的机会。重试时countdown参数可以设置成递增的如60, 120, 300避免失败后立即重试给下游服务造成压力。3. 进度更新如果任务可以分阶段在关键阶段更新数据库中的progress字段。前端轮询时能拿到这个进度并更新UI。3.3 状态查询接口这个接口要简单、快速因为它会被高频轮询。router.get(/tasks/{task_id}/status) async def get_task_status(task_id: str): task await db.tasks.find_one({task_id: task_id}) if not task: raise HTTPException(status_code404, detail任务不存在) # 返回必要信息注意过滤敏感数据 return { task_id: task_id, status: task[status], progress: task.get(progress, 0), result: task.get(result_data) if task[status] SUCCESS else None, error: task.get(error_info) if task[status] FAILED else None }性能优化这个接口查询必须走数据库索引对task_id建立唯一索引。可以考虑给任务记录加一个短期缓存如Redis过期时间5分钟进一步降低数据库压力。4. 前端交互优化智能轮询与用户体验后端准备好了前端的任务是把这种异步机制以最友好的方式呈现给用户。4.1 基础轮询实现前端在收到task_id后启动一个定时器周期性调用状态查询接口。// 前端示例 (Vue3 Composition API) import { ref } from vue; const taskStatus ref(null); const pollInterval ref(null); const startPolling (taskId) { stopPolling(); // 先停止之前的轮询 pollInterval.value setInterval(async () { try { const resp await fetch(/api/tasks/${taskId}/status); const data await resp.json(); taskStatus.value data; // 根据状态决定是否停止轮询 if (data.status SUCCESS || data.status FAILED || data.status CANCELLED) { stopPolling(); if (data.status SUCCESS) { // 处理成功结果 showResult(data.result); } else { // 处理失败或取消 showError(data.error); } } else if (data.status PROCESSING) { // 更新进度条 updateProgress(data.progress); } } catch (error) { console.error(轮询请求失败:, error); // 网络错误处理可以考虑重试几次再报错 } }, 1000); // 1秒轮询一次 }; const stopPolling () { if (pollInterval.value) { clearInterval(pollInterval.value); pollInterval.value null; } };4.2 高级优化策略基础轮询可行但不够优雅。我们可以做得更好1. 指数退避轮询一开始频繁轮询比如每秒一次如果任务长时间处于PENDING或PROCESSING可以逐渐拉长轮询间隔2秒4秒8秒…直到达到一个上限如30秒。这能有效减少对服务器的压力。let pollDelay 1000; // 初始1秒 const maxDelay 30000; // 最大30秒 const pollWithBackoff (taskId) { const poll async () { // ... 执行查询逻辑 ... if (data.status ‘PENDING’ || data.status ‘PROCESSING’) { // 任务未完成下次延迟增加 pollDelay Math.min(pollDelay * 1.5, maxDelay); setTimeout(poll, pollDelay); } else { // 任务完成停止 } }; setTimeout(poll, pollDelay); };2. 长轮询这是对普通轮询的改进。前端发起一个查询请求后端如果任务未完成不是立即返回而是“挂起”这个请求保持连接直到任务状态发生变化或超时比如30秒才返回。这样减少了请求次数实时性也更好。实现上后端可以用类似asyncio.sleep循环检查数据库或者利用数据库的监听功能如Redis的Pub/Sub。3. 服务器发送事件SSE是一种轻量级的、由服务器向客户端推送数据的技术。对于任务状态变更不频繁的场景很合适。前端建立连接后后端可以在任务状态更新时通过这个连接推送事件。// 前端 const eventSource new EventSource(/api/tasks/${taskId}/stream); eventSource.onmessage (event) { const data JSON.parse(event.data); // 处理状态更新... }; eventSource.onerror (error) { // 处理错误尝试重连 };4. WebSocket实时推送这是实时性最高的方案适合需要流式输出结果的AI Skill如AI对话逐字打出。前端建立WebSocket连接后端Worker在生成每个片段时都通过WebSocket推送到前端。实现复杂度最高需要管理连接池和会话状态。实操心得不要一上来就用最复杂的WebSocket。建议从“指数退避轮询”开始它能解决80%的问题实现简单兼容性无敌。当你的Skill确实需要“逐字输出”这种强实时反馈时再考虑引入SSE或WebSocket作为增强功能。可以设计成任务创建后先用轮询当查询到状态进入PROCESSING且任务类型支持流式输出时前端再主动建立WebSocket连接来接收流。4.3 UI/UX设计要点技术实现了UI也要跟上明确的状态提示用清晰的文案和图标告诉用户当前状态。“排队中”、“正在分析您的需求30%”、“即将完成”、“成功”、“抱歉处理失败原因是...”。进度指示器如果有进度一定要展示进度条。没有精确进度可以用无限循环的动画或“正在努力处理中...”这样的动态文案。取消操作提供一个明显的“取消”按钮。点击后前端调用取消接口后端将任务状态标记为CANCELLED并尝试中断Worker中的任务这需要Worker支持任务中断比如检查一个共享的取消标志位。结果预览与交互任务成功后不要只是弹个提示。应该把结果数据优雅地呈现在界面上。如果是文本可以高亮关键信息如果是文件提供下载按钮。5. 进阶考量与生产环境加固一个能上生产环境的异步处理方案还需要考虑更多边边角角的问题。5.1 任务去重与幂等性防止用户因网络问题重复点击导致创建多个相同任务。可以在创建任务前根据“用户ID技能类型输入参数哈希”生成一个唯一键先检查是否存在近期创建的、相同且未完成的PENDING或PROCESSING任务如果存在则直接返回已有的task_id。所有操作创建、查询、取消都要保证幂等性。即同一请求重复执行多次产生的结果与执行一次相同。例如基于task_id取消任务即使调用多次最终结果都是取消。5.2 任务结果清理与存储任务结果尤其是AI生成的内容、文件可能很大不能一直存在数据库里。需要设计清理策略短期缓存任务完成后将结果数据存入对象存储如S3、OSS或缓存Redis数据库中只存一个引用地址。设置一个较短的过期时间如7天。异步清理Job定期运行一个后台Job清理超过保留期限的SUCCESS/FAILED任务记录和对应的存储结果。5.3 监控与告警异步系统黑盒多监控至关重要。队列堆积监控监控Celery/RabbitMQ/Redis中等待处理的任务数量。如果堆积持续增长说明Worker处理能力不足或任务异常。任务耗时分布记录每个任务的开始时间、结束时间统计P50 P95 P99耗时。耗时异常增长可能意味着下游API变慢或内部逻辑有问题。失败率监控监控任务FAILED状态的比例。失败率突增需要立即告警。Worker健康度监控Worker进程是否存活是否因为内存泄漏等原因频繁重启。5.4 与现有架构的集成如果你的Skill后端已经有一套复杂的认证、限流、日志链路需要将异步任务系统集成进去。认证创建任务和查询任务状态的接口都需要验证用户身份。task_id最好与user_id关联查询时校验防止用户查询他人的任务。限流任务创建接口需要限流防止用户恶意刷任务挤爆队列。可以根据用户等级设置不同的并发任务数限制。链路追踪为每个任务生成一个唯一的追踪ID如trace_id并贯穿从HTTP请求到Worker处理的整个链路。这样在排查问题时可以轻松串联所有日志。6. 常见问题排查与实战技巧在实际开发和运维中你会遇到各种各样的问题。这里记录一些典型的坑和解决方法。问题1任务状态一直是PENDING永不执行。排查检查Worker是否启动celery -A app worker -l info命令是否成功执行有无报错。检查消息队列连接Celery配置的Broker URL如Redis地址是否正确网络是否通畅。可以登录Redis用KEYS celery*查看是否有任务消息。检查任务路由Celery的Task名称worker.process_skill_task是否与Worker注册的名称完全一致包括模块路径。技巧在开发环境可以在任务函数开始处打一行日志这是最直接的确认方式。问题2前端轮询时偶尔收到HTTP 404任务不存在。原因可能发生在任务刚创建记录还未写入数据库前端就立刻发起查询的极端情况下。解决状态查询接口对“任务不存在”的情况不要直接返回404错误可以返回一个特定的状态码和数据如{status: NOT_FOUND}让前端友好提示“任务正在创建中请稍后再试”并继续轮询。问题3任务执行一半失败但状态没有更新为FAILED。原因Worker进程崩溃如内存溢出被系统杀死或者任务代码中有未捕获的异常。解决确保Worker任务函数有最外层的try...except并在except中更新数据库状态。为Celery Worker配置进程监控如Supervisor崩溃后自动重启。利用Celery的acks_late设置和死信队列。设置acks_lateTrue可以让任务在执行完毕后才确认消息如果Worker崩溃消息会重新投递给其他Worker。多次重试失败的任务可以进入死信队列方便后续人工排查。问题4用户等待时间过长即使有进度提示也不耐烦。优化预估时间如果可能根据历史数据或任务复杂度给一个粗略的预估完成时间如“大约需要2分钟”。分阶段反馈将一个大任务拆分成多个逻辑阶段每个阶段完成时更新一次状态和文案如“阶段1/3数据准备完成”、“阶段2/3AI模型推理中...”。这比单纯的百分比进度条更有信息量。提供“后台运行”选项对于耗时极长的任务如视频生成可以允许用户关闭当前页面通过通知中心或消息列表在完成后通知用户。问题5如何调试一个正在运行或失败的长任务技巧在任务记录中增加一个debug_log字段类型为数组。在Worker执行任务的关键步骤将日志信息append到这个字段中。这样当任务失败或状态异常时你可以直接查询数据库看到这个任务生命周期的“黑匣子”记录而无需去翻看可能已经被轮转覆盖的服务器日志文件。这套“AI Skill长耗时执行的优雅处理方案”其核心思想是将时间不确定性从用户交互的链路上剥离出去。通过异步化、状态管理和友好的前端交互我们把一个可能失败、可能很慢的“黑盒”过程变成了一个可控、可感知、可中断的“白盒”流程。技术实现上并没有银弹需要根据你的业务规模、团队技术栈和用户体验要求在简单轮询和复杂推送之间找到平衡点。从我个人的经验来看先实现一个带指数退避轮询的稳健基础版本再根据实际需求逐步叠加更实时的特性是一条稳妥且高效的路径。记住目标是让用户感觉一切尽在掌握而不是在未知中空等。