FastAPI 项目实战：LLM 流式对话接口优化与简历投递功能实现

1. 项目背景与需求分析在当前的 AI 应用开发中大语言模型LLM的集成已成为标配功能。本文基于一个真实的 FastAPI 项目分享两个核心功能的实现与优化LLM 流式对话接口优化从单轮对话升级为多轮对话并改进错误处理机制简历投递详情接口为求职者提供详细的投递记录查询功能2. 代码变更概览2.1 文件结构变更37 -28 app/apis/llm/case1_api.py # LLM 流式对话接口优化 1 -1 app/apis/llm/case2_api.py # API 端点基础 URL 更新 16 -3 app/schemas/llm_case1.py # 新增多轮对话数据结构2.2 核心优化点LLM 接口升级单轮对话 → 多轮对话错误处理优化避免 SSE 流异常导致前端无响应API 端点统一标准化阿里云 DashScope API 调用新增简历投递详情接口完善求职功能3. LLM 流式对话接口优化3.1 原单轮对话接口llm_day01_router.post(/case1,summaryLLM-DAY01-CASE1)asyncdefcase1_api(llmCase1Request:LLMCase1):clientOpenAI(api_keyos.getenv(DASHSCOPE_API_KEY),base_urlhttps://ws-d765zw587c5lpqzq.cn-beijing.maas.aliyuncs.com/compatible-mode/v1,)promptf #角色设定{llmCase1Request.role}#用户问题{llmCase1Request.question}completionclient.chat.completions.create(modelqwen-plus,messages[{role:system,content:你是一个智能助手},{role:user,content:prompt},],streamTrue,stream_options{include_usage:True})return{code:1,message:success,data:completion}3.2 优化后的多轮对话接口llm_day01_router.post(/case2,summary多轮对话流式输出)asyncdefcase2_api(req:LLMMultiChat):msgs[{role:m.role,content:m.content}forminreq.messages]returnStreamingResponse(contentstream_chunk(msgs),media_typetext/event-stream)3.3 改进的流式生成器defstream_chunk(messages:list):多轮流式生成器接收完整对话历史system 历史轮次 当前问题逐块 yield SSE data 异常处理OpenAI SDK 抛出的异常如 AuthenticationError、RateLimitError 不会再导致 StreamingResponse 返回 500 前端 SSE 响应体为空。 改为把错误信息作为特殊 SSE 块 yield 给前端前端会展示明确错误。 clientOpenAI(api_keyos.getenv(DASHSCOPE_API_KEY),base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1,)try:completionclient.chat.completions.create(modelqwen-plus,messagesmessages,streamTrue,stream_options{include_usage:True})forchunkincompletion:ifchunk.choices:choicechunk.choices[0]ifchoice.delta:deltachoice.deltaifdelta.content:yieldfdata:{delta.content}\n\nyielddata: [done]\n\nexceptExceptionase:# 把异常信息透传给前端避免 SSE 流截断导致前端看不到具体原因err_typetype(e).__name__ err_msgstr(e).replace(\n, ).strip()[:500]yieldfdata: [ERROR]{err_type}:{err_msg}\n\nyielddata: [done]\n\n3.4 关键优化点解析3.4.1 多轮对话支持原接口仅支持单轮问答每次请求都是独立的对话新接口支持完整的对话历史可以维护上下文连贯性实现方式通过LLMMultiChat数据结构传递完整的消息列表3.4.2 异常处理优化问题原实现中OpenAI SDK 异常会导致 StreamingResponse 返回 500 状态码前端 SSE 连接中断解决方案使用 try-except 包裹流式生成逻辑将异常信息通过 SSE 通道透传给前端优势前端可以正常接收错误信息并展示给用户而不是连接突然中断3.4.3 API 端点标准化# 优化前base_urlhttps://ws-d765zw587c5lpqzq.cn-beijing.maas.aliyuncs.com/compatible-mode/v1# 优化后base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1统一使用阿里云 DashScope 的标准兼容模式端点提高代码的可维护性和可移植性4. 数据结构定义优化4.1 新增多轮对话数据结构fromtypingimportListfrompydanticimportBaseModel,FieldclassLLMCase1(BaseModel):question:strField(...,title问题,description问题)classLLMCase2(BaseModel):user_id:intField(...,title用户ID,description用户ID)session_id:strField(...,title会话ID,description会话ID)message:strField(...,title用户消息,description用户消息)# 新增多轮对话数据结构classLLMMultiChat(BaseModel):messages:List[dict]Field(...,title对话消息列表,description包含角色和内容的完整对话历史)4.2 数据结构设计思路LLMCase1保持向后兼容用于简单的单轮问答场景LLMCase2支持用户会话管理适合需要维护对话状态的场景LLMMultiChat全新的多轮对话支持可以传递完整的对话上下文5. 简历投递详情接口实现5.1 新增接口代码job_router.get(/resume_submission_detail/{id},summary简历投递详情)asyncdefresume_submission_detail(id:int):resawaitJobService.resume_submission_detail(id)return{code:1,message:ok,data:json.loads(res)}5.2 接口设计要点RESTful 设计使用GET /resume_submission_detail/{id}路径参数验证通过路径参数id接收投递记录 ID服务层调用委托给JobService.resume_submission_detail处理业务逻辑数据格式化使用json.loads()确保返回标准化的 JSON 数据5.3 与原接口的对比# 原接口查看我的投递记录列表job_router.get(/get_my_submissions,summary查看我的投递记录)asyncdefget_my_submissions():求职者查看自己的投递记录需候选人Token鉴权resawaitJobService.getMySubmissions(job_seeker_id,page,page_size)return{code:1,message:查询成功,data:res}# 新接口查看单条投递记录详情job_router.get(/resume_submission_detail/{id},summary简历投递详情)asyncdefresume_submission_detail(id:int):resawaitJobService.resume_submission_detail(id)return{code:1,message:ok,data:json.loads(res)}6. 技术实现细节6.1 SSEServer-Sent Events流式传输# 核心实现defstream_chunk(messages:list):# ... 流式生成逻辑 ...forchunkincompletion:ifchunk.choices:choicechunk.choices[0]ifchoice.delta:deltachoice.deltaifdelta.content:yieldfdata:{delta.content}\n\n# SSE 格式yielddata: [done]\n\n# 结束标记SSE 格式要求每行以data:开头每段数据以两个换行符\n\n结束特殊标记[done]表示流结束错误信息格式data: [ERROR] {错误类型}: {错误信息}6.2 异常处理机制try:# 正常的流式生成逻辑completionclient.chat.completions.create(...)# ... 处理正常流 ...exceptExceptionase:# 异常处理将错误信息通过 SSE 通道返回err_typetype(e).__name__ err_msgstr(e).replace(\n, ).strip()[:500]# 限制长度避免过大yieldfdata: [ERROR]{err_type}:{err_msg}\n\nyielddata: [done]\n\n异常类型处理AuthenticationErrorAPI 密钥错误RateLimitError请求频率超限APIConnectionError网络连接问题其他未知异常6.3 依赖注入设计fromfastapiimportDepends,APIRouter,Queryfromapp.core.dependsimportget_job_info,get_job_seeker_info,get_enterprise_info# 依赖注入示例asyncdefget_my_submissions(job_seeker_id:intDepends(get_job_seeker_info),page:intQuery(1,ge1),page_size:intQuery(10,ge1,le100)):# 业务逻辑pass7. 前端集成建议7.1 SSE 客户端实现// 前端 SSE 客户端示例asyncfunctionstreamLLMResponse(messages){consteventSourcenewEventSource(/api/llm-day01/case2);eventSource.onmessage(event){constdataevent.data;if(data[done]){eventSource.close();console.log(流式传输完成);}elseif(data.startsWith([ERROR])){eventSource.close();consterrorMsgdata.substring(8);// 移除 [ERROR] console.error(流式传输错误:,errorMsg);// 显示错误信息给用户}else{// 正常的数据块追加到界面appendToChat(data);}};eventSource.onerror(error){console.error(SSE 连接错误:,error);eventSource.close();};}7.2 错误处理优化// 改进的错误处理functionhandleSSEError(errorData){consterrorPrefix[ERROR] ;if(errorData.startsWith(errorPrefix)){consterrorInfoerrorData.substring(errorPrefix.length);const[errorType,...errorMsgParts]errorInfo.split(: );consterrorMessageerrorMsgParts.join(: );// 根据错误类型提供不同的用户提示switch(errorType){caseAuthenticationError:showToast(API 密钥错误请检查配置);break;caseRateLimitError:showToast(请求频率超限请稍后重试);break;default:showToast(系统错误:${errorMessage});}returntrue;// 已处理错误}returnfalse;// 不是错误信息}8. 部署与配置8.1 环境变量配置# .env 文件配置DASHSCOPE_API_KEYyour_dashscope_api_key_hereBASE_URLhttps://dashscope.aliyuncs.com/compatible-mode/v1MODEL_NAMEqwen-plus8.2 Docker 部署配置# Dockerfile 示例 FROM python:3.9-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, 8000]8.3 Nginx 配置SSE 支持# nginx.conf 片段 server { listen 80; server_name your-domain.com; location /api/ { proxy_pass http://backend:8000; proxy_http_version 1.1; proxy_set_header Connection ; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; # SSE 相关配置 proxy_buffering off; proxy_cache off; proxy_read_timeout 86400s; proxy_send_timeout 86400s; } }9. 性能优化建议9.1 连接池管理# 使用连接池管理 OpenAI 客户端importhttpxfromopenaiimportOpenAIclassLLMClientPool:def__init__(self):self._clients{}defget_client(self,api_key:str)-OpenAI:ifapi_keynotinself._clients:http_clienthttpx.AsyncClient(limitshttpx.Limits(max_connections100,max_keepalive_connections20),timeouthttpx.Timeout(30.0))self._clients[api_key]OpenAI(api_keyapi_key,base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1,http_clienthttp_client)returnself._clients[api_key]9.2 流式响应优化# 添加响应头优化llm_day01_router.post(/case2,summary多轮对话流式输出)asyncdefcase2_api(req:LLMMultiChat):msgs[{role:m.role,content:m.content}forminreq.messages]returnStreamingResponse(contentstream_chunk(msgs),media_typetext/event-stream,headers{Cache-Control:no-cache,X-Accel-Buffering:no,# 禁用 Nginx 缓冲Connection:keep-alive})10. 总结与展望10.1 本次优化的核心价值用户体验提升多轮对话支持让 AI 交互更加自然连贯错误处理完善SSE 流的异常处理机制提高了系统稳定性代码可维护性统一 API 端点和标准化的数据结构设计功能完整性简历投递详情接口完善了求职功能模块10.2 未来优化方向对话状态管理引入 Redis 缓存对话历史支持长期会话流控与限流基于用户或 IP 的请求频率限制监控与日志详细的性能监控和错误日志记录多模型支持扩展支持其他 LLM 提供商OpenAI、Claude 等10.3 最佳实践建议始终使用 try-except包裹外部 API 调用为 SSE 流设置合理的超时时间在前端实现优雅的重连机制定期更新 API 客户端库以获取最新的功能和安全修复通过本次代码优化我们不仅提升了系统的稳定性和用户体验还为未来的功能扩展奠定了良好的架构基础。这种渐进式的优化方式值得在大型项目中推广。