统一异常 + 统一响应，让接口「说人话」

ScreenShot_2026-07-31_083846_669.png

「从零到 AI 应用工程师」专栏 · 第 4 篇

分层之后，下一个痛点很快出现：

成功时你返回：

{"code": 200, "message": "ok", "data": {...}}

失败时 FastAPI 默认给你：

{"detail": "Not authenticated"}

参数校验失败又是另一套：

{"detail":[{"loc":["body","message"],"msg":"...","type":"..."}]}

前端要写三套解析；你自己用 curl 排查也难受。更糟的是：有的错误把堆栈、SQL、厂商原文甩出去——既不安全，也不专业。

今天目标：不管成功还是失败，外层结构统一；HTTP 状态码语义正确；日志里能用 request_id 把一次请求串起来。

一、先定契约：统一响应长什么样

建议固定四个字段：

{
  "code": 200,
  "message": "ok",
  "data": {},
  "request_id": "req_a1b2c3d4e5f6"
}

字段	含义
`code`	业务/协议状态，建议与 HTTP status 对齐：200 / 400 / 401 / 503 / 500
`message`	给人看的短句，可展示
`data`	成功时的业务数据；失败时可为 `null`
`request_id`	本次请求唯一 ID，方便对日志

重要原则：

统一 JSON ≠ 全部返回 HTTP 200。 401 还是 401，503 还是 503。网关、监控、客户端重试都依赖真实状态码。
对外 message 要克制，对内日志要完整。 用户看到「服务暂时不可用」；你在日志里看到真实异常。

二、三件套：业务异常、中间件、全局处理器

1. 定义响应模型与业务异常

# core/response.py
from pydantic import BaseModel
from typing import Any

class ApiResponse(BaseModel):
    code: int
    message: str
    data: Any = None
    request_id: str | None = None


def success(data: Any = None, message: str = "ok", request_id: str | None = None):
    return ApiResponse(code=200, message=message, data=data, request_id=request_id)


class BusinessError(Exception):
    """可预期的业务错误：对外可展示。"""

    def __init__(self, message: str, status_code: int = 400):
        self.message = message
        self.status_code = status_code

什么时候抛 BusinessError？

入参业务不合法（过短、超限、违规）
依赖临时不可用（模型、缓存）且你想给明确提示
鉴权失败（也可直接 401）

什么时候不要用它包装？

你没预料到的 bug → 让它走到统一的 500 处理，打完整日志。

2. 中间件：尽早生成 request_id

# core/middleware.py
from uuid import uuid4
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request.state.request_id = f"req_{uuid4().hex[:12]}"
        response = await call_next(request)
        response.headers["X-Request-Id"] = request.state.request_id
        return response

成功、失败、日志，都读同一个 request.state.request_id。

3. 全局异常处理：把「乱七八糟」收成一种形状

# core/exception_handlers.py
import logging
from fastapi import Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from core.response import ApiResponse, BusinessError

logger = logging.getLogger(__name__)


def _body(request: Request, status_code: int, message: str):
    request_id = getattr(request.state, "request_id", None)
    return JSONResponse(
        status_code=status_code,
        content=ApiResponse(
            code=status_code,
            message=message,
            data=None,
            request_id=request_id,
        ).model_dump(),
    )


async def business_error_handler(request: Request, exc: BusinessError):
    return _body(request, exc.status_code, exc.message)


async def validation_error_handler(request: Request, exc: RequestValidationError):
    # 不把 loc/type 细节直接甩给前端；需要时可记日志
    logger.info("validation failed: %s", exc.errors())
    return _body(request, 400, "请求参数不合法")


async def unhandled_error_handler(request: Request, exc: Exception):
    request_id = getattr(request.state, "request_id", None)
    logger.exception("unhandled error request_id=%s", request_id)
    return _body(request, 500, "服务器内部错误")

在 main.py 注册：

from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from core.middleware import RequestIdMiddleware
from core.exception_handlers import (
    business_error_handler,
    validation_error_handler,
    unhandled_error_handler,
)
from core.response import BusinessError
from routers import chat

app = FastAPI(title="chat-api")
app.add_middleware(RequestIdMiddleware)
app.add_exception_handler(BusinessError, business_error_handler)
app.add_exception_handler(RequestValidationError, validation_error_handler)
app.add_exception_handler(Exception, unhandled_error_handler)
app.include_router(chat.router)

Router 成功返回也带上 request_id：

from fastapi import Request
from core.response import success

@router.post("/chat", dependencies=[Depends(verify_token)])
async def chat(req: ChatRequest, request: Request):
    result = await chat_service.reply(...)
    return success(result, request_id=request.state.request_id)

Service 里抛业务错误：

from core.response import BusinessError

def validate_business_rules(message: str) -> None:
    if len(message) < 2:
        raise BusinessError("问题过短", status_code=400)

三、HTTP 状态码怎么选（别全用 200）

场景	HTTP	message 示例
成功	200	ok
参数/业务不合法	400	请求参数不合法 / 问题过短
未登录或令牌错	401	未提供有效令牌
模型/依赖不可用	503	模型服务暂时不可用
未知异常	500	服务器内部错误

有人喜欢「HTTP 永远 200，只用 body.code 区分」。在网关、监控、CDN、客户端库面前，这会处处别扭。本专栏建议：HTTP 语义保留，body 再给一份可读信息。

四、验收：三种失败也要「长得一样」

1. 未带令牌 → 401 + 统一结构

curl -i -X POST http://127.0.0.1:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"user_id":"demo","session_id":"s1","message":"hello"}'

预期响应体仍有 code、message、request_id，而不是只有 detail。

2. 非法参数 → 400 + 统一结构

curl -i -X POST http://127.0.0.1:8000/chat \
  -H "Authorization: Bearer dev-token" \
  -H "Content-Type: application/json" \
  -d '{"user_id":"demo","session_id":"s1","message":""}'

3. 成功路径仍带 request_id

curl -i -X POST http://127.0.0.1:8000/chat \
  -H "Authorization: Bearer dev-token" \
  -H "Content-Type: application/json" \
  -d '{"user_id":"demo","session_id":"s1","message":"你好"}'

响应头可有 X-Request-Id，body 里也有同值 request_id。拿这个 ID 去日志里搜，应能找到同一次请求的记录。

五、常见坑

成功统一了，异常忘了处理 RequestValidationError——最容易漏，因为默认异常不是你的 BusinessError。
把异常字符串 / SQL / 厂商响应原文返回给用户。
每个路由手写 try/except——格式必然漂移；用全局处理器。
code 写 200，HTTP 却是 500——监控会误判。
request_id 只在成功时生成——恰恰失败时你更需要它。

六、带走这三条

统一响应解决客户端契约；HTTP 状态码解决协议语义——两手都要。
可预期错误对外可展示；未知错误对内打满日志、对外保持克制。
request_id 是响应和日志之间的钉子，没有它，线上排障全靠猜。

下一篇给对话加上「记忆」：用 PostgreSQL 把每次问答落库，并提供分页查询历史。缓存可以省钱，但历史是事实，不能省。

这是专栏第 4 篇。接口开始说人话了。两天一更，下篇见。