漫剧 Agent 生产系统 · 开发文档

目录

1. 系统概述
2. 架构设计
3. 核心模块详解
4. 数据流与接口定义
5. 技术选型
6. 部署方案
7. 开发进度计划
8. 测试说明
9. 常见问题与排障（FAQ）
10. 扩展指南

1. 系统概述

1.1 定位

把「小说原文 / 题材梗概」自动变成「可发布的竖屏漫剧成片」的 多 Agent 流水线。输入一段文字，输出 final.mp4（含配音、字幕、分镜画面）+ 完整生产留痕（每个阶段的 JSON 产物、所用技能、实际模型）。

产品铁律（不可破，开发写代码前必读）： 1. 永不自动发布——系统只产素材/成片，发布走人工（可衔接 publish_tasks 队列，入队≠发布）。 2. 数字回源——所有对外数字只信工具层/库返回值，不编造。 3. AI 与业务隔离——LLM 只经 llm.complete() 网关，不直连业务库。 4. 单段失败留痕中止——失败不吞错，drama_stage_runs.error 记录后停止，不硬跑后续段。

1.2 功能范围（In Scope）

1.3 不做（Out of Scope，有意排除）

• 自动发布到抖音/快手等平台（走人工 + publish_tasks）
• 多集连续剧的跨集剧情记忆（当前每项目单集独立）
• 精剪级编辑（时间轴微调、转场特效）——用剪映等工具后处理
• 自动划款/结算（CPS 模块只记账不划款，属另一模块）

1.4 与周边系统的关系

小说来源(番茄授权/自写) ──► 【漫剧产线 app/drama】──► final.mp4 + 素材包
                                   │
                                   ├─► publish_tasks（人工确认后发布矩阵）
                                   └─► cps_links/cps_commissions（软广植入变现台账，另模块）

2. 架构设计

2.1 四层架构

2.2 目录结构（服务端）

/opt/dsp-matrix-hq/app/backend/
├── app/
│   ├── main.py                    # 入口（补丁注册 drama_router + create_all 建表）
│   ├── drama/                     # ★ 漫剧产线核心包
│   │   ├── __init__.py
│   │   ├── skill_loader.py        # 产线技能加载（产线与技能中台的唯一集成点）
│   │   ├── agents.py              # 6 个 Agent 定义（指令 + JSON 契约）
│   │   ├── pipeline.py            # 六段编排 + 各段执行函数
│   │   ├── tables.py              # drama_projects / drama_stage_runs
│   │   └── skills/                # 13 份自研技能 SKILL.md（按段分目录）
│   │       ├── script/    SKILL.md + AUX_漫剧创作助手v3.md + AUX_短视频脚本大师.md
│   │       ├── storyboard/ SKILL.md + AUX_漫剧分镜技能.md + AUX_漫剧制片浪客.md
│   │       ├── character/ SKILL.md + AUX_3d影视动画漫画生成.md
│   │       ├── voice/     SKILL.md
│   │       ├── video/     SKILL.md + AUX_漫剧全链路制作.md
│   │       └── qc/        SKILL.md + AUX_多平台违禁词查询.md
│   └── routers/drama.py           # REST API
└── web/drama.html                 # 产线控制台 SPA

产物目录（数据，不入库）：

/opt/dsp-matrix-hq/data/drama/<project_id>/
├── characters/   定妆照 <角色名>.png（1024x1536）
├── voice/        d1.mp3, d2.mp3, ...（TTS 逐句）
├── video/        bg.mp4（出画拼接）/ subs.ass（字幕）/ final.mp4（成片）
├── prompts/      <stage>-<时间戳>.txt（完整 prompt 快照，可复现留痕）
└── placeholder.png（无定妆照时的兜底图）

2.3 进程与并发模型

• 单进程 FastAPI（uvicorn 单 worker），产线跑在 BackgroundTasks 后台任务。
• 单项目同时只允许一条产线在跑（status=running 时不重复触发）；不同项目可并行。
• SQLite WAL + session_scope() 短事务；drama_stage_runs 每段一记，写入间互不争锁。
• LLM/媒体调用是 IO 等待型，用 asyncio.to_thread 包住同步媒体函数（generate_image/generate_bg_video/compose）。

3. 核心模块详解

