17.11-批量AI生成任务

要点

批量 AI 生成：一次调用 LLM 生成几十、几百条内容——批量摘要、批量翻译、批量分类
并发控制很重要——同时调 100 次 LLM API 会触发限流
成本控制——批量生成烧钱，需要追踪 token 用量和费用
分批处理——把大批量拆成小批，每批处理完更新进度
失败处理——部分失败时保留已成功的结果，不全部重来

1. 批量生成的场景

批量摘要：100 篇文档，每篇生成 200 字摘要
批量翻译：把网站 500 个页面从中文翻译成英文
批量分类：给 1000 篇文章打标签
批量提取：从 200 份合同里提取关键条款
批量改写：把 50 条产品描述改写成营销文案

这些场景的共同特点：

输入是批量数据（几十到几千条）
每条独立调用 LLM（或批量调用）
总耗时长（几分钟到几小时）
需要异步处理

2. 任务定义

interface BatchGenerateTask {
  batchId: string
  userId: string
  items: Array<{
    id: string
    input: string
    metadata?: Record<string, unknown>
  }>
  prompt: string            // 系统 prompt 或模板
  model: string             // 模型名称
  options: {
    maxTokens: number
    temperature: number
    concurrency: number     // 并发数
  }
}
 
interface BatchResult {
  batchId: string
  status: 'pending' | 'processing' | 'completed' | 'failed'
  totalItems: number
  completedItems: number
  failedItems: number
  totalTokens: number
  totalCost: number
  results: Array<{
    itemId: string
    output?: string
    tokens: number
    error?: string
  }>
}

3. 分批处理

大批量不能一次性处理——需要拆成小批，每批并发处理。

async function batchGenerate(
  task: BatchGenerateTask,
  env: Env
): Promise<BatchResult> {
  const { items, prompt, model, options } = task
  const { concurrency } = options
 
  const result: BatchResult = {
    batchId: task.batchId,
    status: 'processing',
    totalItems: items.length,
    completedItems: 0,
    failedItems: 0,
    totalTokens: 0,
    totalCost: 0,
    results: [],
  }
 
  // 分批处理
  for (let i = 0; i < items.length; i += concurrency) {
    const batch = items.slice(i, i + concurrency)
 
    // 并发处理当前批次
    const promises = batch.map((item) =>
      generateSingle(item, prompt, model, options, env)
        .then((res) => ({ ...res, itemId: item.id }))
        .catch((err) => ({
          itemId: item.id,
          error: err.message,
          tokens: 0,
        }))
    )
 
    const batchResults = await Promise.all(promises)
 
    // 合并结果
    for (const res of batchResults) {
      if (res.error) {
        result.failedItems++
      } else {
        result.completedItems++
        result.totalTokens += res.tokens
        result.totalCost += estimateCost(res.tokens, model)
      }
      result.results.push(res)
    }
 
    // 更新进度
    await updateBatchProgress(env, task.batchId, result)
 
    // 限流：批次之间等待
    if (i + concurrency < items.length) {
      await sleep(1000)  // 1 秒间隔
    }
  }
 
  result.status = result.failedItems === result.totalItems ? 'failed' : 'completed'
  return result
}

4. 单条生成

async function generateSingle(
  item: { id: string; input: string },
  prompt: string,
  model: string,
  options: { maxTokens: number; temperature: number },
  env: Env
): Promise<{ output: string; tokens: number }> {
  // 替换模板变量
  const finalPrompt = prompt.replace('{{input}}', item.input)
 
  // 调用 LLM API
  const response = await env.AI.run(model, {
    messages: [
      { role: 'system', content: finalPrompt },
      { role: 'user', content: item.input },
    ],
    max_tokens: options.maxTokens,
    temperature: options.temperature,
  })
 
  const output = response.choices?.[0]?.message?.content ?? ''
  const tokens = response.usage?.total_tokens ?? estimateTokens(output)
 
  return { output, tokens }
}

5. 并发控制

并发数（concurrency）控制同时调用 LLM 的次数。太高会触发限流，太低处理太慢。

// 根据模型调整并发数
const MODEL_CONCURRENCY: Record<string, number> = {
  '@cf/openai/gpt-4o-mini': 10,     // 小模型，可以高并发
  '@cf/openai/gpt-4o': 5,           // 大模型，控制并发
  '@cf/anthropic/claude-3-haiku': 10,
  '@cf/anthropic/claude-3-sonnet': 5,
  '@cf/anthropic/claude-3-opus': 2,  // 最贵的模型，低并发
}
 
function getConcurrency(model: string): number {
  return MODEL_CONCURRENCY[model] ?? 5
}

动态调整并发

如果检测到限流，降低并发：

