企业官网建设流程全解析

先说结论

早期所有逻辑写在一个文件里 — 人设、选题、生成、质检、输出全塞在一起，改一处牵一发动全身。加个聊天功能要改 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里入口。

好的架构让加功能像搭积木 — 新建一个模块，在编排层接上，完事。

一、为什么需要模块化？

单体时代：所有逻辑在一个文件

agent.py（500行） ├── 人设加载（YAML 解析 + 校验） ├── 选题生成（调 LLM + 解析 JSON） ├── 标题生成（调 LLM + 选最优） ├── 正文生成（调 LLM + 注入人设） ├── 排版（分段 + emoji + 加粗） ├── 质检（敏感词 + 重复检测） ├── 文件输出（写 Markdown） └── main 函数（串起来）

单体的问题：

1. 改一处牵一全身— 改排版逻辑，不小心碰到了选题生成的代码

2. 无法复用— CLI 想调"生成正文"的功能，只能 import 整个 500 行的文件

3. 测试困难— 测质检逻辑要先跑完生成，没法单独测

4. 加功能要改多处— 加聊天功能：改 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行） → 其他模块完全不用改

模块化的三个原则：

1. 单一职责— 每个模块只做一件事

2. 显式依赖— 模块间的依赖通过构造函数注入，不隐式 import

3. 开放扩展— 加功能 = 加模块 + 接线，不改已有模块

二、模块拆分：按职责分

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和逻辑分离— 数据结构定义在 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, cache=None): self.config = config self.repo = repo # 1. 创建基础设施（只创建一次） self.llm = LLMClient( base_url=config.llm.base_url, api_key=config.llm.api_key, model=config.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_filter=SensitiveFilter( sensitive_words_file=config.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, ...)

两种依赖：

基础设施注入给业务模块，业务模块被 PipelineRunner 持有— 层次清晰，不交叉。

四、Repository：数据访问的统一接口

三层存储架构

Repository（仓储封装） ├── MemoryStore（内存引擎）←── dict 存储，零依赖 └── FilePersistence（文件持久化）←── YAML 读写

# storage/repository.py class Repository: """数据访问层 — 统一接口，屏蔽存储实现""" def __init__(self, persist_dir=None, persist_format="yaml"): self.store = MemoryStore() # 内存引擎（必有） self.persistence = None if persist_dir: # 持久化模式 self.persistence = FilePersistence( store_dir=persist_dir, format=persist_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 存储，接口统一。

加一种新数据类型只需三步：

1. MemoryStore加一个 dict + CRUD 方法

2. FilePersistence加序列化/反序列化

3. 业务模块通过repo.store调用

不需要改任何已有模块— 这就是模块化的好处。

五、PipelineRunner：组装中心

职责：编排，不实现

# pipeline/runner.py class PipelineRunner: """V2 主流程：选题 → 标题 → 正文 → 排版 → 四维质检 → 输出"""

PipelineRunner 自己不实现任何业务逻辑— 它只做两件事：

1. 组装— 创建所有模块，注入依赖

2. 编排— 按顺序调用各模块，传递数据

主流程：run()

async def run(self, persona_id, topics=None, topic_count=10, ...): # 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_title=False): # 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_title=topic.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( persona=persona, content=formatted, auto_fix=self.config.quality.auto_fix, existing_contents=existing_contents, ) formatted = quality_result.final_content # 2e: 构建内容对象 content = GeneratedContent( persona_id=persona.id, title=best_title, body=formatted, ... ) # 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", required=True) @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(config=config, repo=repo) topics = topics_str.split(",") if topics_str else None results = asyncio.run(runner.run(persona_id=persona_id, topics=topics)) # ... 输出结果

CLI 做三件事：

1. 加载配置 + 创建 Repository

2. 创建 PipelineRunner（注入 config 和 repo）

3. 调runner.run()执行

Web 入口

# api/app.py class AppState: def __init__(self, config=None): 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(config=state.config, repo=state.repo) async def _generate(): results = await runner.run(persona_id=req.persona_id, topics=req.topics) return [r.model_dump(mode="json") for r in results] task_id = await state.task_engine.submit(name="内容生成", coro_func=_generate) return ContentGenerateResponse(task_id=task_id)

Web 也做三件事：

1. AppState 初始化 config + repo + cache + task_engine

2. 路由里创建 PipelineRunner（注入 config 和 repo）

3. 提交到 task_engine 异步执行

共享同一套业务模块

CLI 入口 (cli/main.py) ──┐ ├──→ PipelineRunner ──→ persona/ + topic/ + content/ + quality/ Web 入口 (api/app.py) ───┘ ↑ │ Repository ─────────────┘ LLMClient ─────────────┘

两种入口共享同一套业务模块— CLI 和 Web 的区别只在"怎么接收请求"和"怎么返回结果"，核心业务逻辑完全一样。

CLI：命令行参数 → PipelineRunner.run() → rich.Table 输出 Web：HTTP 请求 → 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_model=APIResponse) 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(data=result.model_dump(mode="json")) # 4. 包成统一响应

每个路由函数的四步套路：

1. get_state()— 拿全局状态（config、repo、cache、task_engine）

2. 创建业务 Manager —PersonaManager(state.repo)

3. 调业务逻辑 —mgr.create(persona)

4. 包成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_filter=SensitiveFilter( sensitive_words_file=config.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— 接口统一，编排器才能统一处理。

加一个新检查器的步骤：

1. 新建quality/xxx.py，实现check和fix

2. 在QualityOrchestrator.__init__加一个参数

3. 在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(path=None) -> 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_url=config.llm.base_url, api_key=config.llm.api_key, model=config.llm.model, ) # QualityOrchestrator 用 config 创建 SensitiveFilter self.quality_orchestrator = QualityOrchestrator( sensitive_filter=SensitiveFilter( sensitive_words_file=config.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 persona，persona 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和业务逻辑分离不只是好看，是解决循环依赖的必要手段。数据模型是共享的，业务逻辑是私有的。

坑3：PipelineRunner 膨胀

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 → 虽然长，但逻辑线性，容易读

教训：不是所有长方法都需要拆。如果一个方法的逻辑是线性的（一步步往下走，没有分支），保持长方法比强行拆成小方法更可读。拆方法的目的是消除分支和重复，不是为了短。

坑4：AppState 单例 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

1. 按职责拆模块，通过构造函数注入依赖— 每个模块只做一件事，依赖从外部传入，不在模块内部new。PipelineRunner 是唯一的组装中心，所有对象在这里创建和接线。

2. schema 和业务逻辑分离是解循环依赖的关键— 数据模型（Pydantic BaseModel）放在schema.py，可以被其他模块 import；业务逻辑放在各自的文件里，不被其他业务模块 import。数据共享，逻辑私有。

3. 两种入口共享同一套业务模块— CLI 和 Web 的区别只在"怎么接收请求"和"怎么返回结果"，核心业务逻辑完全一样。加新入口（gRPC、消息队列）只需写入口文件，业务模块不用改。

下篇预告

下一篇：《配置驱动：让Agent灵活适配不同场景》

本文讲了架构怎么从单体演进到模块化，但有个问题没讲：同一个 Agent 怎么服务 100 个不同的赛道？美妆赛道和职场赛道，用同一套代码，怎么做到？

硬编码 → 加"北漂"赛道要改 3 个文件 配置驱动 → 加"北漂"赛道只改 1 个 YAML 文件

下一篇讲配置驱动设计 — YAML 配置层叠、人设模板、选项动态加载、环境变量管理，让同一个 Agent 灵活适配不同场景。