用 Colibrì 批量生成营销文案，中断通常不在模型侧，而在调用侧的批次管理和失败恢复。几百条文案一次跑完不现实，正确做法是把单条请求变成可重放的最小任务，再分批执行、失败重试、结果去重。

批量调用 Colibrì 时，先确认单条请求的字段和返回结构，再将输入文件按固定批次拆分循环调用；每次请求设置超时并用 try-except 捕获异常，失败后按指数退避重试，最终按 request_id 去重落盘。适用于本地或内网的批量生成任务。操作上先拿 100 条样例跑批，核对成功与失败总数是否等于输入总数。风险边界是模型输出格式、接口限流策略和鉴权方式需结合实际环境确认。

确认 Colibrì 的输入输出格式

Colibrì 的接入方式可能是 HTTP API，也可能是本地 CLI，但都需要先确认一条请求的字段和返回结构。不要直接写批量循环，先拿一条营销文案做最小调用，打印原始返回。下面是一个通用请求骨架，endpoint 和鉴权头需要替换成实际配置：

{
  "model": "colibri-text",
  "input": "产品：降噪耳机；卖点：通勤时听清语音；语气：简洁",
  "max_tokens": 80,
  "temperature": 0.7,
  "request_id": "req_0001"
}

返回结构不能假设。常见的可能是 output.text，也可能是 choices[0].message.content。拿到结果后先检查状态字段、输出字段、请求 ID 是否完整。通过实际运行验证后再进入批量环节。

拆分输入文件并按批次循环调用

输入文件建议按行组织，每行一条 JSON，字段里包含文案主题、产品卖点、目标语气。这样方便按批次读取，也方便失败后定位哪一条请求出了问题。批次大小从 10 开始，跑通后再根据单批耗时调整；批次太大会让单次循环占用太久，批次太小会增加请求总量。

import json

batch_size = 10
req_id = 0

with open("inputs.jsonl", "r", encoding="utf-8") as f:
    batch = []
    for line in f:
        line = line.strip()
        if not line:
            continue
        item = json.loads(line)
        item["request_id"] = f"req_{req_id:04d}"
        batch.append(item)
        req_id += 1
        if len(batch) >= batch_size:
            process_batch(batch)
            batch = []
    if batch:
        process_batch(batch)

如果输入是 CSV，先用 csv.DictReader 逐行读取，不要用 readlines 一次性全部加载。process_batch 和 call_colibri 在后面的小节定义。

为单次调用设置超时与异常捕获

一次请求可能卡住，也可能返回非预期结构。给每次调用设置超时时间，例如 15 秒，并用 try-except 捕获异常。脚本要明确区分连接失败、超时、解析失败和业务错误，因为不同类型的失败重试策略不同。

import time
import traceback

def call_colibri(item, timeout=15):
    # HTTP 请求或 CLI 调用的通用入口
    # 返回解析后的 JSON
    pass

def process_batch(batch):
    for item in batch:
        try:
            resp = call_colibri(item)
            save_result(item, resp)
        except Exception as exc:
            log_failure(item, exc)
        time.sleep(0.2)

def save_result(item, resp):
    # 按行追加写入 results.jsonl
    pass

日志字段要足够定位问题：request_id、输入片段哈希、异常类型、异常信息、重试次数、本地时间戳。建议写入 logs/colibri_failures.log，一行一条 JSON，而不是只 print。这样失败记录可以被后续脚本读取和重放。

{"request_id":"req_0003","source_hash":"ab12...","error_type":"Timeout","retry_count":0,"timestamp":"2026-01-01T00:00:00Z"}

加入指数退避重试并去重结果

临时性错误不应当直接判失败。重试时用指数退避：第一次等待 2 秒，第二次 4 秒，第三次 8 秒，并加上少量随机抖动，避免多个请求同时重试撞车。最大重试次数建议设为 4 次，超过后写入 failures.jsonl。

import random
import time

def call_with_retry(item, max_retries=4):
    for attempt in range(max_retries):
        try:
            return call_colibri(item, timeout=15)
        except Exception as exc:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt + random.uniform(0, 1)
            time.sleep(wait)
            item["retry_count"] = attempt + 1

去重逻辑要同时处理两种情况：脚本中断后重跑，以及请求超时但 Colibrì 已经生成内容。推荐在生成 request_id 时使用输入内容的 SHA-256 摘要，而不是顺序编号。这样同一输入重试后仍对应同一行结果，保存时用 request_id 作为主键，后写入的覆盖先写入的，避免重复文案堆积。

def load_existing_ids(path="results.jsonl"):
    seen = set()
    try:
        with open(path, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                seen.add(json.loads(line)["request_id"])
    except FileNotFoundError:
        pass
    return seen

用样例集跑批并核对输出数量

正式跑全量数据前，先准备一个 100 条的样例集，里面混入超长标题、空字段、特殊符号，避免只用理想样例。脚本运行结束后，核对四个数量：输入文件总条数、results.jsonl 总行数、failures.jsonl 总行数、唯一 request_id 数。

wc -l inputs.jsonl results.jsonl failures.jsonl
grep -o '"request_id":"[^"]*"' results.jsonl | sort -u | wc -l

如果唯一 request_id 数等于输入总数，且 failures.jsonl 中只有最终失败项，说明本轮跑批覆盖了全部输入。如果成功与失败条目之和小于输入总数，说明有请求既没有结果也没有失败记录，需要检查脚本是否在调用前中断，或者 request_id 在生成后没有保存。这个步骤把整条流程从不可见变成可核对，后续再调批次大小和超时时间就有依据了。