yingqing/core/docs/脚本Agent编排架构方案-动态知识装配.md

脚本 Agent 编排架构方案 · 动态知识装配（流式管道）

状态:设计方案(未落地代码)　·　作者:架构评审　·　日期:2026-06-24 关联代码:apps/ai/script_agent.py · apps/ai/services.py · skills/ecommerce-video-script/ 关联文档:脚本Agent流式SSE技术文档.md(现状) · 出格式实测-模型产出汇总.md(实测数据)

---

1. 背景与要解决的问题

1.1 现状

当前脚本生成把领域知识一次性全量灌入上下文:load_ecommerce_skill()(见 script_agent.py:54) 用 glob("*.md") 无条件遍历全部 references,整篇拼成系统提示词,每次对话(不论全自动/一句话/改稿/改一镜) 都把这 ~28K 字符全量发给模型。

SKILL.md 里虽写了一张「输入模式路由 / 参考资料索引」表(指明哪个品类该读哪几篇),但该路由仅作为提示词 发给模型,后端并未按它选择性加载——检索发生在模型的注意力里,不在后端。

1.2 随业务增长的问题

核心痛点:上下文随电商品类数量线性膨胀。

现在 5 个 references = 28K 字符。未来叠加更多品类话术(美妆/食品/3C/服饰/家居/母婴/宠物/家电…) 与平台调性后,全量灌入会:

• 上下文臃肿:单次请求 system prompt 可能涨到数十万字符,逼近/超出上下文窗口;
• 成本线性上涨:每次都付全量知识的 token,即便这单商品只用得上其中一个品类包;
• 注意力稀释:无关品类的话术挤占模型注意力,可能拉低相关品类的发挥;
• 缓存难命中:动态拼接的大 prompt 难以稳定复用 prefix cache。

1.3 目标

把「静态全量灌入」改为「按需动态装配」:拿到商品需求后,只加载与该商品相关的知识模块, 打包给模型,流式生成,再经过滤/提取输出前端。单次上下文只随"命中的 1-2 个品类包"走,不随品类总量膨胀。

---

2. 设计目标(验收标准)

#	目标	可度量标准
G1	上下文不随品类总数膨胀	单次 system prompt 字数 ≈ 内核 + 命中模块,与品类总数解耦
G2	输出格式永不因知识缺失而塌	任意路由结果下,输出契约恒在 prompt 中
G3	全程流式	路由/装配/生成/提取每阶段都有 SSE 进度事件
G4	运营可扩品类不改代码	加一个品类 = 加一个知识模块(文件/DB),无需改 Python
G5	不引入与任务不匹配的重型框架	沿用生成器流式编排,不上状态图运行时
G6	平滑迁移	分阶段落地,每阶段可独立上线、可回滚

---

3. 总体架构:六段流式管道

商品需求(product + 前置条件)
  │
  ▼ ① 路由 Router ────────── 识别品类/平台 → 决定加载哪些知识模块
  │   SSE: tool router  "识别品类:美妆洗护 · 平台:抖音"
  │
  ▼ ② 装配 Assembler ─────── 取「内核 + 命中品类包 + 命中平台调性」
  │   SSE: tool assemble "装配知识:内核+2模块 共 9.2K 字"   ← 可见地证明未膨胀
  │
  ▼ ③ 打包 Packager ──────── 内核(恒在) + 动态知识 + 商品上下文 + 输出契约
  │
  ▼ ④ 生成 Generator ─────── 约束解码(tool/structured)流式出结构
  │   SSE: reasoning(思考) / delta(口语前言)
  │
  ▼ ⑤ 过滤提取 Extractor ─── 校验/归一/抽实体/扫违规词(强不变量,后端兜底)
  │   SSE: tool extract "提取实体4个 · 自检通过"
  │
  ▼ ⑥ 前端 Sink ──────────── draft / saved / summary / done

