OpenAIAPI限速解析：避开助手请求陷阱

本篇文章向大家介绍《OpenAI API限速解析：避开Assistants请求陷阱》，主要包括，具有一定的参考价值，需要的朋友可以参考一下。

理解OpenAI API限速机制

OpenAI API的限速机制旨在确保服务的公平使用和稳定性。限速通常以每分钟请求数（RPM）和每分钟令牌数（TPM）来衡量。对于新用户或特定模型，限速可能相对较低，例如gpt-3.5-turbo-1106模型可能只有3 RPM的限制。

一个常见的误解是，只有“主要”或“显式”的API调用才会计入限额。然而，所有与API服务器进行的交互都计为一次请求。这意味着，即使是用于检查异步任务状态的轮询调用，也同样会消耗你的请求限额。

案例分析：Assistants API中的隐藏请求

考虑以下使用OpenAI Assistants API处理多个文件的场景。用户希望批量处理10个文本文件，每个文件都通过Assistants API进行分类。为了避免限速，用户在处理完每个文件后，在循环外部设置了20秒的延迟：

import pandas as pd
import time
from openai import OpenAI

# ... (API客户端和助手初始化代码) ...

files = ["file1.txt", "file2.txt", ...] # 假设有10个文件

jacket_classifications = pd.DataFrame(columns = ["jacket", "is_nomination"])

for file in files:
    # 1. 创建文件上传请求
    gpt_file = client.files.create(file=open(file, "rb"), purpose='assistants')

    # 2. 创建消息请求
    message = client.beta.threads.messages.create(
        thread_id=thread.id, role="user", content="...", file_ids=[gpt_file.id]
    )

    # 3. 创建Run请求
    run = client.beta.threads.runs.create(
        thread_id=thread.id, assistant_id=assistant.id
    )

    # 4. 轮询Run状态
    while run.status != "completed":
        run = client.beta.threads.runs.retrieve( # ⚠️ 此处是关键！
            thread_id=thread.id, run_id=run.id
        )
        print(run.status)
        if run.status == "failed":
            print(run.last_error)
            exit()

    # ... (处理结果代码) ...

    print("Sleeping 20 seconds to ensure API call rate limit not surpassed")
    time.sleep(20) # 循环外部的延迟

尽管在每个文件处理周期后有20秒的延迟，用户仍然频繁遇到rate_limit_exceeded错误。错误信息明确指出“Rate limit reached for gpt-3.5-turbo-1106 ... on requests per min (RPM): Limit 3, Used 3, Requested 1.”，这表明在某个1分钟窗口内，API请求数超过了3次。

问题根源在于：while run.status != "completed" 循环内部的 client.beta.threads.runs.retrieve() 调用。 每次循环迭代都会向OpenAI API发送一个请求，以检查Run的最新状态。如果Run的执行时间较长，或者代码执行速度过快，这个循环会在短时间内发出大量的retrieve请求。

例如，在一个文件处理周期内：

• client.files.create()：1次请求
• client.beta.threads.messages.create()：1次请求
• client.beta.threads.runs.create()：1次请求
• client.beta.threads.runs.retrieve()：N次请求（N取决于Run的执行时间）

即使每次文件处理之间有20秒的延迟，如果N次retrieve请求在几秒内完成，那么在1分钟内，很容易就会累积超过3次请求，从而触发限速。

解决方案与优化策略

解决此问题的关键在于，不仅要控制“主”操作之间的间隔，还要控制异步任务轮询的频率。

1. 在轮询循环中引入延迟

最直接的解决方案是在 while 循环内部，每次 run.retrieve() 调用之后添加一个延迟。这将显著降低轮询频率，从而减少在给定时间内发出的API请求总数。

import pandas as pd
import time
from openai import OpenAI

# ... (API客户端和助手初始化代码) ...

files = ["file1.txt", "file2.txt", ...]

jacket_classifications = pd.DataFrame(columns = ["jacket", "is_nomination"])

for file in files:
    gpt_file = client.files.create(file=open(file, "rb"), purpose='assistants')
    message = client.beta.threads.messages.create(
        thread_id=thread.id, role="user", content="...", file_ids=[gpt_file.id]
    )
    run = client.beta.threads.runs.create(
        thread_id=thread.id, assistant_id=assistant.id
    )

    # 轮询Run状态，并在每次轮询后增加延迟
    while run.status != "completed":
        run = client.beta.threads.runs.retrieve(
            thread_id=thread.id, run_id=run.id
        )
        print(run.status)
        if run.status == "failed":
            print(run.last_error)
            exit()

        # ⚠️ 在轮询请求后增加延迟
        # 假设Run通常在几十秒内完成，每次轮询间隔40秒可以有效控制请求频率
        time.sleep(40) 

    # ... (处理结果代码) ...

    # 外部循环的延迟可以根据总请求量和限速进一步调整，甚至可以移除
    # print("Sleeping 20 seconds to ensure API call rate limit not surpassed")
    # time.sleep(20)

