"""全能创作 · Agent 编排循环(契约 §3/§4)。 和 script_agent.py 的根本区别:那边是「单次结构化出稿」,模型只会写脚本; 这边是**真 function calling 循环** —— 模型自己决定这一轮该反问用户、该查素材, 还是该出图/出片。 铁律(踩过就回不来的三条): 1. **视频 5–10 分钟,绝不在 SSE 里等。** 生成工具立刻返回 task_id,落一条 `generating` 消息,发 `task` 事件,收流。前端轮询完成后原地换成 `result`。 2. **闸门必须等人确认。** `ask_user` / `write_strategy` / `write_plan` / `write_prompt` 一旦落卡就中断循环;视频 5 步(澄清→策略→方案→Prompt→出片确认) 不可同轮连跳。 3. **一条用户消息最多计费生成一次。** 对话式会放大调用量,一句「多做几版」 能烧掉一堆积分。 SSE 事件见契约 §3。 """ from __future__ import annotations import json import logging import re from collections.abc import Iterator from dataclasses import dataclass from django.core.serializers.json import DjangoJSONEncoder from django.db import transaction from .creation import append_message, pin_refs from .creation_presets import ( PLOT_TWIST_PRESET, PLOT_TWIST_STORY_DEPTH_OPTIONS, apply_image_preset_prompt, apply_plot_twist_story_contract, apply_video_preset_prompt, is_click_swap_preset, plot_twist_story_depth, plot_twist_story_contract, preset_guidance, preset_workflow_guidance, video_preset_delivery_contract, ) from .mentions import TYPE_LABELS, infer_field_types, resolve_refs, search_mentions from .models import CreationConversation, CreationMessage, ModelConfig from .services import ( build_provider, enforce_no_embedded_captions, get_default_model, get_seed_text_model, resolve_text_model, ) logger = logging.getLogger(__name__) # 单条用户消息的循环上限。8 轮足够「查素材 → 反问 → 写方案 → 出图」, # 再多基本是模型在原地打转。 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") PAIN_POINT_PRESET = "痛点解决演示" PAIN_POINT_DIRECTION_KEY = "pain_point_direction" _STEP_CONFIRM_LABELS = { "strategy": "创作策略已写好。确认后继续写方案；要改就点「我想改」或直接说改哪里。", "plan": "视频方案已写好。确认后我会整理出片细节，并带你确认生成参数；要改就点「我想改」或直接说改哪里。", "prompt": "出片指令已整理好。确认后核对参数并生成；要改就点「我想改」或直接说改哪里。", } _PERSON_SOURCE_PRESETS = { "痛点解决演示", PLOT_TWIST_PRESET, "达人口播种草", "鱼眼换装", } _PERSON_VISUAL_RE = re.compile( r"(人物|角色|模特|主角|达人|主播|出镜|口播|女生|女性|男生|男性|" r"女主|男主|年轻人|手模|手部|真人|换装|穿搭|剧情|短剧)" ) def is_plot_twist_conversation(conversation: CreationConversation) -> bool: return conversation.mode == CreationConversation.Mode.VIDEO and conversation.preset == PLOT_TWIST_PRESET def is_pain_point_conversation(conversation: CreationConversation) -> bool: return conversation.mode == CreationConversation.Mode.VIDEO and conversation.preset == PAIN_POINT_PRESET def is_pain_point_direction_payload(payload: dict) -> bool: fields = [item for item in (payload.get("fields") or []) if isinstance(item, dict)] return bool(fields and str(fields[0].get("key") or "") == PAIN_POINT_DIRECTION_KEY) def apply_pain_point_direction( conversation: CreationConversation, payload: dict, choice: str, ) -> str: """把已选方向直接作为本轮核心痛点/卖点，后续不再重复追问核心卖点。""" fields = [item for item in (payload.get("fields") or []) if isinstance(item, dict)] options = (fields[0].get("options") or []) if fields else [] selected = next( ( item for item in options if isinstance(item, dict) and str(item.get("value") or "") == str(choice or "") ), None, ) direction = str((selected or {}).get("label") or choice or "").strip() memory = dict(conversation.memory or {}) memory["pain_point_direction_ready"] = True memory["pain_point_direction"] = direction memory["selling_point_ready"] = True memory["selling_point_mode"] = "manual" memory["selling_point"] = direction conversation.memory = memory conversation.save(update_fields=["memory", "updated_at"]) return ( f"商家已选择痛点方向：【{direction}】。这个选择同时就是本轮要突出的核心痛点与核心卖点。" "现在只调用 write_strategy 写创作策略，并让痛点、正常使用过程和可见结果都围绕它展开；" "不要再询问核心卖点，不要复述选择，不要直接写方案或出片。" ) def pain_point_direction_options_from_text(text: str) -> list[dict[str, str]]: """模型偶尔只输出三条列表而忘记 ask_user；把列表确定性转成可点击选项。""" items: list[str] = [] for line in str(text or "").splitlines(): match = re.match(r"^\s*(?:[-*+•]|[1-3][.、．)])\s*(.+?)\s*$", line) if not match: continue label = re.sub(r"\*\*", "", match.group(1)).strip() if label and label not in items: items.append(label) if len(items) != 3: return [] return [ {"value": f"direction_{index}", "label": label} for index, label in enumerate(items, start=1) ] def active_plot_twist_story_depth(conversation: CreationConversation) -> str: """时长是最终事实来源；用户中途改时长后，故事结构必须随之切换。""" from_duration = plot_twist_story_depth(str((conversation.params or {}).get("duration") or "")) if from_duration and from_duration["value"] != "smart": return str(from_duration["value"]) memory = conversation.memory if isinstance(conversation.memory, dict) else {} return str(memory.get("plot_twist_story_depth") or "").strip() def set_plot_twist_story_depth(conversation: CreationConversation, value: str) -> dict | None: """卡片或自然语言选时长后，同步会话顶部参数与最终 Prompt 的故事结构。""" if not is_plot_twist_conversation(conversation): return None depth = plot_twist_story_depth(value) if depth is None: return None memory = dict(conversation.memory or {}) params = dict(conversation.params or {}) memory["plot_twist_story_depth"] = depth["value"] if depth["duration"]: params["duration"] = depth["duration"] conversation.memory = memory conversation.params = params conversation.save(update_fields=["memory", "params", "updated_at"]) return depth def append_plot_twist_story_depth_question(conversation: CreationConversation) -> CreationMessage: return append_message( conversation, role="assistant", kind=CreationMessage.Kind.ELICIT, text="你希望这支剧情带货视频做到什么程度？", payload={ "interaction": "plot_twist_story_depth", "fields": [ { "key": "story_depth", "label": "故事深度选择", "type": "single", "required": True, "options": [ {"value": item["value"], "label": f"{item['label']}｜{item['summary']}"} for item in PLOT_TWIST_STORY_DEPTH_OPTIONS ], } ], "submitted": False, "answers": {}, }, ) def _plot_twist_direction_fallback(conversation: CreationConversation) -> list[dict]: """模型遗漏方向卡工具时的平台兜底，不能只留下「我准备了三个方向」的空话。""" refs = [item for item in (conversation.pinned_refs or []) if isinstance(item, dict)] product = next((str(item.get("name") or "").strip() for item in refs if item.get("type") == "product"), "这款商品") return [ { "id": "misunderstanding", "title": "误会翻盘", "conflict": "主角把眼前的麻烦误判成无解，情绪不断升级。", "product_role": f"{product} 作为关键证据或解决工具，在最需要时自然出现。", "reversal": "原来问题的答案一直在眼前，误会被当场化解。", "tone": "轻喜剧、反差感强", }, { "id": "last_chance", "title": "最后一次机会", "conflict": "主角已经试过常规办法仍然失败，只剩最后一次选择。", "product_role": f"{product} 承担最后一次可验证的尝试，完整展示正常使用过程。", "reversal": "看似失败的局面被扭转，结果恰好印证核心卖点。", "tone": "紧张后释然、转化更强", }, { "id": "foreshadowing", "title": "伏笔回收", "conflict": "开头埋下一个不起眼的细节，人物关系或目标因此受阻。", "product_role": f"{product} 先作为日常细节出现，后半段成为推动结局的关键线索。", "reversal": "前面的细节被回收，观众才发现结局早有铺垫。", "tone": "温暖小反转、故事感更强", }, ] def _coerce_plot_twist_directions(raw) -> list[dict]: """把模型给的三条方向收成可直接渲染的卡片字段。""" if not isinstance(raw, list): return [] directions: list[dict] = [] for index, item in enumerate(raw[:3], start=1): if not isinstance(item, dict): continue title = str(item.get("title") or item.get("name") or "").strip() conflict = str(item.get("conflict") or item.get("setup") or "").strip() product_role = str(item.get("product_role") or item.get("product") or "").strip() reversal = str(item.get("reversal") or item.get("twist") or "").strip() tone = str(item.get("tone") or item.get("feel") or "").strip() if not all((title, conflict, product_role, reversal)): continue directions.append({ "id": str(item.get("id") or f"direction_{index}").strip() or f"direction_{index}", "title": title, "conflict": conflict, "product_role": product_role, "reversal": reversal, "tone": tone or "剧情带货", }) return directions if len(directions) == 3 else [] def _append_plot_twist_direction_question(context: "AgentContext", directions: list[dict]) -> CreationMessage: return append_message( context.conversation, role="assistant", kind=CreationMessage.Kind.ELICIT, text="选一个剧情方向，我会按它写成完整方案。", payload={ "interaction": "plot_twist_directions", "directions": directions, "fields": [{ "key": "story_direction", "label": "三个剧情反转方向", "type": "single", "required": True, "options": [ {"value": item["id"], "label": item["title"]} for item in directions ], }], "submitted": False, "answers": {}, }, ) 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 append_selling_point_gate(conversation: CreationConversation) -> CreationMessage: """在视频方案生成前确认卖点来源。 商家可以直接给真实卖点；若交给系统，则后续只从商品资料和参考素材中选择可证实的表达， 不把“系统推荐”误做成无依据的夸大文案。 """ return append_message( conversation, role="assistant", kind=CreationMessage.Kind.ELICIT, text="这条视频准备先讲哪个卖点？你可以直接写真实卖点，也可以让我从商品和素材里推荐一个。", payload={ "interaction": "selling_point_gate", "fields": [ { "key": "selling_point", "label": "商品卖点", "type": "text", "required": False, "placeholder": "例如：油污一喷一擦就干净，适合厨房重油污…", } ], "submitted": False, "answers": {}, }, ) def has_locked_person_reference(conversation: CreationConversation) -> bool: """人物一致性的前提是会话里有可解析的人物 Ref。 本地上传人物图由前端标记为 character，模特库则是 model；普通 asset 不能被默认当人，否则商品图/场景图会误跳过这个闸门。 """ return any( isinstance(ref, dict) and ref.get("type") in {"model", "character"} and ref.get("id") for ref in (conversation.pinned_refs or []) ) def video_needs_person_source(conversation: CreationConversation, user_text: str = "") -> bool: """需要真人/角色的视频在写策略前必须先锁定人物来源。""" if conversation.mode != CreationConversation.Mode.VIDEO or has_locked_person_reference(conversation): return False memory = conversation.memory if isinstance(conversation.memory, dict) else {} if memory.get("person_source_ready") or memory.get("person_source_pending"): return False if conversation.preset in _PERSON_SOURCE_PRESETS: return True recent = list( conversation.messages.order_by("-seq").values_list("text", flat=True)[:12] ) pending_prompt = str(memory.get("pending_video_prompt") or "") return bool(_PERSON_VISUAL_RE.search("\n".join([user_text, pending_prompt, *recent]))) def append_person_source_gate(conversation: CreationConversation) -> CreationMessage: """可视化的人物来源闸门；三个选项分别进文件、模特库和生图流程。""" return append_message( conversation, role="assistant", kind=CreationMessage.Kind.ELICIT, text="先确定这条视频的出镜人物。选定后，所有镜头和分段都会锁定同一位人物。", payload={ "interaction": "person_source_gate", "fields": [{ "key": "person_source", "label": "选择人物来源", "type": "single", "required": True, "options": [ {"value": "local_upload", "label": "本地上传"}, {"value": "model_library", "label": "从模特库选择"}, {"value": "platform_generate", "label": "平台帮忙生成"}, ], }], "submitted": False, "answers": {}, }, ) def click_swap_sequence(conversation: CreationConversation) -> str: memory = conversation.memory if isinstance(conversation.memory, dict) else {} return str(memory.get("click_swap_sequence") or "").strip() def click_swap_needs_sequence(conversation: CreationConversation) -> bool: """点击换款只有在商家明确款式和顺序后才能写脚本。""" if conversation.mode != CreationConversation.Mode.VIDEO: return False if not is_click_swap_preset(conversation.preset): return False memory = conversation.memory if isinstance(conversation.memory, dict) else {} return not bool(memory.get("click_swap_ready") and click_swap_sequence(conversation)) def append_click_swap_sequence_gate(conversation: CreationConversation) -> CreationMessage: return append_message( conversation, role="assistant", kind=CreationMessage.Kind.ELICIT, text="先确认要切换的款式和展示顺序。后续每次手指点击都会严格按这个顺序原位换款。", payload={ "interaction": "click_swap_sku_gate", "fields": [{ "key": "sku_sequence", "label": "款式与切换顺序", "type": "text", "required": True, "placeholder": "例如：黑色 → 白色 → 樱花粉", }], "submitted": False, "answers": {}, }, ) def submit_generated_person_reference(*, conversation: CreationConversation, user) -> CreationMessage: """先生成独立人物定妆参考，完成后再由 creation.py 自动建模特并锁定。""" from .services import enqueue_standalone_images recent_user = list( conversation.messages.filter(role=CreationMessage.Role.USER) .order_by("-seq").values_list("text", flat=True)[:6] ) brief = "\n".join(reversed([item.strip() for item in recent_user if item and item.strip()]))[:700] prompt = ( "为短视频生成一张可反复用于锁定身份的真人模特定妆参考图。" "只出现一位成年人物，正面或轻微三分之四角度，中近景，表情自然，" "五官、发型、肤色、身形和服装细节清晰，简洁中性背景，写实摄影，" "不要文字、水印、拼图、多人、遮挡脸部或夸张滤镜。" ) if conversation.preset: prompt += f" 适配视频预设：{conversation.preset}。" if brief: prompt += f" 参考用户需求：{brief}" tasks = enqueue_standalone_images( team=conversation.team, user=user, prompt=prompt, mode="model", count=1, ratio="portrait", feature="omni_create", ) task = tasks[0] memory = dict(conversation.memory or {}) memory["person_source"] = "platform_generate" memory["person_source_pending"] = True conversation.memory = memory conversation.status = CreationConversation.Status.RUNNING conversation.agent_status = CreationConversation.AgentStatus.IDLE conversation.save(update_fields=["memory", "status", "agent_status", "updated_at"]) return append_message( conversation, role="assistant", kind=CreationMessage.Kind.GENERATING, payload={ "task_id": str(task.id), "kind": "person_reference", "prompt": prompt, "label": "正在生成人物参考", }, task=task, ) 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), ) _GATE_CONTENT_KIND = { "strategy": CreationMessage.Kind.STRATEGY, "plan": CreationMessage.Kind.PLAN, "prompt": CreationMessage.Kind.PROMPT_FILE, } def restore_gated_step_after_cancel(conversation: CreationConversation) -> bool: """取消修订/推进中的规划时,恢复最近闸门的 step_confirm,供用户再点「按这个继续 / 我想改」。 返回 True → 会话应回到 awaiting_user;False → 保持 idle(尚无闸门可恢复)。 幂等:已有未提交的 step_confirm 时只校正 stage。 """ gated = ("strategy", "plan", "prompt") confirms = [] for message in ( conversation.messages.filter(kind=CreationMessage.Kind.ELICIT) .order_by("-seq")[:30] ): payload = message.payload if isinstance(message.payload, dict) else {} step = str(payload.get("step") or "").strip() if payload.get("interaction") == "step_confirm" and step in gated: confirms.append(message) if not confirms: return False latest = confirms[0] payload = dict(latest.payload or {}) step = str(payload.get("step") or "").strip() if step not in gated: return False if not payload.get("submitted"): set_video_gate_stage(conversation, step) return True content_kind = _GATE_CONTENT_KIND.get(step) newer_content = False if content_kind: newer_content = conversation.messages.filter( seq__gt=latest.seq, kind=content_kind ).exists() if newer_content: # 重写已落了新卡但还没出确认条 / 或已出过但本条仍是旧 submitted —— 补一条干净确认 has_open = any( (not (m.payload or {}).get("submitted")) and str((m.payload or {}).get("step") or "") == step for m in confirms ) if not has_open: append_step_confirm(conversation, step) else: payload["submitted"] = False payload["answers"] = {} latest.payload = payload latest.save(update_fields=["payload", "updated_at"]) set_video_gate_stage(conversation, step) if step == "strategy": memory = dict(conversation.memory or {}) if "strategy_confirmed" in memory: memory.pop("strategy_confirmed", None) conversation.memory = memory conversation.save(update_fields=["memory", "updated_at"]) return True 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) prompt = apply_product_reality_guard(prompt) prompt = apply_video_platform_safety_guard(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) prompt = apply_product_reality_guard(prompt) prompt = apply_video_platform_safety_guard(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 KEEP_RECENT_MESSAGES = 12 # 攒够这么多条没压过的消息才重压一次。没有它的话,过了阈值以后**每一轮都要多花 # 一次模型调用**去重压那么两三句话 —— 长会话的成本会翻倍。 COMPRESS_MIN_BATCH = 8 FIELD_TYPES = ("single", "multi", "text", "asset") # 顶栏下拉里的展示名 → 火山模型名。前端给的是人看的label,submit_free_video 只认真名。 VIDEO_MODEL_BY_LABEL = { "Seedance 2.5": "doubao-seedance-2-5-260628", "Seedance 2.0": "doubao-seedance-2-0-260128", "Seedance 2.0 Fast": "doubao-seedance-2-0-fast-260128", "Seedance 2.0 Mini": "doubao-seedance-2-0-mini-260615", } DEFAULT_VIDEO_MODEL = "doubao-seedance-2-5-260628" IMAGE_MODEL_BY_LABEL = { "Seedream5.0": "volcano", "Seedream-5.0-pro": "volcano", "YQ image2": "gpt-image", "影擎-Image2": "gpt-image", } # 「智能时长」没有可读到的脚本时才用的保守兜底。实际方案生成后必须从时间轴推导， # 不能把每条片都悄悄压成 15 秒。 SMART_DURATION = 15 _SMART_DURATION_RE = re.compile( # 结尾必须明确带「秒/s」。原来的「秒?」会把“18–22岁”误当成 22 秒。 r"(? str: """商品拟人默认不长脸;用户明确要求可见卡通五官时尊重其创作选择。""" base = str(prompt or "").strip() recent_user_text = " ".join( conversation.messages.filter( role=CreationMessage.Role.USER, kind=CreationMessage.Kind.TEXT, ).order_by("-seq").values_list("text", flat=True)[:12] ) if _VISIBLE_PRODUCT_FACE_RE.search(recent_user_text): return base needs_guard = ( conversation.preset == "商品拟人广告" or bool(_PRODUCT_VOICE_HINT_RE.search(recent_user_text)) or bool(_PRODUCT_VOICE_HINT_RE.search(base)) ) if not needs_guard or PRODUCT_VOICE_VISUAL_GUARD in base: return base return f"{base}\n{PRODUCT_VOICE_VISUAL_GUARD}".strip() def apply_product_reality_guard(prompt: str) -> str: """所有商品视频在出片前补上物理与用途边界，避免最终模型误演“产品坏了”。""" base = str(prompt or "").strip() if not base or PRODUCT_REALITY_GUARD in base: return base return f"{base}\n{PRODUCT_REALITY_GUARD}".strip() def apply_video_platform_safety_guard(prompt: str) -> str: """将最终出片指令收束为较不易触发视频模型审核的安全版本，且可重复调用。""" base = str(prompt or "").strip() if not base or VIDEO_PLATFORM_SAFETY_GUARD in base: return base for pattern, replacement in _VIDEO_PLATFORM_SAFETY_REWRITES: base = pattern.sub(replacement, base) return f"{base}\n{VIDEO_PLATFORM_SAFETY_GUARD}".strip() class AgentError(Exception): """Agent 循环里的业务错误,已经是可以直接给用户看的中文。""" @dataclass class AgentContext: conversation: CreationConversation user: object model_config: ModelConfig generations_used: int = 0 @property def team(self): return self.conversation.team @property def is_video(self) -> bool: return self.conversation.mode == CreationConversation.Mode.VIDEO _ASSET_PICK_LABEL = { "product": "换成哪个商品？", "character": "换成哪个角色？", "model": "换成哪个模特？", "scene": "换成哪个场景？", } _ASSET_PICK_PATTERNS = ( ("product", re.compile(r"(改|换|修改|更换|重新选|选(一个|个)?|挑).{0,8}商品")), ("character", re.compile(r"(改|换|修改|更换|重新选|选(一个|个)?).{0,8}(角色|人物)")), ("model", re.compile(r"(改|换|修改|更换|重新选|选(一个|个)?).{0,8}模特")), ("scene", re.compile(r"(改|换|修改|更换|重新选|选(一个|个)?).{0,8}场景")), ) _CHITCHAT_RE = re.compile( r"^\s*(" r"hi|hello|hey|yo|hola|" r"你好呀?|您好|嗨|哈喽|嘿|" r"在吗|在不在|有人吗|" r"早+|早安|早上好|午安|晚安|" r"嗯+|哦+|噢+|额+|呃+|" r"好的?|行|可以|ok(?:ay)?|thanks?|thank\s*you|谢谢了?|感谢|" r"收到|知道了|明白了|了解" r")[\s!！.。~～？?…]*$", re.IGNORECASE, ) _GREETING_RE = re.compile( r"^\s*(" r"hi|hello|hey|yo|hola|" r"你好呀?|您好|嗨|哈喽|嘿|" r"在吗|在不在|有人吗|" r"早+|早安|早上好|午安|晚安" r")[\s!！.。~～？?…]*$", re.IGNORECASE, ) def is_greeting(user_text: str) -> bool: """单纯打招呼要立刻回应，不能为了分析引用素材去等模型。""" return bool(_GREETING_RE.match((user_text or "").strip())) def is_pure_chitchat(user_text: str) -> bool: """纯打招呼 / 应答,没有创作意图。这类消息绝不能触发 write_strategy / write_plan。""" text = (user_text or "").strip() if not text or len(text) > 24: return False return bool(_CHITCHAT_RE.match(text)) # ---------------------------------------------------------------- 每轮引导(文字也要告诉用户下一步怎么回) _GUIDANCE_MARKER_RE = re.compile( r"(" r"请回复|回复[：:]|请直接|请告诉我|请选|点「|直接输入|" r"选择一个|发「|例如「" r")" ) def text_already_guides(text: str) -> bool: """正文里是否已经写明用户下一步该怎么回。""" return bool(_GUIDANCE_MARKER_RE.search((text or "").strip())) _NUMERIC_REPLY_INSTRUCTION_RE = re.compile( r"(?:你|请)?\s*(?:直接)?(?:回复|选择|选)\s*(?:数字|编号)?\s*" r"1\s*(?:[/、，,]\s*2)(?:\s*(?:[/、，,]\s*3))?[^。！？!\n]*(?:[。！？!]|$)" ) def strip_numeric_reply_instruction(text: str) -> str: """选择器会把 1/2/3 直接做成按钮，正文不再要求用户手输数字。""" return _NUMERIC_REPLY_INSTRUCTION_RE.sub("", text or "").strip() def default_reply_hint( conversation: CreationConversation | None = None, *, has_context: bool = False, is_video: bool = True, ) -> str: """纯文字收束时的默认「下一步可以这样回」提示。""" stage = "" if conversation is not None: try: stage = get_video_gate_stage(conversation) except Exception: # noqa: BLE001 stage = "" if stage in ("strategy", "plan", "prompt"): return "请点上方确认卡的「按这个继续」，或直接说想改哪里。" if stage == "confirm": return "请在确认卡上核对参数后点「开始生成」，或直接说要改的参数。" if stage == "done": return "可以回复「再出一版」，也可以选下方想调整的部分。" if has_context: if is_video: return "可以回复「继续完善方案」，也可以选下方想调整的方向。" return "可以回复「按这个方向出图」，也可以选下方想调整的方向。" kind = "短视频" if is_video else "商品图" return f"请直接丢一句想法，例如「帮我做一条{kind}」。" def default_reply_options( conversation: CreationConversation | None = None, *, has_context: bool = False, is_video: bool = True, assistant_text: str = "", ) -> list[dict[str, str]]: """只给与上一句相关的快捷回复，绝不塞与当前进度无关的固定话术。""" stage = "" if conversation is not None: try: stage = get_video_gate_stage(conversation) except Exception: # noqa: BLE001 stage = "" if stage == "done": return [ {"label": "再出一版", "text": "按当前方向再出一版"}, {"label": "调整画面", "text": "我想调整画面"}, {"label": "重新来", "text": "重新来"}, ] if has_context: text = str(assistant_text or "") # 创作描述常同时出现商品、人物、场景、颜色。快捷回复只看末尾真正交给用户 # 决定的部分，避免正文里的普通名词抢走最后一句的意图。 paragraphs = [part.strip() for part in re.split(r"\n+", text) if part.strip()] source = paragraphs[-1] if paragraphs else text sentences = [part.strip() for part in re.split(r"(?<=[。！？!?])", source) if part.strip()] guidance = "".join(sentences[-2:])[-240:] if sentences else source[-240:] asks_for_detail = bool(re.search( r"(?:额外|另外|其他|重点).{0,12}(?:突出|强调).{0,12}(?:细节|重点|卖点)|" r"(?:细节|重点|卖点).{0,12}(?:突出|强调)", guidance, re.IGNORECASE, )) asks_for_color_order = bool(re.search( r"配色.{0,12}(?:顺序|排序|调换|调整)|(?:调换|调整).{0,12}配色", guidance, re.IGNORECASE, )) if asks_for_detail and asks_for_color_order: return [ {"label": "补充突出细节", "text": "我想补充需要额外突出的细节"}, {"label": "调整配色顺序", "text": "我想调整配色的展示顺序"}, {"label": "按当前描述继续", "text": "没有其他调整，按当前描述继续"}, ] if re.search(r"颜色|色号|色彩|配色|SKU|款式|几种|展示顺序", guidance, re.IGNORECASE): return [ {"label": "补充颜色和顺序", "text": "我来补充每个颜色和展示顺序"}, {"label": "上传各款实物图", "text": "我补充各颜色/款式的实物图"}, {"label": "先按当前主款做", "text": "先按当前主款做，其他颜色后面再补"}, ] if re.search( r"(?:哪款|哪个|什么).{0,8}(?:商品|产品)|" r"(?:商品|产品).{0,12}(?:选择|选|换|更换|主推|想推|要推)|" r"(?:选择|选|换|更换|主推|想推|要推).{0,12}(?:商品|产品)", guidance, re.IGNORECASE, ): return [ {"label": "发商品列表", "text": "把商品列表发给我选"}, {"label": "我直接说商品名", "text": "我直接告诉你商品名"}, {"label": "你来推荐", "text": "你根据当前需求推荐一款"}, ] if re.search( r"(?:哪位|哪个|什么|选择|选|换|更换|调整|修改|改).{0,12}(?:人物|角色|模特|出镜)|" r"(?:人物|角色|模特|出镜).{0,12}(?:哪位|哪个|选择|选|换|更换|调整|修改|改)", guidance, re.IGNORECASE, ): return [ {"label": "上传人物图", "text": "我上传人物参考图"}, {"label": "由你设定角色", "text": "你先帮我设定一个合适的角色"}, {"label": "不需要人物", "text": "这条先不需要人物出镜"}, ] if re.search( r"(?:哪里|哪儿|什么|选择|选|换|更换|调整|修改|改).{0,12}(?:场景|地点|背景)|" r"(?:场景|地点|背景).{0,12}(?:哪里|哪儿|选择|选|换|更换|调整|修改|改)", guidance, re.IGNORECASE, ): return [ {"label": "上传场景图", "text": "我上传场景参考图"}, {"label": "你来推荐场景", "text": "你按商品和预设推荐场景"}, {"label": "用干净日常场景", "text": "先用干净自然的日常场景"}, ] if re.search( r"(?:补充|选择|选|换|更换|调整|修改|改|突出|强调).{0,12}(?:卖点|功能|效果|优惠|价格)|" r"(?:卖点|功能|效果|优惠|价格).{0,12}(?:补充|选择|选|换|更换|调整|修改|改|突出|强调)", guidance, re.IGNORECASE, ): return [ {"label": "补充真实卖点", "text": "我来补充商品真实卖点"}, {"label": "从素材里判断", "text": "先根据我上传的素材判断可表达的卖点"}, {"label": "先只突出一个点", "text": "先围绕一个最核心的卖点创作"}, ] # 没有足够语境时只留自由输入，不伪造“继续/改卖点”这种人机按钮。 return [] kind = "短视频" if is_video else "商品图" return [ {"label": f"帮我做一条{kind}" if is_video else "帮我做一张商品图", "text": f"帮我做一条{kind}" if is_video else "帮我做一张商品图"}, {"label": "我先说想法", "text": "我想做一个新的创作"}, ] def apply_reply_hint(message: CreationMessage, hint: str, options: list[dict[str, str]]) -> CreationMessage: """给文字气泡挂上回复示例和可点选项，供前端把下一步直接交到用户手里。""" payload = dict(message.payload or {}) changed = False if not str(payload.get("reply_hint") or "").strip(): payload["reply_hint"] = hint changed = True if not isinstance(payload.get("reply_options"), list) or not payload.get("reply_options"): payload["reply_options"] = options changed = True if not changed: return message message.payload = payload message.save(update_fields=["payload"]) return message def ensure_turn_guides( conversation: CreationConversation, *, turn_has_gate: bool, last_text_bubble: CreationMessage | None, has_context: bool, is_video: bool, ) -> list[dict]: """回合收束校验:若本轮只有散文气泡、没有追问/确认闸门,补上回复引导。 优先给已有文字气泡挂 reply_hint;若本轮连文字都没有,再落一条短引导。 不打断 5 步闸门(已有 elicit/confirm 时直接跳过)。 """ if turn_has_gate: return [] # 会话里若仍有未提交的追问/确认卡,用户本就能点卡推进,不必再注脚。 open_gate = conversation.messages.filter( role="assistant", kind__in=(CreationMessage.Kind.ELICIT, CreationMessage.Kind.CONFIRM), ).order_by("-seq").first() if open_gate is not None and not bool((open_gate.payload or {}).get("submitted")): return [] hint = default_reply_hint(conversation, has_context=has_context, is_video=is_video) options = default_reply_options( conversation, has_context=has_context, is_video=is_video, assistant_text=last_text_bubble.text if last_text_bubble is not None else "", ) events: list[dict] = [] if last_text_bubble is not None: if text_already_guides(last_text_bubble.text): return [] updated = apply_reply_hint(last_text_bubble, hint, options) events.append({"type": "message", "message": _message_payload(updated)}) return events guide = append_message( conversation, role="assistant", text=hint, payload={"reply_hint": hint, "reply_options": options}, ) events.append({"type": "message", "message": _message_payload(guide)}) return events # 明确要做内容才算出方案/出图。闲聊、吐槽、问功能都不算。 _CREATIVE_INTENT_RE = re.compile( r"(" r"帮我做|帮我拍|帮我出|帮我写|帮我改|帮我生成|" r"创作[一两]?[条个张]?|创作(?:一条|个|短)?|" r"做[一两]?[条个张](?:视频|片|图|广告)?|拍[一两]?[条个张](?:视频|片|图|广告)?|" r"出[一两]?[条个张](?:视频|片|图|广告)?|出片|出图|出方案|写方案|改方案|重写方案|重新写|" r"生成(?:一下|一张|几张|一条)?(?:视频|图|片|广告|脚本)?|做条|做个片|短视频|带货视频|短广告|广告片|带货片|" r"换卖点|改卖点|换剧情|改剧情|重做|重新出|重新来|再来一次|从头开始|重新做|按这个出|确认出片|" r"分镜|脚本|口播稿|storyboard|拟人|" r"(改|换|修改|更换).{0,8}(时长|秒数|比例|尺寸|画幅|分辨率|清晰度|模型)|" r"改成\s*\d+\s*秒|改成\s*\d+\s*[:：]\s*\d+|改成.{0,8}(竖屏|横屏|比例|分辨率)" r")", re.IGNORECASE, ) # 创作 brief 常见搭配:动词 + 成片名词(「创作一条短广告」) _CREATIVE_VERB_RE = re.compile(r"创作|制作|拍摄|生成|做|拍|出|写|弄|来一条|来个") _CREATIVE_NOUN_RE = re.compile(r"视频|短片|短视频|广告|带货|出片|分镜|脚本|口播|主图|海报|成片") def has_creative_intent(user_text: str, refs: list | None = None) -> bool: """用户本轮是否明确要做片/出图/改方案。 没意图时不把 write_strategy / write_plan / generate_image 塞进 tools, 避免模型把闲聊「整理」成方案卡。仅 @ 素材不算意图。 """ text = (user_text or "").strip() if not text: return False if is_pure_chitchat(text): return False if is_restart_intent(text): return True if _CREATIVE_INTENT_RE.search(text): return True # 「把商品…创作一条…短广告」这类 brief:同时有创作动词和成片名词 if len(text) >= 8 and _CREATIVE_VERB_RE.search(text) and _CREATIVE_NOUN_RE.search(text): return True return False _CONTINUE_INTENT_RE = re.compile( r"^\s*(继续|继续做|接着|接着做|往下做|开始吧|开做吧|就这样|就按这个|按这个来|照这个做|直接做|直接来)[吧啊呀呢。.!！]*\s*$" ) def is_continue_intent(user_text: str) -> bool: """已有创作上下文时,这些短句是在授权继续,不是闲聊。""" return bool(_CONTINUE_INTENT_RE.match((user_text or "").strip())) # 整轮重来:不是回答上一张追问卡,也不是局部「改策略/换模特」。 _RESTART_INTENT_RE = re.compile( r"^\s*(" r"重新来|重来|从头开始|再来一次|重新做|" r"重新开始|从头再来|重做一遍|再做一次|重来一遍|" r"restart|start\s*over|start\s*again" r")[吧啊呀呢]*[\s!！.。~～？?…]*$", re.IGNORECASE, ) _RESTART_CONTINUATION = ( "用户明确要求重新来/从头开始。平台已重置闸门阶段并关闭未完成的追问/确认卡。" "先用一句很短的话确认「好，我们重新来」，然后基于会话最初 brief 与已钉素材，" "从澄清或缺信息时 ask_user、信息够了就 write_strategy 重新开一整轮创作；" "不要再打开上一轮的模特库/商品库追问，不要沿用旧策略/方案/Prompt，不要猜模特。" ) def is_restart_intent(user_text: str) -> bool: """用户明确要整轮重来(重新来/重来/从头开始…)。""" return bool(_RESTART_INTENT_RE.match((user_text or "").strip())) def apply_restart_intent(conversation: CreationConversation) -> int: """重置视频闸门 memory,并关闭未提交的 elicit/confirm 卡。返回关闭张数。""" memory = dict(conversation.memory or {}) memory["stage"] = "clarify" memory.pop("pending_video_prompt", None) memory.pop("strategy_confirmed", None) memory.pop("selling_point_ready", None) memory.pop("selling_point_mode", None) memory.pop("selling_point", None) memory.pop("pain_point_direction_ready", None) memory.pop("pain_point_direction", None) memory.pop("click_swap_ready", None) memory.pop("click_swap_sequence", None) conversation.memory = memory conversation.save(update_fields=["memory", "updated_at"]) closed = 0 candidates = conversation.messages.filter( kind__in=(CreationMessage.Kind.ELICIT, CreationMessage.Kind.CONFIRM), ).order_by("-seq")[:40] for message in candidates: payload = dict(message.payload or {}) if payload.get("submitted"): continue payload["submitted"] = True payload["cancelled"] = True payload["cancelled_reason"] = "restart" message.payload = payload message.save(update_fields=["payload", "updated_at"]) closed += 1 return closed def session_has_creative_context(conversation: CreationConversation) -> bool: """会话里是否已有可继续的创作进度(钉了素材 / 出过策略或方案)。 这个信号只用于判断一句短回复是否承接创作,不拿来主动追问流程确认。 """ if conversation.pinned_refs: return True if conversation.messages.filter( kind__in=( CreationMessage.Kind.STRATEGY, CreationMessage.Kind.PLAN, CreationMessage.Kind.PROMPT_FILE, CreationMessage.Kind.CONFIRM, ) ).exists(): return True # 追问发生在策略卡之前时,会话里可能还没有任何结构化产物。原始 brief 本身 # 就是创作上下文;不认它的话,刷新后一句「继续」会被当成空闲聊天。 recent_user_texts = conversation.messages.filter( role=CreationMessage.Role.USER, kind=CreationMessage.Kind.TEXT, ).order_by("-seq").values_list("text", flat=True)[:8] return any(has_creative_intent(item) for item in recent_user_texts) def wanted_asset_pick(user_text: str, refs: list | None) -> str | None: """用户说「改商品」却没点名时,启动对应素材的自然确认。""" if refs: return None text = user_text or "" for type_, pattern in _ASSET_PICK_PATTERNS: if pattern.search(text): return type_ return None def requested_asset_card_from_context( conversation: CreationConversation, user_text: str, ) -> str | None: """「发一下卡片我选」没有说类型时,沿用最近一次素材追问的类型。""" text = str(user_text or "") wants_card = re.search( r"(发|打开|展示|看看|看下|给我).{0,10}(卡片|列表|商品库|素材库)" r"|(卡片|列表).{0,10}(选|选择|看看|看下)", text, ) if not wants_card: return None # 用户明确要某类素材列表时，不能依赖上一条是否碰巧留下追问卡。 # 否则模型会把检索结果念成名称段落，而不会返回可点击的卡片。 direct_types = ( ("product", ("商品", "产品")), ("character", ("角色", "人物")), ("model", ("模特",)), ("scene", ("场景",)), ) for type_, keywords in direct_types: if any(keyword in text for keyword in keywords): return type_ recent = conversation.messages.filter( kind=CreationMessage.Kind.ELICIT ).order_by("-seq")[:8] for message in recent: payload = message.payload or {} fields = payload.get("pending_fields") or payload.get("fields") or [] for field in fields: if not isinstance(field, dict): continue inferred = infer_field_types(field) if len(inferred) == 1 and inferred[0] in TYPE_LABELS: return inferred[0] return None VIDEO_MODELS = ["Seedance 2.5", "Seedance 2.0", "Seedance 2.0 Fast", "Seedance 2.0 Mini"] IMAGE_MODELS = ["Seedream5.0", "YQ image2"] RATIOS = ["16:9", "9:16", "4:3", "3:4", "1:1"] RESOLUTIONS = ["480p", "720p", "1080p"] VIDEO_DURATIONS = ["智能时长", "4 秒", "5 秒", "6 秒", "8 秒", "10 秒", "12 秒", "15 秒", "30 秒", "60 秒"] IMAGE_COUNTS = ["1 张", "2 张", "4 张", "8 张"] SESSION_PARAM_KEYS = ("duration", "ratio", "resolution", "video_model", "count") _PARAM_TO_STORED = {"video_model": "model"} _PARAM_PICK_LABEL = { "duration": "想改成多少秒？", "ratio": "想换成什么画幅？比如 9:16 竖屏或 16:9 横屏。", "resolution": "想换成什么清晰度？直接说 480p、720p 或 1080p 就行。", "video_model": "想换哪个模型？直接告诉我模型名就行。", "count": "这次想出几张？", } def _param_options(key: str, is_video: bool) -> list[str]: if key == "duration": return VIDEO_DURATIONS if key == "ratio": return RATIOS if key == "resolution": return RESOLUTIONS if key == "count": return IMAGE_COUNTS return VIDEO_MODELS if is_video else IMAGE_MODELS def session_param_fields(keys: list[str], is_video: bool) -> list[dict]: fields = [] # 对话式追问一次只问一件事，避免又退化成参数问卷。 for key in keys[:1]: if key not in _PARAM_PICK_LABEL: continue options = [{"value": item, "label": item} for item in _param_options(key, is_video)] fields.append({ "key": key, "label": _PARAM_PICK_LABEL[key], "type": "single", "required": True, "options": options, }) return fields def wanted_param_keys(user_text: str, *, is_video: bool) -> list[str]: """用户说「改时长」「换模型」时弹出参数卡。不和「改模特」抢。""" text = user_text or "" keys: list[str] = [] if re.search(r"(改|换|修改|更换).{0,8}(时长|秒数)|改成\s*\d+\s*秒", text): keys.append("duration" if is_video else "count") if re.search(r"(改|换|修改|更换).{0,8}(比例|尺寸|画幅)", text): keys.append("ratio") if re.search(r"(改|换|修改|更换).{0,8}(分辨率|清晰度)", text): keys.append("resolution") if re.search(r"(改|换|修改|更换).{0,8}模型", text) and "模特" not in text: keys.append("video_model") if re.search(r"(改|换|修改|更换).{0,8}张数", text): keys.append("count") if not keys and re.search(r"(改|换|修改).{0,6}(参数|设置|规格)", text): keys = ["duration", "video_model", "ratio"] if is_video else ["count", "video_model", "ratio"] seen = set() out = [] for key in keys: if key in seen: continue seen.add(key) out.append(key) return out def apply_session_params(conversation, fields, answers: dict) -> bool: """追问卡里选出的时长/模型等写回会话参数。返回是否有改动。""" current = dict(conversation.params or {}) field_by_key = {str(item.get("key") or ""): item for item in (fields or []) if isinstance(item, dict)} changed = False for key, raw in (answers or {}).items(): field = field_by_key.get(str(key)) or {} if field.get("type") == "asset": continue stored = _PARAM_TO_STORED.get(str(key), str(key)) if stored not in {"model", "ratio", "resolution", "duration", "count"}: continue value = "、".join(raw) if isinstance(raw, list) else str(raw or "").strip() if not value or current.get(stored) == value: continue current[stored] = value changed = True if changed: conversation.params = current conversation.save(update_fields=["params", "updated_at"]) if is_plot_twist_conversation(conversation): # 用户在参数卡里改了时长，剧情结构也必须立即跟着切换。 set_plot_twist_story_depth(conversation, str(current.get("duration") or "")) return changed def snapshot_session_params(conversation) -> dict: params = conversation.params or {} return { "model": str(params.get("model") or ""), "resolution": str(params.get("resolution") or ""), "ratio": str(params.get("ratio") or ""), "duration": str(params.get("duration") or ""), "count": str(params.get("count") or params.get("duration") or ""), } def confirm_param_options(is_video: bool) -> dict: return { "model": VIDEO_MODELS if is_video else IMAGE_MODELS, "resolution": RESOLUTIONS if is_video else [], "ratio": RATIOS, "duration": VIDEO_DURATIONS if is_video else [], "count": IMAGE_COUNTS if not is_video else [], } def _normalize_confirm_duration(value) -> tuple[str, int | str]: """确认卡只按实际秒数判断时长变化，兼容「8秒 / 8 秒」等历史格式。""" raw = str(value or "").strip() match = re.search(r"\d+(?:\.\d+)?", raw) if match: return "seconds", int(float(match.group())) return "label", re.sub(r"\s+", "", raw).lower() def apply_confirm_params(conversation, incoming: dict | None) -> tuple[dict, bool]: """确认卡上改的参数写回会话。返回 (最新 params, 视频时长是否变了)。""" current = dict(conversation.params or {}) old_duration = str(current.get("duration") or "") changed = False for key, raw in (incoming or {}).items(): if key not in {"model", "ratio", "resolution", "duration", "count"}: continue value = str(raw or "").strip() if not value or current.get(key) == value: continue if key == "duration" and _normalize_confirm_duration(current.get(key)) == _normalize_confirm_duration(value): continue current[key] = value changed = True duration_changed = ( conversation.mode == CreationConversation.Mode.VIDEO and _normalize_confirm_duration(current.get("duration")) != _normalize_confirm_duration(old_duration) and bool(str(current.get("duration") or "")) and bool(old_duration) ) if changed: conversation.params = current conversation.save(update_fields=["params", "updated_at"]) if is_plot_twist_conversation(conversation): # 在确认卡改时长也要切换故事契约；随后视图会要求重写旧方案。 set_plot_twist_story_depth(conversation, str(current.get("duration") or "")) return snapshot_session_params(conversation), duration_changed # ---------------------------------------------------------------- 工具 schema def tool_schemas(context: AgentContext, *, allow_plan: bool = True) -> list[dict]: """给模型看的工具清单。图片会话不暴露 generate_video,反之亦然 —— 会话 mode 是定死的(契约 §0),把不该用的工具摆出来只会诱导模型走错路。 allow_plan=False 时隐藏 write_strategy / write_plan / generate_image,闲聊用不出来。""" tools = [ { "type": "function", "function": { "name": "ask_user", "description": ( "缺少必要信息、或需要用户做选择时向用户追问 —— 这是优先的引导方式。" "只在信息**确实缺失且无法合理推断**时用;能自己定的就自己定,别把用户当填表机器。" "用户要选/改/换商品、角色、模特、场景时**必须**调这个工具," "type 用 asset 并填 asset_types。系统会以类Agent自然对话轻量询问是否需要发送商品列表(同时支持直接输入名字或由你推荐),避免一上来粗暴弹出大卡片打断交流。" "需要用户在几个明确选项里选时,用 type=single/multi 并给出 options。" "一次只问 1 项,禁止问「要不要继续」「要不要生成」「是否开始创作」这类流程问题。" "若本轮只写说明文字、不做选择,也必须在文字里写清用户下一句该回什么。" ), "parameters": { "type": "object", "properties": { "fields": { "type": "array", "maxItems": 1, "items": { "type": "object", "properties": { "key": {"type": "string", "description": "英文标识,如 product / duration"}, "label": {"type": "string", "description": "问题原文,中文"}, "type": {"type": "string", "enum": list(FIELD_TYPES)}, "required": {"type": "boolean"}, "options": { "type": "array", "items": { "type": "object", "properties": { "value": {"type": "string"}, "label": {"type": "string"}, }, "required": ["value", "label"], }, }, "asset_types": { "type": "array", "items": {"type": "string", "enum": list(TYPE_LABELS)}, }, "placeholder": {"type": "string"}, }, "required": ["key", "label", "type"], }, } }, "required": ["fields"], }, }, }, { "type": "function", "function": { "name": "search_library", "description": "在团队的商品库/模特库/角色/场景/资产库里找素材。用户说了名字但没 @ 时用它找回来。", "parameters": { "type": "object", "properties": { "query": {"type": "string"}, "types": {"type": "array", "items": {"type": "string", "enum": list(TYPE_LABELS)}}, }, "required": ["query"], }, }, }, ] if not allow_plan: # 闲聊轮次不给 ask_user,避免模型追问「要不要出片」; # 换商品/改参数仍由 wanted_asset_pick / wanted_param_keys 兜底追问。 return [t for t in tools if t.get("function", {}).get("name") == "search_library"] if context.is_video: if is_plot_twist_conversation(context.conversation) and active_plot_twist_story_depth(context.conversation): tools.append({ "type": "function", "function": { "name": "present_story_directions", "description": ( "剧情反转带货预设在用户选定故事深度后，先调用此工具展示 3 个可点击的剧情方向。" "每条都必须有不同的冲突、商品承担的实际作用、反转和情绪；不能只说‘我准备了三个方向’。" "用户点击其一后才可 write_strategy；本工具调用后必须停下来等待选择。" ), "parameters": { "type": "object", "properties": { "directions": { "type": "array", "minItems": 3, "maxItems": 3, "items": { "type": "object", "properties": { "id": {"type": "string"}, "title": {"type": "string", "description": "短标题，不超过 10 个字"}, "conflict": {"type": "string", "description": "人物处境与开场冲突"}, "product_role": {"type": "string", "description": "商品如何自然推进剧情"}, "reversal": {"type": "string", "description": "最终反转如何回收"}, "tone": {"type": "string", "description": "情绪与转化倾向"}, }, "required": ["title", "conflict", "product_role", "reversal", "tone"], }, }, }, "required": ["directions"], }, }, }) tools.append({ "type": "function", "function": { "name": "write_strategy", "description": ( "写「创作策略理解」卡:说清这条片给谁看、他为什么会信、你想让他信什么、整体创作方向。" "四个字段都必须写具体非空文案,禁止空字符串。" "策略从第一稿就使用健康、正向、明确成年的人物与情节表达,不要复述需要规避的原始措辞。" "调完会停下来等用户确认或提出修改,不要同轮接着 write_plan。" "仅当用户明确要做片/出方案时调用;打招呼或闲聊不要调。" ), "parameters": { "type": "object", "properties": { "target": {"type": "string", "description": "这条视频给谁看,要具体到人群特征"}, "trust": {"type": "string", "description": "用户为什么相信,靠什么建立可信度"}, "belief": {"type": "string", "description": "希望用户看完相信什么"}, "direction": {"type": "string", "description": "创作方向一句话,说清是什么类型的片"}, }, "required": ["target", "trust", "belief", "direction"], }, }, }) tools.append({ "type": "function", "function": { "name": "write_plan", "description": ( "写「视频最终方案」卡(USP/卖点/时间轴)并请用户确认。" "**仅当用户已确认策略、或明确要改方案时调用**;打招呼或闲聊不要调。" "调完只出方案卡并停下等人确认 —— 不要同轮出 Prompt 卡或积分确认卡。" "usp / points / timeline 必须写满具体文案;同时把完整 video_prompt 写好存档," "用户确认方案后由平台展示 Prompt。" "video_prompt 按系统里的「制作级交付」写成完整 Prompt 文件:必须有整体规则、" "声音/灯光/场景/参考素材锁定、逐镜四行细节和一致性收束,不要只写大纲。" "第一稿必须已经可直接过平台审核:只写正向安全描述,不要输出风险词清单或否定式免责声明。" "先有已确认的 write_strategy,再调它。" ), "parameters": { "type": "object", "properties": { "usp": {"type": "string", "description": "主打卖点,全片只讲这一个核心价值"}, "points": { "type": "array", "maxItems": 3, "items": {"type": "string"}, "description": "核心支撑卖点,最多 3 条", }, "timeline": { "type": "array", "items": { "type": "object", "properties": { "start": {"type": "number"}, "end": {"type": "number"}, "stage": {"type": "string", "description": "Hook / 过桥 / 正文 / CTA"}, "desc": {"type": "string"}, }, "required": ["start", "end", "stage"], }, }, "voice_chars": { "type": "array", "items": {"type": "integer"}, "description": "口播字数区间 [下限, 上限]", }, "video_prompt": { "type": "string", "description": ( "交给出片模型的制作级完整指令，不是大纲。按系统规定的标题顺序写：" "时长任务、标题、风格、镜头语言、视觉美术、色彩材质、打光、剪辑、声音、场景、主体参考、逐镜脚本、一致性收束。" "每一镜严格包含拍法/画面内容/主体或产品露出/声音四行；15 秒至少 4 镜，含人声原文、拟音和 BGM 节奏。" "已 @ 素材标明用途与需锁定的特征；禁止把口播做成画面文字。" "人物必须明确为成年人且构图得体；只写正向可拍内容，不列风险词或禁用词。" ), }, }, "required": ["usp", "points", "video_prompt"], }, }, }) tools.append({ "type": "function", "function": { "name": "write_prompt", "description": ( "写出片 Prompt 文件卡并请用户确认。" "仅当用户已确认方案、或明确要求改 Prompt 时调用;" "不要在 write_plan 同轮调用。调完停下等人确认,不要同轮出积分确认卡或直接出片。" "video_prompt 必须是系统规定的制作级完整文件，不能只把旧 Prompt 缩写成几行分镜。" ), "parameters": { "type": "object", "properties": { "video_prompt": { "type": "string", "description": ( "交给出片模型的完整制作文件。必须含总体视觉/美术/色彩/打光/声音/场景/参考图锁定规则，" "以及按秒分段、每镜均含拍法/画面内容/主体或产品露出/声音的完整脚本与一致性收束。" ), }, }, "required": ["video_prompt"], }, }, }) else: tools.append({ "type": "function", "function": { "name": "generate_image", "description": ( "生成图片。prompt 必须是完整、可独立执行的画面描述(主体/动作/环境/光线/构图/风格)," "不要写成对用户说的话。已 @ 引用的素材会自动作为参考图带上,不用在 prompt 里重复描述它们的外观。" ), "parameters": { "type": "object", "properties": { "prompt": {"type": "string"}, "count": {"type": "integer", "minimum": 1, "maximum": 4}, }, "required": ["prompt"], }, }, }) return tools # ---------------------------------------------------------------- 工具执行 def _coerce_fields(raw) -> list[dict]: """把模型给的 fields 规整成契约 §2 的 Field。脏数据丢弃而不是抛 —— 模型偶尔漏个 type 不该让整条对话崩掉。""" fields: list[dict] = [] for item in (raw or [])[:1]: if not isinstance(item, dict): continue key = str(item.get("key") or "").strip() label = str(item.get("label") or "").strip() type_ = str(item.get("type") or "").strip() if not key or not label or type_ not in FIELD_TYPES: continue field = { "key": key, "label": label, "type": type_, "required": bool(item.get("required", True)), } options = [ {"value": str(o.get("value")), "label": str(o.get("label"))} for o in (item.get("options") or []) if isinstance(o, dict) and o.get("value") and o.get("label") ] if type_ in ("single", "multi"): if not options: continue # 单选/多选没选项 = 废卡,丢掉 field["options"] = options if type_ == "asset": asset_types = [t for t in (item.get("asset_types") or []) if t in TYPE_LABELS] field["asset_types"] = asset_types or list(TYPE_LABELS) if type_ == "text": field["placeholder"] = str(item.get("placeholder") or "") inferred = infer_field_types(field) # 商品/角色这类必须出素材卡,文字单选钉不上参考图 if key in SESSION_PARAM_KEYS: fields.append(field) continue if len(inferred) == 1 and inferred[0] in TYPE_LABELS and type_ != "asset": field["type"] = "asset" field["asset_types"] = inferred field.pop("options", None) field.pop("placeholder", None) fields.append(field) return fields def _normalize_model_label(value: str) -> str: return "".join(ch for ch in str(value or "").lower() if ch.isalnum()) def image_model_name(params: dict) -> str | None: """出图模型 label → 供应商模型名;目录优先,认不出返回 None 让下游用默认。""" from django.db.models import Q from .models import ModelConfig label = str(params.get("model") or "").strip() if not label: return None mapped = IMAGE_MODEL_BY_LABEL.get(label) if mapped: return mapped hit = ( ModelConfig.objects.filter(capability=ModelConfig.Capability.IMAGE, status=ModelConfig.Status.ACTIVE) .filter(Q(display_name=label) | Q(name=label)) .order_by("created_at") .first() ) return hit.name if hit else None def video_model_name(params: dict) -> str: """会话参数里的模型 label → 供应商模型名。 先认历史写死映射,再按 ModelConfig.display_name / name 查目录 —— 后台新加模型不用改代码。 """ from django.db.models import Q from .models import ModelConfig label = str(params.get("model") or "").strip() if not label: return DEFAULT_VIDEO_MODEL mapped = VIDEO_MODEL_BY_LABEL.get(label) if not mapped: norm = _normalize_model_label(label) for k, v in VIDEO_MODEL_BY_LABEL.items(): if _normalize_model_label(k) == norm: mapped = v break if mapped: return mapped hit = ( ModelConfig.objects.filter(capability=ModelConfig.Capability.VIDEO, status=ModelConfig.Status.ACTIVE) .filter(Q(display_name=label) | Q(name=label)) .order_by("created_at") .first() ) if hit is None: hit = ( ModelConfig.objects.filter(capability=ModelConfig.Capability.VIDEO, status=ModelConfig.Status.ACTIVE) .filter(Q(display_name__icontains=label) | Q(name__icontains=label)) .order_by("created_at") .first() ) return hit.name if hit else DEFAULT_VIDEO_MODEL def _is_smart_duration(params: dict) -> bool: raw = str((params or {}).get("duration") or "").strip().lower() return not raw or "智能" in raw or raw in {"smart", "auto"} def infer_script_duration(*, timeline: list[dict] | None = None, prompt: str = "") -> int | None: """从已写好的方案推算实际片长：优先方案时间轴，其次 Prompt 的秒级分段。""" ends: list[float] = [] for item in timeline or []: if not isinstance(item, dict): continue try: end = float(item.get("end")) except (TypeError, ValueError): continue if end > 0: ends.append(end) for match in _SMART_DURATION_RE.finditer(prompt or ""): try: ends.append(float(match.group(2))) except (TypeError, ValueError): continue for match in _TOTAL_DURATION_RE.finditer(prompt or ""): try: ends.append(float(match.group(1))) except (TypeError, ValueError): continue if not ends: return None # 方案的最后一个时间点就是成片总长；向上取整避免 19.2 秒被截成 19 秒。 return max(4, min(60, int(max(ends) + 0.999))) def _raw_script_duration(*, timeline: list[dict] | None = None, prompt: str = "") -> int | None: """返回方案真实的最大时间点，不在这里截断，供 60 秒硬上限校验使用。""" ends: list[float] = [] for item in timeline or []: if not isinstance(item, dict): continue try: end = float(item.get("end")) except (TypeError, ValueError): continue if end > 0: ends.append(end) for match in _SMART_DURATION_RE.finditer(prompt or ""): try: ends.append(float(match.group(2))) except (TypeError, ValueError): continue for match in _TOTAL_DURATION_RE.finditer(prompt or ""): try: ends.append(float(match.group(1))) except (TypeError, ValueError): continue return int(max(ends) + 0.999) if ends else None def plan_video_segments(duration: int, timeline: list[dict] | None = None) -> list[dict]: """按方案节奏把 31–60 秒视频拆成可由 Seedance 2.5 单独完成的片段。 每段不超过 30 秒；优先落在时间轴的自然转场处，找不到合适转场再均分。 当前模型上限下 31–60 秒稳定拆为两段，避免无意义地把同一支片拆得过碎。 """ total = max(4, min(int(duration or 0), 60)) if total <= 30: return [{"index": 1, "start": 0, "end": total, "duration": total}] candidates: list[int] = [] for item in timeline or []: if not isinstance(item, dict): continue try: end = int(round(float(item.get("end")))) except (TypeError, ValueError): continue if 4 <= end <= total - 4: candidates.append(end) # 两段都必须 <= 30 秒，因此 60 秒时唯一合法分点就是 30 秒。 lower, upper = max(4, total - 30), min(30, total - 4) target = total / 2 legal = [point for point in candidates if lower <= point <= upper] split = min(legal, key=lambda point: abs(point - target)) if legal else int(round(target)) split = max(lower, min(upper, split)) return [ {"index": 1, "start": 0, "end": split, "duration": split}, {"index": 2, "start": split, "end": total, "duration": total - split}, ] def segment_video_prompt(prompt: str, segment: dict, total_duration: int) -> str: """让单段模型只拍本段，不把整条长脚本压回每一个片段里。""" index = int(segment.get("index") or 1) start = int(segment.get("start") or 0) end = int(segment.get("end") or 0) return ( f"{prompt.strip()}\n\n" f"【分段出片约束】这是整支 {total_duration} 秒视频的第 {index} 段，只生成 {start}–{end} 秒的内容。" "仅呈现这一时间段对应的情节与镜头。必须继续使用参考图锁定的同一位人物，" "不得在本段重新设计、随机替换或改变其五官、发型、年龄、身形和服装。" "承接上一段的动作、商品、场景和光线，" "为下一段留出自然动作衔接；不要重演完整故事，不要添加字幕、文字、角标或水印。" ) def apply_person_identity_guard(prompt: str, references: list[dict]) -> str: """把解析后的人物参考编号写进最终出片 Prompt。 resolve_refs 会把人物排在最前，但仍按真实位置计算 @图N，避免混入 场景/商品后编号指错。 """ indexes = [ index for index, item in enumerate(references or [], start=1) if isinstance(item, dict) and item.get("type") in {"model", "character"} ] if not indexes: return prompt labels = "、".join(f"参考图{index}" for index in indexes) return ( f"{prompt.strip()}\n\n【人物一致性硬约束】{labels}定义本片的固定出镜人物。" "整片及所有分段、远景、近景、转场都必须保持同一人；五官比例、脸型、发型、" "肤色、年龄、身形、手部特征和基础服装不得漂移。不得换人、随机造人、合并成新面孔，" "不得因镜头或光线变化而改变身份。" ) def video_duration(params: dict, *, prompt: str = "", timeline: list[dict] | None = None) -> int: """显式时长优先；智能时长从方案/Prompt 推导，完全缺失才回退 15 秒。""" raw = str(params.get("duration") or "") digits = "".join(ch for ch in raw if ch.isdigit()) if not digits: inferred = infer_script_duration(timeline=timeline, prompt=prompt) if inferred is not None: return inferred return SMART_DURATION return max(4, min(int(digits), 60)) def _model_supports_duration(model_name: str, duration: int) -> bool: """仅用于智能时长的自动路由；配置缺失时不武断拦截，交由提交校验给出明确错误。""" model = ModelConfig.objects.filter( name=model_name, capability=ModelConfig.Capability.VIDEO, status=ModelConfig.Status.ACTIVE, ).first() if model is None: return True meta = model.metadata if isinstance(model.metadata, dict) else {} listed = (meta.get("capabilities") or {}).get("durations") or meta.get("durations") or [] values = [int(value) for value in listed if str(value).isdigit()] return not values or min(values) <= duration <= max(values) def resolve_smart_video_duration( conversation: CreationConversation, *, prompt: str = "", timeline: list[dict] | None = None, ) -> int: """把智能时长固化为方案真实时长，并在需要时切换到能承载它的模型。 这一步发生在方案写完、用户看到最终确认卡之前，所以页面显示、计费和实际 API 参数 始终是同一个秒数。 """ params = dict(conversation.params or {}) smart_duration = _is_smart_duration(params) duration = video_duration(params, prompt=prompt, timeline=timeline) if smart_duration: params["duration"] = f"{duration} 秒" selected_model = video_model_name(params) switched = False # Seedance 2.5 是当前唯一可稳定承载 16–30 秒单段、以及 31–60 秒分段的模型。 # 即使总时长 45 秒不在单段能力表里，后续也会拆成两条 <=30 秒的 2.5 任务。 if duration > 15 and selected_model != DEFAULT_VIDEO_MODEL: params["model"] = "Seedance 2.5" switched = True elif not _model_supports_duration(selected_model, duration): # 智能模式可自动选能完成完整脚本的模型；确认卡会清楚展示变更，用户仍能手动调整。 if _model_supports_duration(DEFAULT_VIDEO_MODEL, duration): params["model"] = "Seedance 2.5" switched = True memory = dict(conversation.memory or {}) memory["smart_duration_resolved"] = duration if switched: memory["smart_duration_model_switched"] = True if smart_duration or switched: conversation.params = params conversation.memory = memory conversation.save(update_fields=["params", "memory", "updated_at"]) return duration def _image_count(params: dict, raw) -> int: """出图张数:首页选过「N 张」就用它,否则用模型传的 count,默认 1,上限 8。""" label = str((params or {}).get("count") or (params or {}).get("duration") or "") if "张" in label: digits = "".join(ch for ch in label if ch.isdigit()) if digits: raw = digits try: count = int(raw or 1) except (TypeError, ValueError): count = 1 return max(1, min(count, 8)) def _run_search_library(context: AgentContext, args: dict) -> dict: results = search_mentions( context.team, q=str(args.get("query") or "").strip(), types=[t for t in (args.get("types") or []) if t in TYPE_LABELS] or None, limit=5, ) return { "results": [ {"type": r["type"], "id": r["id"], "name": r["name"], "kind": TYPE_LABELS[r["type"]]} for r in results ] } def _run_generate_image(context: AgentContext, args: dict) -> tuple[dict, list]: """提交出图。返回 (给模型看的结果, AITask 列表)。 出图也是异步的(worker 出图 ~30s),所以这里同样只提交不等待 —— 和视频一条路子, 前端拿 task_id 轮询 GET /api/ai/generate-image/?ids=… """ from .services import enqueue_standalone_images prompt = str(args.get("prompt") or "").strip() if not prompt: raise AgentError("生成失败:模型没有给出画面描述") prompt = apply_image_preset_prompt(context.conversation.preset, prompt) params = context.conversation.params or {} resolved = resolve_refs(context.team, context.conversation.pinned_refs or []) reference_image_ids = [r["asset_id"] for r in resolved.references if r.get("asset_id")] count = _image_count(params, args.get("count")) tasks = enqueue_standalone_images( team=context.team, user=context.user, prompt=prompt, mode="image", count=count, ratio=params.get("ratio") or None, image_model=image_model_name(params) or params.get("model") or None, reference_image_ids=reference_image_ids or None, feature="omni_create", ) context.generations_used += 1 return ( {"submitted": True, "count": len(tasks), "note": "已提交生成,结果稍后回填,不要重复提交"}, list(tasks), ) def _video_submit_params(context: AgentContext, prompt: str) -> tuple[dict, list]: """拼 submit_free_video 的入参。references 直接用 resolve_refs 的产物 —— 它已经排好 角色 → 场景 → 商品 的顺序,那正是出片模型 @图N 的语义依据。""" resolved = resolve_refs(context.team, context.conversation.pinned_refs or []) prompt = apply_video_preset_prompt(context.conversation.preset, prompt) prompt = apply_product_voice_visual_guard(context.conversation, prompt) prompt = apply_product_reality_guard(prompt) prompt = apply_video_platform_safety_guard(prompt) prompt = apply_plot_twist_story_contract( context.conversation.preset, active_plot_twist_story_depth(context.conversation), prompt, ) prompt = apply_person_identity_guard(prompt, resolved.references) duration = resolve_smart_video_duration(context.conversation, prompt=prompt) # resolve_smart_video_duration 可能为了完整脚本切到支持长时长的模型，必须重新取参数。 params = context.conversation.params or {} submit = { "prompt": prompt, "feature": "omni_create", "mode": "universal", "model": video_model_name(params), "aspect_ratio": params.get("ratio") or "9:16", "resolution": params.get("resolution") or "720p", "duration": duration, "generate_audio": True, "references": resolved.references, } if any(item.get("type") in {"model", "character"} for item in resolved.references): # 长视频的多个任务共用同一个确定性 seed，减少两段各自随机采样导致的脸部/服装漂移。 submit["seed"] = int(str(context.conversation.id).replace("-", "")[:8], 16) return submit, resolved.references def estimate_video_credits(context: AgentContext) -> int: """确认按钮旁的预计积分。算不出来返回 0,前端就不显示 —— 估价失败绝不能挡住出片(用户仍会在扣费环节看到真实数字)。""" from apps.billing.pricing import quote_video_estimate params = context.conversation.params or {} submit, references = _video_submit_params(context, "") model_config = ModelConfig.objects.filter( name=submit["model"], capability=ModelConfig.Capability.VIDEO ).first() if model_config is None: return 0 try: _tokens, quote = quote_video_estimate( model_config, aspect_ratio=submit["aspect_ratio"], resolution=submit["resolution"], duration=submit["duration"], references=references, team=context.team, ) return int(quote.points) except Exception: # noqa: BLE001 — 估价挂了不该挡住出片 logger.warning("omni create: video estimate failed", exc_info=True) return 0 def estimate_image_credits(context: AgentContext) -> int: """出图确认卡预计积分:挂牌单价(含团队系数)逐张取整后再 × 张数,与 enqueue 逐任务预留同口径。""" from apps.billing.pricing import quote_flat from apps.ai.services import resolve_image_model, get_default_model params = context.conversation.params or {} model_name = image_model_name(params) or str(params.get("model") or "").strip() or None model_config = resolve_image_model(model_name) if model_name else None if model_config is None: model_config = get_default_model(ModelConfig.Capability.IMAGE) if model_config is None: return 0 count = _image_count(params, None) try: per = quote_flat(model_config, units=1, team=context.team) return int(per.points) * count except Exception: # noqa: BLE001 logger.warning("omni create: image estimate failed", exc_info=True) return 0 def submit_confirmed_video(*, conversation: CreationConversation, user, confirm_message: CreationMessage): """用户点了确认 → 直接按方案卡里存好的 video_prompt 出片。 **这里不再跑一轮模型**:方案已经确认过了,再让模型决定一次既费钱又可能它不调工具。 返回 (生成中消息, 错误文案),两者必有其一。 """ from django.conf import settings from .free_video import IN_FLIGHT_STATUSES, submit_free_video from .models import AITask payload = confirm_message.payload or {} prompt = str(payload.get("video_prompt") or "").strip() if not prompt: return None, "这条方案没有存下出片指令，请让我重新写一次方案。" context = AgentContext(conversation=conversation, user=user, model_config=None) submit, _references = _video_submit_params(context, prompt) total_duration = int(submit["duration"]) segments = plan_video_segments(total_duration) # 先整体检查并发余量，再提交任何一段；否则第一段已扣费、第二段才因额度满失败会留下孤儿片段。 in_flight = AITask.objects.filter( team=conversation.team, task_type=AITask.Type.FREE_VIDEO, status__in=IN_FLIGHT_STATUSES, ).count() max_concurrent = int(getattr(settings, "FREE_VIDEO_MAX_CONCURRENT", 3)) if in_flight + len(segments) > max_concurrent: return None, f"当前视频任务余量不足，需要同时生成 {len(segments)} 段；请等待现有任务完成后再试。" tasks = [] try: for segment in segments: segment_prompt = ( segment_video_prompt(submit["prompt"], segment, total_duration) if len(segments) > 1 else submit["prompt"] ) segment_submit = { **submit, "duration": int(segment["duration"]), # 长视频也必须从完整的最终 Prompt 分段。以前这里误用原始 prompt， # 会丢掉预设约束、商品安全约束和人物锁定，是 60s 前后换人的直接原因。 # 分段之后再执行一次统一清洗：移除模型/用户写进正文的任何屏显指令， # 并让每个独立视频任务的首尾都带最高优先级画面洁净约束。 "prompt": enforce_no_embedded_captions(segment_prompt), "extra_payload": { "omni_segment": { "index": segment["index"], "start": segment["start"], "end": segment["end"], "total_duration": total_duration, } }, } tasks.append(submit_free_video(team=conversation.team, user=user, params=segment_submit)) except ValueError as exc: # 校验类错误(时长/比例/额度),给用户看原文 return None, str(exc) if len(tasks) == 1: message_payload = {"task_id": str(tasks[0].id), "kind": "video", "prompt": prompt} else: message_payload = { "task_id": str(tasks[0].id), "task_ids": [str(task.id) for task in tasks], "kind": "video_segments", "prompt": prompt, "total_duration": total_duration, "segments": [ {**segment, "task_id": str(task.id)} for segment, task in zip(segments, tasks, strict=True) ], } message = append_message( conversation, role="assistant", kind=CreationMessage.Kind.GENERATING, payload=message_payload, task=tasks[0], ) _remember_artifact(conversation, prompt, "video") set_video_gate_stage(conversation, "done", clear_pending_prompt=True) return message, "" def submit_confirmed_image(*, conversation: CreationConversation, user, confirm_message: CreationMessage): """用户点了确认 → 按确认卡里存的画面描述出图。同样不跑一轮模型。""" payload = confirm_message.payload or {} prompt = str(payload.get("prompt") or payload.get("image_prompt") or "").strip() if not prompt: return None, "这条方案没有存下出图指令，请让我重新写一次。" context = AgentContext(conversation=conversation, user=user, model_config=None) try: _result, tasks = _run_generate_image(context, {"prompt": prompt}) except AgentError as exc: return None, str(exc) except ValueError as exc: return None, str(exc) message = None for task in tasks: message = append_message( conversation, role="assistant", kind=CreationMessage.Kind.GENERATING, payload={"task_id": str(task.id), "kind": "image", "prompt": prompt}, task=task, ) if message is None: return None, "出图没有提交成功，请再试一次。" _remember_artifact(conversation, prompt, "image") return message, "" # ---------------------------------------------------------------- 提示词 def get_creation_chat_model(requested: ModelConfig | None = None) -> ModelConfig | None: """全能创作编排固定优先 Seed 2.1 Pro;显式传入的可用模型仍尊重用户选择。""" if requested is not None: return resolve_text_model(requested) return get_seed_text_model() or resolve_text_model(None) def _creation_model_sees_images(model_config: ModelConfig | None) -> bool: """对话模型能不能收图。参考图只在能看图时才塞进 chat messages,避免纯文本模型整轮失败。""" if model_config is None: return False if getattr(model_config, "capability", "") == ModelConfig.Capability.VISION: return True name = str(getattr(model_config, "name", "") or "").lower() # 豆包 Seed 2.x / 1.6 文本档都支持图文;vl / vision 后缀同理。 if name.startswith("doubao-seed-") or "vision" in name or name.endswith("-vl") or "-vl-" in name: return True metadata = model_config.metadata if isinstance(getattr(model_config, "metadata", None), dict) else {} capabilities = metadata.get("capabilities") if isinstance(metadata.get("capabilities"), dict) else {} features = {str(item) for item in capabilities.get("features") or []} return bool({"vision", "image_input", "multimodal"} & features) def _prefer_vision_text_model(current: ModelConfig | None, team, refs: list | None) -> ModelConfig | None: """有参考图时,尽量换成能看图的文本模型(豆包 Seed 等),否则聊天侧完全看不见男女。""" if current is not None and _creation_model_sees_images(current): return current if not _ref_image_urls(team, refs): return current qs = ( ModelConfig.objects.select_related("provider") .filter( capability=ModelConfig.Capability.TEXT, status=ModelConfig.Status.ACTIVE, provider__status="active", ) .order_by("created_at") ) for candidate in qs: if _creation_model_sees_images(candidate): return candidate return get_default_model(ModelConfig.Capability.VISION) or current def _ref_image_urls(team, refs: list | None) -> list[str]: resolved = resolve_refs(team, refs or []) urls: list[str] = [] seen: set[str] = set() for item in resolved.references: url = str(item.get("url") or "").strip() if not url or url in seen: continue seen.add(url) urls.append(url) return urls[:6] def _attach_ref_images(messages: list[dict], image_urls: list[str]) -> list[dict]: """把锁定素材图挂到最近一条 user 消息上(OpenAI image_url 格式)。""" if not image_urls or not messages: return messages note = ( f"【参考图·请亲眼看】下面 {len(image_urls)} 张是用户锁定的素材。" "人物的性别、年龄段、发型、服装必须以图为准;看不清再问用户,禁止凭文件名猜测性别。" ) out = [dict(message) for message in messages] index = next((i for i in range(len(out) - 1, -1, -1) if out[i].get("role") == "user"), None) if index is None: content = [{"type": "text", "text": note}] content.extend({"type": "image_url", "image_url": {"url": url}} for url in image_urls) out.append({"role": "user", "content": content}) return out last = dict(out[index]) raw = last.get("content") if isinstance(raw, list): content = list(raw) text_bits = [str(item.get("text") or "") for item in content if isinstance(item, dict) and item.get("type") == "text"] if not any(note[:8] in bit for bit in text_bits): content.append({"type": "text", "text": note}) existing = { (item.get("image_url") or {}).get("url") for item in content if isinstance(item, dict) and item.get("type") == "image_url" } content.extend( {"type": "image_url", "image_url": {"url": url}} for url in image_urls if url not in existing ) else: content = [{"type": "text", "text": f"{raw or ''}\n\n{note}".strip()}] content.extend({"type": "image_url", "image_url": {"url": url}} for url in image_urls) last["content"] = content out[index] = last return out def build_system_prompt(context: AgentContext, *, allow_plan: bool = True, has_context: bool = False) -> str: conversation = context.conversation params = conversation.params or {} kind = "视频" if context.is_video else "图片" lines = [ "你是影擎「全能创作」的创作 agent,帮电商商家做短视频和商品图。", f"本次会话产出的是**{kind}**,这一点在整个会话里不会改变 —— 用户要另一种就请他新开一个创作。", "", "【怎么说话】", "- 说人话,像个懂创作、会一起把事做完的同事;自然、简短、有判断,不要写成客服话术或需求确认清单。", "- 禁止用「好的」「收到」「明白了」「有需要再说」「我将为你」起手;这些句子没有信息量,也很像机器人。", "- 不要逐字复述用户刚说过的话。需要承接时,只说你的判断或下一步,例如「这个方向能做,我先把反转落在商品登场上。」", "- 一次只推进一步,但不要把能直接做的事停在寒暄、确认或客套话上。", "- 缺信息时一次只问一个真正影响结果的问题,不要把对话做成问卷。", "- 缺少商品、模特、角色、场景等素材时：**绝不要直接弹出大块素材选择卡**打断对话。像懂创作、懂电商的专业伙伴一样先自然询问用户想推什么商品/用哪个角色，询问是否需要把商品列表发给他选，同时说明也可以直接输入商品名或由你推荐。", "- 只有当用户明确说要发列表（如「发给我」「发列表」「给我看看」「我来选」）时，才展示可视化卡片。", "- 用户直接打字输入商品名时，直接采纳该商品并继续推进创作方案，不要强迫用户去卡片里点选。", "- 调性、受众、文案、时长等其他信息用 ask_user 或聊天追问;无论哪种,都必须让用户知道下一句怎么回。", "- 用户刚回答过你的问题时,直接沿着答案继续;不要复述答案,也不要额外回一句「收到」。", "- 用户说「重新来」「重来」「从头开始」「再来一次」「重新做」:整轮重开创作。" " 先短确认,再按最初 brief/已钉素材从澄清或 write_strategy 推进;禁止再打开上一轮模特/商品库追问。", "- 用户选择暂不提供某项素材时,把它当成明确授权:按已有信息和合理默认继续。除非任务客观上无法完成,否则不要再次追问同一素材。", "- 用户说「你来定」「你帮我选」「随便」「都行」时,就是授权你做专业判断;直接选合理方案继续,不要把选择题再抛回去。", "- 禁止问「要不要继续」「要不要生成」「是否开始创作」这类流程问题。缺信息用 ask_user;信息够了就写策略。" " 视频每写完策略或方案，平台会出确认卡；方案确认后平台会在后台整理出片指令，再让用户核对生成参数。不要口头问流程。", "- 用户打招呼或闲聊(hi / 你好 / 在吗 / 你在干什么 / 嗯 / 好的 / ok):" " **禁止**调用 write_strategy、write_plan、generate_image,不要「整理方案」或直接开写脚本。", "- 会话里**还没有**商品/方向时:自然地告诉用户可以直接丢一句想法,别硬推销,也别用客服式结束语。", "- 会话里**已有**商品或方向时:针对用户这句话本身自然回应;别用「要现在生成还是先调细节」把工作又抛回给用户。", "- 没有明确创作指令时不要自己写策略卡/方案卡,也不要主动追问流程确认;等用户给出具体想法或修改。", "", "【每轮必须给引导】", "- 每一轮回复结束时,用户必须知道下一步怎么做:要么调用 ask_user 弹出可选项," "要么在文字里明确告诉用户该回复什么(例如「直接说想改的卖点」)。", "- 当你给出 2–3 个方向时,必须调用 ask_user 生成可点击选项；正文只解释各方向，" "绝不要求用户回复数字、编号或 1/2/3。", "- 禁止只丢一段解释/分析就结束、让用户不知道该回什么。", "- 用户已经给出可直接执行的图片需求(主体、场景或氛围已足够)时,直接调用 generate_image 进入确认卡;" "不要只描述你准备怎么拍,再让用户继续补一句。", "- 禁止用「有想调整可以直接说」这类泛泛收尾代替引导;要么 ask_user 给出可选项," "要么给出一句用户可以直接点击或复制回复的话。", "- 缺信息或要用户做选择时:优先 ask_user(带 options 的 single/multi,或 asset)。", "- 纯说明/判断/闲聊回应:文末必须带一句可执行的回复指引。", "- 视频闸门的 step_confirm / 积分确认卡本身已是引导,不要再口头追问流程。", "", "【什么时候反问】", "- 只有信息**确实缺失且无法合理推断**时才调用 ask_user;能自己定的就自己定。", "- ask_user 一次只问一件事。type=asset 会以类Agent自然追问轻量询问是否发送列表或由你推荐；其他类型显示普通聊天问题，让用户直接输入。", "- 商品是谁、给谁看、什么调性 —— 这些缺了会直接影响成片,值得问。", "- 让用户选商品/角色/模特/场景时,ask_user 必须用 type=asset 并填 asset_types。", "- 用户说「改商品」「换角色」却没点名是哪个:立刻 ask_user 启动素材确认,不要强行出大卡片。", "- 用户说改时长/模型/比例/分辨率但没给新值:立刻 ask_user,type=single 提供可识别的候选值;聊天里只显示问题,用户直接输入。回答后旧方案作废,必须按新参数重新 write_plan。", "- 光线、构图、镜头这些专业判断是你的活,不要反过来问用户。", ] if context.is_video: lines.extend(["", _OMNI_VIDEO_PROMPT_RULES.strip()]) lines.extend([ "", "【视频创作工作原则】", "- 先从文字和已锁定素材整理商品事实、人物/服装、场景、参考视频或音频、现成脚本，以及时长、比例和语言；已经给出的信息不重复问。", "- 用户已给脚本或分镜时，以它为基础补足，不强制从头重写。只改用户点名的镜头、人物、商品、台词或参数，其他内容保持。", "- 可以自己决定转场、灯光、普通镜头细节；商品功能、规格、价格、活动、功效和关键使用边界不能猜，缺失时一次只问一项。", "- 用户上传的人物、商品、服装和场景优先作为参考；如确实需要额外生成角色、场景或道具，先说明用途和预计积分，等用户确认。", "- 出片前检查：主卖点都有画面或台词证据、口播能在时长内说完、脚本中每位人物/商品/服装/场景都有对应素材、商品正常使用、全片一致、所有 SKU 都已安排。内容超出时长时建议删减、延长或拆分，而不是硬塞。", "- 安全检查前置到创作第一稿：人物默认明确为成年人，服装与构图得体；策略、方案和 video_prompt 只写改写后的正向可拍内容，不复述风险情节，不罗列平台禁用词，也不写否定式免责声明。", ]) # 按会话时长给出口播字数锚点(与专业创作 narration_limit 同口径) try: dur = video_duration(params) except Exception: # noqa: BLE001 dur = SMART_DURATION 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( "- 视频 5 步闸门(不可同轮连跳):①缺信息 ask_user 停 → ②write_strategy 停等确认 → " "③用户确认后 write_plan 停等确认 → ④方案确认后展示完整出片指令并停等确认 → ⑤再显示积分确认卡。" ) lines.append( "- 只有用户明确要做片、出方案、改方案、换卖点/剧情时才调用 write_strategy / write_plan;" "闲聊与打招呼绝对不要。用户对某一步提出修改时,只重写那一步,不要跳到后面。" "write_plan 里的 video_prompt 要按上面的秒级分镜规范写满，供平台后台出片使用；不要只给大纲，也禁止只回「好的,有需要再说」。" ) stage = get_video_gate_stage(conversation) memory = conversation.memory if isinstance(conversation.memory, dict) else {} selling_point = str(memory.get("selling_point") or "").strip() selling_mode = str(memory.get("selling_point_mode") or "").strip() if selling_mode == "manual" and selling_point: lines.append( f"- 商家已确认核心卖点：【{selling_point}】。策略、方案、脚本和出片指令必须围绕它展开；" "只补充可从素材或正常使用中证明的支撑，不得替换或夸大。" ) elif selling_mode == "auto": lines.append( "- 商家已授权系统推荐卖点。你必须从商品资料、可见素材和正常使用动作中选择一个最易证明的核心卖点；" "不要虚构功效、价格或规格。" ) strategy_confirmed = bool(memory.get("strategy_confirmed")) stage_hint = { "clarify": "当前阶段=澄清:缺关键信息就 ask_user;信息够了只调 write_strategy。", "strategy": ( "当前阶段=策略已确认,请调用 write_plan 写方案;不要再写策略。" if strategy_confirmed else "当前阶段=等策略确认:不要再写方案或出片;用户确认后才会进入方案。若用户在改策略,只重调 write_strategy。" ), "plan": "当前阶段=等方案确认:不要出片。若用户在改方案,只重调 write_plan。", "prompt": "当前阶段=兼容历史会话的出片指令确认:不要出片，按用户反馈重写内部出片指令。", "confirm": "当前阶段=等出片确认:不要再写策略/方案;用户会在确认卡上点开始生成。", "done": "当前阶段=已出片:等用户新的修改或新需求再行动。用户说重新来/重来/从头开始时,当作新一轮创作,从澄清或 write_strategy 重开,不要再弹旧模特/商品追问。", }.get(stage, "") if stage_hint: lines.append(f"- {stage_hint}") else: lines.extend([ "", "【出图】", "- 决定出图时必须调用 generate_image,不要只口头说「我这就出图」。", "- 一次用户消息只出一轮;张数用会话已定参数,不要自己加张。", ]) if not allow_plan: lines.extend([ "", "【本轮闸门】", "- 用户本轮没有明确说「去做/出方案」。没有 write_strategy / write_plan / generate_image。", "- **禁止**本轮直接写出策略卡或方案卡。", ]) if has_context: lines.extend([ "- 会话已有素材或方向:针对用户本轮实际说的话自然回应;不要问要不要继续、要不要生成。", "- 禁止只回「在呢」「有需要随时招呼」「收到」这种空话。", ]) else: lines.extend([ "- 会话还没有创作进度:短回一句,告诉用户直接说想做什么即可;不要用客服式结束语,也不要硬推销出片。", ]) if params: meta = "、".join(f"{k}:{v}" for k, v in params.items() if v) if meta: lines.append(f"\n【会话已定参数】{meta}(出片按这套;用户改动后必须按新值重写方案)") if conversation.preset: # 只给名字模型只能靠猜;把这个预设的拍法约束一起给它 guidance = preset_guidance(conversation.preset) lines.append(f"\n【创作预设】{conversation.preset}") if context.is_video: lines.append( "预设必须贯穿本次对话：据此判断该补问哪些必要事实和素材、给什么创意选择、" "如何写脚本/分镜、商品如何出现、生成前检查什么；不能把预设当成最后才附加的一行风格词。" ) if guidance: lines.append(guidance) lines.append("用户选了这个预设,就按它的拍法来;要偏离得先问过用户。") if is_plot_twist_conversation(conversation): depth = active_plot_twist_story_depth(conversation) if not depth: lines.append("剧情反转带货尚未选择故事深度：必须先调用 ask_user 让用户选 15 秒、30 秒、60 秒或智能推荐；此时禁止给剧情方向、策略或方案。") else: lines.append(plot_twist_story_contract(depth)) memory = conversation.memory if isinstance(conversation.memory, dict) else {} selected_direction = str(memory.get("plot_twist_story_direction") or "").strip() if selected_direction: lines.append(f"用户已选剧情方向：{selected_direction}。直接围绕此方向写策略，不要重发方向卡。") else: lines.append( "【强制下一步】现在必须调用 present_story_directions，直接展示 3 张完整剧情方向卡。" "每张卡写清冲突、商品如何推进剧情、反转和情绪；不得只在文字中说‘我准备了三个方向’，" "不得先 write_strategy / write_plan，也不要让用户输入编号。" ) if is_pain_point_conversation(conversation): memory = conversation.memory if isinstance(conversation.memory, dict) else {} selected_direction = str(memory.get("pain_point_direction") or "").strip() if selected_direction: lines.append( f"用户已选择痛点方向：【{selected_direction}】。这个方向就是已确认的核心卖点；" "直接围绕它写策略，不得再询问核心卖点。" ) else: lines.append( "【强制下一步·痛点方向三选一】先根据商品事实与可见素材整理恰好 3 个明显不同、可被画面证明的痛点方向。" "必须调用 ask_user：field key 固定为 pain_point_direction，type=single，options 恰好 3 项；" "每个 label 直接写完整的‘具体困扰 + 商品正常使用后可见结果’，让用户点击即选中。" "不得只在正文里列三条，不得要求用户回复编号，不得先 write_strategy，也不得另问核心卖点。" ) workflow_guidance = preset_workflow_guidance(conversation.preset) if context.is_video else "" if workflow_guidance: lines.append(f"【当前预设的工作重点】{workflow_guidance}") delivery_contract = video_preset_delivery_contract(conversation.preset) if context.is_video else "" if delivery_contract: lines.append(f"【当前预设必须贯穿脚本与出片】{delivery_contract}") resolved = resolve_refs(context.team, conversation.pinned_refs or []) if resolved.facts: lines.append("\n【本次会话已锁定的素材事实】") lines.append(resolved.facts_text) lines.append( "以上素材的参考图会附给你看,出片时也会自动锁人锁物。" "人物性别、年龄段、发型、服装、商品颜色外形必须以图为准;" "图上看不清或没附图时,必须问用户,禁止凭文件名猜测男女。" ) memory = conversation.memory or {} if memory.get("summary"): lines.append(f"\n【前情提要】{memory['summary']}") artifacts = memory.get("artifacts") or [] if artifacts: recent = artifacts[-3:] lines.append("\n【本会话已生成过】") for index, item in enumerate(recent, 1): lines.append(f"{index}. {item.get('prompt', '')[:120]}") lines.append( "用户说「改成…」「换成…」时,是要在**最后一次生成**的基础上重新生成一版," "把改动合进完整 prompt 再调生成工具 —— 不要只写改动部分。" ) return "\n".join(lines) def _answer_label(field: dict, raw) -> str: """追问卡的内部值翻成人能读的标签,避免把 UUID / gate key 喂回模型。""" values = raw if isinstance(raw, list) else [raw] options = { str(item.get("value")): str(item.get("label") or item.get("value") or "") for item in (field.get("options") or []) if isinstance(item, dict) } return "、".join(options.get(str(value), str(value)) for value in values if value not in (None, "")) def build_messages(context: AgentContext, *, allow_plan: bool = True, has_context: bool = False) -> list[dict]: """会话历史 → 模型消息。只喂对模型有意义的:文字、追问和它的答案、生成过什么。 策略卡/方案卡这类结构化产物压成一句话,原样塞 JSON 只会挤爆上下文。 长会话只喂最近 KEEP_RECENT_MESSAGES 条原文,更早的靠 system 里的【前情提要】 ——摘要在 compress_memory() 里生成,不在这里现算。 """ messages = [{"role": "system", "content": build_system_prompt(context, allow_plan=allow_plan, has_context=has_context)}] history = list(context.conversation.messages.all()) if len(history) > COMPRESS_AFTER_MESSAGES: history = history[-KEEP_RECENT_MESSAGES:] for message in history: if message.kind == CreationMessage.Kind.TEXT: if message.text.strip(): messages.append({"role": message.role, "content": message.text}) elif message.kind == CreationMessage.Kind.ELICIT: payload = message.payload or {} answers = payload.get("answers") or {} fields = [item for item in (payload.get("fields") or []) if isinstance(item, dict)] field_by_key = {str(item.get("key") or ""): item for item in fields} if payload.get("interaction") == "chat" or ( payload.get("interaction") == "asset_picker" and payload.get("answered_via") == "chat" ): question = message.text.strip() or str((fields[0] if fields else {}).get("label") or "").strip() if question: messages.append({"role": "assistant", "content": question}) # 聊天式回答本身就是下一条 TEXT user 消息,这里不再合成一条伪造答案。 continue if answers: if payload.get("phase") == "gate": choice = str(answers.get("_asset_gate") or "") if choice == "skip": answer_text = "(用户选择暂不添加这项素材,要求按现有信息继续原任务)" elif choice == "auto": answer_text = "(用户希望你帮忙挑选/决定素材,按合理推荐继续推进)" elif choice == "send": answer_text = "(用户希望打开素材列表继续选择)" else: answer_text = f"(用户指定了素材：{choice})" else: joined = "；".join( f"{field_by_key.get(str(key), {}).get('label') or key}：" f"{_answer_label(field_by_key.get(str(key), {}), value)}" for key, value in answers.items() ) answer_text = f"(用户完成了刚才的选择：{joined})" messages.append({"role": "assistant", "content": "(我请用户补充了一项会影响创作的信息)"}) messages.append({"role": "user", "content": answer_text}) else: messages.append({"role": "assistant", "content": "(我正在等用户回答刚才的问题)"}) elif message.kind in (CreationMessage.Kind.GENERATING, CreationMessage.Kind.RESULT): prompt = (message.payload or {}).get("prompt") or "" messages.append({"role": "assistant", "content": f"(我生成了一版,prompt:{prompt[:200]})"}) elif message.kind == CreationMessage.Kind.ERROR: messages.append({"role": "assistant", "content": f"(上一次生成失败:{message.text})"}) if _creation_model_sees_images(context.model_config): messages = _attach_ref_images( messages, _ref_image_urls(context.team, context.conversation.pinned_refs or []), ) return messages # ---------------------------------------------------------------- 流式循环 def _sse(event: dict) -> str: """一帧 SSE。**必须用 DjangoJSONEncoder** —— 消息里带 UUID(task 外键)和 datetime(created_at),标准 json.dumps 直接抛 TypeError,整条流当场断掉。""" return f"data: {json.dumps(event, ensure_ascii=False, cls=DjangoJSONEncoder)}\n\n" def _merge_tool_call_deltas(buffer: dict, deltas: list) -> None: """OpenAI 流式把一次 tool_call 的 arguments 拆成很多片,按 index 拼回来。 name 只在第一片出现,arguments 要逐片累加 —— 直接覆盖会只剩最后一个字符。""" for delta in deltas or []: if not isinstance(delta, dict): continue index = delta.get("index", 0) slot = buffer.setdefault(index, {"name": "", "arguments": ""}) function = delta.get("function") or {} if function.get("name"): slot["name"] = function["name"] if function.get("arguments"): slot["arguments"] += function["arguments"] def _parse_arguments(raw: str) -> dict: """解析工具 arguments。模型偶发包 markdown 围栏或夹杂前后缀,尽量救回 JSON。""" text = (raw or "").strip() if not text: return {} if text.startswith("```"): text = text.strip("`") if text.lower().startswith("json"): text = text[4:].lstrip() text = text.strip() try: parsed = json.loads(text) except ValueError: start, end = text.find("{"), text.rfind("}") if start < 0 or end <= start: return {} try: parsed = json.loads(text[start : end + 1]) except ValueError: return {} return parsed if isinstance(parsed, dict) else {} def _pick_str(args: dict, *keys: str) -> str: """按候选键取非空字符串;兼容中文别名与嵌套一层 dict。""" for key in keys: value = args.get(key) if isinstance(value, dict): # 偶发 {"text": "..."} / {"value": "..."} for nested in ("text", "value", "content", "desc", "description"): inner = value.get(nested) if isinstance(inner, str) and inner.strip(): return inner.strip() continue if value is None: continue text = str(value).strip() if text: return text return "" def _coerce_points(raw) -> list[str]: """points 规整成最多 3 条非空文案。兼容纯字符串、对象数组、dict。""" if raw is None: return [] items: list = [] if isinstance(raw, str): text = raw.strip() if not text: return [] # 中文分号/换行拆条 for part in text.replace("\r", "\n").replace("；", "\n").split("\n"): part = part.strip(" ·•-、，,") if part: items.append(part) elif isinstance(raw, dict): # {"P0": "...", "P1": "..."} 或 {"0": "..."} for key in sorted(raw.keys(), key=lambda k: str(k)): val = raw[key] if isinstance(val, dict): text = _pick_str(val, "text", "value", "content", "desc", "point", "label") else: text = str(val or "").strip() if text: items.append(text) elif isinstance(raw, (list, tuple)): for item in raw: if isinstance(item, dict): text = _pick_str(item, "text", "value", "content", "desc", "point", "label") else: text = str(item or "").strip() if text: items.append(text) else: text = str(raw).strip() if text: items.append(text) # 过滤单字符噪声(字符串被误当成 list 迭代时的残留) cleaned = [p for p in items if len(p) > 1] return cleaned[:3] def _coerce_voice_chars(raw, fallback: list[int] | None = None) -> list[int]: """voice_chars 必须是 [下限, 上限];模型常误塞语气文案或单个整数。""" if isinstance(raw, (list, tuple)) and len(raw) >= 2: try: lo, hi = int(raw[0]), int(raw[1]) if lo > 0 and hi >= lo: return [lo, hi] except (TypeError, ValueError): pass if isinstance(raw, (int, float)) and int(raw) > 0: n = int(raw) return [max(1, n - 5), n + 5] return list(fallback or []) def _coerce_timeline(raw) -> list[dict]: items: list[dict] = [] for item in (raw or []) if isinstance(raw, list) else []: if not isinstance(item, dict): continue try: start = float(item.get("start")) end = float(item.get("end")) except (TypeError, ValueError): continue stage = str(item.get("stage") or "").strip() if not stage: continue entry = {"start": start, "end": end, "stage": stage} desc = str(item.get("desc") or item.get("description") or "").strip() if desc: entry["desc"] = desc items.append(entry) return items def _default_plan_matrix(usp: str, points: list[str]) -> dict: """设计稿里的「卖点覆盖矩阵」:有 USP/支撑点就自动铺一版,避免方案卡干瘪。""" rows = [{"point": "主打卖点 USP", "hits": [1, 3]}] labels = ["体验卖点 P0", "视觉卖点 P0", "转化卖点 P0"] for index, point in enumerate(points[:3]): label = labels[index] if index < len(labels) else f"支撑卖点 P{index}" # 点名用文案前缀,hits 错落分布到 4 镜 hit = (index % 4) + 1 rows.append({"point": label if not point else f"{label}", "hits": [hit]}) return {"shots": 4, "rows": rows} def _coerce_strategy_args(args: dict) -> dict: return { "target": _pick_str( args, "target", "audience", "who", "给谁看", "目标人群", "人群", ), "trust": _pick_str( args, "trust", "credibility", "为什么相信", "信任", "可信度", ), "belief": _pick_str( args, "belief", "希望相信", "想让他信什么", "认知", "takeaway", ), "direction": _pick_str( args, "direction", "创作方向", "方向", "style", "路线", ), } _STRATEGY_TEXT_SECTION_ALIASES = { "目标受众": "target", "目标人群": "target", "这条视频给谁看": "target", "给谁看": "target", "用户为什么相信": "trust", "为什么相信": "trust", "内容逻辑": "trust", "可信依据": "trust", "信任依据": "trust", "希望用户相信什么": "belief", "希望相信": "belief", "核心卖点": "belief", "核心主张": "belief", "创作方向": "direction", "视觉调性": "direction", "视觉风格": "direction", "表达方向": "direction", } def strategy_args_from_text(text: str) -> dict: """模型漏调 write_strategy 时，把带明确栏目名的策略正文救回结构化卡片。 只接受四个栏目都能识别的高置信文本，普通聊天不会被误转成策略卡。 """ sections = {"target": [], "trust": [], "belief": [], "direction": []} current = "" for raw_line in str(text or "").splitlines(): line = re.sub(r"^\s*(?:[-*#>]+\s*)?", "", raw_line).replace("**", "").strip() if not line: continue match = re.match(r"^([^：:\n]{2,36})\s*[：:]\s*(.*)$", line) if match: heading = re.sub(r"[（(].*$", "", match.group(1)).strip().replace(" ", "") key = _STRATEGY_TEXT_SECTION_ALIASES.get(heading) if key: current = key content = match.group(2).strip() if content: sections[key].append(content) continue if heading in {"创作策略", "策略理解", "创作策略理解"}: current = "" continue if not current: continue if re.match(r"^(?:你看|请确认|如果你|是否需要|可以再)", line): continue sections[current].append(line) payload = { key: "\n".join(parts).strip() for key, parts in sections.items() } return payload if all(payload.values()) else {} def _coerce_plan_card_args(args: dict) -> dict: usp = _pick_str(args, "usp", "主打卖点", "卖点", "core_usp", "main_point") points = _coerce_points( args.get("points") if args.get("points") is not None else args.get("支撑点") or args.get("supports") or args.get("selling_points") ) # 兼容 point1/point2/point3 展开写法(设计稿 blueprint) if not points: for key in ("point1", "point2", "point3", "P0", "P1", "P2"): text = _pick_str(args, key) if text: points.append(text) points = points[:3] timeline = _coerce_timeline(args.get("timeline") or args.get("时间轴")) matrix = args.get("matrix") if not isinstance(matrix, dict) or not matrix.get("rows"): matrix = _default_plan_matrix(usp, points) if (usp or points) else {} return { "usp": usp, "points": points, "timeline": timeline, "matrix": matrix, "voice_chars": args.get("voice_chars"), } def apply_click_swap_plan_card( conversation: CreationConversation, card: dict, duration: int, ) -> dict: """把模型可能写偏的方案卡收束成可核对的点击换款时间轴。""" if not is_click_swap_preset(conversation.preset): return card sequence = click_swap_sequence(conversation) if not sequence: return card total = max(4, min(int(duration or 0), 60)) first_end = max(1, round(total * 0.2)) second_end = max(first_end + 1, round(total * 0.45)) third_end = max(second_end + 1, round(total * 0.75)) third_end = min(third_end, total - 1) second_end = min(second_end, third_end - 1) return { **card, "usp": f"手指逐次点击，商品按「{sequence}」在原位连续换款", "points": [ "固定机位、背景、光线与商品中心位置", "每次换款都由一次清楚的手指点击触发", f"严格按「{sequence}」逐款展示，结尾给出全款式总览", ], "timeline": [ { "start": 0, "end": first_end, "stage": "首款定帧", "desc": "固定机位建立首款，商品位置、尺寸、角度和背景作为后续唯一基准。", }, { "start": first_end, "end": second_end, "stage": "首次点击换款", "desc": "手指清晰点击商品，接触瞬间在原位 match cut 为下一款。", }, { "start": second_end, "end": third_end, "stage": "按序连续换款", "desc": f"按「{sequence}」继续一触一换；机位、构图、商品比例和光线完全不变。", }, { "start": third_end, "end": total, "stage": "全款式收束", "desc": "保持同一构图完成全款式总览，不加入口播、剧情、换景或字幕。", }, ], "matrix": { "shots": 4, "rows": [ {"point": "固定构图", "hits": [1, 2, 3, 4]}, {"point": "点击触发", "hits": [2, 3]}, {"point": "款式顺序", "hits": [1, 2, 3, 4]}, ], }, "voice_chars": [0, 0], } def iter_creation_agent_events( *, conversation: CreationConversation, user, text: str, refs: list[dict] | None = None, model_config: ModelConfig | None = None, record_user_message: bool = True, force_creative_turn: bool = False, continuation_instruction: str = "", ) -> Iterator[dict]: """一条用户消息 → 事件 dict 流(message/delta/tool/done/error)。消息已落库。""" refs = refs or [] fast_greeting = record_user_message and is_greeting(text) # 打招呼不依赖模型，模型未配置也能即时回应；其他消息保持原有的先校验模型行为。 if not fast_greeting: model_config = get_creation_chat_model(model_config) if model_config is None: yield {"type": "error", "detail": "没有可用的文本模型,请先在模型库配置"} return context = AgentContext(conversation=conversation, user=user, model_config=model_config) try: user_message = None with transaction.atomic(): pin_refs(conversation, refs) if record_user_message: user_message = append_message(conversation, role="user", text=text, refs=refs) if user_message is not None: yield {"type": "message", "message": _message_payload(user_message)} # 单纯打招呼无论有没有参考素材、有没有既有上下文，都不值得等模型。 # 素材先收下，等用户说清用途再分析，避免「你好」卡住还冒出一大段建议。 if fast_greeting: greet_text = ( "你好，素材我收到了。想拿它做什么，直接说就行。" if refs else "你好。想做什么，直接说就行。" ) reply = append_message( conversation, role="assistant", text=greet_text, payload={ "reply_hint": default_reply_hint( conversation, has_context=False, is_video=conversation.mode == CreationConversation.Mode.VIDEO ), "reply_options": default_reply_options( conversation, has_context=False, is_video=conversation.mode == CreationConversation.Mode.VIDEO ), }, ) yield {"type": "message", "message": _message_payload(reply)} yield {"type": "done"} return # 空会话里的简短应答：短回一句就够。已有商品/方向时走模型， # 针对用户实际说的话自然回应，但不给方案工具。 has_context = session_has_creative_context(conversation) if record_user_message and is_pure_chitchat(text) and not refs and not has_context: chit_text = "我在。想做什么，直接丢一句想法给我就行。" reply = append_message( conversation, role="assistant", text=chit_text, payload={ "reply_hint": default_reply_hint( conversation, has_context=False, is_video=conversation.mode == CreationConversation.Mode.VIDEO ), "reply_options": default_reply_options( conversation, has_context=False, is_video=conversation.mode == CreationConversation.Mode.VIDEO ), }, ) yield {"type": "message", "message": _message_payload(reply)} yield {"type": "done"} return # 剧情反转预设必须先选故事深度。平台直接落选择卡，不能把这一步交给模型猜， # 否则默认 15 秒会吞掉 30/60 秒该有的人物关系和冲突发展。 if is_plot_twist_conversation(conversation) and not active_plot_twist_story_depth(conversation): explicit_depth = plot_twist_story_depth(text) if explicit_depth is not None and explicit_depth["value"] != "smart": set_plot_twist_story_depth(conversation, explicit_depth["value"]) else: question = append_plot_twist_story_depth_question(conversation) conversation.agent_status = CreationConversation.AgentStatus.AWAITING_USER conversation.save(update_fields=["agent_status", "updated_at"]) yield {"type": "message", "message": _message_payload(question)} yield {"type": "done"} return # 点击换款的款式清单和顺序是脚本事实，不能让模型自行猜色号或把预设改成普通展示片。 if click_swap_needs_sequence(conversation): question = append_click_swap_sequence_gate(conversation) set_video_gate_stage(conversation, "clarify") conversation.agent_status = CreationConversation.AgentStatus.AWAITING_USER conversation.save(update_fields=["agent_status", "updated_at"]) yield {"type": "message", "message": _message_payload(question)} yield {"type": "done"} return # 需要真人/角色的视频必须先选定人物来源。这是平台闸门， # 不交给模型自由发挥，否则它会在脚本里随机造人，到 60s 分段时必然漂移。 if video_needs_person_source(conversation, text): question = append_person_source_gate(conversation) set_video_gate_stage(conversation, "clarify") conversation.agent_status = CreationConversation.AgentStatus.AWAITING_USER conversation.save(update_fields=["agent_status", "updated_at"]) yield {"type": "message", "message": _message_payload(question)} yield {"type": "done"} return # 明确要商品/角色/场景列表时，直接生成真实选择卡，绝不先让模型念出素材名称。 requested_card = requested_asset_card_from_context(conversation, text) if requested_card: result, _stop = _dispatch_tool( context, "ask_user", {"fields": [{ "key": requested_card, "label": _ASSET_PICK_LABEL.get( requested_card, _ASSET_CARD_LABELS.get(requested_card, "请选择素材"), ), "type": "asset", "required": True, "asset_types": [requested_card], }]}, allow_pick=True, ) for event in result.get("_events", []): yield event yield {"type": "done"} return model_config = _prefer_vision_text_model(model_config, conversation.team, conversation.pinned_refs or []) context.model_config = model_config resolved = resolve_refs(context.team, refs) if resolved.missing: names = "、".join(r.get("name") or "某个素材" for r in resolved.missing) note = append_message( conversation, role="assistant", text=f"有几个引用的素材已经找不到了({names}),我先按其余信息继续。", ) yield {"type": "message", "message": _message_payload(note)} # 压缩放在**建消息之前**:摘要要进这一轮的 system 提示词才有意义。 # 用户消息已经先回显了,所以这一小段等待不会看起来像卡住。 compress_memory(context) allow_plan = ( force_creative_turn or has_creative_intent(text, refs) or (has_context and is_continue_intent(text)) ) provider = build_provider(model_config) messages = build_messages(context, allow_plan=allow_plan, has_context=has_context) if continuation_instruction.strip(): # 卡片答案已经在历史里,这里仅给本轮一个不落库的执行指令。这样既不会多出 # 一条伪造的用户气泡,也不会让模型把「跳过素材」误判成闲聊后停住。 messages.append({"role": "user", "content": continuation_instruction.strip()}) tools = tool_schemas(context, allow_plan=allow_plan) from .creation import is_agent_cancel_requested turn_has_gate = False last_text_bubble = None for _round in range(MAX_TOOL_ROUNDS): if is_agent_cancel_requested(conversation.id): # 用户终止:干净收束,不落 ERROR,已落库消息保留 yield {"type": "cancelled"} return text_buffer: list[str] = [] tool_buffer: dict = {} for chunk in provider.chat_completion_stream( model=model_config.name, messages=messages, endpoint=model_config.endpoint or "chat/completions", extra_body={"tools": tools}, ): kind = chunk.get("type") if kind == "reasoning": yield {"type": "reasoning", "text": chunk.get("text", "")} elif kind == "delta": piece = chunk.get("text", "") text_buffer.append(piece) yield {"type": "delta", "text": piece} elif kind == "tool_call": _merge_tool_call_deltas(tool_buffer, chunk.get("tool_calls")) said = strip_numeric_reply_instruction("".join(text_buffer)) calls = [tool_buffer[i] for i in sorted(tool_buffer) if tool_buffer[i].get("name")] fallback_fields = None allow_pick = False if not calls: requested_card = requested_asset_card_from_context(conversation, text) wanted_pick = wanted_asset_pick(text, refs) pick = requested_card or wanted_pick param_keys = wanted_param_keys(text, is_video=context.is_video) if pick: fallback_fields = [{ "key": pick, "label": _ASSET_PICK_LABEL.get(pick, _ASSET_CARD_LABELS.get(pick, "请选择素材")), "type": "asset", "required": True, "asset_types": [pick], }] allow_pick = bool(requested_card) elif param_keys: fallback_fields = session_param_fields(param_keys, context.is_video) elif allow_plan and not said: # 部分推理模型会只给 reasoning_content 后结束本轮。不能让用户 # 只看到自己那条消息；没有商品上下文时，收敛成一条自然追问。 has_product = any( isinstance(ref, dict) and ref.get("type") == "product" for ref in (conversation.pinned_refs or []) ) if not has_product: fallback_fields = [{ "key": "product", "label": "这条想推哪款商品？", "type": "asset", "required": True, "asset_types": ["product"], }] # 痛点解决预设若模型只写了三条列表却漏调 ask_user，平台直接把这三条 # 转成单选按钮；不允许用户再手抄一遍，也不进入重复的核心卖点闸门。 memory = conversation.memory if isinstance(conversation.memory, dict) else {} if ( not calls and fallback_fields is None and allow_plan and is_pain_point_conversation(conversation) and not memory.get("selling_point_ready") ): pain_options = pain_point_direction_options_from_text(said) if pain_options: fallback_fields = [{ "key": PAIN_POINT_DIRECTION_KEY, "label": "选择这条视频要重点解决的痛点", "type": "single", "required": True, "options": pain_options, }] # 方向卡是剧情反转预设的固定入口。模型偶尔只会说「我准备了三个方向」而忘了调工具， # 此处直接补上可点击卡，不能让用户面对一段空话再自己追问。 memory = conversation.memory if isinstance(conversation.memory, dict) else {} needs_plot_twist_directions = ( not calls and allow_plan and is_plot_twist_conversation(conversation) and bool(active_plot_twist_story_depth(conversation)) and not str(memory.get("plot_twist_story_direction") or "").strip() ) if needs_plot_twist_directions: result, _stop = _dispatch_tool( context, "present_story_directions", {"directions": _plot_twist_direction_fallback(conversation)}, ) for event in result.get("_events", []): yield event if event.get("type") == "message": turn_has_gate = True break # 有些模型会把完整策略按「核心卖点/目标受众/内容逻辑/视觉调性」写成散文， # 却漏掉 write_strategy 工具调用。高置信识别后直接走同一条结构化策略闸门， # 避免前端退化成一整块普通聊天气泡，也避免缺失「按这个继续」。 current_stage = get_video_gate_stage(conversation) may_write_strategy = ( current_stage == "clarify" or (current_stage == "strategy" and not bool(memory.get("strategy_confirmed"))) ) prose_strategy = ( strategy_args_from_text(said) if not calls and context.is_video and may_write_strategy else {} ) if prose_strategy: result, _stop = _dispatch_tool(context, "write_strategy", prose_strategy) for event in result.get("_events", []): yield event if event.get("type") == "message": msg = event.get("message") or {} if msg.get("kind") in ( CreationMessage.Kind.ELICIT, CreationMessage.Kind.CONFIRM, ): turn_has_gate = True break # ask_user 自己会落一条可追踪的聊天问题。模型同时吐出的过渡文案不再 # 另存一条,否则界面会连续出现两遍几乎相同的问题。 asks_user = bool(fallback_fields) or any(call.get("name") == "ask_user" for call in calls) if said and not asks_user: bubble = append_message(conversation, role="assistant", text=said) last_text_bubble = bubble yield {"type": "message", "message": _message_payload(bubble)} if not calls: if fallback_fields: result, _stop = _dispatch_tool( context, "ask_user", {"fields": fallback_fields}, allow_pick=allow_pick ) for event in result.get("_events", []): yield event if event.get("type") == "message": msg = event.get("message") or {} if msg.get("kind") in ( CreationMessage.Kind.ELICIT, CreationMessage.Kind.CONFIRM, ): turn_has_gate = True break messages.append({ "role": "assistant", "content": said or None, "tool_calls": [ {"id": f"call_{i}", "type": "function", "function": {"name": c["name"], "arguments": c["arguments"]}} for i, c in enumerate(calls) ], }) stop = False for index, call in enumerate(calls): if is_agent_cancel_requested(conversation.id): yield {"type": "cancelled"} return name = call["name"] args = _parse_arguments(call["arguments"]) yield {"type": "tool", "id": name, "label": _TOOL_LABELS.get(name, name), "status": "running"} try: result, stop_after = _dispatch_tool(context, name, args) except AgentError as exc: yield {"type": "tool", "id": name, "label": _TOOL_LABELS.get(name, name), "status": "error"} failure = append_message( conversation, role="assistant", kind=CreationMessage.Kind.ERROR, text=str(exc), ) yield {"type": "message", "message": _message_payload(failure)} yield {"type": "done"} return yield {"type": "tool", "id": name, "label": _TOOL_LABELS.get(name, name), "status": "done"} for event in result.get("_events", []): yield event if event.get("type") == "message": msg = event.get("message") or {} if msg.get("kind") in ( CreationMessage.Kind.ELICIT, CreationMessage.Kind.CONFIRM, ): turn_has_gate = True messages.append({ "role": "tool", "tool_call_id": f"call_{index}", "content": json.dumps(result.get("payload", {}), ensure_ascii=False), }) stop = stop or stop_after # 闸门中断后不再执行同轮后续工具,防止策略+方案+Prompt 一锅端 if stop_after: break if stop: break # 收束校验:本轮若只剩散文、没有追问/确认卡,补 reply_hint 或短引导。 for event in ensure_turn_guides( conversation, turn_has_gate=turn_has_gate, last_text_bubble=last_text_bubble, has_context=has_context, is_video=context.is_video, ): yield event yield {"type": "done"} except Exception as exc: # noqa: BLE001 — SSE 里任何未捕获异常都会变成前端「白屏卡死」 logger.exception("creation agent stream failed: %s", exc) yield {"type": "error", "detail": "生成过程出错了,请再试一次"} def stream_creation_agent( *, conversation: CreationConversation, user, text: str, refs: list[dict] | None = None, model_config: ModelConfig | None = None, record_user_message: bool = True, force_creative_turn: bool = False, continuation_instruction: str = "", ) -> Iterator[str]: """兼容旧 SSE 消费方(单测 / 调试)。生产路径走 Celery + poll。""" for event in iter_creation_agent_events( conversation=conversation, user=user, text=text, refs=refs, model_config=model_config, record_user_message=record_user_message, force_creative_turn=force_creative_turn, continuation_instruction=continuation_instruction, ): yield _sse(event) def run_creation_agent_turn( *, conversation_id: str, user_id: str, text: str = "", refs: list | None = None, model_config_id: str | None = None, record_user_message: bool = True, force_creative_turn: bool = False, continuation_instruction: str = "", ) -> str: """Celery worker 入口:跑完一轮 tool loop,消息落库;更新 agent_status;释放团队锁。 不复用 CreationMessage.kind=generating / AITask —— 那些是确认后出片/出图用的。 """ from apps.accounts.models import User from .creation import finish_agent_planning conversation = ( CreationConversation.objects.select_related("team", "created_by") .filter(id=conversation_id) .first() ) if conversation is None: return conversation_id user = User.objects.filter(id=user_id).first() or conversation.created_by model_config = None if model_config_id: model_config = ( ModelConfig.objects.select_related("provider") .filter(id=model_config_id, capability=ModelConfig.Capability.TEXT, status=ModelConfig.Status.ACTIVE) .first() ) awaiting_user = False user_cancelled = False try: for event in iter_creation_agent_events( conversation=conversation, user=user, text=text or "", refs=refs or [], model_config=model_config, record_user_message=record_user_message, force_creative_turn=force_creative_turn, continuation_instruction=continuation_instruction or "", ): if not isinstance(event, dict): continue if event.get("type") == "cancelled": user_cancelled = True awaiting_user = False break if event.get("type") == "message": message = event.get("message") or {} kind = message.get("kind") if isinstance(message, dict) else None if kind in (CreationMessage.Kind.ELICIT, CreationMessage.Kind.CONFIRM): awaiting_user = True elif event.get("type") == "error": detail = str(event.get("detail") or "生成过程出错了,请再试一次") # 循环内多数错误已落 ERROR 消息;这里兜底一条,避免前端只看到卡在 planning last = conversation.messages.order_by("-seq").first() if last is None or last.kind != CreationMessage.Kind.ERROR: append_message( conversation, role="assistant", kind=CreationMessage.Kind.ERROR, text=detail, ) except Exception as exc: # noqa: BLE001 — worker 不能把异常冒成无限 planning logger.exception("creation agent turn failed: %s", exc) try: append_message( conversation, role="assistant", kind=CreationMessage.Kind.ERROR, text="生成过程出错了,请再试一次", ) except Exception: # noqa: BLE001 logger.exception("creation agent turn: failed to append error message") awaiting_user = False finally: try: conversation.refresh_from_db(fields=["agent_status", "team_id"]) except Exception: # noqa: BLE001 pass # 用户终止:恢复闸门确认条(若有),再 finish;避免卡在「已收到…正在重写」且 agent 变 idle if user_cancelled: try: awaiting_user = restore_gated_step_after_cancel(conversation) except Exception: # noqa: BLE001 logger.exception("creation agent cancel: restore gated step failed") awaiting_user = False finish_agent_planning( conversation, awaiting_user=awaiting_user, ) return conversation_id _TOOL_LABELS = { "ask_user": "向你确认", "present_story_directions": "整理剧情方向", "search_library": "查找素材", "generate_image": "生成图片", "write_strategy": "梳理创作策略", "write_plan": "编排视频方案", "write_prompt": "整理出片指令", } _ASSET_CARD_LABELS = { "product": "这条要展示哪款商品？", "character": "这条想用哪个角色？", "model": "这条想用哪位模特？", "scene": "这条想放在哪个场景里？", "asset": "这一步要使用哪项素材？", } _GATE_LABELS = { "product": "这条想推哪款商品？可以直接告诉我商品名，或者需要我把商品库列表发给你选吗？", "character": "这条想用哪个角色？可以直接告诉我角色名，或者需要我发角色列表给你选吗？", "model": "这条想用哪位模特？可以直接告诉我，或者需要我把模特库发给你选吗？", "scene": "这条想放在哪个场景里？可以直接告诉我，或者需要我把场景库发给你选吗？", "asset": "这一步想用哪项素材？可以直接告诉我，或者需要我把素材列表发给你选吗？", } def _guided_elicit_text(field: dict) -> str: """追问既要问清一件事,也要让用户知道下一句怎么回。""" label = str(field.get("label") or "这项你想怎么定？").strip() if field.get("type") == "asset" or field.get("key") == "_asset_gate": return label if str(field.get("key") or "") in SESSION_PARAM_KEYS: return label if field.get("type") == "single" and field.get("options"): return f"{label} 直接点击一个选项，也可以输入自己的想法。" return f"{label} 直接用一句话告诉我就行，不用整理成完整需求。" def _elicit_payload_for_fields( fields: list[dict], *, allow_pick: bool = False, is_video: bool = True ) -> dict: """素材问题默认出轻量对话闸门；用户明确要求发列表时才出 pick 选择卡。""" field = dict(fields[0]) if field.get("type") == "asset": asset_types = [item for item in (field.get("asset_types") or []) if item in _GATE_LABELS] primary_type = asset_types[0] if asset_types else "product" if allow_pick: field["label"] = _ASSET_CARD_LABELS.get(primary_type, _ASSET_CARD_LABELS["asset"]) return { "interaction": "asset_picker", "phase": "pick", "fields": [field], "submitted": False, "answers": {}, } if primary_type == "product": gate_label = ( "这条视频想推哪款商品？可以直接告诉我商品名，或者需要我把商品库列表发给你选吗？" if is_video else "这次想做哪款商品？可以直接告诉我商品名，或者需要我把商品库列表发给你选吗？" ) else: gate_label = _GATE_LABELS.get(primary_type, _GATE_LABELS["asset"]) asset_label = { "product": "商品", "character": "角色", "model": "模特", "scene": "场景", }.get(primary_type, "素材") gate_field = { "key": "_asset_gate", "label": gate_label, "type": "text", "required": False, "options": [ {"value": "send", "label": f"发{asset_label}列表"}, {"value": "auto", "label": "你来推荐"}, ], } return { # 这是 Agent 的一句自然追问，不是让用户点选流程的卡片。 # 只有用户明确要求发列表后，才会生成上面 allow_pick 分支的商品卡。 "interaction": "chat", "phase": "gate", "fields": [gate_field], "pending_fields": [field], "submitted": False, "answers": {}, } return { "interaction": "chat", "fields": [field], "submitted": False, "answers": {}, } def _dispatch_tool( context: AgentContext, name: str, args: dict, *, allow_pick: bool = False ) -> tuple[dict, bool]: """执行一个工具。返回 (结果, 是否中断循环)。 结果里的 `_events` 会原样转发给前端,`payload` 回喂给模型。 """ if name == "ask_user": fields = _coerce_fields(args.get("fields")) if not fields: return {"payload": {"error": "fields 不合法,请重新组织问题"}}, False payload = _elicit_payload_for_fields(fields, allow_pick=allow_pick, is_video=context.is_video) display_field = (payload.get("fields") or fields)[0] message = append_message( context.conversation, role="assistant", kind=CreationMessage.Kind.ELICIT, text=_guided_elicit_text(display_field), payload=payload, ) # 反问一旦发出就必须停,等人回答。继续跑等于自问自答。 if context.is_video: set_video_gate_stage(context.conversation, "clarify") return { "payload": {"asked": True}, "_events": [{"type": "message", "message": _message_payload(message)}], }, True if name == "present_story_directions": if not is_plot_twist_conversation(context.conversation): return {"payload": {"error": "当前会话不是剧情反转带货预设"}}, False directions = _coerce_plot_twist_directions(args.get("directions")) if not directions: return { "payload": { "error": "必须提供恰好 3 个完整剧情方向，每个都要有标题、冲突、商品作用、反转和情绪。" } }, False message = _append_plot_twist_direction_question(context, directions) set_video_gate_stage(context.conversation, "clarify") return { "payload": {"presented": True, "count": 3}, "_events": [{"type": "message", "message": _message_payload(message)}], }, True if name == "search_library": return {"payload": _run_search_library(context, args)}, False if name == "write_strategy": if context.is_video and click_swap_needs_sequence(context.conversation): gate = append_click_swap_sequence_gate(context.conversation) set_video_gate_stage(context.conversation, "clarify") return { "payload": {"asked": True, "field": "sku_sequence"}, "_events": [{"type": "message", "message": _message_payload(gate)}], }, True if context.is_video and video_needs_person_source( context.conversation, json.dumps(args if isinstance(args, dict) else {}, ensure_ascii=False), ): gate = append_person_source_gate(context.conversation) set_video_gate_stage(context.conversation, "clarify") return { "payload": {"asked": True, "field": "person_source"}, "_events": [{"type": "message", "message": _message_payload(gate)}], }, True memory = context.conversation.memory if isinstance(context.conversation.memory, dict) else {} if context.is_video and is_pain_point_conversation(context.conversation) and not memory.get("selling_point_ready"): return { "payload": { "error": ( "痛点解决演示必须先调用 ask_user 展示痛点方向三选一：" "key=pain_point_direction、type=single、恰好 3 个 options。" "用户点击后该方向会直接成为核心卖点，不要另开卖点确认卡。" ) } }, False if context.is_video and not memory.get("selling_point_ready"): gate = append_selling_point_gate(context.conversation) set_video_gate_stage(context.conversation, "clarify") return { "payload": {"asked": True, "field": "selling_point"}, "_events": [{"type": "message", "message": _message_payload(gate)}], }, True strategy_payload = _coerce_strategy_args(args if isinstance(args, dict) else {}) # 空卡会落成「只有标签没有正文」——拒绝,让模型把四字段写满再调 if not all(strategy_payload.values()): return { "payload": { "error": ( "创作策略卡四字段都不能为空:请填写具体的 target / trust / belief / direction" "(给谁看、为什么信、希望他信什么、创作方向),不要留空。" ) } }, False message = append_message( context.conversation, role="assistant", 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, "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 if context.is_video and click_swap_needs_sequence(context.conversation): gate = append_click_swap_sequence_gate(context.conversation) set_video_gate_stage(context.conversation, "clarify") return { "payload": {"asked": True, "field": "sku_sequence"}, "_events": [{"type": "message", "message": _message_payload(gate)}], }, True if context.is_video and video_needs_person_source(context.conversation, video_prompt): gate = append_person_source_gate(context.conversation) set_video_gate_stage(context.conversation, "clarify") return { "payload": {"asked": True, "field": "person_source"}, "_events": [{"type": "message", "message": _message_payload(gate)}], }, True video_prompt = apply_video_preset_prompt(context.conversation.preset, video_prompt) if is_click_swap_preset(context.conversation.preset): video_prompt = ( f"{video_prompt}\n\n【商家确认的换款顺序】{click_swap_sequence(context.conversation)}\n" "只允许按这个顺序逐款切换；不得跳序、漏款或自行增加颜色和款式。" ) video_prompt = apply_product_voice_visual_guard(context.conversation, video_prompt) video_prompt = apply_product_reality_guard(video_prompt) video_prompt = apply_video_platform_safety_guard(video_prompt) video_prompt = apply_plot_twist_story_contract( context.conversation.preset, active_plot_twist_story_depth(context.conversation), video_prompt, ) card = _coerce_plan_card_args(args if isinstance(args, dict) else {}) if not card["usp"] or not card["points"]: return { "payload": { "error": ( "方案卡缺正文:请填写非空的 usp(主打卖点)和 points(1–3 条核心支撑)," "不要只写 video_prompt。卡片上要让用户看见卖点文案。" ) } }, False selling_memory = context.conversation.memory if isinstance(context.conversation.memory, dict) else {} chosen_selling_point = str(selling_memory.get("selling_point") or "").strip() if chosen_selling_point and str(selling_memory.get("selling_point_mode") or "") == "manual": # 用户亲自给出的真实卖点是本轮的唯一 USP，模型只能围绕它补充画面证据，不能擅自替换。 card["usp"] = chosen_selling_point video_prompt = ( f"{video_prompt}\n\n【商家确认的核心卖点】{chosen_selling_point}\n" "整条视频必须围绕这个卖点展开，并用真实使用动作或素材可见细节证明；不得替换、夸大或新增未经确认的功效。" ) raw_duration = _raw_script_duration(timeline=card["timeline"], prompt=video_prompt) if raw_duration is not None and raw_duration > 60: return { "payload": {"error": "脚本时长不能超过 60 秒。请把方案收束到 60 秒以内，再重新给出完整时间轴和出片指令。"} }, False duration = resolve_smart_video_duration( context.conversation, prompt=video_prompt, timeline=card["timeline"], ) card = apply_click_swap_plan_card(context.conversation, card, duration) lo = max(20, round(duration * 3.4)) hi = max(lo + 1, round(duration * 4)) plan_payload = { "usp": card["usp"], "points": card["points"], "timeline": card["timeline"], "matrix": card["matrix"], "voice_chars": ( [0, 0] if is_click_swap_preset(context.conversation.preset) else _coerce_voice_chars(card["voice_chars"], [lo, hi]) ), "ref_count": len(context.conversation.pinned_refs or []), } events = [] plan = append_message( context.conversation, role="assistant", kind=CreationMessage.Kind.PLAN, payload=plan_payload, ) events.append({"type": "message", "message": _message_payload(plan)}) # video_prompt 先作为方案产物存档；用户确认方案后会展示指令文件供确认。 set_video_gate_stage( context.conversation, "plan", pending_video_prompt=video_prompt ) 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 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 if context.is_video and click_swap_needs_sequence(context.conversation): gate = append_click_swap_sequence_gate(context.conversation) set_video_gate_stage(context.conversation, "clarify") return { "payload": {"asked": True, "field": "sku_sequence"}, "_events": [{"type": "message", "message": _message_payload(gate)}], }, True video_prompt = apply_video_preset_prompt(context.conversation.preset, video_prompt) video_prompt = apply_product_voice_visual_guard(context.conversation, video_prompt) video_prompt = apply_product_reality_guard(video_prompt) video_prompt = apply_video_platform_safety_guard(video_prompt) video_prompt = apply_plot_twist_story_contract( context.conversation.preset, active_plot_twist_story_depth(context.conversation), video_prompt, ) prompt_messages = emit_prompt_gate(context.conversation, video_prompt) if not prompt_messages: return {"payload": {"error": "无法准备出片指令"}}, False events = [ {"type": "message", "message": _message_payload(message)} for message in prompt_messages ] return { "payload": {"awaiting_step": "prompt", "prepared": True}, "_events": events, }, True if name == "generate_image": if context.generations_used >= MAX_BILLED_GENERATIONS: # 一条用户消息只计费一次。模型想连出好几版时在这里挡住。 return {"payload": {"error": "本轮已经生成过一次了,请让用户看过再决定要不要改"}}, True prompt = str(args.get("prompt") or "").strip() if not prompt: return {"payload": {"error": "生成失败:模型没有给出画面描述"}}, False prompt = apply_image_preset_prompt(context.conversation.preset, prompt) # 出图也走确认卡:用户先看当前模型/比例/张数,点了才提交。 credits = estimate_image_credits(context) confirm = append_message( context.conversation, role="assistant", kind=CreationMessage.Kind.CONFIRM, payload={ "kind": "image", "label": "开始生成", "estimated_credits": credits, "prompt": prompt, "submitted": False, "params": snapshot_session_params(context.conversation), "param_options": confirm_param_options(False), }, ) events = [ {"type": "message", "message": _message_payload(confirm)}, {"type": "credits", "estimated": credits}, ] return {"payload": {"awaiting_confirmation": True}, "_events": events}, True return {"payload": {"error": f"未知工具 {name}"}}, False def _summarize_source(messages: list[CreationMessage]) -> str: """要压缩的那批消息 → 喂给模型的纯文本。只取有信息量的部分。""" lines = [] for message in messages: if message.kind == CreationMessage.Kind.TEXT and message.text.strip(): who = "用户" if message.role == "user" else "我" lines.append(f"{who}:{message.text.strip()}") elif message.kind == CreationMessage.Kind.ELICIT: answers = (message.payload or {}).get("answers") or {} if answers: lines.append("用户确认:" + "；".join(f"{k}={v}" for k, v in answers.items())) elif message.kind in (CreationMessage.Kind.GENERATING, CreationMessage.Kind.RESULT): prompt = (message.payload or {}).get("prompt") or "" if prompt: lines.append(f"我生成了一版:{prompt[:120]}") return "\n".join(lines) def compress_memory(context: AgentContext) -> None: """把早期消息压成一段摘要存进 conversation.memory.summary(契约 §5)。 只在消息数超过阈值时做;压缩失败**静默跳过** —— 摘要是锦上添花, 为它把整条对话打断不值得。压缩额外调一次模型,所以按 summarized_upto 记进度,同一批消息不重复压。 """ conversation = context.conversation history = list(conversation.messages.all()) if len(history) <= COMPRESS_AFTER_MESSAGES: return memory = dict(conversation.memory or {}) cutoff = len(history) - KEEP_RECENT_MESSAGES if cutoff - int(memory.get("summarized_upto") or 0) < COMPRESS_MIN_BATCH: return # 没压过的还不够一批,攒着 —— 每轮重压一次太贵 source = _summarize_source(history[:cutoff]) if not source.strip(): memory["summarized_upto"] = cutoff conversation.memory = memory conversation.save(update_fields=["memory", "updated_at"]) return previous = memory.get("summary") or "" instruction = ( "把下面这段创作对话压成一段中文摘要,150 字以内。" "保留:用户明确提过的要求和否决过的方向、已确认的设定、生成过什么。" "丢掉:寒暄、过程性的话。直接输出摘要正文,不要前言。\n\n" + (f"【已有摘要】{previous}\n\n" if previous else "") + f"【新增对话】\n{source}" ) try: provider = build_provider(context.model_config) pieces = [] for chunk in provider.chat_completion_stream( model=context.model_config.name, messages=[{"role": "user", "content": instruction}], endpoint=context.model_config.endpoint or "chat/completions", ): if chunk.get("type") == "delta": pieces.append(chunk.get("text", "")) summary = "".join(pieces).strip() except Exception: # noqa: BLE001 — 摘要失败不该打断对话 logger.warning("omni create: memory compression failed", exc_info=True) return if not summary: return memory["summary"] = summary[:400] memory["summarized_upto"] = cutoff conversation.memory = memory conversation.save(update_fields=["memory", "updated_at"]) def _remember_artifact(conversation: CreationConversation, prompt: str, kind: str) -> None: """记进产物索引,让下一轮「把背景换成夜景」能定位到这一版(契约 §5)。""" memory = dict(conversation.memory or {}) artifacts = list(memory.get("artifacts") or []) artifacts.append({"prompt": prompt, "kind": kind}) memory["artifacts"] = artifacts[-10:] conversation.memory = memory conversation.save(update_fields=["memory", "updated_at"]) def _message_payload(message: CreationMessage) -> dict: from .serializers import CreationMessageSerializer return CreationMessageSerializer(message).data