这次我们来看一个面向 LLM 开发者的 FastAPI 实战教程。如果你正在寻找一个能快速上手、性能出色并且能轻松构建 LLM 应用后端的 Python 框架FastAPI 几乎是当前最直接的选择。它不只是一个 Web 框架更是连接你的创意与大语言模型LLM能力的桥梁从简单的文本生成接口到复杂的 Agent 系统都能高效支撑。本文的核心是“能用”和“怎么用”。我们将跳过冗长的理论铺垫直接进入实战。你会看到如何用 FastAPI 快速搭建一个可运行的 LLM 服务如何处理 prompt 输入与流式输出如何设计 API 以适配不同的模型以及如何应对开发中常见的坑比如请求体验证、并发处理和错误处理。无论你是想为自研的 LLM 应用提供一个可靠的 HTTP 接口还是希望将 OpenAI、通义千问等模型的 API 进行二次封装和增强FastAPI 都能提供一套简洁而强大的工具。本文适合有一定 Python 基础希望快速将 LLM 想法落地为可访问服务的开发者。我们将从环境搭建、第一个 API 创建开始逐步深入到 LLM 集成、流式响应、Prompt 工程化以及项目结构优化最终形成一个可用于实际项目的基础框架。1. 核心能力速览在深入代码之前我们先快速了解 FastAPI 在 LLM 开发场景下的核心优势与关键特性这能帮助你判断它是否适合你的项目。能力项说明与 LLM 开发关联开发速度极快。基于 Python 类型提示Type Hints自动生成 API 文档Swagger UI/ReDoc减少大量手写文档与验证代码的时间让你专注于 LLM 业务逻辑。性能表现基于 Starlette异步和 Pydantic数据验证性能媲美 Node.js 和 Go。对于高并发的 LLM API 请求尤其是流式输出至关重要。异步支持原生支持async/await完美处理 LLM 模型推理通常是 I/O 密集型的等待时间提高服务器并发能力。依赖注入系统强大的依赖注入机制可以优雅地管理 LLM 模型实例、数据库连接、认证信息等使代码更清晰、更易测试。数据验证与序列化通过 Pydantic 自动进行请求/响应数据的验证与转换确保发送给 LLM 的 prompt 格式正确并安全地返回结构化结果。自动 API 文档启动服务后自动在/docs(Swagger) 和/redoc提供交互式文档。前端或测试人员可以直接在浏览器中调用你的 LLM 接口极大提升联调效率。学习门槛较低。如果你熟悉 Python特别是类型提示上手非常快。官方文档清晰社区活跃。适合的 LLM 场景1. 封装第三方 LLM API如 OpenAI、Claude。2. 部署本地开源模型通过 transformers 等库。3. 构建 LLM Agent 或工作流的后端服务。4. 开发基于 Prompt 的各类应用写作助手、代码生成、数据分析。2. 适用场景与使用边界FastAPI 是一个通用 Web 框架但在 LLM 开发领域其特性被放大了。明确它的适用场景和边界能帮助你做出更好的技术选型。它非常适合以下场景快速原型验证你有了一个 LLM 应用的想法比如一个智能客服接口需要最快速度搭建一个后端服务来演示和测试。FastAPI 的快速开发特性让你在几小时内就能看到可运行的 API。生产级 API 服务你需要为移动端、网页端或其他服务提供一个稳定、高性能的 LLM 接口。FastAPI 的异步特性、数据验证和自动文档非常适合构建易于维护和协作的生产接口。复杂 LLM 工作流你的应用涉及多步推理、工具调用Agent、或需要串联多个模型。FastAPI 的路由和依赖注入可以很好地组织这些复杂逻辑。统一 API 网关你可能同时使用多个 LLM 提供商OpenAI、Azure、本地模型。可以用 FastAPI 构建一个统一的网关处理认证、计费、日志、负载均衡和格式转换。需要注意的边界与考量并非机器学习框架FastAPI 本身不提供模型训练或推理功能。你需要集成像transformers、langchain、openai这样的库来实际调用 LLM。WebSocket 支持虽然 FastAPI 支持 WebSocket适用于实时对话场景但其核心优势仍在 HTTP/HTTPS API。对于超大规模、全双工的实时流可能需要结合更专业的网关。超大规模部署对于日调用量亿级以上的场景虽然 FastAPI 性能优秀但整体架构还需要考虑 API 网关、负载均衡、服务发现、容器化等云原生技术栈。前端渲染FastAPI 主要用于构建 API 后端。如果你需要复杂的用户界面通常需要搭配前端框架如 React, Vue或使用专门的模板引擎。3. 环境准备与前置条件开始编码前确保你的开发环境已经就绪。以下是 LLM 开发场景下推荐的基础环境配置。1. 操作系统推荐Linux (Ubuntu 20.04/22.04 LTS) 或 macOS。生产环境部署首选 Linux。也可用Windows 10/11建议使用 WSL2 以获得接近 Linux 的开发体验。2. Python 版本必须Python 3.8 或更高版本。FastAPI 充分利用了 Python 3.6 的类型提示特性3.8 以上版本能获得最佳兼容性。检查命令python --version # 或 python3 --version3. 包管理工具推荐使用pip并搭配虚拟环境venv或conda以隔离项目依赖。创建虚拟环境# 使用 venv python -m venv venv # 激活 (Linux/macOS) source venv/bin/activate # 激活 (Windows) venv\Scripts\activate4. 基础依赖核心依赖就是fastapi和异步服务器uvicorn。安装命令pip install fastapi uvicorn可选但推荐python-multipart用于处理表单数据如上文件httpx用于在异步代码中发出 HTTP 请求例如调用外部 LLM API。5. LLM 相关依赖按需安装调用 OpenAI 等云端 APIpip install openai使用 LangChain 框架pip install langchain langchain-openai部署本地 Hugging Face 模型pip install transformers torch accelerate注意本地模型部署对硬件GPU 显存有要求需根据模型大小准备相应资源。6. 代码编辑器/IDE任何你熟悉的即可如 VS Code推荐对 Python 和 FastAPI 支持好、PyCharm 等。4. 第一个 FastAPI 应用与 LLM “Hello World”让我们从一个最简单的例子开始感受 FastAPI 的便捷并立即将其与 LLM 联系起来。4.1 创建项目文件创建一个名为main.py的文件输入以下代码from fastapi import FastAPI from pydantic import BaseModel # 1. 创建 FastAPI 应用实例 app FastAPI(titleLLM FastAPI Demo, description一个简单的 LLM 服务示例) # 2. 定义请求体模型使用 Pydantic class PromptRequest(BaseModel): prompt: str max_tokens: int 100 # 3. 定义一个模拟的 LLM 生成函数后续替换为真实模型 def mock_llm_generate(prompt: str, max_tokens: int) - str: # 这里模拟一个简单的文本补全 return f你输入的是{prompt}。这是一个模拟的 LLM 回复最大生成长度为 {max_tokens}。 # 4. 定义根路径 app.get(/) async def root(): return {message: 欢迎使用 LLM FastAPI 服务请访问 /docs 查看接口文档。} # 5. 定义核心的 LLM 生成接口 app.post(/generate/) async def generate_text(request: PromptRequest): 接收一个 prompt返回模拟的 LLM 生成结果。 - **prompt**: 输入的提示文本 - **max_tokens**: 最大生成长度默认 100 # 调用模拟生成函数 result mock_llm_generate(request.prompt, request.max_tokens) return {prompt: request.prompt, generated_text: result}4.2 启动服务在终端中进入main.py所在目录运行uvicorn main:app --reload --host 0.0.0.0 --port 8000main:appmain是文件名不含.pyapp是代码中创建的FastAPI实例。--reload开发模式代码修改后自动重启服务器。--host 0.0.0.0允许所有网络接口访问便于局域网测试。--port 8000指定端口为 8000。4.3 访问与测试服务状态浏览器打开http://127.0.0.1:8000你会看到{message:欢迎使用 LLM FastAPI 服务请访问 /docs 查看接口文档。}。交互式文档访问http://127.0.0.1:8000/docs你会看到自动生成的 Swagger UI 界面。这是 FastAPI 最强大的功能之一。测试接口在/docs页面找到POST /generate/接口点击 “Try it out”。在Request body中修改 JSON{ prompt: 请用Python写一个快速排序函数, max_tokens: 200 }点击 “Execute”。你会看到服务器响应其中包含了我们的模拟回复。至此你已经成功创建了一个具有完整请求验证、自动文档的 LLM 服务雏形。接下来我们将用真实的 LLM 替换掉模拟函数。5. 集成真实 LLM以 OpenAI API 为例我们将把上面的模拟函数替换为调用真实的 OpenAI GPT 模型。这演示了如何将第三方 API 集成到 FastAPI 服务中。5.1 安装 OpenAI 库并设置密钥pip install openai你需要一个 OpenAI API 密钥。建议通过环境变量管理避免硬编码在代码中。# Linux/macOS export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here5.2 修改main.py集成 OpenAIfrom fastapi import FastAPI, HTTPException from pydantic import BaseModel import openai import os from typing import Optional app FastAPI(titleLLM FastAPI with OpenAI, description集成 OpenAI API 的 LLM 服务) # 从环境变量读取 API 密钥 openai.api_key os.getenv(OPENAI_API_KEY) if not openai.api_key: raise ValueError(请设置 OPENAI_API_KEY 环境变量) class OpenAIPromptRequest(BaseModel): prompt: str model: str gpt-3.5-turbo # 默认模型 max_tokens: Optional[int] 500 temperature: float 0.7 app.post(/generate/openai/) async def generate_with_openai(request: OpenAIPromptRequest): 调用 OpenAI API 生成文本。 try: # 构造请求消息 messages [{role: user, content: request.prompt}] # 调用 OpenAI ChatCompletion API response openai.ChatCompletion.create( modelrequest.model, messagesmessages, max_tokensrequest.max_tokens, temperaturerequest.temperature, streamFalse # 先使用非流式 ) # 提取回复内容 generated_text response.choices[0].message.content.strip() return { model: request.model, prompt: request.prompt, generated_text: generated_text, usage: response.usage } except openai.error.OpenAIError as e: # 处理 OpenAI API 错误 raise HTTPException(status_code500, detailfOpenAI API 错误: {str(e)}) except Exception as e: # 处理其他未知错误 raise HTTPException(status_code500, detailf服务器内部错误: {str(e)})5.3 测试真实接口重启uvicorn服务如果--reload已开启保存文件会自动重启。访问http://127.0.0.1:8000/docs。找到新的POST /generate/openai/接口尝试发送请求。你应该能收到来自 GPT 模型的真实回复。关键点解析错误处理我们使用try...except捕获了openai.error.OpenAIError和其他异常并通过 FastAPI 的HTTPException返回友好的错误信息这是生产环境必备的。配置化模型、最大 token 数、温度等参数都通过请求体传入使接口非常灵活。依赖管理API 密钥通过环境变量管理安全且便于在不同环境开发、测试、生产切换。6. 实现流式响应 (Streaming)LLM 生成文本时逐字输出流式能极大提升用户体验。FastAPI 通过返回一个StreamingResponse或使用生成器Generator可以轻松实现。6.1 修改接口支持流式输出from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse # 导入 StreamingResponse from pydantic import BaseModel import openai import os from typing import Optional import asyncio app FastAPI(titleLLM FastAPI with Streaming, description支持流式输出的 LLM 服务) openai.api_key os.getenv(OPENAI_API_KEY) class OpenAIPromptRequest(BaseModel): prompt: str model: str gpt-3.5-turbo max_tokens: Optional[int] 500 temperature: float 0.7 app.post(/generate/openai/stream/) async def generate_with_openai_stream(request: OpenAIPromptRequest): 流式调用 OpenAI API 生成文本。 返回一个 Server-Sent Events (SSE) 流。 async def event_generator(): try: messages [{role: user, content: request.prompt}] # 注意这里 streamTrue response_stream openai.ChatCompletion.create( modelrequest.model, messagesmessages, max_tokensrequest.max_tokens, temperaturerequest.temperature, streamTrue # 启用流式 ) for chunk in response_stream: # 检查是否有内容增量 if hasattr(chunk.choices[0].delta, content): content chunk.choices[0].delta.content if content: # 以 SSE 格式 yield 数据 yield fdata: {content}\n\n await asyncio.sleep(0) # 让出控制权避免阻塞 yield data: [DONE]\n\n # 流结束标记 except openai.error.OpenAIError as e: yield fdata: [ERROR] {str(e)}\n\n except Exception as e: yield fdata: [ERROR] 服务器内部错误\n\n # 返回 StreamingResponse指定媒体类型为 text/event-stream return StreamingResponse(event_generator(), media_typetext/event-stream)6.2 测试流式接口重启服务。由于 Swagger UI 对 SSE 流式支持有限我们可以用curl命令或写一个简单的 HTML 页面来测试。使用curl测试curl -N -X POST http://127.0.0.1:8000/generate/openai/stream/ \ -H Content-Type: application/json \ -d {prompt: 请介绍FastAPI框架, max_tokens: 200}你会看到文本逐字输出。前端集成在前端 JavaScript 中可以使用EventSourceAPI 来接收这个流。流式响应的优势低延迟用户无需等待整个响应生成完毕即可看到部分结果。更好的用户体验适用于聊天、长文本生成等场景。节省服务器内存无需在服务器端缓存完整响应再一次性发送。7. 进阶依赖注入与 LLM 客户端管理在真实项目中我们不应在每个请求处理函数中都初始化 LLM 客户端。FastAPI 的依赖注入系统可以优雅地解决这个问题实现客户端的共享和生命周期管理。7.1 创建依赖项创建一个新的文件dependencies.py# dependencies.py import openai import os from functools import lru_cache def get_openai_client(): 返回一个配置好的 OpenAI 客户端实例。 使用 lru_cache 确保在同一个进程中只创建一次客户端。 api_key os.getenv(OPENAI_API_KEY) if not api_key: raise RuntimeError(OPENAI_API_KEY 环境变量未设置) # 注意新版 OpenAI Python SDK 推荐使用 OpenAI 类 from openai import OpenAI client OpenAI(api_keyapi_key) return client # 或者如果你使用其他 LLM 服务例如本地模型 # def get_huggingface_pipeline(): # from transformers import pipeline # # 加载模型这里只是一个示例实际需要根据模型调整 # generator pipeline(text-generation, modelgpt2) # return generator7.2 在主应用中使用依赖项修改main.pyfrom fastapi import FastAPI, Depends, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel from typing import Optional import asyncio from openai import OpenAI # 使用新版客户端 from dependencies import get_openai_client # 导入依赖项 app FastAPI(titleLLM FastAPI with DI, description使用依赖注入管理 LLM 客户端的服务) class OpenAIPromptRequest(BaseModel): prompt: str model: str gpt-3.5-turbo max_tokens: Optional[int] 500 temperature: float 0.7 app.post(/generate/openai/di/) async def generate_with_di( request: OpenAIPromptRequest, openai_client: OpenAI Depends(get_openai_client) # 注入客户端 ): 使用依赖注入的 OpenAI 客户端生成文本。 try: messages [{role: user, content: request.prompt}] response openai_client.chat.completions.create( modelrequest.model, messagesmessages, max_tokensrequest.max_tokens, temperaturerequest.temperature, streamFalse ) generated_text response.choices[0].message.content return { model: request.model, prompt: request.prompt, generated_text: generated_text, usage: response.usage } except Exception as e: raise HTTPException(status_code500, detailf生成失败: {str(e)})依赖注入的好处代码复用客户端初始化逻辑在一处定义多处使用。易于测试在单元测试中可以轻松地用模拟mock客户端替换真实的依赖。生命周期管理结合lru_cache或数据库连接池可以高效管理昂贵资源如模型实例的创建和销毁。配置集中所有与外部服务OpenAI、数据库等的连接配置都在依赖项中管理。8. 项目结构优化与配置管理当项目增长时良好的结构至关重要。以下是一个推荐的适用于中小型 LLM 后端项目的目录结构llm_fastapi_project/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用创建和路由汇总 │ ├── config.py # 配置管理从环境变量或配置文件读取 │ ├── dependencies.py # 依赖项定义LLM客户端、数据库连接等 │ ├── models/ # Pydantic 模型定义 │ │ ├── __init__.py │ │ ├── request.py # 请求体模型 │ │ └── response.py # 响应体模型 │ ├── routers/ # 路由模块按功能拆分 │ │ ├── __init__.py │ │ ├── chat.py # 聊天相关接口 │ │ ├── completion.py # 补全相关接口 │ │ └── admin.py # 管理接口 │ ├── services/ # 业务逻辑层 │ │ ├── __init__.py │ │ ├── llm_service.py # 封装所有 LLM 调用逻辑 │ │ └── cache_service.py # 缓存服务 │ └── utils/ # 工具函数 │ ├── __init__.py │ └── logger.py # 日志配置 ├── tests/ # 测试目录 ├── requirements.txt # 项目依赖 ├── .env.example # 环境变量示例文件 └── README.md8.1 配置管理示例 (app/config.py)# app/config.py from pydantic_settings import BaseSettings # 需要安装 pydantic-settings class Settings(BaseSettings): # 从 .env 文件或环境变量中读取 openai_api_key: str openai_base_url: str https://api.openai.com/v1 # 可配置用于兼容其他兼容API model_default: str gpt-3.5-turbo max_tokens_default: int 1000 server_host: str 0.0.0.0 server_port: int 8000 log_level: str INFO class Config: env_file .env # 指定从 .env 文件加载 # 创建全局配置实例 settings Settings()8.2 使用配置和路由拆分在app/main.py中# app/main.py from fastapi import FastAPI from app.config import settings from app.routers import chat, completion, admin # 导入子路由 app FastAPI(titleLLM API Server, version1.0.0) # 包含子路由 app.include_router(chat.router, prefix/api/v1/chat, tags[chat]) app.include_router(completion.router, prefix/api/v1/completion, tags[completion]) app.include_router(admin.router, prefix/api/v1/admin, tags[admin]) app.get(/) async def root(): return {message: LLM API Server is running.}在app/routers/chat.py中# app/routers/chat.py from fastapi import APIRouter, Depends, HTTPException from app.models.request import ChatRequest from app.models.response import ChatResponse from app.services.llm_service import LLMService from app.dependencies import get_llm_service router APIRouter() router.post(/messages, response_modelChatResponse) async def create_chat_message( request: ChatRequest, llm_service: LLMService Depends(get_llm_service) ): 处理聊天消息。 try: response await llm_service.chat_completion(request.messages, request.model) return ChatResponse(**response) except Exception as e: raise HTTPException(status_code500, detailstr(e))这种结构使得代码职责清晰易于维护和扩展。9. 常见问题与排查方法在开发 FastAPI LLM 应用时你可能会遇到以下典型问题。这里提供排查思路。问题现象可能原因排查方式解决方案启动失败ImportError依赖未安装或虚拟环境未激活。1. 检查终端前缀是否有(venv)。2. 运行pip list查看fastapi和uvicorn是否存在。1. 激活虚拟环境。2. 运行pip install -r requirements.txt。访问127.0.0.1:8000无响应服务未启动或端口被占用。1. 检查终端uvicorn进程是否在运行。2. 运行netstat -ano | findstr :8000(Win) 或lsof -i:8000(Mac/Linux) 查看端口占用。1. 正确启动服务。2. 杀死占用端口的进程或更换端口如--port 8001。API 返回422 Unprocessable Entity请求体格式不符合 Pydantic 模型定义。1. 查看 FastAPI 自动文档/docs确认请求体格式。2. 检查前端发送的 JSON 字段名和类型。1. 严格按照 API 文档的格式发送请求。2. 在代码中为 Pydantic 字段设置合理的默认值或使用Optional。调用 OpenAI API 超时或报错网络问题、API 密钥错误、额度不足、请求频率超限。1. 检查OPENAI_API_KEY环境变量是否正确设置。2. 在 OpenAI 官网检查额度与账单。3. 查看 FastAPI 服务日志或 OpenAI 返回的错误信息。1. 确保密钥有效且有额度。2. 在代码中增加重试机制和更详细的错误日志。3. 考虑使用代理或调整超时时间。流式响应在前端不工作前端未正确使用 Server-Sent Events (SSE) 或跨域问题。1. 先用curl测试后端流式接口是否正常。2. 检查浏览器控制台是否有 CORS 错误。1. 确保后端返回StreamingResponse且media_typetext/event-stream。2. 在 FastAPI 中配置 CORS 中间件。高并发下服务响应慢或崩溃同步阻塞操作、数据库连接未池化、LLM 推理进程阻塞。1. 检查代码中是否有耗时的同步操作如文件读写、复杂计算在异步函数中直接调用。2. 使用async数据库驱动如asyncpg,aiomysql。3. 监控服务器 CPU/内存。1. 将同步阻塞操作放到线程池中执行asyncio.to_thread。2. 使用异步数据库库。3. 对于本地 LLM 推理考虑使用独立进程并通过消息队列通信。自动 API 文档 (/docs) 无法加载网络问题或 Swagger UI 资源加载失败。1. 检查浏览器控制台是否有 JS/CSS 加载错误。2. 尝试访问/redocReDoc 文档看是否正常。1. 通常是暂时的网络问题刷新或稍后再试。2. 可以配置 FastAPI 使用本地或 CDN 资源。10. 最佳实践与项目实战建议遵循以下建议可以让你的 FastAPI LLM 项目更加健壮和可维护。1. 环境变量与配置分离永远不要将 API 密钥、数据库密码等敏感信息硬编码在代码中。使用.env文件配合pydantic-settings或python-dotenv管理配置。将.env文件加入.gitignore并提交一个.env.example模板。2. 全面的日志记录在关键位置请求开始/结束、调用外部 API、发生错误添加日志。使用 Python 标准库logging进行结构化日志记录便于后续排查问题。import logging logger logging.getLogger(__name__) app.post(/generate/) async def generate(...): logger.info(f收到生成请求prompt: {request.prompt[:50]}...) # ... 业务逻辑 logger.info(生成请求处理完毕)3. 实现请求限流与鉴权对于公开的 LLM API必须实施限流Rate Limiting以防止滥用。可以使用slowapi或fastapi-limiter等中间件。为管理接口或付费 API 添加鉴权JWT、OAuth2等。FastAPI 内置了强大的安全工具。4. 使用异步数据库与缓存如果项目涉及用户、对话历史等数据存储务必选择支持异步的数据库驱动如asyncpgfor PostgreSQL,aiomysqlfor MySQL。对于频繁查询且变化不频繁的数据如模型配置、用户额度使用 Redis 等缓存可以极大提升性能。5. 编写单元测试与集成测试为你的路由、服务和工具函数编写测试。FastAPI 提供了TestClient使得测试 API 端点非常方便。测试应覆盖正常流程、边界情况和错误处理。from fastapi.testclient import TestClient from app.main import app client TestClient(app) def test_generate_endpoint(): response client.post(/generate/, json{prompt: Hello}) assert response.status_code 200 assert generated_text in response.json()6. 容器化部署使用 Docker 将你的应用及其所有依赖打包成镜像。这确保了环境一致性简化了部署。编写Dockerfile和docker-compose.yml文件。示例Dockerfile基础部分FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 80]7. 监控与告警在生产环境中集成监控工具如 Prometheus Grafana来跟踪 API 的请求量、延迟、错误率。设置关键指标如 5xx 错误激增、响应时间过长的告警。通过以上步骤你不仅学会了 FastAPI 的基础更掌握了如何构建一个结构清晰、易于维护、适合生产环境的 LLM 后端服务。从第一个简单的 API 到支持流式响应、依赖注入、配置化管理的项目结构这套方法论可以应用到绝大多数 LLM 应用开发中。接下来你可以基于这个骨架集成更复杂的逻辑如多轮对话管理、工具调用Agent、或接入本地大模型打造属于你自己的智能应用。 FastAPI构建LLM应用后端：从入门到实战部署指南
这次我们来看一个面向 LLM 开发者的 FastAPI 实战教程。如果你正在寻找一个能快速上手、性能出色并且能轻松构建 LLM 应用后端的 Python 框架FastAPI 几乎是当前最直接的选择。它不只是一个 Web 框架更是连接你的创意与大语言模型LLM能力的桥梁从简单的文本生成接口到复杂的 Agent 系统都能高效支撑。本文的核心是“能用”和“怎么用”。我们将跳过冗长的理论铺垫直接进入实战。你会看到如何用 FastAPI 快速搭建一个可运行的 LLM 服务如何处理 prompt 输入与流式输出如何设计 API 以适配不同的模型以及如何应对开发中常见的坑比如请求体验证、并发处理和错误处理。无论你是想为自研的 LLM 应用提供一个可靠的 HTTP 接口还是希望将 OpenAI、通义千问等模型的 API 进行二次封装和增强FastAPI 都能提供一套简洁而强大的工具。本文适合有一定 Python 基础希望快速将 LLM 想法落地为可访问服务的开发者。我们将从环境搭建、第一个 API 创建开始逐步深入到 LLM 集成、流式响应、Prompt 工程化以及项目结构优化最终形成一个可用于实际项目的基础框架。1. 核心能力速览在深入代码之前我们先快速了解 FastAPI 在 LLM 开发场景下的核心优势与关键特性这能帮助你判断它是否适合你的项目。能力项说明与 LLM 开发关联开发速度极快。基于 Python 类型提示Type Hints自动生成 API 文档Swagger UI/ReDoc减少大量手写文档与验证代码的时间让你专注于 LLM 业务逻辑。性能表现基于 Starlette异步和 Pydantic数据验证性能媲美 Node.js 和 Go。对于高并发的 LLM API 请求尤其是流式输出至关重要。异步支持原生支持async/await完美处理 LLM 模型推理通常是 I/O 密集型的等待时间提高服务器并发能力。依赖注入系统强大的依赖注入机制可以优雅地管理 LLM 模型实例、数据库连接、认证信息等使代码更清晰、更易测试。数据验证与序列化通过 Pydantic 自动进行请求/响应数据的验证与转换确保发送给 LLM 的 prompt 格式正确并安全地返回结构化结果。自动 API 文档启动服务后自动在/docs(Swagger) 和/redoc提供交互式文档。前端或测试人员可以直接在浏览器中调用你的 LLM 接口极大提升联调效率。学习门槛较低。如果你熟悉 Python特别是类型提示上手非常快。官方文档清晰社区活跃。适合的 LLM 场景1. 封装第三方 LLM API如 OpenAI、Claude。2. 部署本地开源模型通过 transformers 等库。3. 构建 LLM Agent 或工作流的后端服务。4. 开发基于 Prompt 的各类应用写作助手、代码生成、数据分析。2. 适用场景与使用边界FastAPI 是一个通用 Web 框架但在 LLM 开发领域其特性被放大了。明确它的适用场景和边界能帮助你做出更好的技术选型。它非常适合以下场景快速原型验证你有了一个 LLM 应用的想法比如一个智能客服接口需要最快速度搭建一个后端服务来演示和测试。FastAPI 的快速开发特性让你在几小时内就能看到可运行的 API。生产级 API 服务你需要为移动端、网页端或其他服务提供一个稳定、高性能的 LLM 接口。FastAPI 的异步特性、数据验证和自动文档非常适合构建易于维护和协作的生产接口。复杂 LLM 工作流你的应用涉及多步推理、工具调用Agent、或需要串联多个模型。FastAPI 的路由和依赖注入可以很好地组织这些复杂逻辑。统一 API 网关你可能同时使用多个 LLM 提供商OpenAI、Azure、本地模型。可以用 FastAPI 构建一个统一的网关处理认证、计费、日志、负载均衡和格式转换。需要注意的边界与考量并非机器学习框架FastAPI 本身不提供模型训练或推理功能。你需要集成像transformers、langchain、openai这样的库来实际调用 LLM。WebSocket 支持虽然 FastAPI 支持 WebSocket适用于实时对话场景但其核心优势仍在 HTTP/HTTPS API。对于超大规模、全双工的实时流可能需要结合更专业的网关。超大规模部署对于日调用量亿级以上的场景虽然 FastAPI 性能优秀但整体架构还需要考虑 API 网关、负载均衡、服务发现、容器化等云原生技术栈。前端渲染FastAPI 主要用于构建 API 后端。如果你需要复杂的用户界面通常需要搭配前端框架如 React, Vue或使用专门的模板引擎。3. 环境准备与前置条件开始编码前确保你的开发环境已经就绪。以下是 LLM 开发场景下推荐的基础环境配置。1. 操作系统推荐Linux (Ubuntu 20.04/22.04 LTS) 或 macOS。生产环境部署首选 Linux。也可用Windows 10/11建议使用 WSL2 以获得接近 Linux 的开发体验。2. Python 版本必须Python 3.8 或更高版本。FastAPI 充分利用了 Python 3.6 的类型提示特性3.8 以上版本能获得最佳兼容性。检查命令python --version # 或 python3 --version3. 包管理工具推荐使用pip并搭配虚拟环境venv或conda以隔离项目依赖。创建虚拟环境# 使用 venv python -m venv venv # 激活 (Linux/macOS) source venv/bin/activate # 激活 (Windows) venv\Scripts\activate4. 基础依赖核心依赖就是fastapi和异步服务器uvicorn。安装命令pip install fastapi uvicorn可选但推荐python-multipart用于处理表单数据如上文件httpx用于在异步代码中发出 HTTP 请求例如调用外部 LLM API。5. LLM 相关依赖按需安装调用 OpenAI 等云端 APIpip install openai使用 LangChain 框架pip install langchain langchain-openai部署本地 Hugging Face 模型pip install transformers torch accelerate注意本地模型部署对硬件GPU 显存有要求需根据模型大小准备相应资源。6. 代码编辑器/IDE任何你熟悉的即可如 VS Code推荐对 Python 和 FastAPI 支持好、PyCharm 等。4. 第一个 FastAPI 应用与 LLM “Hello World”让我们从一个最简单的例子开始感受 FastAPI 的便捷并立即将其与 LLM 联系起来。4.1 创建项目文件创建一个名为main.py的文件输入以下代码from fastapi import FastAPI from pydantic import BaseModel # 1. 创建 FastAPI 应用实例 app FastAPI(titleLLM FastAPI Demo, description一个简单的 LLM 服务示例) # 2. 定义请求体模型使用 Pydantic class PromptRequest(BaseModel): prompt: str max_tokens: int 100 # 3. 定义一个模拟的 LLM 生成函数后续替换为真实模型 def mock_llm_generate(prompt: str, max_tokens: int) - str: # 这里模拟一个简单的文本补全 return f你输入的是{prompt}。这是一个模拟的 LLM 回复最大生成长度为 {max_tokens}。 # 4. 定义根路径 app.get(/) async def root(): return {message: 欢迎使用 LLM FastAPI 服务请访问 /docs 查看接口文档。} # 5. 定义核心的 LLM 生成接口 app.post(/generate/) async def generate_text(request: PromptRequest): 接收一个 prompt返回模拟的 LLM 生成结果。 - **prompt**: 输入的提示文本 - **max_tokens**: 最大生成长度默认 100 # 调用模拟生成函数 result mock_llm_generate(request.prompt, request.max_tokens) return {prompt: request.prompt, generated_text: result}4.2 启动服务在终端中进入main.py所在目录运行uvicorn main:app --reload --host 0.0.0.0 --port 8000main:appmain是文件名不含.pyapp是代码中创建的FastAPI实例。--reload开发模式代码修改后自动重启服务器。--host 0.0.0.0允许所有网络接口访问便于局域网测试。--port 8000指定端口为 8000。4.3 访问与测试服务状态浏览器打开http://127.0.0.1:8000你会看到{message:欢迎使用 LLM FastAPI 服务请访问 /docs 查看接口文档。}。交互式文档访问http://127.0.0.1:8000/docs你会看到自动生成的 Swagger UI 界面。这是 FastAPI 最强大的功能之一。测试接口在/docs页面找到POST /generate/接口点击 “Try it out”。在Request body中修改 JSON{ prompt: 请用Python写一个快速排序函数, max_tokens: 200 }点击 “Execute”。你会看到服务器响应其中包含了我们的模拟回复。至此你已经成功创建了一个具有完整请求验证、自动文档的 LLM 服务雏形。接下来我们将用真实的 LLM 替换掉模拟函数。5. 集成真实 LLM以 OpenAI API 为例我们将把上面的模拟函数替换为调用真实的 OpenAI GPT 模型。这演示了如何将第三方 API 集成到 FastAPI 服务中。5.1 安装 OpenAI 库并设置密钥pip install openai你需要一个 OpenAI API 密钥。建议通过环境变量管理避免硬编码在代码中。# Linux/macOS export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here5.2 修改main.py集成 OpenAIfrom fastapi import FastAPI, HTTPException from pydantic import BaseModel import openai import os from typing import Optional app FastAPI(titleLLM FastAPI with OpenAI, description集成 OpenAI API 的 LLM 服务) # 从环境变量读取 API 密钥 openai.api_key os.getenv(OPENAI_API_KEY) if not openai.api_key: raise ValueError(请设置 OPENAI_API_KEY 环境变量) class OpenAIPromptRequest(BaseModel): prompt: str model: str gpt-3.5-turbo # 默认模型 max_tokens: Optional[int] 500 temperature: float 0.7 app.post(/generate/openai/) async def generate_with_openai(request: OpenAIPromptRequest): 调用 OpenAI API 生成文本。 try: # 构造请求消息 messages [{role: user, content: request.prompt}] # 调用 OpenAI ChatCompletion API response openai.ChatCompletion.create( modelrequest.model, messagesmessages, max_tokensrequest.max_tokens, temperaturerequest.temperature, streamFalse # 先使用非流式 ) # 提取回复内容 generated_text response.choices[0].message.content.strip() return { model: request.model, prompt: request.prompt, generated_text: generated_text, usage: response.usage } except openai.error.OpenAIError as e: # 处理 OpenAI API 错误 raise HTTPException(status_code500, detailfOpenAI API 错误: {str(e)}) except Exception as e: # 处理其他未知错误 raise HTTPException(status_code500, detailf服务器内部错误: {str(e)})5.3 测试真实接口重启uvicorn服务如果--reload已开启保存文件会自动重启。访问http://127.0.0.1:8000/docs。找到新的POST /generate/openai/接口尝试发送请求。你应该能收到来自 GPT 模型的真实回复。关键点解析错误处理我们使用try...except捕获了openai.error.OpenAIError和其他异常并通过 FastAPI 的HTTPException返回友好的错误信息这是生产环境必备的。配置化模型、最大 token 数、温度等参数都通过请求体传入使接口非常灵活。依赖管理API 密钥通过环境变量管理安全且便于在不同环境开发、测试、生产切换。6. 实现流式响应 (Streaming)LLM 生成文本时逐字输出流式能极大提升用户体验。FastAPI 通过返回一个StreamingResponse或使用生成器Generator可以轻松实现。6.1 修改接口支持流式输出from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse # 导入 StreamingResponse from pydantic import BaseModel import openai import os from typing import Optional import asyncio app FastAPI(titleLLM FastAPI with Streaming, description支持流式输出的 LLM 服务) openai.api_key os.getenv(OPENAI_API_KEY) class OpenAIPromptRequest(BaseModel): prompt: str model: str gpt-3.5-turbo max_tokens: Optional[int] 500 temperature: float 0.7 app.post(/generate/openai/stream/) async def generate_with_openai_stream(request: OpenAIPromptRequest): 流式调用 OpenAI API 生成文本。 返回一个 Server-Sent Events (SSE) 流。 async def event_generator(): try: messages [{role: user, content: request.prompt}] # 注意这里 streamTrue response_stream openai.ChatCompletion.create( modelrequest.model, messagesmessages, max_tokensrequest.max_tokens, temperaturerequest.temperature, streamTrue # 启用流式 ) for chunk in response_stream: # 检查是否有内容增量 if hasattr(chunk.choices[0].delta, content): content chunk.choices[0].delta.content if content: # 以 SSE 格式 yield 数据 yield fdata: {content}\n\n await asyncio.sleep(0) # 让出控制权避免阻塞 yield data: [DONE]\n\n # 流结束标记 except openai.error.OpenAIError as e: yield fdata: [ERROR] {str(e)}\n\n except Exception as e: yield fdata: [ERROR] 服务器内部错误\n\n # 返回 StreamingResponse指定媒体类型为 text/event-stream return StreamingResponse(event_generator(), media_typetext/event-stream)6.2 测试流式接口重启服务。由于 Swagger UI 对 SSE 流式支持有限我们可以用curl命令或写一个简单的 HTML 页面来测试。使用curl测试curl -N -X POST http://127.0.0.1:8000/generate/openai/stream/ \ -H Content-Type: application/json \ -d {prompt: 请介绍FastAPI框架, max_tokens: 200}你会看到文本逐字输出。前端集成在前端 JavaScript 中可以使用EventSourceAPI 来接收这个流。流式响应的优势低延迟用户无需等待整个响应生成完毕即可看到部分结果。更好的用户体验适用于聊天、长文本生成等场景。节省服务器内存无需在服务器端缓存完整响应再一次性发送。7. 进阶依赖注入与 LLM 客户端管理在真实项目中我们不应在每个请求处理函数中都初始化 LLM 客户端。FastAPI 的依赖注入系统可以优雅地解决这个问题实现客户端的共享和生命周期管理。7.1 创建依赖项创建一个新的文件dependencies.py# dependencies.py import openai import os from functools import lru_cache def get_openai_client(): 返回一个配置好的 OpenAI 客户端实例。 使用 lru_cache 确保在同一个进程中只创建一次客户端。 api_key os.getenv(OPENAI_API_KEY) if not api_key: raise RuntimeError(OPENAI_API_KEY 环境变量未设置) # 注意新版 OpenAI Python SDK 推荐使用 OpenAI 类 from openai import OpenAI client OpenAI(api_keyapi_key) return client # 或者如果你使用其他 LLM 服务例如本地模型 # def get_huggingface_pipeline(): # from transformers import pipeline # # 加载模型这里只是一个示例实际需要根据模型调整 # generator pipeline(text-generation, modelgpt2) # return generator7.2 在主应用中使用依赖项修改main.pyfrom fastapi import FastAPI, Depends, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel from typing import Optional import asyncio from openai import OpenAI # 使用新版客户端 from dependencies import get_openai_client # 导入依赖项 app FastAPI(titleLLM FastAPI with DI, description使用依赖注入管理 LLM 客户端的服务) class OpenAIPromptRequest(BaseModel): prompt: str model: str gpt-3.5-turbo max_tokens: Optional[int] 500 temperature: float 0.7 app.post(/generate/openai/di/) async def generate_with_di( request: OpenAIPromptRequest, openai_client: OpenAI Depends(get_openai_client) # 注入客户端 ): 使用依赖注入的 OpenAI 客户端生成文本。 try: messages [{role: user, content: request.prompt}] response openai_client.chat.completions.create( modelrequest.model, messagesmessages, max_tokensrequest.max_tokens, temperaturerequest.temperature, streamFalse ) generated_text response.choices[0].message.content return { model: request.model, prompt: request.prompt, generated_text: generated_text, usage: response.usage } except Exception as e: raise HTTPException(status_code500, detailf生成失败: {str(e)})依赖注入的好处代码复用客户端初始化逻辑在一处定义多处使用。易于测试在单元测试中可以轻松地用模拟mock客户端替换真实的依赖。生命周期管理结合lru_cache或数据库连接池可以高效管理昂贵资源如模型实例的创建和销毁。配置集中所有与外部服务OpenAI、数据库等的连接配置都在依赖项中管理。8. 项目结构优化与配置管理当项目增长时良好的结构至关重要。以下是一个推荐的适用于中小型 LLM 后端项目的目录结构llm_fastapi_project/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用创建和路由汇总 │ ├── config.py # 配置管理从环境变量或配置文件读取 │ ├── dependencies.py # 依赖项定义LLM客户端、数据库连接等 │ ├── models/ # Pydantic 模型定义 │ │ ├── __init__.py │ │ ├── request.py # 请求体模型 │ │ └── response.py # 响应体模型 │ ├── routers/ # 路由模块按功能拆分 │ │ ├── __init__.py │ │ ├── chat.py # 聊天相关接口 │ │ ├── completion.py # 补全相关接口 │ │ └── admin.py # 管理接口 │ ├── services/ # 业务逻辑层 │ │ ├── __init__.py │ │ ├── llm_service.py # 封装所有 LLM 调用逻辑 │ │ └── cache_service.py # 缓存服务 │ └── utils/ # 工具函数 │ ├── __init__.py │ └── logger.py # 日志配置 ├── tests/ # 测试目录 ├── requirements.txt # 项目依赖 ├── .env.example # 环境变量示例文件 └── README.md8.1 配置管理示例 (app/config.py)# app/config.py from pydantic_settings import BaseSettings # 需要安装 pydantic-settings class Settings(BaseSettings): # 从 .env 文件或环境变量中读取 openai_api_key: str openai_base_url: str https://api.openai.com/v1 # 可配置用于兼容其他兼容API model_default: str gpt-3.5-turbo max_tokens_default: int 1000 server_host: str 0.0.0.0 server_port: int 8000 log_level: str INFO class Config: env_file .env # 指定从 .env 文件加载 # 创建全局配置实例 settings Settings()8.2 使用配置和路由拆分在app/main.py中# app/main.py from fastapi import FastAPI from app.config import settings from app.routers import chat, completion, admin # 导入子路由 app FastAPI(titleLLM API Server, version1.0.0) # 包含子路由 app.include_router(chat.router, prefix/api/v1/chat, tags[chat]) app.include_router(completion.router, prefix/api/v1/completion, tags[completion]) app.include_router(admin.router, prefix/api/v1/admin, tags[admin]) app.get(/) async def root(): return {message: LLM API Server is running.}在app/routers/chat.py中# app/routers/chat.py from fastapi import APIRouter, Depends, HTTPException from app.models.request import ChatRequest from app.models.response import ChatResponse from app.services.llm_service import LLMService from app.dependencies import get_llm_service router APIRouter() router.post(/messages, response_modelChatResponse) async def create_chat_message( request: ChatRequest, llm_service: LLMService Depends(get_llm_service) ): 处理聊天消息。 try: response await llm_service.chat_completion(request.messages, request.model) return ChatResponse(**response) except Exception as e: raise HTTPException(status_code500, detailstr(e))这种结构使得代码职责清晰易于维护和扩展。9. 常见问题与排查方法在开发 FastAPI LLM 应用时你可能会遇到以下典型问题。这里提供排查思路。问题现象可能原因排查方式解决方案启动失败ImportError依赖未安装或虚拟环境未激活。1. 检查终端前缀是否有(venv)。2. 运行pip list查看fastapi和uvicorn是否存在。1. 激活虚拟环境。2. 运行pip install -r requirements.txt。访问127.0.0.1:8000无响应服务未启动或端口被占用。1. 检查终端uvicorn进程是否在运行。2. 运行netstat -ano | findstr :8000(Win) 或lsof -i:8000(Mac/Linux) 查看端口占用。1. 正确启动服务。2. 杀死占用端口的进程或更换端口如--port 8001。API 返回422 Unprocessable Entity请求体格式不符合 Pydantic 模型定义。1. 查看 FastAPI 自动文档/docs确认请求体格式。2. 检查前端发送的 JSON 字段名和类型。1. 严格按照 API 文档的格式发送请求。2. 在代码中为 Pydantic 字段设置合理的默认值或使用Optional。调用 OpenAI API 超时或报错网络问题、API 密钥错误、额度不足、请求频率超限。1. 检查OPENAI_API_KEY环境变量是否正确设置。2. 在 OpenAI 官网检查额度与账单。3. 查看 FastAPI 服务日志或 OpenAI 返回的错误信息。1. 确保密钥有效且有额度。2. 在代码中增加重试机制和更详细的错误日志。3. 考虑使用代理或调整超时时间。流式响应在前端不工作前端未正确使用 Server-Sent Events (SSE) 或跨域问题。1. 先用curl测试后端流式接口是否正常。2. 检查浏览器控制台是否有 CORS 错误。1. 确保后端返回StreamingResponse且media_typetext/event-stream。2. 在 FastAPI 中配置 CORS 中间件。高并发下服务响应慢或崩溃同步阻塞操作、数据库连接未池化、LLM 推理进程阻塞。1. 检查代码中是否有耗时的同步操作如文件读写、复杂计算在异步函数中直接调用。2. 使用async数据库驱动如asyncpg,aiomysql。3. 监控服务器 CPU/内存。1. 将同步阻塞操作放到线程池中执行asyncio.to_thread。2. 使用异步数据库库。3. 对于本地 LLM 推理考虑使用独立进程并通过消息队列通信。自动 API 文档 (/docs) 无法加载网络问题或 Swagger UI 资源加载失败。1. 检查浏览器控制台是否有 JS/CSS 加载错误。2. 尝试访问/redocReDoc 文档看是否正常。1. 通常是暂时的网络问题刷新或稍后再试。2. 可以配置 FastAPI 使用本地或 CDN 资源。10. 最佳实践与项目实战建议遵循以下建议可以让你的 FastAPI LLM 项目更加健壮和可维护。1. 环境变量与配置分离永远不要将 API 密钥、数据库密码等敏感信息硬编码在代码中。使用.env文件配合pydantic-settings或python-dotenv管理配置。将.env文件加入.gitignore并提交一个.env.example模板。2. 全面的日志记录在关键位置请求开始/结束、调用外部 API、发生错误添加日志。使用 Python 标准库logging进行结构化日志记录便于后续排查问题。import logging logger logging.getLogger(__name__) app.post(/generate/) async def generate(...): logger.info(f收到生成请求prompt: {request.prompt[:50]}...) # ... 业务逻辑 logger.info(生成请求处理完毕)3. 实现请求限流与鉴权对于公开的 LLM API必须实施限流Rate Limiting以防止滥用。可以使用slowapi或fastapi-limiter等中间件。为管理接口或付费 API 添加鉴权JWT、OAuth2等。FastAPI 内置了强大的安全工具。4. 使用异步数据库与缓存如果项目涉及用户、对话历史等数据存储务必选择支持异步的数据库驱动如asyncpgfor PostgreSQL,aiomysqlfor MySQL。对于频繁查询且变化不频繁的数据如模型配置、用户额度使用 Redis 等缓存可以极大提升性能。5. 编写单元测试与集成测试为你的路由、服务和工具函数编写测试。FastAPI 提供了TestClient使得测试 API 端点非常方便。测试应覆盖正常流程、边界情况和错误处理。from fastapi.testclient import TestClient from app.main import app client TestClient(app) def test_generate_endpoint(): response client.post(/generate/, json{prompt: Hello}) assert response.status_code 200 assert generated_text in response.json()6. 容器化部署使用 Docker 将你的应用及其所有依赖打包成镜像。这确保了环境一致性简化了部署。编写Dockerfile和docker-compose.yml文件。示例Dockerfile基础部分FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 80]7. 监控与告警在生产环境中集成监控工具如 Prometheus Grafana来跟踪 API 的请求量、延迟、错误率。设置关键指标如 5xx 错误激增、响应时间过长的告警。通过以上步骤你不仅学会了 FastAPI 的基础更掌握了如何构建一个结构清晰、易于维护、适合生产环境的 LLM 后端服务。从第一个简单的 API 到支持流式响应、依赖注入、配置化管理的项目结构这套方法论可以应用到绝大多数 LLM 应用开发中。接下来你可以基于这个骨架集成更复杂的逻辑如多轮对话管理、工具调用Agent、或接入本地大模型打造属于你自己的智能应用。