3.1 skill_loader.py — 技能加载器（集成点）

这是产线与「总部自研技能中台」的唯一集成点。 技能本质是角色提示词包（SKILL.md 全文）。

load_bundle(stage: str) -> str
# 返回该段完整角色提示词 = 主技能 SKILL.md 全文 + 各 AUX_*.md（带【辅助技能】分隔头）
# 抛 ValueError(stage 非法) / FileNotFoundError(技能目录缺失)

台账常量： - STAGE_ORDER = ("script", "storyboard", "character", "voice", "video", "qc") - STAGE_SKILLS: dict[stage → {agent, primary, aux[]}] —— 供 /api/drama/skills 与前端展示。

调用方式（agents.build_messages）：system = load_bundle(stage) + 【本节点任务】instruction + 【输出 JSON 契约】output_spec + 【输出铁律】，user = 上游产物 JSON。注入 llm.complete(messages=...)。

替换/升级技能：直接覆盖 skills/<stage>/SKILL.md 或增删 AUX_*.md，重启服务即生效（load_bundle 每次读盘）。文件来源：八仙矩阵总部自研技能（source: baxian-matrix）。

技能台账（当前部署）：

3.2 agents.py — Agent 定义与输出契约

数据类 DramaAgent{key, name, tier, max_tokens, instruction, output_spec}；AGENTS 字典注册 6 节点。tier 传给 LLM 网关选档（large=旗舰，medium=快省）。

输出铁律（拼进每条 system）：只输出一个 JSON 对象、不要 markdown 代码块、字段必须存在、拿不到留空字符串、绝不编造。

各段 JSON 契约（output_spec，开发对接/测试断言以此为准）：

script（tier=large, max_tokens=7000）

{"title":"剧名(≤12字)","logline":"一句话梗概(≤40字)","hook":"开头钩子句",
 "endcard":"片尾字卡(≤8字)","style_tone":"风格基调(≤20字)",
 "characters":[{"name":"","desc":"人物小传(≤40字)","voice_role":"narrator|hexiangu|lvdongbin|yuan|xiaoyu"}],
 "scenes":[{"scene_id":1,"setting":"场景(≤15字)","action":"动作描述(≤40字)",
   "dialogues":[{"role":"","line":"台词(≤18字)","emotion":"happy|sad|angry|fearful|surprised|neutral"}]}]}

storyboard（large, 7000）

{"visual_style":"全局画风锚定(≤60字)",
 "shots":[{"shot_id":1,"scene_id":1,"duration":5,"camera":"景别与运镜(≤15字)",
   "visual_prompt":"视频提示词(60~120字)","subtitle_hint":"对应台词"}]}

约束：每场 1~3 镜、全片 3~6 镜、每镜 4~8 秒；提示词只写可见元素（主体+动作+环境+镜头+光线+画风），禁止台词文字与品牌名。

character（medium, 4000）

{"style_prompt":"全局画风(≤40字)",
 "characters":[{"name":"","voice_role":"音色键",
   "visual_anchor":"视觉锚定(50~100字：发色/瞳色/服装/配饰/体型)",
   "portrait_prompt":"定妆照提示词(≤80字)"}]}

visual_anchor 是全片角色一致性的关键——video 段每条提示词都会复述它。

voice（medium, 5000）

{"lines":[{"role":"音色键","text":"配音句(≤18字，用…表示停顿)",
   "emotion":"happy|sad|angry|fearful|surprised|neutral"}]}

video（medium, 4000）

{"prompts":["最终镜头提示词(80~150字) = 角色锚定 + 本镜动作 + 竖屏9:16 + 画风 + 画面无文字"],
 "total_seconds":30}

qc（medium, 4000）

{"pass":true,"score":85,
 "issues":[{"level":"error|warn|info","stage":"script|storyboard|voice|video",
   "text":"问题描述","suggestion":"整改建议"}],
 "compliance_note":"一句话合规结论"}

3.3 pipeline.py — 六段编排

主编排：

async def run_project(project_id: int, from_stage: str | None = None,
                      to_stage: str | None = None) -> dict
# 返回 {"project_id", "stages":[{"stage","status","model","error","seconds"}], "status"}

