基于AI agent的童话编剧与绘本生成器（二）从脚手架内存到持久化与依赖注入

项目最初的后端脚手架FastAPI 路由拆分、故事创建与 LangChain 对接、静态资源占位接口等由另一位团队成员编写为团队快速完成前后端联调提供了基础。本次完成了数据根路径、SQLite 持久化、Depends依赖注入等工作是在该既有代码上的增量优化即补齐存储与工程边界我们对早期「进程内字典」等方案的反思并非否定当时「联调优先」的取舍而是说明进入现阶段后为何要再向前迈一步。技术栈FastAPI、Pydantic、SQLite。背景团队在「AI 童话 / 绘本」项目中迭代后端时希望先把数据可靠性和扩展边界立住。本文记录这一阶段的技术判断与几处特色代码的设计思路。一、问题意识脚手架阶段的「隐债」早期接口为了快速联调前端往往采用进程内全局字典保存story_id → 详情的映射。这在演示时非常高效但会带来几类结构性问题生命周期与数据生命周期绑定服务一重启用户刚生成的绘本在服务端「消失」与产品上的「历史作品」「再次打开」诉求冲突。容量与运维不可控内存无限增长或需要手写淘汰策略没有单一落点做备份与迁移。测试与替换成本高路由文件里直接new Service()、直接操作 dict后续若要换 PostgreSQL、或注入 Mock都要改路由实现本身违反「稳定边界」。我的目标不是一步到位上 ORM / 微服务而是用最小表结构 明确依赖入口把「数据从哪来」从路由里抽出去同时让**工作目录cwd**不再悄悄影响数据文件位置——这是很多 Python Web 项目在本地「能跑」、部署「找不到库文件」的根源。二、设计选择一固定「数据根」相对backend解析若用相对路径./data在不同入口启动 uvicorn在仓库根、在backend目录、或由 IDE 启动时实际落盘位置会漂移。我的做法是用当前文件位置反推backend目录再拼接配置的data_dir。defdata_root()-Path:解析数据根目录相对路径相对于 backend 目录。pPath(settings.data_dir)ifp.is_absolute():returnp# backend/app/core/paths.py - parents[2] backendbackend_dirPath(__file__).resolve().parents[2]returnbackend_dir/p为什么不用Path.cwd()cwd 是进程属性与「代码安装在哪」无必然关系框架层应锚定代码树而不是「用户从哪敲命令」。为什么仍支持绝对路径容器或挂载盘场景下运维可能直接给/var/lib/app/data此时不应再强行相对backend。代价若未来包被安装为 site-packages 里的 egg锚点语义会变化当前实训 / 单体仓库场景下这是可接受的权衡。配合应用lifespan启动时创建images、exports、audio、fonts等子目录并初始化库表——这样后续无论是 PDF 还是生图写盘都共享同一套「目录契约」减少各模块各自mkdir的重复与竞态。asynccontextmanagerasyncdeflifespan(app:FastAPI):fromapp.api.depsimportget_story_repository rootdata_root()forsubin(images,exports,audio,fonts):(root/sub).mkdir(parentsTrue,exist_okTrue)get_story_repository().ensure_schema()logging.basicConfig(levellogging.INFO,format%(asctime)s %(levelname)s %(name)s: %(message)s,)logger.info(startup data_root%s,root)yieldlogger.info(shutdown)三、设计选择二JSON 整包落库 保留最近 N 条我没有先设计复杂的「故事表 页表 外键」关系模型而是把StoryDetailResponse序列化成 JSON存入 SQLite。理由很务实与 Pydantic 模型round-trip简单减少手写 ORM 映射与迁移成本当前读路径以「按story_id取整本」为主列表页只需要摘要字段可从 JSON 里解析story小节即可。defsave(self,detail:StoryDetailResponse)-None:payloaddetail.model_dump_json()created_atdatetime.now(UTC).isoformat()story_iddetail.story.story_idwithself._connect()asconn:conn.execute( INSERT INTO stories (story_id, payload_json, created_at) VALUES (?, ?, ?) ON CONFLICT(story_id) DO UPDATE SET payload_json excluded.payload_json, created_at excluded.created_at ,(story_id,payload,created_at),)conn.commit()self._prune_older_than(conn,keep20)def_prune_older_than(self,conn:sqlite3.Connection,*,keep:int)-None:curconn.execute(SELECT story_id FROM stories ORDER BY datetime(created_at) DESC)rowscur.fetchall()iflen(rows)keep:returndrop_ids[r[story_id]forrinrows[keep:]]conn.executemany(DELETE FROM stories WHERE story_id ?,[(i,)foriindrop_ids])conn.commit()ON CONFLICT ... DO UPDATE为未来「同 ID 重新生成 / 覆盖写」留口避免插入主键冲突时整条链路失败。裁剪策略放在同一连接的事务后先写入再按时间排序删除多余行逻辑直观若并发升高可再改为「计数触发异步清理」。取舍JSON 整包不利于按页 SQL 查询与索引优化当产品需要「全文检索页级文本」时应增量引入页表或同步索引而不是一开始就过度设计。四、设计选择三FastAPIDepends与单例工厂路由层只描述「需要什么」不描述「怎么构造」。仓储与生成服务通过deps.py暴露工厂函数内部用模块级单例避免无谓重复构造。defget_story_repository()-StorySqliteRepository:global_story_repoif_story_repoisNone:fromapp.core.pathsimportdata_root _story_repoStorySqliteRepository(data_root()/app.db)return_story_repodefget_generation_service()-StoryGenerationService:global_generation_serviceif_generation_serviceisNone:_generation_serviceStoryGenerationService()return_generation_service路由侧使用Annotated[..., Depends(...)]类型检查器与 IDE 都能识别依赖类型router.get(,response_modellist[StoryListItem],summary最近生成记录)asyncdeflist_stories(repo:Annotated[StorySqliteRepository,Depends(get_story_repository)],)-list[StoryListItem]:returnrepo.list_recent(limit20)router.post(,response_modelCreateStoryResponse,summary创建故事)asyncdefcreate_story(payload:CreateStoryRequest,repo:Annotated[StorySqliteRepository,Depends(get_story_repository)],gen:Annotated[StoryGenerationService,Depends(get_generation_service)],)-CreateStoryResponse:story,pagesgen.generate(payload)detailStoryDetailResponse(storystory,pagespages)repo.save(detail)returnCreateStoryResponse(storystory,pagespages)单测时可替换get_story_repository的返回值而不必 import 整个路由模块去 patch 全局 dict。未来若接入多租户可在Depends链上增加「当前用户 → 选择不同 DB 路径」的一层而路由签名基本不变。五、其他改动生成服务与 LLM 配置的「单一入口」生成逻辑原先直接读settings.llm_*与LLMProvider重复。现在由StoryGenerationService持有LLMProvider「是否可调用 LangChain」与「用什么 key/base_url/model」同源避免两处判断漂移。classStoryGenerationService:Story generation with LangChain fallback mock.def__init__(self)-None:self._llmLLMProvider()def_is_langchain_ready(self)-bool:returnbool(ChatOpenAIandself._llm.is_configured())这与「图像、TTS 也用 Provider 封装」的风格一致后续把密钥来源换成配置中心时改动面集中。六、结语这次改动刻意控制在增量优化没有引入重型 ORM也没有重写前端契约但把数据根、持久化、依赖入口三件事说清楚了。下一阶段若上异步任务队列或多 Agent 流水线这些边界会成为自然的挂载点——而不是在路由文件里继续堆全局状态。再次感谢队友搭好的首版后端骨架使本文所述的优化能始终「贴着真实联调场景」推进而不是空中楼阁。