优化全能创作

2026-09-14 11:47:49 +08:00
parent 02e5dbfb60
commit 463613c0d7
7 changed files with 875 additions and 197 deletions
@@ -7,8 +7,9 @@
 铁律(踩过就回不来的三条):
 1. **视频 5–10 分钟,绝不在 SSE 里等。** 生成工具立刻返回 task_id,落一条
   `generating` 消息,发 `task` 事件,收流。前端轮询完成后原地换成 `result`。
-2. **`ask_user` 一旦被调用就中断循环。** 反问的意义是等人回答,继续跑下去
-   等于自问自答。
+2. **闸门必须等人确认。** `ask_user` / `write_strategy` / `write_plan` /
+   `write_prompt` 一旦落卡就中断循环;视频 5 步(澄清→策略→方案→Prompt→出片确认)
+   不可同轮连跳。
 3. **一条用户消息最多计费生成一次。** 对话式会放大调用量,一句「多做几版」
   能烧掉一堆积分。

@@ -39,6 +40,159 @@ MAX_TOOL_ROUNDS = 8
 # 单条用户消息最多触发一次计费生成(契约 §4)
 MAX_BILLED_GENERATIONS = 1

+# 视频闸门阶段(落在 conversation.memory.stage;resume 靠它)
+# clarify → strategy → plan → prompt → confirm → done
+VIDEO_GATE_STAGES = ("clarify", "strategy", "plan", "prompt", "confirm", "done")
+_STEP_CONFIRM_LABELS = {
+    "strategy": "创作策略已写好。确认后继续写方案；要改就点「我想改」或直接说改哪里。",
+    "plan": "视频方案已写好。确认后继续出 Prompt；要改就点「我想改」或直接说改哪里。",
+    "prompt": "出片 Prompt 已写好。确认后选择参数并生成；要改就点「我想改」或直接说改哪里。",
+}
+
+
+def get_video_gate_stage(conversation: CreationConversation) -> str:
+    memory = conversation.memory if isinstance(conversation.memory, dict) else {}
+    stage = str(memory.get("stage") or "clarify").strip()
+    return stage if stage in VIDEO_GATE_STAGES else "clarify"
+
+
+def set_video_gate_stage(
+    conversation: CreationConversation,
+    stage: str,
+    *,
+    pending_video_prompt: str | None = None,
+    clear_pending_prompt: bool = False,
+) -> None:
+    """持久化闸门阶段;可选缓存尚未展示的 video_prompt。"""
+    if stage not in VIDEO_GATE_STAGES:
+        stage = "clarify"
+    memory = dict(conversation.memory or {})
+    memory["stage"] = stage
+    if pending_video_prompt is not None:
+        memory["pending_video_prompt"] = str(pending_video_prompt)
+    if clear_pending_prompt:
+        memory.pop("pending_video_prompt", None)
+    conversation.memory = memory
+    conversation.save(update_fields=["memory", "updated_at"])
+
+
+def get_pending_video_prompt(conversation: CreationConversation) -> str:
+    memory = conversation.memory if isinstance(conversation.memory, dict) else {}
+    return str(memory.get("pending_video_prompt") or "").strip()
+
+
+def _step_confirm_payload(step: str) -> dict:
+    return {
+        "interaction": "step_confirm",
+        "step": step,
+        "fields": [
+            {
+                "key": "step_action",
+                "label": "这一步可以继续吗？",
+                "type": "single",
+                "required": True,
+                "options": [
+                    {"value": "confirm", "label": "按这个继续"},
+                    {"value": "revise", "label": "我想改"},
+                ],
+            }
+        ],
+        "submitted": False,
+        "answers": {},
+    }
+
+
+def append_step_confirm(conversation: CreationConversation, step: str) -> CreationMessage:
+    return append_message(
+        conversation,
+        role="assistant",
+        kind=CreationMessage.Kind.ELICIT,
+        text=_STEP_CONFIRM_LABELS.get(step, "请确认这一步后再继续。"),
+        payload=_step_confirm_payload(step),
+    )
+
+
+def emit_prompt_gate(
+    conversation: CreationConversation, video_prompt: str | None = None
+) -> list[CreationMessage]:
+    """方案确认后:落 prompt_file + 步骤确认卡(确定性,不再跑模型)。"""
+    prompt = (video_prompt or get_pending_video_prompt(conversation) or "").strip()
+    if not prompt:
+        return []
+    prompt = apply_product_voice_visual_guard(conversation, prompt)
+    messages = []
+    prompt_file = append_message(
+        conversation,
+        role="assistant",
+        kind=CreationMessage.Kind.PROMPT_FILE,
+        payload={
+            "title": "视频生成Prompt.md",
+            "body": prompt,
+            "ref_count": len(conversation.pinned_refs or []),
+        },
+    )
+    messages.append(prompt_file)
+    # 缓存供最终确认卡使用
+    set_video_gate_stage(conversation, "prompt", pending_video_prompt=prompt)
+    messages.append(append_step_confirm(conversation, "prompt"))
+    return messages
+
+
+def emit_final_confirm_gate(
+    conversation: CreationConversation,
+    *,
+    context: "AgentContext | None" = None,
+) -> CreationMessage | None:
+    """Prompt 确认后:落最终积分确认卡(确定性)。"""
+    if conversation.mode != CreationConversation.Mode.VIDEO:
+        return None
+    prompt = get_pending_video_prompt(conversation)
+    if not prompt:
+        last = (
+            conversation.messages.filter(kind=CreationMessage.Kind.PROMPT_FILE)
+            .order_by("-seq")
+            .first()
+        )
+        prompt = str((last.payload or {}).get("body") or "").strip() if last else ""
+    if not prompt:
+        return None
+    prompt = apply_product_voice_visual_guard(conversation, prompt)
+    credits = 0
+    if context is not None:
+        try:
+            credits = estimate_video_credits(context)
+        except Exception:  # noqa: BLE001
+            credits = 0
+    else:
+        # 无 AgentContext 时用临时壳估分(model_config 可空)
+        try:
+            credits = estimate_video_credits(
+                AgentContext(
+                    conversation=conversation,
+                    user=conversation.created_by,
+                    model_config=None,  # type: ignore[arg-type]
+                )
+            )
+        except Exception:  # noqa: BLE001
+            credits = 0
+    confirm = append_message(
+        conversation,
+        role="assistant",
+        kind=CreationMessage.Kind.CONFIRM,
+        payload={
+            "kind": "video",
+            "label": "开始生成",
+            "estimated_credits": credits,
+            "video_prompt": prompt,
+            "submitted": False,
+            "params": snapshot_session_params(conversation),
+            "param_options": confirm_param_options(True),
+        },
+    )
+    set_video_gate_stage(conversation, "confirm", pending_video_prompt=prompt)
+    return confirm
+
+
 # 记忆压缩(契约 §5):超过这么多条消息就把最老的一批压成一段摘要,
 # 只保留最近 KEEP_RECENT_MESSAGES 条原文。
 COMPRESS_AFTER_MESSAGES = 24
