FastAPI构建生产级AI应用网关实战指南

1. 项目概述：为什么“从零搭建第一个AI应用”这件事，比你想象中更值得认真对待

我带过不少转行做AI应用开发的朋友，有Java后端干了八年突然想切入大模型赛道的，也有刚毕业的Python新手，还有做UI设计多年想用AI重构工作流的。他们问得最多的问题不是“FastAPI怎么写”，而是：“我照着教程跑通了Hello World，可接下来呢？我的AI应用到底该长成什么样？它和普通Web服务到底差在哪？”——这个问题，恰恰是标题里“从零搭建你的第一个AI应用”最核心的潜台词。

这根本不是一次简单的框架安装教学。FastAPI本身只是一把趁手的刀，而LLM API调用才是你要切的那块肉。真正卡住90%初学者的，从来不是语法错误，而是对“AI应用”这个概念的认知断层：它既不是纯前端调用一个API Key就能完事的玩具，也不是要你从头训练一个大模型的科研项目；它处在中间那个极其关键、也最容易被忽略的“工程化接口层”——既要理解LLM的输入输出边界（比如token限制、流式响应、系统提示词的嵌入逻辑），又要扛住真实用户并发请求的压力（比如超时控制、重试策略、错误降级），还得让非技术同事能看懂、能调试、能改提示词。这三件事叠加起来，就是标题里那个“完整教程”真正的分量。

所以这篇内容，我不会从pip install fastapi开始讲起，而是先带你站在一个已经上线过3个AI内部工具的老手视角，看清整个链条上每个环节的真实职责。你会看到：为什么我们坚持用uvicorn --reload只用于开发，而生产必须加--workers 4；为什么Pydantic BaseModel里一个str | None = None的字段声明，背后是整整两轮用户反馈才定下来的容错设计；为什么健康检查接口/health要拆出verbose=true参数，而不是简单返回{"status":"ok"}。这些细节没有一行代码高深，但每一条都来自线上报错日志、用户投诉截图和凌晨三点的服务器告警。它们共同构成了“第一个AI应用”能活过一周、而不是仅在本地跑通五分钟的关键骨架。

关键词自然融入：AI应用开发、FastAPI、LLM API、教程——这些不是标签，而是你接下来每一分钟都要亲手触摸的真实组件。适合谁？适合所有已经写过print("Hello World")、但还没在浏览器里输入过http://localhost:8000/docs并点下“Execute”按钮的人；也适合那些在招聘JD里反复看到“熟悉FastAPI+LLM集成”的工程师，想搞清楚这六个字背后到底要填多少坑。这不是速成班，而是一份你部署第一个AI服务时，会反复翻出来对照的施工图纸。

2. 整体架构设计与选型逻辑：为什么是FastAPI，而不是Flask、Django或Node.js？

2.1 FastAPI的不可替代性：性能、类型安全与文档自动生成的三角闭环

很多人选FastAPI，是因为网上都说“它快”。但快到什么程度？快是否解决得了AI应用的核心痛点？我们来算一笔账。假设你调用一个主流LLM API（比如OpenAI的gpt-3.5-turbo），平均响应时间是1.2秒。如果后端框架本身处理请求要耗掉300毫秒，那用户感知延迟就是1.5秒；如果框架优化到50毫秒，感知延迟就是1.25秒——省下的250毫秒，在用户连续提问的场景下，就是体验的分水岭。FastAPI的底层依赖Starlette，其ASGI实现比Flask的WSGI快3-5倍，实测在同等硬件下，QPS（每秒查询数）能从Flask的1200提升到FastAPI的4800。但这只是表象。

真正的杀手锏在于类型驱动的自动校验与文档生成。举个具体例子：LLM API的输入通常包含messages数组（含role和content字段）、model名称、temperature浮点数、max_tokens整数等。用Flask写，你得手动写if not isinstance(data.get('temperature'), float)，再写if data['temperature'] < 0 or data['temperature'] > 2.0，最后还要拼接错误提示JSON。而FastAPI只需定义一个Pydantic模型：

from pydantic import BaseModel, Field from typing import List, Dict, Optional class LLMRequest(BaseModel): messages: List[Dict[str, str]] = Field(..., min_items=1) model: str = Field(default="gpt-3.5-turbo", pattern=r"^gpt-[0-9.-]+$") temperature: float = Field(default=0.7, ge=0.0, le=2.0) max_tokens: int = Field(default=1024, ge=1, le=4096)

