OpenAIAPI速率限制优化方法

IT行业相对于一般传统行业，发展更新速度更快，一旦停止了学习，很快就会被行业所淘汰。所以我们需要踏踏实实的不断学习，精进自己的技术，尤其是初学者。今天golang学习网给大家整理了《OpenAI API速率限制：优化Run状态轮询方法》，聊聊，我们一起来看看吧！

在使用OpenAI Assistants API时，因run状态轮询操作被计入API请求速率限制而导致的常见问题。即使在请求间加入固定延迟，用户仍可能遭遇速率限制错误。文章详细分析了问题根源，即client.beta.threads.runs.retrieve调用频繁消耗请求配额，并提供了通过在轮询循环内引入策略性延迟来有效管理和规避速率限制的解决方案，确保API调用的稳定性和可靠性。

理解OpenAI API速率限制机制

OpenAI API为了确保服务的公平性和稳定性，对不同账户和模型设定了严格的速率限制（Rate Limits）。这些限制通常以每分钟请求数（RPM - Requests Per Minute）和每分钟令牌数（TPM - Tokens Per Minute）的形式体现。当您的应用程序在短时间内发出的请求超过了这些预设的限制时，API会返回rate_limit_exceeded错误。

对于许多开发者而言，理解哪些操作会被计入速率限制至关重要。例如，一个常见的误解是，只有显式地创建消息或运行（client.beta.threads.messages.create、client.beta.threads.runs.create）才会计入请求数。然而，实际上，许多辅助性操作，如轮询Run的状态，同样会消耗您的请求配额。

Run状态轮询：隐藏的速率消耗者

在使用OpenAI Assistants API时，一个典型的流程是：

1. 创建文件（client.files.create）
2. 创建消息（client.beta.threads.messages.create）
3. 创建运行（client.beta.threads.runs.create）
4. 轮询运行状态直到完成（client.beta.threads.runs.retrieve）
5. 获取结果（client.beta.threads.messages.list）

问题通常出现在第四步：轮询运行状态。为了确定助手是否已完成其任务，我们需要反复调用client.beta.threads.runs.retrieve来检查run.status。这个retrieve操作本身就是一次API请求，它会实时计入您的速率限制。

考虑以下场景：如果您的速率限制是每分钟3个请求（3 RPM），并且您在每次主请求之间加入了20秒的time.sleep(20)。这看起来足以将主请求频率控制在3 RPM以内。然而，如果在每次主请求的内部，您又在一个while循环中频繁地调用client.beta.threads.runs.retrieve来检查状态，那么这些内部的retrieve调用会迅速耗尽您的请求配额。

例如，一个Run可能需要几秒钟甚至更长时间才能完成。在这期间，while循环可能会每秒钟执行一次retrieve调用。即使一个Run只持续10秒，也可能产生10次额外的API请求。如果您的主请求间隔是20秒，而内部轮询在短时间内产生了大量请求，总请求数很容易超过每分钟3次的限制。这就是为什么有时程序会在处理第一个文件时失败，有时在第三个文件时失败，因为失败的时机取决于Run的实际处理时长以及内部轮询的频率。

优化速率限制管理策略

为了有效管理和规避这种类型的速率限制问题，核心在于控制所有API调用的频率，而不仅仅是主操作。

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

最直接的解决方案是在while循环内部，每次调用client.beta.threads.runs.retrieve之后，也加入一个适当的延迟。这个延迟应该足够长，以确保即使在最坏情况下，内部轮询和外部主请求的总频率也不会超过速率限制。

假设您的速率限制是3 RPM，这意味着平均每20秒才能发出一个请求。如果一个Run平均需要10秒完成，并且您希望在这10秒内只进行少量状态检查，那么每次轮询之间可以设置一个较长的延迟。

示例代码修改：

import pandas as pd
import time
from openai import OpenAI

client = OpenAI(api_key = "[MY API KEY]")
# 建议为每个文件创建一个新的线程，以避免线程内容积累和混淆
# thread = client.beta.threads.create() # 移到循环内部

