AI Prompt Engineering · 课时

批处理与异步执行

Q: 「批处理与异步执行」这节课中我会学到什么？

OpenAI 批处理 API、异步 Python 和并发执行提示词。 你通过在浏览器中直接运行的动手代码来练习 AI Prompt Engineering，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

OpenAI 批处理 API、异步 Python 和并发执行提示词。

批处理与异步执行 是 CoddyKit 上的免费 AI Prompt Engineering 课时。这是第 2 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Prompt Engineering 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Prompt Engineering 课程共包含 4 节课。

为什么要使用批处理和异步执行

按顺序处理数千个 LLM 请求既缓慢又昂贵。批处理将请求分组，可降低 50% 的成本。异步执行会并行处理请求，在速率限制范围内最大化吞吐量。二者结合后，可以大幅降低成本和实际耗时。

OpenAI 批处理接口：成本降低 50%

OpenAI 批处理接口会在后台异步处理请求（最长 24 小时），价格仅为正常接口价格的 50%。它非常适合评估运行、数据集处理以及非实时工作负载。

import openai
import json

client = openai.OpenAI(api_key='YOUR_API_KEY')

# Step 1: Create batch input file (JSONL format)
batch_requests = [
    {
        'custom_id': f'request-{i}',
        'method': 'POST',
        'url': '/v1/chat/completions',
        'body': {
            'model': 'gpt-4o-mini',
            'messages': [
                {'role': 'user', 'content': f'Summarize this document: {doc}'}
            ],
            'max_tokens': 200
        }
    }
    for i, doc in enumerate(['Doc A text...', 'Doc B text...', 'Doc C text...'])
]

# Write to JSONL file
with open('batch_input.jsonl', 'w') as f:
    for req in batch_requests:
        f.write(json.dumps(req) + '\n')

# Step 2: Upload the file
batch_file = client.files.create(
    file=open('batch_input.jsonl', 'rb'),
    purpose='batch'
)
print(f'Batch file uploaded: {batch_file.id}')

提交并轮询批处理任务

上传输入文件后，创建批处理任务并持续轮询，直到任务完成。批处理接口会在 24 小时内处理请求（对于较小的批次，通常会快得多）。

import time

# Step 3: Create batch job
batch = client.batches.create(
    input_file_id=batch_file.id,
    endpoint='/v1/chat/completions',
    completion_window='24h'
)
print(f'Batch created: {batch.id} | Status: {batch.status}')

# Step 4: Poll for completion
def wait_for_batch(batch_id, poll_interval=30, timeout=3600):
    start = time.time()
    while time.time() - start < timeout:
        batch = client.batches.retrieve(batch_id)
        print(f'Status: {batch.status} | '
              f'Completed: {batch.request_counts.completed}/ '
              f'{batch.request_counts.total}')
        if batch.status == 'completed':
            return batch
        if batch.status in ('failed', 'expired', 'cancelling', 'cancelled'):
            raise RuntimeError(f'Batch {batch_id} ended with status: {batch.status}')
        time.sleep(poll_interval)
    raise TimeoutError('Batch polling timed out')

# batch = wait_for_batch(batch.id)

获取批处理结果

批处理完成后，下载输出文件，并将 JSONL 结果解析为可用格式。

def retrieve_batch_results(batch):
    if not batch.output_file_id:
        raise ValueError('No output file — batch may have failed')

    # Download output file
    content = client.files.content(batch.output_file_id).text

    # Parse JSONL: one result per line
    results = {}
    for line in content.strip().split('\n'):
        if not line:
            continue
        result = json.loads(line)
        custom_id = result['custom_id']
        if result.get('error'):
            results[custom_id] = {'error': result['error']}
        else:
            response_body = result['response']['body']
            text = response_body['choices'][0]['message']['content']
            results[custom_id] = {'text': text}

    # Report error rate
    errors = sum(1 for r in results.values() if 'error' in r)
    print(f'Retrieved {len(results)} results, {errors} errors')
    return results

# results = retrieve_batch_results(batch)
# for req_id, result in results.items():
#     print(req_id, result.get('text', result.get('error', ''))[:50])

使用 asyncio.gather() 执行异步 Python

对于实时（非批处理）并行任务，Python 的 asyncio 与 asyncio.gather() 会并发发起多个接口调用，并等待它们全部完成。对于包含多个请求的工作负载，这可以大幅减少总实际耗时。

import asyncio
import openai

async_client = openai.AsyncOpenAI(api_key='YOUR_API_KEY')

async def async_completion(messages, model='gpt-4o-mini', max_tokens=200):
    response = await async_client.chat.completions.create(
        model=model,
        messages=messages,
        max_tokens=max_tokens
    )
    return response.choices[0].message.content

