AI 视频生成接口集成

AI 视频生成是整个平台最核心也最复杂的功能。用户提交一段文本描述，后台调用 AI 模型 API 生成视频，这个过程通常需要数十秒到数分钟。在这个异步、不确定的流程中，如何设计可靠的接口集成、状态同步、错误恢复和用户体验，是本章要解决的问题。

1. AI 视频生成的本质

1.1 异步长时任务

AI 视频生成不是一个请求-响应模型——用户提交 prompt 后，AI 模型需要 30s ~ 5min（甚至更久）才能生成结果。这决定了整个技术方案：

用户提交 prompt → 创建任务 → 立即返回 taskId
                              ↓
                    后台轮询/回调 AI API
                              ↓
                    生成完成 → 更新任务状态
                              ↓
                    前端轮询/SSE → 展示结果

1.2 主流 AI 视频 API 对比

平台API 模式生成时长特点
可灵 (Kling)异步 + 轮询30s ~ 3min国内访问快，中文支持好
Runway Gen-3异步 + Webhook1 ~ 5min画质高，风格多
Sora异步 + 回调1 ~ 10minOpenAI 生态
Pika异步 + 轮询30s ~ 2min快速迭代

所有平台都是异步模式——提交任务后返回 taskId，通过轮询或回调获取结果。

2. BFF 层：适配器模式

2.1 为什么需要 BFF 层

直接在前端调用 AI API 有三个问题：

  1. API Key 暴露：前端代码中无法安全存放 API Key
  2. 接口差异：不同 AI 平台的 API 格式不同，前端不应关心
  3. 业务逻辑：配额检查、任务记录、错误处理属于后端逻辑

BFF 层的核心价值：统一接口，隔离差异。

2.2 适配器接口定义

// server/utils/ai-adapters/types.ts
export interface AiVideoAdapter {
  name: string
  createTask(params: CreateTaskParams): Promise<ExternalTaskId>
  queryTask(externalId: string): Promise<TaskStatus>
  cancelTask(externalId: string): Promise<void>
}
 
export interface CreateTaskParams {
  prompt: string
  style?: string
  duration?: number
  aspectRatio?: '16:9' | '9:16' | '1:1'
}
 
export interface TaskStatus {
  status: 'pending' | 'running' | 'completed' | 'failed'
  progress: number       // 0-100
  videoUrl?: string      // 生成完成的视频地址
  error?: string
}
 
type ExternalTaskId = string

2.3 可灵适配器实现

// server/utils/ai-adapters/kling.ts
export class KlingAdapter implements AiVideoAdapter {
  name = 'kling'
  private baseUrl = 'https://api.klingai.com/v1'
  private apiKey: string
 
  constructor() {
    this.apiKey = useRuntimeConfig().klingApiKey
    if (!this.apiKey) throw new Error('KLING_API_KEY not configured')
  }
 
  async createTask(params: CreateTaskParams): Promise<string> {
    const response = await $fetch(`${this.baseUrl}/videos/generations`, {
      method: 'POST',
      headers: { Authorization: `Bearer ${this.apiKey}` },
      body: {
        prompt: params.prompt,
        model: 'kling-v1',
        duration: params.duration || 5,
        aspect_ratio: params.aspectRatio || '16:9',
        style: params.style,
      },
    })
    return response.data.task_id
  }
 
  async queryTask(externalId: string): Promise<TaskStatus> {
    const response = await $fetch(
      `${this.baseUrl}/videos/generations/${externalId}`,
      { headers: { Authorization: `Bearer ${this.apiKey}` } }
    )
 
    const statusMap: Record<string, TaskStatus['status']> = {
      submitted: 'pending',
      processing: 'running',
      succeed: 'completed',
      failed: 'failed',
    }
 
    return {
      status: statusMap[response.data.status] || 'pending',
      progress: response.data.progress || 0,
      videoUrl: response.data.video_url,
      error: response.data.error_message,
    }
  }
 
  async cancelTask(externalId: string): Promise<void> {
    await $fetch(`${this.baseUrl}/videos/generations/${externalId}/cancel`, {
      method: 'POST',
      headers: { Authorization: `Bearer ${this.apiKey}` },
    })
  }
}

2.4 适配器工厂

// server/utils/ai-adapters/factory.ts
const adapters = new Map<string, AiVideoAdapter>()
 
export function getAiAdapter(model: string): AiVideoAdapter {
  if (!adapters.has(model)) {
    switch (model) {
      case 'kling':
        adapters.set(model, new KlingAdapter())
        break
      case 'runway':
        adapters.set(model, new RunwayAdapter())
        break
      default:
        throw createError({ statusCode: 400, message: `Unknown model: ${model}` })
    }
  }
  return adapters.get(model)!
}

新增 AI 模型只需添加一个适配器类 + 注册到工厂，不需要修改任何业务代码。

3. 任务生命周期

3.1 创建任务

// server/api/ai/generate.post.ts
import { z } from 'zod'
 
const schema = z.object({
  prompt: z.string().min(10).max(500),
  model: z.enum(['kling', 'runway']),
  style: z.enum(['realistic', 'anime', 'cinematic']).optional(),
  duration: z.number().min(3).max(60).default(5),
})
 
