Python 与 AI 开发教程

HTTP 和 HTTPX

Agent 的每一步几乎都在调用接口。这一章用 HTTPX 发请求，处理超时、错误和重试，顺手把 OkHttp 的经验搬过来。

  • Python 服务端 · 第 1 章
  • 16 分钟
  • 5 组对比
  • 基于 HTTPX 0.28 · HTTPX2 2.16 · OkHttp 5.5

发请求

HTTPX 是 Python 常用的 HTTP 客户端，同时支持同步和异步。requests 也很流行，但它只有同步接口，在 async def 里用会卡住事件循环。

终端
pip install httpx

本章的示例用 MockTransport 代替真实网络，存成文件就能运行。真实项目里去掉 transport 参数，请求就会发到网络上。

发送 GET 请求

Python
import httpx

response = httpx.get(
    "https://api.example.com/users/1",
    params={"page": 2},
)

response.status_code  # => 200
response.json()  # => {'name': 'Ada', 'page': '2'}
var client = HttpClient.newHttpClient();
var request = HttpRequest.newBuilder()
    .uri(URI.create("https://api.example.com/users/1?page=2"))
    .build();

var response = client.send(request, BodyHandlers.ofString());
System.out.println(response.statusCode());
System.out.println(response.body());
对比项PythonJava
入口模块函数或 ClientHttpClient
查询参数params 字典，自动编码自己拼进网址
解析 JSONresponse.json()用 Jackson
关闭响应自动自动

response.json() 返回字典和列表。要拿到有类型的对象，再交给下一章的 Pydantic。

POST 用 json= 发送 JSON，用 data= 发送表单，用 headers= 加请求头。

Python
response = httpx.post(
    "https://api.example.com/users",
    json={"name": "Ada"},
    headers={"Authorization": "Bearer token"},
)

response.json()["body"]  # => '{"name":"Ada"}'

复用连接

httpx.get 每次都新建连接，调用一个服务多次时要改用 Client。它带连接池，还能统一设置基础地址、请求头和超时。

Client 相当于 OkHttpClient，一个服务创建一个，全局共用。在 FastAPI 里，把它放进 lifespan 创建和关闭。

共用一个客户端

Python
with httpx.Client(
    base_url="https://api.example.com/v1",
    headers={"User-Agent": "demo/1.0"},
) as client:
    response = client.get("/users")

response.json()  # => {'path': '/v1/users', 'ua': 'demo/1.0'}
var client = HttpClient.newBuilder()
    .connectTimeout(Duration.ofSeconds(5))
    .build();

// 基础地址和公共请求头要自己封装
var request = HttpRequest.newBuilder()
    .uri(URI.create(BASE_URL + "/users"))
    .header("User-Agent", "demo/1.0")
    .build();
对比项PythonJava
连接池Client 自带HttpClient 自带
公共请求头headers 参数自己封装
基础地址base_url自己封装
拦截请求event_hooks 或自定义 transport自己封装
关闭with 或 client.close()进程结束时回收

不要在循环或每次请求里创建 Client，连接池会失去作用。

超时

HTTPX 默认的超时是 5 秒，这和 OkHttp 的默认 10 秒不同。调用大模型时，等待第一个字节就可能超过 5 秒，所以要主动调大读取超时。

设置超时

Python
import httpx

timeout = httpx.Timeout(
    10.0,  # 其余阶段的默认值
    connect=5.0,
    read=60.0,
)
client = httpx.Client(timeout=timeout)
client.timeout.read  # => 60.0
var client = HttpClient.newBuilder()
    .connectTimeout(Duration.ofSeconds(5))
    .build();

// 没有单独的读取超时，timeout 管到响应头返回
var request = HttpRequest.newBuilder()
    .uri(URI.create("https://api.example.com"))
    .timeout(Duration.ofSeconds(60))
    .build();
对比项PythonJava
默认值各阶段 5 秒不超时
读取超时的含义两次收到数据的最长间隔等响应头的总时间
关闭超时timeout=None不设置

读取超时是两次收到数据的间隔，不是整个请求的总时长。流式响应可以很长，只要一直有数据就不会超时。

错误处理

HTTPX 收到 4xx、5xx 不会抛异常，要调用 raise_for_status()。连接失败和超时则直接抛异常，它们的共同父类是 httpx.HTTPError。

处理错误

Python
def get_user(client: httpx.Client, user_id: int) -> dict | None:
    try:
        response = client.get(f"https://api.example.com/users/{user_id}")
        response.raise_for_status()
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            return None
        raise
    except httpx.TimeoutException:
        raise RuntimeError("服务超时")
    return response.json()

