期待已久的第三篇来了。前面两篇分别把 FastAPI 的基础路由、参数校验、模型调用的封装讲完了如果你是从第一篇一路跟过来的现在手上的服务应该已经能跑通了——用户发一个请求模型返回一段文本接口正常响应。但这只是万里长征第一步一旦把服务暴露给真实用户能跑和扛得住之间隔着流式传输、并发治理、可测试性、部署调优这几道坎。这篇文章把这些坎挨个过一遍。1. 第三期选题生成式服务与传统接口的本质差异1.1 前两期的边界能跑通接口不等于能上生产写接口这件事很多有经验的开发者会觉得轻车熟路FastAPI 定义路由、做参数校验、调数据库、返回 JSON这套流程熟得不能再熟。但生成式人工智能服务一旦接入原来那套 REST 接口的直觉马上就失效了。前两期我们做了哪些事建了 FastAPI 项目骨架定义了请求体模型包装了模型调用函数让每一路请求都能拿到完整结果再返回。这在联调阶段完全没问题可实际用起来就会发现普通接口的用户等待时间在 200 毫秒以内而一次大模型生成的耗时动辄几秒到几十秒。让用户对着空白页面干等十几秒体验极其糟糕前端程序员会直接找你吵架。更麻烦的是资源模型完全不同。普通接口的瓶颈一般落在数据库连接池或者下游服务的响应速度上而生成式接口的瓶颈是模型推理本身显存、算力、批量大小。同一个模型进程同一时刻通常只能喂进去有限的请求超过这个数要么排队要么直接撑爆显存。这些都是生成式服务特有的问题靠原来那套加两台机器的思路解决不了因为瓶颈不在 Web 框架而在模型运行时。1.2 流式和一次性返回的差别决定了接口设计思路我把传统接口和生成式接口横向对比着看最核心的三个差异分别是响应时间量级、返回语义、状态管理方式。传统接口是短事务生成式是长事务传统接口返回完整 JSON生成式可以逐字吐传统接口无状态生成式高频场景需要带上下文。下面的表格把差异列得很清楚维度传统 REST 接口生成式 AI 服务接口单次请求耗时几毫秒到几百毫秒几秒到几十秒返回方式一次性 JSON流式逐块返回瓶颈资源数据库连接、网络显存、算力、并发槽位失败处理失败就返回 4xx/5xx可能中途断流需要协议层处理上下文管理通常无状态高频场景需要会话/记忆这三点差异直接决定了我们现在要做的技术选型。如果你已经在生产环境里维护过生成式服务大概率踩过这些坑前端超时报错、并发一高就 OOM、测试不知道该怎么 mock 生成器。这篇文章的后半部分就是针对这些具体问题逐个给方案。2. 接入流式接口SSE 与异步生成器的落地细节2.1 普通 JSON 返回和流式返回到底该选哪个先说结论只要是面向真实用户的对话、写作、内容生成类场景一律优先做流式返回。原因很简单用户心理等待时间是有极限的超过 1 秒没有反馈就会焦虑而流式输出在生成开始后的几百毫秒内就能看到第一个字后续内容连续出现用户体感上会觉得服务在干活而不是卡死了。实现上SSEServer-Sent Events是最合适的方式没有之一。它基于 HTTP 长连接服务端可以持续推送数据客户端只要用 EventSource 或者 fetch reader 就能接住。FastAPI 对这一套支持得很好核心就是响应的media_type和生成器类型。有人问为什么不直接用 WebSocket对于纯文本生成场景WebSocket 是杀鸡用牛刀而且需要维护双向通道状态SSE 是单向推送服务端写到哪儿客户端读到哪儿天然匹配用户提问模型回答的对话模型。2.2 同步生成器还是异步生成器选错类型流程就断这里有个容易踩的坑就是生成器类型的选择。假设你的模型 SDK 提供了一个同步生成器model.generate_stream(prompt)每次next()返回一段文本。如果直接在 FastAPI 的 StreamingResponse 里返回这个同步生成器框架其实能处理它会把生成器放到线程池里逐个取结果。但这会引入两个问题一是每个请求占一个线程池工作线程并发一高线程切换开销显著二是当你需要做异步超时控制、异步清理时同步生成器很难配合。最佳实践是不管底层模型 SDK 是同步还是异步统一在服务层包装成异步生成器。同步 SDK 可以用asyncio.to_thread包一层把耗时的生成调用丢到后台线程再把结果通过await取回来上层代码保持全异步。下面这段代码是服务层的推荐写法import asyncio from fastapi import FastAPI from fastapi.responses import StreamingResponse from typing import AsyncGenerator app FastAPI() async def stream_answer(prompt: str) - AsyncGenerator[str, None]: # 这里示例用同步生成器模拟模型实际项目里换成你的模型 SDK def sync_generate(): chunks [你好, 我是, 一个, 生成式, 服务。] for chunk in chunks: yield chunk gen sync_generate() loop asyncio.get_running_loop() try: while True: chunk await loop.run_in_executor(None, next, gen) yield chunk except StopIteration: return app.post(/chat) async def chat(prompt: str): return StreamingResponse(stream_answer(prompt), media_typetext/event-stream)注意上面的run_in_executor(None, next, gen)会把每次取块的操作放到默认线程池执行避免阻塞事件循环。如果你的模型 SDK 本身就是异步的直接用async for chunk in model.stream_async(prompt)即可连线程池都不用。2.3 事件流格式与断连清理前端能不能看清边界SSE 的text/event-stream媒体类型在协议层是确定的但数据格式有讲究。标准的 SSE 事件每条以data:开头以空行结尾。前端的 EventSource 会按照空行拆分事件所以如果你直接把裸文本抛出去前端收到的可能是整块拼好的字符串而不是逐字效果。想要真正实现打字机效果最好每个 chunk 都作为独立事件推送from fastapi.responses import StreamingResponse import json async def stream_chunk_events(prompt: str): async for chunk in stream_answer(prompt): data json.dumps({content: chunk}, ensure_asciiFalse) yield fdata: {data}\n\n app.post(/chat/event) async def chat_event(prompt: str): return StreamingResponse(stream_chunk_events(prompt), media_typetext/event-stream)前端拿着每一条data里的content字段往界面上追加文本就能得到逐字输出的效果。这里还有一层容易忽略的地方如果生成过程中发生了异常后端不能直接甩一个 500 状态码因为连接已经建立请求已经进入流式传输过程。正确做法是在流里发一个状态事件比如data: {type: error, message: ...}让前端感知到并优雅提示。下面这段代码演示了中间异常时如何兜底async def stream_chunk_events_with_error(prompt: str): try: async for chunk in stream_answer(prompt): yield fdata: {json.dumps({type: chunk, content: chunk})}\n\n yield fdata: {json.dumps({type: done})}\n\n except Exception as exc: yield fdata: {json.dumps({type: error, message: str(exc)})}\n\n另一个高频事故是客户端突然断连。移动端切后台、浏览器刷新连接的接收端没了但服务端还在继续生成。浪费算力还是小事严重的是生成循环可能永远不退出。为了解决这个问题FastAPI 的 StreamingResponse 会在客户端断开后取消异步生成器生成器的finally块会执行你可以在这个钩子里做状态清理。但是很多模型 SDK 不响应取消所以还要给生成器套一层心跳检查逻辑每次迭代检查某个任务是否被取消。实际项目中我习惯写一个小工具函数来包装生成器import asyncio async def guarded_stream(gen: AsyncGenerator[str, None]) - AsyncGenerator[str, None]: try: async for chunk in gen: if asyncio.current_task() and asyncio.current_task().cancelled(): break yield chunk finally: await gen.aclose()这段代码在finally里主动调用aclose()确保底层生成器的资源被释放不会出现僵尸任务。3. 并发治理异步信号量、超时与队列的配合3.1 把并发数限制在模型能力范围内而不是框架能力范围内FastAPI 本身的异步框架能扛住成千上万个并发连接但这不意味着模型也能扛住。真实场景中你总希望用户请求尽量不排队但如果让它们全冲进模型进程显存瞬间爆炸服务直接不可用。这里有一个朴素的道理并发上限应该由模型推理服务的承载能力决定而不是由 Web 框架铺开多少个连接决定。先把模型服务的承载能力搞清楚。以常见的 GPU 推理服务为例一个 batch 槽位能处理一个请求批处理可以提升吞吐但不会无限扩展。最简单的压测方法写一个并发脚本不停增加同时请求数量观察显存占用和首 token 延迟找到拐点。这个拐点的并发数就是信号量大小的依据。没有压测数据就别拍脑袋设一个夸张的数值。3.2 asyncio.Semaphore 到底该放在哪里很多教程教你在 FastAPI 路由函数里用asyncio.Semaphore但放在路由函数里有个问题如果路由函数直接返回StreamingResponse信号量很可能在生成器真正开始前就被释放了。正确的做法是让生成器自身持有一个信号量引用在开始生成前获取生成结束后释放。下面这段代码把信号量封装成一个依赖import asyncio from fastapi import Depends from typing import AsyncGenerator class GenerationLimiter: def __init__(self, max_concurrent_gen: int 4): self._sem asyncio.Semaphore(max_concurrent_gen) async def acquire(self): await self._sem.acquire() def release(self): self._sem.release() async def run(self, gen: AsyncGenerator[str, None]) - AsyncGenerator[str, None]: await self.acquire() try: async for chunk in gen: yield chunk finally: self.release() async def get_limiter() - GenerationLimiter: return GenerationLimiter(max_concurrent_gen4) app.state.generation_limiter GenerationLimiter(max_concurrent_gen4)注意Semaphore的acquire()是协程必须在事件循环中执行所以我把acquire和release封装进了GenerationLimiter由生成器自己负责生命周期。路由里的写法变成app.post(/chat) async def chat(prompt: str): limiter app.state.generation_limiter return StreamingResponse(limiter.run(stream_answer(prompt)), media_typetext/event-stream)这样每个请求在进入生成循环之前都会排队等待一个槽位如果已经有 4 个生成任务在跑第 5 个请求会在信号量上挂起而不是直接涌入模型。用户看到的现象是这个请求迟迟没返回但至少服务本身不会被打崩。3.3 超时、背压与队列给排队中的请求一个明确命运生产环境中排队本身就是一种体验问题。用户等了 8 秒钟前面 4 个请求还在生成第 5 个请求一直在信号量上悬着最后前端超时断连后端的请求任务却还停在信号量上。这个问题必须用一个外层超时来解决等在信号量上的时间也算在请求时间内超了就返回 503不要让请求无限期挂起。实现方案是给整个生成流程包上一层asyncio.wait_for但注意wait_for和流式生成之间需要谨慎配合。一旦超时wait_for会取消当前任务如果任务正卡在模型 SDK 的同步调用里可能不是立刻取消所以要在取消之后确保信号量一定被释放。下面的工具类把所有细节串在一起import asyncio from contextlib import asynccontextmanager asynccontextmanager async def limited_generation(limiter: GenerationLimiter, timeout: float 30.0): try: await asyncio.wait_for(limiter.acquire(), timeouttimeout) yield finally: limiter.release()配合这个上下文管理器路由里就能写出更清晰的逻辑app.post(/chat) async def chat(prompt: str): limiter app.state.generation_limiter async def stream_with_timeout(): async with limited_generation(limiter, timeout10.0) as _: async for chunk in stream_answer(prompt): yield chunk yield fdata: {json.dumps({type: done})}\n\n return StreamingResponse(stream_with_timeout(), media_typetext/event-stream)再进阶一点还可以在信号量前面加一个有界队列信号量控制的是正在生成的并发数队列控制的是排队等待的最大数量。超过队列容量的请求直接拒绝避免服务在极端流量下把所有请求都挂在等待状态里。实现上队列可以用简单的计数器模拟比如等待信号量的请求数超过 N 就返回 429。不过到了这一步你已经需要做全局的限流策略了FastAPI 的中间件或者独立限流组件都可以接进来目标很单一超过承载能力时快速失败而不是慢慢拖死。4. 让测试真正可信模拟生成器与流式故障注入4.1 为什么常规的接口测试方式在流式接口上失灵写单元测试的习惯很多人都有但到了流式接口这就卡壳了。以前测试一个 POST 接口发请求、拿 JSON、断言字段值三步搞定。现在接口返回的是StreamingResponse不能直接断言整个响应体等于某个 JSON你需要读流、按事件切分、再逐条校验。如果你还停留在发一个请求等 response的思路里测试大概率会超时或拿不到预期结果。正确的测试路径是使用 FastAPI 的TestClient或者 httpx 的异步 ASGI 传输层。前者是同步阻塞方式适合快速验证但读取流式响应时不够灵活后者能完全控制请求还支持async for读取流适合对流式接口做精确断言。推荐优先用后者。4.2 mock 一个生成器三块内容缺一不可测试生成式服务的核心是把真实模型调用替换成一个可控的假生成器。这个假生成器至少要有三块行为正常逐块返回、中断抛异常、结束后释放资源。下面的代码是一个可配置的假生成器from typing import AsyncGenerator, Optional class MockModelStream: staticmethod async def good_stream(chunks: list[str]) - AsyncGenerator[str, None]: for chunk in chunks: yield chunk staticmethod async def fail_stream(chunks: list[str], error_at: int) - AsyncGenerator[str, None]: for i, chunk in enumerate(chunks): if i error_at: raise RuntimeError(模拟生成中断) yield chunk单元测试时依赖注入替换模型客户端对象后面路由内部调用的是stream_answer(prompt)你可以在测试里把stream_answer临时替换为MockModelStream.good_stream。如果路由把生成函数写在模块顶层可以用 pytest 的 monkeypatch 直接替换模块属性import pytest from httpx import AsyncClient, ASGITransport async def test_streaming_success(monkeypatch): async def fake_stream(prompt: str): async for chunk in MockModelStream.good_stream([Hello, World]): yield chunk monkeypatch.setattr(myapp.main.stream_answer, fake_stream) transport ASGITransport(appapp) async with AsyncClient(transporttransport, base_urlhttp://test) as client: async with client.stream(POST, /chat, json{prompt: hi}) as response: assert response.status_code 200 chunks [item async for item in response.aiter_text()] assert .join(chunks) Hello World4.3 故障注入测试挂在连接建立后才能验证协议兜底流式接口的故障点分为三类连接建立前的参数错误、连接建立后的生成中断、生成结束后的资源清理。第一类用普通参数断言就能覆盖第二类是最值得写的因为它验证的是协议层的行为——客户端已经建立流服务端不能在生成中途返回 HTTP 状态码错误只能通过 SSE 事件传达。第三类用aclose()的 mock 断言就能验证。故障注入测试可以这样写用一个在第三个块处抛异常的假生成器请求已经进入流式传输然后逐个读事件断言读到typeerror或类似的事件并且连接没有中途关闭。这个测试通过才能说明你的错误事件机制真的有效。另外还要测一个大多数人忽略的场景客户端中途断开。测试时不能真的把一个短连接客户端塞进去等待因为 FastAPI 的 StreamingResponse 只有在写事件时发现连接已关闭才会触发取消。可以把生成器包在guarded_stream里然后在生成器内部人为asyncio.sleep同时客户端只读两个块就退出最后断言生成器的finally块执行到了并且信号量成功释放。这一步能防止生产环境中出现僵尸生成任务。5. 部署参数与运维配置worker、超时与观测5.1 Uvicorn worker 数量和模型并发池的匹配部署生成式服务时最常见的问题是 worker 数量设得过于激进。默认uvicorn main:app --workers 4看上去是合理的多进程部署但如果每个 worker 都是一个独立的模型调用进程而模型推理又占大量显存4 个 worker 意味着显存占用翻四倍可能直接 OOM。所以你要搞清楚这里的层级关系每个 Uvicorn worker 是一个 Python 进程进程内跑着 FastAPI 应用和事件循环模型并发池比如前面用信号量控制的生成槽位是进程内的软限制而显存是硬资源。务实的部署策略是GPU 机器上Uvicorn worker 数量通常设成 1 或 2 就够了真正控制吞吐的是进程内的信号量并发槽位。对于 CPU 推理的小型模型可以适当调多 worker但如果每个 worker 都加载同样一份模型内存翻倍的压力也不容忽视。启动命令可以这样写uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1 --limit-concurrency 500配合代码里的信号量限流真正的并发天花板就不会被进程数冲破了。--limit-concurrency限制的是 HTTP 连接并发数只用于保护 Web 层不要把它和模型生成并发混为一谈。5.2 多层超时与优雅停机生成式服务天然长耗时超时设置必须分层。第一层是客户端与 Web 服务器之间的请求超时第二层是应用内信号量等待超时第三层是模型调用的内部超时。这些超时逐层收紧才能保证任何一层出现故障时不至于无限等待。具体到我实测过的经验客户端侧超时建议设置成单次生成耗时的 1.5 倍比如模型生成最慢 20 秒那客户端超时就给 30 秒。应用内asyncio.wait_for的超时则要稍微小于客户端超时给协议层留出错误返回的时间。模型调用的内部超时在最底层可以控制在单次流式迭代的合理时间内比如每块 10 秒没返回就判定调用异常。优雅停机是很容易被忽略的一环。服务更新版本时你可能直接杀进程但正在执行的生成任务会被硬生生打断造成前端收到断流。正确做法是给 Uvicorn 配置优雅超时时间让它先停止接收新请求再等待存量生成任务完成。进程管理器里配置的停止超时建议设在 30 秒以上因为一次生成可能本身就长达二三十秒。停下后先等一等再 kill 强杀。5.3 观测日志链路与关键指标不能靠感觉生成式服务的排查难度比普通服务高一个量级因为一个请求从进来到最终写完可能经过十几秒期间发生在生成器内部的任何异常在上层看都是流断了或响应超时。如果日志里没有请求 ID 贯穿全链路你根本无从定位是哪次生成出了问题。我在项目里会给每个请求生成一个request_id放进日志上下文同时在生成器入口、每个 chunk 输出、错误事件、信号量获取/释放四个位置打日志。这里给一个最简单的日志结构示例import logging import uuid import time logger logging.getLogger(genai-service) async def stream_answer_with_trace(prompt: str, request_id: str): start time.monotonic() logger.info(generation_start, extra{request_id: request_id, prompt_len: len(prompt)}) chunk_count 0 try: async for chunk in actual_model_stream(prompt): chunk_count 1 yield chunk except Exception as exc: logger.warning(generation_error, extra{request_id: request_id, error: str(exc), chunk_count: chunk_count}) raise finally: duration time.monotonic() - start logger.info(generation_end, extra{request_id: request_id, duration: duration, chunk_count: chunk_count})观测指标方面除了常规的请求量、错误率、延迟之外建议额外记录排队等待信号量的时间、首 token 延迟、每块间隔时长、并发生成数。前两个直接影响用户体验后两个能告诉你模型服务的健康状态。把prometheus_client和 FastAPI 中间件接起来以后在 Grafana 里看这几个指标的变化比翻日志要直观得多。6. 从服务到编排提示模板与工具调用的扩展思路6.1 生成式服务不只是请求-应答还有多轮会话与状态管理做到这一步你的服务已经具备基本的生成能力和并发治理能力但真实业务通常还有更高一层的需求多轮对话。用户不是每次都发一个独立问题而是带着上下文的连续提问。如果每次请求都把整段历史重新发给模型上下文越长越浪费 token 和算力如果完全不传历史又答非所问。这里就要引入会话状态管理。最简单的方案是用 Redis 存会话窗口每次请求只取出最近若干轮消息拼接成大模型输入格式。FastAPI 里可以在依赖中读取user_id、session_id再走一遍序列化与拼接的流程。完整的会话管理器不难写但要注意三个细节窗口大小上限通常限制在最近 20-40 条消息、过期时间给每个 session 设 TTL、以及上下文的 token 截断策略。6.2 提示模板管理别再在代码里写死提示词了很多人写提示词直接在业务代码里拼字符串调试了几轮之后代码里到处都是f你现在是一个...用户的问题是{question}。一旦产品要调整话术或者做多个模型版本对比改代码、发布版本整个流程效率极低。我推荐把提示模板抽出来单独做一个模块或者目录用模板引擎管起来。Jinja2 就够用模板文件按场景命名渲染逻辑只负责传变量。模板管理的另一个好处是方便走配置化。同一个模型服务可以给不同业务方提供不同提示模板只要在请求里带一个template_name参数服务端从模板库拉取并渲染即可。这样运营人员改提示也完全不用动代码只要模板文件允许热加载。6.3 工具调用的编排思路从生成文本到生成行为最后提一个进阶方向让模型不只是吐文本而是根据用户意图调用外部工具。比如用户问帮我查一下今天的天气模型先判断需要调用天气查询接口生成一个结构化调用参数然后服务端执行查询再把结果回填给模型生成最终回复。这个模式业界通常叫函数回调或工具调用核心是把模型直接返回文本变成模型先规划服务端再执行。FastAPI 在这套编排里的角色是可靠的执行层接收意图、路由到工具、拿回结果、组织下一次模型请求。你可以先定义一个统一的工具函数注册表比如用字典把工具函数和对应 schema 关联起来然后在流式生成前先跑一轮意图解析请求。这部分逻辑虽然比单纯的请求-应答复杂但它能真正把生成式服务从玩具变成业务工具。等你的第一版跑通之后再往这个方向演进不会白费功夫。我在实际生成式服务项目里体会最深的是流式接口改造、并发控制、测试模拟这三件事如果不在一开始就做好后面返工的成本会成倍放大。尤其是信号量的位置和错误事件的协议改起来牵扯前端联调牵一发动全身。所以这一篇的每段代码建议你都拿到本地跑一遍先跑通过再根据实际模型调整参数。等请求真正能被稳定地流式处理、并发控制得心应手的时候你的服务就从一个能跑通代码的教学项目变成一个有底气交给用户的正式服务了。