大模型项目： 学习FastAPI 服务器开发

FastAPI 服务器开发之前的课程学习了 RAG 流程和向量数据库。今天进入Web 服务器开发领域——学习 FastAPI 框架掌握接口定义、参数接收、子路由嵌套、流式输出等核心技能将 LLM 能力封装为可外部调用的 API 接口。一、服务器基础概念1. 什么是服务器服务器在我们的生活中无处不在如果需要下载客户端再使用 →C/S 模式客户端/服务器模式如果不需要下载客户端就可以使用 →B/S 模式浏览器/服务器模式我们做的Web 应用开发基于B/S 模式实现——通过构造网页来实现项目中的内容和功能展示。服务器开发的重点只有通过服务器我们才可以操作数据库中的内容注意服务器开发时的分包思想结构化组织代码注意服务器开发中的术语——一个内容可以有不同称呼如接口、端点、路由2. Python 服务器开发框架选型框架说明Django重量级全栈框架适合大型项目Flask轻量级微框架适合中小项目FastAPI现代高性能框架异步原生自动生成 API 文档——本项目选型FastAPI 官网https://fastapi.tiangolo.com/zh/3. 接口术语接口服务器中提供给客户端实现某个功能访问的函数。和普通 Python 函数定义没有区别只是多了一个请求路径配置告诉客户端通过什么请求地址才能访问到这个函数接口函数不能像普通函数一样直接调用需要通过请求地址去访问常用接口测试工具Apipost、PostmanFastAPI 内置集成了Swagger UI直接通过xxx/docs地址即可在浏览器中测试所有接口二、FastAPI 环境搭建1. 安装激活虚拟环境后执行pipinstallfastapiuvicorn[standard]-ihttps://repo.huaweicloud.com/repository/pypi/simple/项目类型选择上非普通 Python 项目需创建为 FastAPI 项目类型。2. 启动方式方式一通过 IDE调整内置参数后点击运行按钮直接启动启动参数说明main:appmain 文件名main.pyapp FastAPI 实例变量名--reload热重载模式代码修改后自动重启开发时启用--host监听地址0.0.0.0允许所有 IP 访问--port监听端口默认 8000方式二通过命令行启动if __name__ __main__: import uvicorn as uv uv.run( appmain:app, hostlocalhost, port8000, reloadFalse, )值效果reloadTrue开发模式 — 代码文件一保存修改服务器自动重启reloadFalse生产模式 — 改完代码必须手动停服务再重启才生效三、接口定义1. 接口定义格式接口定义分为两种情况第一种直接在main.py中定义接口项目不使用仅用于演示第二种在其他文件中定义接口项目中使用的方案2. 请求方式请求方式使用场景参数传递方式GET只查数据、下拉列表、详情页浏览器地址栏直接访问用于获取数据。它应该是安全的只读且幂等的多次请求结果一致不会改变服务器状态。没有请求体Body。参数拼接在请求地址后面表单格式 kvPOST登录注册、上传文件、新增/修改/删除数据库记录用于创建或提交数据。它既不安全也不幂等多次提交可能会创建多个资源比如重复下单。拥有请求体Body。参数包装在请求体里面JSON 格式PUT更新操作参数在请求体中JSON 格式DELETE删除操作参数通常在 URL 路径中3. 返回值接口返回值统一以 JSON 格式返回不需要手动调用json.dumps()转换把返回值设为字典即可自动完成 JSON 序列化4. 命名规范项目规范接口函数名蛇形命名snake_casesay_hello请求路径小驼峰命名/sayHello形参小驼峰命名userName5. main.py 中直接定义接口演示 定义一个 GET 请求的接口假设需要返回 msg: hello 给客户端 1、先直接定义一个函数 2、通过装饰器配置访问路径和请求方式 3、设置函数内容逻辑处理、返回值等 fromfastapiimportFastAPI appFastAPI()app.get(/say)defsay():print(say 接口函数执行了)# 设置返回值 --- 字典自动转 JSONreturn{msg:hello}app.post(/say2)defsay2():print(say2 函数执行了)return{msg:hello2}关键点app.get()/app.post()通过装饰器将普通函数变为接口函数app是 FastAPI 实例get/post设置请求方式请求路径如/say需要拼接在服务器地址http://localhost:8000后面返回值直接写字典即可FastAPI 自动转为 JSON四、MVC 分包思想【重要】按照不同的项目模块和功能代码进行分包处理降低代码的耦合度使得代码分层清晰、便于测试和维护。1. 标准分包结构项目根目录 ├── users用户模块 │ ├── controller/ --- 定义接口接收和响应客户端请求 │ ├── service/ --- 业务逻辑处理供 controller 调用 │ ├── dao/ --- 数据库操作层只操作数据库不做逻辑处理 │ ├── utils/ --- 当前模块的工具函数 │ └── entity/ --- 实体类数据验证、接收 JSON ├── chat对话模块 │ ├── controller/ │ ├── service/ │ ├── dao/ │ ├── utils/ │ └── entity/ ├── common公共模块 --- 多个模块共用的工具代码 └── aiAI 模块 --- 大模型相关封装核心原则同一模块下不同业务创建不同文件来实现不需要创建类直接定义函数接口比如 chat 模块既有聊天业务、也有加载历史对话记录业务 → 创建两套文件分别处理2. 各层职责层职责核心任务controller接口层定义接口、接收客户端参数、调用 service、返回响应service业务层实现具体业务逻辑处理调用 dao 操作数据dao数据层只负责数据库的增删改查不做逻辑处理entity实体层定义数据模型类用于接收 JSON 参数和数据验证utils工具层抽取冗余代码形成工具函数五、父子路由嵌套【重点】因为采用分包分模块思想接口不在main.py中定义而在各模块的controller包下。但 controller 包中没有 FastAPI 对象无法直接定义接口。解决方案将 controller 中的接口定义为子路由然后在main.py中注册子路由。1. 子路由定义controller 层# users/controller/TestController_1.pyfromfastapiimportAPIRouter# 创建子路由对象users_routerAPIRouter()# 定义子路由接口 --- 配置的路径并非最终接口访问路径users_router.get(/sayHello)defsay_hello():return{msg:hello}关键点APIRouter()创建子路由对象代替 FastAPI 实例装饰器使用users_router.get()而非app.get()子路由中配置的路径不是最终路径需要通过main.py注册后才完整2. 子路由注册main.py# main.pyfromfastapiimportFastAPI# 导入子路由fromusers.controller.TestController_1importusers_router appFastAPI()# 注册子路由 --- 访问路径为/users/sayHelloapp.include_router(users_router,prefix/users tags[users])关键点app.include_router(子路由对象, prefix/模块名)注册子路由prefix/users设置路由前缀最终接口路径 prefix 子路由中配置的路径子路由注册后在 controller 中定义的接口才能被外部访问六、接口接收客户端请求参数【核心】参数传递方式分为三种取决于请求方式和数据格式方式一GET 请求 keyvalue 表单格式参数通过kv格式拼接在请求地址后面接口直接用形参接收形参名必须和 key 一致。 Way 1: GET keyvalue URL: localhost:8001/users/getParams?usernameadminpassword111 形参名必须和 key 相同否则接收不到数据 users_router.get(/getParams)defget_params(username:strNone,password:strNone):print(f接收到的数据为username{username}, password{password})return{code:200,msg:success,data:{username:username,password:password}}关键点直接用函数形参接收形参名必须与 URL 中的 key 一致设置默认值 None使参数可选避免客户端不传时报错客户端访问示例/getParams?usernameadminpassword111方式二GET 请求 参数在请求路径中参数直接写在请求路径里没有 key需要在路径中定义{变量}占位符。 Way 2: GET URL 路径参数 URL: localhost:8001/users/getParamsTwo/admin/111 顺序匹配路径中的占位符 常用于查询、删除操作 users_router.get(/getParamsTwo/{username}/{password})defget_params_two(username:str,password:str):print(f接收到数据为username{username}, password{password})return{code:200,msg:success,data:{username:username,password:password}}关键点路径中使用{变量名}占位客户端按顺序传入值⚠️ 形参名必须和路径中的占位符名字一致否则返回 422 错误有顺序问题/getParamsTwo/admin/111按路径顺序匹配 usernameadmin, password111方式三POST 请求 JSON 格式数据参数在请求体中传输Content-Type: application/json需要定义一个数据类来接收。 Way 3: POST JSON 需要定义类来接收类属性名必须和 JSON 中的 key 一致 客户端 curl -X POST localhost:8001/users/postParams \ -H Content-Type: application/json \ -d {username:admin,password:111} frompydanticimportBaseModel,Field# 定义接收数据的类 --- 直接继承 BaseModelclassTestClass(BaseModel):# Field(..., title用户名) 表示这是一个必填字段username:strField(...,title用户名)password:strField(...,title密码)users_router.post(/postParams)defpost_params(testClass:TestClass):print(f接收到的数据为{testClass})print(testClass.username,testClass.password)return{code:200,msg:success,data:testClass}关键点继承BaseModelPydantic定义数据类属性名必须和 JSON 的 key 一致Field(..., title用户名)中的...表示该字段为必填改为默认值则为可选接口形参直接用类类型接收FastAPI 自动解析 JSON 并验证数据访问示例POST /postParamsBody 为{username:admin,password:111}方式四POST 请求 文件上传文件类型的参数必须用 POST 请求使用UploadFile类型接收文件其他额外参数用Form接收。 Way 4: POST file file 类型数据必须用 POST 请求 文件用 UploadFile 接收额外参数用 Form 接收 fromfastapiimportUploadFile,File,Formusers_router.post(/postFile)defpost_file(file:UploadFileFile(...),username:strForm(...)):print(f接收到的数据为file{file}, \n username{username})# 重新定义文件名字 --- 时间戳唯一标识文件filenamestr(int(time.time())).file.filename.split(.)[-1]# 文件存储地址 文件名字save_pathrD\stu_fastapi\static\upload\\filename# 存储文件wb是w(写入)b(二进制形式)withopen(save_path,wb)asf:# file.file.read() 读取文件内容f.write(file.file.read())return{code:200,msg:success,data:}关键点UploadFileFastAPI 提供的文件类型自动处理上传文件File(...)表示这是一个文件类型的必填参数Form(...)接收文件外的普通表单字段文件名用时间戳重命名防止冲突str(int(time.time())) . 扩展名file.file.read()读取上传文件的内容file.filename获取原始文件名四种传参方式对比方式请求方式数据格式接收方式适用场景方式一GET表单kv直接形参接收查询列表、简单参数传递方式二GETURL 路径参数路径占位符 形参查询/删除单个资源方式三POSTJSONPydantic 数据类新增/登录/复杂参数方式四POSTmultipart/form-dataUploadFile Form文件上传核心原则无论选择什么方式传递数据给服务器一定要满足key 对得上——客户端和服务器通过 key:value 交互数据只能通过 key 找 value。七、流式输出 — StreamingResponse在 FastAPI 中通过StreamingResponse实现流式输出核心是返回一个生成器迭代器对象。fromstarlette.responsesimportStreamingResponseimporttimeimportjson方案一基础流式输出fetch 请求客户端使用 fetch 请求接收服务器直接返回结果服务端代码简单客户端代码较难写。users_router.get(/testStream)deftest_stream():基础流式输出假设模型返回 0-9 十个数字defgenerator():foriinrange(9):yieldf{i}time.sleep(0.1)# 模拟模型逐 token 生成returnStreamingResponse(contentgenerator(),# 迭代器对象media_typetext/event-stream,# 媒体类型)方案二SSE 流式输出标准方案客户端使用SSEServer-Sent Events请求服务器必须将数据包装成data: 内容\n\n格式推荐方案。users_router.get(/testStreamSSE)deftest_stream_sse():SSE 流式输出标准 data: 格式便于客户端处理result你好 很高兴见到你。有什么我可以帮你的吗defgenerator():foriinresult:# 包装为 SSE 标准格式内容转为 JSON 便于客户端解析yieldfdata:{json.dumps({content:i})}\n\ntime.sleep(0.1)# 模拟耗时# 发送结束标记yieldfdata:{json.dumps({content:[DONE]})}\n\nreturnStreamingResponse(contentgenerator(),# 迭代器对象media_typetext/event-stream,# SSE 媒体类型)关键点StreamingResponse的content参数接收一个生成器/迭代器对象函数中使用yield逐次返回数据media_typetext/event-stream指定 SSE 媒体类型告知客户端以流式事件接收方案二SSE数据必须包装成data: 内容\n\n字符串格式否则客户端报错通常将数据转为 JSON 格式返回便于客户端处理需要告诉客户端流式输出何时结束发送一个约定的结束标识符如[DONE]八、综合实战LLM 流式回复接口将前面所学知识点串联——结合 LLM 模型调用实现一个完整的流式对话 API。1. 封装 LLM 加载工具ai/TestLLM.py# ai/TestLLM.pyimportosfromlangchain_openaiimportChatOpenAIdefLLM_Model(question:str):封装 LLM 加载和调用返回流式生成器chatLLMChatOpenAI(api_keyos.getenv(DASHSCOPE_API_KEY),base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1,modelqwen3.7-max-preview,streamingTrue,)messages[{role:user,content:question}]# stream() 返回生成器逐 token 产出forchunkinchatLLM.stream(messages):yieldchunk.content2. 实现流式对话接口TestController_1.py# users/controller/TestController_1.pyimportjsonimporttimefromfastapiimportAPIRouterfromstarlette.responsesimportStreamingResponsefromai.TestLLMimportLLM_Model users_routerAPIRouter()users_router.get(/StreamSSE)deftest_stream_sse(question:str你好): 用户输入问题 → LLM 生成回复 → 流式 SSE 输出给客户端 客户端访问/users/StreamSSE?question你好 print(f接收到的数据为question{question})# 调用 LLM 获取流式生成器resultLLM_Model(questionquestion)# 生成器 --- 包装为 SSE 格式输出defgenerator():foriinresult:yieldfdata:{json.dumps({content:i})}\n\ntime.sleep(0.1)# 模拟网络传输延迟# 数据结束标记yieldfdata:{json.dumps({content:[DONE]})}\n\nreturnStreamingResponse(contentgenerator(),media_typetext/event-stream,)关键点LLM_Model()返回生成器通过yield逐 token 产出的回复内容接口使用 GET 请求 kv 参数方式一接收用户问题将 LLM 的流式输出包装为 SSE 标准格式逐 token 推送给客户端客户端接收完所有数据后通过[DONE]标记判断流是否结束九、完整开发流程总结FastAPI 接口开发完整流程① 环境搭建 pip install fastapi uvicorn[standard] ② 创建项目、分包 users/ controller/ ← 定义接口 service/ ← 业务逻辑 dao/ ← 数据库操作 entity/ ← 数据模型 ③ 在 controller 中定义子路由 router APIRouter() router.get(/path) → 接口函数 ④ 在 main.py 注册子路由 app.include_router(router, prefix/users) ⑤ 启动服务器 uvicorn main:app --reload --host 0.0.0.0 --port 8001 ⑥ 访问 Swagger UI 测试 http://localhost:8001/docs接口定义规范速查请求方式参数传递接收方式示例GETURL 查询参数kv直接形参/getParams?usernameadminGETURL 路径参数路径占位符/getParamsTwo/{id}POSTJSON 请求体Pydantic 数据类{name: 张三}POST文件上传UploadFile Formmultipart/form-dataGET/POST流式输出StreamingResponse 生成器SSE 格式逐 token 推送核心要点接口 普通函数 请求路径配置不能直接调用必须通过 HTTP 请求访问MVC 分包降低耦合度controller → service → dao 三层职责分明子路由是项目开发的标配方案APIRouter()app.include_router()客户端和服务器通过key:value交互形参名必须和 key 一致接口返回值统一字典格式FastAPI 自动转 JSON流式输出使用StreamingResponse 生成器yieldSSE 格式需要data: 内容\n\n包装Swagger UI/docs是 FastAPI 最强大的特性之一无需第三方测试工具FastAPI常见错误码查路径对不对→404路径错了 /405路径对了但 Method 错了。查请求体格式错没错→ JSON 结构坏了给400字段类型错了给422。查登录没→ 没 Token 给401有 Token 但没权限给403。最终思考从RAG 知识库搭建到FastAPI 服务器开发的全栈链路学习也学会了将 LLM 对话功能封装为 API 接口通过浏览器或客户端调用实现完整的 AI 应用服务。