get_user(client, 9)  # => None
Map<String, Object> getUser(HttpClient client, HttpRequest request)
        throws Exception {
    try {
        var response = client.send(request, BodyHandlers.ofString());
        if (response.statusCode() == 404) {
            return null;
        }
        return mapper.readValue(response.body(), Map.class);
    } catch (HttpTimeoutException e) {
        throw new RuntimeException("服务超时", e);
    }
}
对比项PythonJava
4xx、5xx不抛，调用 raise_for_status()不抛，自己判断
超时httpx.TimeoutExceptionHttpTimeoutException
连接失败httpx.ConnectErrorConnectException

三种客户端都不把 4xx、5xx 当异常，这一点和 Retrofit 的挂起函数不同。忘记检查状态码时，错误页面会被当成正常内容解析。

重试

HTTPTransport(retries=3) 只在连接失败时重试。429、5xx 这类响应要自己处理，通常用指数退避：每次多等一倍，再加一点随机抖动，避免大量客户端同时重试。

Python
import random
import time

RETRYABLE = {429, 500, 502, 503, 504}

def get_with_retry(client: httpx.Client, url: str, attempts: int = 4) -> httpx.Response:
    for attempt in range(attempts):
        response = client.get(url)
        if response.status_code not in RETRYABLE or attempt == attempts - 1:
            return response
        retry_after = response.headers.get("Retry-After")
        wait = float(retry_after) if retry_after else 2 ** attempt * 0.5
        time.sleep(wait + random.random() * 0.1)
    raise AssertionError("不会走到这里")

get_with_retry(client, "https://api.example.com").status_code  # => 200

异步并发

在 async def 里用 AsyncClient，方法名和 Client 一样，多一个 await。配合 TaskGroup 或 gather 就能同时发多个请求。

并发请求

Python
import asyncio

async def get_user(client: httpx.AsyncClient, user_id: int) -> dict:
    response = await client.get(f"/users/{user_id}")
    response.raise_for_status()
    return response.json()

async def main():
    async with httpx.AsyncClient(
        base_url="https://api.example.com",
    ) as client:
        users = await asyncio.gather(*[get_user(client, i) for i in [1, 2]])
    print(users)  # => [{'id': 1}, {'id': 2}]

asyncio.run(main())
var first = client.sendAsync(a, BodyHandlers.ofString());
var second = client.sendAsync(b, BodyHandlers.ofString());

CompletableFuture.allOf(first, second).join();
System.out.println(first.join().body());
对比项PythonJava
客户端AsyncClientsendAsync
并发gatherCompletableFuture.allOf
限制并发数asyncio.SemaphoreSemaphore

同步的 Client 不能用在 async def 里，它会阻塞事件循环。

测试

测试时不要访问真实网络。MockTransport 接收一个函数，由它决定每个请求返回什么，作用类似 MockWebServer，但不用启动服务。

Python
import httpx

def handler(request: httpx.Request) -> httpx.Response:
    if request.url.path == "/users/1":
        return httpx.Response(200, json={"name": "Ada"})
    return httpx.Response(404)

client = httpx.Client(
    transport=httpx.MockTransport(handler),
    base_url="https://api.example.com",
)

client.get("/users/1").json()  # => {'name': 'Ada'}
client.get("/users/2").status_code  # => 404
HTTPX 和 HTTPX2选读

HTTPX 的 0.28 版发布于 2024 年底，之后 Pydantic 团队维护了一个新的分支 HTTPX2，用法基本相同。OpenAI 的 Python SDK 已经改用 HTTPX2，FastAPI 的 TestClient 也优先使用它。

自己的代码可以继续用 httpx，这是目前教程和项目里最常见的写法。只有要把客户端对象传给这些 SDK 时要注意，它们需要 httpx2 里的对象，httpx.Client 不能直接传进去。

HTTP 客户端写法对照
PythonJava说明
httpx.ClientHttpClient全局共用一个
client.get(url, params=...)HttpRequest.newBuilder()HTTPX 自动编码参数
response.raise_for_status()判断 statusCode()都不会自动抛异常
httpx.Timeout(read=60)request.timeout()HTTPX 默认只有 5 秒
httpx.AsyncClientsendAsync协程里用它
MockTransportMockRestServiceServer不访问真实网络

本章要点

  • 调用同一个服务时用 Client 复用连接，全局共用一个
  • async def 里用 AsyncClient，不要用 requests 和同步 Client
  • 默认超时只有 5 秒，调用大模型要调大 read
  • 4xx、5xx 不抛异常，记得调用 raise_for_status()
  • 429 和 5xx 用指数退避重试，POST 只有幂等时才重试
  • 测试用 MockTransport，不访问真实网络

参考资料：HTTPX 快速入门客户端超时异步支持HTTPX2