OpenAI API速率限制管理：理解并优化Run状态轮询机制

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.")