yingqing/core/docs/脚本Agent流式SSE技术文档.md

脚本生成 Agent · 流式 SSE 编排技术文档

对象代码：core/backend/apps/ai/script_agent.py 的 stream_script_agent() 端点：POST /api/projects/{id}/script-agent-stream/ → text/event-stream 一句话定位：把一次「调大模型出脚本」的过程，包装成一条可见的、可计费的、可断点回滚的 SSE 事件流，给前端真 agent 体感，同时由后端而非模型保证结构化结果的可靠性。

---

0. 全局视角

前端 fetch(POST script-agent-stream)  ──SSE──▶  浏览器逐帧消费
                │                                      ▲
                ▼                                      │ data: {json}\n\n
        Django StreamingHttpResponse                   │
                │ 包裹                                  │
                ▼                                      │
        stream_script_agent()  ← 同步生成器，每 yield 一帧
                │
   ┌────────────┼─────────────────────────────────────────┐
   │            │                                          │
 加载skill   建AITask+预扣额度                        调豆包流式SSE
 (tool)      (reserve_credit)                        (reasoning/delta)
   │            │                                          │
   └──▶ 抽取JSON+规范化 ──▶ 落库ScriptVersion ──▶ charge额度 ──▶ saved/summary/done

核心思想三条：

1. 进度即事件：内部每一步（加载技能 / 分析商品 / 生成分镜 / 提取实体 / 自检）都吐一张「工具卡」，让用户看到 agent 在干活，而不是对着一个转圈等几十秒。
2. 结构化结果由后端兜底：模型只管「生成」，JSON 的抽取、字段对齐、镜数补齐全在后端做，不信任模型的排版纪律。
3. 计费与流式生命周期绑定：额度预扣（reserve）→ 成功结算（charge）/ 失败或断连释放（release），用 try/finally 覆盖包括客户端断连在内的所有退出路径。

---

1. SSE 帧格式与事件协议

1.1 帧编码

每一帧都是标准 SSE：

def _sse(obj: dict) -> str:
    return f"data: {json.dumps(obj, ensure_ascii=False)}\n\n"

• ensure_ascii=False：中文不转义，前端直接拿到可读文本。
• 每帧一个 JSON 对象，必带 type 字段，前端按 type 分派渲染。
• 结尾 \n\n 是 SSE 规范的事件分隔符。

1.2 事件类型清单

type	载荷	语义	是否进入「答案」
tool	{id, label?, status: running|done|error}	工具卡：内部步骤可视化	否（纯进度）
reasoning	{text}	推理模型思考流，逐字	否（纯展示）
delta	{text}	模型自然语言前言	否（前言，JSON 不外露）
draft	{draft}	规范化后的 ScriptDraft	是（结构化渲染）
saved	{script_version_id, version}	已落库的 ScriptVersion	是
summary	{text}	模型写的收尾交付语	是（当 AI 回复气泡）
done	{}	正常结束	—
error	{detail}	失败（额度已回滚）	—

---

2. 逐帧时序详解

下面按生成器实际 yield 顺序拆解，标注每一步的技术意图。

阶段 A · 加载技能（同步、毫秒级）

yield _sse({"type": "tool", "id": "skill", "label": "加载电商脚本技能", "status": "running"})
skill_loaded = bool(load_ecommerce_skill())
yield _sse({"type": "tool", "id": "skill", "status": "done" if skill_loaded else "error"})