assistant = client.beta.assistants.create(
    name = "Nomination Hearing Identifier",
    instructions = "Given a complete transcript of a US Senate hearing, determine if this hearing was or was not a nomination hearing. Respond with only 'YES' or 'NO' and do not provide justification.",
    tools = [{"type": "retrieval"}],
    model = "gpt-3.5-turbo-1106"
)

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:
    # 为每个文件创建一个新的线程，确保隔离性
    thread = client.beta.threads.create()

    gpt_file = client.files.create(
        file = open(file, "rb"),
        purpose = 'assistants'
    )

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

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

    # 在这里引入一个更长的初始等待，以避免立即开始频繁轮询
    print(f"Waiting for run {run.id} to complete for file {file}...")
    # time.sleep(5) # 可以在这里加一个初始等待，但更重要的是循环内的等待

    while run.status != "completed":
        # 每次轮询前都进行等待，确保retrieve调用频率受控
        # 假设每次retrieve调用需要至少20秒的间隔来满足3 RPM的限制
        # 如果Run本身很快，可以适当缩短，但要保守估计
        print(f"Run status: {run.status}. Sleeping for 10 seconds before next check.")
        time.sleep(10) # 关键：在每次retrieve调用前等待

        run = client.beta.threads.runs.retrieve(
            thread_id=thread.id,
            run_id=run.id
        )

        if run.status == "failed":
            print(f"Run failed for file {file}: {run.last_error}")
            # 可以在这里添加重试逻辑或跳过当前文件
            break # 跳出当前文件的轮询循环

    if run.status == "completed":
        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) # 使用ignore_index=True

        print(f"Processed file {file}. Result: {output}")
    else:
        print(f"Skipping file {file} due to failed run.")

    # 外部循环的延迟可以根据整体请求频率和模型处理速度调整
    # 如果内部轮询已经有了足够的延迟，这里可以根据需要调整
    print("Sleeping 20 seconds before processing next file to ensure overall API call rate limit not surpassed.")
    time.sleep(20) # 确保下一个文件的初始请求不会立即触发速率限制

jacket_classifications.to_csv("[MY FILE PATH]/test.csv", index=False) # index=False避免写入额外索引列
print("Processing complete. Results saved to CSV.")

代码改进说明：

• 内部轮询延迟： 在while run.status != "completed"循环内部，每次调用client.beta.threads.runs.retrieve之前添加time.sleep(10)。这个值需要根据您的具体速率限制和Run的平均完成时间进行调整。目标是确保retrieve调用的频率低于速率限制。
• 线程管理： 将thread = client.beta.threads.create()移到for循环内部。虽然原始问题不直接与此相关，但在处理多个独立文件时，为每个文件创建新线程是更好的实践，可以避免上下文混淆和潜在的令牌使用问题。
• 错误处理： 增加了对run.status == "failed"的检查，以便在Run失败时能够优雅地处理。
• 字符串比较： if "yes" in output.lower(): 使得判断不区分大小写，更健壮。
• DataFrame拼接： pd.concat时使用ignore_index=True是一个好的实践，可以避免索引重复。
• CSV保存： index=False可以避免将DataFrame的索引写入CSV文件。

2. 指数退避（Exponential Backoff）

对于更健壮的生产系统，建议使用指数退避策略来处理速率限制。当API返回速率限制错误时，不是立即重试，而是等待一个逐渐增长的时间间隔后再重试。许多Python库（如tenacity）都提供了开箱即用的指数退避功能。

3. 监控API使用情况

定期查看OpenAI平台上的账户使用情况和速率限制仪表板（https://platform.openai.com/account/rate-limits）是至关重要的。这能帮助您了解当前的限制，并根据实际使用模式调整您的代码。

总结与最佳实践

• 理解所有API调用： 明确知道哪些操作会计入您的API请求配额，即使是看似辅助性的操作（如状态轮询）。
• 策略性延迟： 在所有可能导致高频率API调用的循环中，尤其是轮询操作，引入适当的time.sleep延迟。
• 指数退避： 在生产环境中，结合指数退避机制来处理临时的速率限制错误，提高应用程序的韧性。
• 监控与调整： 定期检查您的API使用情况，并根据OpenAI的速率限制政策和您的实际需求调整代码中的延迟参数。
• 优化请求设计： 尽量减少不必要的API调用。例如，如果可能，考虑批量处理数据以减少API请求次数，但这在Assistants API的Run模式下可能不直接适用。

通过上述策略，您可以更有效地管理OpenAI API的速率限制，确保您的应用程序在扩展时能够稳定、可靠地运行。

终于介绍完啦！小伙伴们，这篇关于《OpenAIAPI速率限制优化方法》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布文章相关知识，快来关注吧！