如何使用ChatGPT批量总结大量文本？

问题背景 #

现有30万+条文本需要生成单句总结，当前采用循环调用openai.ChatCompletion.create的方式处理，因单条请求叠加网络延迟和API调用开销，速度极慢。需要实现类似openai.Completion的批量处理能力提升效率。

现有低效代码 #

for text_to_summarize in all_text:
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[ 
            {"role": "user", "content": f"Summarize into one sentence: {text_to_summarize}"},
        ],
    )
    summary = response["choices"][0]["message"]["content"] 

两种优化方案 #

方案1：合并文本为单请求批量生成 #

将多条文本打包成一个请求，让模型按指定格式返回对应总结，大幅减少请求次数。

实现代码

import openai

openai.api_key = "你的API密钥"

# 按批次处理，批次大小根据模型token限制调整（gpt-3.5-turbo-16k支持16k token）
batch_size = 20
results = []

for i in range(0, len(all_text), batch_size):
    batch_texts = all_text[i:i+batch_size]
    # 构造批量请求的提示词，明确格式要求
    prompt = "请为以下每条文本生成一句总结，每条总结前对应原文本的编号：\n"
    for idx, text in enumerate(batch_texts, 1):
        prompt += f"{idx}. {text}\n"
    prompt += "请严格按照「编号. 总结内容」的格式返回，不要添加额外说明。"
    
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo-16k",  # 可选16k版本支持更多文本批量处理
        messages=[{"role": "user", "content": prompt}],
        temperature=0  # 降低随机性，保证格式稳定
    )
    
    # 解析返回结果，拆分每条总结
    raw_output = response["choices"][0]["message"]["content"].strip().split("\n")
    for line in raw_output:
        if ". " in line:
            _, summary = line.split(". ", 1)
            results.append(summary)

注意点

• 需根据模型上下文token限制计算批次大小，避免超出上限
• 提示词需明确格式要求，方便后续解析结果
• 建议使用gpt-3.5-turbo-16k提升单批次处理量

方案2：异步并行请求处理 #

通过异步IO并行发送多个请求，同时处理多条文本，大幅缩短总耗时。

实现代码

import asyncio
import aiohttp
import openai

openai.api_key = "你的API密钥"

async def summarize_single(session, text):
    url = "https://api.openai.com/v1/chat/completions"
    headers = {
        "Authorization": f"Bearer {openai.api_key}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "gpt-3.5-turbo",
        "messages": [{"role": "user", "content": f"Summarize into one sentence: {text}"}]
    }
    async with session.post(url, json=payload) as resp:
        result = await resp.json()
        return result["choices"][0]["message"]["content"]

async def batch_summarize(all_text, max_concurrent=50):
    # 控制并发数，避免触发API速率限制
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def bounded_summarize(text):
        async with semaphore:
            return await summarize_single(session, text)
    
    async with aiohttp.ClientSession() as session:
        tasks = [bounded_summarize(text) for text in all_text]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

# 调用方式
# summaries = asyncio.run(batch_summarize(all_text))

注意点

• 设置合理的max_concurrent值，匹配你的OpenAI API速率配额，避免被限流
• 添加return_exceptions=True避免单个请求失败导致整个批量任务中断，后续可单独处理失败的请求
• 建议配合重试机制（如tenacity库）处理网络波动或API临时报错

通用注意事项 #

• 处理30万+条文本时，需分阶段保存结果，避免中途出错丢失进度
• 优先使用方案1减少请求次数，降低API成本；若对实时性要求高，可结合方案2并行处理
• 所有批量操作前，先使用小样本测试格式和兼容性

内容的提问来源于stack exchange，提问作者Penguin