第26篇|网络请求与 HTTP:requests/httpx、调用大模型 API
本篇定位:《Python 教程》专栏二「Python 进阶与工程」第 26 篇。承接第 25 篇 asyncio「如何叠等待」之后,正面讲清 HTTP 客户端:同步
requests、双模httpx、流式响应,以及如何调用 OpenAI 兼容 的 Chat Completions。为第 27 篇 FastAPI 推理服务、第 28 篇 Agent 工具循环提供「发请求」的肌肉记忆。读完你将能够:用自己的话解释一次 HTTP 往返(方法、状态码、Headers、Body);用
requests/httpx发 GET/POST 并解析 JSON;写出httpx.AsyncClient+gather+Semaphore的并发调用;实现非流式与流式(SSE)两种大模型调用;处理超时、429 与密钥环境变量。预计用时:阅读 35~45 分钟,动手 50~70 分钟。
第 25 篇把「协程如何叠等待」说清了;本篇把等待的对象换成真实网络。一句话定心:HTTP 就是「发一封信、等回信」——方法与 URL 说去哪儿,Headers 说身份与格式,Body 说内容,状态码说成败。 大模型 API 并没有发明新协议,只是把 prompt 塞进 JSON、把 token(或流式 chunk)塞进响应。
建议边读边在练习目录动手:uv init --no-package,再 uv add requests httpx。没有真实 API Key 时,可用 httpbin.org / 本地 mock 练协议;有 Key 时再对接任一 OpenAI 兼容网关。阅读顺序:HTTP 心智 → requests → httpx 双模 → 流式 → Chat Completions → 并发限流与 AI Infra → 踩坑。
和第 10 篇对照:文件 with 管资源;本篇 Session / Client / async with 同样管连接池。和第 11 篇对照:网络失败是常态,raise_for_status、超时、重试要写进路径。和第 25 篇对照:环内禁止同步 requests;异步路径用 httpx.AsyncClient + await。
动手时建议沿用「假 API → 真 HTTP → 真 LLM」三步:先对 https://httpbin.org/json 练解析;再对兼容端点练 chat/completions;最后把第 25 篇的 gather + Semaphore 换成真请求。做完后打开任意 Agent 评测脚本,你会认出同一套形状。
本篇不穷尽 HTTP/2 帧细节、不替代完整 REST 设计教程、不绑定某一家云厂商 SDK。主线是:协议心智 → 同步客户端 → 异步客户端 → 流式 → 兼容 API → 生产三件套(限流 / 超时 / 重试)。和「装一个 SDK 就完事」相比,正确价值是:你能看懂任何兼容端点的请求与响应,并能在 SDK 不够用时自己拼。
写请求时保持工程直觉:密钥进环境变量;超时必设;并发必有界;流式边收边处理;错误按状态码分支。 「把 API Key 写进仓库」与「async 路由里狂调同步 requests」——两个月后的夜班你会感谢今天的肌肉记忆。
也可以把本篇看成专栏二「联网面」的开篇:前面把并发模型收齐,本篇把「出站 HTTP」说透,下一篇把「入站服务」用 FastAPI 立住。Agent 工具调用、评测批量打分、网关转发,都假设你会发请求。
读官方 requests / httpx 文档时,按本篇目录对照即可。入门阶段把「会设超时、会解析 JSON、会 async 并发、会流式读、会藏密钥」做扎实,比背完整参数表更重要。
写请求代码时再记一条:HTTP 客户端的第一原则仍是正确优先于炫——错误的「全异步包装」比同步更糟,因为它摧毁可读性并引入取消、重试与连接泄漏新坑。排障时一条最小可复现(一次带超时的 GET + 打印状态码),往往比一百次「再加并发」更值钱。
和「一切都上官方 SDK」的神话相比,更值钱的是看懂兼容协议:你会换 base_url、会读 choices、会处理流式,SDK 只是薄封装。把这个品味带进 code review,比强制「必须用某某库」温和且有效。
若你刚从第 25 篇「协程与事件环」过来,本篇前两节会换一种体感:同样是叠等待,但挂钩点从「假 sleep」换成「真 TCP / TLS / JSON」。把三次串行 requests.get 改成 AsyncClient + gather,把「环里 requests」改成 await client.get——两条路径走完,联网直觉就长出来了。
和「在聊天里口头说会调 API」相比,可运行示例的价值是:机器可重复、可测、可对比墙钟与状态码——本地与 CI 上同一套结论。排障时你面对的是「谁超时、谁 429、谁堵环」,而不是「我记得好像调通了」。
动手前再立一条纪律:先同步正确,再异步;先非流式正确,再开 stream。 先对单一端点跑通并打印状态码与 body 片段,再引入并发与流式。很多人倒过来「先写 gather 再修超时」,修的时候容易把阻塞客户端塞进协程——事件环又变成复杂度放大器。
一、HTTP 心智模型:请求、响应与状态码
1.1 一张图看清一次往返

