Web 框架FastAPI / Flask核心概念AI 应用开发岗初级面试突击指南目标岗位AI 应用开发工程师初级内容聚焦围绕大模型 API 服务化、流式响应、文件上传、异步推理等高频场景讲解 FastAPI为主与 Flask 的路由、中间件、依赖注入、请求生命周期、流式响应、异步视图、CORS、文件上传与后台任务。每个模块均包含面试常见问题 → 类比理解 → 原理与术语说明 → 详细注释的 AI 场景伪代码 →完整运行结果→ 常见面试题及参考回答。一、路由、中间件与依赖注入【面试问题】“你要用 FastAPI 搭建一个 AI 服务路由怎么划分如何给某些接口加 API Key 校验怎么让每个处理函数自动获取当前用户信息而不重复代码”【类比】餐厅的服务流程路由是菜单上的分类冷菜、热菜、汤每个菜名对应一个后厨操作。中间件是餐厅门口的迎宾员不管顾客去哪桌都要先通过他检查比如量体温、查预约。依赖注入是顾客落座后服务员自动端上来的餐具和茶水——你不用每次喊按规矩就给你备好。【原理与术语】路由 (Route)把 URL 路径和 HTTP 方法映射到具体的处理函数FastAPI 使用装饰器app.get(/path)声明。中间件 (Middleware)在请求进入路由处理函数之前以及响应返回给客户端之前执行公共逻辑如日志、鉴权、跨域。FastAPI 中通过app.middleware(http)定义Flask 中通过app.before_request和app.after_request。依赖注入 (Dependency Injection)FastAPI 的Depends允许把函数声明为“依赖”视图函数需要时自动调用并注入结果常用于提取当前用户、获取数据库会话等实现可复用组件。 术语速查术语解释路由URL 和处理函数的映射关系。比如/chat对应大模型聊天接口。中间件一个“切面”所有请求和响应都会经过它。像安检闸机不符合要求就直接拦住。依赖注入把需要的外部资源用户信息、配置等以参数形式自动传给视图函数无需手动在每个函数里重复获取。DependsFastAPI 提供的核心工具声明一个函数为依赖项。【AI 场景伪代码路由 API Key 校验依赖】fromfastapiimportFastAPI,Depends,HTTPException,Header appFastAPI(titleAI 模型服务)# ---------- 1. 定义依赖验证 API Key ----------defverify_api_key(x_api_key:strHeader(None)): 从请求头提取 x-api-key 并校验 如果缺少或错误则直接返回 401 ifnotx_api_key:raiseHTTPException(status_code401,detail缺少 API Key)ifx_api_key!secret-2026:raiseHTTPException(status_code403,detail无效的 API Key)# 这里可以返回用户信息或权限注入给视图return{user:authorized_user}# ---------- 2. 路由定义 ----------app.get(/health)defhealth_check():健康检查路由无需 API Keyreturn{status:ok}app.post(/chat)defchat(prompt:str,user_info:dictDepends(verify_api_key)): 受保护的聊天接口自动获取 API Key 验证后的用户信息 # 模拟调用大模型replyf用户{user_info[user]}说:{prompt}。模型回复: 你好return{reply:reply}【运行结果】启动服务后使用curl测试# 1. 无 API Key 请求聊天接口curl-XPOSThttp://127.0.0.1:8000/chat?prompt你好# 响应{detail:缺少 API Key} 状态码 401# 2. 使用正确 API Keycurl-XPOSThttp://127.0.0.1:8000/chat?prompt你好-Hx-api-key: secret-2026# 响应{reply:用户 authorized_user 说: 你好。模型回复: 你好}说明Depends(verify_api_key)让每个需要认证的路由都能自动得到用户信息避免在每个视图里写重复的校验代码。【常见面试题】Q1FastAPI 的依赖注入有什么优势和中间件有何不同参考回答依赖注入可以精细控制到路由或路由组返回具体数据给视图使用中间件是全局的主要用于请求/响应的通用处理如日志、CORS。Depends还可嵌套、缓存、用于权限细分。Q2Flask 如何实现类似 FastAPI 的依赖注入参考回答Flask 本身没有内置 Depends但可以通过flask.g、before_request钩子和自定义装饰器模拟例如在before_request中校验 API Key 并将用户存入g视图再从g获取。更复杂的场景可借助flask-injector等扩展。Q3中间件和依赖注入的执行顺序是怎样的参考回答中间件 → 依赖注入 → 视图函数 → 依赖退出 → 中间件返回。中间件包裹在最外层所有请求先过中间件再进入依赖解析最后到视图。响应时逆序。二、请求生命周期与上下文管理【面试问题】“一个 HTTP 请求从进到 FastAPI 服务到返回响应中间会经历哪些环节如何把一次请求相关的数据比如请求 ID全程记录下来并传递给所有内部调用”【类比】快递包裹的流转快递收件请求到达→ 分拣中心中间件、路由匹配→ 派件员拿到包裹信息依赖注入、上下文→ 派送并签收视图处理→ 信息回传响应返回。每个环节都能在包裹单上盖章记录最后汇总。【原理与术语】请求生命周期接收连接 → ASGI/WSGI 解析 → 中间件处理 → 路由匹配 → 依赖解析 → 视图函数执行 → 响应 → 中间件出栈 → 发送给客户端。上下文管理在一次请求中共享数据FastAPI/Starlette 使用request.state线程安全或协程安全Flask 使用request全局代理内部通过_app_ctx_stack实现线程/协程隔离。request.state可在中间件中设置request.state.request_id uuid4()后续所有依赖和视图都能访问该属性实现请求级别的数据传递。【AI 场景伪代码请求 ID 全链路跟踪】importuuidimporttimefromfastapiimportFastAPI,Request appFastAPI()# ---------- 中间件注入请求 ID 并记录耗时 ----------app.middleware(http)asyncdefadd_request_id(request:Request,call_next): 在每个请求到达时生成唯一 ID存储在 request.state 中 然后记录处理时间并在响应头返回请求 ID request_idstr(uuid.uuid4())[:8]# 简短 IDrequest.state.request_idrequest_id starttime.time()# 调用下一个处理链路由、依赖、视图responseawaitcall_next(request)# 请求处理完成后的后置处理durationtime.time()-start response.headers[X-Request-ID]request_idprint(f[{request_id}]{request.method}{request.url.path}耗时{duration:.3f}s)returnresponse# ---------- 视图使用上下文 ----------app.get(/predict)asyncdefpredict(prompt:str,request:Request): 使用 request.state 获取请求 ID 并输出到日志 req_idrequest.state.request_idprint(f[{req_id}] 收到预测请求:{prompt})# 模拟调用模型awaitasyncio.sleep(0.1)return{prompt:prompt,result:预测结果示例,request_id:req_id}【运行结果】启动服务后访问http://127.0.0.1:8000/predict?prompthello服务端日志和控制台可能输出[abc12345] GET /predict 耗时 0.112s [abc12345] 收到预测请求: hello客户端响应头包含X-Request-ID: abc12345响应体中也返回了 request_id方便排查问题。【常见面试题】Q1FastAPI 里request.state和 Flask 的g有什么区别参考回答request.state生命周期是一次请求请求结束即销毁适合存储请求级属性Flask 的g也是请求级但 Flask 通过应用上下文和请求上下文代理来实现两者用法相似都是安全的临时存储区。Q2为什么不能在中间件或视图中使用全局变量来保存请求数据参考回答因为服务器特别是异步模式下同时处理多个请求全局变量会被所有请求共享造成数据错乱。必须使用请求级别的上下文如request.state或ContextVar来隔离数据。Q3依赖注入中能否获取request对象参考回答可以。只需在依赖函数中声明request: Request参数FastAPI 会自动注入当前请求对象从而获取request.state等上下文信息。三、流式响应StreamingResponse 与 SSEServer-Sent Events【面试问题】“你的大模型 API 返回的是打字机效果需要把生成的 token 实时推送到前端。如何用 FastAPI 实现流式响应SSE 是什么和 WebSocket 有何不同”【类比】网络视频播放普通下载是等整部电影下完再看流式传输则是边下边播用户不用等待。SSE 就像一条单向传输的数据流服务器不断地把最新 token 推给前端前端实时渲染。【原理与术语】StreamingResponseFastAPI/Starlette 提供的响应类接受一个生成器作为内容可逐步发送数据块适用于文件下载、大模型输出流等。Server-Sent Events (SSE)基于 HTTP 的单向流协议服务器以text/event-stream格式持续推送数据每条消息以data:开头用两个换行分隔。前端使用EventSourceAPI 接收。打字机效果大模型逐 token 生成通过生成器yield每个 tokenStreamingResponse 即时发送前端逐字展示。SSE 与 WebSocket 的区别SSE 单向服务器→客户端基于 HTTP 更轻量适合单向推送如模型输出WebSocket 全双工适合双向实时通信。【AI 场景伪代码大模型流式输出 SSE 接口】importasynciofromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponse appFastAPI()asyncdefgenerate_tokens(prompt:str): 异步生成器模拟大模型逐 token 生成 每个 token 等待 0.2 秒后产出并包装成 SSE 格式 # 模拟一个句子被拆成 tokentokens[今天,天气,非常,好,,适合,出游,。]fortokenintokens:awaitasyncio.sleep(0.2)# 模拟生成间隔# SSE 格式要求: data: 内容\n\nyieldfdata:{token}\n\n# 发送结束标记非必须但便于前端识别yielddata: [DONE]\n\napp.get(/stream-chat)asyncdefstream_chat(prompt:str): 返回 StreamingResponsemedia_type 设为 text/event-stream returnStreamingResponse(generate_tokens(prompt),media_typetext/event-stream,headers{Cache-Control:no-cache,Connection:keep-alive,})【运行结果】启动服务后浏览器或curl访问curl-Nhttp://127.0.0.1:8000/stream-chat?prompthello输出逐行缓慢出现data: 今天 data: 天气 data: 非常 data: 好 data: data: 适合 data: 出游 data: 。 data: [DONE]前端 JavaScript 使用EventSource监听即可逐字显示实现打字机效果。【常见面试题】Q1StreamingResponse 和普通 JSON 响应有什么区别参考回答普通响应先生成完整数据再一次性返回客户端需等待全部数据。StreamingResponse 可以边生成边发送降低首字节时间适合大文件传输或实时数据推送。Q2SSE 和 WebSocket 怎么选参考回答如果只需要服务器向客户端推送数据如大模型输出、通知SSE 更简单基于 HTTP 无需额外握手还能利用 HTTP/2。如果客户端也需要频繁向服务器发数据如聊天消息双向发送WebSocket 更合适。Q3异步生成器内如果发生异常怎么办参考回答可以在生成器中用try/except捕获异常通过yield发送错误信息的 SSE 消息如data: {error: ...}\n\n前端根据消息内容决定是否重连。FastAPI 也会在生成器退出时关闭连接。四、异步视图 vs 同步视图的阻塞问题【面试问题】“FastAPI 里如果我把一个同步函数定义为def而不是async def它会阻塞事件循环吗我在async def里不小心调用了time.sleep(1)会怎样”【类比】餐厅里的两种服务员同步服务员一次只服务一桌端菜、等顾客吃完再收碗期间完全不理其他桌阻塞。异步服务员利用顾客自己吃菜的时间去服务其他桌一旦需要主动等待比如等后厨出菜就立刻切换去做别的非阻塞。如果你把一个慢悠悠的同步服务员def里干重活塞进快速轮转的异步团队里他一个人就把整个团队卡住了。【原理与术语】同步视图 (def)FastAPI 会将其放入线程池执行不会阻塞事件循环本身但过多同步任务会耗尽线程池导致请求排队。异步视图 (async def)在同一个事件循环中运行如果内部执行了阻塞同步调用如time.sleep、同步 HTTP 请求会卡住整个事件循环所有并发请求全部停止。解决方案对于同步 I/O应使用async版本的库如aiohttp代替requests对于无法避免的同步代码使用run_in_executor放入线程池或进程池。【AI 场景伪代码模拟阻塞对比】importtimeimportasynciofromfastapiimportFastAPI appFastAPI()# ---------- 同步视图被 FastAPI 在线程池中执行不会阻塞事件循环 ----------app.get(/sync-block)defsync_block():time.sleep(1)# 模拟 CPU 计算或同步阻塞return{msg:sync done}# ---------- 异步视图内错误使用同步阻塞 ----------app.get(/async-bad)asyncdefasync_bad():time.sleep(1)# 危险会阻塞事件循环 1 秒return{msg:async bad}# ---------- 正确做法异步 sleep ----------app.get(/async-good)asyncdefasync_good():awaitasyncio.sleep(1)# 释放控制权不阻塞return{msg:async good}【运行结果】若同时向/async-bad发出两个请求第二个请求必须等第一个的time.sleep(1)结束后才能开始处理总耗时 2 秒。而/async-good和/sync-block都能并发处理各自耗时约 1 秒/sync-block在线程池中运行也不阻塞主循环。验证可以打开两个终端分别curl测试/async-bad会发现它们是串行响应的。【常见面试题】Q1FastAPI 是如何处理同步def视图的参考回答FastAPI 检测到视图是普通def会利用 Starlette 的run_in_threadpool将其放到线程池中运行主事件循环立即恢复不会阻塞。但线程池大小有限大量慢同步任务仍会导致请求排队。Q2在异步视图中调用第三方 SDK 的同步方法怎么办参考回答使用loop.run_in_executor(None, func, *args)将同步调用提交到线程池然后await结果。最好将这类调用封装到一个异步辅助函数中保持视图代码简洁。Q3如何判断一个操作是否需要异步参考回答如果操作主要涉及网络 I/O、磁盘 I/O 且等待时间较长则适合异步如果是纯 Python 计算异步并不能加速反而应放进进程池避免阻塞事件循环。五、CORS、文件上传与后台任务【面试问题】“前端在不同域名下调用你的 AI 服务报跨域错误怎么解决用户上传一张图片给模型推理FastAPI 如何处理文件推理耗时 5 秒你想立即返回给用户‘排队中’怎么做”【类比】CORS就像小区的访客登记浏览器默认不允许外来请求除非服务器明确说“允许来自 example.com 的访客进入”。文件上传就像快递收到包裹需要先保存到临时仓库再交给处理中心。后台任务就像餐厅的“先下单再慢慢做”你给顾客一个取餐号然后让后厨在后台慢慢处理顾客不用站在柜台等。【原理与术语】CORS (跨域资源共享)浏览器安全策略通过响应头Access-Control-Allow-Origin等告知浏览器是否允许跨域请求。FastAPI 中使用CORSMiddleware中间件添加。文件上传FastAPI 使用UploadFile类型可以异步读写上传的文件支持大文件流式存储。后台任务 (BackgroundTasks)FastAPI 提供的工具在返回响应后继续执行某些任务比如异步记录日志、发送邮件、启动模型推理。【AI 场景伪代码CORS 图片上传 后台推理】fromfastapiimportFastAPI,UploadFile,File,BackgroundTasksfromfastapi.middleware.corsimportCORSMiddlewareimporttimeimportuuid appFastAPI()# ---------- 1. CORS 中间件配置 ----------app.add_middleware(CORSMiddleware,allow_origins[https://your-frontend.com],# 允许的域名allow_credentialsTrue,allow_methods[*],allow_headers[*],)# 模拟一个任务状态存储生产环境应使用 Redis 等task_store{}# ---------- 2. 后台推理函数 ----------defrun_model_inference(task_id:str,image_path:str):模拟长时间模型推理更新任务状态time.sleep(3)# 模拟 GPU 推理耗时# 推理完成更新状态task_store[task_id]completed# ---------- 3. 文件上传接口 ----------app.post(/upload-and-predict)asyncdefupload_image(file:UploadFileFile(...),# 接收上传文件background_tasks:BackgroundTasksNone):# 为本次推理生成任务 IDtask_idstr(uuid.uuid4())[:8]# 保存上传文件到临时位置image_pathf/tmp/{file.filename}withopen(image_path,wb)asf:f.write(awaitfile.read())# 异步读取文件内容# 立即返回给客户端同时注册后台任务执行真正的推理background_tasks.add_task(run_model_inference,task_id,image_path)task_store[task_id]processingreturn{task_id:task_id,status:processing,message:图片已上传模型推理在后台执行}# ---------- 4. 查询任务结果 ----------app.get(/task/{task_id})defget_task_status(task_id:str):statustask_store.get(task_id,not found)return{task_id:task_id,status:status}【运行结果】使用curl上传图片curl-XPOST-Ffilephoto.jpghttp://127.0.0.1:8000/upload-and-predict# 即时响应{task_id:abc123,status:processing,message:图片已上传模型推理在后台执行}3 秒后查询状态curlhttp://127.0.0.1:8000/task/abc123# {task_id:abc123,status:completed}这样用户无需等待推理完成立刻得到反馈提升体验。【常见面试题】Q1CORS 预检请求 (OPTIONS) 是什么FastAPI 如何处理参考回答浏览器在发出非简单跨域请求前会先发一个 OPTIONS 请求询问服务器是否允许。FastAPI 的CORSMiddleware会自动响应这些预检请求添加合适的跨域头开发者无需额外处理。Q2UploadFile和普通bytes有什么区别参考回答UploadFile支持异步读写和流式存储适合大文件不会一次性加载全部内容到内存。bytes会把整个文件读到内存小文件更方便。AI 场景上传大图片或视频应优先用UploadFile。Q3BackgroundTasks 和 Celery 这类任务队列有什么不同参考回答BackgroundTasks是 FastAPI 内置的轻量级后台任务运行在同一个进程里没有持久化服务重启任务丢失。Celery 是分布式任务队列支持持久化、重试、调度适合长时间或大量的后台任务。对于轻量级推理后处理可用 BackgroundTasks重要的异步任务应选 Celery。总结作为 AI 应用开发初级工程师你需要掌握用 FastAPI 构建服务的基础能力包括路由设计、鉴权中间件/依赖注入、请求上下文管理、流式响应SSE 打字机效果、正确处理异步与同步阻塞以及文件上传与后台任务。熟悉这些模式后你可以快速搭建生产可用的模型推理 API从容应对相关面试。
Web 框架（FastAPI / Flask）核心概念
发布时间：2026/5/31 15:03:21
Web 框架FastAPI / Flask核心概念AI 应用开发岗初级面试突击指南目标岗位AI 应用开发工程师初级内容聚焦围绕大模型 API 服务化、流式响应、文件上传、异步推理等高频场景讲解 FastAPI为主与 Flask 的路由、中间件、依赖注入、请求生命周期、流式响应、异步视图、CORS、文件上传与后台任务。每个模块均包含面试常见问题 → 类比理解 → 原理与术语说明 → 详细注释的 AI 场景伪代码 →完整运行结果→ 常见面试题及参考回答。一、路由、中间件与依赖注入【面试问题】“你要用 FastAPI 搭建一个 AI 服务路由怎么划分如何给某些接口加 API Key 校验怎么让每个处理函数自动获取当前用户信息而不重复代码”【类比】餐厅的服务流程路由是菜单上的分类冷菜、热菜、汤每个菜名对应一个后厨操作。中间件是餐厅门口的迎宾员不管顾客去哪桌都要先通过他检查比如量体温、查预约。依赖注入是顾客落座后服务员自动端上来的餐具和茶水——你不用每次喊按规矩就给你备好。【原理与术语】路由 (Route)把 URL 路径和 HTTP 方法映射到具体的处理函数FastAPI 使用装饰器app.get(/path)声明。中间件 (Middleware)在请求进入路由处理函数之前以及响应返回给客户端之前执行公共逻辑如日志、鉴权、跨域。FastAPI 中通过app.middleware(http)定义Flask 中通过app.before_request和app.after_request。依赖注入 (Dependency Injection)FastAPI 的Depends允许把函数声明为“依赖”视图函数需要时自动调用并注入结果常用于提取当前用户、获取数据库会话等实现可复用组件。 术语速查术语解释路由URL 和处理函数的映射关系。比如/chat对应大模型聊天接口。中间件一个“切面”所有请求和响应都会经过它。像安检闸机不符合要求就直接拦住。依赖注入把需要的外部资源用户信息、配置等以参数形式自动传给视图函数无需手动在每个函数里重复获取。DependsFastAPI 提供的核心工具声明一个函数为依赖项。【AI 场景伪代码路由 API Key 校验依赖】fromfastapiimportFastAPI,Depends,HTTPException,Header appFastAPI(titleAI 模型服务)# ---------- 1. 定义依赖验证 API Key ----------defverify_api_key(x_api_key:strHeader(None)): 从请求头提取 x-api-key 并校验 如果缺少或错误则直接返回 401 ifnotx_api_key:raiseHTTPException(status_code401,detail缺少 API Key)ifx_api_key!secret-2026:raiseHTTPException(status_code403,detail无效的 API Key)# 这里可以返回用户信息或权限注入给视图return{user:authorized_user}# ---------- 2. 路由定义 ----------app.get(/health)defhealth_check():健康检查路由无需 API Keyreturn{status:ok}app.post(/chat)defchat(prompt:str,user_info:dictDepends(verify_api_key)): 受保护的聊天接口自动获取 API Key 验证后的用户信息 # 模拟调用大模型replyf用户{user_info[user]}说:{prompt}。模型回复: 你好return{reply:reply}【运行结果】启动服务后使用curl测试# 1. 无 API Key 请求聊天接口curl-XPOSThttp://127.0.0.1:8000/chat?prompt你好# 响应{detail:缺少 API Key} 状态码 401# 2. 使用正确 API Keycurl-XPOSThttp://127.0.0.1:8000/chat?prompt你好-Hx-api-key: secret-2026# 响应{reply:用户 authorized_user 说: 你好。模型回复: 你好}说明Depends(verify_api_key)让每个需要认证的路由都能自动得到用户信息避免在每个视图里写重复的校验代码。【常见面试题】Q1FastAPI 的依赖注入有什么优势和中间件有何不同参考回答依赖注入可以精细控制到路由或路由组返回具体数据给视图使用中间件是全局的主要用于请求/响应的通用处理如日志、CORS。Depends还可嵌套、缓存、用于权限细分。Q2Flask 如何实现类似 FastAPI 的依赖注入参考回答Flask 本身没有内置 Depends但可以通过flask.g、before_request钩子和自定义装饰器模拟例如在before_request中校验 API Key 并将用户存入g视图再从g获取。更复杂的场景可借助flask-injector等扩展。Q3中间件和依赖注入的执行顺序是怎样的参考回答中间件 → 依赖注入 → 视图函数 → 依赖退出 → 中间件返回。中间件包裹在最外层所有请求先过中间件再进入依赖解析最后到视图。响应时逆序。二、请求生命周期与上下文管理【面试问题】“一个 HTTP 请求从进到 FastAPI 服务到返回响应中间会经历哪些环节如何把一次请求相关的数据比如请求 ID全程记录下来并传递给所有内部调用”【类比】快递包裹的流转快递收件请求到达→ 分拣中心中间件、路由匹配→ 派件员拿到包裹信息依赖注入、上下文→ 派送并签收视图处理→ 信息回传响应返回。每个环节都能在包裹单上盖章记录最后汇总。【原理与术语】请求生命周期接收连接 → ASGI/WSGI 解析 → 中间件处理 → 路由匹配 → 依赖解析 → 视图函数执行 → 响应 → 中间件出栈 → 发送给客户端。上下文管理在一次请求中共享数据FastAPI/Starlette 使用request.state线程安全或协程安全Flask 使用request全局代理内部通过_app_ctx_stack实现线程/协程隔离。request.state可在中间件中设置request.state.request_id uuid4()后续所有依赖和视图都能访问该属性实现请求级别的数据传递。【AI 场景伪代码请求 ID 全链路跟踪】importuuidimporttimefromfastapiimportFastAPI,Request appFastAPI()# ---------- 中间件注入请求 ID 并记录耗时 ----------app.middleware(http)asyncdefadd_request_id(request:Request,call_next): 在每个请求到达时生成唯一 ID存储在 request.state 中 然后记录处理时间并在响应头返回请求 ID request_idstr(uuid.uuid4())[:8]# 简短 IDrequest.state.request_idrequest_id starttime.time()# 调用下一个处理链路由、依赖、视图responseawaitcall_next(request)# 请求处理完成后的后置处理durationtime.time()-start response.headers[X-Request-ID]request_idprint(f[{request_id}]{request.method}{request.url.path}耗时{duration:.3f}s)returnresponse# ---------- 视图使用上下文 ----------app.get(/predict)asyncdefpredict(prompt:str,request:Request): 使用 request.state 获取请求 ID 并输出到日志 req_idrequest.state.request_idprint(f[{req_id}] 收到预测请求:{prompt})# 模拟调用模型awaitasyncio.sleep(0.1)return{prompt:prompt,result:预测结果示例,request_id:req_id}【运行结果】启动服务后访问http://127.0.0.1:8000/predict?prompthello服务端日志和控制台可能输出[abc12345] GET /predict 耗时 0.112s [abc12345] 收到预测请求: hello客户端响应头包含X-Request-ID: abc12345响应体中也返回了 request_id方便排查问题。【常见面试题】Q1FastAPI 里request.state和 Flask 的g有什么区别参考回答request.state生命周期是一次请求请求结束即销毁适合存储请求级属性Flask 的g也是请求级但 Flask 通过应用上下文和请求上下文代理来实现两者用法相似都是安全的临时存储区。Q2为什么不能在中间件或视图中使用全局变量来保存请求数据参考回答因为服务器特别是异步模式下同时处理多个请求全局变量会被所有请求共享造成数据错乱。必须使用请求级别的上下文如request.state或ContextVar来隔离数据。Q3依赖注入中能否获取request对象参考回答可以。只需在依赖函数中声明request: Request参数FastAPI 会自动注入当前请求对象从而获取request.state等上下文信息。三、流式响应StreamingResponse 与 SSEServer-Sent Events【面试问题】“你的大模型 API 返回的是打字机效果需要把生成的 token 实时推送到前端。如何用 FastAPI 实现流式响应SSE 是什么和 WebSocket 有何不同”【类比】网络视频播放普通下载是等整部电影下完再看流式传输则是边下边播用户不用等待。SSE 就像一条单向传输的数据流服务器不断地把最新 token 推给前端前端实时渲染。【原理与术语】StreamingResponseFastAPI/Starlette 提供的响应类接受一个生成器作为内容可逐步发送数据块适用于文件下载、大模型输出流等。Server-Sent Events (SSE)基于 HTTP 的单向流协议服务器以text/event-stream格式持续推送数据每条消息以data:开头用两个换行分隔。前端使用EventSourceAPI 接收。打字机效果大模型逐 token 生成通过生成器yield每个 tokenStreamingResponse 即时发送前端逐字展示。SSE 与 WebSocket 的区别SSE 单向服务器→客户端基于 HTTP 更轻量适合单向推送如模型输出WebSocket 全双工适合双向实时通信。【AI 场景伪代码大模型流式输出 SSE 接口】importasynciofromfastapiimportFastAPIfromfastapi.responsesimportStreamingResponse appFastAPI()asyncdefgenerate_tokens(prompt:str): 异步生成器模拟大模型逐 token 生成 每个 token 等待 0.2 秒后产出并包装成 SSE 格式 # 模拟一个句子被拆成 tokentokens[今天,天气,非常,好,,适合,出游,。]fortokenintokens:awaitasyncio.sleep(0.2)# 模拟生成间隔# SSE 格式要求: data: 内容\n\nyieldfdata:{token}\n\n# 发送结束标记非必须但便于前端识别yielddata: [DONE]\n\napp.get(/stream-chat)asyncdefstream_chat(prompt:str): 返回 StreamingResponsemedia_type 设为 text/event-stream returnStreamingResponse(generate_tokens(prompt),media_typetext/event-stream,headers{Cache-Control:no-cache,Connection:keep-alive,})【运行结果】启动服务后浏览器或curl访问curl-Nhttp://127.0.0.1:8000/stream-chat?prompthello输出逐行缓慢出现data: 今天 data: 天气 data: 非常 data: 好 data: data: 适合 data: 出游 data: 。 data: [DONE]前端 JavaScript 使用EventSource监听即可逐字显示实现打字机效果。【常见面试题】Q1StreamingResponse 和普通 JSON 响应有什么区别参考回答普通响应先生成完整数据再一次性返回客户端需等待全部数据。StreamingResponse 可以边生成边发送降低首字节时间适合大文件传输或实时数据推送。Q2SSE 和 WebSocket 怎么选参考回答如果只需要服务器向客户端推送数据如大模型输出、通知SSE 更简单基于 HTTP 无需额外握手还能利用 HTTP/2。如果客户端也需要频繁向服务器发数据如聊天消息双向发送WebSocket 更合适。Q3异步生成器内如果发生异常怎么办参考回答可以在生成器中用try/except捕获异常通过yield发送错误信息的 SSE 消息如data: {error: ...}\n\n前端根据消息内容决定是否重连。FastAPI 也会在生成器退出时关闭连接。四、异步视图 vs 同步视图的阻塞问题【面试问题】“FastAPI 里如果我把一个同步函数定义为def而不是async def它会阻塞事件循环吗我在async def里不小心调用了time.sleep(1)会怎样”【类比】餐厅里的两种服务员同步服务员一次只服务一桌端菜、等顾客吃完再收碗期间完全不理其他桌阻塞。异步服务员利用顾客自己吃菜的时间去服务其他桌一旦需要主动等待比如等后厨出菜就立刻切换去做别的非阻塞。如果你把一个慢悠悠的同步服务员def里干重活塞进快速轮转的异步团队里他一个人就把整个团队卡住了。【原理与术语】同步视图 (def)FastAPI 会将其放入线程池执行不会阻塞事件循环本身但过多同步任务会耗尽线程池导致请求排队。异步视图 (async def)在同一个事件循环中运行如果内部执行了阻塞同步调用如time.sleep、同步 HTTP 请求会卡住整个事件循环所有并发请求全部停止。解决方案对于同步 I/O应使用async版本的库如aiohttp代替requests对于无法避免的同步代码使用run_in_executor放入线程池或进程池。【AI 场景伪代码模拟阻塞对比】importtimeimportasynciofromfastapiimportFastAPI appFastAPI()# ---------- 同步视图被 FastAPI 在线程池中执行不会阻塞事件循环 ----------app.get(/sync-block)defsync_block():time.sleep(1)# 模拟 CPU 计算或同步阻塞return{msg:sync done}# ---------- 异步视图内错误使用同步阻塞 ----------app.get(/async-bad)asyncdefasync_bad():time.sleep(1)# 危险会阻塞事件循环 1 秒return{msg:async bad}# ---------- 正确做法异步 sleep ----------app.get(/async-good)asyncdefasync_good():awaitasyncio.sleep(1)# 释放控制权不阻塞return{msg:async good}【运行结果】若同时向/async-bad发出两个请求第二个请求必须等第一个的time.sleep(1)结束后才能开始处理总耗时 2 秒。而/async-good和/sync-block都能并发处理各自耗时约 1 秒/sync-block在线程池中运行也不阻塞主循环。验证可以打开两个终端分别curl测试/async-bad会发现它们是串行响应的。【常见面试题】Q1FastAPI 是如何处理同步def视图的参考回答FastAPI 检测到视图是普通def会利用 Starlette 的run_in_threadpool将其放到线程池中运行主事件循环立即恢复不会阻塞。但线程池大小有限大量慢同步任务仍会导致请求排队。Q2在异步视图中调用第三方 SDK 的同步方法怎么办参考回答使用loop.run_in_executor(None, func, *args)将同步调用提交到线程池然后await结果。最好将这类调用封装到一个异步辅助函数中保持视图代码简洁。Q3如何判断一个操作是否需要异步参考回答如果操作主要涉及网络 I/O、磁盘 I/O 且等待时间较长则适合异步如果是纯 Python 计算异步并不能加速反而应放进进程池避免阻塞事件循环。五、CORS、文件上传与后台任务【面试问题】“前端在不同域名下调用你的 AI 服务报跨域错误怎么解决用户上传一张图片给模型推理FastAPI 如何处理文件推理耗时 5 秒你想立即返回给用户‘排队中’怎么做”【类比】CORS就像小区的访客登记浏览器默认不允许外来请求除非服务器明确说“允许来自 example.com 的访客进入”。文件上传就像快递收到包裹需要先保存到临时仓库再交给处理中心。后台任务就像餐厅的“先下单再慢慢做”你给顾客一个取餐号然后让后厨在后台慢慢处理顾客不用站在柜台等。【原理与术语】CORS (跨域资源共享)浏览器安全策略通过响应头Access-Control-Allow-Origin等告知浏览器是否允许跨域请求。FastAPI 中使用CORSMiddleware中间件添加。文件上传FastAPI 使用UploadFile类型可以异步读写上传的文件支持大文件流式存储。后台任务 (BackgroundTasks)FastAPI 提供的工具在返回响应后继续执行某些任务比如异步记录日志、发送邮件、启动模型推理。【AI 场景伪代码CORS 图片上传 后台推理】fromfastapiimportFastAPI,UploadFile,File,BackgroundTasksfromfastapi.middleware.corsimportCORSMiddlewareimporttimeimportuuid appFastAPI()# ---------- 1. CORS 中间件配置 ----------app.add_middleware(CORSMiddleware,allow_origins[https://your-frontend.com],# 允许的域名allow_credentialsTrue,allow_methods[*],allow_headers[*],)# 模拟一个任务状态存储生产环境应使用 Redis 等task_store{}# ---------- 2. 后台推理函数 ----------defrun_model_inference(task_id:str,image_path:str):模拟长时间模型推理更新任务状态time.sleep(3)# 模拟 GPU 推理耗时# 推理完成更新状态task_store[task_id]completed# ---------- 3. 文件上传接口 ----------app.post(/upload-and-predict)asyncdefupload_image(file:UploadFileFile(...),# 接收上传文件background_tasks:BackgroundTasksNone):# 为本次推理生成任务 IDtask_idstr(uuid.uuid4())[:8]# 保存上传文件到临时位置image_pathf/tmp/{file.filename}withopen(image_path,wb)asf:f.write(awaitfile.read())# 异步读取文件内容# 立即返回给客户端同时注册后台任务执行真正的推理background_tasks.add_task(run_model_inference,task_id,image_path)task_store[task_id]processingreturn{task_id:task_id,status:processing,message:图片已上传模型推理在后台执行}# ---------- 4. 查询任务结果 ----------app.get(/task/{task_id})defget_task_status(task_id:str):statustask_store.get(task_id,not found)return{task_id:task_id,status:status}【运行结果】使用curl上传图片curl-XPOST-Ffilephoto.jpghttp://127.0.0.1:8000/upload-and-predict# 即时响应{task_id:abc123,status:processing,message:图片已上传模型推理在后台执行}3 秒后查询状态curlhttp://127.0.0.1:8000/task/abc123# {task_id:abc123,status:completed}这样用户无需等待推理完成立刻得到反馈提升体验。【常见面试题】Q1CORS 预检请求 (OPTIONS) 是什么FastAPI 如何处理参考回答浏览器在发出非简单跨域请求前会先发一个 OPTIONS 请求询问服务器是否允许。FastAPI 的CORSMiddleware会自动响应这些预检请求添加合适的跨域头开发者无需额外处理。Q2UploadFile和普通bytes有什么区别参考回答UploadFile支持异步读写和流式存储适合大文件不会一次性加载全部内容到内存。bytes会把整个文件读到内存小文件更方便。AI 场景上传大图片或视频应优先用UploadFile。Q3BackgroundTasks 和 Celery 这类任务队列有什么不同参考回答BackgroundTasks是 FastAPI 内置的轻量级后台任务运行在同一个进程里没有持久化服务重启任务丢失。Celery 是分布式任务队列支持持久化、重试、调度适合长时间或大量的后台任务。对于轻量级推理后处理可用 BackgroundTasks重要的异步任务应选 Celery。总结作为 AI 应用开发初级工程师你需要掌握用 FastAPI 构建服务的基础能力包括路由设计、鉴权中间件/依赖注入、请求上下文管理、流式响应SSE 打字机效果、正确处理异步与同步阻塞以及文件上传与后台任务。熟悉这些模式后你可以快速搭建生产可用的模型推理 API从容应对相关面试。
