先说结论早期所有逻辑写在一个文件里 — 人设、选题、生成、质检、输出全塞在一起改一处牵一发动全身。加个聊天功能要改 4 个文件加个热点功能要在 3 个地方插代码。在 self-media-agent 项目里最终演进为14 个子模块 2 个入口 1 个组装中心src/self_media_agent/ ├── config/ 配置加载 ├── persona/ 人设管理 ├── topic/ 选题管理 ├── content/ 内容生成 ├── quality/ 质检系统 ├── hotspot/ 热点感知 ├── chat/ 对话修改 ├── storage/ 数据存储 ├── llm/ LLM 客户端 ├── cache/ 缓存 ├── task/ 异步任务 ├── analytics/ 数据分析 ├── pipeline/ 流水线编排组装中心 ├── api/ Web 入口 └── cli/ CLI 入口每个模块按职责拆分通过Repository和LLMClient传递依赖在PipelineRunner里组装在app.py或cli/main.py里入口。架构层职责对应代码入口层接收请求调用业务cli/main.pyapi/app.py编排层组装模块控制流程pipeline/runner.py业务层各自独立的业务逻辑persona/topic/content/quality/chat/基础设施层通用能力llm/storage/cache/task/好的架构让加功能像搭积木 — 新建一个模块在编排层接上完事。一、为什么需要模块化单体时代所有逻辑在一个文件agent.py500行 ├── 人设加载YAML 解析 校验 ├── 选题生成调 LLM 解析 JSON ├── 标题生成调 LLM 选最优 ├── 正文生成调 LLM 注入人设 ├── 排版分段 emoji 加粗 ├── 质检敏感词 重复检测 ├── 文件输出写 Markdown └── main 函数串起来单体的问题改一处牵一全身— 改排版逻辑不小心碰到了选题生成的代码无法复用— CLI 想调生成正文的功能只能 import 整个 500 行的文件测试困难— 测质检逻辑要先跑完生成没法单独测加功能要改多处— 加聊天功能改 agent.py加逻辑 改 main加调用 改输出加保存想加对话修改功能 → ① 在 agent.py 里加 chat 函数200行混进去 ② 在 main 里加 if 判断调 chat 还是 generate ③ 修改输出逻辑保存修改记录 ④ 改人设加载加 style_preferences 字段 → 改了 4 个地方每个地方都可能引入 bug模块化的目标加对话修改功能 → ① 新建 chat/ 模块schema.py 逻辑 ② 在 api/routes/ 加 chat.py 路由 ③ 在 app.py 注册路由1行 → 其他模块完全不用改模块化的三个原则单一职责— 每个模块只做一件事显式依赖— 模块间的依赖通过构造函数注入不隐式 import开放扩展— 加功能 加模块 接线不改已有模块二、模块拆分按职责分14 个模块的职责业务层7个— 各管一个业务领域 ├── persona/ 人设定义、CRUD、风格学习 ├── topic/ 选题生成、选题池管理 ├── content/ 内容标题、正文、脚本、排版 ├── quality/ 质检合规、去重、口语化、逻辑 ├── hotspot/ 热点爬取、分析、多平台适配 ├── chat/ 对话会话、修改记录 └── analytics/ 分析发布数据追踪 基础设施层4个— 通用能力 ├── llm/ LLM 客户端调用、Prompt 管理、Token 计算 ├── storage/ 存储内存引擎、文件持久化、仓储封装 ├── cache/ 缓存LRU TTL └── task/ 任务异步任务引擎 编排层1个 └── pipeline/ 流水线组装所有模块控制执行顺序 入口层2个 ├── cli/ CLI 入口Click 命令行 └── api/ Web 入口FastAPI SPA 配置层1个 └── config/ 配置YAML 加载、Pydantic 模型每个模块的内部结构以content/模块为例content/ ├── __init__.py ├── schema.py 数据模型GeneratedContent, QualityReport ├── body.py 正文生成器BodyGenerator ├── title.py 标题生成器TitleGenerator ├── script.py 脚本生成器ScriptGenerator └── formatter.py 排版器Formatter每个模块的固定套路文件职责命名约定schema.py数据模型Pydantic BaseModel必有xxx.py业务逻辑一个类按功能命名__init__.py模块导出可空schema.py和逻辑分离— 数据结构定义在 schema业务逻辑在各自的文件。这样 schema 可以被其他模块 import比如quality/需要content/schema.py的QualityReport不会循环依赖。模块间不互相 import 业务逻辑✅ 正确quality/ import content/schema.py数据模型 ❌ 错误quality/ import content/body.py业务逻辑 ✅ 正确pipeline/runner.py import 所有模块组装中心 ❌ 错误content/body.py import quality/业务层互相依赖业务模块之间不直接依赖—content/不 importquality/topic/不 importcontent/。它们之间的协作由PipelineRunner编排。唯一的例外是数据模型—quality/需要content/schema.py的QualityReport来返回质检结果这是数据依赖不是逻辑依赖。三、依赖注入通过构造函数传递问题模块怎么拿到 LLM 和存储BodyGenerator 需要 LLMClient 来生成正文 TopicGenerator 需要 LLMClient 来生成选题 PersonaManager 需要 Repository 来存取人设 TopicPool 需要 Repository 来存取选题如果每个模块自己创建依赖# ❌ 错误每个模块自己 new class BodyGenerator: def __init__(self): self.llm LLMClient(base_url..., api_key...) # 自己创建 class TopicGenerator: def __init__(self): self.llm LLMClient(base_url..., api_key...) # 又创建一个问题创建了多个 LLMClient 实例配置散落在各处换配置要改每个模块。解决构造函数注入# ✅ 正确从外部注入 class BodyGenerator: def __init__(self, llm: LLMClient, prompt_mgr: PromptManager): self.llm llm # 外部传入 self.prompt_mgr prompt_mgr class TopicGenerator: def __init__(self, llm: LLMClient, prompt_mgr: PromptManager): self.llm llm # 同一个实例 self.prompt_mgr prompt_mgr在 PipelineRunner 里统一创建和注入# pipeline/runner.py class PipelineRunner: def __init__(self, config: AppConfig, repo: Repository, cacheNone): self.config config self.repo repo # 1. 创建基础设施只创建一次 self.llm LLMClient( base_urlconfig.llm.base_url, api_keyconfig.llm.api_key, modelconfig.llm.model, ) self.prompt_mgr PromptManager() # 2. 注入到业务模块 self.persona_mgr PersonaManager(repo) # repo 注入 self.topic_pool TopicPool(repo) # repo 注入 self.topic_gen TopicGenerator(self.llm, self.prompt_mgr) # llm 注入 self.title_gen TitleGenerator(self.llm, self.prompt_mgr) # llm 注入 self.body_gen BodyGenerator(self.llm, self.prompt_mgr) # llm 注入 self.script_gen ScriptGenerator(self.llm, self.prompt_mgr) self.formatter Formatter() # 无依赖 # 3. 质检编排器注入子检查器 self.quality_orchestrator QualityOrchestrator( sensitive_filterSensitiveFilter( sensitive_words_fileconfig.quality.sensitive_words_file, ), )PipelineRunner 是组装中心— 它创建所有基础设施LLMClient、PromptManager然后注入到每个业务模块。整个项目只有这里new对象。依赖注入图PipelineRunner组装中心 │ ├── LLMClient ←─────── 注入给 TopicGenerator, TitleGenerator, BodyGenerator, ScriptGenerator ├── PromptManager ←─── 注入给同上 ├── Repository ←────── 注入给 PersonaManager, TopicPool │ ├── PersonaManager(repo) ├── TopicPool(repo) ├── TopicGenerator(llm, prompt_mgr) ├── TitleGenerator(llm, prompt_mgr) ├── BodyGenerator(llm, prompt_mgr) ├── ScriptGenerator(llm, prompt_mgr) ├── Formatter() └── QualityOrchestrator(sensitive_filter, dedup_checker, ...)两种依赖依赖类型注入对象谁接收基础设施LLMClient, PromptManager, Repository所有需要的业务模块业务模块PersonaManager, TopicGenerator...PipelineRunner 自己持有基础设施注入给业务模块业务模块被 PipelineRunner 持有— 层次清晰不交叉。四、Repository数据访问的统一接口三层存储架构Repository仓储封装 ├── MemoryStore内存引擎←── dict 存储零依赖 └── FilePersistence文件持久化←── YAML 读写# storage/repository.py class Repository: 数据访问层 — 统一接口屏蔽存储实现 def __init__(self, persist_dirNone, persist_formatyaml): self.store MemoryStore() # 内存引擎必有 self.persistence None if persist_dir: # 持久化模式 self.persistence FilePersistence( store_dirpersist_dir, formatpersist_format, ) self.persistence.load(self.store) # 启动时从文件恢复 def _maybe_persist(self): 操作后可选落盘 if self.persistence: self.persistence.save(self.store)Repository 封装了两个引擎MemoryStore— 进程内 dict 存储所有读写都走这里快FilePersistence— YAML 文件读写启动时加载修改后落盘持久业务模块只跟 Repository 打交道不知道底层是内存还是文件# persona/manager.py class PersonaManager: def __init__(self, repo: Repository): self.repo repo def create(self, config: PersonaConfig) - PersonaConfig: result self.repo.store.save_persona(config) # 写内存 self.repo._maybe_persist() # 可选落盘 return result def get(self, persona_id: str) - PersonaConfig | None: return self.repo.store.get_persona(persona_id) # 读内存PersonaManager 不关心数据存哪— 它只调repo.store.save_persona()和repo._maybe_persist()至于存到内存还是文件由 Repository 决定。MemoryStore统一的数据接口# storage/memory.py class MemoryStore: def __init__(self): self._personas: dict[str, PersonaConfig] {} self._topics: dict[str, Topic] {} self._contents: dict[str, GeneratedContent] {} self._chat_sessions: dict[str, ChatSession] {} # --- Persona --- def save_persona(self, persona): ... def get_persona(self, persona_id): ... def list_personas(self): ... def delete_persona(self, persona_id): ... # --- Topic --- def save_topic(self, topic): ... def get_topic(self, topic_id): ... def list_topics(self): ... # --- Content --- def save_content(self, content): ... def get_content(self, content_id): ... def list_contents(self): ... # --- ChatSession --- def save_chat_session(self, session): ... def get_chat_session_by_content(self, content_id): ...四类数据每类一套 CRUD— 人设、选题、内容、会话全用 dict 存储接口统一。加一种新数据类型只需三步MemoryStore加一个 dict CRUD 方法FilePersistence加序列化/反序列化业务模块通过repo.store调用不需要改任何已有模块— 这就是模块化的好处。五、PipelineRunner组装中心职责编排不实现# pipeline/runner.py class PipelineRunner: V2 主流程选题 → 标题 → 正文 → 排版 → 四维质检 → 输出PipelineRunner 自己不实现任何业务逻辑— 它只做两件事组装— 创建所有模块注入依赖编排— 按顺序调用各模块传递数据主流程run()async def run(self, persona_id, topicsNone, topic_count10, ...): # Step 1: 获取/生成选题 persona self.persona_mgr.get(persona_id) if topics: topic_list [Topic(...) for t in topics] for t in topic_list: self.topic_pool.add(t) else: topic_list await self.topic_gen.generate_topics(persona, topic_count) # Step 2: 批量生成内容 results [] for topic in topic_list: content await self._generate_single(persona, topic, ...) results.append(content) self.topic_pool.mark_used(topic.id) return results两步走先选题再逐篇生成。每篇生成完标记选题为已使用。单篇生成_generate_single()async def _generate_single(self, persona, topic, output_dir, use_as_titleFalse): # 2a: 生成标题 if use_as_title and topic.source TopicSource.MANUAL: best_title topic.title else: titles await self.title_gen.generate(persona, topic) best_title titles[0] # 2b: 生成正文/脚本 if persona.content_format short_video: script await self.script_gen.generate_script(persona, topic, best_title) raw_body script.narration else: raw_body await self.body_gen.generate_text(persona, topic, best_title) # 2c: 排版 formatted self.formatter.format(persona, raw_body, topic_titletopic.title) # 2d: 四维质检 existing_contents [c.body for c in self.repo.store.list_contents() if c.persona_id persona.id] quality_result await self.quality_orchestrator.check_and_fix( personapersona, contentformatted, auto_fixself.config.quality.auto_fix, existing_contentsexisting_contents, ) formatted quality_result.final_content # 2e: 构建内容对象 content GeneratedContent( persona_idpersona.id, titlebest_title, bodyformatted, ... ) # 2f: 存储 self.repo.store.save_content(content) self.repo._maybe_persist() # 2g: 导出文件 await self._export(content, output_dir) return content七个子步骤每个调一个模块2a: title_gen.generate() → 标题生成 2b: body_gen / script_gen → 正文/脚本生成 2c: formatter.format() → 排版 2d: quality_orchestrator → 质检 2e: GeneratedContent() → 构建对象 2f: repo.store.save_content() → 存储 2g: _export() → 导出文件PipelineRunner 只负责按顺序调这些模块不关心每个模块内部怎么实现— 换个排版器、换个质检器PipelineRunner 的代码不用改只要接口不变。六、两种入口同一套业务CLI 入口# cli/main.py click.group() click.pass_context def cli(ctx): Self-Media-Agent — 全自动批量内容生产 AI Agent ctx.ensure_object(dict) ctx.obj[config] load_config() ctx.obj[repo] _get_repo(ctx.obj[config]) cli.command() click.option(--persona, -p, persona_id, requiredTrue) click.option(--topics, -t, topics_str) click.pass_context def run(ctx, persona_id, topics_str): 执行内容生产 config ctx.obj[config] repo ctx.obj[repo] runner PipelineRunner(configconfig, reporepo) topics topics_str.split(,) if topics_str else None results asyncio.run(runner.run(persona_idpersona_id, topicstopics)) # ... 输出结果CLI 做三件事加载配置 创建 Repository创建 PipelineRunner注入 config 和 repo调runner.run()执行Web 入口# api/app.py class AppState: def __init__(self, configNone): self.config config or load_config() self.repo Repository(...) self.cache init_cache(...) self.task_engine TaskEngine(...) # api/routes/content.py router.post(/generate) async def generate_content(req: ContentGenerateRequest): state get_state() runner PipelineRunner(configstate.config, repostate.repo) async def _generate(): results await runner.run(persona_idreq.persona_id, topicsreq.topics) return [r.model_dump(modejson) for r in results] task_id await state.task_engine.submit(name内容生成, coro_func_generate) return ContentGenerateResponse(task_idtask_id)Web 也做三件事AppState 初始化 config repo cache task_engine路由里创建 PipelineRunner注入 config 和 repo提交到 task_engine 异步执行共享同一套业务模块CLI 入口 (cli/main.py) ──┐ ├──→ PipelineRunner ──→ persona/ topic/ content/ quality/ Web 入口 (api/app.py) ───┘ ↑ │ Repository ─────────────┘ LLMClient ─────────────┘两种入口共享同一套业务模块— CLI 和 Web 的区别只在怎么接收请求和怎么返回结果核心业务逻辑完全一样。CLI命令行参数 → PipelineRunner.run() → rich.Table 输出 WebHTTP 请求 → PipelineRunner.run() → JSON 响应这就是模块化的核心价值— 业务逻辑和入口解耦。加第三种入口比如 gRPC、消息队列只需要写一个新入口文件业务模块完全不用改。七、路由注册模式每个模块一个 APIRouter路由与业务模块的对应# api/app.py from .routes import persona, topic, content, task, hotspot, analytics, options, chat app.include_router(persona.router, prefix/api/personas, tags[人设管理]) app.include_router(topic.router, prefix/api/topics, tags[选题管理]) app.include_router(content.router, prefix/api/content, tags[内容生产]) app.include_router(task.router, prefix/api/tasks, tags[任务监控]) app.include_router(hotspot.router, prefix/api/hotspots, tags[热点]) app.include_router(analytics.router, prefix/api/analytics, tags[数据分析]) app.include_router(options.router, prefix/api/options, tags[选项管理]) app.include_router(chat.router, prefix/api/chat, tags[AI 对话])8 个路由模块8 个 APIRouter在 app.py 统一注册— 每个路由模块对应一个业务领域。路由模块的固定结构# api/routes/persona.py router APIRouter() router.post(, response_modelAPIResponse) async def create_persona(req: PersonaCreateRequest) - APIResponse: state get_state() # 1. 拿全局状态 persona PersonaConfig(**req.model_dump()) mgr PersonaManager(state.repo) # 2. 创建业务 Manager result mgr.create(persona) # 3. 调业务逻辑 return APIResponse(dataresult.model_dump(modejson)) # 4. 包成统一响应每个路由函数的四步套路get_state()— 拿全局状态config、repo、cache、task_engine创建业务 Manager —PersonaManager(state.repo)调业务逻辑 —mgr.create(persona)包成APIResponse返回路由模块不持有状态不做业务逻辑— 它只是 HTTP 和业务之间的薄薄一层适配层。加一个新路由的步骤加收藏夹功能 → ① 新建 api/routes/favorite.py路由 CRUD ② 在 app.py 加一行app.include_router(favorite.router, prefix/api/favorites, ...) → 完事其他路由模块完全不用改一行注册代码— 这就是路由注册模式的好处。不用在一个巨大的app.py里塞所有路由每个路由模块独立维护。八、质检编排器策略模式的体现编排器持有四个检查器# quality/orchestrator.py class QualityOrchestrator: def __init__( self, sensitive_filter: Optional[SensitiveFilter] None, dedup_checker: Optional[DedupChecker] None, colloquial_optimizer: Optional[ColloquialOptimizer] None, logic_checker: Optional[LogicChecker] None, ): self.sensitive_filter sensitive_filter or SensitiveFilter() self.dedup_checker dedup_checker or DedupChecker() self.colloquial_optimizer colloquial_optimizer or ColloquialOptimizer() self.logic_checker logic_checker or LogicChecker()四个检查器通过构造函数注入— 可以传自定义的也可以用默认的。这就是策略模式。# pipeline/runner.py 里的组装 self.quality_orchestrator QualityOrchestrator( sensitive_filterSensitiveFilter( sensitive_words_fileconfig.quality.sensitive_words_file, ), )PipelineRunner 只定制了 SensitiveFilter传了敏感词文件路径其他三个用默认值。如果未来要换一个更强的去重检查器只需在这里传一个新的DedupChecker实现QualityOrchestrator 的代码不用改。统一接口check fix# 每个检查器都有相同的接口 class SensitiveFilter: async def check(self, persona, content) - QualityReport: ... async def auto_fix(self, persona, content, report) - str: ... class DedupChecker: async def check(self, persona, content, existing_contents) - QualityReport: ... async def fix(self, persona, content, report) - str: ... class ColloquialOptimizer: async def check(self, persona, content) - QualityReport: ... async def auto_fix(self, persona, content) - str: ...每个检查器都返回QualityReport都有check和fix— 接口统一编排器才能统一处理。加一个新检查器的步骤新建quality/xxx.py实现check和fix在QualityOrchestrator.__init__加一个参数在check_and_fix里加一步调用不用改任何已有检查器— 新检查器独立开发独立测试。九、配置层Pydantic 模型驱动配置模型# config/models.py class AppConfig(BaseModel): llm: LLMConfig storage: StorageConfig cache: CacheConfig task: TaskConfig quality: QualityConfig web: WebConfig所有配置用 Pydantic 模型定义— 类型校验、默认值、嵌套结构全由 Pydantic 保证。# config/loader.py def load_config(pathNone) - AppConfig: 加载配置 with open(path or config/default.yaml) as f: data yaml.safe_load(f) return AppConfig(**data)YAML → Pydantic 模型—load_config()读 YAML 文件Pydantic 自动校验类型和默认值。配置错了启动就报错不会等到运行时才发现。配置注入# PipelineRunner 用 config 创建 LLMClient self.llm LLMClient( base_urlconfig.llm.base_url, api_keyconfig.llm.api_key, modelconfig.llm.model, ) # QualityOrchestrator 用 config 创建 SensitiveFilter self.quality_orchestrator QualityOrchestrator( sensitive_filterSensitiveFilter( sensitive_words_fileconfig.quality.sensitive_words_file, ), )config 在最顶层加载逐层注入到需要的地方— 不在业务模块里直接读配置文件而是由 PipelineRunner 拿到 config 后取需要的部分传给每个模块。好处业务模块不知道配置文件长什么样只知道构造函数传进来的参数。换配置源YAML → 环境变量 → 数据库只改load_config()业务模块不用改。踩坑记录坑1加聊天功能要改 4 个文件单体时代加对话修改功能 ① agent.py 加 chat 函数200 行混进 500 行的文件 ② main 函数加 if 判断调 chat 还是 generate ③ 输出逻辑加保存修改记录 ④ 人设加载加 style_preferences 字段 → 改了 4 个地方每个地方都可能引入 bug → 而且 chat 的代码和 generate 的代码混在一起很难维护模块化后加对话修改功能 ① 新建 chat/ 模块schema.py 定义 ChatSession RevisionRecord ② 新建 api/routes/chat.py路由 ③ app.py 加一行路由注册 → 其他模块完全不用改 → chat 的代码独立在 chat/ 目录不影响其他功能教训模块化的核心收益不是代码好看而是加功能时不碰已有代码。已有代码不动就没有回归 bug。坑2循环依赖content/body.py 需要 persona/schema.py 的 PersonaConfig persona/style_learner.py 需要 content/schema.py 的 GeneratedContent → content import personapersona import content → Python 报错ImportError: cannot import name根因业务模块之间互相 import形成循环。修复把数据模型抽到schema.py业务模块只 import schema不 import 业务逻辑。✅ content/body.py import persona/schema.py只拿数据模型 ✅ persona/style_learner.py import content/schema.py只拿数据模型 ❌ content/body.py import persona/manager.py拿业务逻辑教训schema.py和业务逻辑分离不只是好看是解决循环依赖的必要手段。数据模型是共享的业务逻辑是私有的。坑3PipelineRunner 膨胀PipelineRunner 从 100 行膨胀到 280 行 → _generate_single 方法 90 行做了 7 件事 → 想拆成 _generate_title, _generate_body, _format, _quality_check... → 但拆了之后方法间要传一堆中间变量更乱了权衡后的决定保持_generate_single作为一个长方法。拆开的代价 → 7 个子方法每个 10 行 → 方法间传递 title, raw_body, formatted, quality_result 等中间变量 → 变量传递的代码比业务逻辑还多 保持长方法的好处 → 所有中间变量都在一个作用域里不用传来传去 → 执行顺序一目了然2a → 2b → 2c → 2d → 2e → 2f → 2g → 虽然长但逻辑线性容易读教训不是所有长方法都需要拆。如果一个方法的逻辑是线性的一步步往下走没有分支保持长方法比强行拆成小方法更可读。拆方法的目的是消除分支和重复不是为了短。坑4AppState 单例 vs 依赖注入路由里用 get_state() 拿全局状态 state get_state() mgr PersonaManager(state.repo) → 每个路由函数都 new 一个 PersonaManager → PersonaManager 是无状态的new 一次和 new 多次没区别 → 但如果 PersonaManager 有状态呢比如缓存 → 每次请求 new 一个缓存就失效了当前没问题— PersonaManager、TopicPool 等都是无状态的状态全在 repo 里每次 new 不会有问题。潜在风险— 如果未来某个 Manager 需要持有状态比如本地缓存每次请求 new 一个就会出问题。到时候需要改成单例或在 AppState 里持有。教训get_state() 每次 new Manager 的模式适用于无状态 Manager。如果 Manager 有状态应该在 AppState 里创建一次路由里通过state.xxx_mgr访问。关键 Takeaway按职责拆模块通过构造函数注入依赖— 每个模块只做一件事依赖从外部传入不在模块内部new。PipelineRunner 是唯一的组装中心所有对象在这里创建和接线。schema 和业务逻辑分离是解循环依赖的关键— 数据模型Pydantic BaseModel放在schema.py可以被其他模块 import业务逻辑放在各自的文件里不被其他业务模块 import。数据共享逻辑私有。两种入口共享同一套业务模块— CLI 和 Web 的区别只在怎么接收请求和怎么返回结果核心业务逻辑完全一样。加新入口gRPC、消息队列只需写入口文件业务模块不用改。下篇预告下一篇《配置驱动让Agent灵活适配不同场景》本文讲了架构怎么从单体演进到模块化但有个问题没讲同一个 Agent 怎么服务 100 个不同的赛道美妆赛道和职场赛道用同一套代码怎么做到硬编码 → 加北漂赛道要改 3 个文件 配置驱动 → 加北漂赛道只改 1 个 YAML 文件下一篇讲配置驱动设计 — YAML 配置层叠、人设模板、选项动态加载、环境变量管理让同一个 Agent 灵活适配不同场景。 Agent架构模式：从单体到模块化 — 好的架构让加功能像搭积木，而不是拆房子
先说结论早期所有逻辑写在一个文件里 — 人设、选题、生成、质检、输出全塞在一起改一处牵一发动全身。加个聊天功能要改 4 个文件加个热点功能要在 3 个地方插代码。在 self-media-agent 项目里最终演进为14 个子模块 2 个入口 1 个组装中心src/self_media_agent/ ├── config/ 配置加载 ├── persona/ 人设管理 ├── topic/ 选题管理 ├── content/ 内容生成 ├── quality/ 质检系统 ├── hotspot/ 热点感知 ├── chat/ 对话修改 ├── storage/ 数据存储 ├── llm/ LLM 客户端 ├── cache/ 缓存 ├── task/ 异步任务 ├── analytics/ 数据分析 ├── pipeline/ 流水线编排组装中心 ├── api/ Web 入口 └── cli/ CLI 入口每个模块按职责拆分通过Repository和LLMClient传递依赖在PipelineRunner里组装在app.py或cli/main.py里入口。架构层职责对应代码入口层接收请求调用业务cli/main.pyapi/app.py编排层组装模块控制流程pipeline/runner.py业务层各自独立的业务逻辑persona/topic/content/quality/chat/基础设施层通用能力llm/storage/cache/task/好的架构让加功能像搭积木 — 新建一个模块在编排层接上完事。一、为什么需要模块化单体时代所有逻辑在一个文件agent.py500行 ├── 人设加载YAML 解析 校验 ├── 选题生成调 LLM 解析 JSON ├── 标题生成调 LLM 选最优 ├── 正文生成调 LLM 注入人设 ├── 排版分段 emoji 加粗 ├── 质检敏感词 重复检测 ├── 文件输出写 Markdown └── main 函数串起来单体的问题改一处牵一全身— 改排版逻辑不小心碰到了选题生成的代码无法复用— CLI 想调生成正文的功能只能 import 整个 500 行的文件测试困难— 测质检逻辑要先跑完生成没法单独测加功能要改多处— 加聊天功能改 agent.py加逻辑 改 main加调用 改输出加保存想加对话修改功能 → ① 在 agent.py 里加 chat 函数200行混进去 ② 在 main 里加 if 判断调 chat 还是 generate ③ 修改输出逻辑保存修改记录 ④ 改人设加载加 style_preferences 字段 → 改了 4 个地方每个地方都可能引入 bug模块化的目标加对话修改功能 → ① 新建 chat/ 模块schema.py 逻辑 ② 在 api/routes/ 加 chat.py 路由 ③ 在 app.py 注册路由1行 → 其他模块完全不用改模块化的三个原则单一职责— 每个模块只做一件事显式依赖— 模块间的依赖通过构造函数注入不隐式 import开放扩展— 加功能 加模块 接线不改已有模块二、模块拆分按职责分14 个模块的职责业务层7个— 各管一个业务领域 ├── persona/ 人设定义、CRUD、风格学习 ├── topic/ 选题生成、选题池管理 ├── content/ 内容标题、正文、脚本、排版 ├── quality/ 质检合规、去重、口语化、逻辑 ├── hotspot/ 热点爬取、分析、多平台适配 ├── chat/ 对话会话、修改记录 └── analytics/ 分析发布数据追踪 基础设施层4个— 通用能力 ├── llm/ LLM 客户端调用、Prompt 管理、Token 计算 ├── storage/ 存储内存引擎、文件持久化、仓储封装 ├── cache/ 缓存LRU TTL └── task/ 任务异步任务引擎 编排层1个 └── pipeline/ 流水线组装所有模块控制执行顺序 入口层2个 ├── cli/ CLI 入口Click 命令行 └── api/ Web 入口FastAPI SPA 配置层1个 └── config/ 配置YAML 加载、Pydantic 模型每个模块的内部结构以content/模块为例content/ ├── __init__.py ├── schema.py 数据模型GeneratedContent, QualityReport ├── body.py 正文生成器BodyGenerator ├── title.py 标题生成器TitleGenerator ├── script.py 脚本生成器ScriptGenerator └── formatter.py 排版器Formatter每个模块的固定套路文件职责命名约定schema.py数据模型Pydantic BaseModel必有xxx.py业务逻辑一个类按功能命名__init__.py模块导出可空schema.py和逻辑分离— 数据结构定义在 schema业务逻辑在各自的文件。这样 schema 可以被其他模块 import比如quality/需要content/schema.py的QualityReport不会循环依赖。模块间不互相 import 业务逻辑✅ 正确quality/ import content/schema.py数据模型 ❌ 错误quality/ import content/body.py业务逻辑 ✅ 正确pipeline/runner.py import 所有模块组装中心 ❌ 错误content/body.py import quality/业务层互相依赖业务模块之间不直接依赖—content/不 importquality/topic/不 importcontent/。它们之间的协作由PipelineRunner编排。唯一的例外是数据模型—quality/需要content/schema.py的QualityReport来返回质检结果这是数据依赖不是逻辑依赖。三、依赖注入通过构造函数传递问题模块怎么拿到 LLM 和存储BodyGenerator 需要 LLMClient 来生成正文 TopicGenerator 需要 LLMClient 来生成选题 PersonaManager 需要 Repository 来存取人设 TopicPool 需要 Repository 来存取选题如果每个模块自己创建依赖# ❌ 错误每个模块自己 new class BodyGenerator: def __init__(self): self.llm LLMClient(base_url..., api_key...) # 自己创建 class TopicGenerator: def __init__(self): self.llm LLMClient(base_url..., api_key...) # 又创建一个问题创建了多个 LLMClient 实例配置散落在各处换配置要改每个模块。解决构造函数注入# ✅ 正确从外部注入 class BodyGenerator: def __init__(self, llm: LLMClient, prompt_mgr: PromptManager): self.llm llm # 外部传入 self.prompt_mgr prompt_mgr class TopicGenerator: def __init__(self, llm: LLMClient, prompt_mgr: PromptManager): self.llm llm # 同一个实例 self.prompt_mgr prompt_mgr在 PipelineRunner 里统一创建和注入# pipeline/runner.py class PipelineRunner: def __init__(self, config: AppConfig, repo: Repository, cacheNone): self.config config self.repo repo # 1. 创建基础设施只创建一次 self.llm LLMClient( base_urlconfig.llm.base_url, api_keyconfig.llm.api_key, modelconfig.llm.model, ) self.prompt_mgr PromptManager() # 2. 注入到业务模块 self.persona_mgr PersonaManager(repo) # repo 注入 self.topic_pool TopicPool(repo) # repo 注入 self.topic_gen TopicGenerator(self.llm, self.prompt_mgr) # llm 注入 self.title_gen TitleGenerator(self.llm, self.prompt_mgr) # llm 注入 self.body_gen BodyGenerator(self.llm, self.prompt_mgr) # llm 注入 self.script_gen ScriptGenerator(self.llm, self.prompt_mgr) self.formatter Formatter() # 无依赖 # 3. 质检编排器注入子检查器 self.quality_orchestrator QualityOrchestrator( sensitive_filterSensitiveFilter( sensitive_words_fileconfig.quality.sensitive_words_file, ), )PipelineRunner 是组装中心— 它创建所有基础设施LLMClient、PromptManager然后注入到每个业务模块。整个项目只有这里new对象。依赖注入图PipelineRunner组装中心 │ ├── LLMClient ←─────── 注入给 TopicGenerator, TitleGenerator, BodyGenerator, ScriptGenerator ├── PromptManager ←─── 注入给同上 ├── Repository ←────── 注入给 PersonaManager, TopicPool │ ├── PersonaManager(repo) ├── TopicPool(repo) ├── TopicGenerator(llm, prompt_mgr) ├── TitleGenerator(llm, prompt_mgr) ├── BodyGenerator(llm, prompt_mgr) ├── ScriptGenerator(llm, prompt_mgr) ├── Formatter() └── QualityOrchestrator(sensitive_filter, dedup_checker, ...)两种依赖依赖类型注入对象谁接收基础设施LLMClient, PromptManager, Repository所有需要的业务模块业务模块PersonaManager, TopicGenerator...PipelineRunner 自己持有基础设施注入给业务模块业务模块被 PipelineRunner 持有— 层次清晰不交叉。四、Repository数据访问的统一接口三层存储架构Repository仓储封装 ├── MemoryStore内存引擎←── dict 存储零依赖 └── FilePersistence文件持久化←── YAML 读写# storage/repository.py class Repository: 数据访问层 — 统一接口屏蔽存储实现 def __init__(self, persist_dirNone, persist_formatyaml): self.store MemoryStore() # 内存引擎必有 self.persistence None if persist_dir: # 持久化模式 self.persistence FilePersistence( store_dirpersist_dir, formatpersist_format, ) self.persistence.load(self.store) # 启动时从文件恢复 def _maybe_persist(self): 操作后可选落盘 if self.persistence: self.persistence.save(self.store)Repository 封装了两个引擎MemoryStore— 进程内 dict 存储所有读写都走这里快FilePersistence— YAML 文件读写启动时加载修改后落盘持久业务模块只跟 Repository 打交道不知道底层是内存还是文件# persona/manager.py class PersonaManager: def __init__(self, repo: Repository): self.repo repo def create(self, config: PersonaConfig) - PersonaConfig: result self.repo.store.save_persona(config) # 写内存 self.repo._maybe_persist() # 可选落盘 return result def get(self, persona_id: str) - PersonaConfig | None: return self.repo.store.get_persona(persona_id) # 读内存PersonaManager 不关心数据存哪— 它只调repo.store.save_persona()和repo._maybe_persist()至于存到内存还是文件由 Repository 决定。MemoryStore统一的数据接口# storage/memory.py class MemoryStore: def __init__(self): self._personas: dict[str, PersonaConfig] {} self._topics: dict[str, Topic] {} self._contents: dict[str, GeneratedContent] {} self._chat_sessions: dict[str, ChatSession] {} # --- Persona --- def save_persona(self, persona): ... def get_persona(self, persona_id): ... def list_personas(self): ... def delete_persona(self, persona_id): ... # --- Topic --- def save_topic(self, topic): ... def get_topic(self, topic_id): ... def list_topics(self): ... # --- Content --- def save_content(self, content): ... def get_content(self, content_id): ... def list_contents(self): ... # --- ChatSession --- def save_chat_session(self, session): ... def get_chat_session_by_content(self, content_id): ...四类数据每类一套 CRUD— 人设、选题、内容、会话全用 dict 存储接口统一。加一种新数据类型只需三步MemoryStore加一个 dict CRUD 方法FilePersistence加序列化/反序列化业务模块通过repo.store调用不需要改任何已有模块— 这就是模块化的好处。五、PipelineRunner组装中心职责编排不实现# pipeline/runner.py class PipelineRunner: V2 主流程选题 → 标题 → 正文 → 排版 → 四维质检 → 输出PipelineRunner 自己不实现任何业务逻辑— 它只做两件事组装— 创建所有模块注入依赖编排— 按顺序调用各模块传递数据主流程run()async def run(self, persona_id, topicsNone, topic_count10, ...): # Step 1: 获取/生成选题 persona self.persona_mgr.get(persona_id) if topics: topic_list [Topic(...) for t in topics] for t in topic_list: self.topic_pool.add(t) else: topic_list await self.topic_gen.generate_topics(persona, topic_count) # Step 2: 批量生成内容 results [] for topic in topic_list: content await self._generate_single(persona, topic, ...) results.append(content) self.topic_pool.mark_used(topic.id) return results两步走先选题再逐篇生成。每篇生成完标记选题为已使用。单篇生成_generate_single()async def _generate_single(self, persona, topic, output_dir, use_as_titleFalse): # 2a: 生成标题 if use_as_title and topic.source TopicSource.MANUAL: best_title topic.title else: titles await self.title_gen.generate(persona, topic) best_title titles[0] # 2b: 生成正文/脚本 if persona.content_format short_video: script await self.script_gen.generate_script(persona, topic, best_title) raw_body script.narration else: raw_body await self.body_gen.generate_text(persona, topic, best_title) # 2c: 排版 formatted self.formatter.format(persona, raw_body, topic_titletopic.title) # 2d: 四维质检 existing_contents [c.body for c in self.repo.store.list_contents() if c.persona_id persona.id] quality_result await self.quality_orchestrator.check_and_fix( personapersona, contentformatted, auto_fixself.config.quality.auto_fix, existing_contentsexisting_contents, ) formatted quality_result.final_content # 2e: 构建内容对象 content GeneratedContent( persona_idpersona.id, titlebest_title, bodyformatted, ... ) # 2f: 存储 self.repo.store.save_content(content) self.repo._maybe_persist() # 2g: 导出文件 await self._export(content, output_dir) return content七个子步骤每个调一个模块2a: title_gen.generate() → 标题生成 2b: body_gen / script_gen → 正文/脚本生成 2c: formatter.format() → 排版 2d: quality_orchestrator → 质检 2e: GeneratedContent() → 构建对象 2f: repo.store.save_content() → 存储 2g: _export() → 导出文件PipelineRunner 只负责按顺序调这些模块不关心每个模块内部怎么实现— 换个排版器、换个质检器PipelineRunner 的代码不用改只要接口不变。六、两种入口同一套业务CLI 入口# cli/main.py click.group() click.pass_context def cli(ctx): Self-Media-Agent — 全自动批量内容生产 AI Agent ctx.ensure_object(dict) ctx.obj[config] load_config() ctx.obj[repo] _get_repo(ctx.obj[config]) cli.command() click.option(--persona, -p, persona_id, requiredTrue) click.option(--topics, -t, topics_str) click.pass_context def run(ctx, persona_id, topics_str): 执行内容生产 config ctx.obj[config] repo ctx.obj[repo] runner PipelineRunner(configconfig, reporepo) topics topics_str.split(,) if topics_str else None results asyncio.run(runner.run(persona_idpersona_id, topicstopics)) # ... 输出结果CLI 做三件事加载配置 创建 Repository创建 PipelineRunner注入 config 和 repo调runner.run()执行Web 入口# api/app.py class AppState: def __init__(self, configNone): self.config config or load_config() self.repo Repository(...) self.cache init_cache(...) self.task_engine TaskEngine(...) # api/routes/content.py router.post(/generate) async def generate_content(req: ContentGenerateRequest): state get_state() runner PipelineRunner(configstate.config, repostate.repo) async def _generate(): results await runner.run(persona_idreq.persona_id, topicsreq.topics) return [r.model_dump(modejson) for r in results] task_id await state.task_engine.submit(name内容生成, coro_func_generate) return ContentGenerateResponse(task_idtask_id)Web 也做三件事AppState 初始化 config repo cache task_engine路由里创建 PipelineRunner注入 config 和 repo提交到 task_engine 异步执行共享同一套业务模块CLI 入口 (cli/main.py) ──┐ ├──→ PipelineRunner ──→ persona/ topic/ content/ quality/ Web 入口 (api/app.py) ───┘ ↑ │ Repository ─────────────┘ LLMClient ─────────────┘两种入口共享同一套业务模块— CLI 和 Web 的区别只在怎么接收请求和怎么返回结果核心业务逻辑完全一样。CLI命令行参数 → PipelineRunner.run() → rich.Table 输出 WebHTTP 请求 → PipelineRunner.run() → JSON 响应这就是模块化的核心价值— 业务逻辑和入口解耦。加第三种入口比如 gRPC、消息队列只需要写一个新入口文件业务模块完全不用改。七、路由注册模式每个模块一个 APIRouter路由与业务模块的对应# api/app.py from .routes import persona, topic, content, task, hotspot, analytics, options, chat app.include_router(persona.router, prefix/api/personas, tags[人设管理]) app.include_router(topic.router, prefix/api/topics, tags[选题管理]) app.include_router(content.router, prefix/api/content, tags[内容生产]) app.include_router(task.router, prefix/api/tasks, tags[任务监控]) app.include_router(hotspot.router, prefix/api/hotspots, tags[热点]) app.include_router(analytics.router, prefix/api/analytics, tags[数据分析]) app.include_router(options.router, prefix/api/options, tags[选项管理]) app.include_router(chat.router, prefix/api/chat, tags[AI 对话])8 个路由模块8 个 APIRouter在 app.py 统一注册— 每个路由模块对应一个业务领域。路由模块的固定结构# api/routes/persona.py router APIRouter() router.post(, response_modelAPIResponse) async def create_persona(req: PersonaCreateRequest) - APIResponse: state get_state() # 1. 拿全局状态 persona PersonaConfig(**req.model_dump()) mgr PersonaManager(state.repo) # 2. 创建业务 Manager result mgr.create(persona) # 3. 调业务逻辑 return APIResponse(dataresult.model_dump(modejson)) # 4. 包成统一响应每个路由函数的四步套路get_state()— 拿全局状态config、repo、cache、task_engine创建业务 Manager —PersonaManager(state.repo)调业务逻辑 —mgr.create(persona)包成APIResponse返回路由模块不持有状态不做业务逻辑— 它只是 HTTP 和业务之间的薄薄一层适配层。加一个新路由的步骤加收藏夹功能 → ① 新建 api/routes/favorite.py路由 CRUD ② 在 app.py 加一行app.include_router(favorite.router, prefix/api/favorites, ...) → 完事其他路由模块完全不用改一行注册代码— 这就是路由注册模式的好处。不用在一个巨大的app.py里塞所有路由每个路由模块独立维护。八、质检编排器策略模式的体现编排器持有四个检查器# quality/orchestrator.py class QualityOrchestrator: def __init__( self, sensitive_filter: Optional[SensitiveFilter] None, dedup_checker: Optional[DedupChecker] None, colloquial_optimizer: Optional[ColloquialOptimizer] None, logic_checker: Optional[LogicChecker] None, ): self.sensitive_filter sensitive_filter or SensitiveFilter() self.dedup_checker dedup_checker or DedupChecker() self.colloquial_optimizer colloquial_optimizer or ColloquialOptimizer() self.logic_checker logic_checker or LogicChecker()四个检查器通过构造函数注入— 可以传自定义的也可以用默认的。这就是策略模式。# pipeline/runner.py 里的组装 self.quality_orchestrator QualityOrchestrator( sensitive_filterSensitiveFilter( sensitive_words_fileconfig.quality.sensitive_words_file, ), )PipelineRunner 只定制了 SensitiveFilter传了敏感词文件路径其他三个用默认值。如果未来要换一个更强的去重检查器只需在这里传一个新的DedupChecker实现QualityOrchestrator 的代码不用改。统一接口check fix# 每个检查器都有相同的接口 class SensitiveFilter: async def check(self, persona, content) - QualityReport: ... async def auto_fix(self, persona, content, report) - str: ... class DedupChecker: async def check(self, persona, content, existing_contents) - QualityReport: ... async def fix(self, persona, content, report) - str: ... class ColloquialOptimizer: async def check(self, persona, content) - QualityReport: ... async def auto_fix(self, persona, content) - str: ...每个检查器都返回QualityReport都有check和fix— 接口统一编排器才能统一处理。加一个新检查器的步骤新建quality/xxx.py实现check和fix在QualityOrchestrator.__init__加一个参数在check_and_fix里加一步调用不用改任何已有检查器— 新检查器独立开发独立测试。九、配置层Pydantic 模型驱动配置模型# config/models.py class AppConfig(BaseModel): llm: LLMConfig storage: StorageConfig cache: CacheConfig task: TaskConfig quality: QualityConfig web: WebConfig所有配置用 Pydantic 模型定义— 类型校验、默认值、嵌套结构全由 Pydantic 保证。# config/loader.py def load_config(pathNone) - AppConfig: 加载配置 with open(path or config/default.yaml) as f: data yaml.safe_load(f) return AppConfig(**data)YAML → Pydantic 模型—load_config()读 YAML 文件Pydantic 自动校验类型和默认值。配置错了启动就报错不会等到运行时才发现。配置注入# PipelineRunner 用 config 创建 LLMClient self.llm LLMClient( base_urlconfig.llm.base_url, api_keyconfig.llm.api_key, modelconfig.llm.model, ) # QualityOrchestrator 用 config 创建 SensitiveFilter self.quality_orchestrator QualityOrchestrator( sensitive_filterSensitiveFilter( sensitive_words_fileconfig.quality.sensitive_words_file, ), )config 在最顶层加载逐层注入到需要的地方— 不在业务模块里直接读配置文件而是由 PipelineRunner 拿到 config 后取需要的部分传给每个模块。好处业务模块不知道配置文件长什么样只知道构造函数传进来的参数。换配置源YAML → 环境变量 → 数据库只改load_config()业务模块不用改。踩坑记录坑1加聊天功能要改 4 个文件单体时代加对话修改功能 ① agent.py 加 chat 函数200 行混进 500 行的文件 ② main 函数加 if 判断调 chat 还是 generate ③ 输出逻辑加保存修改记录 ④ 人设加载加 style_preferences 字段 → 改了 4 个地方每个地方都可能引入 bug → 而且 chat 的代码和 generate 的代码混在一起很难维护模块化后加对话修改功能 ① 新建 chat/ 模块schema.py 定义 ChatSession RevisionRecord ② 新建 api/routes/chat.py路由 ③ app.py 加一行路由注册 → 其他模块完全不用改 → chat 的代码独立在 chat/ 目录不影响其他功能教训模块化的核心收益不是代码好看而是加功能时不碰已有代码。已有代码不动就没有回归 bug。坑2循环依赖content/body.py 需要 persona/schema.py 的 PersonaConfig persona/style_learner.py 需要 content/schema.py 的 GeneratedContent → content import personapersona import content → Python 报错ImportError: cannot import name根因业务模块之间互相 import形成循环。修复把数据模型抽到schema.py业务模块只 import schema不 import 业务逻辑。✅ content/body.py import persona/schema.py只拿数据模型 ✅ persona/style_learner.py import content/schema.py只拿数据模型 ❌ content/body.py import persona/manager.py拿业务逻辑教训schema.py和业务逻辑分离不只是好看是解决循环依赖的必要手段。数据模型是共享的业务逻辑是私有的。坑3PipelineRunner 膨胀PipelineRunner 从 100 行膨胀到 280 行 → _generate_single 方法 90 行做了 7 件事 → 想拆成 _generate_title, _generate_body, _format, _quality_check... → 但拆了之后方法间要传一堆中间变量更乱了权衡后的决定保持_generate_single作为一个长方法。拆开的代价 → 7 个子方法每个 10 行 → 方法间传递 title, raw_body, formatted, quality_result 等中间变量 → 变量传递的代码比业务逻辑还多 保持长方法的好处 → 所有中间变量都在一个作用域里不用传来传去 → 执行顺序一目了然2a → 2b → 2c → 2d → 2e → 2f → 2g → 虽然长但逻辑线性容易读教训不是所有长方法都需要拆。如果一个方法的逻辑是线性的一步步往下走没有分支保持长方法比强行拆成小方法更可读。拆方法的目的是消除分支和重复不是为了短。坑4AppState 单例 vs 依赖注入路由里用 get_state() 拿全局状态 state get_state() mgr PersonaManager(state.repo) → 每个路由函数都 new 一个 PersonaManager → PersonaManager 是无状态的new 一次和 new 多次没区别 → 但如果 PersonaManager 有状态呢比如缓存 → 每次请求 new 一个缓存就失效了当前没问题— PersonaManager、TopicPool 等都是无状态的状态全在 repo 里每次 new 不会有问题。潜在风险— 如果未来某个 Manager 需要持有状态比如本地缓存每次请求 new 一个就会出问题。到时候需要改成单例或在 AppState 里持有。教训get_state() 每次 new Manager 的模式适用于无状态 Manager。如果 Manager 有状态应该在 AppState 里创建一次路由里通过state.xxx_mgr访问。关键 Takeaway按职责拆模块通过构造函数注入依赖— 每个模块只做一件事依赖从外部传入不在模块内部new。PipelineRunner 是唯一的组装中心所有对象在这里创建和接线。schema 和业务逻辑分离是解循环依赖的关键— 数据模型Pydantic BaseModel放在schema.py可以被其他模块 import业务逻辑放在各自的文件里不被其他业务模块 import。数据共享逻辑私有。两种入口共享同一套业务模块— CLI 和 Web 的区别只在怎么接收请求和怎么返回结果核心业务逻辑完全一样。加新入口gRPC、消息队列只需写入口文件业务模块不用改。下篇预告下一篇《配置驱动让Agent灵活适配不同场景》本文讲了架构怎么从单体演进到模块化但有个问题没讲同一个 Agent 怎么服务 100 个不同的赛道美妆赛道和职场赛道用同一套代码怎么做到硬编码 → 加北漂赛道要改 3 个文件 配置驱动 → 加北漂赛道只改 1 个 YAML 文件下一篇讲配置驱动设计 — YAML 配置层叠、人设模板、选项动态加载、环境变量管理让同一个 Agent 灵活适配不同场景。