执行语义（测试必读）： 1. 阶段序列 = STAGE_ORDER 切片 [from..to]（都可空 = 全链）；from_stage > to_stage 顺序非法 → 400 拒绝。 2. 上游产物复用：开跑前从 drama_stage_runs 里取每段最近一条 status=ok 的 output_json 装进 ctx——所以可跳段跑（如只补跑 video），但上游缺产物的段会返回 error 而非抛异常（防御 .get，2026-10-31 修复）。 3. 并发与残留防护（2026-10-31 加固）： - 触发端用原子条件 UPDATE 抢占（status!='running' OR updated_at<超时线），杜绝「查完再改」竞态双跑；running 且新鲜 → 409，running 但超时（>30 分钟）→ 允许接管重跑； - 每段开工刷新 updated_at（心跳），长段出画属正常窗口不会被误接管； - 进程启动即清扫（路由模块 import 时执行）：BackgroundTasks 随进程消亡，遗留 running 一律复位 error（段记录标注「进程重启中断」）——防重启后项目永久卡死。 4. 每段执行：先插 status=running 记录 → 跑 _STAGERS[st](ctx) → ok 则写 output_json/model/artifact_path；异常则写 error = "ExcType: msg[:400]" 并中止后续段。 5. 可复现留痕：每段把完整 prompt（system=技能包全文+指令+契约 / user=上游产物）快照到 <project_dir>/prompts/<stage>-<时间戳>.txt；input_json 记录 skill_hash（技能包内容指纹）——换技能后历史运行仍可溯源。 6. 收尾回写 drama_projects：全部 ok → status=done，result_video ← video 段 video_path，result_summary ← qc 段 compliance_note，标题未命名时用 script.title；任一段失败 → status=error。

ctx（阶段间数据总线）结构：

ctx = {
  "project_id": int, "source_text": str, "style": str,   # 项目初始
  "script": {...}, "storyboard": {...}, "character": {...},
  "voice": {...}, "video": {...}, "qc": {...},           # 各段 output_json 解析后
}

各段执行函数行为：

_run_llm_stage(agent, user_content) 统一封装：llm.complete(build_messages(...), tier, temperature=0.6, max_tokens=agent.max_tokens) → _parse_json_obj(res.text)。LLM 不可用时返回 {"error":"LLM 不可用（未配 Key）"} 而非崩溃。

同步包装 run_project_sync(project_id, **kw) 供脚本/测试直接调。

3.4 tables.py — 数据表

建表机制：SQLModel 定义 → main.py import 注册 metadata → db.init_db() 的 create_all 自动建（加列走 _auto_add_columns 自动 ALTER，不用写迁移）。

drama_projects（一部剧一条）：

drama_stage_runs（每段执行一条，全程留痕）：

3.5 routers/drama.py — REST 层

见 §4.2。安全约定：全部端点经 bxsid 会话 + current_user/tenant_of 鉴权、租户隔离；写操作由主入口 router_write_guard 纳管；文件下载防路径穿越（拒 /、\、..）。

3.6 web/drama.html — 产线控制台

