第5章 HTTP 客户端与 Web

学习目标

• 使用 httpx 发起同步与异步 HTTP 请求，掌握连接池复用
• 理解前后端交互模式（REST、SSE、WebSocket）的适用场景
• 实现 Server-Sent Events 与 WebSocket 的流式通信
• 设计健壮的错误处理、重试与退避策略
• 调用 LLM API 完成流式输出与工具调用

问题导入

AI Agent 极少"单机作战"：它要调用 LLM 推理、检索向量库、触发外部工具、向前端推送中间状态——每一次交互都是一次网络通信。如何高效发起请求、如何处理流式响应、如何在网络抖动时优雅重试、如何在长连接场景下保持低延迟，直接决定了 Agent 的体验与稳定性。

Python 生态中，httpx 因其"同步/异步双模、HTTP/2 支持、与 requests API 高度兼容"的特性，已成为现代 HTTP 客户端首选。配合 SSE、WebSocket 等流式协议，以及 tenacity 等重试库，可以构建出生产级 Agent 的通信层。本章将系统讲解这些技术，并以"调用 LLM 流式 API"作为贯穿案例。

核心概念

httpx 同步与异步

httpx 提供两套 API：httpx.Client（同步）与 httpx.AsyncClient（异步）。两者 API 几乎一致，便于在同步脚本与异步服务间迁移。关键实践是复用客户端实例：每个 Client 内部维护连接池，复用可避免反复 TCP 握手与 TLS 协商，大幅降低延迟与资源消耗。Client 应作为长生命周期对象（如应用级单例），并通过 with / async with 确保正确关闭。

请求构造与响应处理

client.get/post/put/delete 方法支持 params（查询参数）、json（JSON 体）、headers、timeout、auth 等。响应对象 Response 提供 .status_code、.json()、.text、.headers、.iter_bytes() / .aiter_bytes()（流式字节）、.iter_lines() / .aiter_lines()（流式行）。设置 follow_redirects=True 跟随重定向。

流式响应与 SSE

流式响应（streaming response）指服务端不一次性返回完整 body，而是持续推送数据块。Server-Sent Events（SSE）是基于 HTTP 的服务器推送协议：响应 Content-Type: text/event-stream，每条消息形如 data: <payload>\n\n，客户端持续读取直至连接关闭。SSE 是 LLM 流式输出的标准协议（OpenAI、Anthropic 均采用）。客户端用 client.stream("POST", url, json=...) 上下文管理器获取流，逐块解析。

WebSocket 双向通信

WebSocket 提供全双工、长连接通信，适合需要双向实时交互的场景（如多人协作、语音对话）。与 SSE（服务端单向推送）不同，WebSocket 允许客户端随时发送消息。Python 中可用 websockets 库或 Starlette/FastAPI 的 WebSocket 支持。在 Agent 场景，WebSocket 常用于"边说话边接收"的语音 Agent。

错误处理与重试

网络通信必然面临超时、连接错误、5xx、限流（429）等问题。健壮策略包括： - 超时分层：连接超时、读取超时、整体超时分别设置。 - 幂等性判断：仅对幂等请求（GET、PUT）自动重试，POST 需谨慎。 - 退避策略：指数退避 + 抖动（jitter），避免重试风暴；429 时尊重 Retry-After 头。 - 重试库：tenacity 提供装饰器式重试，支持 @retry(stop=stop_after_attempt(3), wait=wait_exponential())。 - 熔断：连续失败达阈值后短路，避免雪崩。

原理图

flowchart LR
    Agent[AI Agent 客户端] -->|POST /chat| Client[httpx.AsyncClient 连接池]
    Client -->|HTTP/2| Gateway[LLM 网关]
    Gateway -->|SSE 流| Stream[流式响应解析]
    Stream --> Parse[逐 token 解析 data: 行]
    Parse --> Sink[业务消费]
    Agent -.->|WS 双向| WSClient[WebSocket 长连接]
    WSClient -.->|实时交互| VoiceServer[语音 Agent 服务]
    Client -->|失败/超时| Retry[tenacity 重试+退避]
    Retry --> Client
    Client -->|限流 429| Backoff[尊重 Retry-After]
    Backoff --> Client

图：客户端-服务端交互，httpx 复用连接池，SSE 流式解析 LLM token，失败经 tenacity 退避重试