与现状的本质差异:②③ 从"glob 全部"变为"只装命中"。①②⑤ 是真实工作步骤,不再是装样子的工具卡。

---

4. 核心设计:知识分两层

把现有单块 28K 知识拆成**内核(恒在)+ 模块(动态)**两层。

4.1 内核 Kernel(每次必带,小而稳)

内容	来源(现状)
黄金结构(钩子→痛点→卖点→CTA)、档位×结构映射	methodology.md 的结构部分
输出契约(铁律1):字段名锚定、JSON 形状、镜数规则	SKILL.md 铁律1 + _OUTPUT_PROTOCOL
写作红线:≤55字、违规词清单、口语化	methodology.md 红线 + SKILL.md 铁律3
字段纪律:tone/role 枚举、entity 引用合法性	SKILL.md 铁律1

⚠️ 输出契约必须在内核里,永远在。 这是你们踩过的坑(skills 没进镜像→契约丢失→模型吐散文解析失败) 的正式解。无论路由加载了哪些品类包,格式硬底线都不会塌。现有 _EXTRACT_OUTPUT_CONTRACT (见 services.py:385)写死兜底,就是这一思想的雏形——把它正式化为"内核"。

4.2 品类模块 Module(按商品命中才加载,多而长)

模块类型	例	frontmatter 选择维度
品类话术	美妆/食品/3C/服饰/家居…	applies_to: [category...]
平台调性	抖音/快手/小红书/视频号	platforms: [...]
钩子库分册	痛点提问/反差/数字冲击…	tone: [...] 或 always

每个模块是一个独立的、带元数据索引的知识单元(不再是一坨大 concat)。

---

5. 路由机制:怎么选模块

三种机制,按本场景适配度排序。推荐 rule-first 混合。

5.1 元数据路由(主力 · 先落地这个)

把 SKILL.md 的路由表从提示词搬进代码:每个模块 frontmatter 声明它服务的品类/平台, select_knowledge(product) 按 product.category / 平台前置条件命中。

• 优点:零额外调用、确定性、可解释、可单测。电商商品基本都有 category 字段,80% 情况足够。
• 缺点:新品类要维护映射——但加品类 = 加一个 .md 模块 + 写 frontmatter,不改 Python(满足 G4)。

5.2 向量检索 RAG(扩容兜底 · 品类破百再上)

把知识块 embedding,用商品上下文检索 top-K 相关片段。

• 优点:处理模糊/新品类(novel category 自动匹配近邻),可无限扩。
• 缺点:引入检索失败模式(检错块→知识缺失)、需 embedding 基建与运维。别过早引入。

5.3 LLM 路由(灵活但加跳)

用便宜快模型(如 doubao-lite)先分类"该商品属哪类、用哪套话术"。

• 优点:最灵活,能理解复杂商品描述。缺点:多一次调用 + 延迟。

5.4 推荐:rule-first 混合

select_knowledge(product):
    modules = [Kernel]                      # 恒在
    hit = rule_match(product.category, platform)   # 5.1 规则命中
    if hit:
        modules += hit
    else:
        modules += fallback()               # 命中不到:回落(全量核心包 or 5.2 检索)
    return modules

先只做 5.1 规则版,留好 fallback() 接口;品类规模或模糊度上来时,把 fallback 换成检索/LLM 路由。

---

6. 接口设计

6.1 知识模块结构(frontmatter 规范)

每个 reference 模块在文件头加 YAML frontmatter(或等价 DB 字段):

---
id: playbook-beauty
type: category            # core | category | platform | hook
applies_to: [美妆, 护肤, 洗护, 彩妆]   # 命中这些 category 时加载
platforms: []             # 限定平台(空=不限)
keywords: [精华, 面膜, 口红, 防晒]     # 检索/模糊命中用
priority: 10
enabled: true
---
（正文:该品类的话术、语气、卖点侧重…）

