OpenAIAPI速率限制优化技巧

小伙伴们对文章编程感兴趣吗？是否正在学习相关知识点？如果是，那么本文《优化OpenAI助手API速率限制方法》，就很适合你，本篇文章讲解的知识点主要包括。在之后的文章中也会多多分享相关知识点，希望对大家的知识积累有所帮助！

本文旨在解决OpenAI Assistants API中常见的速率限制错误，尤其是在用户认为已正确实施延迟策略时仍遭遇限制的问题。核心洞察在于，不仅是创建运行（run）的API调用，其后续状态检索（retrieve run）操作也计入速率限制。教程将深入分析这一机制，提供包含代码示例的有效解决方案，并探讨更高级的速率限制管理策略，以确保API调用的稳定性和效率。

1. 理解OpenAI API的速率限制机制

在使用OpenAI API，特别是其Assistant API时，开发者经常会遇到rate_limit_exceeded错误。这通常意味着在特定时间窗口内，您的API请求数量超过了账户或模型所允许的上限。OpenAI为不同账户级别和模型设定了不同的速率限制，通常以每分钟请求数（RPM）和每分钟令牌数（TPM）来衡量。

对于许多新用户而言，一个常见的误解是，只有“主要”的操作（例如创建文件、创建消息或创建运行）才会计入速率限制。然而，实际上，每一次与OpenAI服务器的交互，即每一次API调用，都会被计入您的速率限制。这包括但不限于：

• client.files.create()：上传文件
• client.beta.threads.messages.create()：向线程添加消息
• client.beta.threads.runs.create()：启动一个运行
• client.beta.threads.runs.retrieve()：检索一个运行的状态
• client.beta.threads.messages.list()：列出线程中的消息

特别是对于Assistant API，run对象的生命周期管理是导致速率限制问题的常见根源。一个run从创建到完成可能需要一定时间，在此期间，开发者通常会通过循环调用client.beta.threads.runs.retrieve()来轮询其状态。这些频繁的状态检索调用，如果不加以控制，会迅速耗尽您的速率限制。

2. 识别隐藏的API请求：Run状态检索

考虑以下使用OpenAI Assistant API处理文本文件的Python脚本片段：

import pandas as pd
import time
from openai import OpenAI

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

files = ["CHRG-108shrg1910401.txt", ...] # 假设有10个文件

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,
    )

    while run.status != "completed":
        run = client.beta.threads.runs.retrieve( # <-- 这是一个API请求！
        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) # <-- 这个延迟是在文件处理之间

在这个示例中，开发者在处理完一个文件并进入下一个文件之前，添加了一个time.sleep(20)的延迟。然而，即使有了这个延迟，仍然可能遇到速率限制错误。原因在于，while run.status != "completed"循环内部的client.beta.threads.runs.retrieve()调用。如果一个run需要较长时间才能完成，这个循环可能会在短时间内执行多次retrieve操作，每次操作都计入速率限制。

假设您的速率限制是3 RPM，而一个run需要10秒才能完成。如果retrieve操作每秒执行一次，那么在一个run的生命周期内，就会产生10个API请求。再加上文件上传、消息创建和运行创建等操作，很容易就会超过3 RPM的限制。

3. 解决方案：策略性地管理API请求延迟

为了有效解决这个问题，我们需要在所有可能触发API请求的地方考虑延迟，尤其是在频繁轮询状态的循环中。

3.1. 方案一：在轮询循环内增加延迟

最直接的解决方案是在run状态检索的循环内部添加一个延迟。这样可以确保每次状态检查之间有足够的时间间隔，从而避免短时间内产生过多的retrieve请求。

import time
from openai import OpenAI
import pandas as pd # 确保导入所有需要的库

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

files = ["CHRG-108shrg1910401.txt","CHRG-108shrg1910403.txt", "CHRG-108shrg1910406.txt", "CHRG-108shrg1910407.txt", "CHRG-108shrg1910408.txt", "CHRG-108shrg1910409.txt", "CHRG-108shrg1910410.txt", "CHRG-108shrg1910411.txt", "CHRG-108shrg1910413.txt", "CHRG-108shrg1910414.txt"]
jacket_classifications = pd.DataFrame(columns = ["jacket", "is_nomination"])

