使用 FastAPI 封装统一的大模型调用服务

系列：Python + 大模型应用开发（第 2 篇）
目标：把大模型调用代码封装成统一的 HTTP 服务，为网页、小程序、企业微信侧边栏和业务系统提供后端接口。

1. 为什么要增加一层后端服务

上一篇中，Python 程序直接调用了模型服务。如果未来需要接入网页、CRM、小程序或企业微信，一种危险做法是让每个客户端都直接携带模型 API Key。

更合理的基础架构是：

网页 / 小程序 / 企业微信 / CRM
              ↓
        自己的 FastAPI 服务
              ↓
     鉴权、参数校验、日志、限流
              ↓
          大模型服务商

增加 FastAPI 层可以解决以下问题：

• 模型密钥只保存在后端；
• 多个业务系统使用统一接口；
• 集中完成输入校验、异常转换和日志记录；
• 后续可以统一增加鉴权、限流、缓存和成本统计；
• 更换模型服务商时，前端不需要跟着修改。

本文使用异步 HTTP 客户端 httpx.AsyncClient 调用模型，避免在 FastAPI 的异步接口中使用同步网络请求阻塞事件循环。

2. 目标接口

我们准备提供两个接口：

方法	路径	用途
GET	/health	判断当前 FastAPI 进程是否正常运行
POST	/api/v1/chat	接收用户问题并返回模型回答

聊天请求：

{
  "user_message": "请解释 Python 装饰器。",
  "temperature": 0.2
}

聊天响应：

{
  "content": "模型生成的回答",
  "model": "your-model-id",
  "usage": {
    "prompt_tokens": 20,
    "completion_tokens": 80,
    "total_tokens": 100
  }
}

usage 是否存在、包含哪些字段，由实际模型服务决定，因此代码必须允许它为 null。

3. 创建项目

本文使用 Python 3.10 及以上版本。

项目结构：

llm_fastapi_service/
├── app/
│   ├── __init__.py
│   ├── config.py
│   ├── schemas.py
│   ├── llm_client.py
│   └── main.py
└── requirements.txt

创建虚拟环境：

python -m venv .venv
.\.venv\Scripts\python.exe -m pip install --upgrade pip

requirements.txt：

fastapi>=0.115,<1
uvicorn[standard]>=0.30,<1
httpx>=0.27,<1
pydantic>=2.7,<3

安装依赖：

.\.venv\Scripts\python.exe -m pip install -r requirements.txt

app/__init__.py 可以是空文件，它表示 app 是一个 Python 包。

4. 编写配置模块

新建 app/config.py：

import os
from dataclasses import dataclass


@dataclass(frozen=True)
class Settings:
    """保存大模型服务配置。"""

    api_key: str
    base_url: str
    model: str


def load_settings() -> Settings:
    """读取环境变量，并尽早发现缺失配置。"""

    # 所有敏感配置都从运行环境读取，不写入代码仓库
    api_key = os.getenv("LLM_API_KEY", "").strip()
    base_url = os.getenv("LLM_BASE_URL", "").strip().rstrip("/")
    model = os.getenv("LLM_MODEL", "").strip()

    missing_variables = []
    if not api_key:
        missing_variables.append("LLM_API_KEY")
    if not base_url:
        missing_variables.append("LLM_BASE_URL")
    if not model:
        missing_variables.append("LLM_MODEL")

    if missing_variables:
        names = ", ".join(missing_variables)
        raise RuntimeError(f"缺少环境变量：{names}")

    # 远程传输密钥时必须使用 HTTPS
    if not base_url.startswith("https://"):
        raise RuntimeError("LLM_BASE_URL 必须使用 https:// 地址")

    return Settings(
        api_key=api_key,
        base_url=base_url,
        model=model,
    )

设置环境变量：

$env:LLM_API_KEY = "替换为真实密钥"
$env:LLM_BASE_URL = "https://替换为模型服务地址/v1"
$env:LLM_MODEL = "替换为真实模型标识"

这些示例值不能直接使用，必须根据实际服务商文档填写。

5. 使用 Pydantic 定义数据结构

新建 app/schemas.py：

from typing import Any

from pydantic import BaseModel, Field, field_validator