图1 HTTP:一次请求与一次响应
# 最小「读懂一次响应」——无需第三方库(演示用)
import json
from urllib.request import urlopen, Request
req = Request(
"https://httpbin.org/json",
headers={"User-Agent": "python-course/26"},
)
with urlopen(req, timeout=10) as resp:
print("status", resp.status)
data = json.loads(resp.read().decode())
print("keys", list(data.keys())[:5])
status 200
keys ['slideshow']
(若网络不可达,可改成本地 mock;重点是状态码 + JSON 解析这条路径。)
要点:
- 无状态:每次请求自带身份与内容;服务端不默认「记得你」。
- 状态码是分支入口:先看码,再决定解析、重试还是放弃。
- Body 格式靠 Headers 约定:发 JSON 就声明
Content-Type;收 JSON 用.json()。
1.2 和后续篇章的挂钩
第 27 篇 FastAPI 会站在服务端收同样的请求;本篇站在客户端发。协议两侧对称——你会用客户端调试自己写的接口。第 28 篇 Agent 的工具调用,多数就是「拼 JSON → POST → 解析」。
AI Infra 视角:推理网关、vLLM / SGLang 的 OpenAI 兼容端点、评测批量打分、Agent 并行工具——出站全是 HTTP。先把客户端层写对(超时、连接复用、限流),比先纠结「哪家 SDK」更划算。
1.3 安装与环境
uv init --no-package
uv add requests httpx
# 可选:把密钥放进环境,不要写进代码
export OPENAI_API_KEY="sk-..."
export OPENAI_BASE_URL="https://api.openai.com/v1" # 或兼容网关
统一约定:下文示例用环境变量 OPENAI_API_KEY / OPENAI_BASE_URL;没有 Key 时用 httpbin 练协议,用注释标出「换成真实端点」的位置。
1.4 常见状态码速查
把表贴在显示器边也值回票价:很多「模型坏了」其实是 401 密钥错了 或 404 多写了 /v1。先读状态码,再怀疑模型。
也可以把 HTTP 看成「有类型的远程函数调用」:URL 是函数名,JSON 是参数,状态码是异常通道,Body 是返回值。这样想,第 11 篇的异常分层与本篇的状态码分支就对上了。
二、requests:同步客户端入门
2.1 GET / POST 与 JSON
import requests
# GET
r = requests.get("https://httpbin.org/get", params={"q": "python"}, timeout=10)
print(r.status_code, r.headers.get("content-type", "")[:32])
print(r.json()["args"])
# POST JSON
r = requests.post(
"https://httpbin.org/post",
json={"hello": "world"},
timeout=10,
)
r.raise_for_status()
print(r.json()["json"])
200 application/json
{'q': 'python'}
{'hello': 'world'}
timeout=10 是秒;不设超时,请求可能永久挂起——生产里这是事故。raise_for_status() 在 4xx/5xx 时抛 HTTPError,避免「静默拿到错误页还当成功」。
2.2 Session:复用连接与默认头
import os
import requests
session = requests.Session()
session.headers.update({
"Authorization": f"Bearer {os.environ.get('OPENAI_API_KEY', 'dummy')}",
"Content-Type": "application/json",
})
# 多次请求复用 TCP / TLS;脚本结束可 session.close()
r = session.get("https://httpbin.org/headers", timeout=10)
print("has_auth", "Authorization" in r.json()["headers"])
session.close()
has_auth True
要点:短脚本用模块级 requests.get 也行;同一主机多次调用,优先 Session——少握手、好统一 Headers。
2.3 何时用 requests、何时不用
AI Infra 视角:很多「假异步」事故,根因就是 async 路由里塞了同步
requests。第 25 篇说过:阻塞请挪走;本篇给出替代物——httpx.AsyncClient。
2.4 错误处理最小模板
import requests
def fetch_json(url: str) -> dict:
try:
r = requests.get(url, timeout=10)
r.raise_for_status()
return r.json()
except requests.Timeout:
raise RuntimeError(f"timeout: {url}") from None
except requests.HTTPError as e:
raise RuntimeError(f"http {e.response.status_code}: {url}") from e
except requests.RequestException as e:
raise RuntimeError(f"network: {e}") from e
print(list(fetch_json("https://httpbin.org/json").keys())[:3])
['slideshow']
模板价值不在「多写几行」,而在把超时、HTTP 错误、其它网络错误分开——日志与告警才能对症。评测脚本里若所有失败都叫 Exception,你永远分不清是配额还是代码 bug。
三、httpx:同步与异步双模
3.1 对照图