示例代码

异步 HTTP 客户端基础

import asyncio
import httpx

async def fetch_example() -> None:
    # 复用客户端，连接池生效
    async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client:
        resp = await client.get("https://httpbin.org/get", params={"q": "agent"})
        print(f"状态: {resp.status_code}")
        data = resp.json()
        print(f"请求参数回显: {data['args']}")

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

并发请求与连接池

import asyncio
import httpx
import time

async def fetch_one(client: httpx.AsyncClient, url: str) -> dict:
    resp = await client.get(url)
    resp.raise_for_status()
    return resp.json()

async def main() -> None:
    urls = [f"https://httpbin.org/delay/{i}" for i in range(1, 4)]
    # 复用单个 AsyncClient，连接池自动管理
    async with httpx.AsyncClient(timeout=30.0) as client:
        start = time.perf_counter()
        results = await asyncio.gather(*(fetch_one(client, u) for u in urls))
        print(f"并发耗时: {time.perf_counter() - start:.2f}s, 共 {len(results)} 个结果")

asyncio.run(main())

SSE 流式解析 LLM 输出

import asyncio
import httpx
import json

async def stream_llm(prompt: str) -> None:
    """模拟调用 LLM 流式接口，逐 token 打印"""
    payload = {
        "model": "gpt-4o-mini",
        "messages": [{"role": "user", "content": prompt}],
        "stream": True,
    }
    headers = {"Authorization": "Bearer demo-token"}
    async with httpx.AsyncClient(timeout=httpx.Timeout(60.0, connect=5.0)) as client:
        async with client.stream(
            "POST", "https://api.example.com/v1/chat/completions",
            json=payload, headers=headers,
        ) as resp:
            resp.raise_for_status()
            full = ""
            async for line in resp.aiter_lines():
                if not line or not line.startswith("data: "):
                    continue
                data = line.removeprefix("data: ")
                if data == "[DONE]":
                    break
                chunk = json.loads(data)
                delta = chunk["choices"][0]["delta"].get("content", "")
                full += delta
                print(delta, end="", flush=True)
            print(f"\n完整回复: {full}")

if __name__ == "__main__":
    asyncio.run(stream_llm("用一句话介绍 AI Agent"))

tenacity 重试与退避

import asyncio
import random
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=0.5, min=0.5, max=5),
    retry=retry_if_exception_type((httpx.RequestError, httpx.HTTPStatusError)),
    reraise=True,
)
async def call_api(client: httpx.AsyncClient, url: str) -> dict:
    resp = await client.get(url)
    # 5xx 自动抛 HTTPStatusError 触发重试
    resp.raise_for_status()
    return resp.json()

async def main() -> None:
    async with httpx.AsyncClient() as client:
        try:
            data = await call_api(client, "https://httpbin.org/status/500")
        except httpx.HTTPStatusError as e:
            print(f"最终失败: {e.response.status_code}")

asyncio.run(main())

WebSocket 客户端

import asyncio
import websockets

async def ws_client() -> None:
    uri = "ws://localhost:8000/ws/agent"
    async with websockets.connect(uri) as ws:
        await ws.send("你好，Agent")
        while True:
            msg = await ws.recv()
            print(f"收到: {msg}")
            if msg == "[DONE]":
                break

# asyncio.run(ws_client())

企业案例

案例 1：LLM API 流式调用

某团队封装统一的 LLM 客户端，支持 OpenAI、Anthropic、本地模型。客户端用 httpx AsyncClient 单例复用连接池，对所有厂商的 SSE 流做统一解析（兼容 data: 行格式差异），用 asyncio.Queue 将 token 分发给上层业务（Web 前端 SSE 推送、语音 TTS、日志记录）。重试策略用 tenacity：仅对连接错误与 5xx 重试，429 按 Retry-After 头退避，避免无谓重试。

案例 2：Agent 工具调用网关

Agent 需调用多个外部 HTTP 工具（天气、搜索、知识库）。团队用 httpx 封装 ToolClient，为每个工具配置独立超时与重试策略，并通过 asyncio.Semaphore 限制对同一工具的并发，防止下游被压垮。失败工具返回结构化错误而非抛异常，Agent 据此决定是否换用其他工具，提升鲁棒性。

案例 3：语音 Agent 的 WebSocket 通道