这段代码同时完成了三件事：① 自动校验messages非空、temperature在0-2之间、model符合正则；② 自动生成OpenAPI规范，让Swagger UI能渲染出带输入示例、参数说明、错误码的交互式页面；③ 为IDE提供精准类型提示，写request.temperature.时直接弹出.ge.le等方法。这省下的不是代码行数，而是每次修改API时重新写校验逻辑、更新文档、同步前端Mock数据的重复劳动。我见过一个团队，用Flask开发LLM接口，光是维护参数校验和文档就占了后端30%工时；切换FastAPI后，这部分时间归零，且前端联调周期缩短了60%。

2.2 为什么不是Django？——轻量级AI服务不需要“全栈重装”

Django的ORM、Admin后台、用户认证系统，对AI应用开发是典型的“杀鸡用牛刀”。一个LLM API网关的核心职责只有三件：接收请求→转发给LLM→返回结果。它不需要管理数据库里的用户表，不需要渲染HTML模板，更不需要复杂的权限分级（API Key鉴权足够）。强行套Django，你会陷入无尽的配置：settings.py里堆满INSTALLED_APPS，urls.py里写满path('api/v1/', include('llm_api.urls'))，还要为每个API写APIView类继承。而FastAPI的路由是函数级的，@app.post("/chat")一行就绑定好，逻辑全部在函数体内，干净利落。更重要的是，Django的同步阻塞模型在处理LLM这种长IO操作时，会严重拖慢并发能力——你得额外引入Celery或ASGI改造，而FastAPI原生支持async def，await client.post()调用LLM API时，线程能立刻释放去处理其他请求。

2.3 为什么不是Node.js？——Python生态对AI开发的绝对统治力

Node.js的Express确实快，生态也成熟。但当你需要对接Hugging Face的Transformers库、LangChain的链式调用、或是自定义的向量检索模块时，Python的SDK丰富度和文档质量是碾压级的。OpenAI官方SDK、Anthropic的Claude SDK、Google Vertex AI SDK，全部优先支持Python。更关键的是调试体验：在Jupyter里调试一个LLM调用链，print(response.choices[0].message.content)直接看到结果；而在Node.js里，你得处理Promise链、console.log(JSON.stringify(res.data))，还容易漏掉res.data?.choices?.[0]?.message?.content的深层空值判断。我曾帮一个Node.js团队迁移LLM服务，他们花两天时间写的错误处理逻辑（网络超时、token截断、rate limit重试），在Python里用tenacity库加三行装饰器就搞定。这不是语言优劣，而是生态适配——AI应用开发的主战场在Python，FastAPI是它最顺手的Web接口层。

2.4 架构图解：一个生产级AI应用的最小可行结构

┌─────────────────┐ ┌───────────────────┐ ┌───────────────────────┐ │ 用户端 │───▶│ FastAPI 网关 │───▶│ 第三方 LLM API │ │ (Web/移动端) │ │ • 请求路由 & 校验 │ │ (OpenAI / Anthropic等) │ └─────────────────┘ │ • 提示词预处理 │ └───────────────────────┘ │ • 流式响应封装 │ │ • 错误统一降级 │ └───────────────────┘ ▲ │ ┌───────────────────┐ │ 运维监控层 │ │ • /health 健康检查 │ │ • 日志结构化输出 │ │ • Prometheus指标 │ └───────────────────┘

这个结构里，FastAPI网关是绝对核心，它不碰业务逻辑（如知识库检索、多步Agent编排），只做三件事：协议转换（把HTTP请求转成LLM API所需的JSON）、流量管控（限流、熔断、重试）、体验增强（流式SSE响应、错误友好化）。所有复杂逻辑都应下沉到独立服务，保持网关轻量化。这也是为什么标题强调“从零搭建第一个”，它必须是可复制、可演进的起点，而不是一个包罗万象的巨石应用。

3. 核心细节解析与实操要点：从Hello World到生产可用的七道坎

3.1 环境隔离：为什么pip install "fastapi[all]"是唯一推荐的安装方式