class ChatRequest(BaseModel):
    """调用聊天接口时，客户端允许提交的数据。"""

    user_message: str = Field(
        min_length=1,
        max_length=4000,
        description="用户问题，长度为 1 到 4000 个字符",
    )
    temperature: float = Field(
        default=0.2,
        ge=0,
        le=2,
        description="常见的模型随机性参数，真实范围以模型文档为准",
    )

    @field_validator("user_message")
    @classmethod
    def message_must_not_be_blank(cls, value: str) -> str:
        """阻止只包含空格、换行符的无效问题。"""

        cleaned_value = value.strip()
        if not cleaned_value:
            raise ValueError("user_message 不能为空白字符串")

        # 返回清理后的内容，后续业务代码无需重复 strip()
        return cleaned_value


class ChatResponse(BaseModel):
    """聊天接口成功时的统一响应。"""

    content: str
    model: str

    # 不同模型服务商返回的 usage 字段可能不同，因此用 Any 表示值类型
    usage: dict[str, Any] | None = None


class HealthResponse(BaseModel):
    """健康检查接口响应。"""

    status: str

为什么不让调用者提交 system_message？

因为系统提示词属于服务端规则。如果任意前端用户都能修改它，就可能把“严谨的业务助手”改成其他身份，绕过原有业务约束。本文把系统提示词固定在后端，后续如果存在多个业务助手，可以使用服务端维护的 assistant_id 白名单进行选择。

6. 编写异步模型客户端

新建 app/llm_client.py：

from typing import Any

import httpx

from app.config import Settings


class LLMServiceError(RuntimeError):
    """表示调用上游模型服务时发生的可预期错误。"""

    def __init__(self, message: str, http_status: int = 502) -> None:
        super().__init__(message)

        # http_status 是自己的 FastAPI 服务要返回给调用方的状态码
        self.http_status = http_status


class AsyncLLMClient:
    """基于 httpx.AsyncClient 的异步模型客户端。"""

    def __init__(self, settings: Settings) -> None:
        self.settings = settings

        # 60 秒是默认总超时，建立连接最多等待 5 秒
        timeout = httpx.Timeout(60.0, connect=5.0)

        # AsyncClient 应在应用生命周期内复用，不能每次请求都重新创建
        self.http_client = httpx.AsyncClient(timeout=timeout)

    async def chat(
        self,
        user_message: str,
        system_message: str,
        temperature: float,
    ) -> tuple[str, dict[str, Any] | None]:
        """异步调用模型，返回回答文本和可选的 Token 用量。"""

        request_url = f"{self.settings.base_url}/chat/completions"
        request_headers = {
            "Authorization": f"Bearer {self.settings.api_key}",
            "Content-Type": "application/json",
        }
        request_body = {
            "model": self.settings.model,
            "messages": [
                {"role": "system", "content": system_message},
                {"role": "user", "content": user_message},
            ],
            "temperature": temperature,
        }

        try:
            # await 表示当前协程等待网络结果时，可以让出执行权
            response = await self.http_client.post(
                request_url,
                headers=request_headers,
                json=request_body,
            )
        except httpx.TimeoutException as exc:
            # 504 表示作为网关等待上游服务超时
            raise LLMServiceError(
                "等待模型服务响应超时",
                http_status=504,
            ) from exc
        except httpx.ConnectError as exc:
            raise LLMServiceError(
                "无法连接模型服务",
                http_status=502,
            ) from exc
        except httpx.HTTPError as exc:
            raise LLMServiceError(
                "调用模型服务时发生网络异常",
                http_status=502,
            ) from exc

        # 上游鉴权失败属于后端配置问题，不向前端暴露密钥细节
        if response.status_code == 401:
            raise LLMServiceError("模型服务鉴权失败", http_status=502)

        # 自己的服务当前无法满足请求，统一转换成 503
        if response.status_code == 429:
            raise LLMServiceError(
                "模型服务繁忙或额度受限，请稍后重试",
                http_status=503,
            )

        if 400 <= response.status_code < 500:
            raise LLMServiceError(
                f"模型服务拒绝了请求，上游状态码：{response.status_code}",
                http_status=502,
            )

        if response.status_code >= 500:
            raise LLMServiceError(
                "模型服务暂时不可用",
                http_status=503,
            )

        try:
            # httpx 的 response.json() 会把 JSON 转成 Python 对象
            data: dict[str, Any] = response.json()
        except ValueError as exc:
            raise LLMServiceError("模型服务返回了无效 JSON") from exc

        try:
            content = data["choices"][0]["message"]["content"]
        except (KeyError, IndexError, TypeError) as exc:
            raise LLMServiceError("模型响应缺少预期字段") from exc

        if not isinstance(content, str) or not content.strip():
            raise LLMServiceError("模型返回了空内容")

        # usage 不是所有服务都提供，因此使用 get() 安全读取
        raw_usage = data.get("usage")
        usage = raw_usage if isinstance(raw_usage, dict) else None

        return content.strip(), usage

    async def close(self) -> None:
        """关闭异步客户端，释放连接池资源。"""

        await self.http_client.aclose()