export default defineEventHandler(async (event) => {
  const session = await requireAuth(event)
  const body = await readValidatedBody(event, schema.parse)
 
  // 1. 配额检查
  const quota = await checkUserQuota(session.user.id)
  if (quota.remaining <= 0) {
    throw createError({ statusCode: 429, message: '生成配额已用完' })
  }
 
  // 2. 调用 AI API 创建任务
  const adapter = getAiAdapter(body.model)
  const externalTaskId = await adapter.createTask({
    prompt: body.prompt,
    style: body.style,
    duration: body.duration,
  })
 
  // 3. 记录任务到数据库
  const db = useDB()
  const [task] = await db.insert(aiTasks).values({
    userId: session.user.id,
    prompt: body.prompt,
    model: body.model,
    style: body.style,
    externalId: externalTaskId,
    status: 'pending',
  }).returning()
 
  // 4. 扣减配额
  await deductQuota(session.user.id)
 
  return { taskId: task.id }
})

3.2 后台轮询：Task Runner

AI 任务创建后，需要一个后台进程定期检查任务状态：

// server/plugins/task-runner.ts
export default defineNitroPlugin((nitro) => {
  const POLL_INTERVAL = 5000 // 5 秒
 
  async function pollPendingTasks() {
    const db = useDB()
    const pendingTasks = await db.select()
      .from(aiTasks)
      .where(inArray(aiTasks.status, ['pending', 'running']))
      .limit(20)
 
    for (const task of pendingTasks) {
      try {
        const adapter = getAiAdapter(task.model)
        const status = await adapter.queryTask(task.externalId)
 
        await db.update(aiTasks)
          .set({
            status: status.status,
            progress: status.progress,
            ...(status.status === 'completed' && {
              completedAt: new Date(),
            }),
            ...(status.error && {
              errorMessage: status.error,
            }),
          })
          .where(eq(aiTasks.id, task.id))
 
        // 生成完成：创建视频记录
        if (status.status === 'completed' && status.videoUrl) {
          await createVideoFromAiTask(task, status.videoUrl)
        }
      } catch (error) {
        console.error(`Task ${task.id} poll failed:`, error)
      }
    }
  }
 
  // 定时执行
  const timer = setInterval(pollPendingTasks, POLL_INTERVAL)
 
  // 服务关闭时清理
  nitro.hooks.hook('close', () => clearInterval(timer))
})

3.3 从 AI 结果创建视频

// server/utils/ai-task-handler.ts
async function createVideoFromAiTask(task: AiTask, videoUrl: string) {
  const db = useDB()
 
  const [video] = await db.insert(videos).values({
    title: `AI 生成 - ${task.prompt.slice(0, 30)}`,
    userId: task.userId,
    sourceType: 'ai_generated',
    sourceUrl: videoUrl,
    status: 'processing', // 还需要转码处理
  }).returning()
 
  // 关联任务和视频
  await db.update(aiTasks)
    .set({ videoId: video.id })
    .where(eq(aiTasks.id, task.id))
 
  // 触发转码流水线（下一章详述）
  await triggerTranscoding(video.id, videoUrl)
}

4. 前端状态同步

4.1 轮询 vs SSE vs WebSocket

方案复杂度实时性可靠性适用场景
轮询低秒级高AI 任务（不需要毫秒级实时）
SSE中亚秒中实时进度流
WebSocket高毫秒低双向通信（弹幕、聊天）

对于 AI 视频生成（30s ~ 5min），轮询是最佳选择——实现简单、不怕断连、服务端无状态。

4.2 前端轮询 Composable

// composables/useAiTask.ts
export function useAiTask(taskId: Ref<string | null>) {
  const task = ref<AiTask | null>(null)
  const isPolling = ref(false)
 
  const { data, refresh } = useFetch(() =>
    taskId.value ? `/api/ai/tasks/${taskId.value}` : null,
    { watch: false }
  )
 
  let timer: ReturnType<typeof setInterval> | null = null
 
  function startPolling() {
    if (isPolling.value) return
    isPolling.value = true
 
    timer = setInterval(async () => {
      await refresh()
      task.value = data.value
 
      // 终态停止轮询
      if (task.value && ['completed', 'failed', 'cancelled'].includes(task.value.status)) {
        stopPolling()
      }
    }, 3000) // 3 秒间隔
  }
 
  function stopPolling() {
    if (timer) {
      clearInterval(timer)
      timer = null
    }
    isPolling.value = false
  }
 
  // 组件卸载时清理
  onUnmounted(stopPolling)
 
  return { task, isPolling, startPolling, stopPolling }
}

4.3 生成页面 UI

<!-- pages/create/index.vue -->
<script setup>
const prompt = ref('')
const selectedModel = ref('kling')
const taskId = ref<string | null>(null)
 
const { task, isPolling, startPolling } = useAiTask(taskId)
 
async function handleGenerate() {
  const { data } = await useFetch('/api/ai/generate', {
    method: 'POST',
    body: {
      prompt: prompt.value,
      model: selectedModel.value,
    },
  })
 
  if (data.value?.taskId) {
    taskId.value = data.value.taskId
    startPolling()
  }
}
</script>
 
