基于AI agent的童话编剧与绘本生成器(二)从脚手架内存到持久化与依赖注入
2026/4/20 5:11:14 网站建设 项目流程

项目最初的后端脚手架(FastAPI 路由拆分、故事创建与 LangChain 对接、静态资源占位接口等)由另一位团队成员编写,为团队快速完成前后端联调提供了基础。本次完成了数据根路径、SQLite 持久化、Depends依赖注入等工作,是在该既有代码上的增量优化:即补齐存储与工程边界;我们对早期「进程内字典」等方案的反思,并非否定当时「联调优先」的取舍,而是说明进入现阶段后为何要再向前迈一步。
技术栈:FastAPI、Pydantic、SQLite。
背景:团队在「AI 童话 / 绘本」项目中迭代后端时,希望先把数据可靠性扩展边界立住。本文记录这一阶段的技术判断与几处特色代码的设计思路。


一、问题意识:脚手架阶段的「隐债」

早期接口为了快速联调前端,往往采用进程内全局字典保存story_id → 详情的映射。这在演示时非常高效,但会带来几类结构性问题:

  1. 生命周期与数据生命周期绑定:服务一重启,用户刚生成的绘本在服务端「消失」,与产品上的「历史作品」「再次打开」诉求冲突。
  2. 容量与运维不可控:内存无限增长或需要手写淘汰策略;没有单一落点做备份与迁移。
  3. 测试与替换成本高:路由文件里直接new Service()、直接操作 dict,后续若要换 PostgreSQL、或注入 Mock,都要改路由实现本身,违反「稳定边界」。

我的目标不是一步到位上 ORM / 微服务,而是:用最小表结构 + 明确依赖入口,把「数据从哪来」从路由里抽出去,同时让**工作目录(cwd)**不再悄悄影响数据文件位置——这是很多 Python Web 项目在本地「能跑」、部署「找不到库文件」的根源。


二、设计选择一:固定「数据根」相对backend解析

若用相对路径./data,在不同入口启动 uvicorn(在仓库根、在backend目录、或由 IDE 启动)时,实际落盘位置会漂移。我的做法是:用当前文件位置反推backend目录,再拼接配置的data_dir

defdata_root()->Path:"""解析数据根目录:相对路径相对于 backend 目录。"""p=Path(settings.data_dir)ifp.is_absolute():returnp# backend/app/core/paths.py -> parents[2] == backendbackend_dir=Path(__file__).resolve().parents[2]returnbackend_dir/p
  • 为什么不用Path.cwd():cwd 是进程属性,与「代码安装在哪」无必然关系;框架层应锚定代码树,而不是「用户从哪敲命令」。
  • 为什么仍支持绝对路径:容器或挂载盘场景下,运维可能直接给/var/lib/app/data,此时不应再强行相对backend
  • 代价:若未来包被安装为 site-packages 里的 egg,锚点语义会变化;当前实训 / 单体仓库场景下这是可接受的权衡。

配合应用lifespan,启动时创建imagesexportsaudiofonts等子目录,并初始化库表——这样后续无论是 PDF 还是生图写盘,都共享同一套「目录契约」,减少各模块各自mkdir的重复与竞态。

@asynccontextmanagerasyncdeflifespan(app:FastAPI):fromapp.api.depsimportget_story_repository root=data_root()forsubin("images","exports","audio","fonts"):(root/sub).mkdir(parents=True,exist_ok=True)get_story_repository().ensure_schema()logging.basicConfig(level=logging.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:payload=detail.model_dump_json()created_at=datetime.now(UTC).isoformat()story_id=detail.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,keep=20)def_prune_older_than(self,conn:sqlite3.Connection,*,keep:int)->None:cur=conn.execute("SELECT story_id FROM stories ORDER BY datetime(created_at) DESC")rows=cur.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_repo=StorySqliteRepository(data_root()/"app.db")return_story_repodefget_generation_service()->StoryGenerationService:global_generation_serviceif_generation_serviceisNone:_generation_service=StoryGenerationService()return_generation_service

路由侧使用Annotated[..., Depends(...)],类型检查器与 IDE 都能识别依赖类型:

@router.get("",response_model=list[StoryListItem],summary="最近生成记录")asyncdeflist_stories(repo:Annotated[StorySqliteRepository,Depends(get_story_repository)],)->list[StoryListItem]:returnrepo.list_recent(limit=20)@router.post("",response_model=CreateStoryResponse,summary="创建故事")asyncdefcreate_story(payload:CreateStoryRequest,repo:Annotated[StorySqliteRepository,Depends(get_story_repository)],gen:Annotated[StoryGenerationService,Depends(get_generation_service)],)->CreateStoryResponse:story,pages=gen.generate(payload)detail=StoryDetailResponse(story=story,pages=pages)repo.save(detail)returnCreateStoryResponse(story=story,pages=pages)
  • 单测时可替换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._llm=LLMProvider()def_is_langchain_ready(self)->bool:returnbool(ChatOpenAIandself._llm.is_configured())

这与「图像、TTS 也用 Provider 封装」的风格一致,后续把密钥来源换成配置中心时,改动面集中。


六、结语

这次改动刻意控制在增量优化:没有引入重型 ORM,也没有重写前端契约,但把数据根、持久化、依赖入口三件事说清楚了。下一阶段若上异步任务队列或多 Agent 流水线,这些边界会成为自然的挂载点——而不是在路由文件里继续堆全局状态。
再次感谢队友搭好的首版后端骨架,使本文所述的优化能始终「贴着真实联调场景」推进,而不是空中楼阁。


需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询