一个语音对话 Agent 采用 WebSocket 双向通道：客户端边采集音频边发送，服务端边识别边用 LLM 生成回复并推送 token，TTS 流式合成播放。相比 HTTP 轮询，WebSocket 将端到端延迟从 800ms 降至 300ms。团队用 websockets 库的连接心跳与断线重连机制保证长连接稳定性。

常见问题

Q1：httpx 和 requests 该选哪个？ 新项目优先 httpx：支持异步与 HTTP/2、API 与 requests 高度兼容、可流式与代理。requests 仅适合纯同步、无异步需求的小脚本。需注意 httpx 默认不跟随重定向，需显式 follow_redirects=True。

Q2：为什么不能每次请求都新建一个 Client？ 新建 Client 会重建连接池，每次请求都需 TCP 握手 + TLS 协商，延迟增加数十到数百毫秒，且文件描述符消耗大，高并发下可能耗尽端口。应将 Client 作为长生命周期对象复用。

Q3：SSE 和 WebSocket 有什么区别？如何选？ SSE 基于 HTTP，单向（服务端→客户端），自动重连，文本协议，简单易调试，适合 LLM token 流、通知推送。WebSocket 是独立协议，双向，需手动处理重连，支持二进制，适合实时双向交互（语音、协作）。LLM 文本流首选 SSE，双向实时选 WebSocket。

Q4：重试会不会导致请求重复执行？ 对非幂等请求（如扣款 POST）盲目重试会重复执行。应：仅对幂等方法（GET/PUT/DELETE）自动重试；对 POST 设计幂等键（Idempotency-Key 头）让服务端去重；或先查询再决定是否重试。生产中务必评估幂等性。

面试题

题1：解释 httpx 的 Timeout 各字段含义。 参考答案：Timeout(timeout) 整体超时；Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0) 分别设置连接建立、读取、写入、等待连接池可用连接的超时。分层超时能更精细控制，例如 connect 超时短以快速失败，read 超时长以容忍慢响应。

题2：如何解析 SSE 流中的 data: 行？要注意什么？ 参考答案：逐行读取，跳过空行与非 data: 开头行（如 event:、id:、注释 :）；去除 data: 前缀后，若为 [DONE] 表示结束，否则按 JSON 解析。注意一条 SSE 消息可能跨多个 data: 行，需按空行分隔合并；流可能中断，需处理重连与 Last-Event-ID。

题3：指数退避为什么要加抖动（jitter）？ 参考答案：纯指数退避会让多个客户端在同一时刻集中重试（"惊群"），加剧服务端压力。加入随机抖动使各客户端重试时刻分散，平滑负载，提高整体成功率。tenacity 的 wait_exponential_jitter 即为此设计。

题4：HTTP/2 对 Agent 调用 LLM 有什么好处？ 参考答案：HTTP/2 支持多路复用，单一 TCP 连接可并发多个请求，避免 HTTP/1.1 的队头阻塞与连接数限制；头部压缩（HPACK）减少开销；对同一 LLM 网关的高并发调用尤其受益。httpx 通过 http2=True 启用（需安装 h2 包）。

练习题

1. 上机：用 httpx.AsyncClient 编写一个 MultiSourceFetcher，并发从 3 个 URL 抓取数据，复用同一客户端，统计总耗时与各请求耗时，对比"每次新建客户端"的耗时差异。

2. 思考：调用 LLM 流式接口时，若中途网络断开，已收到的 token 如何保留？请设计一个支持断点续传的方案（提示：结合 Last-Event-ID 或会话状态持久化）。

3. 上机：用 tenacity 封装一个 robust_post(client, url, json) 函数，对 429 与 5xx 退避重试，429 时读取 Retry-After 头决定等待时长，最多重试 5 次，并打印每次重试原因。

小结

HTTP 客户端与流式协议是 AI Agent 与外界通信的血管。httpx 以同步/异步双模与连接池复用成为首选；SSE 是 LLM 流式输出的标准；WebSocket 服务双向实时交互；tenacity 提供退避重试保障鲁棒性。理解这些技术的适用边界与陷阱，才能构建低延迟、高可用的 Agent 通信层。下一章将聚焦如何为这些组件编写可靠测试。延伸阅读：httpx 文档、tenacity 文档、websockets 文档。