为什么 FastAPI 中使用异步客户端

模型请求的大部分时间都消耗在网络等待上。如果异步接口内部使用阻塞式网络请求，事件循环会被阻塞，其他请求也可能受到影响。

这里的关键不是把函数名称前面简单加上 async，而是内部网络库本身也要支持异步，并且在调用时使用 await。

7. 创建 FastAPI 应用

新建 app/main.py：

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from app.config import Settings, load_settings
from app.llm_client import AsyncLLMClient, LLMServiceError
from app.schemas import ChatRequest, ChatResponse, HealthResponse


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """管理应用启动和关闭时需要创建、释放的资源。"""

    # 启动阶段读取配置；配置有误时服务会直接启动失败
    settings = load_settings()

    # 创建一个供整个应用复用的异步模型客户端
    llm_client = AsyncLLMClient(settings)

    # 将对象放入 app.state，路由函数可以通过 Request 取得它们
    app.state.settings = settings
    app.state.llm_client = llm_client

    # yield 之前是启动逻辑，yield 之后是关闭逻辑
    yield

    # 服务关闭时释放 HTTP 连接池
    await llm_client.close()


app = FastAPI(
    title="统一大模型调用服务",
    version="1.0.0",
    lifespan=lifespan,
)


@app.exception_handler(LLMServiceError)
async def handle_llm_service_error(
    request: Request,
    exc: LLMServiceError,
) -> JSONResponse:
    """把内部模型异常转换成统一且可理解的 HTTP 响应。"""

    # request 在当前示例中未读取，后续可用它取得请求 ID 或用户身份
    return JSONResponse(
        status_code=exc.http_status,
        content={"detail": str(exc)},
    )


@app.get("/health", response_model=HealthResponse)
async def health() -> HealthResponse:
    """判断当前 FastAPI 进程是否正常运行。"""

    # 这里只验证自己的服务进程，不代表上游模型一定可用
    return HealthResponse(status="running")


@app.post("/api/v1/chat", response_model=ChatResponse)
async def chat(
    chat_request: ChatRequest,
    request: Request,
) -> ChatResponse:
    """接收用户问题，通过统一模型客户端生成回答。"""

    # 系统提示词由服务端控制，调用者不能通过请求任意覆盖
    system_message = (
        "你是一名严谨的 Python 教师。"
        "请基于事实回答；不确定时明确说明，不要编造。"
    )

    # 从应用状态中取得启动时创建的共享对象
    llm_client: AsyncLLMClient = request.app.state.llm_client
    settings: Settings = request.app.state.settings

    # 等待异步模型调用完成
    content, usage = await llm_client.chat(
        user_message=chat_request.user_message,
        system_message=system_message,
        temperature=chat_request.temperature,
    )

    # response_model 会再次校验接口输出结构
    return ChatResponse(
        content=content,
        model=settings.model,
        usage=usage,
    )

8. 启动并测试服务

在项目根目录执行：

.\.venv\Scripts\python.exe -m uvicorn app.main:app --reload

--reload 会在代码变化后自动重启，只适合本地开发，不应直接作为生产环境启动方式。

浏览器访问：

http://127.0.0.1:8000/docs

FastAPI 会生成交互式 API 文档，可以直接测试 /health 和 /api/v1/chat。

