Files
yingqing/core/backend/apps/ai/creation_agent.py
T
2026-09-16 16:07:56 +08:00

3339 lines
153 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""全能创作 · Agent 编排循环(契约 §3/§4)。
和 script_agent.py 的根本区别:那边是「单次结构化出稿」,模型只会写脚本;
这边是**真 function calling 循环** —— 模型自己决定这一轮该反问用户、该查素材,
还是该出图/出片。
铁律(踩过就回不来的三条):
1. **视频 510 分钟,绝不在 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,
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, 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")
_STEP_CONFIRM_LABELS = {
"strategy": "创作策略已写好。确认后继续写方案;要改就点「我想改」或直接说改哪里。",
"plan": "视频方案已写好。确认后我会整理出片细节,并带你确认生成参数;要改就点「我想改」或直接说改哪里。",
"prompt": "出片指令已整理好。确认后核对参数并生成;要改就点「我想改」或直接说改哪里。",
}
def is_plot_twist_conversation(conversation: CreationConversation) -> bool:
return conversation.mode == CreationConversation.Mode.VIDEO and conversation.preset == PLOT_TWIST_PRESET
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 _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)
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)
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(
r"(?<!\d)(\d{1,2}(?:\.\d+)?)\s*(?:-|—||~|至|到)\s*(\d{1,2}(?:\.\d+)?)\s*秒?"
)
_TOTAL_DURATION_RE = re.compile(
r"(?:总时长|成片时长|时长)\s*[:]?\s*(\d{1,2}(?:\.\d+)?)\s*秒",
re.IGNORECASE,
)
# 全能创作 video_prompt 写作规范。目标是把 Prompt 写成导演、摄影、声音与出片模型都能
# 直接执行的制作文件,而不是只有几行「0-3 秒做什么」的提纲。
_OMNI_VIDEO_PROMPT_RULES = """
【视频生成 Prompt · 制作级交付】
video_prompt 是交给出片模型的完整制作文件,不是方案摘要、更不是几句分镜大纲。保留用户选定的
预设、人物、商品和创意方向;下面规定的是文档完整度与写法,不能照搬任何其他项目的商品、人设或台词。
请按以下顺序完整输出,每个标题都要有具体内容:
时长:X 秒以内,比例:X。任务:一句话说明要生成什么、引用哪些参考素材。
标题:《本片标题》
风格与视觉参考:说明成片质感、真实度、拍摄媒介感与 4–6 个风格关键词。
镜头语言:说明主景别、人物/商品与镜头关系、机位、手持或稳定方式、镜头切换节奏。
视觉与美术规则:说明场景陈设、主体外观连续性、商品外观连续性;有参考图时要说清「只取什么特征、忽略什么背景/构图」。
色彩与材质系统:写出主色、材质、皮肤/产品/环境的可见质感。
打光规则:光源方向、软硬、色温、人物和产品分别如何受光。
剪辑节奏:列出时间段与 Hook → 证据/体验 → 转化收束的推进逻辑。
声音方向:人声身份、语气、语速、环境声/拟音、背景音乐的进入和收束;人声原文必须分配到对应分镜。
场景:逐一写清可见地点、前中后景、环境道具与景深。
主体与参考素材:逐一说明角色、商品、场景的可见身份和一致性要求。已 @ 的素材按 @图片1、@图片2 … 标注其用途;仅引用实际提供的素材。
各个分镜的具体内容:按时间顺序逐段展开。
分镜不能省略细节:每一段都必须严格用下面四行写完,不得只写一句画面描述:
「03 秒」
拍法:景别 + 机位 + 运镜 + 节奏。
画面内容:谁在何处、哪只手/身体如何动作、商品如何进入画面、前后发生什么可见变化。
主体/产品露出:本段产品、人物或关键物件处在什么位置,哪些外观细节必须清晰稳定。
声音:人声(仅音频,由人物口型与配音表达,不作为画面文字):{本段原话};再写清对应拟音和背景音乐状态。
硬性质量要求:
- 15 秒成片至少写 4 段连续分镜;更长时按约 3–5 秒一段拆开,至少切换两种景别与两种机位/运镜。
- 15 秒 Prompt 通常不少于约 1200 个汉字;短片按时长同比例展开。每一段都要有可拍的连续动作、物理反馈或产品证据,不能用「高级感、展示质感、氛围拉满」代替动作。
- 人声必须是自然口语,约 5.0–5.7 字/秒;钩子前 15 个字禁止「大家好 / 今天分享 / 给你们推荐」。CTA 像朋友提醒,禁止小黄车、立即购买、闭眼入等平台指令腔。
- 全片围绕一个具体情境和一个主卖点推进,卖点必须有可见证据,例如质地、使用动作、前后变化或真实反应。
- 商品演示先过“真实用途与状态”检查:只展示商品在其已知用途和已提供事实范围内的正常、完整状态;
禁止无依据出现破损、漏液、渗水、失效、异常变形、脏污,或把商品拿去做超出用途的压力测试。
不确定防水、防漏、承重、耐热、容量、材质等关键能力时,不编造测试和结果;要么问用户,要么改用可观察的正常使用动作。
- 画面中不出现新增字幕、花字、标题贴片、弹幕、角标、水印、购物浮层或说明性文字;口播只存在于声音,包装本身原有印刷字除外。
- 结尾单列「全片一致性与禁用项」:重申角色、商品、场景、光线、服装/材质的连续性,以及本片最重要的画面禁用项。
- 用户说「商品说话 / 商品自述 / 商品拟人」时,默认无脸拟人:商品声音是画外角色声,商品本体不做口型、不新增卡通五官;性格只通过整体倾斜、转向、弹跳、进退、镜头和音效表达。只有用户明确要求可见卡通五官时才例外。
"""
_PRODUCT_VOICE_HINT_RE = re.compile(
r"(商品|产品|包装|瓶|机身).{0,8}(说话|开口|自述|拟人)"
r"|拟人.{0,8}(商品|产品|包装|瓶|机身)"
)
_VISIBLE_PRODUCT_FACE_RE = re.compile(
r"(商品|产品|包装|瓶|机身).{0,12}(卡通)?(五官|眼睛|眼珠|嘴巴|嘴|口型)"
r"|(卡通)?(五官|眼睛|眼珠|嘴巴|嘴|口型).{0,12}(商品|产品|包装|瓶|机身)"
r"|(给|让).{0,12}(长出|加上|出现).{0,6}(卡通)?(五官|眼睛|眼珠|嘴巴|嘴|口型)"
)
PRODUCT_VOICE_VISUAL_GUARD = (
"【商品角色表现·最高优先级】商品始终是参考图里的真实物件。包装正面、瓶身、机身和全部可见表面"
"保持原有设计,只保留参考图本来就有的标签、图案与结构。拟人感完全通过整个商品轻微倾斜、"
"转向、弹跳、进退,配合镜头、光影、环境反应和音效表达。商品台词采用画外角色声,声源不在"
"画面内,商品保持完整物件形态;场景里的其他物件也保持真实原貌。整体采用精致实拍广告质感"
"与克制幽默。前文若有改造商品外观的动作描述,统一改成商品整体运动的对应表达。"
)
# 这段在最终出片 Prompt 上再加一道确定性约束,避免模型虽然在方案里写了“真实使用”,
# 但生成时又把商品拍进不合理的失效状态。它不替用户补商品性能,只限制无依据的错误演示。
PRODUCT_REALITY_GUARD = (
"【商品真实使用约束·最高优先级】商品只按已知真实用途和用户提供的事实展示,"
"全程保持正常、完整、可用的状态。没有用户明确提供的性能依据时,不出现破损、漏液、渗水、"
"失效、异常变形、脏污、超载或超出用途的测试;也不把不确定的防水、防漏、承重、耐热、容量、"
"材质等能力拍成已被验证的结果。若关键信息不足,使用可观察的正常操作替代夸张测试。"
)
def apply_product_voice_visual_guard(conversation: CreationConversation, prompt: str) -> 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()
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 "")
if re.search(r"颜色|色号|色彩|SKU|款式|几种", text, re.IGNORECASE):
return [
{"label": "补充颜色和顺序", "text": "我来补充每个颜色和展示顺序"},
{"label": "上传各款实物图", "text": "我补充各颜色/款式的实物图"},
{"label": "先按当前主款做", "text": "先按当前主款做,其他颜色后面再补"},
]
if re.search(r"商品|产品|主推|哪款", text, re.IGNORECASE):
return [
{"label": "发商品列表", "text": "把商品列表发给我选"},
{"label": "我直接说商品名", "text": "我直接告诉你商品名"},
{"label": "你来推荐", "text": "你根据当前需求推荐一款"},
]
if re.search(r"人物|角色|模特|出镜", text, re.IGNORECASE):
return [
{"label": "上传人物图", "text": "我上传人物参考图"},
{"label": "由你设定角色", "text": "你先帮我设定一个合适的角色"},
{"label": "不需要人物", "text": "这条先不需要人物出镜"},
]
if re.search(r"场景|地点|背景|在哪", text, re.IGNORECASE):
return [
{"label": "上传场景图", "text": "我上传场景参考图"},
{"label": "你来推荐场景", "text": "你按商品和预设推荐场景"},
{"label": "用干净日常场景", "text": "先用干净自然的日常场景"},
]
if re.search(r"卖点|功能|效果|优惠|价格", text, 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)
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 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
current[key] = value
changed = True
duration_changed = (
conversation.mode == CreationConversation.Mode.VIDEO
and str(current.get("duration") or "") != 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]:
"""按方案节奏把 3160 秒视频拆成可由 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 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_plot_twist_story_contract(
context.conversation.preset,
active_plot_twist_story_depth(context.conversation),
prompt,
)
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,
}
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_submit = {
**submit,
"duration": int(segment["duration"]),
"prompt": segment_video_prompt(prompt, segment, total_duration) if len(segments) > 1 else submit["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 弹出可选项,"
"要么在文字里明确告诉用户该回复什么(例如「直接说想改的卖点」)。",
"- 当你给出 23 个方向时,必须调用 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 都已安排。内容超出时长时建议删减、延长或拆分,而不是硬塞。",
])
# 按会话时长给出口播字数锚点(与专业创作 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,也不要让用户输入编号。"
)
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", "路线",
),
}
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 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
# 明确要商品/角色/场景列表时,直接生成真实选择卡,绝不先让模型念出素材名称。
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"],
}]
# 方向卡是剧情反转预设的固定入口。模型偶尔只会说「我准备了三个方向」而忘了调工具,
# 此处直接补上可点击卡,不能让用户面对一段空话再自己追问。
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
# 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
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":
memory = context.conversation.memory if isinstance(context.conversation.memory, dict) else {}
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
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_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(13 条核心支撑),"
"不要只写 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"],
)
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": _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
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_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