通过在 while 循环内部添加 time.sleep(40)，每次 retrieve 请求之间至少间隔40秒。结合一个文件处理周期中其他3个请求，如果Run通常在1-2次轮询内完成，那么处理一个文件可能总共发出 3（创建）+ 1-2（轮询）= 4-5个请求。如果每个文件处理间隔较长，或者总处理时间较长，就能有效避免限速。

2. 考虑更健壮的重试机制：指数退避

对于生产环境或更复杂的应用，仅仅依靠固定的 time.sleep() 可能不够灵活。指数退避（Exponential Backoff） 是一种更推荐的重试策略，它在每次重试失败后，逐渐增加等待时间。这不仅有助于遵守速率限制，还能优雅地处理临时的API服务中断。

Python库如 tenacity 或 backoff 可以轻松实现指数退避：

import time
from tenacity import retry, wait_exponential, stop_after_attempt, RetriableError
from openai import OpenAI

# ... (API客户端和助手初始化代码) ...

# 定义一个带有指数退避的重试函数
@retry(wait=wait_exponential(multiplier=1, min=4, max=60), stop=stop_after_attempt(10))
def call_openai_api_with_retry(api_call_func, *args, **kwargs):
    try:
        return api_call_func(*args, **kwargs)
    except Exception as e: # 捕获OpenAI API可能抛出的限速或其他错误
        print(f"API call failed, retrying... Error: {e}")
        raise RetriableError(e) # 抛出可重试错误，让tenacity捕获

# 在轮询Run状态时使用重试机制
def get_run_status_with_backoff(thread_id, run_id):
    while True:
        try:
            run = call_openai_api_with_retry(client.beta.threads.runs.retrieve, thread_id=thread_id, run_id=run_id)
            if run.status != "completed":
                print(f"Run status: {run.status}. Waiting before next check...")
                # 在轮询之间仍然可以有基础的延迟，防止过于频繁的重试
                time.sleep(5) 
            else:
                return run
        except RetriableError:
            # tenacity 会处理重试逻辑，这里可以记录日志
            print("Encountered retriable error, tenacity will handle backoff.")
            time.sleep(1) # 短暂等待，避免无限循环的日志输出
        except Exception as e:
            print(f"An unrecoverable error occurred: {e}")
            break

# ... (在主循环中使用) ...
# run = get_run_status_with_backoff(thread.id, run.id)

3. 异步处理与Webhook（高级）

对于需要处理大量请求且对延迟敏感的场景，可以考虑使用异步编程结合Webhook。当Run完成时，OpenAI API可以向你的服务器发送一个通知，而不是你持续轮询。这可以极大地减少API请求数量，但需要更复杂的架构来接收和处理Webhook。

注意事项

• 理解不同模型的限速： 不同的OpenAI模型（如GPT-3.5 Turbo、GPT-4）和不同的账户级别（免费、付费、企业）都有不同的速率限制。务必查阅OpenAI官方文档中关于你所使用模型和账户的最新限速信息。
• 监控API使用情况： OpenAI平台提供了API使用情况仪表板，你可以通过它实时监控你的请求量和令牌使用情况，帮助你更好地理解和调整你的调用策略。
• 考虑请求并发性： 如果你的应用是多线程或多进程的，每个线程/进程都会独立地向API发送请求，这会更快地触及限速。在这种情况下，需要一个全局的限速器来协调所有请求。
• API文档是你的朋友： 仔细阅读OpenAI的API文档，特别是关于限速和异步操作的部分，可以帮助你避免许多常见问题。

总结

在使用OpenAI Assistants API时，避免速率限制错误的关键在于对所有API调用的全面理解，包括那些用于轮询异步任务状态的“隐藏”请求。通过在轮询循环中引入适当的延迟，或采用更高级的指数退避策略，可以有效管理API请求频率，确保应用稳定运行并遵守API使用政策。对API行为的深入洞察和代码的细致优化，是构建健壮、高效AI应用的基础。

理论要掌握，实操不能落！以上关于《OpenAIAPI限速解析：避开助手请求陷阱》的详细介绍，大家都掌握了吧！如果想要继续提升自己的能力，那么就来关注golang学习网公众号吧！