如何在不超出Azure事务限制的前提下向Azure自定义视觉API批量发送图片？

看起来你已经尝试了多线程和异步限流两种方式，但还是会遇到超出请求限制的报错，我来帮你分析问题出在哪，以及怎么优化你的方案。

首先，先梳理下你当前遇到的核心问题：

• 多线程方案：靠延迟启动线程控制请求频率，但系统负载波动时，线程实际执行API调用的时间容易偏离预期，高并发时就会“破防”
• 异步+AsyncLimiter方案：理论上每秒限制2次请求，但实际还是有35%的失败率，这背后其实有几个容易忽略的细节

为什么你的异步限流代码会失败？ #

你的代码逻辑本身没大问题，但Azure的限流判断是基于请求到达Azure服务器的UTC时间，而不是你本地代码里记录的时间，这里有几个偏差点：

1. 本地与Azure的时间差：你本地的UTC时间和Azure服务器的UTC时间可能有毫秒级偏差，导致你认为在“下一秒”的请求，实际还是落在上一个限流窗口里
2. 网络延迟的影响：你本地获取限流许可的时间，和请求实际抵达Azure的时间可能有延迟，比如两个请求在本地0.9秒和0.95秒获取许可，实际都在1秒窗口内到达Azure
3. 滑动窗口的计数逻辑：AsyncLimiter的滑动窗口限流是基于本地时间计算的，和Azure服务器的滑动窗口可能不同步，导致同一时间窗口内的请求数超标

优化方案：精准限流+重试兜底 #

最稳妥的方式是结合“本地精准限流”和“429错误自动重试”，同时考虑Azure的实际限流规则，下面是调整后的代码和思路：

1. 优化后的异步限流代码（带重试）

这个版本会：

• 用Azure返回的响应头时间来校准请求到达时间
• 增加本地二次限流检查，确保同一秒内最多2个请求抵达Azure
• 自动处理429错误，根据Azure返回的Retry-After头等待后重试
• 给网络波动留缓冲，调整限流窗口为1.1秒

import asyncio
import aiohttp
import time
import glob
from aiolimiter import AsyncLimiter
import json
from datetime import datetime, timezone
from dateutil.parser import parse  # 需要先安装：pip install python-dateutil

# 加载配置
with open('Config.json','r') as f:
    data = json.load(f)

headers = {
    'Content-Type': 'application/octet-stream',
    'Prediction-Key': data['customvisionPredict_key']
}
predict_url = data['customvision_predict_image']

# 更保守的限流设置：每秒最多2次，窗口放大到1.1秒给网络留缓冲
rate_limiter = AsyncLimiter(max_rate=2, time_period=1.1)
# 维护最近1秒内抵达Azure的请求时间戳（用Azure服务器时间）
recent_azure_times = []

async def upload_image(img_data):
    while True:
        # 先清理超过1秒的旧请求记录
        now_utc = datetime.now(timezone.utc).timestamp()
        recent_azure_times[:] = [t for t in recent_azure_times if t >= now_utc - 1]
        
        # 本地二次检查：确保当前1秒窗口内不超过2个请求
        if len(recent_azure_times) >= 2:
            wait_seconds = (recent_azure_times[0] + 1) - now_utc
            if wait_seconds > 0:
                await asyncio.sleep(wait_seconds)
            continue
        
        async with rate_limiter:
            try:
                async with aiohttp.request('POST', predict_url, data=img_data, headers=headers) as response:
                    # 从响应头获取Azure服务器的UTC时间（最精准的限流依据）
                    azure_date = parse(response.headers.get('Date')).replace(tzinfo=timezone.utc).timestamp()
                    recent_azure_times.append(azure_date)
                    
                    # 处理429限流错误：按Azure提示的时间等待后重试
                    if response.status == 429:
                        retry_after = int(response.headers.get('Retry-After', 1))
                        print(f"触发限流，将在{retry_after}秒后重试")
                        await asyncio.sleep(retry_after)
                        continue
                    
                    return await response.json(), azure_date
            except aiohttp.ClientError as e:
                print(f"请求失败：{e}，1秒后重试")
                await asyncio.sleep(1)
                continue

async def run_batch_upload():
    n = 20
    files = glob.glob("Input/*.png")[:n]
    start_time = time.monotonic()
    
    # 批量创建任务
    tasks = []
    for file_path in files:
        with open(file_path, 'rb') as f:
            tasks.append(upload_image(f.read()))
    
    responses = await asyncio.gather(*tasks)
    
    end_time = time.monotonic()
    # 统计结果
    failure_count = 0
    times = []
    for result, azure_time in sorted(responses, key=lambda x: x[1]):
        times.append(azure_time)
        current_window_count = sum(1 for t in times if t >= azure_time - 1)
        interval = times[-1] - times[-2] if len(times) > 1 else 0
        if 'error' in result:
            failure_count += 1
            print(f"{azure_time:.6f} | 窗口内请求数：{current_window_count} | 间隔：{interval:.6f} | 错误")
        else:
            print(f"{azure_time:.6f} | 窗口内请求数：{current_window_count} | 间隔：{interval:.6f}")
    
    print(f"\n总失败数：{failure_count} / {len(responses)}")
    print(f"总执行时间：{end_time - start_time:.2f} 秒")

if __name__ == "__main__":
    asyncio.run(run_batch_upload())

2. 更省心的替代方案：使用Azure批量预测API

如果你的业务场景允许，优先使用自定义视觉的批量预测API，这比自己控制并发要高效得多：

• 你只需要把所有图片打包上传到Azure Blob存储，然后提交一个批量预测任务
• Azure会异步处理所有图片，不用你自己控制TPS，也不会触发单请求的限流
• 最后你可以直接下载批量预测的结果文件

你可以通过Azure Python SDK调用批量预测，核心逻辑是：

from azure.cognitiveservices.vision.customvision.prediction import CustomVisionPredictionClient
from azure.storage.blob import BlobServiceClient
# 初始化客户端后，提交批量任务
prediction_client = CustomVisionPredictionClient(endpoint, key)
batch_prediction = prediction_client.start_batch_prediction(project_id, published_name, input_blob_uri, output_blob_uri)

最后几个小建议 #

1. 确认Azure自定义视觉的限流规则：免费层和付费层的限制不同，有些是“每秒请求数”，有些是“每分钟请求数”，可以在Azure门户的自定义视觉资源页查看
2. 避免突发流量：即使你设置了每秒2次，也不要一次性提交上百个任务，最好分批提交（比如每1秒提交2个）
3. 监控请求状态：保留失败请求的日志，方便后续排查问题

备注：内容来源于stack exchange，提问作者Faris