DEBUG.WIKI
京东云;让 AI 进入真实业务;零售 · 物流 · 客服 · 硬件;查看方案 →
ZgoCloud:多地区部署,按需选节点;日本、美国、香港、德国
Evoxt;最高 6.0GHz CPU;高频云服务器 · 免费每周备份;查看配置 →

AsyncOpenAI 配了代理却报 Response 不能 await,先查传进去的客户端

AsyncOpenAI 需要异步 HTTP 客户端。若底层异常是 Response 不能 await,检查是否误传 httpx.Client,改用 AsyncClient,并在同一事件循环内创建、使用和关闭。

AsyncOpenAI 配了代理却报 Response 不能 await,先查传进去的客户端

最外层异常是 APIConnectionError,容易让人先排查代理。把堆栈往上翻,如果还能看到下面这行,排查顺序就该变一下:

TypeError: object Response can't be used in 'await' expression

官方 issue #1095 的复现环境是 openai 1.9.0、Python 3.9.13。触发代码把同步 httpx.Client 交给了 AsyncOpenAI。异步 SDK 等待 send(),同步客户端却直接交回了 Response,于是异常发生在 await 上。原始复现与修正

这个历史堆栈不代表所有新版都会报同样的外层异常。有的版本会更早拒绝错误类型。真正值得核对的是:SDK 和传输客户端是否都走异步接口。

检查创建客户端的地方,别只检查请求函数

下面这组搭配才一致:

  • OpenAI 配 httpx.Client,请求直接调用
  • AsyncOpenAI 配 httpx.AsyncClient,请求使用 await

给请求函数加 async def 不会改变传入对象的类型。也不要删掉 SDK 请求前的 await 来绕过错误,那样只是把未执行的协程继续传给业务代码。

若客户端来自依赖注入,临时在注入点记录 type(http_client).__name__ 即可,不要打印代理地址或认证信息。HTTPX 官方异步接口说明把请求方法、客户端关闭方法都列为需要异步处理的操作。HTTPX 异步接口

把创建、请求和关闭放进同一个生命周期

下面的独立脚本以 openai 1.109.1、httpx 0.28.1 为接口基线。升级跨越大版本时,先核对 SDK 接受的自定义传输客户端类型。模型和代理从环境变量读取。执行请求会联网并可能计费,请在自己的测试环境确认后运行。

import asyncio
import os
import httpx
from openai import AsyncOpenAI

async def main():
    transport = httpx.AsyncClient(
        proxy=os.environ["APP_HTTP_PROXY"],
        timeout=30.0,
    )
    async with AsyncOpenAI(
        api_key=os.environ["OPENAI_API_KEY"],
        http_client=transport,
        max_retries=0,
    ) as client:
        result = await client.chat.completions.create(
            model=os.environ["OPENAI_MODEL"],
            messages=[{"role": "user", "content": "只回复 OK"}],
        )
        print(result.choices[0].finish_reason)

if __name__ == "__main__":
    asyncio.run(main())

这里把重试设为零,是为了在诊断时尽快看到第一次失败;修复后再按项目要求配置重试。生产 Web 服务应复用生命周期内的客户端,避免每个请求都建立新连接池。在已有事件循环的 Notebook 或异步框架里,调用 await main(),不要再嵌套 asyncio.run()。

旧 issue 使用的 proxies= 属于当时的 HTTPX 接口。复制历史代码时要按本地版本改代理参数;新版 HTTPX 单代理示例使用 proxy=。HTTPX 代理说明

不访问代理,也能先验证异步传输

下面用内存中的 MockTransport 检查异步请求和关闭流程,不会访问外网。HTTPX 模拟传输说明

import asyncio
import httpx

async def check_transport():
    def reply(request):
        return httpx.Response(200, json={"ok": True})

    async with httpx.AsyncClient(
        transport=httpx.MockTransport(reply)
    ) as client:
        response = await client.get("https://example.invalid/check")
        assert response.json() == {"ok": True}

asyncio.run(check_transport())

这能检查本地异步用法,不能证明代理认证、TLS 或 API 权限正确。换回真实传输后若出现 407、连接超时或证书异常,应沿着新的具体错误继续查。只有底层仍是“Response 不能 await”时,才继续围绕同步/异步错配排查。