也可以使用 PowerShell 调用聊天接口：

# 使用哈希表构造请求体，再转换为 JSON
$body = @{
    user_message = "请用三个要点解释 Python 列表和元组的区别"
    temperature = 0.2
} | ConvertTo-Json

# 调用自己的 FastAPI 服务，而不是让前端直接调用模型厂商
Invoke-RestMethod `
    -Method Post `
    -Uri "http://127.0.0.1:8000/api/v1/chat" `
    -ContentType "application/json" `
    -Body $body

9. FastAPI 自动完成了哪些工作

当请求到达 /api/v1/chat 时，FastAPI 和 Pydantic 会：

1. 读取 JSON 请求体；
2. 检查 user_message 是否存在；
3. 检查字符串长度；
4. 拦截只有空格的输入；
5. 检查 temperature 是否在规定范围内；
6. 把合法数据转换成 ChatRequest 对象；
7. 使用 ChatResponse 校验响应结构；
8. 自动生成 OpenAPI 接口文档。

例如，请求中把 temperature 设置为 5，FastAPI 会直接返回 422，而不会继续调用模型并产生费用。

10. 为什么不能每个请求都创建 AsyncClient

下面的写法虽然可能运行，但不适合高频调用：

# 不推荐：每个业务请求都创建和关闭新的连接池
async with httpx.AsyncClient() as client:
    response = await client.post("模型地址")

本文通过 lifespan 在应用启动时创建一次 AsyncClient，在关闭服务时统一释放。这样可以复用连接池，职责也更加清晰。

11. 对抗性审查：当前服务还有哪些风险

1. 自己的接口还没有身份认证

当前示例用于本地学习，任何能够访问该端口的人都可以调用接口并消耗模型额度。生产环境至少需要用户身份认证、权限校验和调用额度控制。

2. 健康检查不代表模型可用

/health 只说明 FastAPI 进程在运行。若要检查模型服务是否可用，应单独设计 Readiness（就绪检查），并谨慎控制检查频率，避免不断产生模型费用。

3. 错误信息不能泄露敏感数据

不应直接把上游完整响应、请求头、API Key 或内部堆栈返回给前端。详细异常可以写入经过脱敏的内部日志，对外只返回必要信息。

4. 输入长度限制不等于成本控制

字符数与 Token 数不是完全相同的概念。生产服务还需要根据实际模型的计费和上下文限制统计 Token，并设置用户级额度。

5. 系统提示词不是安全边界

把系统提示词保存在后端可以减少被直接修改的风险，但不能只依赖一句 Prompt 实现权限控制。数据库查询、文件访问和外部操作必须在代码层执行身份认证和权限判断。

6. 模型回答仍然可能错误

HTTP 200 只代表服务正常处理了请求，不代表回答符合事实。高风险业务需要增加 RAG、规则校验、结构化输出和人工审核。

12. 下一步如何扩展

这个统一服务可以继续增加：

• API 用户鉴权；
• 请求 ID 和结构化日志；
• 429、5xx 的有限重试；
• Redis 限流和缓存；
• Token 用量与成本统计；
• 流式输出 SSE（服务器发送事件）；
• 多轮对话和历史消息存储；
• 多模型适配器；
• RAG 企业知识库；
• Prompt 和模型效果评测。

13. 总结

本文完成了从“Python 直接调用模型”到“统一后端模型服务”的升级：

客户端
  ↓
FastAPI 参数校验
  ↓
服务端系统规则
  ↓
异步模型客户端
  ↓
模型服务
  ↓
统一异常与响应结构

真正有价值的不是多包装了一层接口，而是建立了明确的系统边界：密钥属于后端、规则由服务端控制、外部输入必须校验、上游错误需要转换、网络资源需要统一管理。

14. 练习题

1. 将 user_message 最大长度改为 2000，并测试超长请求；
2. 将 temperature 设置为 3，观察 FastAPI 返回的 422；
3. 删除一个必需环境变量，观察服务如何在启动阶段失败；
4. 为 /health 增加当前服务版本字段；
5. 为成功响应增加一个由后端生成的 request_id；
6. 思考：如果接入三个模型厂商，怎样避免在路由函数中编写大量 if/else？