core 类型 = 内核,恒加载;其余按 applies_to/platforms/keywords 命中。

6.2 选择函数(替换 load_ecommerce_skill)

# apps/ai/knowledge.py(新增)
@dataclass
class KnowledgeModule:
    id: str
    type: str               # core|category|platform|hook
    applies_to: list[str]
    platforms: list[str]
    keywords: list[str]
    body: str

def load_registry() -> list[KnowledgeModule]:
    """扫 skills/ 下模块(含 frontmatter),或读 DB。缓存。"""

def select_knowledge(*, product, platform: str | None, mode: str) -> list[KnowledgeModule]:
    """rule-first:内核恒在 + 按 category/platform 命中品类包/平台调性;
    命中不到走 fallback(全量核心 or 检索)。改稿模式可少带选题类模块。"""

def assemble_system_prompt(modules: list[KnowledgeModule]) -> tuple[str, int]:
    """拼 system prompt = 内核(置顶稳定,利于 prefix cache) + 动态模块。
    返回 (prompt, 字数) —— 字数用于 SSE 上报,可见证明未膨胀。"""

build_agent_messages(见 script_agent.py:111) 的 system = load_ecommerce_skill() + _OUTPUT_PROTOCOL 改为 system, n = assemble_system_prompt(select_knowledge(...)),其中输出契约并入内核。

6.3 SSE 事件扩展

在现有 tool/reasoning/delta/draft/saved/summary/done 基础上,让 ①②⑤ 成为真实工具卡:

事件	新增/变化	载荷
tool: router	新增	{label:"识别品类:美妆·抖音", status, meta:{category, platform}}
tool: assemble	新增	{label:"装配知识 内核+2模块 9.2K字", status, meta:{module_ids, chars}}
tool: generate	不变	约束解码流式
tool: extract	强化	真实体提取 + 违规词自检结果

assemble 卡把"这次只装了 9.2K 而非 28K"可观测地展示给用户/运维,是 G1 的活体证明。

---

7. 生成与过滤提取(④⑤)

7.1 生成:约束解码,让 normalize 退居安全网

结合 出格式实测-模型产出汇总.md 的实测结论:

• 三模型(豆包/GPT-5.5/Gemini-3.1-pro)的 structured/tool 均可产出 raw_clean✓ 的契约 JSON;
• freeform 下三家原始输出全 raw_clean✗(靠 normalize_draft fuzzy 抢救);
• tool 策略跨模型 segments 键集完全同构(最稳)。

→ 生成阶段改用 tool/structured 约束解码(schema 须补全 dialogue/speaker/voice_ref 等契约字段), 内核保留输出契约文字作双保险。normalize_draft 从"主力解析器"降为"安全网"。

⚠️ 约束解码与"先写口语前言"的 _OUTPUT_PROTOCOL 有张力(实测 GPT structured 被前言污染)。 落地时对话气泡(前言/收尾)与结构稿分离:结构走纯约束,气泡另起轻量一跳或用支持混合流的部件协议。

7.2 过滤提取:强不变量后端兜底

沿用并简化现有逻辑(约束解码后原始已干净,兜底压力骤降):

• 镜数对齐(=时长/15)、role/tone 枚举归一、entity_refs 合法性 —— normalize_draft 现有能力;
• 实体抽取(角色/场景)—— 复用 services.py 的 run_extract_entities_task;
• 违规词自检 —— 可前置为真校验节点(发现即标记/可触发重生成)。

---

8. 为什么不用 LangGraph

本管道是线性流水线(router→assemble→generate→extract→sink):无环、无 reflect-retry、无多 agent 对话。 线性 + 流式正是现有生成器范式的最佳 altitude。LangGraph 的状态图是为"有环/有分支/要回退/human-in-loop" 设计的,此处上图属过度设计。真正的编排升级点是"选择性装配"这一层抽象,而非更换运行时。