async function batchGenerateWithAdaptiveConcurrency(
  task: BatchGenerateTask,
  env: Env
): Promise<BatchResult> {
  let concurrency = task.options.concurrency
  const result = initResult(task)
 
  for (let i = 0; i < task.items.length; i += concurrency) {
    const batch = task.items.slice(i, i + concurrency)
 
    try {
      const batchResults = await Promise.all(
        batch.map((item) => generateSingle(item, task.prompt, task.model, task.options, env))
      )
 
      // 成功，合并结果
      mergeResults(result, batchResults)
 
      // 连续成功 3 次，尝试提高并发
      if (consecutiveSuccesses >= 3 && concurrency < 20) {
        concurrency++
      }
    } catch (err) {
      if (err.message.includes('rate limit')) {
        // 限流，降低并发
        concurrency = Math.max(1, Math.floor(concurrency / 2))
        await sleep(5000)  // 等 5 秒再试
      } else {
        // 其他错误
        mergeError(result, err)
      }
    }
 
    await updateBatchProgress(env, task.batchId, result)
    await sleep(1000)
  }
 
  return result
}

6. 成本控制

批量生成烧钱。需要追踪 token 用量和费用。

// 估算 token 数（粗略）
function estimateTokens(text: string): number {
  const cjk = (text.match(/[一-鿿]/g) || []).length
  return Math.ceil(cjk / 1.5 + (text.length - cjk) / 4)
}
 
// 估算费用
function estimateCost(tokens: number, model: string): number {
  const PRICING: Record<string, number> = {
    '@cf/openai/gpt-4o-mini': 0.00015,     // $0.15 / 1M tokens
    '@cf/openai/gpt-4o': 0.0025,            // $2.50 / 1M tokens
    '@cf/anthropic/claude-3-haiku': 0.00025,
    '@cf/anthropic/claude-3-sonnet': 0.003,
    '@cf/anthropic/claude-3-opus': 0.015,
  }
 
  const pricePerToken = PRICING[model] ?? 0.001
  return tokens * pricePerToken
}
 
// 检查预算
async function checkBudget(userId: string, estimatedCost: number, env: Env): Promise<boolean> {
  const user = await env.DB.prepare(`
    SELECT budget_limit, budget_used FROM users WHERE id = ?
  `).bind(userId).first()
 
  if (!user) return false
  if (user.budget_limit === null) return true  // 无限制
 
  return (user.budget_used + estimatedCost) <= user.budget_limit
}

在批量任务开始前检查预算：

app.post('/api/batch-generate', async (c) => {
  const { items, prompt, model } = await c.req.json()
 
  // 估算总 token 和费用
  const estimatedTokens = items.reduce((sum, item) => {
    return sum + estimateTokens(item.input) + 200  // 200 是输出预估
  }, 0)
  const estimatedCost = estimateCost(estimatedTokens, model)
 
  // 检查预算
  if (!await checkBudget(c.get('user').id, estimatedCost, c.env)) {
    return c.json({ error: '预算不足' }, 402)
  }
 
  // 创建任务...
})

7. 进度追踪

批量任务的进度：已完成 X / Y 条，失败 Z 条。

CREATE TABLE batch_tasks (
  id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL,
  total_items INTEGER NOT NULL,
  completed_items INTEGER NOT NULL DEFAULT 0,
  failed_items INTEGER NOT NULL DEFAULT 0,
  total_tokens INTEGER NOT NULL DEFAULT 0,
  total_cost REAL NOT NULL DEFAULT 0,
  status TEXT NOT NULL DEFAULT 'pending',
  created_at TEXT NOT NULL,
  updated_at TEXT NOT NULL
);
 
CREATE TABLE batch_results (
  id TEXT PRIMARY KEY,
  batch_id TEXT NOT NULL,
  item_id TEXT NOT NULL,
  output TEXT,
  tokens INTEGER NOT NULL DEFAULT 0,
  error TEXT,
  created_at TEXT NOT NULL
);

async function updateBatchProgress(env: Env, batchId: string, result: BatchResult): Promise<void> {
  await env.DB.prepare(`
    UPDATE batch_tasks
    SET completed_items = ?, failed_items = ?, total_tokens = ?,
        total_cost = ?, status = ?, updated_at = ?
    WHERE id = ?
  `).bind(
    result.completedItems,
    result.failedItems,
    result.totalTokens,
    result.totalCost,
    result.status,
    new Date().toISOString(),
    batchId,
  ).run()
}

前端轮询：

async function pollBatchProgress(batchId: string): Promise<void> {
  const interval = setInterval(async () => {
    const data = await fetch(`/api/batch-tasks/${batchId}`).then(r => r.json())
 
    updateProgressBar(data.completedItems / data.totalItems * 100)
    updateStats({
      completed: data.completedItems,
      failed: data.failedItems,
      total: data.totalItems,
      cost: data.total_cost,
    })
 
    if (data.status === 'completed' || data.status === 'failed') {
      clearInterval(interval)
    }
  }, 3000)
}