• load_ecommerce_skill() 用 @lru_cache(maxsize=1)：把 SKILL.md + references/*.md 拼成系统提示词，进程内只读一次磁盘。
• 同一个 id: "skill" 先发 running 再发 done，前端据 id 原地更新同一张卡的状态，而不是堆两张卡。
• 缺文件不致命：load_ecommerce_skill 有兜底字符串，skill_loaded 仍为 True。

阶段 B · 分析商品 + 构建消息（同步）

yield _sse({"type": "tool", "id": "analyze", "label": f"分析商品:{project.product.title}", "status": "running"})
# ... 加载基准稿(改稿)、校验镜号、build_agent_messages ...
yield _sse({"type": "tool", "id": "analyze", "status": "done"})

这一阶段做了几件关键的前置判断：

1. 改稿才加载基准稿：mode == "revise" and base_version_id 时 _load_base_draft 读出历史稿。基准稿拿不到则 target_index = None，单镜改无从谈起，退回整版生成。
2. 改稿用基准稿的真实时长：effective_duration = base_draft.total_duration or total_duration。这是修「90s/6镜稿被请求侧默认 60 挤掉尾镜」的关键——前端可能硬编码 60，但改稿必须尊重原稿镜数。
3. 镜号越界先于建任务：target_index 非空时校验 0 <= target_index < seg_n，越界直接 yield error 并 return，绝不建任务/扣费，避免计费空转的静默 no-op。

注意所有 target_index 判断一律用 is None，不能用真值判断——0 是合法镜号（第 1 镜），if target_index: 会把第 1 镜误当未指定。

阶段 C · 建任务 + 预扣额度

task_type = AITask.Type.SCRIPT_OPTIMIZATION if mode == "revise" else AITask.Type.SCRIPT_GENERATION
try:
    task = create_ai_task(project=..., task_type=task_type, model_config=..., request_payload={...})
except Exception as exc:
    yield _sse({"type": "error", "detail": f"任务创建失败(可能额度不足):{exc}"})
    return
reservation = task.credit_reservation

• create_ai_task 内部 @transaction.atomic：建 AITask（CREATED）→ reserve_credit 预扣 → 置 RESERVED。预扣失败（余额不足）抛异常，这里转成 error 事件优雅返回。
• reservation 句柄留到后面 charge/release 用。

阶段 D · 调模型流式生成（核心，耗时几十秒）

这是整个流程最重的一段，包在 try/finally（计费兜底）+ 内层 try/except（生成失败处理）里。

settled = False          # 额度是否已结算
try:
    yield _sse({"type": "tool", "id": "generate", "label": "按黄金结构生成分镜", "status": "running"})
    full: list[str] = []  # 累积模型正文
    shown = 0             # 已外露给前端的可见字符数
    forwarding = True     # 是否仍在转发前言(遇到 JSON 起点后置 False)
    try:
        task.status = AITask.Status.SUBMITTED; task.save(...)
        provider = build_provider(model_config)
        for ev in provider.chat_completion_stream(model=..., messages=messages, temperature=0.85):
            et = ev.get("type")
            if et == "reasoning":
                rpiece = ev.get("text") or ""
                if rpiece:
                    yield _sse({"type": "reasoning", "text": rpiece})
                continue
            if et == "delta":
                full.append(ev["text"])
                if forwarding:
                    text = "".join(full)
                    cut = _visible_cut(text)
                    if cut < len(text):
                        forwarding = False
                    visible = text[:cut]
                    if len(visible) > shown:
                        piece = visible[shown:]
                        shown = len(visible)
                        if piece.strip():
                            yield _sse({"type": "delta", "text": piece})
            elif et == "done":
                break
        raw = "".join(full)
        draft = normalize_draft(raw, aspect_ratio=..., total_duration=effective_duration)
        if target_index is not None and base_draft:
            draft = _merge_single_segment(base_draft, draft, target_index, ...)
    except Exception as exc:
        _fail_task(task, reservation, str(exc)); settled = True
        yield _sse({"type": "tool", "id": "generate", "status": "error"})
        yield _sse({"type": "error", "detail": f"脚本生成失败:{exc}"})
        return

D.1 两种 delta 的区分（reasoning vs content）

底层 chat_completion_stream 把 OpenAI 兼容 SSE 的 delta 拆成两路：

• delta.reasoning_content → {type: "reasoning"}
• delta.content → {type: "delta"}

豆包 seed-pro 这类推理模型在出 JSON 前会先思考几十秒，思考期只发 reasoning_content、不发 content。如果不单独转发 reasoning，整个思考期前端零输出 = 假死（用户看到「按黄金结构生成分镜」卡了几十秒以为崩了）。所以 reasoning 逐字下发、纯展示、continue 掉不进 full（它不是答案正文）。

D.2 前言可见区裁剪（_visible_cut）

模型按运行时输出协议会先说一句口语前言（"在为这款保温杯生成 4 镜痛点脚本…"），紧接着吐 ```json 代码块。前端只该看到前言，不该看到刷屏的 JSON。

def _visible_cut(text: str) -> int:
    """可见区终点 = JSON 起点(``` 或第一个 {)。"""
    cands = []
    for marker in ("```", "{"):
        i = text.find(marker)
        if i != -1:
            cands.append(i)
    return min(cands) if cands else len(text)

转发逻辑用三个游标协作：

• full：累积全部模型正文（含 JSON），用于最后解析。
• shown：已经 delta 出去的可见字符数，保证只增量发新字符、不重发。
• forwarding：一旦 cut < len(text)（即出现了 ``` 或 {），说明前言结束、JSON 开始，置 False，此后不再转发任何 delta（JSON 不外露）。

只发 piece.strip() 非空的片段，避免把纯空白也当帧发出去。

D.3 抽取与规范化

raw = "".join(full) 是模型完整正文。normalize_draft(raw, ...) 负责「不信任模型排版」的全部兜底（抽 JSON、配平括号、挑内容最丰富的 segments 数组、字段模糊匹配、镜数对齐补齐），详见 script_agent.py 的 _extract_json / _resolve_segments / _pick_field。

D.4 精准改一镜的合并

target_index is not None and base_draft 时，模型虽被要求只改第 N 镜并输出完整稿，但后端不信它会乖乖保留其余镜——_merge_single_segment 以基准稿深拷贝为底，只用新稿的第 N 镜替换，其余镜逐字保持，再整体规范化。模型没产出目标镜则抛错（不静默返回 base 空转计费）。

阶段 E · 自检卡 + draft 事件

yield _sse({"type": "tool", "id": "generate", "status": "done"})
yield _sse({"type": "tool", "id": "extract", "label": f"提取实体 {len(draft['entities'])} 个 · {len(draft['segments'])} 镜", "status": "done"})
yield _sse({"type": "tool", "id": "check", "label": "自检:镜数 / ≤55字 / 违规词", "status": "done"})
yield _sse({"type": "draft", "draft": draft})

注意 extract / check 卡直接发 done——它们是对已完成结果的事后陈述（实体数、镜数都已知），不是真有独立的耗时步骤，目的是补齐 agent 工作流的叙事完整性。draft 事件把结构化稿交给前端做卡片化渲染。

阶段 F · 落库 + 结算额度

try:
    with transaction.atomic():
        task.status = AITask.Status.SUCCEEDED
        task.response_payload = {"raw": raw[:8000]}
        task.actual_cost = task.estimated_cost
        task.completed_at = timezone.now(); task.save(...)
        charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
        source = "revise" if mode == "revise" else ("theme" if mode == "theme" else "ai")
        script = persist_script_draft(project=..., task=task, draft=draft, source=source)
    settled = True   # charge 已提交
except Exception as exc:
    _fail_task(task, reservation, f"保存脚本失败:{exc}"); settled = True
    yield _sse({"type": "error", "detail": f"保存脚本失败:{exc}"})
    return

• charge 与落库同一事务：charge_reserved_credit 和 persist_script_draft 在同一个 transaction.atomic() 里。落库失败则 atomic 回滚 charge，_fail_task 补释放预留——钱和数据强一致。
• settled = True 标记额度已结算，给最外层 finally 看（见第 3 节）。
• persist_script_draft 建 ScriptVersion + ScriptSegment，并把 entities 回填 project.metadata（cast/scenes/script_entities），把 SCRIPT 阶段标 NEEDS_REVIEW。

阶段 G · saved / summary / done

yield _sse({"type": "saved", "script_version_id": str(script.id),
            "version": ScriptVersionSerializer(script).data})
summary = _closing_summary(raw)
if summary:
    yield _sse({"type": "summary", "text": summary})
yield _sse({"type": "done"})

_closing_summary 取「最后一个 JSON 对象之后的文字」当 AI 回复气泡——这是模型在协议第 3 步写的口语交付语（"这版主打熬夜痛点，钩子用了反差，你可以再让我调 CTA"）。去掉收尾的 ``` 围栏，若残留 { 或太短（<4 字）则返回空串，由前端兜底默认句。

---

3. 计费生命周期与断连兜底（最易踩坑处）

整段生成包在：

settled = False
try:
    ...  # 阶段 D~G
finally:
    if not settled:
        _fail_task(task, reservation, "stream aborted (client disconnected)")

为什么必须用 finally 而不是普通 except：

客户端中途断连时，Django 会对生成器调用 .close()，在当前 yield 处抛 GeneratorExit。它继承自 BaseException 而非 Exception，普通 except Exception 抓不到。若不处理，预扣的额度会永久冻结（既没 charge 也没 release）。

settled 标志覆盖所有路径：

退出路径	settled	finally 动作
正常完成（charge 成功）	True	不动
生成异常（D.4 except）	True（已 _fail_task）	不动
落库失败（F except）	True（已 _fail_task）	不动
客户端断连（GeneratorExit）	False	_fail_task 释放预扣

_fail_task 自身也防御性 try/finally：先置任务 FAILED，再 release_credit，release 失败也吞掉（不让兜底逻辑自身抛异常）。

---

4. 关键技术细节备忘

细节	说明
同步生成器 + StreamingHttpResponse	stream_script_agent 是普通同步 def + yield，不是 async。Django 的 StreamingHttpResponse 直接迭代它，每 yield 一帧立即下发。
关 nginx 缓冲	端点设 X-Accel-Buffering: no + Cache-Control: no-cache，否则 nginx 会攒够 buffer 才下发，破坏逐帧体感。
DRF 必须挂 SSE renderer	@action(..., renderer_classes=[ServerSentEventRenderer])，否则 DRF 内容协商返回 406。
UTF-8 强制	底层 chat_completion_stream 设 response.encoding = "utf-8"，SSE 不带 charset 时 requests 默认 latin-1 会让中文乱码。
temperature 0.85	脚本生成要发散有创意；对比实体提取那条用 0.3（结构化抽取要稳，降 JSON 漂移）。
同 id 工具卡原地更新	running → done/error 复用同一 id，前端据 id 更新而非新增卡片。
reasoning 不进 full	思考流纯展示，continue 跳过累积，避免污染待解析正文。
raw 截断存档	task.response_payload = {"raw": raw[:8000]}，存证据但限长，失败时可回看模型到底吐了啥。

---

5. 前端消费契约（给前端对接者）

按 type 分派即可：

• tool：维护一个 Map<id, {label, status}>，渲染成进度卡列表；同 id 更新状态。
• reasoning：追加到「思考过程」可折叠区（灰字、逐字滚动）。
• delta：追加到 AI 前言气泡。
• draft：用结构化数据渲染分镜卡片（hook/tone/segments），可直接编辑。
• saved：拿 script_version_id 标记当前稿，version 是完整序列化对象可直接入列表。
• summary：作为 AI 的收尾回复气泡（没有则用默认句兜底）。
• done：关闭 loading。
• error：弹 detail，此时后端已回滚额度，前端无需补偿。

---

6. 涉及文件索引

文件	角色
apps/ai/script_agent.py	本文主体：stream_script_agent 编排 + normalize/merge/persist
apps/projects/views.py	script_agent_stream 端点 + ServerSentEventRenderer
apps/ai/providers/volcano.py	chat_completion_stream 底层 SSE，reasoning/delta 分流
apps/ai/services.py	build_provider 可插拔分流、create_ai_task 预扣
apps/billing/services/ledger.py	reserve/charge/release_credit
skills/ecommerce-video-script/SKILL.md	领域知识（系统提示词）