若未来产品要做「生成→自检违规词→自动修正→再检」的真闭环,或「编剧/审查/提取」多 agent 协作, 那时再评估 LangGraph / PydanticAI(类型契约+重试)/ 轻量自写 retry。当前不需要。

---

9. 分阶段迁移清单

每阶段可独立上线、独立回滚。

Phase 1 · 知识分层 + 规则路由(止血膨胀,优先级最高)

• references 加 frontmatter(type/applies_to/platforms/keywords);抽出 core 内核。
• 新增 apps/ai/knowledge.py:load_registry / select_knowledge(规则版)/ assemble_system_prompt。
• build_agent_messages 改用 assemble_system_prompt(select_knowledge(...));输出契约并入内核。
• SSE 增 router/assemble 真实工具卡(含字数)。
• 单测:命中/未命中/改稿模式各自加载了哪些模块;内核必含契约。
• 验收:单次 system prompt 字数与品类总数解耦(G1);格式零回归。

Phase 2 · 约束解码(让 normalize 退居安全网)

• 定义完整 ScriptDraft JSON Schema(含 dialogue/speaker/voice_ref)。
• 生成阶段切 tool/structured(按 provider 能力分流,实测已验证三家可行)。
• 对话气泡与结构稿分离,解决 _OUTPUT_PROTOCOL 与约束的张力。
• normalize_draft 降级为兜底;保留以防个别模型/中转站不合规。
• 验收:三模型原始输出 raw_clean✓;normalize 命中率(需抢救比例)大幅下降。

Phase 3 · 检索扩容(品类规模化后才做)

• 知识块 embedding + 向量库;fallback() 接入检索。
• 模糊/新品类召回评估。
• 验收:新增品类无需改路由规则即可被正确召回。

Phase 4(可选)· 自检闭环

• 违规词/字数校验做成真节点,不过则带错误自动重生成(轻量 retry,非全图)。

---

10. 风险与对策

风险	对策
R1 内核漏放契约 → 某品类下格式塌	内核必含完整输出契约;单测断言"任意路由结果都含契约";保留写死兜底。
R2 路由漏召(该加载却没加载)→ 模型瞎编	漏召比误召危险。规则命中不到必须回落(全量核心包 or 检索),严禁裸奔。
R3 约束解码与对话气泡冲突	结构稿走纯约束、气泡分离(见 7.1);或用支持混合流的部件协议。
R4 prefix cache 未命中,内核成本没摊薄	内核置顶且稳定;实测豆包/中转是否支持 prefix cache 再定。
R5 检索引入新失败模式	Phase 3 才上;上之前用规则兜底;检索结果可解释、可回退规则。

---

11. 与现有代码映射(速查)

设计组件	现状	落点
内核 + 模块拆分	load_ecommerce_skill() glob 全部	新 apps/ai/knowledge.py
路由 select_knowledge	SKILL.md 路由表(给模型看)	新 knowledge.py,规则实现
装配 assemble_system_prompt	字符串拼全部	新 knowledge.py,内核+命中
打包	build_agent_messages system 拼接	改 script_agent.py:122
生成(约束解码)	freeform + _OUTPUT_PROTOCOL	改 provider 调用,见 providers/
过滤提取	normalize_draft 当主力	降为安全网 script_agent.py:307
流式编排	stream_script_agent 生成器	沿用,增 router/assemble 事件 script_agent.py:561
模块运营管理	references/*.md 文件	frontmatter 文件,或复用 PromptTemplate admin DB 模式

---

12. 一句话总结

把「静态全量灌」改成「内核恒在 + 品类模块按需装配」的线性流式管道:路由先用规则(SKILL 路由表搬进代码)、 扩容再上检索;生成换约束解码让 normalize_draft 退居安全网;流式编排沿用现有生成器,不需要 LangGraph。 如此品类再叠,单次上下文也只随"命中的一两个包"走,不随品类总量膨胀。