如何在 Python 中异步调用 OpenAI 的 API？

要在 Python 中异步调用 OpenAI 的 API，您需要使用 asyncio 库和官方的 openai 包，该包支持异步操作。 首先，确保您已安装最新的 OpenAI 库 (pip install openai)。该库提供了一个 AsyncOpenAI 客户端，专为非阻塞请求而设计，使您的代码能够并发处理多个 API 调用，而无需等待每个调用完成才开始下一个调用。 这种方法非常适合聊天机器人或批量处理等需要效率和响应速度的应用程序。

首先导入 asyncio 并使用您的 API 密钥初始化异步客户端。 这是一个基本示例

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI(api_key="your-api-key")

async def get_completion(prompt):
 response = await client.chat.completions.create(
 model="gpt-3.5-turbo",
 messages=[{"role": "user", "content": prompt}]
 )
 return response.choices[0].message.content

async def main():
 result = await get_completion("Explain async programming in Python.")
 print(result)

asyncio.run(main())

此代码定义了一个异步函数 get_completion，该函数使用 await 发送请求，从而在等待 API 响应时释放事件循环以处理其他任务。 main 函数等待结果并将其打印出来。 对于错误处理，请将 API 调用包装在 try/except 块中，以捕获网络错误或速率限制等异常。

对于批量或并行请求，请使用 asyncio.gather() 一次运行多个调用。 例如

async def main():
 prompts = ["Prompt 1", "Prompt 2", "Prompt 3"]
 tasks = [get_completion(prompt) for prompt in prompts]
 results = await asyncio.gather(*tasks)
 for res in results:
 print(res)

与顺序同步调用相比，这减少了总执行时间。 请注意，异步代码需要仔细管理 API 速率限制等资源——避免使用过多的并发请求来压垮 API。 如果需要，使用信号量或批量延迟。 总的来说，使用官方库可以轻松实现与 OpenAI API 的异步集成，从而轻松构建可扩展的应用程序。