单页 SPA：项目列表 → 建项目（题材粘贴/画风选择）→ 一键全链 / 分段跑 → 六段进度卡（状态/模型/耗时/错误）→ 产物查看（成片播放、定妆照、配音单、质检报告）→ 技能台账。纯静态，挂 /static/drama.html，API 走同域 /api/drama/*。

4. 数据流与接口定义

4.1 全链数据流

source_text ──[script]──► 剧本JSON ──┬──[storyboard]──► 分镜表 ──────────┐
                                     ├──[character]──► 角色锚定+定妆照 ──┤
                                     └──[voice]──► 配音单 ──TTS──► .mp3 ─┤
                                                                          ▼
                                     [video] = 分镜prompts×角色锚定 → 出画 → 拼接
                                              → 时间轴+字幕ass+compose → final.mp4
                                                                          ▼
                                     [qc] = 全链摘要 → 违禁词/限流/侵权体检 → 报告

LLM 产物一律走「JSON 契约 → _parse_json_obj → ctx」；媒体产物一律走 media_service 落盘到 /data/drama/<id>/<stage>/。

4.2 REST API（/api/drama，全部需登录 bxsid）

curl 示例（服务器本机，测试用免登 session 见 §8.1）：

# 建项目
curl -s -X POST -H 'Content-Type: application/json' -H "Cookie: bxsid=$T" \
  -d '{"title":"测试剧","source_text":"少年在长安城追查画灵奇案…","style":"国漫"}' \
  http://127.0.0.1:8010/api/drama/projects

# 全链跑
curl -s -X POST -H 'Content-Type: application/json' -H "Cookie: bxsid=$T" \
  -d '{}' http://127.0.0.1:8010/api/drama/projects/1/run

# 只补跑 video 段（上游产物自动复用）
curl -s -X POST -H 'Content-Type: application/json' -H "Cookie: bxsid=$T" \
  -d '{"from_stage":"video","to_stage":"video"}' \
  http://127.0.0.1:8010/api/drama/projects/1/run

# 查状态 / 下载成片
curl -s -H "Cookie: bxsid=$T" http://127.0.0.1:8010/api/drama/projects/1
curl -s -H "Cookie: bxsid=$T" -o out.mp4 \
  'http://127.0.0.1:8010/api/drama/projects/1/file?name=final.mp4'

4.3 内部接口（媒体层，media_service）

video/voice/character 段复用的底座函数（签名以生产代码为准）：

async def synth_lines(lines: list[dict], workdir: Path) -> list[dict]
    # lines=[{"text","role","emotion"}]；全引擎降级：豆包seedtts→edge-tts→piper→静音
    # 返回 [{path, seconds(实测时长), ...}] 供 plan_timeline 排时间轴

def plan_timeline(items, total=30.0, endcard="") -> rows   # 配音排布 + 尾部字卡
def build_ass(rows, out_path: Path, title=...) -> Path     # 生成 ASS 字幕
def compose(rows, avatar: Path, out_mp4: Path, ..., total, ass, bg) -> None
    # ffmpeg 合成：画面(出画/静帧) + 配音轨 + 字幕 → final.mp4

def generate_bg_video(prompts: list[str], workdir: Path, total: float) -> Path
    # 逐镜出画 + 拼接；通道降级链：秘塔(METASO) → MiniMax → 可灵(KLING)

def generate_image(prompt, out_path: Path, size="1024x1536") -> None   # 定妆照
def image_generation_available() -> bool
def video_generation_available() -> bool

VOICES 音色表（media_service.VOICES）：narrator / hexiangu / lvdongbin / yuan / xiaoyu 等键 → 不同 TTS 音色+风格；未知键回退 DEFAULT_ROLE。

4.4 LLM 网关接口

from ..agent import llm
res = await llm.complete(messages, tier="large"|"medium",
                         temperature=0.6, max_tokens=N,
                         offline_fallback='{"error":"...","data":{}}')
res.text, res.model   # 文本与实际命中模型名（落 drama_stage_runs.model）

供应商降级链（llm.PROVIDERS）：deepseek → doubao（含 seed-2-1-lite）→ qwen（百炼）。tier=large 用旗舰档，medium 用快省档。

5. 技术选型

选型原则：复用 > 新建；确定性代码（合成/时间轴/落盘）与不确定性 LLM（剧本/分镜/质检）严格分界——LLM 只产 JSON，文件产物一律代码生成。

6. 部署方案

6.1 一键部署（可复现）

本地暂存目录 D:\八仙矩阵总部\_drama_deploy\，核心脚本 deploy_drama.py：

cd D:\八仙矩阵总部\_drama_deploy
python deploy_drama.py push      # 推送 20 个文件（代码+13 份 SKILL.md+drama.html）
python deploy_drama.py patch     # main.py 补丁：注册 drama_router + import tables（带时间戳备份）
python deploy_drama.py restart   # systemctl restart dsp-hq.service
python deploy_drama.py all       # 以上三步

推送机制：paramiko SSH + base64 传文件（SFTP 对 root 不可用）；Windows 下传 /opt 路径必须加 MSYS_NO_PATHCONV=1（Git Bash 会把 /opt 转成 Windows 路径）。本地辅助工具：_test/ssh_run.py（执行远程命令）、_test/ssh_file.py（推文件）。

6.2 服务纳管

systemctl restart dsp-hq.service
systemctl is-active dsp-hq.service          # 期望 active
curl -s 127.0.0.1:8010/api/health           # 健康检查
curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:8010/static/drama.html  # 期望 200

日志：journalctl -u dsp-hq -o short-iso --since '10 min ago'（app.log 已弃用，一律走 journal）。

6.3 配置

零新增密钥——LLM/TTS/出画全复用矩阵总部既有通道（.env 既有配置）。换出画通道只改 .env 的 METASO/MINIMAX/KLING Key（video_generation_available() 自动探测）。产物根目录 ARTIFACT_ROOT = /opt/dsp-matrix-hq/data/drama（改路径动 pipeline.py 顶部常量）。

6.4 回滚

patch 前自动备份 main.py.bak-<时间戳>；回滚 = 恢复备份 + 重启。文件级回滚：用 git/暂存目录旧版重新 push。表结构只加不删，无需迁移回滚。

7. 开发进度计划

7.1 已完成（2026-10-31，v1.0 上线）

7.2 迭代路线（v1.x，按优先级）

说明：以上为产品规划节奏（非项目排期承诺）；每项上线前走冒烟清单（§8）。

8. 测试说明

8.1 免登冒烟手法（测试专用）

平台 API 需登录。测试可直接往 sessions 表插一行（created_at 必填 NOT NULL），cookie 名 bxsid：

TOKEN=smoke-$(date +%s)
EXP=$(date -u -d '+2 hours' '+%Y-%m-%d %H:%M:%S'); NOW=$(date -u '+%Y-%m-%d %H:%M:%S')
sqlite3 /opt/dsp-matrix-hq/data/hq.db \
  "INSERT OR REPLACE INTO sessions(token,user_id,tenant_id,expires_at,created_at) VALUES('$TOKEN',1,1,'$EXP','$NOW');"
# 用完清理：DELETE FROM sessions WHERE token LIKE 'smoke-%';

8.2 验收基准（v1.0 实测记录，项目#1《长安画灵疑案》）

8.3 测试用例清单（回归用）

T13~T18 由 smoke_drama3.py 覆盖（2026-10-31 实测 13/13 PASS，含 T0 建项目与收口断言）。

8.4 自动化回归套件（pytest，已上线）

# 服务器上运行（venv 已装 pytest 9.x）：
/opt/dsp-matrix-hq/venv/bin/python -m pytest /opt/dsp-matrix-hq/tests/test_drama.py -v
# 跳过 LLM 调用（纯静态用例）：SKIP_LLM=1 ...

26 个用例 = 单元层（JSON 截断抢救/技能包/契约/护栏真值表）+ 鉴权校验（T4~T7/T13/T14）+ 生命周期（T2/T3/T11/T18，含少量 LLM）+ 并发护栏（T15/T16/run_count 上限）+ 成本护栏（video 重跑 force）+ 清扫（T17/T17b 活跑不误杀）。默认不出画不花钱；全链出片（T1/T9/T10）为计费人工用例，不进自动化。2026-10-31 实测 26/26 PASS（3m49s）。

套件在测试中抓出并已修复的真 bug：①_parse_json_obj 截断抢救对「有键无值」尾巴失效+补括号嵌套顺序错（现按栈逆序补全）；②bg 写回对已删行不判空（AttributeError 打崩后台任务）；③清扫误杀活跑（现双条件：启动前遗留 且 行龄>120s）；④重跑用例撞「新行未插入」竞态（wait 需 after_id 基线）。

8.5 本地静态检查（提交前必跑）

cd D:\八仙矩阵总部\_drama_deploy
python -m py_compile app/drama/*.py app/routers/drama.py   # Python 语法
# 注意：服务器 Python 3.10 —— f-string 内禁反斜杠表达式！本地 3.12 能过不代表服务器能过

9. 常见问题与排障（FAQ）

Q1. 服务起不来 / import 报错？ 看 journalctl -u dsp-hq -o short-iso | tail -30。历史坑：① routers/drama.py 引用产线包必须 from ..drama import ...（相对导入两层，写成 .drama 会 ImportError）；② f-string 里不能有反斜杠（Python 3.10）——正则类表达式提出来先赋值。

Q2. 某段卡住很久不出结果？ 先查库：SELECT stage,status,datetime(started_at) FROM drama_stage_runs WHERE project_id=N ORDER BY id DESC LIMIT 3;。character/video 段含出图出画（单镜 20~60s，4 镜可到 3 分钟），属正常窗口，勿急着重启。journal 无异常 + 文件在增长 = 在跑。真卡死（>30 分钟无心跳）时再 run 会自动接管重跑。

Q2b. 服务重启后项目显示 running 但跑不动？ 旧版本会永久卡死；现在启动清扫会自动把遗留 running 复位为 error（段记录标注「进程重启中断」），直接重新 run 即可。若仍见 running 且 updated_at 新鲜但确信已死，等 30 分钟超时接管或手工置 error。

Q3. 跳段跑报「缺少分镜产物」？ 预期行为（T3）：上游无 ok 产物时该段返回 error 不硬跑。先补跑上游段，或全链跑。

Q4. video 段有 video_error: 未配置视频生成通道？ .env 缺 METASO/MINIMAX/KLING Key（video_generation_available() 为 False）。系统降级出「配音+字幕版」成片。配好任一通道 Key 重启即可。

Q5. 定妆照没出？ image_generation_available() 为 False 或出图超时——character 段不阻塞，portrait_error 留痕，video 段自动用 placeholder.png 兜底。

Q6. LLM 输出解析失败 / 字段缺失？ _parse_json_obj 已容错代码块与杂文；仍有缺字段段内 setdefault 兜底（title/scenes/shots/characters/lines/issues）。持续失败查 model 是否命中降级到离线 fallback（res.model 落在 stage_runs 里可查）。

Q7. 改了 SKILL.md 不生效？ load_bundle 每次读盘，但运行中产线已把 system 消息发给 LLM——新项目/重跑该段才用新技能。若完全不生效，确认文件在 app/drama/skills/<stage>/ 且文件名为 SKILL.md（主）或 AUX_*.md（辅）。

Q8. 重跑某段会覆盖旧产物吗？ DB 是追加式（每次新插 drama_stage_runs，取最近 ok）；磁盘产物按段目录覆盖写。成片要看历史版本请先拷走。

Q9. SSH 部署时路径变成 C:/Program Files/Git/...？ Git Bash 的 MSYS 路径转换。所有 ssh_run/ssh_file 调用前缀 MSYS_NO_PATHCONV=1。

Q10. 单集成本大概多少？ LLM 六段约 2~6 万 token（deepseek/doubao 档）；出画 4 镜走秘塔约 ¥3~5/集；TTS edge-tts 免费。合计 ¥5 以内/集（不含人工审片）。

10. 扩展指南

10.1 换/加技能（不改代码）

覆盖 skills/<stage>/SKILL.md 或新增 AUX_xxx.md → 重启（或下次跑段生效）。建议同步更新 skill_loader.STAGE_SKILLS 的展示名（台账）。

10.2 加一个新产线段（如「配乐」）

1. skills/music/SKILL.md 放技能；
2. agents.py 加 AGENTS["music"]（instruction+output_spec）；
3. pipeline.py 加 _stage_music(ctx) 并注册 _STAGERS、插入 STAGE_ORDER（skill_loader）；
4. 上下游段从 ctx 取产物（用 .get 防御）；
5. 补 §8.3 用例。

10.3 换 LLM 档位/模型

只动 agents.py 的 tier/max_tokens；模型名单在 agent/llm.py::PROVIDERS，不建议产线直改。

10.4 接自动发布（谨慎）

原则「永不自动发布」。可做的最大步：video 段完成后把成片入 publish_tasks 队列（入队≠发布，发布仍人工确认）。接口对接见矩阵总部 publish_service。

10.5 对接变现（CPS 软广植入）

成片可关联 cps_links.drama_project_id（placement_type 承载植入方式），走 CPS 模块建挂载链接 + 分佣台账（见《漫剧商业变现与推广执行手册》第六章）。

附录 A · 关键文件索引

附录 B · 术语

• 段（stage）：产线六步之一（script/storyboard/character/voice/video/qc）
• ctx：段间数据总线（dict，上游产物 JSON）
• 视觉锚定（visual_anchor）：角色外貌定型描述，全片提示词复述以保一致性
• 出画：文本提示词→AI 视频帧段（render_ai_clip 通道链）
• 成片：video 段 final.mp4（画面+配音+字幕）