本文旨在解决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调用的稳定性和效率。开发者应持续关注官方文档，并结合实际使用情况调整延迟策略，以实现最佳实践。
python csv ai openai gpt api调用 python脚本 red Python while 循环 线程 并发 对象 异步