零基础入门python66：FastAPI AI标题、摘要和标签

零基础入门python66：FastAPI AI标题、摘要和标签

一、上一篇课后练习讲解

上一篇练习围绕“安全调用大模型API”。参考做法是先运行上一篇的测试，再用一个成功请求和一个失败请求验证边界；本篇在同一项目上增加新能力。

上一篇课后练习完整答案

上一篇练习已经落实到完整文件，运行下面代码可以观察本篇要求的成功和失败状态；数据库写入全部放在明确事务边界内。
答案要点：safe_generate 截断脱敏 prompt，用 asyncio.wait_for 控制超时；超时只把 ai_status 置 failed，不回滚正文。

完整答案文件：app/ai.py

完整参考答案文件

完整文件：app/ai.py

importasyncioasyncdefsafe_generate(client,prompt,timeout=10):prompt=prompt[:4000]try:result=awaitasyncio.wait_for(client.generate(prompt),timeout)return{"status":"ok","text":result.text}exceptasyncio.TimeoutError:return{"status":"failed","error":"ai_timeout"}

验收：运行项目测试（FastAPI/Flask 使用 python -m pytest -q，Django 使用 python manage.py test），再按本文 curl 或 Docker 命令检查预期状态码。

完整参考答案文件

下面是交付项目中真实存在的完整文件 fastapi-blog/app/ai.py。它覆盖本篇新增逻辑以及前文已经完成的依赖代码；复制单个函数会丢失上下文，因此这里提供整份文件。

fromtypingimportProtocolfromfastapiimportAPIRouter,Dependsfrom.dependenciesimportget_current_userfrom.modelsimportUserfrom.schemasimportWritingRequestclassAIClient(Protocol):defsuggest(self,content:str)->dict:...classLocalAIClient:defsuggest(self,content):return{'title':content[:20],'summary':content[:80],'tags':['Python']}defget_ai_client():returnLocalAIClient()router=APIRouter(prefix='/api/ai',tags=['ai'])@router.post('/writing-assistant')defwriting_assistant(body:WritingRequest,user:User=Depends(get_current_user),client:AIClient=Depends(get_ai_client)):returnclient.suggest(body.content)

正文长度限制控制成本；邮箱、token 和内部提示词不能写进日志；模型输出永远是不可信输入，不能执行其中的 SQL、HTML 或 Python。

二、本篇要解决的真实问题

把 AI 输出解析为结构化 schema，校验标题长度和标签数量；解析失败返回可追踪错误并写审计日志。

三、请求是怎样走完整条链路的

客户端请求 → 路由匹配 → Pydantic 校验 → Depends 注入用户/Session → SQLAlchemy 查询或业务服务 → 提交事务 → 响应模型序列化。每一步都有明确责任，排错时按这个顺序定位。

四、先看一个最小代码片段

timeout_seconds=10# 外部服务必须有超时，失败时返回可解释的降级结果

这里的注释不是装饰：它说明数据从哪里来、为什么不能相信客户端，以及失败时系统应保持什么状态。

五、完整文件代码

app/ai.py（当前阶段完整文件）

fromtypingimportProtocolfromfastapiimportAPIRouter,Dependsfrom.dependenciesimportget_current_userfrom.modelsimportUserfrom.schemasimportWritingRequestclassAIClient(Protocol):defsuggest(self,content:str)->dict:...classLocalAIClient:defsuggest(self,content):return{'title':content[:20],'summary':content[:80],'tags':['Python']}defget_ai_client():returnLocalAIClient()router=APIRouter(prefix='/api/ai',tags=['ai'])@router.post('/writing-assistant')defwriting_assistant(body:WritingRequest,user:User=Depends(get_current_user),client:AIClient=Depends(get_ai_client)):returnclient.suggest(body.content)

阅读方法：先找路由函数，再找它的 Depends、输入 schema、数据库操作和 response_model；这五处合起来才是一个功能。

app/schemas.py（当前阶段完整文件）