很多教程教pip install fastapi，然后单独pip install uvicorn。这是危险的。FastAPI的[all]选项不仅包含Uvicorn，还强制安装了pydantic最新兼容版、jinja2（用于模板）、python-multipart（文件上传）、aiofiles（异步文件IO）等一揽子依赖。我踩过的坑：某次升级FastAPI到0.110，pydantic版本没同步，导致BaseModel的Field(default_factory=list)报TypeError: 'list' object is not callable——因为旧版Pydantic不支持default_factory。而[all]确保所有组件版本锁死，避免这种隐性冲突。

实操步骤：

1. 创建项目目录：mkdir my-first-ai-app && cd my-first-ai-app
2. 初始化虚拟环境：python -m venv venv && source venv/bin/activate（Mac/Linux）或venv\Scripts\activate（Windows）
3. 关键一步：pip install "fastapi[all]"—— 注意引号，防止shell解析空格
4. 验证安装：pip list | grep -E "(fastapi|uvicorn|pydantic)"应显示类似：

   fastapi 0.111.0 pydantic 2.7.4 uvicorn 0.29.0

提示：永远不要在全局Python环境装包。虚拟环境是AI应用开发的生命线——不同项目可能依赖不同版本的Transformers或LangChain，混装必崩。

3.2 路由设计哲学：GET/POST/PUT的语义边界，如何影响LLM调用的健壮性

初学者常犯的错误：把所有LLM请求都塞进@app.post("/chat")。这违反RESTful原则，也埋下隐患。正确做法是按操作语义拆分：

• POST /chat/completions：标准OpenAI兼容接口，接收messages数组，返回完整响应。这是主干道。
• GET /models：获取当前支持的模型列表（如["gpt-3.5-turbo", "claude-3-haiku"]）。用GET因为它是幂等的、可缓存的。
• POST /chat/stream：专门处理SSE（Server-Sent Events）流式响应。POST因为流式需要客户端明确发起请求，且不能被CDN缓存。
• PUT /chat/prompt：更新系统提示词（System Prompt）。用PUT因为它是幂等的更新操作。

为什么这么重要？以/models为例，如果用POST，前端每次刷新页面都要发一次请求；而GET接口可被浏览器缓存、CDN缓存，甚至前端用useSWR自动做增量更新。再比如流式接口，如果和普通接口共用一个路由，你得在代码里判断Accept: text/event-stream头，逻辑混乱；独立路由则清晰分离关注点。

3.3 Pydantic模型：不只是校验，更是API契约与前端协作的基石

LLM API的输入结构远比想象中复杂。一个看似简单的messages字段，实际需约束：

• 至少1条消息（min_items=1）
• 每条消息必须有role（"user"/"assistant"/"system"）和content（非空字符串）
• role值必须是枚举（Literal["user", "assistant", "system"]）
• content长度不能超过模型限制（如gpt-3.5-turbo是16K token，但中文字符数≈token数×1.3，需预留缓冲）

完整模型定义：

from pydantic import BaseModel, Field, validator from typing import List, Literal, Optional, Dict, Any class Message(BaseModel): role: Literal["user", "assistant", "system"] content: str = Field(..., min_length=1, max_length=8000) class LLMRequest(BaseModel): messages: List[Message] = Field(..., min_items=1, max_items=100) model: str = Field(..., pattern=r"^gpt-[0-9.-]+|claude-.*|gemini-.*$") temperature: float = Field(default=0.7, ge=0.0, le=2.0) max_tokens: int = Field(default=1024, ge=1, le=4096) stream: bool = Field(default=False) @validator('messages') def validate_messages(cls, v): # 系统消息只能在开头 if len(v) > 0 and v[0].role != "system": raise ValueError("First message must be system role") return v

这个模型带来的好处是：① Swagger UI自动生成带枚举选项、范围提示的表单；② 前端TypeScript可直接用zod库生成对应Schema，保证前后端类型100%一致；③ 当用户传入{"role": "bot", "content": ""}时，FastAPI自动返回422 Unprocessable Entity及详细错误路径messages/0/role，前端可精确定位问题。

3.4 异常处理：为什么HTTPException比return {"error": "xxx"}更能救你的命

LLM调用失败有无数种可能：网络超时、API Key无效、模型不存在、token超限、服务端503。如果都用return {"error": "xxx"}，前端无法区分是用户输错内容（400）还是服务崩了（500），监控系统也抓不到错误码。正确姿势是分层抛出HTTPException：

