01. Web 应用基础:HTTP、API、异步
读完这篇你能:
- 用 FastAPI 写一个能跑的 API,浏览器/curl 能调通
- 看懂一次 HTTP 请求的完整流程(method、URL、header、body、status code)
- 知道为什么 LLM 调用必须用 async,会写
async def对应架构层:应用层(入门)
前置知识:会 Python,调过
openai.ChatCompletion.create预计用时:2 小时(含动手)
0. 为什么算法工程师要学这个
你写过这样的代码:
# notebook 里
response = openai.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "总结:..."}]
)
print(response.choices[0].message.content)
这是脚本——一次执行、一个用户、你自己看着结果。
但用户用的不是 notebook。用户在浏览器里点一下"生成"按钮,他的浏览器要通过互联网把请求送到你的服务器,你的服务器再调 OpenAI,拿到结果,再通过互联网送回他的浏览器。
这个"通过互联网交换数据"的协议,叫 HTTP。 你这个"在服务器上等请求"的程序,叫 Web 服务。
这一篇就是讲这两件事。学会了你才能:
- 把 notebook 里的代码变成"任何人在浏览器里都能调"的 API
- 理解为什么 LLM 调用要异步(不然 100 个用户同时点,你的服务卡死)
- 看懂后端工程师在说什么("这个接口是 POST,body 是 JSON,返回 200")
1. 核心概念(最少必要理论)
1.1 前端、后端、全栈
| 角色 | 干什么 | 技术栈 |
|---|---|---|
| 前端 | 用户看到的部分:按钮、表单、动画 | HTML/CSS/JS、React/Vue |
| 后端 | 用户看不到的部分:存数据、调 LLM、算结果 | Python、Go、Java、Node.js |
| 全栈 | 两个都做 | 一个人干两人的活 |
算法工程师大多要做"后端为主 + 前端够用"。前端可以用 Gradio / Streamlit / Vercel v0 快速搭,后端必须自己写。
1.2 HTTP 协议一次请求的全过程
当你在浏览器输入 https://api.example.com/summarize 按下回车,发生了什么:
浏览器(客户端) 服务器(你的 FastAPI)
│ │
│ 1. DNS 解析:example.com → 1.2.3.4 │
│ 2. TCP 三次握手 │
│ 3. TLS 握手(HTTPS) │
│ │
│ 4. 发送 HTTP 请求 ────────────────▶ │
│ ┌──────────────────────────────┐ │
│ │ POST /summarize HTTP/1.1 │ │
│ │ Host: api.example.com │ │
│ │ Content-Type: application/json│ │
│ │ Authorization: Bearer xxx │ │
│ │ │ │
│ │ {"text":"长文本..."} │ │
│ └──────────────────────────────┘ │
│ │
│ │ 5. FastAPI 路由匹配
│ │ 6. 调函数处理
│ │ 7. 调 OpenAI
│ │ 8. 拿到结果
│ │
│ ◀──────────────── 9. 返回 HTTP 响应 │
│ ┌──────────────────────────────┐ │
│ │ HTTP/1.1 200 OK │ │
│ │ Content-Type: application/json│ │
│ │ │ │
│ │ {"summary":"摘要..."} │ │
│ └──────────────────────────────┘ │
│ │
│ 10. 浏览器渲染结果 │
记四个关键部分:
- Method(方法):
GET/POST/PUT/DELETE,告诉服务器"我要干什么" - URL(路径):
/summarize,告诉服务器"在哪个接口干" - Header(头):
Content-Type: application/json,元信息 - Body(体):
{"text":"..."},真正的数据
1.3 HTTP Method(语义)
| Method | 含义 | 例子 | 有 body? |
|---|---|---|---|
| GET | 获取资源 | GET /users/123 | 一般没有 |
| POST | 创建资源 | POST /users(body: 新用户信息) | 有 |
| PUT | 全量更新资源 | PUT /users/123(body: 完整用户) | 有 |
| PATCH | 部分更新 | PATCH /users/123(body: {"name":"新名字"}) | 有 |
| DELETE | 删除资源 | DELETE /users/123 | 一般没有 |
记忆法:CRUD(Create / Read / Update / Delete)对应 POST / GET / PUT PATCH / DELETE。
1.4 HTTP Status Code(状态码)
服务器返回的"结果状态"。记住这五类:
| 范围 | 含义 | 常见 |
|---|---|---|
| 2xx | 成功 | 200 OK、201 Created、204 No Content |
| 3xx | 重定向 | 301 永久跳转、302 临时跳转 |
| 4xx | 客户端错了 | 400 Bad Request、401 Unauthorized、403 Forbidden、404 Not Found、429 Too Many Requests |
| 5xx | 服务器错了 | 500 Internal Error、502 Bad Gateway、503 Service Unavailable |
最重要的区分:
- 401 Unauthorized:你没登录(token 没传或失效)
- 403 Forbidden:你登录了,但没权限(普通用户想删别人的数据)
- 500 Internal Error:你的代码抛异常了
1.5 JSON 和 Schema
HTTP body 最常用的格式是 JSON(JavaScript Object Notation)。长得像 Python dict:
{
"user_id": 123,
"name": "sanbu",
"skills": ["python", "pytorch"],
"metadata": {"created_at": "2026-06-23"}
}
为什么用 JSON 不用 Python pickle?
因为 JSON 是跨语言的——Python、JS、Go、Java 都能读写。pickle 只有 Python 能读。
Schema 是什么?
Schema 是"数据长什么样的契约"。比如"summarize 接口的 body 必须有 text 字段,字符串,长度 ≥ 10"。
Python 里用 Pydantic 定义 Schema:
from pydantic import BaseModel, Field
class SummarizeRequest(BaseModel):
text: str = Field(..., min_length=10, description="要总结的长文本")
max_length: int = Field(200, ge=50, le=1000, description="摘要最大长度")
class SummarizeResponse(BaseModel):
summary: str
tokens_used: int
Field(..., min_length=10) 表示必填、最小长度 10。FastAPI 自动校验,不通过直接返回 422。
1.6 REST API 设计原则
REST(Representational State Transfer)是一套 API 设计约定,不是强制的协议。核心原则:
- URL 表示资源:
/users、/users/123、/users/123/orders - Method 表示动作:
GET /users/123(读)、DELETE /users/123(删) - 用 HTTP 状态码表示结果:200 成功、404 没找到、500 服务器错
- 无状态:每个请求自带身份(token),服务器不记 session
反例(不要这么写):
POST /getUserById?id=123 ❌ URL 里有动词
POST /deleteUser ❌ URL 里有 delete 动词
GET /users?op=delete&id=123 ❌ 用 GET 做删除操作
正例:
GET /users/123 ✅ 获取 id=123 的用户
POST /users ✅ 创建新用户
DELETE /users/123 ✅ 删除 id=123 的用户
1.7 同步 vs 异步(算法工程师最容易卡的地方)
同步(sync):
def summarize(text):
result = openai.chat.completions.create(...) # 阻塞!等 3 秒
return result
# 100 个用户同时调 → 排队 → 第 100 个用户等 300 秒
异步(async):
async def summarize(text):
result = await openai.chat.completions.create(...) # 不阻塞!让出 CPU
return result
# 100 个用户同时调 → 并发等待 → 几乎同时拿到结果
为什么异步更快?
LLM 调用 99% 时间在等网络(等 OpenAI 服务器返回),CPU 闲着。同步代码让 CPU 一起干等。异步代码让 CPU 在等待时去服务下一个请求。
类比 GPU:同步像 batch_size=1 跑 100 次推理,异步像 batch_size=100 跑 1 次推理——同样是 100 个样本,异步吞吐量高 100 倍。
Python 异步的关键字:
async def:声明异步函数await:等一个异步操作完成(等待时让出 CPU)
最重要的规则:LLM API、数据库、HTTP 调用都要用 async 客户端。如果你在 async 函数里调同步的 requests.get,整个事件循环卡死。
# ❌ 错误:async 函数里调 sync 库
async def bad():
resp = requests.get("https://api.openai.com/...") # 卡死事件循环!
# ✅ 正确:async 函数里调 async 库
async def good():
resp = await httpx.AsyncClient().get("https://api.openai.com/...")
2. 实战:用 FastAPI 写一个能跑的 API
2.1 装环境
# 创建项目目录
mkdir my-llm-app && cd my-llm-app
# 建虚拟环境
python -m venv venv
source venv/bin/activate # Mac/Linux
# venv\Scripts\activate # Windows
# 装包
pip install fastapi uvicorn[standard] httpx pydantic openai python-dotenv
把 OpenAI key 放进 .env(不要写死在代码里):
echo 'OPENAI_API_KEY=sk-...' > .env
echo '.env' > .gitignore
2.2 Hello World
新建 main.py:
from fastapi import FastAPI
app = FastAPI(title="My LLM App")
@app.get("/")
def root():
return {"message": "Hello, World!"}
启动:
uvicorn main:app --reload --port 8000
main:app=main.py里的app变量--reload= 改代码自动重启(开发用)--port 8000= 监听 8000 端口
测试:
curl http://localhost:8000/
# {"message":"Hello, World!"}
浏览器打开 http://localhost:8000/docs,你会看到 Swagger UI——FastAPI 自动生成的 API 文档。这就是用它的一大理由。
2.3 写一个 /summarize 接口
升级 main.py:
import os
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from openai import AsyncOpenAI
from dotenv import load_dotenv
load_dotenv()
app = FastAPI(title="My LLM App", version="0.1.0")
client = AsyncOpenAI() # 注意是 AsyncOpenAI!
# ====== Schema ======
class SummarizeRequest(BaseModel):
text: str = Field(..., min_length=10, max_length=10000, description="要总结的文本")
max_length: int = Field(200, ge=50, le=1000)
class SummarizeResponse(BaseModel):
summary: str
tokens_used: int
# ====== Routes ======
@app.get("/")
def root():
return {"message": "Hello, World!"}
@app.get("/health")
def health():
return {"status": "ok"}
@app.post("/api/summarize", response_model=SummarizeResponse)
async def summarize(req: SummarizeRequest):
try:
resp = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": f"用不超过{req.max_length}字总结以下文本。"},
{"role": "user", "content": req.text},
],
)
return SummarizeResponse(
summary=resp.choices[0].message.content,
tokens_used=resp.usage.total_tokens,
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
关键点解读:
AsyncOpenAI()用异步客户端,await client.chat.completions.create(...)不阻塞@app.post("/api/summarize")声明 POST 路由req: SummarizeRequest—— FastAPI 自动把 body 反序列化成 Pydantic 对象,自动校验response_model=SummarizeResponse—— 返回值自动按 Schema 序列化HTTPException抛出后,FastAPI 自动返回对应状态码
2.4 测试接口
启动服务:
uvicorn main:app --reload --port 8000
方式一:curl
curl -X POST http://localhost:8000/api/summarize \
-H "Content-Type: application/json" \
-d '{"text": "FastAPI 是一个现代的 Python Web 框架,基于 Starlette 和 Pydantic。它由 Sebastián Ramírez 创建,主打高性能、易用性和自动文档生成。FastAPI 的设计哲学是让开发者用最少的代码完成最多的功能,同时保持类型安全和性能。", "max_length": 100}'
返回:
{
"summary": "FastAPI 是基于 Starlette 和 Pydantic 的现代 Python Web 框架...",
"tokens_used": 187
}
方式二:Swagger UI
浏览器打开 http://localhost:8000/docs,点击 /api/summarize → "Try it out" → 填 body → "Execute"。
方式三:VS Code REST Client 插件
新建 test.http:
### 测试 summarize
POST http://localhost:8000/api/summarize
Content-Type: application/json
{
"text": "FastAPI 是一个现代的 Python Web 框架...",
"max_length": 100
}
文件上方点 "Send Request"。
2.5 试试错误场景
参数校验失败:
curl -X POST http://localhost:8000/api/summarize \
-H "Content-Type: application/json" \
-d '{"text": "短"}'
返回 422,body 里告诉用户"text 太短":
{
"detail": [
{
"type": "string_too_short",
"loc": ["body", "text"],
"msg": "String should have at least 10 characters",
"input": "短"
}
]
}
这就是 Pydantic 的威力——你写一遍 Schema,自动校验、自动报错、自动文档,完全不用手写。
3. 进阶:生产级改造
刚才的代码能跑,但离生产级还差几件事。
3.1 CORS(跨域资源共享)
前端跑在 http://localhost:3000,后端跑在 http://localhost:8000——浏览器认为这是跨域,默认拦截。需要后端明确允许:
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "https://yourapp.com"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
生产环境:allow_origins 不要写 ["*"],要明确列出允许的前端域名。
3.2 配置管理(不要把 key 写死)
用 Pydantic Settings:
# config.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
openai_api_key: str
openai_model: str = "gpt-4o-mini"
max_text_length: int = 10000
class Config:
env_file = ".env"
settings = Settings()
# main.py
from config import settings
client = AsyncOpenAI(api_key=settings.openai_api_key)
好处:
- 配置有类型
- 默认值
- 不同环境(dev/staging/prod)用不同
.env文件
3.3 全局异常处理
不要每个接口都 try/except。注册全局处理器:
from fastapi import Request
from fastapi.responses import JSONResponse
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
# 这里应该记日志,不只是返回
return JSONResponse(
status_code=500,
content={"detail": "Internal Server Error", "message": str(exc)},
)
3.4 结构化日志
算法工程师习惯 print()。生产环境要结构化日志(JSON 格式),方便后续聚合查询。
import structlog
logger = structlog.get_logger()
@app.post("/api/summarize")
async def summarize(req: SummarizeRequest):
logger.info("summarize_request", text_length=len(req.text))
try:
# ...
logger.info("summarize_success", tokens=resp.usage.total_tokens)
return ...
except Exception as e:
logger.exception("summarize_failed", error=str(e))
raise
详细配置见 06. 可观测性。
3.5 启动和关闭事件(生命周期)
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# 启动时:连数据库、预热缓存
logger.info("app_starting")
yield
# 关闭时:清理资源
logger.info("app_shutting_down")
app = FastAPI(lifespan=lifespan)
4. 常见踩坑(算法工程师常犯)
坑 1:async 函数里用 requests 库
# ❌ 阻塞事件循环
async def bad():
resp = requests.get("https://api.openai.com/...")
# ✅ 用 async 客户端
async def good():
async with httpx.AsyncClient() as client:
resp = await client.get("https://api.openai.com/...")
症状:服务跑起来 QPS 极低,一个慢请求拖垮整个服务。
坑 2:把 API key 硬编码
# ❌ 推到 GitHub 立刻被扫走,OpenAI 自动撤销 key,还可能扣光余额
client = OpenAI(api_key="sk-prod-xxx...")
# ✅ 用 .env + 环境变量
client = OpenAI() # 自动读 OPENAI_API_KEY
GitHub 有机器人专门扫公开仓库里的 API key,几分钟就扫到。
坑 3:GET 请求带敏感数据
# ❌ URL 会被浏览器历史、代理日志、CDN 日志记录
@app.get("/api/user")
def get_user(token: str): # token 出现在 URL 里
# ✅ 用 Header 传
@app.get("/api/user")
def get_user(authorization: str = Header()):
坑 4:同步的 LLM 调用写成 async
# ❌ 这是同步调用,在 async 函数里会阻塞
@app.post("/summarize")
async def summarize(req):
resp = openai.chat.completions.create(...) # 注意:OpenAI() 不是 AsyncOpenAI()
记住:AsyncOpenAI 的方法要 await,OpenAI 的方法是同步阻塞。两者不能混用。
坑 5:不设超时
# ❌ LLM 卡住时,请求无限等待,连接被耗尽
resp = await client.chat.completions.create(...)
# ✅ 设超时(默认 OpenAI SDK 是 600 秒,太长)
client = AsyncOpenAI(timeout=30.0)
生产环境 LLM 调用建议 30 秒超时,超了直接返回降级结果或 503。
5. 检查清单
读完这篇,你能勾选下面全部吗?
- 我能解释 GET 和 POST 的区别
- 我知道 200、401、403、404、500、429 分别是什么意思
- 我能用 Pydantic 写一个带字段校验的 Schema
- 我能用 FastAPI 写一个 POST 接口,自动文档能打开
- 我知道为什么 LLM 调用要用 AsyncOpenAI
- 我会区别
def和async def,知道什么时候用await - 我用 curl / Postman / Swagger 测过自己的接口
- 我知道 CORS 是干嘛的,会配置
- 我把 API key 放在
.env里,没硬编码 - 我会给 LLM 调用设超时
全部勾选 → 进入下一篇,加数据库。
6. 下一步
这篇你写了应用层最基础的 Hello World。下一篇我们要去数据层——给应用装上"记忆":
- 02. 数据库与持久化:用 PostgreSQL + SQLModel 把用户数据、查询历史存下来
- 03. 缓存与对象存储:用 Redis 加速、用 S3 存文件
学完 02 + 03,你的应用就具备了完整的存储层,可以开始考虑认证、权限、流式(04 篇)。
进阶学习资源
- FastAPI 官方教程:
https://fastapi.tiangolo.com/tutorial/(必读,质量很高) - Real Python HTTP 入门:https://realpython.com/python-https/
- Pydantic 文档:https://docs.pydantic.dev/
- HTTP 协议图解:https://developer.mozilla.org/zh-CN/docs/Web/HTTP/Overview
- async/await 详解:https://docs.python.org/3/library/asyncio.html