在实际项目中快速构建一个高性能、易于维护的Web API是后端开发的核心需求。FastAPI凭借其现代的设计理念、极致的性能基于Starlette和Pydantic以及自动生成的交互式API文档已成为Python领域构建API的首选框架之一。与此同时大型语言模型LLM的应用开发正从简单的对话接口转向需要复杂业务逻辑、状态管理和精准流程控制的Agent或工具调用场景。将FastAPI与LLM开发结合能够为模型提供一个稳定、可扩展的后端服务层处理认证、会话管理、异步任务、数据持久化等工程问题让开发者更专注于Prompt设计、模型调用和业务逻辑本身。本文面向有一定Python基础希望快速掌握FastAPI并将其应用于实际LLM项目开发的开发者。我们将从零开始完成一个完整的LLM问答服务后端项目。这个项目不仅会涵盖FastAPI的核心用法还会深入如何集成LLM以OpenAI API为例、设计可维护的Prompt模板、处理流式响应、管理对话上下文并最终部署一个可用的服务。学完后你将能够独立搭建具备生产级潜力的LLM应用后端。1. 理解FastAPI的核心优势与LLM服务架构在开始编码之前需要明确为什么选择FastAPI来构建LLM服务以及一个典型的LLM服务后端包含哪些组件。1.1 FastAPI为何适合LLM项目LLM应用后端通常需要处理高并发、低延迟的请求特别是流式输出场景。FastAPI的异步支持async/await天生适合这类I/O密集型任务。其基于Pydantic的数据验证能确保发送给LLM的Prompt参数结构正确、类型安全自动生成的OpenAPI文档则方便前端或测试人员理解接口。此外FastAPI的依赖注入系统能优雅地管理LLM客户端、数据库连接等全局资源。一个常见的误区是认为LLM开发只需要调用模型API。实际上工程化部分同样重要包括请求验证与预处理确保用户输入符合预期并进行必要的清洗或转换。上下文管理维护多轮对话的历史记录并处理可能超长的上下文。异步与流式处理避免阻塞主线程并实现Token-by-Token的流式返回以提升用户体验。错误处理与重试优雅地处理模型服务不可用、超时、内容过滤等异常。可观测性记录日志、监控指标便于问题排查。FastAPI为上述所有需求提供了简洁而强大的基础框架。1.2 LLM服务后端的基本组件我们将构建的服务包含以下核心组件应用入口与路由FastAPI应用实例定义API端点。数据模型使用Pydantic定义请求体和响应体的结构。LLM客户端与服务层封装对OpenAI等模型API的调用包含Prompt构建逻辑。依赖注入管理LLM客户端、配置等的生命周期和共享。中间件用于处理CORS、请求日志等横切关注点。配置管理从环境变量或配置文件加载API密钥等敏感信息。项目的技术栈如下Python: 3.8Web框架: FastAPIASGI服务器: Uvicorn (用于运行FastAPI)HTTP客户端:httpx或openai官方库环境管理:python-dotenv数据验证: Pydantic (已包含在FastAPI中)2. 环境准备与项目初始化让我们从创建一个干净的项目环境开始。2.1 创建虚拟环境与安装依赖首先为项目创建一个独立的虚拟环境这是管理项目依赖的最佳实践。# 创建项目目录并进入 mkdir fastapi-llm-project cd fastapi-llm-project # 创建虚拟环境以venv为例 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate激活虚拟环境后命令行提示符前通常会出现(venv)标识。接下来安装核心依赖。# 安装FastAPI及其ASGI服务器Uvicorn pip install fastapi uvicorn # 安装HTTP客户端和配置管理库 pip install httpx python-dotenv # 安装OpenAI官方库用于调用GPT模型 pip install openai # 可选安装用于开发自动重载和调试的库 pip install watchfiles2.2 项目结构设计一个清晰的项目结构有助于代码维护。我们采用以下结构fastapi-llm-project/ ├── .env # 环境变量文件不提交到Git ├── .gitignore ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用创建和主路由 │ ├── config.py # 配置加载 │ ├── dependencies.py # 依赖注入项 │ ├── models.py # Pydantic数据模型 │ ├── services/ # 业务逻辑层 │ │ ├── __init__.py │ │ └── llm_service.py # LLM服务封装 │ └── routers/ # 路由模块 │ ├── __init__.py │ └── chat.py # 聊天相关路由 ├── requirements.txt # 项目依赖清单 └── README.md现在创建这些文件和目录。# 创建目录 mkdir -p app/services app/routers # 创建文件 touch .env .gitignore app/__init__.py app/main.py app/config.py app/dependencies.py app/models.py touch app/services/__init__.py app/services/llm_service.py touch app/routers/__init__.py app/routers/chat.py touch requirements.txt README.md2.3 管理环境变量与配置LLM项目通常需要API密钥等敏感信息。我们使用.env文件来管理它们并通过python-dotenv加载。首先在.env文件中添加你的OpenAI API密钥或其他LLM服务商密钥。# .env OPENAI_API_KEYsk-your-actual-openai-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果使用其他兼容API可修改此处 MODEL_NAMEgpt-3.5-turbo # 默认使用的模型注意务必在.gitignore中添加.env避免将密钥提交到版本控制系统。# .gitignore venv/ __pycache__/ *.pyc .env接下来创建app/config.py来集中管理配置。# app/config.py from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): 应用配置从环境变量加载 openai_api_key: str openai_base_url: str https://api.openai.com/v1 model_name: str gpt-3.5-turbo # 可以添加其他配置如数据库URL、日志级别等 # database_url: Optional[str] None # log_level: str INFO class Config: env_file .env # 指定从.env文件加载 env_file_encoding utf-8 # 创建全局配置实例 settings Settings()这里使用了Pydantic的BaseSettings它能自动从环境变量和.env文件加载配置并进行验证。如果OPENAI_API_KEY没有设置启动应用时会直接报错便于及早发现问题。3. 构建核心LLM服务与数据模型在定义API路由之前我们需要先封装LLM调用逻辑并设计好API的数据契约。3.1 定义Pydantic数据模型Pydantic模型定义了API接口的“形状”确保输入输出的数据类型和结构正确。在app/models.py中定义。# app/models.py from pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime class Message(BaseModel): 单条消息模型 role: str Field(..., description消息角色如 user, assistant, system) content: str Field(..., description消息内容) class ChatRequest(BaseModel): 聊天请求体 messages: List[Message] Field(..., description对话历史消息列表) model: Optional[str] Field(None, description指定使用的模型不填则使用配置默认值) stream: bool Field(False, description是否启用流式输出) temperature: float Field(0.7, ge0.0, le2.0, description采样温度控制随机性) max_tokens: Optional[int] Field(None, gt0, description回复的最大token数) class ChatResponse(BaseModel): 聊天响应体非流式 id: str object: str chat.completion created: int model: str choices: List[dict] usage: dict class StreamResponseChunk(BaseModel): 流式响应块 id: str object: str chat.completion.chunk created: int model: str choices: List[dict] class HealthResponse(BaseModel): 健康检查响应 status: str timestamp: datetimeField类用于提供字段的元数据如描述、默认值和验证规则ge大于等于le小于等于。这不仅能生成清晰的API文档还能在请求到达业务逻辑前就拦截非法数据。3.2 封装LLM服务我们将对OpenAI API的调用封装在一个服务类中以提高代码的可测试性和可维护性。创建app/services/llm_service.py。# app/services/llm_service.py import httpx import json from typing import AsyncGenerator, Optional from app.config import settings from app.models import ChatRequest import logging logger logging.getLogger(__name__) class LLMService: LLM服务封装类 def __init__(self): self.api_key settings.openai_api_key self.base_url settings.openai_base_url self.default_model settings.model_name self.headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } async def create_chat_completion( self, request: ChatRequest, timeout: float 30.0 ) - dict: 创建非流式聊天补全 model request.model or self.default_model payload { model: model, messages: [msg.dict() for msg in request.messages], temperature: request.temperature, max_tokens: request.max_tokens, stream: False } # 移除为None的字段 payload {k: v for k, v in payload.items() if v is not None} async with httpx.AsyncClient(timeouttimeout) as client: try: response await client.post( f{self.base_url}/chat/completions, headersself.headers, jsonpayload ) response.raise_for_status() # 如果状态码不是2xx抛出HTTPError return response.json() except httpx.HTTPStatusError as e: logger.error(fLLM API HTTP错误: {e.response.status_code} - {e.response.text}) raise except httpx.RequestError as e: logger.error(fLLM API 请求错误: {e}) raise async def create_chat_completion_stream( self, request: ChatRequest, timeout: float 60.0 ) - AsyncGenerator[str, None]: 创建流式聊天补全返回一个异步生成器 model request.model or self.default_model payload { model: model, messages: [msg.dict() for msg in request.messages], temperature: request.temperature, max_tokens: request.max_tokens, stream: True } payload {k: v for k, v in payload.items() if v is not None} async with httpx.AsyncClient(timeouttimeout) as client: try: async with client.stream( POST, f{self.base_url}/chat/completions, headersself.headers, jsonpayload ) as response: response.raise_for_status() async for chunk in response.aiter_lines(): if chunk: # OpenAI流式响应格式 data: {...}\n\n if chunk.startswith(data: ): data chunk[6:] # 去掉data: 前缀 if data.strip() [DONE]: break try: yield data except json.JSONDecodeError: logger.warning(f无法解析流式响应块: {data}) except httpx.HTTPStatusError as e: logger.error(fLLM流式API HTTP错误: {e.response.status_code}) raise except httpx.RequestError as e: logger.error(fLLM流式API 请求错误: {e}) raise关键点解释依赖注入配置服务从全局settings获取API密钥和URL便于统一管理。异常处理使用httpx的异常机制并记录日志便于排查网络或API错误。流式处理create_chat_completion_stream方法是一个异步生成器使用client.stream来逐步接收响应并通过yield逐个返回JSON字符串块。这是实现SSEServer-Sent Events的基础。超时控制为流式和非流式请求设置了不同的超时时间流式请求通常需要更长时间。3.3 设置依赖注入为了在路由中方便地使用LLMService我们通过FastAPI的依赖注入系统来管理它的实例。创建app/dependencies.py。# app/dependencies.py from app.services.llm_service import LLMService def get_llm_service() - LLMService: 获取LLM服务实例的依赖函数 # 这里可以扩展为更复杂的生命周期管理如连接池 return LLMService()4. 实现API路由与流式响应现在我们将创建具体的API端点包括健康检查、普通聊天和流式聊天。4.1 创建聊天路由在app/routers/chat.py中定义聊天相关的端点。# app/routers/chat.py from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import StreamingResponse import json import time from typing import Optional from app.dependencies import get_llm_service from app.models import ChatRequest, ChatResponse, StreamResponseChunk, HealthResponse from app.services.llm_service import LLMService router APIRouter(prefix/api/v1, tags[chat]) router.get(/health, response_modelHealthResponse) async def health_check(): 健康检查端点 return HealthResponse(statushealthy, timestampdatetime.now()) router.post(/chat/completions, response_modelChatResponse) async def create_chat_completion( request: ChatRequest, llm_service: LLMService Depends(get_llm_service) ): 非流式聊天补全端点。 请求体需符合ChatRequest模型返回完整的响应。 try: result await llm_service.create_chat_completion(request) return result except Exception as e: # 记录详细日志但返回给客户端的错误信息要简化 raise HTTPException(status_code500, detailf服务内部错误: {str(e)}) router.post(/chat/completions/stream) async def create_chat_completion_stream( request: ChatRequest, llm_service: LLMService Depends(get_llm_service) ): 流式聊天补全端点。 返回Server-Sent Events (SSE)流。 async def event_generator(): try: async for chunk in llm_service.create_chat_completion_stream(request): # 按照SSE格式返回数据 yield fdata: {chunk}\n\n except Exception as e: # 流式响应中发生错误可以发送一个错误事件如果客户端支持 error_chunk json.dumps({ error: { message: f流式响应生成失败: {str(e)}, type: internal_error } }) yield fdata: {error_chunk}\n\n finally: # 可选发送结束标记 yield data: [DONE]\n\n return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no # 禁用Nginx缓冲 } )关键点解释路由组织使用APIRouter并设置prefix和tags有助于在大型项目中模块化路由并使API文档更清晰。依赖注入使用在端点函数参数中声明llm_service: LLMService Depends(get_llm_service)FastAPI会自动调用get_llm_service函数并将返回的LLMService实例注入进来。流式响应/chat/completions/stream端点返回一个StreamingResponse。其核心是一个异步生成器函数event_generator它从llm_service获取数据块并格式化为SSE标准格式data: ...\n\n后yield出去。客户端如浏览器EventSource或Fetch API可以逐步接收这些数据。错误处理在流式响应中处理异常比较棘手因为响应头已经发送。这里我们尝试将错误信息也包装成SSE事件发送给客户端。生产环境中可能需要更复杂的错误恢复和重试逻辑。4.2 创建主应用并集成路由最后在app/main.py中创建FastAPI应用实例并集成我们定义的路由、中间件等。# app/main.py from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware import logging from app.routers import chat # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 创建FastAPI应用实例 app FastAPI( titleLLM Chat API, description一个基于FastAPI构建的LLM聊天API服务, version1.0.0 ) # 添加CORS中间件允许前端跨域访问 app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境应替换为具体的前端域名 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 包含路由 app.include_router(chat.router) app.on_event(startup) async def startup_event(): 应用启动时执行 logger.info(LLM Chat API 服务启动...) app.on_event(shutdown) async def shutdown_event(): 应用关闭时执行 logger.info(LLM Chat API 服务关闭...)5. 运行、测试与API文档验证服务代码已经完成现在让我们启动它并进行测试。5.1 启动开发服务器在项目根目录下运行以下命令启动Uvicorn服务器。uvicorn app.main:app --reload --host 0.0.0.0 --port 8000参数说明app.main:app指定FastAPI应用实例的位置app模块下的main.py文件中的app对象。--reload启用热重载代码修改后服务器会自动重启。仅用于开发。--host 0.0.0.0监听所有网络接口方便从其他设备访问。--port 8000指定端口号。看到类似以下输出说明服务已启动INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRLC to quit) INFO: Started reloader process [12345] using WatchFiles INFO: Started server process [12346] INFO: Waiting for application startup. INFO: Application startup complete.5.2 使用自动生成的交互式API文档FastAPI的一大亮点是自动生成OpenAPI文档。启动服务后访问以下URL交互式API文档 (Swagger UI): http://localhost:8000/docs替代API文档 (ReDoc): http://localhost:8000/redoc在Swagger UI界面你可以看到我们定义的所有端点/health,/api/v1/chat/completions,/api/v1/chat/completions/stream。你可以直接在这个界面上点击“Try it out”来测试API无需额外工具。5.3 使用curl或Postman进行测试测试健康检查端点curl -X GET http://localhost:8000/api/v1/health预期返回{status:healthy,timestamp:2023-10-27T08:00:00.123456}测试非流式聊天端点curl -X POST http://localhost:8000/api/v1/chat/completions \ -H Content-Type: application/json \ -d { messages: [ {role: user, content: 你好请介绍一下FastAPI。} ], stream: false }预期返回一个完整的JSON响应包含模型生成的回复。测试流式聊天端点curl -X POST http://localhost:8000/api/v1/chat/completions/stream \ -H Content-Type: application/json \ -H Accept: text/event-stream \ -d { messages: [ {role: user, content: 用Python写一个简单的Hello World程序。} ], stream: true }你会看到一系列以data:开头的行每行都是一个JSON块最后一行是data: [DONE]。这就是SSE流式响应。5.4 编写一个简单的Python测试客户端为了更好地理解流式响应可以编写一个简单的客户端脚本。# test_stream_client.py import asyncio import aiohttp import json async def test_stream(): url http://localhost:8000/api/v1/chat/completions/stream payload { messages: [{role: user, content: 讲一个笑话}], stream: True } async with aiohttp.ClientSession() as session: async with session.post(url, jsonpayload) as resp: async for line in resp.content: line line.decode(utf-8).strip() if line.startswith(data: ): data line[6:] # Remove data: if data [DONE]: print(\n[Stream finished]) break try: chunk json.loads(data) # 提取并打印delta content if chunk.get(choices): delta chunk[choices][0].get(delta, {}) if content in delta: print(delta[content], end, flushTrue) except json.JSONDecodeError: print(f\n[Invalid JSON: {data}]) if __name__ __main__: asyncio.run(test_stream())运行这个脚本你会看到笑话被逐词打印出来模拟了类似ChatGPT的流式输出效果。6. 生产环境部署与进阶配置开发环境运行正常后我们需要考虑如何将其部署到生产环境并优化配置。6.1 使用Gunicorn管理Uvicorn Worker对于生产环境通常使用Gunicorn作为进程管理器来管理多个Uvicorn工作进程以提高并发能力和稳定性。首先安装Gunicornpip install gunicorn然后创建一个Gunicorn配置文件gunicorn_conf.py# gunicorn_conf.py import multiprocessing # 服务器套接字 bind 0.0.0.0:8000 # 工作进程数通常建议为 (2 * CPU核心数) 1 workers multiprocessing.cpu_count() * 2 1 # 每个工作进程的线程数Uvicorn是ASGI服务器通常worker_class用uvicorn.workers.UvicornWorker线程数设置无效 # 但我们可以指定worker类 worker_class uvicorn.workers.UvicornWorker # 工作进程最大请求数防止内存泄漏 max_requests 1000 max_requests_jitter 50 # 超时设置 timeout 120 keepalive 5 # 日志配置 accesslog - # 访问日志输出到stdout errorlog - # 错误日志输出到stderr loglevel info使用Gunicorn启动服务gunicorn -c gunicorn_conf.py app.main:app6.2 环境变量与配置管理生产环境不应使用.env文件而应使用环境变量或专业的配置管理服务如Vault。在Docker或Kubernetes中可以方便地注入环境变量。确保你的app/config.py中的Settings类能从环境变量正确读取。在服务器上你可以这样设置export OPENAI_API_KEYsk-prod-xxx export MODEL_NAMEgpt-4 # 然后启动应用6.3 添加认证与鉴权目前的API是公开的。生产环境必须添加认证。FastAPI支持多种方式如OAuth2、JWT等。以下是一个简单的API密钥认证示例在app/dependencies.py中添加# app/dependencies.py (新增) from fastapi import HTTPException, Security, Depends from fastapi.security import APIKeyHeader from app.config import settings API_KEY_NAME X-API-Key api_key_header APIKeyHeader(nameAPI_KEY_NAME, auto_errorFalse) async def verify_api_key(api_key: str Security(api_key_header)): if not api_key: raise HTTPException(status_code403, detail未提供API密钥) # 这里应该从数据库或配置中验证密钥此处简化 if api_key ! settings.admin_api_key: # 假设我们在settings里配置了ADMIN_API_KEY raise HTTPException(status_code403, detail无效的API密钥) return api_key然后在需要保护的路由上添加依赖# app/routers/chat.py from app.dependencies import verify_api_key router.post(/chat/completions, response_modelChatResponse, dependencies[Depends(verify_api_key)]) async def create_chat_completion(...): ...6.4 实现对话上下文管理当前的端点要求客户端每次发送完整的对话历史messages。对于多轮对话应用服务端应该维护会话状态。这可以通过数据库如Redis存储会话来实现。简化版思路创建一个/session端点用于创建新会话并返回session_id。修改/chat/completions端点接受session_id参数。服务端根据session_id从Redis中读取历史消息将新消息追加调用LLM然后将LLM的回复也存入历史最后返回回复。需要设置上下文长度限制和消息裁剪策略如只保留最近N条或最近N个token。6.5 监控与日志生产环境需要完善的监控和日志。日志使用Python的logging模块配置不同的Handler文件、Syslog、ELK等记录请求、响应、错误和性能数据。指标可以集成Prometheus客户端库如prometheus-fastapi-instrumentator来暴露应用指标请求数、延迟、错误率等。链路追踪对于分布式系统可以考虑集成OpenTelemetry。7. 常见问题排查与优化在开发和部署过程中你可能会遇到以下问题。7.1 请求报错 422 Unprocessable Entity这是FastAPI/Pydantic数据验证失败。检查点请求体格式错误确保JSON格式正确字段名拼写无误。字段类型不匹配例如temperature字段传了字符串而不是数字。缺少必填字段例如messages数组为空或未提供。排查查看FastAPI自动生成的/docs页面确认每个端点的请求体模型。或者查看服务日志FastAPI会输出详细的验证错误信息。7.2 流式响应不工作或中断可能原因代理或负载均衡器缓冲Nginx等默认会缓冲代理响应。需要在响应头中添加X-Accel-Buffering: no我们已在代码中添加并在Nginx配置中设置proxy_buffering off;。客户端未正确处理SSE确保客户端使用正确的方式接收SSE如JavaScript的EventSource或Fetch API的response.body.getReader()。超时设置过短LLM生成长回复可能需要超过30秒。确保服务器Uvicorn/Gunicorn和客户端超时设置足够长。网络连接不稳定检查网络。7.3 LLM API调用失败或超时可能原因API密钥无效或额度不足检查.env文件或环境变量中的OPENAI_API_KEY并确认账户状态。网络问题服务器可能无法访问OpenAI API需检查网络策略。请求频率超限免费或试用账号有速率限制。需要实现重试机制和退避策略。提示词过长导致上下文溢出错误信息可能包含context overflow。需要检查并限制客户端发送的messages总长度或在服务端实现自动裁剪。7.4 性能优化建议连接池为httpx.AsyncClient使用连接池避免为每个请求创建新连接。可以在LLMService的__init__中创建客户端实例并复用。异步全局状态如果使用数据库如Redis确保其客户端也是异步的如aioredis避免阻塞事件循环。缓存对于常见或重复的查询可以考虑在服务端添加缓存如使用cachetools库。监控与调优使用uvicorn --log-level debug或添加性能中间件来监控请求处理时间找出瓶颈。通过以上步骤你已经完成了一个从零到一、具备生产级潜力的FastAPI LLM服务后端。这个项目骨架涵盖了配置管理、依赖注入、路由设计、流式响应、错误处理等核心工程实践。在实际项目中你可以在此基础上扩展会话管理、多模型支持、异步任务队列、更复杂的Prompt工程等高级功能。记住良好的架构和清晰的代码分层是应对LLM应用快速迭代的关键。
FastAPI构建LLM问答服务：从零搭建生产级AI应用后端
在实际项目中快速构建一个高性能、易于维护的Web API是后端开发的核心需求。FastAPI凭借其现代的设计理念、极致的性能基于Starlette和Pydantic以及自动生成的交互式API文档已成为Python领域构建API的首选框架之一。与此同时大型语言模型LLM的应用开发正从简单的对话接口转向需要复杂业务逻辑、状态管理和精准流程控制的Agent或工具调用场景。将FastAPI与LLM开发结合能够为模型提供一个稳定、可扩展的后端服务层处理认证、会话管理、异步任务、数据持久化等工程问题让开发者更专注于Prompt设计、模型调用和业务逻辑本身。本文面向有一定Python基础希望快速掌握FastAPI并将其应用于实际LLM项目开发的开发者。我们将从零开始完成一个完整的LLM问答服务后端项目。这个项目不仅会涵盖FastAPI的核心用法还会深入如何集成LLM以OpenAI API为例、设计可维护的Prompt模板、处理流式响应、管理对话上下文并最终部署一个可用的服务。学完后你将能够独立搭建具备生产级潜力的LLM应用后端。1. 理解FastAPI的核心优势与LLM服务架构在开始编码之前需要明确为什么选择FastAPI来构建LLM服务以及一个典型的LLM服务后端包含哪些组件。1.1 FastAPI为何适合LLM项目LLM应用后端通常需要处理高并发、低延迟的请求特别是流式输出场景。FastAPI的异步支持async/await天生适合这类I/O密集型任务。其基于Pydantic的数据验证能确保发送给LLM的Prompt参数结构正确、类型安全自动生成的OpenAPI文档则方便前端或测试人员理解接口。此外FastAPI的依赖注入系统能优雅地管理LLM客户端、数据库连接等全局资源。一个常见的误区是认为LLM开发只需要调用模型API。实际上工程化部分同样重要包括请求验证与预处理确保用户输入符合预期并进行必要的清洗或转换。上下文管理维护多轮对话的历史记录并处理可能超长的上下文。异步与流式处理避免阻塞主线程并实现Token-by-Token的流式返回以提升用户体验。错误处理与重试优雅地处理模型服务不可用、超时、内容过滤等异常。可观测性记录日志、监控指标便于问题排查。FastAPI为上述所有需求提供了简洁而强大的基础框架。1.2 LLM服务后端的基本组件我们将构建的服务包含以下核心组件应用入口与路由FastAPI应用实例定义API端点。数据模型使用Pydantic定义请求体和响应体的结构。LLM客户端与服务层封装对OpenAI等模型API的调用包含Prompt构建逻辑。依赖注入管理LLM客户端、配置等的生命周期和共享。中间件用于处理CORS、请求日志等横切关注点。配置管理从环境变量或配置文件加载API密钥等敏感信息。项目的技术栈如下Python: 3.8Web框架: FastAPIASGI服务器: Uvicorn (用于运行FastAPI)HTTP客户端:httpx或openai官方库环境管理:python-dotenv数据验证: Pydantic (已包含在FastAPI中)2. 环境准备与项目初始化让我们从创建一个干净的项目环境开始。2.1 创建虚拟环境与安装依赖首先为项目创建一个独立的虚拟环境这是管理项目依赖的最佳实践。# 创建项目目录并进入 mkdir fastapi-llm-project cd fastapi-llm-project # 创建虚拟环境以venv为例 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate激活虚拟环境后命令行提示符前通常会出现(venv)标识。接下来安装核心依赖。# 安装FastAPI及其ASGI服务器Uvicorn pip install fastapi uvicorn # 安装HTTP客户端和配置管理库 pip install httpx python-dotenv # 安装OpenAI官方库用于调用GPT模型 pip install openai # 可选安装用于开发自动重载和调试的库 pip install watchfiles2.2 项目结构设计一个清晰的项目结构有助于代码维护。我们采用以下结构fastapi-llm-project/ ├── .env # 环境变量文件不提交到Git ├── .gitignore ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI应用创建和主路由 │ ├── config.py # 配置加载 │ ├── dependencies.py # 依赖注入项 │ ├── models.py # Pydantic数据模型 │ ├── services/ # 业务逻辑层 │ │ ├── __init__.py │ │ └── llm_service.py # LLM服务封装 │ └── routers/ # 路由模块 │ ├── __init__.py │ └── chat.py # 聊天相关路由 ├── requirements.txt # 项目依赖清单 └── README.md现在创建这些文件和目录。# 创建目录 mkdir -p app/services app/routers # 创建文件 touch .env .gitignore app/__init__.py app/main.py app/config.py app/dependencies.py app/models.py touch app/services/__init__.py app/services/llm_service.py touch app/routers/__init__.py app/routers/chat.py touch requirements.txt README.md2.3 管理环境变量与配置LLM项目通常需要API密钥等敏感信息。我们使用.env文件来管理它们并通过python-dotenv加载。首先在.env文件中添加你的OpenAI API密钥或其他LLM服务商密钥。# .env OPENAI_API_KEYsk-your-actual-openai-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果使用其他兼容API可修改此处 MODEL_NAMEgpt-3.5-turbo # 默认使用的模型注意务必在.gitignore中添加.env避免将密钥提交到版本控制系统。# .gitignore venv/ __pycache__/ *.pyc .env接下来创建app/config.py来集中管理配置。# app/config.py from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): 应用配置从环境变量加载 openai_api_key: str openai_base_url: str https://api.openai.com/v1 model_name: str gpt-3.5-turbo # 可以添加其他配置如数据库URL、日志级别等 # database_url: Optional[str] None # log_level: str INFO class Config: env_file .env # 指定从.env文件加载 env_file_encoding utf-8 # 创建全局配置实例 settings Settings()这里使用了Pydantic的BaseSettings它能自动从环境变量和.env文件加载配置并进行验证。如果OPENAI_API_KEY没有设置启动应用时会直接报错便于及早发现问题。3. 构建核心LLM服务与数据模型在定义API路由之前我们需要先封装LLM调用逻辑并设计好API的数据契约。3.1 定义Pydantic数据模型Pydantic模型定义了API接口的“形状”确保输入输出的数据类型和结构正确。在app/models.py中定义。# app/models.py from pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime class Message(BaseModel): 单条消息模型 role: str Field(..., description消息角色如 user, assistant, system) content: str Field(..., description消息内容) class ChatRequest(BaseModel): 聊天请求体 messages: List[Message] Field(..., description对话历史消息列表) model: Optional[str] Field(None, description指定使用的模型不填则使用配置默认值) stream: bool Field(False, description是否启用流式输出) temperature: float Field(0.7, ge0.0, le2.0, description采样温度控制随机性) max_tokens: Optional[int] Field(None, gt0, description回复的最大token数) class ChatResponse(BaseModel): 聊天响应体非流式 id: str object: str chat.completion created: int model: str choices: List[dict] usage: dict class StreamResponseChunk(BaseModel): 流式响应块 id: str object: str chat.completion.chunk created: int model: str choices: List[dict] class HealthResponse(BaseModel): 健康检查响应 status: str timestamp: datetimeField类用于提供字段的元数据如描述、默认值和验证规则ge大于等于le小于等于。这不仅能生成清晰的API文档还能在请求到达业务逻辑前就拦截非法数据。3.2 封装LLM服务我们将对OpenAI API的调用封装在一个服务类中以提高代码的可测试性和可维护性。创建app/services/llm_service.py。# app/services/llm_service.py import httpx import json from typing import AsyncGenerator, Optional from app.config import settings from app.models import ChatRequest import logging logger logging.getLogger(__name__) class LLMService: LLM服务封装类 def __init__(self): self.api_key settings.openai_api_key self.base_url settings.openai_base_url self.default_model settings.model_name self.headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } async def create_chat_completion( self, request: ChatRequest, timeout: float 30.0 ) - dict: 创建非流式聊天补全 model request.model or self.default_model payload { model: model, messages: [msg.dict() for msg in request.messages], temperature: request.temperature, max_tokens: request.max_tokens, stream: False } # 移除为None的字段 payload {k: v for k, v in payload.items() if v is not None} async with httpx.AsyncClient(timeouttimeout) as client: try: response await client.post( f{self.base_url}/chat/completions, headersself.headers, jsonpayload ) response.raise_for_status() # 如果状态码不是2xx抛出HTTPError return response.json() except httpx.HTTPStatusError as e: logger.error(fLLM API HTTP错误: {e.response.status_code} - {e.response.text}) raise except httpx.RequestError as e: logger.error(fLLM API 请求错误: {e}) raise async def create_chat_completion_stream( self, request: ChatRequest, timeout: float 60.0 ) - AsyncGenerator[str, None]: 创建流式聊天补全返回一个异步生成器 model request.model or self.default_model payload { model: model, messages: [msg.dict() for msg in request.messages], temperature: request.temperature, max_tokens: request.max_tokens, stream: True } payload {k: v for k, v in payload.items() if v is not None} async with httpx.AsyncClient(timeouttimeout) as client: try: async with client.stream( POST, f{self.base_url}/chat/completions, headersself.headers, jsonpayload ) as response: response.raise_for_status() async for chunk in response.aiter_lines(): if chunk: # OpenAI流式响应格式 data: {...}\n\n if chunk.startswith(data: ): data chunk[6:] # 去掉data: 前缀 if data.strip() [DONE]: break try: yield data except json.JSONDecodeError: logger.warning(f无法解析流式响应块: {data}) except httpx.HTTPStatusError as e: logger.error(fLLM流式API HTTP错误: {e.response.status_code}) raise except httpx.RequestError as e: logger.error(fLLM流式API 请求错误: {e}) raise关键点解释依赖注入配置服务从全局settings获取API密钥和URL便于统一管理。异常处理使用httpx的异常机制并记录日志便于排查网络或API错误。流式处理create_chat_completion_stream方法是一个异步生成器使用client.stream来逐步接收响应并通过yield逐个返回JSON字符串块。这是实现SSEServer-Sent Events的基础。超时控制为流式和非流式请求设置了不同的超时时间流式请求通常需要更长时间。3.3 设置依赖注入为了在路由中方便地使用LLMService我们通过FastAPI的依赖注入系统来管理它的实例。创建app/dependencies.py。# app/dependencies.py from app.services.llm_service import LLMService def get_llm_service() - LLMService: 获取LLM服务实例的依赖函数 # 这里可以扩展为更复杂的生命周期管理如连接池 return LLMService()4. 实现API路由与流式响应现在我们将创建具体的API端点包括健康检查、普通聊天和流式聊天。4.1 创建聊天路由在app/routers/chat.py中定义聊天相关的端点。# app/routers/chat.py from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import StreamingResponse import json import time from typing import Optional from app.dependencies import get_llm_service from app.models import ChatRequest, ChatResponse, StreamResponseChunk, HealthResponse from app.services.llm_service import LLMService router APIRouter(prefix/api/v1, tags[chat]) router.get(/health, response_modelHealthResponse) async def health_check(): 健康检查端点 return HealthResponse(statushealthy, timestampdatetime.now()) router.post(/chat/completions, response_modelChatResponse) async def create_chat_completion( request: ChatRequest, llm_service: LLMService Depends(get_llm_service) ): 非流式聊天补全端点。 请求体需符合ChatRequest模型返回完整的响应。 try: result await llm_service.create_chat_completion(request) return result except Exception as e: # 记录详细日志但返回给客户端的错误信息要简化 raise HTTPException(status_code500, detailf服务内部错误: {str(e)}) router.post(/chat/completions/stream) async def create_chat_completion_stream( request: ChatRequest, llm_service: LLMService Depends(get_llm_service) ): 流式聊天补全端点。 返回Server-Sent Events (SSE)流。 async def event_generator(): try: async for chunk in llm_service.create_chat_completion_stream(request): # 按照SSE格式返回数据 yield fdata: {chunk}\n\n except Exception as e: # 流式响应中发生错误可以发送一个错误事件如果客户端支持 error_chunk json.dumps({ error: { message: f流式响应生成失败: {str(e)}, type: internal_error } }) yield fdata: {error_chunk}\n\n finally: # 可选发送结束标记 yield data: [DONE]\n\n return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no # 禁用Nginx缓冲 } )关键点解释路由组织使用APIRouter并设置prefix和tags有助于在大型项目中模块化路由并使API文档更清晰。依赖注入使用在端点函数参数中声明llm_service: LLMService Depends(get_llm_service)FastAPI会自动调用get_llm_service函数并将返回的LLMService实例注入进来。流式响应/chat/completions/stream端点返回一个StreamingResponse。其核心是一个异步生成器函数event_generator它从llm_service获取数据块并格式化为SSE标准格式data: ...\n\n后yield出去。客户端如浏览器EventSource或Fetch API可以逐步接收这些数据。错误处理在流式响应中处理异常比较棘手因为响应头已经发送。这里我们尝试将错误信息也包装成SSE事件发送给客户端。生产环境中可能需要更复杂的错误恢复和重试逻辑。4.2 创建主应用并集成路由最后在app/main.py中创建FastAPI应用实例并集成我们定义的路由、中间件等。# app/main.py from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware import logging from app.routers import chat # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 创建FastAPI应用实例 app FastAPI( titleLLM Chat API, description一个基于FastAPI构建的LLM聊天API服务, version1.0.0 ) # 添加CORS中间件允许前端跨域访问 app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境应替换为具体的前端域名 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 包含路由 app.include_router(chat.router) app.on_event(startup) async def startup_event(): 应用启动时执行 logger.info(LLM Chat API 服务启动...) app.on_event(shutdown) async def shutdown_event(): 应用关闭时执行 logger.info(LLM Chat API 服务关闭...)5. 运行、测试与API文档验证服务代码已经完成现在让我们启动它并进行测试。5.1 启动开发服务器在项目根目录下运行以下命令启动Uvicorn服务器。uvicorn app.main:app --reload --host 0.0.0.0 --port 8000参数说明app.main:app指定FastAPI应用实例的位置app模块下的main.py文件中的app对象。--reload启用热重载代码修改后服务器会自动重启。仅用于开发。--host 0.0.0.0监听所有网络接口方便从其他设备访问。--port 8000指定端口号。看到类似以下输出说明服务已启动INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRLC to quit) INFO: Started reloader process [12345] using WatchFiles INFO: Started server process [12346] INFO: Waiting for application startup. INFO: Application startup complete.5.2 使用自动生成的交互式API文档FastAPI的一大亮点是自动生成OpenAPI文档。启动服务后访问以下URL交互式API文档 (Swagger UI): http://localhost:8000/docs替代API文档 (ReDoc): http://localhost:8000/redoc在Swagger UI界面你可以看到我们定义的所有端点/health,/api/v1/chat/completions,/api/v1/chat/completions/stream。你可以直接在这个界面上点击“Try it out”来测试API无需额外工具。5.3 使用curl或Postman进行测试测试健康检查端点curl -X GET http://localhost:8000/api/v1/health预期返回{status:healthy,timestamp:2023-10-27T08:00:00.123456}测试非流式聊天端点curl -X POST http://localhost:8000/api/v1/chat/completions \ -H Content-Type: application/json \ -d { messages: [ {role: user, content: 你好请介绍一下FastAPI。} ], stream: false }预期返回一个完整的JSON响应包含模型生成的回复。测试流式聊天端点curl -X POST http://localhost:8000/api/v1/chat/completions/stream \ -H Content-Type: application/json \ -H Accept: text/event-stream \ -d { messages: [ {role: user, content: 用Python写一个简单的Hello World程序。} ], stream: true }你会看到一系列以data:开头的行每行都是一个JSON块最后一行是data: [DONE]。这就是SSE流式响应。5.4 编写一个简单的Python测试客户端为了更好地理解流式响应可以编写一个简单的客户端脚本。# test_stream_client.py import asyncio import aiohttp import json async def test_stream(): url http://localhost:8000/api/v1/chat/completions/stream payload { messages: [{role: user, content: 讲一个笑话}], stream: True } async with aiohttp.ClientSession() as session: async with session.post(url, jsonpayload) as resp: async for line in resp.content: line line.decode(utf-8).strip() if line.startswith(data: ): data line[6:] # Remove data: if data [DONE]: print(\n[Stream finished]) break try: chunk json.loads(data) # 提取并打印delta content if chunk.get(choices): delta chunk[choices][0].get(delta, {}) if content in delta: print(delta[content], end, flushTrue) except json.JSONDecodeError: print(f\n[Invalid JSON: {data}]) if __name__ __main__: asyncio.run(test_stream())运行这个脚本你会看到笑话被逐词打印出来模拟了类似ChatGPT的流式输出效果。6. 生产环境部署与进阶配置开发环境运行正常后我们需要考虑如何将其部署到生产环境并优化配置。6.1 使用Gunicorn管理Uvicorn Worker对于生产环境通常使用Gunicorn作为进程管理器来管理多个Uvicorn工作进程以提高并发能力和稳定性。首先安装Gunicornpip install gunicorn然后创建一个Gunicorn配置文件gunicorn_conf.py# gunicorn_conf.py import multiprocessing # 服务器套接字 bind 0.0.0.0:8000 # 工作进程数通常建议为 (2 * CPU核心数) 1 workers multiprocessing.cpu_count() * 2 1 # 每个工作进程的线程数Uvicorn是ASGI服务器通常worker_class用uvicorn.workers.UvicornWorker线程数设置无效 # 但我们可以指定worker类 worker_class uvicorn.workers.UvicornWorker # 工作进程最大请求数防止内存泄漏 max_requests 1000 max_requests_jitter 50 # 超时设置 timeout 120 keepalive 5 # 日志配置 accesslog - # 访问日志输出到stdout errorlog - # 错误日志输出到stderr loglevel info使用Gunicorn启动服务gunicorn -c gunicorn_conf.py app.main:app6.2 环境变量与配置管理生产环境不应使用.env文件而应使用环境变量或专业的配置管理服务如Vault。在Docker或Kubernetes中可以方便地注入环境变量。确保你的app/config.py中的Settings类能从环境变量正确读取。在服务器上你可以这样设置export OPENAI_API_KEYsk-prod-xxx export MODEL_NAMEgpt-4 # 然后启动应用6.3 添加认证与鉴权目前的API是公开的。生产环境必须添加认证。FastAPI支持多种方式如OAuth2、JWT等。以下是一个简单的API密钥认证示例在app/dependencies.py中添加# app/dependencies.py (新增) from fastapi import HTTPException, Security, Depends from fastapi.security import APIKeyHeader from app.config import settings API_KEY_NAME X-API-Key api_key_header APIKeyHeader(nameAPI_KEY_NAME, auto_errorFalse) async def verify_api_key(api_key: str Security(api_key_header)): if not api_key: raise HTTPException(status_code403, detail未提供API密钥) # 这里应该从数据库或配置中验证密钥此处简化 if api_key ! settings.admin_api_key: # 假设我们在settings里配置了ADMIN_API_KEY raise HTTPException(status_code403, detail无效的API密钥) return api_key然后在需要保护的路由上添加依赖# app/routers/chat.py from app.dependencies import verify_api_key router.post(/chat/completions, response_modelChatResponse, dependencies[Depends(verify_api_key)]) async def create_chat_completion(...): ...6.4 实现对话上下文管理当前的端点要求客户端每次发送完整的对话历史messages。对于多轮对话应用服务端应该维护会话状态。这可以通过数据库如Redis存储会话来实现。简化版思路创建一个/session端点用于创建新会话并返回session_id。修改/chat/completions端点接受session_id参数。服务端根据session_id从Redis中读取历史消息将新消息追加调用LLM然后将LLM的回复也存入历史最后返回回复。需要设置上下文长度限制和消息裁剪策略如只保留最近N条或最近N个token。6.5 监控与日志生产环境需要完善的监控和日志。日志使用Python的logging模块配置不同的Handler文件、Syslog、ELK等记录请求、响应、错误和性能数据。指标可以集成Prometheus客户端库如prometheus-fastapi-instrumentator来暴露应用指标请求数、延迟、错误率等。链路追踪对于分布式系统可以考虑集成OpenTelemetry。7. 常见问题排查与优化在开发和部署过程中你可能会遇到以下问题。7.1 请求报错 422 Unprocessable Entity这是FastAPI/Pydantic数据验证失败。检查点请求体格式错误确保JSON格式正确字段名拼写无误。字段类型不匹配例如temperature字段传了字符串而不是数字。缺少必填字段例如messages数组为空或未提供。排查查看FastAPI自动生成的/docs页面确认每个端点的请求体模型。或者查看服务日志FastAPI会输出详细的验证错误信息。7.2 流式响应不工作或中断可能原因代理或负载均衡器缓冲Nginx等默认会缓冲代理响应。需要在响应头中添加X-Accel-Buffering: no我们已在代码中添加并在Nginx配置中设置proxy_buffering off;。客户端未正确处理SSE确保客户端使用正确的方式接收SSE如JavaScript的EventSource或Fetch API的response.body.getReader()。超时设置过短LLM生成长回复可能需要超过30秒。确保服务器Uvicorn/Gunicorn和客户端超时设置足够长。网络连接不稳定检查网络。7.3 LLM API调用失败或超时可能原因API密钥无效或额度不足检查.env文件或环境变量中的OPENAI_API_KEY并确认账户状态。网络问题服务器可能无法访问OpenAI API需检查网络策略。请求频率超限免费或试用账号有速率限制。需要实现重试机制和退避策略。提示词过长导致上下文溢出错误信息可能包含context overflow。需要检查并限制客户端发送的messages总长度或在服务端实现自动裁剪。7.4 性能优化建议连接池为httpx.AsyncClient使用连接池避免为每个请求创建新连接。可以在LLMService的__init__中创建客户端实例并复用。异步全局状态如果使用数据库如Redis确保其客户端也是异步的如aioredis避免阻塞事件循环。缓存对于常见或重复的查询可以考虑在服务端添加缓存如使用cachetools库。监控与调优使用uvicorn --log-level debug或添加性能中间件来监控请求处理时间找出瓶颈。通过以上步骤你已经完成了一个从零到一、具备生产级潜力的FastAPI LLM服务后端。这个项目骨架涵盖了配置管理、依赖注入、路由设计、流式响应、错误处理等核心工程实践。在实际项目中你可以在此基础上扩展会话管理、多模型支持、异步任务队列、更复杂的Prompt工程等高级功能。记住良好的架构和清晰的代码分层是应对LLM应用快速迭代的关键。