from fastapi import HTTPException import httpx @app.post("/chat/completions") async def chat_completions(request: LLMRequest): try: # 1. 业务校验（用户可控错误） if len(request.messages[-1].content) > 5000: raise HTTPException( status_code=400, detail="User message too long. Max 5000 characters." ) # 2. 调用LLM（外部依赖错误） async with httpx.AsyncClient() as client: response = await client.post( "https://api.openai.com/v1/chat/completions", headers={"Authorization": f"Bearer {OPENAI_API_KEY}"}, json=request.dict(exclude={"stream"}), timeout=30.0 # 关键！必须设超时 ) response.raise_for_status() # 触发4xx/5xx异常 return response.json() # 3. 外部服务错误（映射为503） except httpx.TimeoutException: raise HTTPException( status_code=503, detail="LLM service timeout. Please try again later." ) except httpx.HTTPStatusError as e: # 映射OpenAI错误码 if e.response.status_code == 401: raise HTTPException(status_code=401, detail="Invalid API key") elif e.response.status_code == 429: raise HTTPException(status_code=429, detail="Rate limit exceeded") else: raise HTTPException( status_code=502, detail=f"Bad gateway: {e.response.text}" ) # 4. 未预期错误（500兜底） except Exception as e: logger.error(f"Unexpected error: {e}", exc_info=True) raise HTTPException( status_code=500, detail="Internal server error. Contact admin." )

这样做的价值：① Nginx或云WAF可根据状态码做不同路由（如429重定向到限流页）；② Prometheus监控可统计各错误码分布，快速定位是用户问题还是服务问题；③ 前端fetch可针对response.status做差异化处理（400弹窗提示，503显示维护公告）。

3.5 日志与监控：没有日志的AI应用，就像没有刹车的汽车

FastAPI默认日志太简陋。生产环境必须结构化日志，记录关键上下文：请求ID、模型名、输入长度、响应时间、token用量。我用structlog替代原生日志：

import structlog from fastapi import Request # 配置structlog structlog.configure( processors=[ structlog.stdlib.filter_by_level, structlog.stdlib.add_logger_name, structlog.stdlib.add_log_level, structlog.stdlib.PositionalArgumentsFormatter(), structlog.processors.TimeStamper(fmt="iso"), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.UnicodeDecoder(), structlog.processors.JSONRenderer() # 输出JSON格式 ], context_class=dict, logger_factory=structlog.stdlib.LoggerFactory(), ) logger = structlog.get_logger() @app.middleware("http") async def log_requests(request: Request, call_next): # 生成唯一请求ID request_id = str(uuid.uuid4()) logger = logger.bind(request_id=request_id) start_time = time.time() try: response = await call_next(request) process_time = (time.time() - start_time) * 1000 logger.info( "Request completed", method=request.method, url=str(request.url), status_code=response.status_code, process_time_ms=round(process_time, 2) ) return response except Exception as e: process_time = (time.time() - start_time) * 1000 logger.error( "Request failed", method=request.method, url=str(request.url), process_time_ms=round(process_time, 2), error=str(e) ) raise

配合/health接口，你就能在Grafana里画出实时图表：QPS、P95延迟、错误率。当/chat/completions的503错误突增，结合日志查LLM service timeout，立刻知道是OpenAI服务抖动，而非自己代码问题。

4. 实操过程与核心环节实现：手把手搭建可运行的AI网关

4.1 项目初始化：从零创建可部署的目录结构

拒绝“一个py文件走天下”。生产级项目必须有清晰分层。我的标准结构：

my-first-ai-app/ ├── app/ # FastAPI应用核心 │ ├── __init__.py │ ├── main.py # ASGI入口，app实例化 │ ├── api/ # API路由定义 │ │ ├── __init__.py │ │ ├── v1/ # 版本化路由 │ │ │ ├── __init__.py │ │ │ ├── chat.py # /chat/* 路由 │ │ │ └── health.py # /health 路由 │ ├── core/ # 配置、依赖注入 │ │ ├── __init__.py │ │ ├── config.py # 环境变量加载 │ │ └── dependencies.py # 数据库/LLM客户端依赖 │ └── schemas/ # Pydantic模型 │ ├── __init__.py │ └── chat.py # LLMRequest/Response定义 ├── tests/ # 单元测试 ├── requirements.txt ├── .env # 环境变量（API Key等） └── README.md

