使用 FastAPI 封装统一的大模型调用服务系列Python 大模型应用开发第 2 篇目标把大模型调用代码封装成统一的 HTTP 服务为网页、小程序、企业微信侧边栏和业务系统提供后端接口。1. 为什么要增加一层后端服务上一篇中Python 程序直接调用了模型服务。如果未来需要接入网页、CRM、小程序或企业微信一种危险做法是让每个客户端都直接携带模型 API Key。更合理的基础架构是网页 / 小程序 / 企业微信 / CRM ↓ 自己的 FastAPI 服务 ↓ 鉴权、参数校验、日志、限流 ↓ 大模型服务商增加 FastAPI 层可以解决以下问题模型密钥只保存在后端多个业务系统使用统一接口集中完成输入校验、异常转换和日志记录后续可以统一增加鉴权、限流、缓存和成本统计更换模型服务商时前端不需要跟着修改。本文使用异步 HTTP 客户端httpx.AsyncClient调用模型避免在 FastAPI 的异步接口中使用同步网络请求阻塞事件循环。2. 目标接口我们准备提供两个接口方法路径用途GET/health判断当前 FastAPI 进程是否正常运行POST/api/v1/chat接收用户问题并返回模型回答聊天请求{user_message:请解释 Python 装饰器。,temperature:0.2}聊天响应{content:模型生成的回答,model:your-model-id,usage:{prompt_tokens:20,completion_tokens:80,total_tokens:100}}usage是否存在、包含哪些字段由实际模型服务决定因此代码必须允许它为null。3. 创建项目本文使用 Python 3.10 及以上版本。项目结构llm_fastapi_service/ ├── app/ │ ├── __init__.py │ ├── config.py │ ├── schemas.py │ ├── llm_client.py │ └── main.py └── requirements.txt创建虚拟环境python-m venv.venv.\.venv\Scripts\python.exe-m pip install--upgrade piprequirements.txtfastapi0.115,1 uvicorn[standard]0.30,1 httpx0.27,1 pydantic2.7,3安装依赖.\.venv\Scripts\python.exe-m pip install-r requirements.txtapp/__init__.py可以是空文件它表示app是一个 Python 包。4. 编写配置模块新建app/config.pyimportosfromdataclassesimportdataclassdataclass(frozenTrue)classSettings:保存大模型服务配置。api_key:strbase_url:strmodel:strdefload_settings()-Settings:读取环境变量并尽早发现缺失配置。# 所有敏感配置都从运行环境读取不写入代码仓库api_keyos.getenv(LLM_API_KEY,).strip()base_urlos.getenv(LLM_BASE_URL,).strip().rstrip(/)modelos.getenv(LLM_MODEL,).strip()missing_variables[]ifnotapi_key:missing_variables.append(LLM_API_KEY)ifnotbase_url:missing_variables.append(LLM_BASE_URL)ifnotmodel:missing_variables.append(LLM_MODEL)ifmissing_variables:names, .join(missing_variables)raiseRuntimeError(f缺少环境变量{names})# 远程传输密钥时必须使用 HTTPSifnotbase_url.startswith(https://):raiseRuntimeError(LLM_BASE_URL 必须使用 https:// 地址)returnSettings(api_keyapi_key,base_urlbase_url,modelmodel,)设置环境变量$env:LLM_API_KEY 替换为真实密钥$env:LLM_BASE_URL https://替换为模型服务地址/v1$env:LLM_MODEL 替换为真实模型标识这些示例值不能直接使用必须根据实际服务商文档填写。5. 使用 Pydantic 定义数据结构新建app/schemas.pyfromtypingimportAnyfrompydanticimportBaseModel,Field,field_validatorclassChatRequest(BaseModel):调用聊天接口时客户端允许提交的数据。user_message:strField(min_length1,max_length4000,description用户问题长度为 1 到 4000 个字符,)temperature:floatField(default0.2,ge0,le2,description常见的模型随机性参数真实范围以模型文档为准,)field_validator(user_message)classmethoddefmessage_must_not_be_blank(cls,value:str)-str:阻止只包含空格、换行符的无效问题。cleaned_valuevalue.strip()ifnotcleaned_value:raiseValueError(user_message 不能为空白字符串)# 返回清理后的内容后续业务代码无需重复 strip()returncleaned_valueclassChatResponse(BaseModel):聊天接口成功时的统一响应。content:strmodel:str# 不同模型服务商返回的 usage 字段可能不同因此用 Any 表示值类型usage:dict[str,Any]|NoneNoneclassHealthResponse(BaseModel):健康检查接口响应。status:str为什么不让调用者提交system_message因为系统提示词属于服务端规则。如果任意前端用户都能修改它就可能把“严谨的业务助手”改成其他身份绕过原有业务约束。本文把系统提示词固定在后端后续如果存在多个业务助手可以使用服务端维护的assistant_id白名单进行选择。6. 编写异步模型客户端新建app/llm_client.pyfromtypingimportAnyimporthttpxfromapp.configimportSettingsclassLLMServiceError(RuntimeError):表示调用上游模型服务时发生的可预期错误。def__init__(self,message:str,http_status:int502)-None:super().__init__(message)# http_status 是自己的 FastAPI 服务要返回给调用方的状态码self.http_statushttp_statusclassAsyncLLMClient:基于 httpx.AsyncClient 的异步模型客户端。def__init__(self,settings:Settings)-None:self.settingssettings# 60 秒是默认总超时建立连接最多等待 5 秒timeouthttpx.Timeout(60.0,connect5.0)# AsyncClient 应在应用生命周期内复用不能每次请求都重新创建self.http_clienthttpx.AsyncClient(timeouttimeout)asyncdefchat(self,user_message:str,system_message:str,temperature:float,)-tuple[str,dict[str,Any]|None]:异步调用模型返回回答文本和可选的 Token 用量。request_urlf{self.settings.base_url}/chat/completionsrequest_headers{Authorization:fBearer{self.settings.api_key},Content-Type:application/json,}request_body{model:self.settings.model,messages:[{role:system,content:system_message},{role:user,content:user_message},],temperature:temperature,}try:# await 表示当前协程等待网络结果时可以让出执行权responseawaitself.http_client.post(request_url,headersrequest_headers,jsonrequest_body,)excepthttpx.TimeoutExceptionasexc:# 504 表示作为网关等待上游服务超时raiseLLMServiceError(等待模型服务响应超时,http_status504,)fromexcexcepthttpx.ConnectErrorasexc:raiseLLMServiceError(无法连接模型服务,http_status502,)fromexcexcepthttpx.HTTPErrorasexc:raiseLLMServiceError(调用模型服务时发生网络异常,http_status502,)fromexc# 上游鉴权失败属于后端配置问题不向前端暴露密钥细节ifresponse.status_code401:raiseLLMServiceError(模型服务鉴权失败,http_status502)# 自己的服务当前无法满足请求统一转换成 503ifresponse.status_code429:raiseLLMServiceError(模型服务繁忙或额度受限请稍后重试,http_status503,)if400response.status_code500:raiseLLMServiceError(f模型服务拒绝了请求上游状态码{response.status_code},http_status502,)ifresponse.status_code500:raiseLLMServiceError(模型服务暂时不可用,http_status503,)try:# httpx 的 response.json() 会把 JSON 转成 Python 对象data:dict[str,Any]response.json()exceptValueErrorasexc:raiseLLMServiceError(模型服务返回了无效 JSON)fromexctry:contentdata[choices][0][message][content]except(KeyError,IndexError,TypeError)asexc:raiseLLMServiceError(模型响应缺少预期字段)fromexcifnotisinstance(content,str)ornotcontent.strip():raiseLLMServiceError(模型返回了空内容)# usage 不是所有服务都提供因此使用 get() 安全读取raw_usagedata.get(usage)usageraw_usageifisinstance(raw_usage,dict)elseNonereturncontent.strip(),usageasyncdefclose(self)-None:关闭异步客户端释放连接池资源。awaitself.http_client.aclose()为什么 FastAPI 中使用异步客户端模型请求的大部分时间都消耗在网络等待上。如果异步接口内部使用阻塞式网络请求事件循环会被阻塞其他请求也可能受到影响。这里的关键不是把函数名称前面简单加上async而是内部网络库本身也要支持异步并且在调用时使用await。7. 创建 FastAPI 应用新建app/main.pyfromcollections.abcimportAsyncIteratorfromcontextlibimportasynccontextmanagerfromfastapiimportFastAPI,Requestfromfastapi.responsesimportJSONResponsefromapp.configimportSettings,load_settingsfromapp.llm_clientimportAsyncLLMClient,LLMServiceErrorfromapp.schemasimportChatRequest,ChatResponse,HealthResponseasynccontextmanagerasyncdeflifespan(app:FastAPI)-AsyncIterator[None]:管理应用启动和关闭时需要创建、释放的资源。# 启动阶段读取配置配置有误时服务会直接启动失败settingsload_settings()# 创建一个供整个应用复用的异步模型客户端llm_clientAsyncLLMClient(settings)# 将对象放入 app.state路由函数可以通过 Request 取得它们app.state.settingssettings app.state.llm_clientllm_client# yield 之前是启动逻辑yield 之后是关闭逻辑yield# 服务关闭时释放 HTTP 连接池awaitllm_client.close()appFastAPI(title统一大模型调用服务,version1.0.0,lifespanlifespan,)app.exception_handler(LLMServiceError)asyncdefhandle_llm_service_error(request:Request,exc:LLMServiceError,)-JSONResponse:把内部模型异常转换成统一且可理解的 HTTP 响应。# request 在当前示例中未读取后续可用它取得请求 ID 或用户身份returnJSONResponse(status_codeexc.http_status,content{detail:str(exc)},)app.get(/health,response_modelHealthResponse)asyncdefhealth()-HealthResponse:判断当前 FastAPI 进程是否正常运行。# 这里只验证自己的服务进程不代表上游模型一定可用returnHealthResponse(statusrunning)app.post(/api/v1/chat,response_modelChatResponse)asyncdefchat(chat_request:ChatRequest,request:Request,)-ChatResponse:接收用户问题通过统一模型客户端生成回答。# 系统提示词由服务端控制调用者不能通过请求任意覆盖system_message(你是一名严谨的 Python 教师。请基于事实回答不确定时明确说明不要编造。)# 从应用状态中取得启动时创建的共享对象llm_client:AsyncLLMClientrequest.app.state.llm_client settings:Settingsrequest.app.state.settings# 等待异步模型调用完成content,usageawaitllm_client.chat(user_messagechat_request.user_message,system_messagesystem_message,temperaturechat_request.temperature,)# response_model 会再次校验接口输出结构returnChatResponse(contentcontent,modelsettings.model,usageusage,)8. 启动并测试服务在项目根目录执行.\.venv\Scripts\python.exe-m uvicorn app.main:app--reload--reload会在代码变化后自动重启只适合本地开发不应直接作为生产环境启动方式。浏览器访问http://127.0.0.1:8000/docsFastAPI 会生成交互式 API 文档可以直接测试/health和/api/v1/chat。也可以使用 PowerShell 调用聊天接口# 使用哈希表构造请求体再转换为 JSON$body {user_message 请用三个要点解释 Python 列表和元组的区别temperature 0.2}|ConvertTo-Json# 调用自己的 FastAPI 服务而不是让前端直接调用模型厂商Invoke-RestMethod-Method Post -Urihttp://127.0.0.1:8000/api/v1/chat-ContentTypeapplication/json-Body$body9. FastAPI 自动完成了哪些工作当请求到达/api/v1/chat时FastAPI 和 Pydantic 会读取 JSON 请求体检查user_message是否存在检查字符串长度拦截只有空格的输入检查temperature是否在规定范围内把合法数据转换成ChatRequest对象使用ChatResponse校验响应结构自动生成 OpenAPI 接口文档。例如请求中把temperature设置为 5FastAPI 会直接返回 422而不会继续调用模型并产生费用。10. 为什么不能每个请求都创建 AsyncClient下面的写法虽然可能运行但不适合高频调用# 不推荐每个业务请求都创建和关闭新的连接池asyncwithhttpx.AsyncClient()asclient:responseawaitclient.post(模型地址)本文通过lifespan在应用启动时创建一次AsyncClient在关闭服务时统一释放。这样可以复用连接池职责也更加清晰。11. 对抗性审查当前服务还有哪些风险1. 自己的接口还没有身份认证当前示例用于本地学习任何能够访问该端口的人都可以调用接口并消耗模型额度。生产环境至少需要用户身份认证、权限校验和调用额度控制。2. 健康检查不代表模型可用/health只说明 FastAPI 进程在运行。若要检查模型服务是否可用应单独设计 Readiness就绪检查并谨慎控制检查频率避免不断产生模型费用。3. 错误信息不能泄露敏感数据不应直接把上游完整响应、请求头、API Key 或内部堆栈返回给前端。详细异常可以写入经过脱敏的内部日志对外只返回必要信息。4. 输入长度限制不等于成本控制字符数与 Token 数不是完全相同的概念。生产服务还需要根据实际模型的计费和上下文限制统计 Token并设置用户级额度。5. 系统提示词不是安全边界把系统提示词保存在后端可以减少被直接修改的风险但不能只依赖一句 Prompt 实现权限控制。数据库查询、文件访问和外部操作必须在代码层执行身份认证和权限判断。6. 模型回答仍然可能错误HTTP 200 只代表服务正常处理了请求不代表回答符合事实。高风险业务需要增加 RAG、规则校验、结构化输出和人工审核。12. 下一步如何扩展这个统一服务可以继续增加API 用户鉴权请求 ID 和结构化日志429、5xx 的有限重试Redis 限流和缓存Token 用量与成本统计流式输出 SSE服务器发送事件多轮对话和历史消息存储多模型适配器RAG 企业知识库Prompt 和模型效果评测。13. 总结本文完成了从“Python 直接调用模型”到“统一后端模型服务”的升级客户端 ↓ FastAPI 参数校验 ↓ 服务端系统规则 ↓ 异步模型客户端 ↓ 模型服务 ↓ 统一异常与响应结构真正有价值的不是多包装了一层接口而是建立了明确的系统边界密钥属于后端、规则由服务端控制、外部输入必须校验、上游错误需要转换、网络资源需要统一管理。14. 练习题将user_message最大长度改为 2000并测试超长请求将temperature设置为 3观察 FastAPI 返回的 422删除一个必需环境变量观察服务如何在启动阶段失败为/health增加当前服务版本字段为成功响应增加一个由后端生成的request_id思考如果接入三个模型厂商怎样避免在路由函数中编写大量if/else
)