<template>
  <div class="max-w-2xl mx-auto p-6">
    <!-- 输入区域 -->
    <div v-if="!taskId">
      <UTextarea v-model="prompt" placeholder="描述你想生成的视频..." rows="4" />
      <UButton @click="handleGenerate" :disabled="!prompt" class="mt-4">
        生成视频
      </UButton>
    </div>
 
    <!-- 进度展示 -->
    <div v-else-if="task">
      <div v-if="task.status === 'running'" class="text-center">
        <UProgress :value="task.progress" class="mb-4" />
        <p class="text-gray-500">生成中... {{ task.progress }}%</p>
      </div>
 
      <div v-else-if="task.status === 'completed'">
        <p class="text-green-600 mb-4">生成完成！</p>
        <NuxtLink :to="`/video/${task.videoId}`">
          <UButton>查看视频</UButton>
        </NuxtLink>
      </div>
 
      <div v-else-if="task.status === 'failed'">
        <p class="text-red-600">生成失败：{{ task.errorMessage }}</p>
        <UButton variant="outline" @click="taskId = null">重新生成</UButton>
      </div>
    </div>
  </div>
</template>

5. 错误处理与重试

5.1 重试策略

AI API 调用可能因为网络波动、服务过载等原因失败。采用指数退避重试：

// server/utils/retry.ts
export async function withRetry<T>(
  fn: () => Promise<T>,
  options: { maxRetries?: number; baseDelay?: number } = {}
): Promise<T> {
  const { maxRetries = 3, baseDelay = 1000 } = options
 
  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await fn()
    } catch (error: any) {
      if (attempt === maxRetries) throw error
 
      // 不重试的错误
      if (error.statusCode === 400 || error.statusCode === 401) throw error
 
      const delay = baseDelay * Math.pow(2, attempt) + Math.random() * 1000
      await new Promise(resolve => setTimeout(resolve, delay))
    }
  }
  throw new Error('Unreachable')
}

5.2 在任务轮询中使用

// Task Runner 中增加重试计数
if (status.status === 'failed' && task.retryCount < 3) {
  // 重新提交到 AI API
  const adapter = getAiAdapter(task.model)
  const newExternalId = await withRetry(() =>
    adapter.createTask({ prompt: task.prompt, style: task.style })
  )
 
  await db.update(aiTasks).set({
    externalId: newExternalId,
    status: 'pending',
    retryCount: task.retryCount + 1,
    errorMessage: null,
  }).where(eq(aiTasks.id, task.id))
}

5.3 配额补偿

如果任务最终失败（重试次数用完），应该退还用户配额：

if (status.status === 'failed' && task.retryCount >= 3) {
  await db.update(aiTasks).set({
    status: 'failed',
    errorMessage: status.error || '生成失败，已自动退还配额',
  }).where(eq(aiTasks.id, task.id))
 
  // 退还配额
  await refundQuota(task.userId)
}

6. 安全与限流

6.1 接口限流

AI 生成接口是昂贵的——必须限流：

// server/middleware/rate-limit.ts
const rateLimitMap = new Map<string, { count: number; resetAt: number }>()
 
export default defineEventHandler((event) => {
  if (!event.path.startsWith('/api/ai/generate')) return
 
  const userId = event.context.session?.user?.id
  if (!userId) return
 
  const now = Date.now()
  const limit = rateLimitMap.get(userId)
 
  if (limit && now < limit.resetAt) {
    if (limit.count >= 10) { // 每小时 10 次
      throw createError({
        statusCode: 429,
        message: '请求过于频繁，请稍后再试',
      })
    }
    limit.count++
  } else {
    rateLimitMap.set(userId, { count: 1, resetAt: now + 3600000 })
  }
})

6.2 Prompt 安全过滤

用户输入的 prompt 需要检查：

// server/utils/prompt-filter.ts
const BLOCKED_KEYWORDS = ['violence', 'explicit', /* ... */]
 
export function validatePrompt(prompt: string): { valid: boolean; reason?: string } {
  const lower = prompt.toLowerCase()
 
  for (const keyword of BLOCKED_KEYWORDS) {
    if (lower.includes(keyword)) {
      return { valid: false, reason: '内容包含违规关键词' }
    }
  }
 
  if (prompt.length < 10) {
    return { valid: false, reason: '描述太短，请提供更多细节' }
  }
 
  return { valid: true }
}

本章小结

  • 异步架构：AI 视频生成是长时异步任务，采用"创建任务 → 后台轮询 → 前端同步"三段式
  • 适配器模式：BFF 层统一封装不同 AI 平台 API，新增模型只需添加适配器类
  • 任务生命周期：pending → running → completed/failed，Nitro Plugin 后台定时轮询
  • 前端轮询：3s 间隔的 useAiTask Composable，终态自动停止，组件卸载自动清理
  • 可靠性：指数退避重试（最多 3 次）+ 配额补偿（失败退还）
  • 安全：接口限流（每小时 10 次）+ Prompt 安全过滤