async def process_parallel(prompts):
    tasks = [
        async_completion([{'role': 'user', 'content': p}])
        for p in prompts
    ]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

# Usage
async def main():
    prompts = ['Explain photosynthesis.', 'Explain gravity.', 'Explain evolution.']
    results = await process_parallel(prompts)
    for prompt, result in zip(prompts, results):
        if isinstance(result, Exception):
            print(f'ERROR: {result}')
        else:
            print(f'{prompt[:30]}... -> {result[:60]}...')

# asyncio.run(main())
print('asyncio.gather: all 3 requests fire simultaneously')

遵守速率限制的并发调用

发起过多并发请求会触发速率限制错误。Semaphore 会限制并发量，使其保持在速率限制范围内，同时最大化吞吐量。

import asyncio

# Rate limits (example for gpt-4o-mini):
# RPM (requests per minute): 500
# TPM (tokens per minute): 200,000

MAX_CONCURRENT = 20  # stay well below rate limit

async def process_with_rate_limit(prompts, max_concurrent=MAX_CONCURRENT):
    semaphore = asyncio.Semaphore(max_concurrent)
    results = [None] * len(prompts)

    async def bounded_completion(i, prompt):
        async with semaphore:
            try:
                result = await async_completion(
                    [{'role': 'user', 'content': prompt}]
                )
                results[i] = result
            except openai.RateLimitError as e:
                print(f'Rate limited on prompt {i}: {e}')
                await asyncio.sleep(60)  # back off and retry
                result = await async_completion(
                    [{'role': 'user', 'content': prompt}]
                )
                results[i] = result

    await asyncio.gather(*[
        bounded_completion(i, p) for i, p in enumerate(prompts)
    ])
    return results

print('Semaphore limits to', MAX_CONCURRENT, 'concurrent requests')

Anthropic 批处理接口

Anthropic 也提供 Message Batches 接口，其经济性与 OpenAI 的批处理接口相似。批次会被异步处理，结果可以通过轮询或流式传输获取。

import anthropic

client = anthropic.Anthropic(api_key='YOUR_API_KEY')

# Create a batch of messages
batch = client.messages.batches.create(
    requests=[
        {
            'custom_id': f'doc-{i}',
            'params': {
                'model': 'claude-haiku-4-5',
                'max_tokens': 200,
                'messages': [{
                    'role': 'user',
                    'content': f'Classify the sentiment of: {text}'
                }]
            }
        }
        for i, text in enumerate([
            'Amazing product, exceeded expectations!',
            'Terrible quality, broke after one use.',
            'It works as described.'
        ])
    ]
)
print(f'Batch created: {batch.id} | Status: {batch.processing_status}')

# Poll for completion
# while (batch := client.messages.batches.retrieve(batch.id)).processing_status != 'ended':
#     time.sleep(30)

# Retrieve results
# for result in client.messages.batches.results(batch.id):
#     print(result.custom_id, result.result.message.content[0].text[:50])

吞吐量优化：批处理策略

根据延迟要求和工作负载特征选择合适的批处理策略，以最大化吞吐量。

throughput_strategies = {
    'API Batch (OpenAI/Anthropic)': {
        'cost': '50% of normal price',
        'latency': 'Minutes to hours (background processing)',
        'best_for': 'Offline workloads: eval runs, dataset labeling, report generation',
        'max_batch_size': '50,000 requests per batch'
    },
    'asyncio.gather()': {
        'cost': 'Normal price',
        'latency': 'Same as slowest individual request',
        'best_for': 'Real-time parallel enrichment, multi-step pipelines',
        'max_concurrent': '10-50 depending on rate limits'
    },
    'Streaming + Concurrent': {
        'cost': 'Normal price',
        'latency': 'First token arrives faster, total similar',
        'best_for': 'User-facing applications needing perceived speed',
        'pattern': 'asyncio with stream=True per request'
    },
    'Worker Queue (Celery, RQ)': {
        'cost': 'Normal price',
        'latency': 'Variable (depends on queue depth)',
        'best_for': 'High-volume production with auto-scaling workers',
        'backends': 'Redis, RabbitMQ'
    }
}

for strategy, details in throughput_strategies.items():
    print(f'{strategy}: {details["best_for"][:60]}')

批处理/异步任务中的错误处理与重试逻辑

并发工作负载和批处理工作负载都需要健壮的错误处理。单个请求失败不应导致整个批次崩溃——记录失败、采用退避策略重试，并报告整体成功率。

import asyncio
import random