8. 结果查询

批量完成后，查询每条结果：

app.get('/api/batch-tasks/:batchId/results', async (c) => {
  const batchId = c.req.param('batchId')
 
  const results = await c.env.DB.prepare(`
    SELECT item_id, output, tokens, error
    FROM batch_results
    WHERE batch_id = ?
    ORDER BY created_at
  `).bind(batchId).all()
 
  return c.json({
    batchId,
    results: results.results,
  })
})
 
// 导出为 CSV
app.get('/api/batch-tasks/:batchId/export', async (c) => {
  const results = await c.env.DB.prepare(`
    SELECT item_id, output FROM batch_results WHERE batch_id = ? AND output IS NOT NULL
  `).bind(c.req.param('batchId')).all()
 
  const csv = [
    'item_id,output',
    ...results.results.map((r) => `${r.item_id},"${r.output.replace(/"/g, '""')}"`),
  ].join('\n')
 
  return new Response(csv, {
    headers: {
      'Content-Type': 'text/csv',
      'Content-Disposition': `attachment; filename="batch-${c.req.param('batchId')}.csv"`,
    },
  })
})

9. 失败处理

部分失败时保留已成功的结果：

async function batchGenerateWithRetry(
  task: BatchGenerateTask,
  env: Env
): Promise<BatchResult> {
  const result = initResult(task)
  const failedItems: typeof task.items = []
 
  // 第一轮处理
  for (let i = 0; i < task.items.length; i += task.options.concurrency) {
    const batch = task.items.slice(i, i + task.options.concurrency)
 
    const promises = batch.map(async (item) => {
      try {
        const res = await generateSingle(item, task.prompt, task.model, task.options, env)
        return { ...res, itemId: item.id, success: true }
      } catch (err) {
        return { itemId: item.id, error: err.message, success: false }
      }
    })
 
    const batchResults = await Promise.all(promises)
 
    for (const res of batchResults) {
      if (res.success) {
        result.completedItems++
        result.totalTokens += res.tokens
        saveResult(env, task.batchId, res)
      } else {
        result.failedItems++
        const failedItem = task.items.find((i) => i.id === res.itemId)!
        failedItems.push(failedItem)
      }
    }
 
    await updateBatchProgress(env, task.batchId, result)
  }
 
  // 第二轮重试失败项
  if (failedItems.length > 0) {
    await sleep(5000)  // 等 5 秒
 
    for (const item of failedItems) {
      try {
        const res = await generateSingle(item, task.prompt, task.model, task.options, env)
        result.completedItems++
        result.failedItems--
        result.totalTokens += res.tokens
        saveResult(env, task.batchId, { ...res, itemId: item.id })
      } catch (err) {
        // 重试还是失败，记录错误
        saveError(env, task.batchId, item.id, err.message)
      }
 
      await updateBatchProgress(env, task.batchId, result)
    }
  }
 
  return result
}

10. 完整流程

// 1. 提交批量任务
app.post('/api/batch-generate', async (c) => {
  const { items, prompt, model } = await c.req.json()
 
  const batchId = crypto.randomUUID()
  const now = new Date().toISOString()
 
  await c.env.DB.prepare(`
    INSERT INTO batch_tasks (id, user_id, total_items, status, created_at, updated_at)
    VALUES (?, ?, ?, 'pending', ?, ?)
  `).bind(batchId, c.get('user').id, items.length, now, now).run()
 
  await c.env.BATCH_QUEUE.send({
    batchId,
    userId: c.get('user').id,
    items,
    prompt,
    model,
    options: {
      maxTokens: 500,
      temperature: 0.7,
      concurrency: getConcurrency(model),
    },
  })
 
  return c.json({ batchId, status: 'pending' }, 202)
})
 
// 2. Queue 消费者
export async function batchConsumer(
  batch: MessageBatch<BatchGenerateTask>,
  env: Env
): Promise<void> {
  for (const msg of batch.messages) {
    const result = await batchGenerate(msg.body, env)
    await saveBatchResult(env, result)
    msg.ack()
  }
}

总结

回顾这一节的要点：

批量 AI 生成：一次调 LLM 生成几十到几千条内容
并发控制很重要——太高触发限流，太低处理太慢
分批处理：拆成小批，每批并发，批次间等待
动态调整并发：检测到限流时降低并发，连续成功时提高
成本控制：估算 token 和费用，检查预算
进度追踪：已完成 X / Y 条，失败 Z 条，总费用
失败处理：部分失败保留已成功的，失败项可重试
结果导出：支持 JSON、CSV 格式

下一篇讲工作流编排——多个任务如何串联成复杂流程。