for file in files:
    # 每次文件处理前，确保有足够的间隔
    # 如果上一个文件的处理（包括轮询）可能导致接近限额，这里可以放置一个更长的初始延迟
    # 或者，更推荐的是在每次API调用后都进行检查和延迟

    gpt_file = client.files.create(
        file = open(file, "rb"),
        purpose = 'assistants'
    )
    # 考虑在这里也添加一个小的延迟，如果文件上传也是一个高频操作
    # time.sleep(1) 

    message = client.beta.threads.messages.create(
        thread_id=thread.id,
        role="user",
        content="Determine if the transcript in this file does or does not describe a nomination hearing. Respond with only 'YES' or 'NO' and do not provide justification.",
        file_ids=[gpt_file.id]
    )
    # time.sleep(1)

    run = client.beta.threads.runs.create(
        thread_id=thread.id,
        assistant_id=assistant.id,
    )
    # time.sleep(1)

    # 关键改进：在轮询循环内部添加延迟
    while run.status != "completed":
        # 每次检索前等待，以避免短时间内的连续请求
        time.sleep(5) # 例如，每5秒检查一次，具体值根据您的速率限制和run的平均完成时间调整
        run = client.beta.threads.runs.retrieve(
            thread_id=thread.id,
            run_id=run.id
        )
        print(f"Run status: {run.status}")
        if run.status == "failed":
            print(f"Run failed: {run.last_error}")
            exit()
        elif run.status == "expired": # 增加对过期状态的处理
            print(f"Run expired: {run.last_error}")
            # 可以选择重新创建run或跳过当前文件
            exit()

    messages = client.beta.threads.messages.list(
        thread_id=thread.id
    )
    # time.sleep(1)

    output = messages.data[0].content[0].text.value
    is_nomination = 0 # 默认值
    if "yes" in output.lower(): # 统一转换为小写进行判断
        is_nomination = 1

    row = pd.DataFrame({"jacket":[file], "is_nomination":[is_nomination]})
    jacket_classifications = pd.concat([jacket_classifications, row], ignore_index=True) # 使用ignore_index=True避免索引问题

    print(f"Finished processing {file}. Preparing for next file.")
    # 如果所有API调用（包括轮询）的总时长接近您的RPM限制，这里可能还需要额外的延迟
    # 例如，如果您的限制是3 RPM，那么平均每次请求之间需要20秒。
    # 假设一个文件处理过程中总共产生了N个API请求，那么您可能需要 N * 20秒 / N = 20秒的平均间隔。
    # 更准确的做法是计算一个文件处理的总请求数，然后确保在整个循环中平均分配这些请求的延迟。
    # 简单粗暴但有效的方法是确保一个完整的文件处理流程（包括所有API调用）的总时长大于 (总请求数 / RPM限制)
    # 例如，如果每个文件处理大约产生 5 个API请求 (create_file, create_message, create_run, N*retrieve_run, list_messages)
    # 且限制是 3 RPM，那么 5 个请求需要 5/3 分钟 = 100秒。
    # 如果您的轮询延迟已经足够，这里的`time.sleep(20)`可能就不需要，或者需要根据实际情况调整。
    # 最佳实践是结合指数退避策略。

jacket_classifications.to_csv("[MY FILE PATH]/test.csv", index=False) # index=False避免写入DataFrame索引

注意事项：

• time.sleep(5)：这个值需要根据您的具体速率限制（RPM）和run的平均完成时间进行调整。如果run通常很快完成，可以适当缩短；如果run耗时较长，可以适当延长。
• 总请求数：请记住，每个文件处理循环中，所有的API调用（创建文件、创建消息、创建运行、多次检索运行状态、列出消息）都会计入速率限制。您需要确保所有这些请求的总频率不超过限制。

3.2. 方案二：指数退避（Exponential Backoff）

对于更健壮和高效的API交互，推荐使用指数退避策略。这种方法在遇到API错误（如速率限制）时，会等待一个逐渐增长的时间间隔后重试。对于run状态轮询，也可以应用类似的逻辑，即在每次检索之间逐渐增加延迟，直到run完成。