fromdatetimeimportdatetimefrompydanticimportBaseModel,ConfigDict,EmailStr,Field,field_validatorclassUserCreate(BaseModel):email:EmailStr password:str=Field(min_length=8,max_length=128)classUserRead(BaseModel):model_config=ConfigDict(from_attributes=True)id:intemail:strclassToken(BaseModel):access_token:strtoken_type:str='bearer'classArticleCreate(BaseModel):title:str=Field(min_length=2,max_length=120)content:str=Field(min_length=10,max_length=20000)@field_validator('title')@classmethoddefclean_title(cls,value):returnvalue.strip()classArticleRead(ArticleCreate):model_config=ConfigDict(from_attributes=True)id:intauthor_id:intcreated_at:datetimeclassCommentCreate(BaseModel):content:str=Field(min_length=1,max_length=500)classWritingRequest(BaseModel):content:str=Field(min_length=20,max_length=10000)

阅读方法：先找路由函数，再找它的 Depends、输入 schema、数据库操作和 response_model；这五处合起来才是一个功能。

六、安装、启动与验收

..\..\.venv\Scripts\python.exe-m pip install-r requirements.txt..\..\.venv\Scripts\python.exe-m pytest-q..\..\.venv\Scripts\python.exe-m uvicorn app.main:app--reload

打开/docs，按顺序完成注册、登录、创建文章、分页查询、修改删除、评论点赞。错误请求必须看到明确状态码：校验错误 422，未登录 401，无权限 403，重复点赞 409。测试应全部通过。

七、本篇验收清单

• 能指出输入校验发生在哪个 schema。
• 能解释 Session 何时提交、何时回滚。
• 能用第二个用户验证资源隔离。
• 能复现一个失败场景并说明原因。

八、课后练习

围绕“AI标题、摘要和标签”新增一个测试用例，写出请求、预期响应和断言；下一篇开头会给出参考实现，并继续使用本项目。

五、AI 标题、摘要和标签：结果服务于编辑工作流

流程是正文提交、模型建议、字段校验、用户确认、最终保存。模型建议默认不覆盖数据库。

@router.post("/writing-assistant")defwriting_assistant(body:WritingRequest,user:User=Depends(get_current_user),client:AIClient=Depends(get_ai_client)):suggestion=client.suggest(body.content)return{"requested_by":user.id,"title":suggestion["title"].strip(),"summary":suggestion["summary"].strip(),"tags":list(dict.fromkeys(suggestion.get("tags",[])))[:5]}

标签去重保留顺序，最多五个。对同一正文可按内容哈希缓存，但 key 要包含提示词版本。让 FakeAI 返回空标题或 20 个标签，服务端应拒绝或降级，而不是保存脏数据。

六、当前项目中的完整文件（对照阅读）

七、AI 输出的人工确认点

把 AI 结果展示为草稿，用户点击确认后才写入 articles。保存时再次检查标题长度、标签数量和当前用户权限，因为客户端可能篡改确认请求。模型输出的 Markdown 不能直接当 HTML 渲染，先做安全过滤，防止脚本注入。

可以把 prompt_version、model_name 和 latency_ms 保存到审计表，未来才有可能比较不同提示词的质量和成本。

八、补充代码文件

九、质量评估不能只看格式

为 AI 建立人工抽样表：标题是否概括正文、摘要是否出现正文没有的事实、标签是否过宽、是否包含敏感信息。每周抽样记录通过率和平均延迟，提示词升级前后对比同一批文章。

模型输出先用 Pydantic 校验，再做业务过滤；例如标签去重、禁止空标题、限制 Markdown 长度。保存用户确认后的结果，并保留原始正文 hash，出现投诉时能够复现当时的 prompt_version。

本篇结束：完整模块文件

本节不是代码片段，而是本篇结束时该模块的完整版本。请先备份旧文件，再整体替换；替换后重新运行本篇命令和测试。阅读时重点看本篇新增的函数、事务边界和错误处理，未涉及的代码先不要自行删减。

app/ai.py

fromtypingimportProtocolfromfastapiimportAPIRouter,Dependsfrom.dependenciesimportget_current_userfrom.modelsimportUserfrom.schemasimportWritingRequestclassAIClient(Protocol):defsuggest(self,content:str)->dict:...classLocalAIClient:defsuggest(self,content):return{'title':content[:20],'summary':content[:80],'tags':['Python']}defget_ai_client():returnLocalAIClient()router=APIRouter(prefix='/api/ai',tags=['ai'])@router.post('/writing-assistant')defwriting_assistant(body:WritingRequest,user:User=Depends(get_current_user),client:AIClient=Depends(get_ai_client)):returnclient.suggest(body.content)