图2 requests(同步)vs httpx(同步 / 异步)
import httpx
# 同步 API 形状接近 requests
with httpx.Client(timeout=10.0) as client:
r = client.get("https://httpbin.org/get", params={"lib": "httpx"})
r.raise_for_status()
print("sync", r.json()["args"])
sync {'lib': 'httpx'}
3.2 AsyncClient:可 await 的出站
import asyncio
import httpx
async def main() -> None:
async with httpx.AsyncClient(timeout=10.0) as client:
r = await client.get("https://httpbin.org/get", params={"mode": "async"})
r.raise_for_status()
print("async", r.json()["args"])
asyncio.run(main())
async {'mode': 'async'}
心智:async with 管理连接池生命周期;请求用 await client.get/post;不要每次请求 AsyncClient() 新建又扔掉——握手成本会被放大。
3.3 超时分层
import httpx
timeout = httpx.Timeout(10.0, connect=5.0) # 总预算 10s,连接最多 5s
with httpx.Client(timeout=timeout) as client:
r = client.get("https://httpbin.org/delay/1")
print(r.status_code)
200
大模型生成可能很慢:可以把 read / write 超时调大,但连接超时仍宜短——连不上就快失败,别空等。
3.4 和第 25 篇编排拼接
import asyncio
import httpx
async def fetch(client: httpx.AsyncClient, i: int) -> int:
r = await client.get("https://httpbin.org/get", params={"i": i})
r.raise_for_status()
return int(r.json()["args"]["i"])
async def main() -> None:
sem = asyncio.Semaphore(5)
async def bounded(i: int) -> int:
async with sem:
return await fetch(client, i)
async with httpx.AsyncClient(timeout=10.0) as client:
results = await asyncio.gather(*(bounded(i) for i in range(10)))
print(sorted(results))
asyncio.run(main())
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
这就是第 25 篇 Semaphore + gather 的落地版:共享一个 Client,有界并发,统一超时。
3.5 迁移口诀
从 requests 迁到 httpx 同步:requests.get → client.get;r.json() 依旧;注意 httpx 默认更严格(例如对 HTTP/2、重定向策略细节)。迁到异步:所有发送改 await,Client 改 AsyncClient,入口 asyncio.run。口诀:先保持语义不变,再换运输层。
和「复制 StackOverflow 第一段」相比,自己维护一张「项目出站约定」更值钱:统一超时、统一 Header、统一日志字段。新人入职第一周就能对着约定写,而不是每个人发明一套。
四、流式响应:边收边显
4.1 为什么要流式

