处理大量图片时，逐个调用 LingBot-Vision 接口生成描述确实低效，而且一旦网络抖动或接口限流，中间失败的数据很容易丢。建议先做一个带唯一图片 ID 的输入清单，再用线程池控制并发，把每次请求的结果和失败项都落到本地 JSON 或 CSV 里，这样整批任务即使中断，也能从上次断点继续跑。下面给出一个不绑定具体 SDK 的 Python 脚本骨架，所有接口地址、模型名和密钥都留成可替换配置。
适用场景：本地或服务器上有一批图片，需要调用 LingBot-Vision 这类 OpenAI 兼容的视觉接口生成描述。操作动作：先整理图片清单，构造 base64 或文件 URL 的请求，用线程池限制并发，并对失败项做带退避的重试。验证方式：先跑 3-5 张图片确认描述内容正确，再放开全量任务；中途中断后重新执行脚本，通过已处理记录跳过已完成项。风险边界：图片过大或并发过高可能导致接口超时，重试次数和最大并发数需要结合具体接口限制调整，不能当作无限提速的手段。
读取待处理图片列表与元数据
批量处理的第一步不是调用接口，而是把输入整理成可追踪的清单。每张图片需要一个稳定的唯一标识，建议用相对路径或文件哈希，避免重名文件互相覆盖。下面的函数会遍历指定目录，过滤出 .jpg、.jpeg、.png、.webp 等常见格式，输出包含 id 和 path 的字典列表。
import os
import hashlib
IMAGE_EXTS = {'.jpg', '.jpeg', '.png', '.webp', '.bmp'}
def load_images(image_dir):
items = []
for root, _, files in os.walk(image_dir):
for name in sorted(files):
ext = os.path.splitext(name)[1].lower()
if ext not in IMAGE_EXTS:
continue
path = os.path.join(root, name)
# 用相对路径做 ID，保证唯一且可读
rel = os.path.relpath(path, image_dir)
with open(path, 'rb') as f:
digest = hashlib.md5(f.read()).hexdigest()[:8]
items.append({'id': f'{rel}:{digest}', 'path': path})
return items
if __name__ == '__main__':
for item in load_images('./images'):
print(item['id'], item['path'])这里把文件内容哈希拼进 ID，主要是防止相同路径下图片被替换后结果仍然沿用旧描述。如果图片目录很大，计算 MD5 会占用额外时间，你可以只保留相对路径；但需要确认目录内不会出现同名覆盖的情况。
构造图像描述请求
LingBot-Vision 如果兼容 OpenAI 的 Chat Completions 接口，那么请求体通常包含 model、messages 和 max_tokens。图片有两种常见传法：一是将图片转成 base64 放在 image_url 的 data URL 里；二是传一个可访问的文件 URL。base64 方式适合本地文件，但会增加请求体积；文件 URL 方式更适合图片已上传到对象存储的场景。下面给出可替换的请求构造函数：
import base64
import requests
def encode_image_base64(path):
with open(path, 'rb') as f:
return base64.b64encode(f.read()).decode('utf-8')
def build_payload(image_path, prompt='请详细描述这张图片的内容', model='lingbot-vision'):
b64 = encode_image_base64(image_path)
return {
'model': model,
'messages': [
{
'role': 'user',
'content': [
{'type': 'text', 'text': prompt},
{
'type': 'image_url',
'image_url': {
'url': f'data:image/jpeg;base64,{b64}'
}
}
]
}
],
'max_tokens': 500
}需要注意：image_url.url 里是否要加 detail 参数，取决于具体接口实现。有些接口要求 base64 字符串不带 data:image/jpeg;base64, 前缀，有些则要求必须带。建议先用一张小图测试接口返回，再按实际要求调整。如果图片是 PNG 或 WebP，需要把 data URL 里的 MIME 类型对应改成 image/png 或 image/webp。
并发调用与限流控制
并发能明显缩短总耗时，但盲目提高并发数容易触发接口限流或内存溢出。稳妥的做法是先设置一个较小的最大并发数（比如 4），跑通后再逐步调大。下面使用 ThreadPoolExecutor 实现并发，同时把请求封装成带重试的函数：遇到网络错误或 429/5xx 时，按指数退避等待后重试。
import time
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
MAX_WORKERS = 4
MAX_RETRIES = 3
API_URL = 'https://your-api-endpoint/v1/chat/completions'
API_KEY = 'your-api-key'
def call_vision_api(payload):
headers = {'Authorization': f'Bearer {API_KEY}'}
for attempt in range(MAX_RETRIES):
try:
resp = requests.post(API_URL, json=payload, headers=headers, timeout=60)
if resp.status_code == 200:
return resp.json()
# 429 限流或 5xx 服务端错误时重试
if resp.status_code in (429, 500, 502, 503):
wait = 2 ** attempt
time.sleep(wait)
continue
resp.raise_for_status()
except requests.RequestException as e:
if attempt == MAX_RETRIES - 1:
raise
time.sleep(2 ** attempt)
raise RuntimeError(f'Failed after {MAX_RETRIES} retries')
def process_one(item, prompt):
payload = build_payload(item['path'], prompt)
result = call_vision_api(payload)
try:
description = result['choices'][0]['message']['content']
except (KeyError, IndexError):
raise ValueError(f'Unexpected response: {result}')
return {'id': item['id'], 'path': item['path'], 'description': description}主控循环里，可以先用一个小列表跑通，再跑全量。下面的示例会遍历输入项，把每个任务提交给线程池，并实时打印完成状态：
def run_batch(items, prompt):
results = []
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
future_map = {executor.submit(process_one, item, prompt): item for item in items}
for future in as_completed(future_map):
item = future_map[future]
try:
result = future.result()
results.append(result)
print(f'OK: {item["id"]}')
except Exception as e:
print(f'FAIL: {item["id"]} - {e}')
return results这里把失败打印出来，但还没做持久化。真正的批处理需要把成功和失败分开记录，否则脚本一退出，内存里的结果就丢了。
结果持久化与异常重试
整批任务执行过程中，网络闪断、单张图片过大都可能导致个别请求失败。如果脚本直接退出，重新跑全量会浪费之前成功的部分。所以建议维护两个文件：一个是结果 JSON（或 CSV），一个是断点记录文件，里面保存已成功处理的图片 ID。每次成功后立刻追加写入，这样中断后再次启动时，直接跳过已完成项。
import json
import os
OUTPUT_JSON = 'results.json'
DONE_IDS_FILE = 'done_ids.json'
def load_done_ids():
if os.path.exists(DONE_IDS_FILE):
with open(DONE_IDS_FILE, 'r', encoding='utf-8') as f:
return set(json.load(f))
return set()
def append_result(result):
records = []
if os.path.exists(OUTPUT_JSON):
with open(OUTPUT_JSON, 'r', encoding='utf-8') as f:
records = json.load(f)
records.append(result)
with open(OUTPUT_JSON, 'w', encoding='utf-8') as f:
json.dump(records, f, ensure_ascii=False, indent=2)
def mark_done(item_id):
done = load_done_ids()
done.add(item_id)
with open(DONE_IDS_FILE, 'w', encoding='utf-8') as f:
json.dump(sorted(done), f, ensure_ascii=False)
def run_batch_resumable(items, prompt):
done_ids = load_done_ids()
pending = [item for item in items if item['id'] not in done_ids]
print(f'Total: {len(items)}, skipped: {len(items) - len(pending)}')
for item in pending:
try:
result = process_one(item, prompt)
append_result(result)
mark_done(item['id'])
print(f'OK: {item["id"]}')
except Exception as e:
# 记录失败项到单独文件，也可以直接打印后继续
print(f'FAIL: {item["id"]} - {e}')
with open('failed_ids.txt', 'a', encoding='utf-8') as f:
f.write(item['id'] + '\n')这个版本是串行循环，便于理解断点逻辑。如果希望同时保留并发和断点，可以把 run_batch_resumable 里的循环改成线程池提交，但每个任务完成后都要调用 append_result 和 mark_done。需要注意：多线程同时写 JSON 文件会互相覆盖，所以需要给文件写入加锁，或者改用 SQLite 这类支持并发的存储。
断点续跑的验证方式很简单：跑完一部分后直接 Ctrl+C 中断，再重新执行脚本，观察是否跳过已完成项。重试策略和并发数需要结合具体接口的限流阈值调整，没有统一最优值；建议先把 MAX_WORKERS 设为 2-4，确认稳定后再逐步增加。如果能拿到接口的速率限制响应头，可以据此动态调整退避时间，但那些数字需要在你自己的环境里实测确认，不建议照搬别人的经验值。