创建命令：

mkdir -p app/{api/v1,core,schemas} tests touch app/{__init__.py,main.py} touch app/api/{__init__.py,v1/__init__.py,v1/chat.py,v1/health.py} touch app/core/{__init__.py,config.py,dependencies.py} touch app/schemas/{__init__.py,chat.py} echo "fastapi[all]==0.111.0" > requirements.txt

注意：.env文件必须加入.gitignore！API Key绝不能提交到Git。用python-dotenv加载：

# app/core/config.py from pydantic import BaseSettings class Settings(BaseSettings): OPENAI_API_KEY: str OPENAI_BASE_URL: str = "https://api.openai.com/v1" DEBUG: bool = False class Config: env_file = ".env" settings = Settings()

4.2 核心路由实现：/chat/completions的完整代码与逐行注释

app/api/v1/chat.py：

from fastapi import APIRouter, Depends, HTTPException, status from httpx import AsyncClient, Timeout from typing import Any, Dict, List, Optional import asyncio import json from app.core.config import settings from app.schemas.chat import LLMRequest, LLMResponse from app.core.dependencies import get_llm_client router = APIRouter(prefix="/v1", tags=["Chat"]) @router.post( "/chat/completions", response_model=LLMResponse, summary="OpenAI兼容的聊天补全接口", description="接收messages数组，返回LLM生成结果。支持gpt-3.5-turbo等模型。", responses={ 400: {"description": "请求参数错误，如消息为空或模型不支持"}, 401: {"description": "API Key无效"}, 429: {"description": "请求频率超限"}, 503: {"description": "LLM服务不可用"} } ) async def chat_completions( request: LLMRequest, client: AsyncClient = Depends(get_llm_client) ): """ 主要逻辑： 1. 对request做业务校验（已在Pydantic模型中完成） 2. 构造LLM API请求体（排除stream参数，因本接口不支持流式） 3. 调用LLM服务，设置30秒超时 4. 处理响应，映射为LLMResponse模型 """ # 步骤1：业务校验（示例：禁止系统消息在非首位） if request.messages and request.messages[0].role != "system": raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="First message must be system role" ) # 步骤2：构造请求体，移除stream参数（本接口不处理流式） payload = request.dict(exclude={"stream"}) # 步骤3：调用LLM API try: response = await client.post( url=f"{settings.OPENAI_BASE_URL}/chat/completions", headers={"Authorization": f"Bearer {settings.OPENAI_API_KEY}"}, json=payload, timeout=Timeout(30.0, read=30.0, connect=10.0) # 细粒度超时 ) response.raise_for_status() # 抛出4xx/5xx异常 # 步骤4：解析响应，映射为Pydantic模型 result = response.json() return LLMResponse(**result) # 步骤5：错误分类处理 except asyncio.TimeoutError: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="LLM service timeout. Please try again later." ) except httpx.HTTPStatusError as e: if e.response.status_code == 401: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key" ) elif e.response.status_code == 429: raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="Rate limit exceeded for your API key" ) else: # 兜底：记录原始错误，返回通用502 error_detail = e.response.text[:200] # 截取前200字符防日志爆炸 raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Bad gateway: {error_detail}" ) except Exception as e: # 未预期错误 raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error" )

关键点解析：

• response_model=LLMResponse：FastAPI自动序列化返回值，并在Swagger中生成响应示例。
• timeout=Timeout(30.0, read=30.0, connect=10.0)：连接超时10秒，读取超时30秒，总超时30秒。避免LLM服务挂起时整个进程卡死。
• Depends(get_llm_client)：依赖注入，确保每次请求复用同一个AsyncClient实例（连接池复用），而非每次都新建。

4.3 依赖注入：get_llm_client的实现与连接池优化

app/core/dependencies.py：

