batch-processing

批量处理

Agent Skills™ 的一部分，由 googleadsagent.ai™ 提供

描述

批量处理支持并行执行 AI 任务，并提供进度跟踪、错误处理、速率限制和结果汇总。Agent 会通过 AI 流水线并发处理大量项目——文档、图像、代码文件、API 请求——管理吞吐量、故障和部分结果，确保工作不会丢失。

处理单个项目的 AI 任务很简单；大规模批量处理则会带来故障、速率限制、内存约束以及支持断点续跑的需求。这项技能可以应对这些生产环境中的实际问题：可配置的并发限制、速率限制错误时的指数退避、崩溃后的基于检查点的断点续跑，以及结构化的进度报告，准确显示哪些项目已成功、失败或待处理。

这项技能同时支持同质批次（对每个项目执行相同的操作）和异质批次（根据项目类型分配不同的操作）。结果会汇总为结构化报告，其中包含每个项目的状态、耗时和错误详情。失败的项目会自动通过退避机制重试，永久失败的项目则会收集到死信队列中，以供人工检查。

使用场景

• 通过 AI 流水线处理数百或数千个项目
• 翻译、总结或分类大量文档
• 为文档语料库生成嵌入
• 对整个代码仓库运行代码分析
• 批量生成图像、描述或元数据
• 任何按顺序处理项目，但能从并行处理中获益的任务

工作原理

graph TD
    A[Input Batch: N Items] --> B[Load Checkpoint: Resume if Exists]
    B --> C[Partition into Work Chunks]
    C --> D[Parallel Workers: Concurrency Limit]
    D --> E[Worker 1: Process Item]
    D --> F[Worker 2: Process Item]
    D --> G[Worker K: Process Item]
    E --> H{Success?}
    F --> H
    G --> H
    H -->|Yes| I[Record Result]
    H -->|Rate Limited| J[Backoff + Retry]
    H -->|Failed| K[Retry Queue]
    J --> D
    K --> L{Retries Exhausted?}
    L -->|No| D
    L -->|Yes| M[Dead Letter Queue]
    I --> N[Save Checkpoint]
    N --> O[Progress Report]
    O --> P{All Done?}
    P -->|No| D
    P -->|Yes| Q[Final Aggregation Report]

引擎会在并发工作线程之间分配任务，同时遵守速率限制和重试预算。每个任务块完成后都会持久化检查点，以便在崩溃后恢复。死信队列会收集永久失败的项目，供人工审核。

实现

import asyncio
from dataclasses import dataclass, field
from time import time
import json

@dataclass
class BatchItem:
    id: str
    input: dict
    status: str = "pending"
    result: dict | None = None
    error: str | None = None
    attempts: int = 0
    duration_ms: float = 0

@dataclass
class BatchResult:
    total: int
    succeeded: int
    failed: int
    dead_letter: int
    duration_s: float
    items: list[BatchItem] = field(default_factory=list)

class BatchProcessor:
    def __init__(self, concurrency: int = 5, max_retries: int = 3, checkpoint_file: str = "batch_checkpoint.json"):
        self.concurrency = concurrency
        self.max_retries = max_retries
        self.checkpoint_file = checkpoint_file
        self.semaphore = asyncio.Semaphore(concurrency)

    async def process(self, items: list[BatchItem], processor_fn) -> BatchResult:
        items = self._load_checkpoint(items)
        start = time()
        pending = [i for i in items if i.status == "pending"]

        tasks = [self._process_item(item, processor_fn) for item in pending]
        await asyncio.gather(*tasks, return_exceptions=True)

        return BatchResult(
            total=len(items),
            succeeded=sum(1 for i in items if i.status == "succeeded"),
            failed=sum(1 for i in items if i.status == "failed"),
            dead_letter=sum(1 for i in items if i.status == "dead_letter"),
            duration_s=time() - start,
            items=items,
        )

    async def _process_item(self, item: BatchItem, processor_fn):
        async with self.semaphore:
            while item.attempts < self.max_retries:
                item.attempts += 1
                start = time()
                try:
                    item.result = await processor_fn(item.input)
                    item.status = "succeeded"
                    item.duration_ms = (time() - start) * 1000
                    self._save_checkpoint_item(item)
                    return
                except RateLimitError:
                    await asyncio.sleep(2 ** item.attempts)
                except Exception as e:
                    item.error = str(e)
                    item.duration_ms = (time() - start) * 1000

            item.status = "dead_letter"
            self._save_checkpoint_item(item)

    def _save_checkpoint_item(self, item: BatchItem):
        try:
            data = json.loads(open(self.checkpoint_file).read()) if Path(self.checkpoint_file).exists() else {}
            data[item.id] = {"status": item.status, "result": item.result, "error": item.error}
            with open(self.checkpoint_file, "w") as f:
                json.dump(data, f)
        except Exception:
            pass

    def _load_checkpoint(self, items: list[BatchItem]) -> list[BatchItem]:
        try:
            data = json.loads(open(self.checkpoint_file).read())
            for item in items:
                if item.id in data:
                    item.status = data[item.id]["status"]
                    item.result = data[item.id]["result"]
        except FileNotFoundError:
            pass
        return items

最佳实践

• 根据 API 速率限制设置并发上限，而不只是根据 CPU 核心数
• 对任何超过 100 个项目的批次，实现基于检查点的断点续跑
• 对速率限制重试使用指数退避（2^attempt 秒）
• 定期报告进度（每 10% 或每 30 秒）
• 单独收集死信项目，以便人工审核和重新处理
• 记录每个项目的耗时，以找出拖慢吞吐量的慢速项目

平台兼容性

平台	支持情况	备注
Cursor	完全支持	Python/TS 异步执行
VS Code	完全支持	基于终端运行批处理
Windsurf	完全支持	支持批处理工作流
Claude Code	完全支持	脚本执行
Cline	完全支持	批量任务管理
aider	部分支持	仅支持顺序处理

相关技能

• 工作流编排
• 低代码生成
• 并行 Agent 编排
• Token 优化

关键词

batch-processing parallel-execution rate-limiting retry-logic checkpointing dead-letter-queue progress-tracking concurrency

---

© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License