1. 先搞清楚 FastAPI 和 LLM 项目到底能解决什么问题如果你正在找一个能快速把想法变成 API 服务并且想用它来对接大语言模型LLM做点实际应用那 FastAPI 几乎是目前 Python 领域最直接的选择。它不是一个需要花几个月去学的重型框架核心优势就是“快”开发快、运行快、上手也快。很多人被“三天挑战”、“从入门到实战”这类标题吸引核心诉求其实很明确不想在 Web 框架本身耗费太多时间想尽快把精力放在 LLM 的应用逻辑上比如处理用户提问、构建智能对话、或者做一个带界面的工具。FastAPI 正好契合这个需求。它用 Python 的类型提示Type Hints来定义接口的输入输出自动生成交互式 API 文档并且天生支持异步Async这对于需要等待 LLM API 响应的场景非常友好。你不用再花大量时间去手动写参数校验、文档说明或者处理复杂的并发请求。所以这个组合的学习路径应该是先用 FastAPI 搭一个最简可用的 Web 服务架子然后立刻把 LLM 的调用逻辑“装”进去最后再考虑怎么优化 prompt、处理流式响应、管理对话状态这些更实际的问题。我建议一开始不要把目标定成“学会 FastAPI 的全部”而是聚焦在“如何用 FastAPI 为 LLM 项目提供服务”这条主线上。这样你学的每一个功能比如路径参数、请求体、依赖注入都能立刻对应到 LLM 项目中的一个具体需求上比如根据用户ID获取历史对话、接收复杂的 prompt 输入、或者验证 API 密钥。2. 环境准备别在配置上卡住第一步动手之前环境配置是第一个门槛但目标不是配一个“完美”的环境而是一个“能跑通”的环境。很多教程卡住新手的地方就在这里。核心就三样Python、FastAPI 和一个 ASGI 服务器。Python 版本确保你的 Python 版本是 3.7 及以上。这是 FastAPI 的最低要求。在命令行输入python --version或python3 --version确认一下。如果版本太低去 Python 官网下载安装新版本安装时记得勾选“Add Python to PATH”。创建虚拟环境这是强烈建议的一步它能避免不同项目间的包版本冲突。在你项目的根目录下执行# 创建虚拟环境 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate激活后命令行提示符前通常会显示(venv)表示你正在这个独立的环境里操作。安装核心包在激活的虚拟环境中运行以下命令pip install fastapi uvicornfastapi: 框架本身。uvicorn: 一个轻量级的 ASGI 服务器用于运行 FastAPI 应用。它是开发和生产环境常用的选择。可选但重要的包根据你的 LLM 项目需求可能还需要pip install httpx # 用于异步 HTTP 客户端调用外部 LLM API pip install python-dotenv # 管理环境变量存放 API Key 等敏感信息 pip install pydantic # FastAPI 深度集成用于数据验证虽然 fastapi 会自带但有时需要独立使用验证安装创建一个最简单的main.py文件from fastapi import FastAPI app FastAPI() app.get(/) def read_root(): return {Hello: World}然后在终端运行uvicorn main:app --reload访问http://127.0.0.1:8000看到{Hello:World}就成功了。访问http://127.0.0.1:8000/docs你会看到自动生成的交互式 API 文档Swagger UI。这个“开箱即用”的文档功能是 FastAPI 在开发效率上的一大体现。3. FastAPI 核心概念快速上手为 LLM 接口铺路不需要面面俱到我们只学马上能用于 LLM 项目的部分。3.1 定义路由和请求方法LLM 应用最常见的就是接收用户的文本Prompt所以POST方法是主角。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() # 定义一个数据模型用来描述请求体 class PromptRequest(BaseModel): prompt: str max_tokens: int 100 # 默认值 temperature: float 0.7 app.post(/chat/) async def chat_completion(request: PromptRequest): 接收一个 prompt返回 LLM 的补全结果。 这里先用模拟逻辑后续替换为真实 LLM API 调用。 # 模拟 LLM 处理 simulated_response fReceived: {request.prompt}. Thinking... Here is a simulated response. return { response: simulated_response, input_tokens: len(request.prompt), parameters: {max_tokens: request.max_tokens, temperature: request.temperature} }关键点PromptRequest类继承BaseModel定义了客户端必须发送的 JSON 结构。FastAPI 会自动校验如果prompt不是字符串或temperature不是数字会直接返回 422 错误这正是搜索热词里提到的报错422 Unprocessable Entity的常见原因——请求体格式不符合定义。async def定义了异步函数。虽然这个例子是模拟但当你真正调用 OpenAI、DeepSeek 等外部 API 时使用async/await可以避免在等待网络响应时阻塞服务器提升并发能力。3.2 路径参数和查询参数除了请求体还有其他方式传递信息。路径参数用于标识特定资源比如获取某次对话的历史。app.get(/conversations/{conversation_id}) async def get_conversation(conversation_id: int): # 根据 conversation_id 从数据库或缓存中查找 return {conversation_id: conversation_id, history: [...]}查询参数用于过滤、分页或提供可选参数比如流式输出。app.post(/chat/stream) async def chat_stream(prompt: str, stream: bool False): if stream: # 这里可以实现 Server-Sent Events (SSE) 流式返回 # 模拟逐词输出 words fStreaming response for: {prompt}.split() for word in words: yield {data: word} # 模拟延迟 # await asyncio.sleep(0.1) return {response: Non-streaming response}调用时可以是/chat/stream?prompt你好streamtrue。3.3 依赖注入管理共享逻辑这是 FastAPI 非常强大的特性适合处理 LLM 项目中的通用逻辑比如验证 API 密钥from fastapi import Depends, FastAPI, HTTPException, Header app FastAPI() async def verify_token(x_api_key: str Header(...)): # 这里应该从数据库或环境变量验证密钥 if x_api_key ! expected-secret-key: raise HTTPException(status_code403, detailInvalid API Key) return x_api_key app.post(/secure-chat/) async def secure_chat(request: PromptRequest, api_key: str Depends(verify_token)): # 只有通过了 verify_token 依赖项验证的请求才能执行到这里 return {response: This is a secure endpoint., used_key: api_key}获取 LLM 客户端避免在每个路由函数里重复初始化。import httpx from fastapi import Depends async def get_llm_client(): # 初始化一个可复用的异步 HTTP 客户端 async with httpx.AsyncClient(timeout30.0) as client: yield client app.post(/call-openai/) async def call_openai(prompt: str, client: httpx.AsyncClient Depends(get_llm_client)): # 使用注入的 client 调用 OpenAI API # 注意此处为示例实际需要替换为正确的 API 地址和参数 # resp await client.post(https://api.openai.com/v1/chat/completions, ...) return {message: fWould call LLM with: {prompt}}依赖注入让代码更清晰、更易测试也更容易管理资源如数据库连接、HTTP 客户端。4. 连接 LLM从模拟到真实 API 调用框架搭好了接下来就是把 LLM 接进来。我们分两步走先模拟再对接真实服务。4.1 模拟 LLM 响应快速验证在项目初期直接调用收费或需要网络权限的 LLM API 可能不方便。我们可以先创建一个模拟服务确保 FastAPI 的请求/响应流程完全正确。import asyncio from typing import AsyncGenerator from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): message: str # 模拟一个慢速的、流式的响应生成器 async def mock_llm_streamer(text: str) - AsyncGenerator[str, None]: words text.split() for word in words: await asyncio.sleep(0.2) # 模拟生成延迟 yield fdata: {word}\n\n app.post(/mock-chat/stream) async def mock_chat_stream(request: ChatRequest): # 构造一个模拟的“思考”过程 simulated_thought fIm thinking about: {request.message}. Let me generate a response word by word. generator mock_llm_streamer(simulated_thought) return StreamingResponse(generator, media_typetext/event-stream)这个端点返回的是text/event-stream格式适合前端使用 EventSource 进行流式接收。用 curl 测试一下curl -N -X POST http://127.0.0.1:8000/mock-chat/stream -H Content-Type: application/json -d {message:Hello}你会看到数据逐词返回。这一步验证了你的 FastAPI 后端已经具备了处理流式请求的能力。4.2 对接真实 LLM API以 OpenAI 格式为例当模拟流程跑通后替换为真实的 API 调用。这里以兼容 OpenAI API 格式的服务为例如 OpenAI、DeepSeek、本地部署的 vLLM 服务等。import os import httpx from fastapi import FastAPI, HTTPException from pydantic import BaseModel from dotenv import load_dotenv load_dotenv() # 从 .env 文件加载环境变量 app FastAPI() class LLMRequest(BaseModel): model: str gpt-3.5-turbo # 指定模型 messages: list[dict] # 对话历史 stream: bool False app.post(/v1/chat/completions) async def chat_completions(request: LLMRequest): api_key os.getenv(LLM_API_KEY) api_base os.getenv(LLM_API_BASE, https://api.openai.com/v1) if not api_key: raise HTTPException(status_code500, detailAPI key not configured) headers { Authorization: fBearer {api_key}, Content-Type: application/json, } payload request.dict() async with httpx.AsyncClient(timeout60.0) as client: try: if request.stream: # 流式请求 async with client.stream( POST, f{api_base}/chat/completions, headersheaders, jsonpayload, ) as response: if response.status_code ! 200: error_detail await response.aread() raise HTTPException(status_coderesponse.status_code, detailerror_detail.decode()) # 将上游的流式响应直接转发给客户端 return StreamingResponse(response.aiter_bytes(), media_typetext/event-stream) else: # 非流式请求 resp await client.post(f{api_base}/chat/completions, headersheaders, jsonpayload) resp.raise_for_status() return resp.json() except httpx.RequestError as e: raise HTTPException(status_code503, detailfLLM service request failed: {str(e)})关键配置和避坑点环境变量将LLM_API_KEY和LLM_API_BASE放在项目根目录的.env文件中不要硬编码在代码里。LLM_API_KEYsk-your-actual-key-here LLM_API_BASEhttps://api.openai.com/v1超时设置LLM 生成可能很慢httpx.AsyncClient和请求的超时timeout60.0要设置得足够长。错误处理网络错误、API 密钥错误、模型不可用等情况都要捕获并返回清晰的错误信息给前端而不是让服务崩溃。流式转发代码中流式处理的部分直接将上游的字节流转发下去避免在中间层进行不必要的拼接和解析这能减少延迟和内存占用。5. Prompt 处理与工程化实践LLM 项目的核心不仅是调用 API更是如何构建和管理 prompt。FastAPI 可以作为你 prompt 工程化的“工作台”。5.1 结构化 Prompt 输入不要让用户只传一个字符串而是设计结构化的输入便于添加系统指令、上下文、示例等。from pydantic import BaseModel, Field from typing import Optional, List class ConversationMessage(BaseModel): role: str Field(..., description角色: system, user, assistant) content: str class StructuredPromptRequest(BaseModel): system_prompt: Optional[str] 你是一个有帮助的助手。 conversation_history: List[ConversationMessage] [] current_query: str temperature: float Field(0.7, ge0.0, le2.0, description温度参数范围0-2) top_p: Optional[float] Field(None, ge0.0, le1.0) def to_openai_format(self): 将结构化数据转换为 OpenAI API 所需的 messages 格式 messages [] if self.system_prompt: messages.append({role: system, content: self.system_prompt}) messages.extend([{role: msg.role, content: msg.content} for msg in self.conversation_history]) messages.append({role: user, content: self.current_query}) return messages app.post(/structured-chat/) async def structured_chat(request: StructuredPromptRequest): openai_messages request.to_openai_format() # ... 调用 LLM API使用 openai_messages return {formatted_messages: openai_messages}这样做的好处是前端可以清晰地构建多轮对话后端也能方便地对不同字段做校验比如用Field的ge,le限制参数范围。5.2 实现 Prompt 模板与变量替换对于固定场景如客服、代码生成可以使用模板。from string import Template PROMPT_TEMPLATES { code_review: Template( 请审查以下 $language 代码\n$language\n$code\n\n 请重点关注$focus_areas。请用中文给出审查意见。 ), summary: Template(请用中文总结以下文本要求不超过 $max_words 字\n$text) } app.post(/generate-from-template/) async def generate_from_template(template_name: str, variables: dict): if template_name not in PROMPT_TEMPLATES: raise HTTPException(status_code400, detailTemplate not found) template PROMPT_TEMPLATES[template_name] try: prompt template.safe_substitute(variables) except KeyError as e: raise HTTPException(status_code400, detailfMissing template variable: {e}) # 使用生成的 prompt 调用 LLM return {generated_prompt: prompt, final_response: Simulated LLM output based on prompt.}调用示例POST /generate-from-template/with JSON body{template_name: code_review, variables: {language: python, code: def foo(): pass, focus_areas: 可读性和异常处理}}。5.3 处理常见 Prompt 相关问题从搜索热词看大家常遇到context overflow上下文溢出和prompt injection提示注入问题。上下文溢出当对话历史太长超过模型的最大上下文长度时需要实现一个“剪裁”策略。简单的做法是保留最近的 N 条消息或者优先保留system提示和最近的消息。def trim_conversation_history(messages: List[dict], max_tokens_estimate: int 3000): # 这是一个简化的示例实际需要根据 token 数精确计算 if len(messages) 5: # 假设5条以内不会超长 return messages # 保留 system 消息和最新的4条对话 system_msg [m for m in messages if m[role] system] recent_msgs messages[-4:] return system_msg recent_msgs提示注入防御如果允许用户输入部分系统指令就需要防范用户输入覆盖你的系统设定。一个基本策略是严格区分用户可控部分和系统指令并在最终拼接时将系统指令放在一个用户无法修改的字段中而不是让用户输入包含“忽略之前指令”这类文本。在结构化请求中system_prompt由后端控制或严格过滤是更安全的做法。6. 项目实战构建一个简单的 LLM 问答服务我们把前面的知识点串起来构建一个具备基本功能的 LLM 问答后端。项目结构llm_fastapi_project/ ├── .env # 环境变量 ├── .gitignore ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用实例和路由 │ ├── dependencies.py # 依赖项如认证、客户端 │ ├── models.py # Pydantic 数据模型 │ ├── services/ # 业务逻辑 │ │ ├── __init__.py │ │ └── llm_service.py # LLM 调用封装 │ └── utils.py # 工具函数如 prompt 处理 ├── requirements.txt └── README.md核心文件app/main.pyfrom fastapi import FastAPI, Depends from app.dependencies import get_llm_service from app.models import ChatRequest, ChatResponse from app.services.llm_service import LLMService app FastAPI(titleLLM QA Service, descriptionA simple QA backend powered by FastAPI and LLM.) app.post(/ask, response_modelChatResponse) async def ask_question( request: ChatRequest, llm_service: LLMService Depends(get_llm_service) ): 核心问答接口。 1. 接收用户问题。 2. 可选检索上下文RAG。 3. 构造 prompt。 4. 调用 LLM。 5. 返回响应。 # 这里可以加入上下文检索逻辑RAG # context retrieve_context(request.question) # 构造最终 prompt final_prompt llm_service.construct_prompt(request.question, contextNone) # 调用 LLM llm_response await llm_service.generate_response(final_prompt, request.stream) return ChatResponse( answerllm_response[content], usagellm_response.get(usage, {}), streamrequest.stream ) app.get(/health) async def health_check(): return {status: healthy}服务层app/services/llm_service.pyimport os import httpx from typing import Optional, AsyncGenerator from dotenv import load_dotenv load_dotenv() class LLMService: def __init__(self): self.api_key os.getenv(LLM_API_KEY) self.api_base os.getenv(LLM_API_BASE) self.default_model os.getenv(LLM_DEFAULT_MODEL, gpt-3.5-turbo) if not self.api_key: raise ValueError(LLM_API_KEY not set in environment variables.) def construct_prompt(self, question: str, context: Optional[str]) - str: # 简单的 prompt 工程 if context: prompt f基于以下上下文信息回答用户的问题。\n上下文{context}\n\n问题{question}\n回答 else: prompt f请回答以下问题{question} return prompt async def generate_response(self, prompt: str, stream: bool False): headers { Authorization: fBearer {self.api_key}, Content-Type: application/json, } payload { model: self.default_model, messages: [{role: user, content: prompt}], stream: stream, temperature: 0.7, } async with httpx.AsyncClient(timeout60.0) as client: try: if stream: async with client.stream( POST, f{self.api_base}/chat/completions, headersheaders, jsonpayload, ) as response: response.raise_for_status() async for chunk in response.aiter_lines(): # 这里需要解析 SSE 格式的 chunk if chunk.startswith(data: ): yield chunk else: resp await client.post(f{api_base}/chat/completions, headersheaders, jsonpayload) resp.raise_for_status() return resp.json() except httpx.HTTPStatusError as e: # 处理 HTTP 错误如 429 限速 401 鉴权失败 error_detail e.response.json().get(error, {}).get(message, str(e)) raise Exception(fLLM API error: {error_detail}) except httpx.RequestError as e: # 处理网络错误 raise Exception(fNetwork error contacting LLM service: {str(e)})运行与测试配置好.env文件。安装依赖pip install -r requirements.txt(内容为fastapi,uvicorn,httpx,python-dotenv,pydantic)。启动服务uvicorn app.main:app --reload --host 0.0.0.0 --port 8000。打开http://127.0.0.1:8000/docs在/ask接口的交互界面中测试。使用 curl 测试流式接口curl -N -X POST http://127.0.0.1:8000/ask \ -H Content-Type: application/json \ -d {question: Python中staticmethod装饰器有什么用, stream: true}7. 部署与生产环境考量本地开发跑通后如果要对外服务需要考虑以下几点ASGI 服务器与性能开发时用的uvicorn --reload不适合生产。生产环境通常使用uvicorn配合多进程--workers或搭配gunicorn一个 WSGI/ASGI 服务器管理器。# 使用 gunicorn 管理多个 uvicorn worker 进程 gunicorn -w 4 -k uvicorn.workers.UvicornWorker app.main:app --bind 0.0.0.0:8000-w 4表示启动 4 个 worker 进程。worker 数量通常设置为 CPU 核心数的 1-2 倍。搜索热词中有人问fastapi默认多少线程需要澄清FastAPI 本身不管理线程它运行在 ASGI 服务器如 uvicorn上。Uvicorn 默认是单进程单线程异步通过异步 IO 处理并发。使用gunicorn后才是多进程模型。环境变量与配置生产环境绝不能使用.env文件。应使用 Docker 的-e参数、Kubernetes 的 ConfigMap/Secret、或云服务商的环境变量管理功能。日志记录FastAPI 使用标准的 Python logging。配置日志记录请求、响应和错误便于排查问题。import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) app.post(/ask) async def ask_question(request: ChatRequest): logger.info(fReceived question: {request.question[:50]}...) # ... 业务逻辑 logger.info(fQuestion answered successfully.) return response超时与重试在调用外部 LLM API 时网络不稳定可能导致超时。在生产代码中考虑为httpx.AsyncClient增加重试机制可以使用tenacity库并设置合理的超时时间。速率限制与队列如果用户量较大直接无限制地转发请求到 LLM API 可能导致对方速率限制Rate Limit。需要在 FastAPI 层实现简单的速率限制如使用slowapi库或任务队列如使用Celery或RQ将请求排队处理。容器化部署Docker这是最通用的部署方式。一个简单的DockerfileFROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [gunicorn, -w, 4, -k, uvicorn.workers.UvicornWorker, app.main:app, --bind, 0.0.0.0:8000]构建并运行docker build -t llm-fastapi .和docker run -p 8000:8000 --env-file .env llm-fastapi。8. 常见问题排查清单当你的 FastAPI LLM 项目跑不起来或者行为异常时按这个顺序排查服务根本起不来uvicorn报错检查 Python 和包版本python --version,pip list | grep fastapi。确保版本兼容。检查导入路径uvicorn main:app中的main是文件名不含.pyapp是文件内的FastAPI()实例变量名。检查端口占用默认 8000 端口是否被其他程序占用可以换端口--port 8001。API 请求返回 422 Unprocessable Entity这是最常见的问题之一。几乎总是因为请求体的 JSON 格式不符合你定义的 Pydantic 模型。打开/docs页面直接用 Swagger UI 的“Try it out”功能测试它会自动生成符合格式的请求。检查字段名和类型是否拼写错误prompt写成了promtmax_tokens传了字符串100而不是数字100使用curl或 Postman 时确保Content-Type: application/json头已设置且 JSON 格式正确。调用 LLM API 失败返回 4xx/5xx 错误检查 API 密钥和环境变量print(os.getenv(LLM_API_KEY))看看是否成功加载。生产环境确保变量已注入。检查 API 基础地址如果是自托管模型如使用vllm地址可能是http://localhost:8000/v1。查看 LLM 服务方的错误信息日志或返回的 JSON 中通常有error字段会明确告知是额度不足、模型不存在还是参数错误。流式响应不工作或中断前端是否正确处理 SSE后端返回StreamingResponse前端需要使用EventSource或fetch以流式方式读取。检查超时设置服务器、反向代理如 Nginx和客户端都可能有关闭空闲连接的超时设置。需要适当调大。网络稳定性不稳定的网络连接可能导致流中断。性能差响应慢定位瓶颈使用中间件记录每个请求的处理时间。慢是在 FastAPI 逻辑部分还是在调用 LLM API 的网络等待部分LLM API 本身慢考虑使用更快的模型、调整参数如降低max_tokens、或实现客户端缓存对相同问题缓存答案。数据库或上下文检索慢如果你集成了 RAG检查向量数据库的检索速度。并发请求下内存或 CPU 占用高调整 worker 数量gunicorn -w参数不要设置得远超 CPU 核心数。检查内存泄漏长时间运行后内存是否持续增长可能是全局变量缓存了过多数据或者 HTTP 客户端未正确复用/关闭。使用异步客户端确保使用httpx.AsyncClient并在依赖中正确管理其生命周期如使用yield。这个学习路径的核心是“快速验证逐步深化”。不要试图第一天就搭建一个完美的生产系统。先用 FastAPI 把最简单的“接收 prompt - 调用 LLM - 返回结果”的链路跑通然后逐步加入错误处理、流式响应、结构化 prompt、认证、日志和部署。每加一个功能都确保之前的还能工作。这样三天时间足够你从一个完全的新手到一个能搭建出可用 LLM 服务后端的状态。 FastAPI快速构建LLM应用：从环境配置到生产部署实战指南
1. 先搞清楚 FastAPI 和 LLM 项目到底能解决什么问题如果你正在找一个能快速把想法变成 API 服务并且想用它来对接大语言模型LLM做点实际应用那 FastAPI 几乎是目前 Python 领域最直接的选择。它不是一个需要花几个月去学的重型框架核心优势就是“快”开发快、运行快、上手也快。很多人被“三天挑战”、“从入门到实战”这类标题吸引核心诉求其实很明确不想在 Web 框架本身耗费太多时间想尽快把精力放在 LLM 的应用逻辑上比如处理用户提问、构建智能对话、或者做一个带界面的工具。FastAPI 正好契合这个需求。它用 Python 的类型提示Type Hints来定义接口的输入输出自动生成交互式 API 文档并且天生支持异步Async这对于需要等待 LLM API 响应的场景非常友好。你不用再花大量时间去手动写参数校验、文档说明或者处理复杂的并发请求。所以这个组合的学习路径应该是先用 FastAPI 搭一个最简可用的 Web 服务架子然后立刻把 LLM 的调用逻辑“装”进去最后再考虑怎么优化 prompt、处理流式响应、管理对话状态这些更实际的问题。我建议一开始不要把目标定成“学会 FastAPI 的全部”而是聚焦在“如何用 FastAPI 为 LLM 项目提供服务”这条主线上。这样你学的每一个功能比如路径参数、请求体、依赖注入都能立刻对应到 LLM 项目中的一个具体需求上比如根据用户ID获取历史对话、接收复杂的 prompt 输入、或者验证 API 密钥。2. 环境准备别在配置上卡住第一步动手之前环境配置是第一个门槛但目标不是配一个“完美”的环境而是一个“能跑通”的环境。很多教程卡住新手的地方就在这里。核心就三样Python、FastAPI 和一个 ASGI 服务器。Python 版本确保你的 Python 版本是 3.7 及以上。这是 FastAPI 的最低要求。在命令行输入python --version或python3 --version确认一下。如果版本太低去 Python 官网下载安装新版本安装时记得勾选“Add Python to PATH”。创建虚拟环境这是强烈建议的一步它能避免不同项目间的包版本冲突。在你项目的根目录下执行# 创建虚拟环境 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate激活后命令行提示符前通常会显示(venv)表示你正在这个独立的环境里操作。安装核心包在激活的虚拟环境中运行以下命令pip install fastapi uvicornfastapi: 框架本身。uvicorn: 一个轻量级的 ASGI 服务器用于运行 FastAPI 应用。它是开发和生产环境常用的选择。可选但重要的包根据你的 LLM 项目需求可能还需要pip install httpx # 用于异步 HTTP 客户端调用外部 LLM API pip install python-dotenv # 管理环境变量存放 API Key 等敏感信息 pip install pydantic # FastAPI 深度集成用于数据验证虽然 fastapi 会自带但有时需要独立使用验证安装创建一个最简单的main.py文件from fastapi import FastAPI app FastAPI() app.get(/) def read_root(): return {Hello: World}然后在终端运行uvicorn main:app --reload访问http://127.0.0.1:8000看到{Hello:World}就成功了。访问http://127.0.0.1:8000/docs你会看到自动生成的交互式 API 文档Swagger UI。这个“开箱即用”的文档功能是 FastAPI 在开发效率上的一大体现。3. FastAPI 核心概念快速上手为 LLM 接口铺路不需要面面俱到我们只学马上能用于 LLM 项目的部分。3.1 定义路由和请求方法LLM 应用最常见的就是接收用户的文本Prompt所以POST方法是主角。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() # 定义一个数据模型用来描述请求体 class PromptRequest(BaseModel): prompt: str max_tokens: int 100 # 默认值 temperature: float 0.7 app.post(/chat/) async def chat_completion(request: PromptRequest): 接收一个 prompt返回 LLM 的补全结果。 这里先用模拟逻辑后续替换为真实 LLM API 调用。 # 模拟 LLM 处理 simulated_response fReceived: {request.prompt}. Thinking... Here is a simulated response. return { response: simulated_response, input_tokens: len(request.prompt), parameters: {max_tokens: request.max_tokens, temperature: request.temperature} }关键点PromptRequest类继承BaseModel定义了客户端必须发送的 JSON 结构。FastAPI 会自动校验如果prompt不是字符串或temperature不是数字会直接返回 422 错误这正是搜索热词里提到的报错422 Unprocessable Entity的常见原因——请求体格式不符合定义。async def定义了异步函数。虽然这个例子是模拟但当你真正调用 OpenAI、DeepSeek 等外部 API 时使用async/await可以避免在等待网络响应时阻塞服务器提升并发能力。3.2 路径参数和查询参数除了请求体还有其他方式传递信息。路径参数用于标识特定资源比如获取某次对话的历史。app.get(/conversations/{conversation_id}) async def get_conversation(conversation_id: int): # 根据 conversation_id 从数据库或缓存中查找 return {conversation_id: conversation_id, history: [...]}查询参数用于过滤、分页或提供可选参数比如流式输出。app.post(/chat/stream) async def chat_stream(prompt: str, stream: bool False): if stream: # 这里可以实现 Server-Sent Events (SSE) 流式返回 # 模拟逐词输出 words fStreaming response for: {prompt}.split() for word in words: yield {data: word} # 模拟延迟 # await asyncio.sleep(0.1) return {response: Non-streaming response}调用时可以是/chat/stream?prompt你好streamtrue。3.3 依赖注入管理共享逻辑这是 FastAPI 非常强大的特性适合处理 LLM 项目中的通用逻辑比如验证 API 密钥from fastapi import Depends, FastAPI, HTTPException, Header app FastAPI() async def verify_token(x_api_key: str Header(...)): # 这里应该从数据库或环境变量验证密钥 if x_api_key ! expected-secret-key: raise HTTPException(status_code403, detailInvalid API Key) return x_api_key app.post(/secure-chat/) async def secure_chat(request: PromptRequest, api_key: str Depends(verify_token)): # 只有通过了 verify_token 依赖项验证的请求才能执行到这里 return {response: This is a secure endpoint., used_key: api_key}获取 LLM 客户端避免在每个路由函数里重复初始化。import httpx from fastapi import Depends async def get_llm_client(): # 初始化一个可复用的异步 HTTP 客户端 async with httpx.AsyncClient(timeout30.0) as client: yield client app.post(/call-openai/) async def call_openai(prompt: str, client: httpx.AsyncClient Depends(get_llm_client)): # 使用注入的 client 调用 OpenAI API # 注意此处为示例实际需要替换为正确的 API 地址和参数 # resp await client.post(https://api.openai.com/v1/chat/completions, ...) return {message: fWould call LLM with: {prompt}}依赖注入让代码更清晰、更易测试也更容易管理资源如数据库连接、HTTP 客户端。4. 连接 LLM从模拟到真实 API 调用框架搭好了接下来就是把 LLM 接进来。我们分两步走先模拟再对接真实服务。4.1 模拟 LLM 响应快速验证在项目初期直接调用收费或需要网络权限的 LLM API 可能不方便。我们可以先创建一个模拟服务确保 FastAPI 的请求/响应流程完全正确。import asyncio from typing import AsyncGenerator from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): message: str # 模拟一个慢速的、流式的响应生成器 async def mock_llm_streamer(text: str) - AsyncGenerator[str, None]: words text.split() for word in words: await asyncio.sleep(0.2) # 模拟生成延迟 yield fdata: {word}\n\n app.post(/mock-chat/stream) async def mock_chat_stream(request: ChatRequest): # 构造一个模拟的“思考”过程 simulated_thought fIm thinking about: {request.message}. Let me generate a response word by word. generator mock_llm_streamer(simulated_thought) return StreamingResponse(generator, media_typetext/event-stream)这个端点返回的是text/event-stream格式适合前端使用 EventSource 进行流式接收。用 curl 测试一下curl -N -X POST http://127.0.0.1:8000/mock-chat/stream -H Content-Type: application/json -d {message:Hello}你会看到数据逐词返回。这一步验证了你的 FastAPI 后端已经具备了处理流式请求的能力。4.2 对接真实 LLM API以 OpenAI 格式为例当模拟流程跑通后替换为真实的 API 调用。这里以兼容 OpenAI API 格式的服务为例如 OpenAI、DeepSeek、本地部署的 vLLM 服务等。import os import httpx from fastapi import FastAPI, HTTPException from pydantic import BaseModel from dotenv import load_dotenv load_dotenv() # 从 .env 文件加载环境变量 app FastAPI() class LLMRequest(BaseModel): model: str gpt-3.5-turbo # 指定模型 messages: list[dict] # 对话历史 stream: bool False app.post(/v1/chat/completions) async def chat_completions(request: LLMRequest): api_key os.getenv(LLM_API_KEY) api_base os.getenv(LLM_API_BASE, https://api.openai.com/v1) if not api_key: raise HTTPException(status_code500, detailAPI key not configured) headers { Authorization: fBearer {api_key}, Content-Type: application/json, } payload request.dict() async with httpx.AsyncClient(timeout60.0) as client: try: if request.stream: # 流式请求 async with client.stream( POST, f{api_base}/chat/completions, headersheaders, jsonpayload, ) as response: if response.status_code ! 200: error_detail await response.aread() raise HTTPException(status_coderesponse.status_code, detailerror_detail.decode()) # 将上游的流式响应直接转发给客户端 return StreamingResponse(response.aiter_bytes(), media_typetext/event-stream) else: # 非流式请求 resp await client.post(f{api_base}/chat/completions, headersheaders, jsonpayload) resp.raise_for_status() return resp.json() except httpx.RequestError as e: raise HTTPException(status_code503, detailfLLM service request failed: {str(e)})关键配置和避坑点环境变量将LLM_API_KEY和LLM_API_BASE放在项目根目录的.env文件中不要硬编码在代码里。LLM_API_KEYsk-your-actual-key-here LLM_API_BASEhttps://api.openai.com/v1超时设置LLM 生成可能很慢httpx.AsyncClient和请求的超时timeout60.0要设置得足够长。错误处理网络错误、API 密钥错误、模型不可用等情况都要捕获并返回清晰的错误信息给前端而不是让服务崩溃。流式转发代码中流式处理的部分直接将上游的字节流转发下去避免在中间层进行不必要的拼接和解析这能减少延迟和内存占用。5. Prompt 处理与工程化实践LLM 项目的核心不仅是调用 API更是如何构建和管理 prompt。FastAPI 可以作为你 prompt 工程化的“工作台”。5.1 结构化 Prompt 输入不要让用户只传一个字符串而是设计结构化的输入便于添加系统指令、上下文、示例等。from pydantic import BaseModel, Field from typing import Optional, List class ConversationMessage(BaseModel): role: str Field(..., description角色: system, user, assistant) content: str class StructuredPromptRequest(BaseModel): system_prompt: Optional[str] 你是一个有帮助的助手。 conversation_history: List[ConversationMessage] [] current_query: str temperature: float Field(0.7, ge0.0, le2.0, description温度参数范围0-2) top_p: Optional[float] Field(None, ge0.0, le1.0) def to_openai_format(self): 将结构化数据转换为 OpenAI API 所需的 messages 格式 messages [] if self.system_prompt: messages.append({role: system, content: self.system_prompt}) messages.extend([{role: msg.role, content: msg.content} for msg in self.conversation_history]) messages.append({role: user, content: self.current_query}) return messages app.post(/structured-chat/) async def structured_chat(request: StructuredPromptRequest): openai_messages request.to_openai_format() # ... 调用 LLM API使用 openai_messages return {formatted_messages: openai_messages}这样做的好处是前端可以清晰地构建多轮对话后端也能方便地对不同字段做校验比如用Field的ge,le限制参数范围。5.2 实现 Prompt 模板与变量替换对于固定场景如客服、代码生成可以使用模板。from string import Template PROMPT_TEMPLATES { code_review: Template( 请审查以下 $language 代码\n$language\n$code\n\n 请重点关注$focus_areas。请用中文给出审查意见。 ), summary: Template(请用中文总结以下文本要求不超过 $max_words 字\n$text) } app.post(/generate-from-template/) async def generate_from_template(template_name: str, variables: dict): if template_name not in PROMPT_TEMPLATES: raise HTTPException(status_code400, detailTemplate not found) template PROMPT_TEMPLATES[template_name] try: prompt template.safe_substitute(variables) except KeyError as e: raise HTTPException(status_code400, detailfMissing template variable: {e}) # 使用生成的 prompt 调用 LLM return {generated_prompt: prompt, final_response: Simulated LLM output based on prompt.}调用示例POST /generate-from-template/with JSON body{template_name: code_review, variables: {language: python, code: def foo(): pass, focus_areas: 可读性和异常处理}}。5.3 处理常见 Prompt 相关问题从搜索热词看大家常遇到context overflow上下文溢出和prompt injection提示注入问题。上下文溢出当对话历史太长超过模型的最大上下文长度时需要实现一个“剪裁”策略。简单的做法是保留最近的 N 条消息或者优先保留system提示和最近的消息。def trim_conversation_history(messages: List[dict], max_tokens_estimate: int 3000): # 这是一个简化的示例实际需要根据 token 数精确计算 if len(messages) 5: # 假设5条以内不会超长 return messages # 保留 system 消息和最新的4条对话 system_msg [m for m in messages if m[role] system] recent_msgs messages[-4:] return system_msg recent_msgs提示注入防御如果允许用户输入部分系统指令就需要防范用户输入覆盖你的系统设定。一个基本策略是严格区分用户可控部分和系统指令并在最终拼接时将系统指令放在一个用户无法修改的字段中而不是让用户输入包含“忽略之前指令”这类文本。在结构化请求中system_prompt由后端控制或严格过滤是更安全的做法。6. 项目实战构建一个简单的 LLM 问答服务我们把前面的知识点串起来构建一个具备基本功能的 LLM 问答后端。项目结构llm_fastapi_project/ ├── .env # 环境变量 ├── .gitignore ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用实例和路由 │ ├── dependencies.py # 依赖项如认证、客户端 │ ├── models.py # Pydantic 数据模型 │ ├── services/ # 业务逻辑 │ │ ├── __init__.py │ │ └── llm_service.py # LLM 调用封装 │ └── utils.py # 工具函数如 prompt 处理 ├── requirements.txt └── README.md核心文件app/main.pyfrom fastapi import FastAPI, Depends from app.dependencies import get_llm_service from app.models import ChatRequest, ChatResponse from app.services.llm_service import LLMService app FastAPI(titleLLM QA Service, descriptionA simple QA backend powered by FastAPI and LLM.) app.post(/ask, response_modelChatResponse) async def ask_question( request: ChatRequest, llm_service: LLMService Depends(get_llm_service) ): 核心问答接口。 1. 接收用户问题。 2. 可选检索上下文RAG。 3. 构造 prompt。 4. 调用 LLM。 5. 返回响应。 # 这里可以加入上下文检索逻辑RAG # context retrieve_context(request.question) # 构造最终 prompt final_prompt llm_service.construct_prompt(request.question, contextNone) # 调用 LLM llm_response await llm_service.generate_response(final_prompt, request.stream) return ChatResponse( answerllm_response[content], usagellm_response.get(usage, {}), streamrequest.stream ) app.get(/health) async def health_check(): return {status: healthy}服务层app/services/llm_service.pyimport os import httpx from typing import Optional, AsyncGenerator from dotenv import load_dotenv load_dotenv() class LLMService: def __init__(self): self.api_key os.getenv(LLM_API_KEY) self.api_base os.getenv(LLM_API_BASE) self.default_model os.getenv(LLM_DEFAULT_MODEL, gpt-3.5-turbo) if not self.api_key: raise ValueError(LLM_API_KEY not set in environment variables.) def construct_prompt(self, question: str, context: Optional[str]) - str: # 简单的 prompt 工程 if context: prompt f基于以下上下文信息回答用户的问题。\n上下文{context}\n\n问题{question}\n回答 else: prompt f请回答以下问题{question} return prompt async def generate_response(self, prompt: str, stream: bool False): headers { Authorization: fBearer {self.api_key}, Content-Type: application/json, } payload { model: self.default_model, messages: [{role: user, content: prompt}], stream: stream, temperature: 0.7, } async with httpx.AsyncClient(timeout60.0) as client: try: if stream: async with client.stream( POST, f{self.api_base}/chat/completions, headersheaders, jsonpayload, ) as response: response.raise_for_status() async for chunk in response.aiter_lines(): # 这里需要解析 SSE 格式的 chunk if chunk.startswith(data: ): yield chunk else: resp await client.post(f{api_base}/chat/completions, headersheaders, jsonpayload) resp.raise_for_status() return resp.json() except httpx.HTTPStatusError as e: # 处理 HTTP 错误如 429 限速 401 鉴权失败 error_detail e.response.json().get(error, {}).get(message, str(e)) raise Exception(fLLM API error: {error_detail}) except httpx.RequestError as e: # 处理网络错误 raise Exception(fNetwork error contacting LLM service: {str(e)})运行与测试配置好.env文件。安装依赖pip install -r requirements.txt(内容为fastapi,uvicorn,httpx,python-dotenv,pydantic)。启动服务uvicorn app.main:app --reload --host 0.0.0.0 --port 8000。打开http://127.0.0.1:8000/docs在/ask接口的交互界面中测试。使用 curl 测试流式接口curl -N -X POST http://127.0.0.1:8000/ask \ -H Content-Type: application/json \ -d {question: Python中staticmethod装饰器有什么用, stream: true}7. 部署与生产环境考量本地开发跑通后如果要对外服务需要考虑以下几点ASGI 服务器与性能开发时用的uvicorn --reload不适合生产。生产环境通常使用uvicorn配合多进程--workers或搭配gunicorn一个 WSGI/ASGI 服务器管理器。# 使用 gunicorn 管理多个 uvicorn worker 进程 gunicorn -w 4 -k uvicorn.workers.UvicornWorker app.main:app --bind 0.0.0.0:8000-w 4表示启动 4 个 worker 进程。worker 数量通常设置为 CPU 核心数的 1-2 倍。搜索热词中有人问fastapi默认多少线程需要澄清FastAPI 本身不管理线程它运行在 ASGI 服务器如 uvicorn上。Uvicorn 默认是单进程单线程异步通过异步 IO 处理并发。使用gunicorn后才是多进程模型。环境变量与配置生产环境绝不能使用.env文件。应使用 Docker 的-e参数、Kubernetes 的 ConfigMap/Secret、或云服务商的环境变量管理功能。日志记录FastAPI 使用标准的 Python logging。配置日志记录请求、响应和错误便于排查问题。import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) app.post(/ask) async def ask_question(request: ChatRequest): logger.info(fReceived question: {request.question[:50]}...) # ... 业务逻辑 logger.info(fQuestion answered successfully.) return response超时与重试在调用外部 LLM API 时网络不稳定可能导致超时。在生产代码中考虑为httpx.AsyncClient增加重试机制可以使用tenacity库并设置合理的超时时间。速率限制与队列如果用户量较大直接无限制地转发请求到 LLM API 可能导致对方速率限制Rate Limit。需要在 FastAPI 层实现简单的速率限制如使用slowapi库或任务队列如使用Celery或RQ将请求排队处理。容器化部署Docker这是最通用的部署方式。一个简单的DockerfileFROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [gunicorn, -w, 4, -k, uvicorn.workers.UvicornWorker, app.main:app, --bind, 0.0.0.0:8000]构建并运行docker build -t llm-fastapi .和docker run -p 8000:8000 --env-file .env llm-fastapi。8. 常见问题排查清单当你的 FastAPI LLM 项目跑不起来或者行为异常时按这个顺序排查服务根本起不来uvicorn报错检查 Python 和包版本python --version,pip list | grep fastapi。确保版本兼容。检查导入路径uvicorn main:app中的main是文件名不含.pyapp是文件内的FastAPI()实例变量名。检查端口占用默认 8000 端口是否被其他程序占用可以换端口--port 8001。API 请求返回 422 Unprocessable Entity这是最常见的问题之一。几乎总是因为请求体的 JSON 格式不符合你定义的 Pydantic 模型。打开/docs页面直接用 Swagger UI 的“Try it out”功能测试它会自动生成符合格式的请求。检查字段名和类型是否拼写错误prompt写成了promtmax_tokens传了字符串100而不是数字100使用curl或 Postman 时确保Content-Type: application/json头已设置且 JSON 格式正确。调用 LLM API 失败返回 4xx/5xx 错误检查 API 密钥和环境变量print(os.getenv(LLM_API_KEY))看看是否成功加载。生产环境确保变量已注入。检查 API 基础地址如果是自托管模型如使用vllm地址可能是http://localhost:8000/v1。查看 LLM 服务方的错误信息日志或返回的 JSON 中通常有error字段会明确告知是额度不足、模型不存在还是参数错误。流式响应不工作或中断前端是否正确处理 SSE后端返回StreamingResponse前端需要使用EventSource或fetch以流式方式读取。检查超时设置服务器、反向代理如 Nginx和客户端都可能有关闭空闲连接的超时设置。需要适当调大。网络稳定性不稳定的网络连接可能导致流中断。性能差响应慢定位瓶颈使用中间件记录每个请求的处理时间。慢是在 FastAPI 逻辑部分还是在调用 LLM API 的网络等待部分LLM API 本身慢考虑使用更快的模型、调整参数如降低max_tokens、或实现客户端缓存对相同问题缓存答案。数据库或上下文检索慢如果你集成了 RAG检查向量数据库的检索速度。并发请求下内存或 CPU 占用高调整 worker 数量gunicorn -w参数不要设置得远超 CPU 核心数。检查内存泄漏长时间运行后内存是否持续增长可能是全局变量缓存了过多数据或者 HTTP 客户端未正确复用/关闭。使用异步客户端确保使用httpx.AsyncClient并在依赖中正确管理其生命周期如使用yield。这个学习路径的核心是“快速验证逐步深化”。不要试图第一天就搭建一个完美的生产系统。先用 FastAPI 把最简单的“接收 prompt - 调用 LLM - 返回结果”的链路跑通然后逐步加入错误处理、流式响应、结构化 prompt、认证、日志和部署。每加一个功能都确保之前的还能工作。这样三天时间足够你从一个完全的新手到一个能搭建出可用 LLM 服务后端的状态。