@@ -549,7 +703,8 @@ def tool_schemas(context: AgentContext, *, allow_plan: bool = True) -> list[dict
                "description": (
                    "写「创作策略理解」卡:说清这条片给谁看、他为什么会信、你想让他信什么、整体创作方向。"
                    "四个字段都必须写具体非空文案,禁止空字符串。"
-                    "仅当用户明确要做片/出方案时,在 write_plan 之前调一次;打招呼或闲聊不要调。"
+                    "调完会停下来等用户确认或提出修改,不要同轮接着 write_plan。"
+                    "仅当用户明确要做片/出方案时调用;打招呼或闲聊不要调。"
                ),
                "parameters": {
                    "type": "object",
@@ -568,11 +723,13 @@ def tool_schemas(context: AgentContext, *, allow_plan: bool = True) -> list[dict
            "function": {
                "name": "write_plan",
                "description": (
-                    "写「视频最终方案」卡并请用户确认。**仅当用户明确要做片/出方案/改方案时调用**;"
-                    "打招呼或闲聊不要调。调完会等用户点确认,确认后平台直接按 video_prompt 出片。"
-                    "usp / points / timeline 是给用户看的卡片正文,必须写满具体文案,禁止空着只交 video_prompt。"
-                    "video_prompt 按系统里的「出片脚本写法」写成口播秒级分镜(专业创作同口径),不要只写大纲。"
-                    "先调 write_strategy 再调它。"
+                    "写「视频最终方案」卡(USP/卖点/时间轴)并请用户确认。"
+                    "**仅当用户已确认策略、或明确要改方案时调用**;打招呼或闲聊不要调。"
+                    "调完只出方案卡并停下等人确认 —— 不要同轮出 Prompt 卡或积分确认卡。"
+                    "usp / points / timeline 必须写满具体文案;同时把完整 video_prompt 写好存档,"
+                    "用户确认方案后由平台展示 Prompt。"
+                    "video_prompt 按系统里的「出片脚本写法」写成口播秒级分镜,不要只写大纲。"
+                    "先有已确认的 write_strategy,再调它。"
                ),
                "parameters": {
                    "type": "object",
@@ -613,6 +770,31 @@ def tool_schemas(context: AgentContext, *, allow_plan: bool = True) -> list[dict
                },
            },
        })
+        tools.append({
+            "type": "function",
+            "function": {
+                "name": "write_prompt",
+                "description": (
+                    "写出片 Prompt 文件卡并请用户确认。"
+                    "仅当用户已确认方案、或明确要求改 Prompt 时调用;"
+                    "不要在 write_plan 同轮调用。调完停下等人确认,不要同轮出积分确认卡或直接出片。"
+                    "video_prompt 必须是完整可执行的口播秒级分镜。"
+                ),
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "video_prompt": {
+                            "type": "string",
+                            "description": (
+                                "交给出片模型的完整口播带货指令。"
+                                "必须含总时长与画幅、按秒分段的景别/机位/运镜/动作/信息变化、人声(仅音频抬头)。"
+                            ),
+                        },
+                    },
+                    "required": ["video_prompt"],
+                },
+            },
+        })
    else:
        tools.append({
            "type": "function",
@@ -914,6 +1096,7 @@ def submit_confirmed_video(*, conversation: CreationConversation, user, confirm_
        payload={"task_id": str(task.id), "kind": "video", "prompt": prompt}, task=task,
    )
    _remember_artifact(conversation, prompt, "video")
+    set_video_gate_stage(conversation, "done", clear_pending_prompt=True)
    return message, ""


@@ -1068,7 +1251,8 @@ def build_system_prompt(context: AgentContext, *, allow_plan: bool = True, has_c
        "- 用户刚回答过你的问题时,直接沿着答案继续;不要复述答案,也不要额外回一句「收到」。",
        "- 用户选择暂不提供某项素材时,把它当成明确授权:按已有信息和合理默认继续。除非任务客观上无法完成,否则不要再次追问同一素材。",
        "- 用户说「你来定」「你帮我选」「随便」「都行」时,就是授权你做专业判断;直接选合理方案继续,不要把选择题再抛回去。",
-        "- 禁止问「要不要继续」「要不要生成」「是否开始创作」这类流程问题。用户已经提出创作需求且信息够了,就直接写策略/方案;真正扣积分前另有确认卡。",
+        "- 禁止问「要不要继续」「要不要生成」「是否开始创作」这类流程问题。缺信息用 ask_user;信息够了就写策略。"
+        "  视频每写完一步(策略/方案/Prompt)平台会出确认卡,由用户点「按这个继续」或提出修改;不要口头问流程。",
        "- 用户打招呼或闲聊(hi / 你好 / 在吗 / 你在干什么 / 嗯 / 好的 / ok):"
        "  **禁止**调用 write_strategy、write_plan、generate_image,不要「整理方案」或直接开写脚本。",
        "- 会话里**还没有**商品/方向时:自然地告诉用户可以直接丢一句想法,别硬推销,也别用客服式结束语。",
@@ -1094,7 +1278,33 @@ def build_system_prompt(context: AgentContext, *, allow_plan: bool = True, has_c
        lo = max(1, int(dur * 5.0))
        hi = max(lo, min(85, int(dur * 5.7)))
        lines.append(f"- 当前按约 {dur} 秒出片,口播建议 {lo}–{hi} 字;write_plan 的 voice_chars 填这个区间。")
-        lines.append("- 只有用户明确要做片、出方案、改方案、换卖点/剧情时才调用 write_strategy / write_plan;闲聊与打招呼绝对不要。真要写方案时必须先 write_strategy 再 write_plan,video_prompt 按上面的秒级分镜规范写满,不要只给大纲,也禁止只回「好的,有需要再说」。")
+        lines.append(
+            "- 视频 5 步闸门(不可同轮连跳):①缺信息 ask_user 停 → ②write_strategy 停等确认 → "
+            "③用户确认后 write_plan 停等确认 → ④用户确认后 write_prompt(或平台代出 Prompt)停等确认 → "
+            "⑤用户确认后才出现积分确认卡出片。"
+        )
+        lines.append(
+            "- 只有用户明确要做片、出方案、改方案、换卖点/剧情时才调用 write_strategy / write_plan / write_prompt;"
+            "闲聊与打招呼绝对不要。用户对某一步提出修改时,只重写那一步,不要跳到后面。"
+            "video_prompt 按上面的秒级分镜规范写满,不要只给大纲,也禁止只回「好的,有需要再说」。"
+        )
+        stage = get_video_gate_stage(conversation)
+        memory = conversation.memory if isinstance(conversation.memory, dict) else {}
+        strategy_confirmed = bool(memory.get("strategy_confirmed"))
+        stage_hint = {
+            "clarify": "当前阶段=澄清:缺关键信息就 ask_user;信息够了只调 write_strategy。",
+            "strategy": (
+                "当前阶段=策略已确认,请调用 write_plan 写方案;不要再写策略,不要 write_prompt。"
+                if strategy_confirmed else
+                "当前阶段=等策略确认:不要再 write_plan/write_prompt;用户确认后才会进入方案。若用户在改策略,只重调 write_strategy。"
+            ),
+            "plan": "当前阶段=等方案确认:不要 write_prompt 或出片。若用户在改方案,只重调 write_plan。",
+            "prompt": "当前阶段=等 Prompt 确认:不要出积分确认/出片。若用户在改 Prompt,只重调 write_prompt。",
+            "confirm": "当前阶段=等出片确认:不要再写策略/方案/Prompt;用户会在确认卡上点开始生成。",
+            "done": "当前阶段=已出片:等用户新的修改或新需求再行动。",
+        }.get(stage, "")
+        if stage_hint:
+            lines.append(f"- {stage_hint}")
    else:
        lines.extend([
            "",
@@ -1631,6 +1841,9 @@ def iter_creation_agent_events(
                    "content": json.dumps(result.get("payload", {}), ensure_ascii=False),
                })
                stop = stop or stop_after
+                # 闸门中断后不再执行同轮后续工具,防止策略+方案+Prompt 一锅端
+                if stop_after:
+                    break
            if stop:
                break

@@ -1768,6 +1981,7 @@ _TOOL_LABELS = {
    "generate_image": "生成图片",
    "write_strategy": "梳理创作策略",
    "write_plan": "编排视频方案",
+    "write_prompt": "编写出片 Prompt",
 }


@@ -1870,6 +2084,8 @@ def _dispatch_tool(
            payload=payload,
        )
        # 反问一旦发出就必须停,等人回答。继续跑等于自问自答。
+        if context.is_video:
+            set_video_gate_stage(context.conversation, "clarify")
        return {
            "payload": {"asked": True},
            "_events": [{"type": "message", "message": _message_payload(message)}],
@@ -1895,16 +2111,26 @@ def _dispatch_tool(
            kind=CreationMessage.Kind.STRATEGY,
            payload=strategy_payload,
        )
-        # 策略卡只是「我理解对了吗」,不打断 —— 模型接着就该写方案
+        confirm = append_step_confirm(context.conversation, "strategy")
+        set_video_gate_stage(context.conversation, "strategy")
+        memory = dict(context.conversation.memory or {})
+        memory.pop("strategy_confirmed", None)
+        context.conversation.memory = memory
+        context.conversation.save(update_fields=["memory", "updated_at"])
+        # 策略闸门:必须停下等人确认,禁止同轮连写方案
        return {
-            "payload": {"written": True},
-            "_events": [{"type": "message", "message": _message_payload(message)}],
-        }, False
+            "payload": {"written": True, "awaiting_step": "strategy"},
+            "_events": [
+                {"type": "message", "message": _message_payload(message)},
+                {"type": "message", "message": _message_payload(confirm)},
+            ],
+        }, True

    if name == "write_plan":
        video_prompt = str(args.get("video_prompt") or "").strip()
        if not video_prompt:
            return {"payload": {"error": "video_prompt 不能为空,请把完整出片指令写进去"}}, False
+        video_prompt = apply_product_voice_visual_guard(context.conversation, video_prompt)
        card = _coerce_plan_card_args(args if isinstance(args, dict) else {})
        if not card["usp"] or not card["points"]:
            return {
@@ -1936,34 +2162,32 @@ def _dispatch_tool(
            kind=CreationMessage.Kind.PLAN, payload=plan_payload,
        )
        events.append({"type": "message", "message": _message_payload(plan)})
-
-        prompt_file = append_message(
-            context.conversation, role="assistant",
-            kind=CreationMessage.Kind.PROMPT_FILE,
-            payload={"title": "视频生成Prompt.md", "body": video_prompt,
-                     "ref_count": plan_payload["ref_count"]},
+        # video_prompt 先缓存:用户确认方案后由平台展示 Prompt 卡,本步不出 Prompt/积分卡
+        set_video_gate_stage(
+            context.conversation, "plan", pending_video_prompt=video_prompt
        )
-        events.append({"type": "message", "message": _message_payload(prompt_file)})
+        step = append_step_confirm(context.conversation, "plan")
+        events.append({"type": "message", "message": _message_payload(step)})
+        return {
+            "payload": {"awaiting_step": "plan", "stored_prompt": True},
+            "_events": events,
+        }, True

-        credits = estimate_video_credits(context)
-        # video_prompt 存在确认卡里:用户点确认后直接照它出片,不再跑一轮模型
-        confirm = append_message(
-            context.conversation, role="assistant",
-            kind=CreationMessage.Kind.CONFIRM,
-            payload={
-                "kind": "video",
-                "label": "开始生成",
-                "estimated_credits": credits,
-                "video_prompt": video_prompt,
-                "submitted": False,
-                "params": snapshot_session_params(context.conversation),
-                "param_options": confirm_param_options(True),
-            },
-        )
-        events.append({"type": "message", "message": _message_payload(confirm)})
-        events.append({"type": "credits", "estimated": credits})
-        # 方案卡写着「仅需确认一次」—— 停在这里等人点,别自己往下出片
-        return {"payload": {"awaiting_confirmation": True}, "_events": events}, True
+    if name == "write_prompt":
+        video_prompt = str(args.get("video_prompt") or "").strip()
+        if not video_prompt:
+            video_prompt = get_pending_video_prompt(context.conversation)
+        if not video_prompt:
+            return {"payload": {"error": "video_prompt 不能为空,请把完整出片指令写进去"}}, False
+        video_prompt = apply_product_voice_visual_guard(context.conversation, video_prompt)
+        emitted = emit_prompt_gate(context.conversation, video_prompt)
+        if not emitted:
+            return {"payload": {"error": "无法写出 Prompt 卡"}}, False
+        events = [{"type": "message", "message": _message_payload(m)} for m in emitted]
+        return {
+            "payload": {"awaiting_step": "prompt", "written": True},
+            "_events": events,
+        }, True

    if name == "generate_image":
        if context.generations_used >= MAX_BILLED_GENERATIONS:
@@ -836,23 +836,32 @@ class VideoPlanAndConfirmTests(CreationAgentBaseTests):
        names = {t["function"]["name"] for t in fake.calls[0]["extra_body"]["tools"]}
        self.assertIn("write_strategy", names)
        self.assertIn("write_plan", names)
+        self.assertIn("write_prompt", names)
        self.assertNotIn("generate_image", names)

-    def test_strategy_card_does_not_stop_the_loop(self):
+    def test_strategy_card_stops_with_step_confirm(self):
        fake = FakeProvider([
            _tool_chunks("write_strategy", {"target": "油皮通勤人群", "trust": "真实使用反馈",
                                            "belief": "值得一试", "direction": "达人 UGC 口播"}),
-            _text_chunks("方案我这就写"),
+            _text_chunks("不该跑到这一轮"),
        ])
        with patch("apps.ai.creation_agent.build_provider", return_value=fake):
            events = _events(stream_creation_agent(conversation=self.conversation, user=self.user,
                                                   text="做条视频", model_config=self.model))
-        strategy = [e for e in events if e.get("type") == "message" and e["message"]["kind"] == "strategy"]
-        self.assertEqual(strategy[0]["message"]["payload"]["target"], "油皮通勤人群")
-        # 策略卡只是「我理解对了吗」,不该停下来
-        self.assertEqual(len(fake.calls), 2)
+        kinds = [e["message"]["kind"] for e in events if e.get("type") == "message"]
+        self.assertIn("strategy", kinds)
+        elicits = [
+            e["message"] for e in events
+            if e.get("type") == "message" and e["message"]["kind"] == "elicit"
+        ]
+        self.assertTrue(elicits)
+        self.assertEqual(elicits[-1]["payload"].get("interaction"), "step_confirm")
+        self.assertEqual(elicits[-1]["payload"].get("step"), "strategy")
+        self.assertEqual((self.conversation.memory or {}).get("stage"), "strategy")
+        # 策略闸门必须停下等人确认,不能同轮连写方案
+        self.assertEqual(len(fake.calls), 1)

-    def test_plan_emits_three_cards_and_stops_for_confirmation(self):
+    def test_plan_emits_plan_and_step_confirm_only(self):
        fake = FakeProvider([
            _tool_chunks("write_plan", self._plan_args()),
            _text_chunks("不该跑到这一轮"),
@@ -861,9 +870,76 @@ class VideoPlanAndConfirmTests(CreationAgentBaseTests):
            events = _events(stream_creation_agent(conversation=self.conversation, user=self.user,
                                                   text="做条视频", model_config=self.model))
        kinds = [e["message"]["kind"] for e in events if e.get("type") == "message"]
-        self.assertEqual(kinds[-3:], ["plan", "prompt_file", "confirm"])
-        self.assertTrue(any(e.get("type") == "credits" for e in events))
-        # 「仅需确认一次」—— 必须停下等人点,不能自己往下烧钱出片
+        self.assertIn("plan", kinds)
+        self.assertNotIn("prompt_file", kinds)
+        self.assertNotIn("confirm", kinds)
+        elicits = [
+            e["message"] for e in events
+            if e.get("type") == "message" and e["message"]["kind"] == "elicit"
+        ]
+        self.assertTrue(elicits)
+        self.assertEqual(elicits[-1]["payload"].get("interaction"), "step_confirm")
+        self.assertEqual(elicits[-1]["payload"].get("step"), "plan")
+        self.conversation.refresh_from_db()
+        self.assertEqual((self.conversation.memory or {}).get("stage"), "plan")
+        self.assertEqual(
+            (self.conversation.memory or {}).get("pending_video_prompt"),
+            "0-3秒 近景手持商品…",
+        )
+        # 方案闸门停下,不能同轮出 Prompt/积分卡
+        self.assertEqual(len(fake.calls), 1)
+
+    def test_write_prompt_emits_prompt_and_step_confirm(self):
+        fake = FakeProvider([
+            _tool_chunks("write_prompt", {"video_prompt": "0-3秒 近景手持商品…"}),
+            _text_chunks("不该跑到这一轮"),
+        ])
+        with patch("apps.ai.creation_agent.build_provider", return_value=fake):
+            events = _events(stream_creation_agent(conversation=self.conversation, user=self.user,
+                                                   text="做条视频", model_config=self.model))
+        kinds = [e["message"]["kind"] for e in events if e.get("type") == "message"]
+        self.assertIn("prompt_file", kinds)
+        self.assertNotIn("confirm", kinds)
+        elicits = [
+            e["message"] for e in events
+            if e.get("type") == "message" and e["message"]["kind"] == "elicit"
+        ]
+        self.assertEqual(elicits[-1]["payload"].get("step"), "prompt")
+        self.assertEqual(len(fake.calls), 1)
+
+    def test_strategy_and_plan_same_round_stops_after_strategy(self):
+        """模型若同轮连调 write_strategy+write_plan,只落策略闸门。"""
+        from apps.ai.creation_agent import _parse_arguments  # noqa: F401
+
+        # FakeProvider 一轮里多个 tool_call:模拟两个独立 rounds 不行,需单轮多 call。
+        # 用手动合并:第一帧带两个 tool call deltas。
+        def _multi_tool_chunks(*pairs):
+            chunks = []
+            for index, (name, args) in enumerate(pairs):
+                import json as _json
+                chunks.append({
+                    "type": "tool_call",
+                    "tool_calls": [{
+                        "index": index,
+                        "function": {"name": name, "arguments": _json.dumps(args, ensure_ascii=False)},
+                    }],
+                })
+            return chunks
+
+        fake = FakeProvider([
+            _multi_tool_chunks(
+                ("write_strategy", {"target": "油皮通勤人群", "trust": "真实使用反馈",
+                                    "belief": "值得一试", "direction": "达人 UGC 口播"}),
+                ("write_plan", self._plan_args()),
+            ),
+            _text_chunks("不该跑到这一轮"),
+        ])
+        with patch("apps.ai.creation_agent.build_provider", return_value=fake):
+            events = _events(stream_creation_agent(conversation=self.conversation, user=self.user,
+                                                   text="做条视频", model_config=self.model))
+        kinds = [e["message"]["kind"] for e in events if e.get("type") == "message"]
+        self.assertIn("strategy", kinds)
+        self.assertNotIn("plan", kinds)
        self.assertEqual(len(fake.calls), 1)

    def test_plan_without_video_prompt_is_rejected_without_emitting_cards(self):
@@ -37,7 +37,10 @@ from .creation_agent import (
    _ASSET_CARD_LABELS,
    apply_confirm_params,
    apply_session_params,
+    emit_final_confirm_gate,
+    emit_prompt_gate,
    is_greeting,
+    set_video_gate_stage,
    submit_confirmed_image,
    submit_confirmed_video,
 )
@@ -144,6 +147,119 @@ def _pending_chat_question(conversation: CreationConversation) -> CreationMessag
    return None


+_STEP_CONTINUE_INSTRUCTIONS = {
+    "strategy": (
+        "用户已确认创作策略。现在只调用 write_plan 写方案卡(含完整 video_prompt 存档);"
+        "不要再写策略,不要调用 write_prompt,不要出片。"
+    ),
+    "plan": (
+        "用户已确认视频方案。若尚未写出 Prompt,调用 write_prompt 展示出片 Prompt;"
+        "不要重写策略/方案,不要出积分确认卡或直接出片。"
+    ),
+    "prompt": (
+        "用户已确认出片 Prompt。不要再写策略/方案/Prompt;"
+        "平台会出积分确认卡,等用户点开始生成。"
+    ),
+}
+
+_STEP_REVISE_INSTRUCTIONS = {
+    "strategy": (
+        "用户要求修改创作策略。根据反馈只重新调用 write_strategy 写一版修订策略;"
+        "写完即停,不要同轮 write_plan / write_prompt。"
+    ),
+    "plan": (
+        "用户要求修改视频方案。根据反馈只重新调用 write_plan 写一版修订方案"
+        "(含完整 video_prompt 存档);写完即停,不要同轮 write_prompt 或出片。"
+    ),
+    "prompt": (
+        "用户要求修改出片 Prompt。根据反馈只重新调用 write_prompt 写一版修订 Prompt;"
+        "写完即停,不要出积分确认卡或直接出片。"
+    ),
+}
+
+
+def _handle_step_confirm_answer(
+    conversation: CreationConversation,
+    *,
+    card: CreationMessage,
+    answers: dict,
+    user,
+) -> tuple[JsonResponse | None, bool, str]:
+    """处理步骤确认卡。返回 (短路径响应|None, force_creative, continuation_instruction)。"""
+    payload = dict(card.payload or {})
+    step = str(payload.get("step") or "").strip()
+    action = str(answers.get("step_action") or answers.get("action") or "").strip().lower()
+    feedback = str(answers.get("feedback") or "").strip()
+
+    # 兼容前端只传 confirm/revise 作为 answers 值
+    if action not in {"confirm", "revise"}:
+        raw = " ".join(str(v) for v in answers.values()).lower()
+        if "revise" in raw or "改" in raw:
+            action = "revise"
+        else:
+            action = "confirm"
+
+    payload["answers"] = {
+        "step_action": action,
+        **({"feedback": feedback} if feedback else {}),
+    }
+    payload["submitted"] = True
+    card.payload = payload
+    card.save(update_fields=["payload", "updated_at"])
+
+    if action == "revise":
+        set_video_gate_stage(conversation, step if step in {"strategy", "plan", "prompt"} else "clarify")
+        instruction = _STEP_REVISE_INSTRUCTIONS.get(
+            step,
+            "用户要求修改上一步产出。只重写被指出的那一步,不要跳到后续闸门。",
+        )
+        if feedback:
+            instruction = f"{instruction} 用户反馈：{feedback}"
+        return None, True, instruction
+
+    # confirm
+    if step == "strategy":
+        # 仍标 strategy,但打上已确认标记;write_plan 落库时会切到 plan
+        memory = dict(conversation.memory or {})
+        memory["stage"] = "strategy"
+        memory["strategy_confirmed"] = True
+        conversation.memory = memory
+        conversation.save(update_fields=["memory", "updated_at"])
+        return None, True, _STEP_CONTINUE_INSTRUCTIONS["strategy"]
+
+    if step == "plan":
+        emitted = emit_prompt_gate(conversation)
+        if not emitted:
+            # 缓存丢了:让 agent 补 write_prompt
+            return None, True, _STEP_CONTINUE_INSTRUCTIONS["plan"]
+        conversation.agent_status = CreationConversation.AgentStatus.AWAITING_USER
+        conversation.save(update_fields=["agent_status", "updated_at"])
+        return JsonResponse({
+            "conversation_id": str(conversation.id),
+            "agent_status": conversation.agent_status,
+            "messages": [CreationMessageSerializer(m).data for m in emitted],
+        }, status=200), False, ""
+
+    if step == "prompt":
+        confirm = emit_final_confirm_gate(conversation)
+        if confirm is None:
+            return None, True, _STEP_CONTINUE_INSTRUCTIONS["prompt"]
+        conversation.agent_status = CreationConversation.AgentStatus.AWAITING_USER
+        conversation.save(update_fields=["agent_status", "updated_at"])
+        credits = int((confirm.payload or {}).get("estimated_credits") or 0)
+        body = {
+            "conversation_id": str(conversation.id),
+            "agent_status": conversation.agent_status,
+            "messages": [CreationMessageSerializer(confirm).data],
+        }
+        if credits:
+            body["estimated_credits"] = credits
+        return JsonResponse(body, status=200), False, ""
+
+    # 未知 step:当普通继续
+    return None, True, "用户已确认上一步。继续推进创作,不要复述确认。"
+
+
 class GenerateImageView(APIView):
    """独立生图(不绑项目)· 图片创作/模特图/平台套图共用 —— **异步**。

@@ -1560,7 +1676,18 @@ class CreationConversationViewSet(TeamScopedViewSetMixin, ModelViewSet):
            pending = _pending_chat_question(conversation)
            if pending is not None:
                payload = dict(pending.payload or {})
-                if payload.get("phase") == "gate":
+                if payload.get("interaction") == "step_confirm":
+                    # 用户对着步骤确认卡直接打字 = 修改意见
+                    short, force_creative_turn, continuation_instruction = _handle_step_confirm_answer(
+                        conversation,
+                        card=pending,
+                        answers={"step_action": "revise", "feedback": text},
+                        user=request.user,
+                    )
+                    if short is not None:
+                        return short
+                    # 用户原文仍落库,方便回看;continuation 已带反馈
+                elif payload.get("phase") == "gate":
                    pending_fields = [item for item in (payload.get("pending_fields") or []) if isinstance(item, dict)]
                    primary_field = pending_fields[0] if pending_fields else {}
                    asset_types = [t for t in (primary_field.get("asset_types") or []) if t in TYPE_LABELS] or ["product"]
@@ -1779,13 +1906,28 @@ class CreationConversationViewSet(TeamScopedViewSetMixin, ModelViewSet):
                # 追问卡是一次性的:重复提交会让同一个问题在上下文里出现两次答案
                return JsonResponse({"detail": "这个问题已经回答过了"}, status=409)
            payload = dict(card.payload or {})
-            payload["answers"] = answers
-            payload["submitted"] = True
-            card.payload = payload
-            card.save(update_fields=["payload", "updated_at"])
+            if payload.get("interaction") == "step_confirm":
+                short, force_creative_turn, continuation_instruction = _handle_step_confirm_answer(
+                    conversation,
+                    card=card,
+                    answers=answers if isinstance(answers, dict) else {},
+                    user=request.user,
+                )
+                if short is not None:
+                    return short
+                text = ""
+                record_user_message = False
+                # force_creative / continuation 已设;跳过后面普通 elicit 逻辑
+            else:
+                payload["answers"] = answers
+                payload["submitted"] = True
+                card.payload = payload
+                card.save(update_fields=["payload", "updated_at"])

            # 素材选择闸门:用户愿意时再展开列表;跳过则直接沿原任务继续。
-            if payload.get("phase") == "gate":
+            if payload.get("interaction") == "step_confirm":
+                pass  # 已在上面处理
+            elif payload.get("phase") == "gate":
                choice = str(answers.get("_asset_gate") or "").strip()
                if choice == "send":
                    pending = payload.get("pending_fields") or []