Files
yingqing/core/backend/apps/ai/creation_agent.py
T
2026-09-11 19:11:27 +08:00

2084 lines
93 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` 一旦被调用就中断循环。** 反问的意义是等人回答,继续跑下去
等于自问自答。
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 preset_guidance
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
# 记忆压缩(契约 §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",
}
# 「智能时长」= 交给我们定,取一个口播讲得完又不烧钱的中间值
SMART_DURATION = 15
# 全能创作 video_prompt 写作规范:从专业创作 / 一键成片(口播脚本)抽硬规则,
# 适配「整段出片指令」而不是 ScriptDraft JSON。只要求口径接近,不改工具形态。
_OMNI_VIDEO_PROMPT_RULES = """
【出片脚本写法 · 对齐专业口播】
表现形式默认**口播带货**,像真人在讲刚遇到的一件事,不要详情页朗读或主播念稿。
video_prompt 必须能被出片模型直接执行,按时间轴写秒级分镜,建议结构:
- 片头写清:总时长、画幅、整体光线/色调、口播语气(口语、有停顿与立场)
- 然后按「0-3s / 3-8s / …」逐段写,每段同时写清这五项(不能省):
1. 景别(大特写/特写/近景/中景/全景;一条片里至少切两次景别)
2. 机位(平视/俯拍/仰拍/过肩/桌面视角)
3. 运镜(手持跟拍/推近/拉远/横摇/环绕/固定 —— 每段至少一个运镜词)
4. 动作(谁、哪只手、对什么、做什么;要连贯可拍,禁止「展示质感」等抽象词)
5. 信息变化(这几秒画面上多了/变了什么)
- 人声原文单独写清,必须用「人声(仅音频,由人物口型与配音表达,不出现在画面上):…」,禁止写成「口播:…」(裸写口播会被出片模型烧成字幕);字数贴近会话时长:大约 5.0–5.7 字/秒,
15 秒约 75–85 字;太短撑不满,太长会赶。
- 钩子段画面不要「对着镜头说话」静态开场;前 15 个口播字禁止「大家好/今天分享/给你们推荐」。
- 全片只围绕一个具体情境推进一个主卖点;卖点要有看得见的证据(质地/前后变化/用法结果)。
- CTA 像跟朋友说话;禁止小黄车/立即购买/闭眼入等平台指令腔。
- **不要写字幕/花字/标题贴片/弹幕/角标/水印/购物浮层**,也不要写「无字幕」
(否定说法也容易把字画上屏)。口播只存在于声音;包装上原有印刷字除外。
- 已 @ 的角色/商品/场景参考图会自动附上,不要在 prompt 里重描长相;以图锁定性别年龄服装外形。
- 用户说「商品说话 / 商品自述 / 商品拟人」时,默认采用**无脸拟人**:商品声音是画外角色声,
商品本体不做口型、不新增卡通五官;性格靠整体倾斜、转向、弹跳、进退、镜头和音效表达。
只有用户明确要求可见的卡通五官时才例外。
- 同一场戏保持地点、光线、服装连续;要换环境就明确写下一时间段切换。
"""
_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 = (
"【商品角色表现·最高优先级】商品始终是参考图里的真实物件。包装正面、瓶身、机身和全部可见表面"
"保持原有设计,只保留参考图本来就有的标签、图案与结构。拟人感完全通过整个商品轻微倾斜、"
"转向、弹跳、进退,配合镜头、光影、环境反应和音效表达。商品台词采用画外角色声,声源不在"
"画面内,商品保持完整物件形态;场景里的其他物件也保持真实原貌。整体采用精致实拍广告质感"
"与克制幽默。前文若有改造商品外观的动作描述,统一改成商品整体运动的对应表达。"
)
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()
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))
# 明确要做内容才算出方案/出图。闲聊、吐槽、问功能都不算。
_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 _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()))
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
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 秒"]
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"])
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"])
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自然对话轻量询问是否需要发送商品列表(同时支持直接输入名字或由你推荐),避免一上来粗暴弹出大卡片打断交流。"
"一次只问 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:
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": (
"写「视频最终方案」卡并请用户确认。**仅当用户明确要做片/出方案/改方案时调用**;"
"打招呼或闲聊不要调。调完会等用户点确认,确认后平台直接按 video_prompt 出片。"
"usp / points / timeline 是给用户看的卡片正文,必须写满具体文案,禁止空着只交 video_prompt。"
"video_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": (
"交给出片模型的完整口播带货指令(对齐专业创作口径)。"
"必须含:总时长与画幅、整体光线色调、按 0-Ns 分段的秒级分镜"
"(每段写清景别/机位/运镜/具体动作/信息变化)、人声(仅音频抬头)原文、一个主卖点与可见证据、口语 CTA。禁止「口播:」字样。"
"禁止字幕/花字/贴片及「无字幕」字样;禁止详情页腔与「大家好」开场。"
"已 @ 素材会自动作参考图,勿重描长相。"
),
},
},
"required": ["usp", "points", "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 video_duration(params: dict) -> int:
"""「15 秒」→ 15;「智能时长」/ 解析不出 → SMART_DURATION。"""
raw = str(params.get("duration") or "")
digits = "".join(ch for ch in raw if ch.isdigit())
if not digits:
return SMART_DURATION
return max(4, min(int(digits), 30))
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("生成失败:模型没有给出画面描述")
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 的语义依据。"""
params = context.conversation.params or {}
resolved = resolve_refs(context.team, context.conversation.pinned_refs or [])
prompt = apply_product_voice_visual_guard(context.conversation, prompt)
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": video_duration(params),
"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 .free_video import submit_free_video
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)
try:
task = submit_free_video(team=conversation.team, user=user, params=submit)
except ValueError as exc: # 校验类错误(时长/比例/额度),给用户看原文
return None, str(exc)
message = append_message(
conversation, role="assistant", kind=CreationMessage.Kind.GENERATING,
payload={"task_id": str(task.id), "kind": "video", "prompt": prompt}, task=task,
)
_remember_artifact(conversation, prompt, "video")
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}**,这一点在整个会话里不会改变 —— 用户要另一种就请他新开一个创作。",
"",
"【怎么说话】",
"- 说人话,像个懂创作、会一起把事做完的同事;自然、简短、有判断,不要写成客服话术或需求确认清单。",
"- 禁止用「好的」「收到」「明白了」「有需要再说」「我将为你」起手;这些句子没有信息量,也很像机器人。",
"- 不要逐字复述用户刚说过的话。需要承接时,只说你的判断或下一步,例如「这个方向能做,我先把反转落在商品登场上。」",
"- 一次只推进一步,但不要把能直接做的事停在寒暄、确认或客套话上。",
"- 缺信息时一次只问一个真正影响结果的问题,不要把对话做成问卷。",
"- 缺少商品、模特、角色、场景等素材时:**绝不要直接弹出大块素材选择卡**打断对话。像懂创作、懂电商的专业伙伴一样先自然询问用户想推什么商品/用哪个角色,询问是否需要把商品列表发给他选,同时说明也可以直接输入商品名或由你推荐。",
"- 只有当用户明确说要发列表(如「发给我」「发列表」「给我看看」「我来选」)时,才展示可视化卡片。",
"- 用户直接打字输入商品名时,直接采纳该商品并继续推进创作方案,不要强迫用户去卡片里点选。",
"- 调性、受众、文案、时长等其他信息才用聊天追问,问题末尾顺手告诉用户下一句怎么回。",
"- 用户刚回答过你的问题时,直接沿着答案继续;不要复述答案,也不要额外回一句「收到」。",
"- 用户选择暂不提供某项素材时,把它当成明确授权:按已有信息和合理默认继续。除非任务客观上无法完成,否则不要再次追问同一素材。",
"- 用户说「你来定」「你帮我选」「随便」「都行」时,就是授权你做专业判断;直接选合理方案继续,不要把选择题再抛回去。",
"- 禁止问「要不要继续」「要不要生成」「是否开始创作」这类流程问题。用户已经提出创作需求且信息够了,就直接写策略/方案;真正扣积分前另有确认卡。",
"- 用户打招呼或闲聊(hi / 你好 / 在吗 / 你在干什么 / 嗯 / 好的 / ok):"
" **禁止**调用 write_strategy、write_plan、generate_image,不要「整理方案」或直接开写脚本。",
"- 会话里**还没有**商品/方向时:自然地告诉用户可以直接丢一句想法,别硬推销,也别用客服式结束语。",
"- 会话里**已有**商品或方向时:针对用户这句话本身自然回应;别用「要现在生成还是先调细节」把工作又抛回给用户。",
"- 没有明确创作指令时不要自己写策略卡/方案卡,也不要主动追问流程确认;等用户给出具体想法或修改。",
"",
"【什么时候反问】",
"- 只有信息**确实缺失且无法合理推断**时才调用 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()])
# 按会话时长给出口播字数锚点(与专业创作 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("- 只有用户明确要做片、出方案、改方案、换卖点/剧情时才调用 write_strategy / write_plan;闲聊与打招呼绝对不要。真要写方案时必须先 write_strategy 再 write_plan,video_prompt 按上面的秒级分镜规范写满,不要只给大纲,也禁止只回「好的,有需要再说」。")
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 guidance:
lines.append(guidance)
lines.append("用户选了这个预设,就按它的拍法来;要偏离得先问过用户。")
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:
reply = append_message(
conversation,
role="assistant",
text=(
"你好,素材我收到了。想拿它做什么,直接说就行。"
if refs else "你好。想做什么,直接说就行。"
),
)
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:
reply = append_message(
conversation,
role="assistant",
text="我在。想做什么,直接丢一句想法给我就行。",
)
yield {"type": "message", "message": _message_payload(reply)}
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
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 = "".join(text_buffer).strip()
calls = [tool_buffer[i] for i in sorted(tool_buffer) if tool_buffer[i].get("name")]
fallback_fields = None
allow_pick = False
if not calls:
requested_card = requested_asset_card_from_context(conversation, text)
wanted_pick = wanted_asset_pick(text, refs)
pick = requested_card or wanted_pick
param_keys = wanted_param_keys(text, is_video=context.is_video)
if pick:
fallback_fields = [{
"key": pick,
"label": _ASSET_PICK_LABEL.get(pick, _ASSET_CARD_LABELS.get(pick, "请选择素材")),
"type": "asset",
"required": True,
"asset_types": [pick],
}]
allow_pick = bool(requested_card)
elif param_keys:
fallback_fields = session_param_fields(param_keys, context.is_video)
elif allow_plan and not said:
# 部分推理模型会只给 reasoning_content 后结束本轮。不能让用户
# 只看到自己那条消息;没有商品上下文时,收敛成一条自然追问。
has_product = any(
isinstance(ref, dict) and ref.get("type") == "product"
for ref in (conversation.pinned_refs or [])
)
if not has_product:
fallback_fields = [{
"key": "product",
"label": "这条想推哪款商品?",
"type": "asset",
"required": True,
"asset_types": ["product"],
}]
# ask_user 自己会落一条可追踪的聊天问题。模型同时吐出的过渡文案不再
# 另存一条,否则界面会连续出现两遍几乎相同的问题。
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)
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
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
messages.append({
"role": "tool",
"tool_call_id": f"call_{index}",
"content": json.dumps(result.get("payload", {}), ensure_ascii=False),
})
stop = stop or stop_after
if stop:
break
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
# 用户已终止时 API 侧多半已 idle+放锁;这里再收一次保证幂等,且不落入 awaiting_user
finish_agent_planning(
conversation,
awaiting_user=(awaiting_user and not user_cancelled),
)
return conversation_id
_TOOL_LABELS = {
"ask_user": "向你确认",
"search_library": "查找素材",
"generate_image": "生成图片",
"write_strategy": "梳理创作策略",
"write_plan": "编排视频方案",
}
_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"])
gate_field = {
"key": "_asset_gate",
"label": gate_label,
"type": "text",
"required": False,
}
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,
)
# 反问一旦发出就必须停,等人回答。继续跑等于自问自答。
return {
"payload": {"asked": True},
"_events": [{"type": "message", "message": _message_payload(message)}],
}, True
if name == "search_library":
return {"payload": _run_search_library(context, args)}, False
if name == "write_strategy":
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,
)
# 策略卡只是「我理解对了吗」,不打断 —— 模型接着就该写方案
return {
"payload": {"written": True},
"_events": [{"type": "message", "message": _message_payload(message)}],
}, False
if name == "write_plan":
video_prompt = str(args.get("video_prompt") or "").strip()
if not video_prompt:
return {"payload": {"error": "video_prompt 不能为空,请把完整出片指令写进去"}}, False
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
duration = 15
try:
duration = int(float(str((context.conversation.params or {}).get("duration") or "15").replace("秒", "").strip() or "15"))
except (TypeError, ValueError):
duration = 15
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)})
prompt_file = append_message(
context.conversation, role="assistant",
kind=CreationMessage.Kind.PROMPT_FILE,
payload={"title": "视频生成Prompt.md", "body": video_prompt,
"ref_count": plan_payload["ref_count"]},
)
events.append({"type": "message", "message": _message_payload(prompt_file)})
credits = estimate_video_credits(context)
# video_prompt 存在确认卡里:用户点确认后直接照它出片,不再跑一轮模型
confirm = append_message(
context.conversation, role="assistant",
kind=CreationMessage.Kind.CONFIRM,
payload={
"kind": "video",
"label": "开始生成",
"estimated_credits": credits,
"video_prompt": video_prompt,
"submitted": False,
"params": snapshot_session_params(context.conversation),
"param_options": confirm_param_options(True),
},
)
events.append({"type": "message", "message": _message_payload(confirm)})
events.append({"type": "credits", "estimated": credits})
# 方案卡写着「仅需确认一次」—— 停在这里等人点,别自己往下出片
return {"payload": {"awaiting_confirmation": True}, "_events": events}, True
if name == "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
# 出图也走确认卡:用户先看当前模型/比例/张数,点了才提交。
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