import time
from openai import OpenAI
import pandas as pd
import random

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

files = ["CHRG-108shrg1910401.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)

    # 关键改进：指数退避轮询
    max_retries = 10 # 最大重试次数
    initial_delay = 5 # 初始延迟秒数
    for i in range(max_retries):
        if run.status == "completed":
            break

        # 增加随机抖动，避免所有客户端在同一时间重试
        sleep_time = initial_delay * (2 ** i) + random.uniform(0, 2) 
        print(f"Run status: {run.status}. Waiting for {sleep_time:.2f} seconds before next check.")
        time.sleep(min(sleep_time, 60)) # 设置最大等待时间，防止无限增长

        try:
            run = client.beta.threads.runs.retrieve(
                thread_id=thread.id,
                run_id=run.id
            )
        except OpenAI.APIRateLimitError as e:
            print(f"Rate limit hit during retrieval. Retrying with increased delay. Error: {e}")
            # 如果在检索时也遇到速率限制，这里可以进一步增加延迟或退出
            time.sleep(initial_delay * (2 ** i) * 2) # 更长的延迟
            continue # 继续循环，再次尝试检索

        if run.status == "failed":
            print(f"Run failed: {run.last_error}")
            exit()
        elif run.status == "expired":
            print(f"Run expired: {run.last_error}")
            exit()
    else: # 如果循环结束但run未完成
        print(f"Run did not complete after {max_retries} retries. Last status: {run.status}")
        # 可以选择跳过当前文件或记录错误

    # ... (处理输出和数据存储代码不变) ...
    messages = client.beta.threads.messages.list(thread_id=thread.id)
    output = messages.data[0].content[0].text.value
    is_nomination = 0
    if "yes" in output.lower():
        is_nomination = 1
    row = pd.DataFrame({"jacket":[file], "is_nomination":[is_nomination]})
    jacket_classifications = pd.concat([jacket_classifications, row], ignore_index=True)

    print(f"Finished processing {file}. Preparing for next file.")
    # 在处理下一个文件前，可以根据需要添加一个较长的固定延迟，以确保整体RPM符合要求
    # time.sleep(20)

jacket_classifications.to_csv("[MY FILE PATH]/test.csv", index=False)

指数退避的优势：

• 适应性强： 能够根据run的实际完成时间动态调整轮询频率。
• 减少不必要的请求： 在run完成之前，逐渐拉长轮询间隔，减少API调用次数。
• 鲁棒性： 在遇到临时性错误（如速率限制）时能够自动重试，提高程序的稳定性。

4. 最佳实践与注意事项

• 查阅官方文档： 始终以OpenAI官方文档中关于速率限制的最新信息为准。限制可能会根据模型、账户类型和使用情况而变化。
• 监控API使用情况： 在OpenAI平台仪表盘中监控您的API使用情况和速率限制，这有助于您了解实际的请求模式。
• 错误处理： 除了速率限制错误，还应处理其他API错误（如认证错误、服务器错误等），使您的脚本更健壮。
• 异步处理： 对于高吞吐量的应用，可以考虑使用异步编程（如asyncio）来并发处理多个run，同时仍需精心管理每个run的轮询频率和总体的API请求速率。
• 批量处理： 如果可能，尽量将多个任务打包成更少的API请求，例如，如果Assistant API支持，可以考虑一次性上传多个文件或处理多个请求。
• 升级账户： 如果您的业务需求持续超出当前速率限制，考虑升级您的OpenAI账户或联系OpenAI支持以申请更高的速率限制。

总结

解决OpenAI Assistants API的速率限制问题，关键在于全面理解所有API调用（包括run状态检索）都会计入限制。通过在run状态轮询循环内部策略性地添加延迟，或采用更高级的指数退避策略，可以有效控制API请求频率，避免rate_limit_exceeded错误，从而确保API调用的稳定性和效率。开发者应持续关注官方文档，并结合实际使用情况调整延迟策略，以实现最佳实践。

本篇关于《OpenAIAPI速率限制优化技巧》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于文章的相关知识，请关注golang学习网公众号！