图3 流式响应:边收边显 · 不等整包 JSON
非流式:等模型整段生成完再返回——首字延迟等于整段时间。流式(常见为 SSE,text/event-stream):服务端按 chunk 推送,客户端边收边显示,体验像「打字机」。
4.2 httpx 同步流式读行
import httpx
# 演示:按行读任意流式端点;真实 LLM 见第五节
url = "https://httpbin.org/stream/3"
with httpx.Client(timeout=30.0) as client:
with client.stream("GET", url) as resp:
resp.raise_for_status()
for line in resp.iter_lines():
if line:
print("chunk", line[:60])
chunk {"url": "https://httpbin.org/stream/3", ...
chunk {"url": "https://httpbin.org/stream/3", ...
chunk {"url": "https://httpbin.org/stream/3", ...
4.3 异步 aiter_lines
import asyncio
import httpx
async def main() -> None:
async with httpx.AsyncClient(timeout=30.0) as client:
async with client.stream("GET", "https://httpbin.org/stream/2") as resp:
resp.raise_for_status()
async for line in resp.aiter_lines():
if line:
print("async-chunk", line[:50])
asyncio.run(main())
async-chunk {"url": "https://httpbin.org/stream/2", ...
async-chunk {"url": "https://httpbin.org/stream/2", ...
要点:流式路径不要先 resp.json() 把整包读进内存;用 iter_lines / aiter_lines 增量处理。LLM 的 SSE 行通常以 data: 开头,空行分隔事件,结束标记常见为 data: [DONE]。
五、调用 OpenAI 兼容的 Chat Completions
5.1 协议形状

POST {base_url}/v1/chat/completions,请求时的身份认证方式为Authorization: Bearer $API_KEY;中间分为两个区域,左侧标注为“请求JSON”,示例内容包含model、messages、temperature、stream等关键参数,右侧标注为“响应JSON(节选)”,示例展示了助理角色返回的对话内容等信息;底部的黄色区域说明该接口的兼容特性,即修改base_url和model参数,就可以对接多数国内或开源网关,还提示了密钥的相关注意事项。图4 OpenAI 兼容 Chat Completions
多数国内网关、本地 vLLM / SGLang、云厂商「兼容模式」都认:
POST {base_url}/chat/completions(base_url常已含/v1)- Header:
Authorization: Bearer <key> - Body:
model、messages、temperature、stream等
5.2 非流式最小可运行
import os
import httpx
API_KEY = os.environ.get("OPENAI_API_KEY", "")
BASE = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1").rstrip("/")
payload = {
"model": os.environ.get("OPENAI_MODEL", "gpt-4o-mini"),
"messages": [
{"role": "system", "content": "用一句话回答。"},
{"role": "user", "content": "2+2等于几?"},
],
"temperature": 0,
"stream": False,
}
if not API_KEY:
print("skip: set OPENAI_API_KEY to run against a real endpoint")
else:
with httpx.Client(timeout=60.0) as client:
r = client.post(
f"{BASE}/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json=payload,
)
r.raise_for_status()
data = r.json()
print(data["choices"][0]["message"]["content"])
print("usage", data.get("usage"))
skip: set OPENAI_API_KEY to run against a real endpoint
(配置真实 Key 后,预期类似 4 或「等于 4」。)把 BASE 换成兼容网关即可,代码几乎不用改——这就是「兼容面」的工程价值。
5.3 流式:解析 SSE data 行
import json
import os
import httpx
API_KEY = os.environ.get("OPENAI_API_KEY", "")
BASE = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1").rstrip("/")
payload = {
"model": os.environ.get("OPENAI_MODEL", "gpt-4o-mini"),
"messages": [{"role": "user", "content": "数到 3,每个数字空格分隔"}],
"stream": True,
}
def iter_tokens() -> None:
if not API_KEY:
print("skip: set OPENAI_API_KEY for streaming demo")
return
with httpx.Client(timeout=60.0) as client:
with client.stream(
"POST",
f"{BASE}/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json=payload,
) as resp:
resp.raise_for_status()
for line in resp.iter_lines():
if not line or not line.startswith("data: "):
continue
data = line[len("data: ") :]
if data.strip() == "[DONE]":
break
chunk = json.loads(data)
delta = chunk["choices"][0].get("delta", {})
piece = delta.get("content") or ""
if piece:
print(piece, end="", flush=True)
print()
iter_tokens()
skip: set OPENAI_API_KEY for streaming demo
异步版把 Client 换成 AsyncClient,iter_lines 换成 aiter_lines,外层 async def + await 即可——形状与第 25 篇一致。
5.4 messages 角色约定
温度 temperature:越低越稳(评测、抽取);略高更发散(创意)。生产抽取任务常见 0~0.2。
5.5 本地 mock 与契约测试
没有 Key 时,不要停更:用 httpx.MockTransport 或一个最小 FastAPI(下篇)返回固定 JSON,先把客户端解析与重试测绿。契约测试的价值是——网关变更字段时,你的解析层先红。 这也和第 22 篇 pytest 同向:可重复的红绿循环,比「手动点一下聊天框」可靠。
多轮对话时,记得把历史 user / assistant 按顺序放进 messages;不要只发最新一句却指望服务端「记得」——HTTP 无状态,上下文在你自己的列表里。
六、AI Infra:并发打推理、限流与重试
6.1 生产三件套

图5 并发打推理:AsyncClient + Semaphore + 超时
import asyncio
import os
import random
import httpx
API_KEY = os.environ.get("OPENAI_API_KEY", "")
BASE = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1").rstrip("/")
async def chat_once(client: httpx.AsyncClient, prompt: str) -> str:
if not API_KEY:
await asyncio.sleep(0.05) # 无 Key 时用假延迟演示编排
return f"echo:{prompt}"
r = await client.post(
f"{BASE}/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json={
"model": os.environ.get("OPENAI_MODEL", "gpt-4o-mini"),
"messages": [{"role": "user", "content": prompt}],
"temperature": 0,
},
)
if r.status_code == 429:
raise httpx.HTTPStatusError("rate limited", request=r.request, response=r)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"]
async def with_retry(client: httpx.AsyncClient, prompt: str, retries: int = 3) -> str:
delay = 0.5
for attempt in range(retries):
try:
return await chat_once(client, prompt)
except httpx.HTTPStatusError as e:
if e.response is not None and e.response.status_code == 429 and attempt + 1 < retries:
await asyncio.sleep(delay + random.random() * 0.2)
delay *= 2
continue
raise
raise RuntimeError("unreachable")
async def main() -> None:
prompts = [f"ping-{i}" for i in range(8)]
sem = asyncio.Semaphore(3)
async with httpx.AsyncClient(timeout=60.0) as client:
async def one(p: str) -> str:
async with sem:
return await with_retry(client, p)
out = await asyncio.gather(*(one(p) for p in prompts))
print(len(out), out[0])
asyncio.run(main())
8 echo:ping-0
(有 Key 时会打印真实模型回复;无 Key 时走 echo 分支,仍能验证限流编排。)
6.2 选型表
6.3 和后续篇章的分工
- 第 25 篇:事件环、
gather、Semaphore、禁阻塞。 - 本篇:真实 HTTP、流式、兼容 Chat Completions、重试。
- 第 27 篇:FastAPI 站在服务端暴露
/v1/...一类接口。 - 第 28 篇:Agent 循环里把工具调用当成一次次 HTTP / 函数。
读到这里自检:你能否不看文档写出「带 Bearer 的 POST JSON」?能否解释为何 async 路由里不能直接 requests.get?能否画出 Semaphore 闸门与 429 退避?卡壳就回到图 1、图 2、图 5。
和「等线上偶发再加超时」相比,现在付出的代价极低:Client 级 timeout、有界并发、密钥进环境变量。事故夜里你会感谢今天的自己。
再强调边界:本篇教你用通用 HTTP 客户端对接兼容协议;各厂商 SDK、鉴权刷新、多模态上传是下一层细节。不要因为「听说有更炫的 SDK」就省略超时与限流。
若你带团队做 code review,可收成三条门槛:新增出站是否设超时;async 路径是否真用异步客户端;批量是否有 Semaphore / 退避。门槛不在「必须用某库」,而在于不许用无超时、无界并发换虚假吞吐。
6.4 连接池与生命周期
进程内长时间服务(即将在第 27 篇出现)应把 AsyncClient 放在应用 lifespan 里创建与关闭,而不是每个请求新建。短脚本则 async with 包住 main 即可。泄漏的典型现象是:文件描述符缓涨、CLOSE_WAIT 堆积——用 lsof / 指标盯住,比偶尔重启治本。
批量评测还要注意计费与配额:限流不仅是为了对方 429,也是为了自己的账单与公平性。把 QPS、并发、日预算写成配置,而不是硬编码在循环里。
动手时再记一条:联网的第一原则仍是可观测优先——每次出站至少能回答:打了谁、状态码多少、花了多久、是否重试、用了哪个模型。缺这五问,线上「偶发变慢」会变成玄学。
和「等 SDK 帮你处理好一切」相比,自己会拼兼容请求的价值是:SDK 版本打架、网关字段微调、临时换端点时,你不至于停摆。 薄封装 + 清晰协议,比厚魔法更耐折腾。
七、踩坑大全
通用
坑 1:不设 timeout。 现象:请求偶发永久挂起、worker 耗尽。原因:默认可能无限等。解决:Client / 单次请求都显式 timeout=。
坑 2:忽略状态码,直接 .json()。 现象:解析报错或逻辑静默错。解决:先 raise_for_status() 或按码分支。
坑 3:API Key 写进代码并提交仓库。 现象:密钥泄露、账号被盗刷。解决:环境变量 / 密钥管理;轮换泄露的 Key。
坑 4:在 async def 里用同步 requests。 现象:服务延迟尖刺、吞吐崩。解决:httpx.AsyncClient;或改同步 def 让框架丢线程池(知代价)。
坑 5:每次请求新建 Client / Session。 现象:握手过多、慢且耗文件描述符。解决:复用;async with 包住一批任务。
坑 6:流式却整包 resp.text。 现象:失去首字优势、大响应撑爆内存。解决:iter_lines / aiter_bytes。
坑 7:无界 gather 打爆对方 429。 现象:大面积失败。解决:Semaphore;尊重 Retry-After;指数退避 + 抖动。
macOS / Linux / Windows
坑 8:公司代理 / HTTPS 拦截。 现象:证书错误。解决:按环境配置 SSL_CERT_FILE / httpx 的 verify;切勿随意 verify=False 进生产。
坑 9:Linux ulimit 过低。 现象:高并发 Too many open files。解决:提限额;降并发;复用连接。
坑 10:Windows 控制台打印流式中文乱码。 现象:SSE 片段显示异常。解决:确认终端 UTF-8;PYTHONIOENCODING=utf-8。
与 AI / 服务叠加
坑 11:base_url 多写或少写 /v1。 现象:404。解决:打印最终 URL;对照网关文档;统一 .rstrip("/") 再拼接。
坑 12:把 stream=True 的响应当非流式 .json()。 现象:解析失败或一直阻塞。解决:按流式路径读 SSE。
坑 13:重试时不幂等却狂 POST。 现象:重复计费 / 重复副作用。解决:读请求可大胆重试;写请求要幂等键或谨慎次数。
八、小结
合上文章后,试着用自己的话说明:一次 HTTP 往返有哪些零件、为何要设超时、异步客户端如何与第 25 篇编排拼接、流式 SSE 如何逐行解析。讲得清,才算真的会了。动手至少跑通:一次 requests GET → 一次 httpx.AsyncClient gather → 一次(或 mock)Chat Completions → 一次流式读行 → 一次 Semaphore 限流。
- HTTP 心智:方法 / URL / Headers / Body / 状态码;大模型 API 是同一协议。
- requests:同步脚本利器;
Session复用;必设timeout;善用raise_for_status。 - httpx:同步
Client与异步AsyncClient;API 接近 requests;异步路径可 await。 - 流式:
stream=True+ 逐行 SSE;边收边显;勿整包缓冲。 - 兼容 Chat Completions:
base_url+ Bearer +messages;改网关少改代码。 - 生产三件套:有界并发、显式超时、可观测退避重试;密钥进环境变量。
把故事连起来:第 23~25 篇 concurrent 三件套 → 本篇真实出站 HTTP → 下一站 FastAPI 入站服务。专栏二工程刀具继续成套。
若只能带走一句话:出站必超时;异步用 AsyncClient;并发必限流;密钥不进仓库。
也可以用「三个不要」自检:不要无 timeout;不要在环里跑同步 HTTP;不要无界海量打 LLM。
再留一个自测:对 httpbin 跑通 GET/POST;把第 25 篇 gather 换成 AsyncClient;用环境变量拼一次(或 mock)chat/completions——三问都顺,本篇就站稳了。
本篇技能会立刻复利:读开源先看它如何设超时与限流;PR 里要求说清是否真 async 出站;评测脚本先 Semaphore 再加大并发。后续 FastAPI、Agent,都假设这层已稳。
和专栏一串起来复习也不亏:文件与 with(10)对应 Client 生命周期;异常(11)要在状态码与超时里浮出;上下文(17)对应 async with;asyncio(25)提供编排;本篇是专栏二把「网络请求」说透的一课。
也可以把收获压成检查清单:会读状态码;会 requests/httpx;会 AsyncClient + Semaphore;会 SSE 流式;会拼兼容 Chat Completions;密钥与超时纪律到位。六项过关,即可进入 FastAPI 篇。
若把本篇压成给同事的三十秒讲解:先画「请求→响应」;再强调「超时与限流」;最后提「async 里禁止同步 requests」。三张草图比贴一篇博客更管用。等对方能独立把串行脚本改成有界 AsyncClient,并解释 429 为何要退避,就可以一起 code review 出站 PR 了。
动手清单再压缩一版:requests.get + raise_for_status;AsyncClient + gather;假/真 Chat Completions;流式 iter_lines;Semaphore + 退避。五步做完,比只「看完文章」扎实一个数量级。
九、课后练习
- (基础) 用
requests.get访问https://httpbin.org/get,打印状态码与args;故意设timeout=0.001观察异常类型。 - (基础) 用
requests.Session带自定义 Header 连续请求两次https://httpbin.org/headers,确认 Header 被带回。 - (基础) 用
httpx.Client对https://httpbin.org/post发 JSON,断言回显的json字段。 - (理解) 把第 25 篇三个
asyncio.sleep示例改成三个httpx.AsyncClientGET(可指向 httpbin),对照串行 await 与gather的墙钟。 - (理解) 用
client.stream+iter_lines读取https://httpbin.org/stream/5,统计非空行数。 - (理解) 写一个函数:从环境变量读 Key / Base URL,发非流式
chat/completions;无 Key 时清晰跳过并返回提示,有 Key 时打印content与usage。 - (进阶) 实现
Semaphore(3)+ 指数退避:对 8 个假任务(或真 API)并发调用,模拟一次 429 并重试成功。 - (进阶 · AI Infra) 画一张你当前项目的出站图:哪些用同步、哪些用 AsyncClient、限流值多少、超时如何分层;写 200 字说明「假异步出站」的危害与改法。
下一篇预告
下一篇是《第27篇|FastAPI 写一个推理服务接口》——在搞清「客户端如何发 HTTP、如何打大模型」之后,站到对面:用 FastAPI 暴露一个可被本篇客户端调用的推理接口。你会看到:路由与 Pydantic 模型、同步/异步端点、流式响应返回、与 OpenAI 兼容形状的一点点对齐,以及如何用本篇的 httpx 做自测。
专栏二后续路线:FastAPI → 最小 Agent。带着本篇的超时、限流与兼容协议纪律,写服务时记得:入站校验用模型;出站仍守 Client 纪律;流式两端要配对。
若想提前热身:把本篇非流式 / 流式 Chat Completions 客户端保存成小模块,下篇直接拿它打你自己写的 FastAPI——对接时你会带着具体体感。
也可以把第 23~26 篇对照:线程叠同进程 I/O、进程扛 CPU、协程叠海量等待、本篇把等待变成真实 HTTP——工程刀具的轮廓已经清晰,下一篇把「服务端接口」说透。
专栏一用十四篇把「对象与协议」立住;专栏二从迭代器(15)走到 HTTP(26),下一站是 FastAPI。下一篇见。
若时间充裕,再多做一件事:打开你最近用过的一个评测或 Agent 脚本,找到出站 HTTP 代码,记三条「它如何设超时、如何限流、密钥从哪来」。这种阅读比空背参数名更能把纪律刻进肌肉记忆。
进入专栏二至此,可验证面、可观测面、并发面与联网面开始交汇:会自动验回归,会在正确层级叠任务,也会安全地发请求。下一篇开始把「可被调用的服务」配齐。若某一节读着抽象,回到对应插图把箭头走一遍,再手打 AsyncClient 与 Semaphore——多数卡住的地方,其实是「同步脚本习惯」被原样搬进了异步服务。
写出站或读延迟报告时再记一条:错误往往来自无超时与假异步(该 await 却用 requests、该限流却无界 gather)。和「栈很长但根因很浅」相比,先设超时与有界并发能省很多时间。
也可以把本篇看成专栏二的「出站 HTTP」补课:前面把协议、积木、环境、眼睛、断言与并发收齐,本篇把网络请求收齐。后面 FastAPI 篇教你如何收请求;Agent 篇则把 HTTP 打进工具循环。
若阅读时觉得「requests / httpx / SDK」总搅在一起,回到图 2 与图 4:先定同步还是异步,再定是否走兼容 HTTP。日常编程三者都可能用到,但职责不同——混谈是多数初学者卡住的根源。
收官前对照练习:同一段「打 3 次假 HTTP」,分别用串行 requests、线程池、httpx.AsyncClient + gather。你会清楚感受到:异步客户端不是万能加速器,而是在事件环上叠真实网络等待的工具——也是专栏二与 FastAPI / Agent 对接的入场券之一。
写完练习后,不妨在自己的仓库里开一个小 PR:只把串行出站循环改成有界 AsyncClient + Semaphore,并补上 timeout,不改业务语义。这种「小步联网」比空读文档更能把工作流刻进肌肉记忆,也更适合团队推广。
若你带新人,可以把本篇插图打印成一页速查:左列场景(脚本 / FastAPI 出站 / 流式聊天 / 批量评测),右列策略(requests / AsyncClient / SSE / Semaphore+退避)。新人写出站时对着贴,两周后就会自然说出「先设超时」与「先禁止环内 requests」。工具文化是带出来的,不是靠一次分享会喊出来的。
最后把专栏二的位置再钉一次:协议篇(15~18)解决「怎么正确接线」;标准库与依赖篇(19~20)解决「积木与环境」;可观测与可验证篇(21~22)解决「如何被看见与自动验证」;并发篇(23~25)解决「线程、进程、协程如何分层」;本篇解决「真实 HTTP 与大模型 API 怎么发」;下一篇解决「推理服务接口怎么暴露」。六者齐备,脚本才既写得快,又连得敢——下一篇再给它配上 FastAPI 的实话。
