简介
本设计文档围绕 DouPHP 的 AI 异步任务表 dou_ai_task，系统阐述其表结构设计、任务生命周期、调度与轮询机制、结果缓存与过期策略、超时处理、重试机制以及监控告警方案。面向 AI 任务开发者，提供从提交任务到查询结果、从失败恢复到可观测性的完整实践指南。
项目结构
AI 异步任务相关代码主要分布在以下位置：
- 数据库升级脚本：定义 dou_ai_task 表结构与任务形态迁移
- 仓储层：统一读写 ai_task/chat_task 两表
- 编排层：负责提交、轮询、超时判定、结果落库
- 网关层：模型/密钥解析、驱动选择、调用上游
- 控制器与服务：后台入口、参数校验、用量记录、图片产物获取
- 配置：提示词、图像尺寸策略等
graph TB
A["后台控制器<br/>TaskController"] --> B["后台服务<br/>TaskService"]
B --> C["网关<br/>AiGateway"]
C --> D["驱动工厂/驱动<br/>DriverFactory/AsyncDriverInterface"]
C --> E["仓储<br/>AsyncTaskRepository"]
E --> F["数据库<br/>dou_ai_task / chat_task"]
C --> G["供应商API"]核心组件
- 表设计与索引：基于升级脚本定义，包含供应商ID、模型ID、密钥ID、应用ID、管理员ID、用户ID（预留）、任务类型、上游任务ID、状态、请求体、结果、错误信息、过期时间、创建/更新时间、最近轮询时间，并建立常用查询索引。
- 仓储层：封装任务行创建、提交回写、成功/失败/超时标记、结果更新、轮询时间刷新、分页列表等。
- 编排层：实现“提交→落库→调驱动submit→回写provider_task_id→轮询→提取结果→落库”的完整流程；支持同步直出补录为成功任务；惰性超时判定。
- 网关层：解析供应商/模型/密钥，选择驱动，执行非流式对话或结构化输出；在轮询路径中按任务自身关联重建配置，避免并发熔断影响结果回收。
- 控制器与服务：后台任务提交、轮询、列表、删除、图片产物获取；成功后补记用量。
架构总览
AI 异步任务采用“浏览器懒轮询 + 轻量后端”的无常驻进程模式：
- 提交阶段：控制器→服务→网关→驱动 submitTask → 仓储写入 provider_task_id 与初始状态
- 轮询阶段：控制器→网关→仓储读取任务→驱动 pollTask → 仓储更新结果/状态
- 超时阶段：按任务类型计算阈值，超过阈值未终态则标记 timeout
- 结果缓存：结果以 JSON 持久化，附带 expires_at 控制有效期
- 用量记录：成功终态时由控制器侧补记用量
sequenceDiagram
participant U as "后台界面"
participant C as "TaskController"
participant S as "TaskService"
participant G as "AiGateway"
participant P as "AsyncTaskPoller"
participant R as "AsyncTaskRepository"
participant D as "驱动/供应商"
U->>C : POST 提交任务
C->>S : submit(app_id, prompt, admin_id, post)
S->>G : submitAsyncTask(model_id, params, meta)
G->>P : submit(config, params, meta)
P->>R : create(写入pending, request_payload)
P->>D : submitTask()
D-->>P : {provider_task_id, status?, result?}
P->>R : markSubmitted(provider_task_id, status)
C-->>U : {task_id, status}
U->>C : GET 轮询 ?id=
C->>G : pollAsyncTask(id)
G->>P : poll(id)
P->>R : find(id)
alt 终态
P-->>C : {status, result, expires_at}
else 进行中
P->>D : pollTask(provider_task_id)
D-->>P : {status, result?}
P->>R : markSucceeded/markFailed/touchPolled
P-->>C : {status, result?}
end详细组件分析
数据模型与索引
- 主键与基础字段：id、provider_id、model_id、key_id、app_id、admin_id、user_id（预留）
- 任务标识与类型：task_type（image/video/audio），provider_task_id（上游任务ID）
- 状态机：pending → running → succeeded/failed/timeout
- 载荷与结果：request_payload（提交时的请求体 JSON），result（结果 JSON，含URL与有效期）
- 错误与时效：error（错误信息），expires_at（结果过期时间）
- 审计与追踪：created_at、updated_at、last_polled_at（最近轮询时间）
- 索引：idx_status_created、idx_provider_task、idx_user_created、idx_admin_created
仓储层（AsyncTaskRepository）
职责：
- 确保表存在，缺失时抛出可读异常
- 创建任务行（默认 pending），写入 provider_id/model_id/key_id/app_id/admin_id/user_id/task_type/request_payload
- 提交后回写 provider_task_id 与初始状态
- 成功/失败/超时标记，结果原地更新
- 轮询时间刷新
- 分页列表（支持 status/task_type/归属字段过滤）
复杂度与优化：
- 单次操作 O(1) 数据库访问
- 表存在性检查每请求一次缓存，降低高频轮询开销
编排层（AsyncTaskPoller）
职责：
- 提交：创建任务行 → 调用驱动 submitTask → 若上游返回同步结果则直接标记成功并设置过期时间 → 否则回写 provider_task_id 与运行状态
- 轮询：读取任务 → 终态直接返回 → 非终态重建配置（绕过配额/熔断）→ 惰性超时判定 → 调用驱动 pollTask → 根据上游状态更新本地
- 同步结果补录：将同步生成的图片产物以任务行形式落库，便于统一前端轮询与使用此图流程
超时策略：
- 按 task_type 设定超时阈值（video/image），未命中默认15分钟
- 基于 created_at 计算耗时，超过阈值即标记 timeout
重试机制：
- 当上游拒绝 size 参数时，自动移除 size 重试一次，避免尺寸限制导致失败
网关层（AiGateway）
职责：
- 解析供应商/模型/密钥，选择驱动
- 暴露 submitAsyncTask/pollAsyncTask 供控制器调用
- 在轮询路径中通过任务自身关联重建配置，确保即使密钥被熔断或配额耗尽，也不影响在途任务的结果回收
控制器与服务（TaskController/TaskService）
职责：
- 提交：校验应用与提示词，白名单透传驱动参数，调用网关提交
- 轮询：调用网关轮询，成功终态补记用量
- 列表：分页展示任务，补齐模型/供应商/管理员名称映射
- 删除：终态任务可删，二次确认
- 图片产物：仅从任务结果中取 URL，data URI 直接截取 base64，HTTP URL 经安全策略拉取并校验格式与大小，防止 SSRF 与恶意内容
状态机管理
状态流转：
- pending：任务已入库等待上游响应
- running：上游已接受并开始处理
- succeeded：上游返回成功，结果已落库
- failed：上游或系统侧判定失败，错误信息已落库
- timeout：惰性超时判定触发，任务长时间无终态
stateDiagram-v2
[*] --> 待处理 : "create()"
待处理 --> 运行中 : "markSubmitted()"
运行中 --> 成功 : "markSucceeded()"
运行中 --> 失败 : "markFailed()"
运行中 --> 超时 : "markTimeout()"
成功 --> [*]
失败 --> [*]
超时 --> [*]结果缓存与过期
- 结果以 JSON 持久化于 result 字段，包含 URLs 等产物信息
- expires_at 用于标记结果有效期；同步直出场景默认设置24小时过期
- 轮询返回时携带 expires_at，前端可按需判断是否继续展示
超时处理与惰性判定
- 按任务类型设置超时阈值（video/image），未命中默认15分钟
- 每次轮询计算 elapsed = now - created_at，超过阈值即标记 timeout 并返回错误
重试机制
- 当上游拒绝 size 参数时，自动移除 size 并重试一次，避免因尺寸约束导致失败
- 该重试发生在提交阶段，不影响后续轮询逻辑
长耗时任务处理建议
- 合理设置 task_type 以匹配超时阈值（如 video 更长）
- 前端轮询间隔建议随任务时长动态调整（例如前30秒每2秒，之后每5-10秒）
- 对超长任务可在业务层增加进度回调或分片处理
监控与告警
- 日志：驱动失败、图片生成失败等关键路径已记录日志通道
- 用量：成功终态时补记用量，便于统计与计费
- 指标建议：
- 任务成功率、失败率、超时率
- 各供应商/模型的任务量与平均耗时
- 结果过期命中率
- 告警建议：
- 连续失败阈值触发熔断（网关内置失败阈值与冷却时间）
- 超时率突增告警
- 供应商不可用或配额耗尽告警
依赖关系分析
classDiagram
class AsyncTaskRepository {
+create(data) int
+find(id) array|null
+delete(id) void
+markSubmitted(id, providerTaskId, status) void
+markSucceeded(id, result, expiresAt) void
+updateResult(id, result) void
+markFailed(id, error) void
+markTimeout(id) void
+touchPolled(id) void
+paginate(filters, page, url, pageSize) array
}
class AsyncTaskPoller {
+submit(config, params, meta) array
+recordSyncResult(config, params, meta, urls, expiresAt) int
+poll(taskId) array
-configForTask(task) array|null
-timeoutMinutes(taskType) int
}
class AiGateway {
+submitAsyncTask(modelId, params, meta) array
+pollAsyncTask(taskId) array
}
class TaskController {
+index(request) Response
+store(request) void
+show(request) void
+image(request) void
+destroy(request) Response
}
class TaskService {
+submit(appId, prompt, adminId, post) array
+getIndexPageData(req) array
+findTask(id) array|null
+delete(id, post) array
}
AsyncTaskPoller --> AsyncTaskRepository : "读写任务行"
TaskController --> TaskService : "提交/列表/删除"
TaskController --> AiGateway : "轮询"
TaskService --> AiGateway : "提交"性能与扩展性
- 索引优化：基于 status/created_at、provider_id/provider_task_id、user_id/admin_id/created_at 的复合索引提升常见查询效率
- 轮询频率：建议前端自适应轮询，避免高频请求造成压力
- 结果体积：result 使用 longtext，注意大对象存储成本；必要时拆分至对象存储并以URL引用
- 可扩展性：通过驱动抽象（AsyncDriverInterface）接入不同供应商；任务类型扩展需在超时阈值与提示词配置中补充
故障排查指南
常见问题与定位：
- 任务不存在：检查 taskId 是否正确，仓储 find 是否返回空
- 配置不可用：轮询时重建配置失败（模型/供应商/密钥缺失），需检查关联记录是否存在
- 供应商不支持异步：驱动未实现异步接口，需更换模型或供应商
- 上游未返回任务ID：提交失败，检查驱动 submitTask 返回值与错误信息
- 图片产物获取失败：data URI 格式不符或 HTTP URL 拉取失败，检查 RemoteUrlPolicy 与客户端大小限制
结论
DouPHP 的 AI 异步任务体系以 dou_ai_task 为核心，结合仓储、编排、网关与控制器，实现了稳定可靠的异步任务提交、轮询、结果缓存与超时处理。通过任务类型区分、惰性超时、结果过期、用量记录与安全产物获取，满足长耗时任务的工程化需求。建议在业务侧配合合理的轮询策略与监控告警，进一步提升可用性与可观测性。
附录：字段与状态说明
- 供应商ID（provider_id）：任务所属供应商
- 模型ID（model_id）：使用的模型
- 密钥ID（key_id）：调用密钥
- 应用ID（app_id）：对齐 usage_log 的应用维度
- 管理员ID（admin_id）：后台生成任务的管理员
- 用户ID（user_id）：前台会员ID（预留）
- 任务类型（task_type）：image/video/audio
- 上游任务ID（provider_task_id）：供应商返回的任务ID
- 状态（status）：pending/running/succeeded/failed/timeout
- 请求体（request_payload）：提交时的请求体 JSON
- 结果（result）：结果 JSON（含URL与有效期）
- 错误信息（error）：失败或超时的错误信息
- 过期时间（expires_at）：结果有效期
- 最近轮询时间（last_polled_at）：最近一次轮询时间