from httpx import AsyncClient from fastapi import Depends from app.core.config import settings # 全局客户端实例（单例） _llm_client: AsyncClient | None = None def get_llm_client() -> AsyncClient: """ 依赖注入函数，返回复用的AsyncClient。 优势：连接池复用，避免频繁建连开销；自动处理HTTP/2、keep-alive。 """ global _llm_client if _llm_client is None: # 配置连接池：最大连接数100，每个host最多20连接 _llm_client = AsyncClient( limits=httpx.Limits( max_connections=100, max_keepalive_connections=20, keepalive_expiry=60.0 ), timeout=httpx.Timeout(30.0, read=30.0, connect=10.0) ) return _llm_client # 应用启动时创建客户端，关闭时清理 from fastapi import FastAPI def create_app() -> FastAPI: app = FastAPI(title="My First AI App", version="0.1.0") @app.on_event("startup") async def startup(): # 启动时初始化客户端 global _llm_client if _llm_client is None: _llm_client = get_llm_client() @app.on_event("shutdown") async def shutdown(): # 关闭时清理连接 global _llm_client if _llm_client is not None: await _llm_client.aclose() _llm_client = None return app

为什么不用@lru_cache？因为AsyncClient是异步对象，lru_cache不支持异步函数缓存。全局变量+事件钩子是FastAPI官方推荐模式。

4.4 启动与部署：从uvicorn开发到gunicorn+uvicorn生产

开发阶段：

# 安装依赖 pip install -r requirements.txt # 启动（自动重载） uvicorn app.main:app --reload --host 0.0.0.0:8000 --port 8000

生产部署（Linux服务器）：

# 安装gunicorn（WSGI/ASGI进程管理器） pip install gunicorn # 启动命令（4个工作进程，每个进程1个uvicorn worker） gunicorn app.main:app \ --worker-class uvicorn.workers.UvicornWorker \ --workers 4 \ --bind 0.0.0.0:8000 \ --bind 127.0.0.1:8001 \ --worker-tmp-dir /dev/shm \ --access-logfile - \ --error-logfile - \ --log-level info

关键参数说明：

• --workers 4：CPU核心数×2，避免单进程成为瓶颈。
• --bind 0.0.0.0:8000：对外提供服务。
• --bind 127.0.0.1:8001：内部健康检查端口（Nginx可轮询此端口）。
• --worker-tmp-dir /dev/shm：将临时文件放在内存，加速IO。

提示：永远不要用--reload启动生产环境！它会监控文件变化并重启，导致服务中断。

4.5 本地验证：用curl和Swagger UI双重确认

启动服务后，打开http://localhost:8000/docs，你会看到自动生成的交互式文档。点击/v1/chat/completions→Try it out→ 输入：

{ "messages": [ {"role": "system", "content": "你是一个专业的技术文档助手，回答要简洁准确。"}, {"role": "user", "content": "FastAPI和Flask的主要区别是什么？"} ], "model": "gpt-3.5-turbo", "temperature": 0.5 }

点击Execute，看到成功响应即表示通路打通。

命令行验证（更贴近真实调用）：

curl -X POST "http://localhost:8000/v1/chat/completions" \ -H "Content-Type: application/json" \ -d '{ "messages": [{"role": "user", "content": "你好"}], "model": "gpt-3.5-turbo" }'

5. 常见问题与排查技巧实录：那些让你熬夜到三点的真问题

5.1 问题速查表：高频报错与根因定位

5.2 独家避坑技巧：来自线上事故的血泪总结

技巧1：API Key泄露的“双保险”机制
曾经有同事误将.env提交到GitHub，触发GitHub Secret扫描告警。现在所有项目强制执行：
① 在.gitignore中添加*.env、*.key；
② 在CI流程中加入grep -r "sk-" .检测硬编码Key；
③ 使用Vault或AWS Secrets Manager管理生产Key，.env只用于本地开发。

技巧2：LLM响应截断的静默处理
OpenAI的max_tokens是生成上限，但实际返回可能被截断（如token用尽）。用户看到半句话会困惑。解决方案：在LLMResponse模型中增加truncated: bool字段，并在解析时检查response.choices[0].finish_reason == "length"：

class LLMResponse(BaseModel): id: str object: str created: int model: str choices: List[Choice] usage: Usage truncated: bool = False # 新增字段 # 在chat.py中解析后： if result.get("choices") and result["choices"][0].get("finish_reason") == "length": result["truncated"] = True

前端据此显示“内容过长，已截断”提示。

技巧3：流式响应的“心跳保活”
SSE流式响应中，若LLM生成缓慢，客户端可能因超时断开连接。解决方案：在流式接口中定期发送:ping\n\n（SSE心跳）：

@app.post("/chat/stream") async def chat_stream(request: LLMRequest): async with httpx.AsyncClient() as client: async with client.stream( "POST",