AI API 并发卡住却没报 429？HTTPX 连接池超时与流式释放排查

并发调用 AI API 时，单条请求正常，多条一起跑却出现 httpx.PoolTimeout，不一定是服务商变慢。这个异常表示客户端在指定时间内没有从连接池取得连接。它与服务器返回 HTTP 429、已经建立连接后等不到数据，是不同的故障阶段。

本文针对 Python 的 HTTPX 异步客户端。我们用 Python 3.12.10、HTTPX 0.28.1 和本机 HTTP/1.1 模拟服务验证；没有测试真实中转站、模型速度或付费 API。其他 SDK 是否采用 HTTPX、是否改过连接池与重试参数，要查实际依赖及配置，不能直接套用这些数值。

1. 先保留异常类型，再判断卡在哪一层

不要只记录“请求超时”。HTTPX 区分四种超时：[1]

read=30 不等于整个任务最多30秒：它限制等待数据块的时间。持续收到数据的长流可能远超30秒。业务队列等待也不属于 HTTPX 的 pool 超时，两层应分开记录。

对于一次直接的 HTTPX 传输尝试，池等待超时发生在取得连接之前。但用户点一次按钮，SDK、代理或业务代码可能已经做过别的尝试；不能根据最后一个 PoolTimeout 就认定整项任务从未发送或必定不计费。控制台暂时没有记录也不是充分证据。保留逻辑任务编号与每次尝试编号，必要时按脱敏工单与最小复现核对。

2. 三个连接参数，不是三个并发配额

HTTPX 的 Limits 主要包含：[2]

• max_connections：连接池允许的最大连接数。
• max_keepalive_connections：允许保留的空闲保活连接数，不是活跃请求上限。
• keepalive_expiry：空闲保活连接可保留多久，不是模型生成超时。

因此，把 max_keepalive_connections 从5调到50，不等于把业务并发从5提升到50。把 max_connections 设为无限，也不能增加上游额度，反而可能扩大本机资源占用和请求压力。

在本文的 HTTP/1.1 实验里，一条未读完的流占用一条连接。HTTP/2 支持在一条连接上复用多个流，所以“连接数等于并发请求数”的算法不能直接迁移；还要看实际协商的协议与流限制。即使配置 http2=True，双方没有共同使用 HTTP/2 时仍可能使用 HTTP/1.1，应检查响应的 http_version。[4]

连接池的作用域也要核对。同一客户端中的任务会竞争其池资源；每个进程、每个客户端实例又可能有自己的池。单实例限制不能当成整个部署的全局限额。应在清楚生命周期的范围内复用客户端，并在应用关闭时释放；不要在高频循环里为每条请求新建客户端来“绕开”池等待。[3]

3. 流式响应的关闭责任不能丢

长流正常占满连接池，和代码遗漏关闭响应，都可能导致后来请求等待。先检查这三条路径：

1. 正常读完：调用方是否读到了完整响应，是否完成应用协议的结束校验。
2. 提前返回或解析报错：响应上下文是否仍会退出。
3. 用户取消或业务总超时：任务是否执行清理，而不是只停止界面展示。

HTTPX 推荐使用 async with client.stream(...)。离开上下文时会关闭响应；如果使用 client.send(..., stream=True) 的手动模式，就必须确保最终调用 await response.aclose()。[3] 不要从已经退出的上下文里返回一个尚未消费的流，让另一层稍后再读。

关闭未读完的 HTTP/1.1 响应不保证原 TCP 连接还能复用，但应释放这次响应占用的池资源，让后续请求有机会建立或使用连接。它也不保证远端模型停止工作，取消与费用的边界见停止生成后的请求和账单核对。

4. 在本机复现一次池等待，不发送真实 API 请求

将下面代码保存为 pool_demo.py，在安装了 HTTPX 的 Python 3.11+ 环境执行 python pool_demo.py。脚本只监听 127.0.0.1 的临时端口，不需要 API Key。

它故意把连接池缩到1，并让第一条响应停在半途；第二条请求只等0.1秒。随后退出第一条响应上下文，验证新请求恢复。模拟服务的等待和客户端操作都有上限，结束时关闭监听与连接。

import asyncio
import httpx


async def demo():
    seen = []
    release = asyncio.Event()
    handlers = set()

    async def handle(reader, writer):
        task = asyncio.current_task()
        handlers.add(task)
        try:
            header = await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), 1)
            path = header.split(b" ", 2)[1].decode("ascii")
            seen.append(path)
            writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n"
                         b"Connection: close\r\n\r\n")
            if path == "/hold":
                writer.write(b"a")
                await writer.drain()
                await asyncio.wait_for(release.wait(), 2)
                writer.write(b"b")
            else:
                writer.write(b"ok")
            await writer.drain()
        except (ConnectionError, TimeoutError, asyncio.IncompleteReadError):
            pass
        finally:
            writer.close()
            try:
                await writer.wait_closed()
            except ConnectionError:
                pass
            handlers.discard(task)

    server = await asyncio.start_server(handle, "127.0.0.1", 0, limit=4096)
    port = server.sockets[0].getsockname()[1]
    url = f"http://127.0.0.1:{port}"
    limits = httpx.Limits(max_connections=1, max_keepalive_connections=1)
    timeout = httpx.Timeout(1.0, pool=0.1)
    try:
        async with server, httpx.AsyncClient(
            limits=limits, timeout=timeout, http2=False,
            trust_env=False, follow_redirects=False
        ) as client:
            async with asyncio.timeout(5):
                async with client.stream("GET", url + "/hold") as response:
                    response.raise_for_status()
                    try:
                        await client.get(url + "/blocked")
                    except httpx.PoolTimeout:
                        print("PoolTimeout: waiting for the occupied slot")
                    else:
                        raise AssertionError("Expected PoolTimeout")
                result = await client.get(url + "/ok")
                assert result.status_code == 200 and result.text == "ok"
                assert seen == ["/hold", "/ok"], seen
                print("After stream close: 200; /blocked never reached server")
    finally:
        release.set()
        if handlers:
            await asyncio.gather(*list(handlers))


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