async def resilient_completion(prompt, max_retries=3, base_delay=1.0):
    for attempt in range(max_retries):
        try:
            return await async_completion(
                [{'role': 'user', 'content': prompt}]
            )
        except openai.RateLimitError:
            wait = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f'Rate limited. Waiting {wait:.1f}s (attempt {attempt+1})')
            await asyncio.sleep(wait)
        except openai.APITimeoutError:
            print(f'Timeout on attempt {attempt+1}')
            await asyncio.sleep(base_delay)
        except openai.APIError as e:
            if e.status_code >= 500:
                await asyncio.sleep(base_delay * (attempt + 1))
            else:
                raise  # Don't retry 4xx errors
    raise RuntimeError(f'Failed after {max_retries} attempts')

async def batch_with_error_reporting(prompts):
    tasks = [resilient_completion(p) for p in prompts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    successes = sum(1 for r in results if not isinstance(r, Exception))
    print(f'Batch complete: {successes}/{len(prompts)} succeeded')
    return results

对大型输入分块以进行批处理

大于模型上下文窗口的文档必须在批处理前进行分块。每个分块都会成为一个独立的批处理请求，之后再合并或总结结果。

def chunk_document(text, max_tokens=3000, overlap_tokens=200):
    '''
    Split a long document into overlapping chunks for batch processing.
    Approximate: 1 token ~ 4 characters
    '''
    max_chars = max_tokens * 4
    overlap_chars = overlap_tokens * 4
    chunks = []
    start = 0
    while start < len(text):
        end = min(start + max_chars, len(text))
        # Try to break at a sentence boundary
        if end < len(text):
            last_period = text.rfind('.', start, end)
            if last_period > start + max_chars // 2:
                end = last_period + 1
        chunks.append({'text': text[start:end], 'start': start, 'end': end})
        start = end - overlap_chars  # overlap for context continuity
    return chunks

def batch_summarize_long_document(document_text, summary_prompt):
    chunks = chunk_document(document_text)
    print(f'Document split into {len(chunks)} chunks')
    # Create one batch request per chunk
    batch_inputs = [
        {'custom_id': f'chunk-{i}',
         'content': summary_prompt + '\n\n' + chunk['text']}
        for i, chunk in enumerate(chunks)
    ]
    # Submit all chunks as one batch job
    return batch_inputs

long_doc = 'Lorem ipsum ' * 5000  # ~20K character document
chunks = chunk_document(long_doc)
print(f'Chunks: {len(chunks)}, first chunk length: {len(chunks[0]["text"])} chars')

大型批次的进度跟踪

对于大型批处理任务（包含数千个请求），应显示实时进度，以便操作人员监控吞吐量并估计完成时间。

import asyncio
import time

async def batch_with_progress(prompts, max_concurrent=20):
    semaphore = asyncio.Semaphore(max_concurrent)
    completed = 0
    total = len(prompts)
    start_time = time.time()
    results = [None] * total

    async def process_one(i, prompt):
        nonlocal completed
        async with semaphore:
            results[i] = await resilient_completion(prompt)
            completed += 1

        elapsed = time.time() - start_time
        rate = completed / elapsed if elapsed > 0 else 0
        eta = (total - completed) / rate if rate > 0 else float('inf')

        if completed % 10 == 0 or completed == total:
            print(f'Progress: {completed}/{total} '
                  f'({completed/total:.0%}) | '
                  f'{rate:.1f} req/s | '
                  f'ETA: {eta:.0f}s')

    await asyncio.gather(*[
        process_one(i, p) for i, p in enumerate(prompts)
    ])
    return results

print('Progress tracking: reports every 10 completions with ETA.')

快速检查

您需要在一夜之间为 10,000 份文档标注情感。您希望将成本降至最低，并且不需要实时结果。哪种方法最合适？

批处理与异步执行总结

批处理和异步执行对于大规模提示词工程至关重要：

OpenAI/Anthropic 批处理接口：成本降低 50%，后台处理，每批最多 5 万个请求
asyncio.gather()：并发执行实时请求，同时发起所有请求并等待所有结果
Semaphore：遵守速率限制的并发控制（通常为 10–50 个并发请求）
指数退避：遇到速率限制或超时错误时，以逐次加倍的延迟进行重试
容错式并发收集：将异常作为返回值可防止单个失败导致整个批次崩溃
进度跟踪：对于大型任务，报告完成数量、处理速率和 ETA

常见问题解答

「批处理与异步执行」课时是免费的吗？

是的 — 「批处理与异步执行」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Prompt Engineering 课程的其余内容，请升级到 CoddyKit PRO。 AI Prompt Engineering 课程共包含 4 节课。

「批处理与异步执行」这节课中我会学到什么？

OpenAI 批处理 API、异步 Python 和并发执行提示词。你通过在浏览器中直接运行的动手代码来练习 AI Prompt Engineering，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Prompt Engineering 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Prompt Engineering 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 2 节课，共 4 节。

「批处理与异步执行」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Prompt Engineering 课中编写并运行代码吗？

能。每节 AI Prompt Engineering 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Prompt Engineering