异步调用OpenAI Embedding接口遇APIConnectionError问题求助

1. 检查并升级OpenAI库版本 #

如果你的OpenAI库版本是v1.x之前的旧版本，异步接口的用法可能存在兼容性问题。建议升级到最新稳定版：

pip install --upgrade openai

升级后使用新版异步客户端写法：

from openai import AsyncOpenAI
import asyncio

async def get_openai_api_key():
    # 替换为你的密钥获取逻辑（如从环境变量/配置文件读取）
    return "your-secret-key"

async def embeddings_acreate(input: list[str]):
    client = AsyncOpenAI(
        api_key=await get_openai_api_key(),
        timeout=60,
        # 若需代理，添加以下配置
        # proxy="http://your-proxy-address:port"
    )
    return await client.embeddings.create(
        model='text-embedding-ada-002',
        input=input
    )

# 测试调用示例
async def main():
    result = await embeddings_acreate(["tell me your name"])
    print(result.data)

asyncio.run(main())

2. 验证API密钥的正确性 #

确保get_openai_api_key()返回的密钥与同步代码中使用的完全一致，可临时添加打印验证：

async def embeddings_acreate(input: list[str]):
    api_key = await get_openai_api_key()
    print(f"当前使用的API密钥前缀：{api_key[:10]}...") # 仅打印前10位避免泄露
    client = AsyncOpenAI(api_key=api_key, timeout=60)
    return await client.embeddings.create(model='text-embedding-ada-002', input=input)

3. 配置异步客户端的网络参数 #

若网络需代理，同步代码的全局代理配置不会自动应用到异步客户端，需单独设置：

client = AsyncOpenAI(
    api_key=await get_openai_api_key(),
    timeout=60,
    proxy="http://your-proxy-host:your-proxy-port"
)

若超时设置不生效，可细化超时规则：

from openai import AsyncOpenAI
from openai.types import Timeout

client = AsyncOpenAI(
    api_key=await get_openai_api_key(),
    timeout=Timeout(connect=10, read=60, write=60, pool=60)
)

4. 测试网络环境的异步兼容性 #

部分网络环境（如公司内网防火墙）可能限制异步HTTP请求，可单独用aiohttp测试连通性：

import aiohttp
import asyncio

async def test_openai_connection():
    url = "https://api.openai.com/v1/embeddings"
    api_key = await get_openai_api_key()
    async with aiohttp.ClientSession() as session:
        async with session.post(
            url,
            headers={"Authorization": f"Bearer {api_key}"},
            json={"model": "text-embedding-ada-002", "input": ["test"]}
        ) as response:
            print(f"响应状态码：{response.status}")
            print(f"响应内容：{await response.text()}")

asyncio.run(test_openai_connection())

若此测试也报错，需联系网络管理员调整异步请求的放行规则。

内容的提问来源于stack exchange，提问作者Leon YAo