预期看到 PoolTimeout: waiting for the occupied slot，随后是 After stream close: 200; /blocked never reached server。断言还确认模拟服务只收到了 /hold 与 /ok。

这里没有模拟响应解析器或完整 SSE 协议，也没有证明某个生产环境存在泄漏。第一条流是故意保持开启的，正常长流同样能占满小池。测试使用 trust_env=False 是为了防止本机实验经过环境代理，不是在建议生产环境绕过必要的代理或证书配置。脚本没有关闭 TLS 验证；本机实验使用明文回环地址。

5. 把业务队列、完整流生命周期与总预算放在一起

生产修复通常按这个顺序进行：先补齐响应关闭路径，再限制入口负载，最后根据测量调整连接限额与池等待时间。只延长 pool 会让排队更久，不能解决响应一直未关闭的问题。

下面是一个小响应读取函数，需要前面的 asyncio、httpx 导入。调用方传入同一事件循环内复用的客户端、共享的 asyncio.Semaphore，以及已核验的固定服务地址。它只用 GET 统计原始响应字节，不是模型调用或 SSE 解析器。

async def read_small(client, gate, url):
    async with asyncio.timeout(10):
        async with gate:
            async with client.stream("GET", url) as response:
                response.raise_for_status()
                size = 0
                async for chunk in response.aiter_raw():
                    size += len(chunk)
                    if size > 65536:
                        raise ValueError("response-size-limit")
                return size

把 asyncio.timeout(10) 放在信号量外面，才能把“等业务名额”也计入本次函数的预算。超时通过协作式取消生效，不是抢占式的精确10秒墙钟保证；阻塞事件循环或过慢的清理仍可能拖延。信号量覆盖整个读取阶段，不能收到响应头就提前释放名额。Python 的超时上下文会把自身触发的取消转换为 TimeoutError；在上下文外处理，并保留取消传播，不要吞掉 CancelledError。[5]

函数在超过65,536字节时抛错并关闭响应。这只是原始响应体的计数上限，不是进程内存硬上限，也没有校验内容或解压后的大小。这里不累计正文；HTTPX 本身仍可能先交付一块较大的数据。普通异常、超时和调用方取消都应触发上下文清理。边界值只是示例，要按真实响应规模、时限和协议调整。

信号量只限制进入执行区的数量，不限制已经创建的等待任务总数。批量任务还要有有界队列或固定工作者，超过容量明确拒绝或延期；不要先生成几十万个协程再指望信号量解决内存占用。HTTPX 的连接数、业务并发、RPM/TPM 配额是不同层的控制，不能互相替代。

6. 验收修复时，记录阶段而不是只看成功率

建议至少记录：业务入队时间、取得业务名额时间、调用 HTTPX 时间、收到响应头时间、读取结束/取消时间、异常类、客户端实例标识及尝试编号。用单调时钟测耗时，用带时区的时间关联服务端记录。不要打印完整密钥、请求正文或带敏感查询参数的 URL。

“调用 HTTPX 到响应头”包含池等待、建连、发送和服务端等待等多种时间，不能把这一整段直接叫作模型首字延迟。缺少传输层证据时，写明无法进一步拆分。

本地回归应覆盖：池被长流占满；正常读完；中途取消；读取超时；HTTP错误；响应大小超限；业务队列等待超过总预算。每次失败后都用同一客户端发一条固定的小请求，检查名额和连接资源能否恢复。这里的200只代表模拟服务响应成功，不证明真实模型任务已完整结束。

本文两段代码已通过12类本机用例，包含上述路径、空响应、大小边界、重定向拒绝和响应体提前断开；失败后验证了同一客户端及信号量可继续使用。这些是回环网络上的功能验证，不是性能跑分或服务商测试。

最后再观察真实业务允许的低负载窗口：没有新增 PoolTimeout、队列可回落、流结束与清理次数吻合、没有隐藏重试放大。若错误仍然存在，按连接池作用域检查其他任务和客户端生命周期，不要仅因单条请求能成功就给服务商下“稳定”或“不稳定”的结论。

7. 官方资料与核对日期

核对日期：2026-11-05。示例在 Python 3.12.10 / HTTPX 0.28.1 本地运行；这些版本是验证环境，不宣称是最新版。

1. HTTPX Timeouts：四种超时及读取数据块、取得连接的不同阶段；同时核对0.28.1官方仓库文档。
2. HTTPX Resource Limits：总连接、空闲保活连接及过期时间。
3. HTTPX Async Support：客户端复用、流式上下文与手动关闭责任；核对0.28.1官方仓库文档。
4. HTTPX HTTP/2：多路复用、协商与实际协议检查。
5. Python asyncio Tasks：Timeouts 与 Task Cancellation：总预算、取消和清理边界。