"""全能创作 · Agent 编排循环(契约 §3/§4)。 和 script_agent.py 的根本区别:那边是「单次结构化出稿」,模型只会写脚本; 这边是**真 function calling 循环** —— 模型自己决定这一轮该反问用户、该查素材, 还是该出图/出片。 铁律(踩过就回不来的三条): 1. **视频 5–10 分钟,绝不在 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 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 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}场景")), ) 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 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": "改成什么比例？", "resolution": "改成什么分辨率？", "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[:3]: 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 # ---------------------------------------------------------------- 工具 schema def tool_schemas(context: AgentContext) -> list[dict]: """给模型看的工具清单。图片会话不暴露 generate_video,反之亦然 —— 会话 mode 是定死的(契约 §0),把不该用的工具摆出来只会诱导模型走错路。""" tools = [ { "type": "function", "function": { "name": "ask_user", "description": ( "缺少必要信息时反问用户。会在对话里渲染成可点选/可填写的卡片。" "只在信息**确实缺失且无法合理推断**时用;能自己定的就自己定,别把用户当填表机器。" "用户要选/改/换商品、角色、模特、场景时**必须**调这个工具," "type 用 asset 并填 asset_types;禁止只在气泡里问「换成哪个」。" "一次最多问 3 项。" ), "parameters": { "type": "object", "properties": { "fields": { "type": "array", "maxItems": 3, "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 context.is_video: tools.append({ "type": "function", "function": { "name": "write_strategy", "description": ( "写「创作策略理解」卡:说清这条片给谁看、他为什么会信、你想让他信什么、整体创作方向。" "在动手写方案之前调它一次,让用户先确认你理解对了。" ), "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 出片,你不会再有插话机会 —— 所以 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"], }, }, "matrix": { "type": "object", "description": "卖点覆盖矩阵:哪个卖点落在第几个镜头", "properties": { "shots": {"type": "integer"}, "rows": { "type": "array", "items": { "type": "object", "properties": { "point": {"type": "string"}, "hits": {"type": "array", "items": {"type": "integer"}}, }, "required": ["point", "hits"], }, }, }, }, "voice_chars": { "type": "array", "items": {"type": "integer"}, "description": "口播字数区间 [下限, 上限]", }, "video_prompt": { "type": "string", "description": ( "交给出片模型的完整指令:秒级分镜(每镜画面/动作/机位/光线)、口播原文、" "风格锚点、一致性要求。已 @ 的素材会自动作为参考图附上," "不要在这里重复描述它们的长相。**不要写字幕相关要求。**" ), }, }, "required": ["usp", "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 [])[:3]: 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 video_model_name(params: dict) -> str: """会话参数里的模型 label → 火山模型名。认不出就回落 2.5(最长 30 秒那档)。""" return VIDEO_MODEL_BY_LABEL.get(str(params.get("model") or ""), 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_BY_LABEL.get(str(params.get("model") or ""), params.get("model") or None), reference_image_ids=reference_image_ids or None, ) 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 []) 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 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 build_system_prompt(context: AgentContext) -> str: conversation = context.conversation params = conversation.params or {} kind = "视频" if context.is_video else "图片" lines = [ "你是影擎「全能创作」的创作 agent,帮电商商家做短视频和商品图。", f"本次会话产出的是**{kind}**,这一点在整个会话里不会改变 —— 用户要另一种就请他新开一个创作。", "", "【怎么说话】", "- 说人话,像个有经验的同事,不要写成客服话术或需求确认清单。", "- 不要复述用户刚说过的话,不要说「好的,我明白了」这种空句。", "- 一次只推进一步。", "", "【什么时候反问】", "- 只有信息**确实缺失且无法合理推断**时才调用 ask_user;能自己定的就自己定。", "- 商品是谁、给谁看、什么调性 —— 这些缺了会直接影响成片,值得问。", "- 让用户选商品/角色/模特/场景时,ask_user 必须用 type=asset 并填 asset_types。", "- 用户说「改商品」「换角色」却没点名是哪个:立刻 ask_user 出点选卡,禁止只在气泡里问「换成哪个、填个名字」。用户不必 @。", "- 用户说改时长/模型/比例/分辨率:立刻 ask_user,type=single 给出选项;选完后旧方案作废,必须按新参数重新 write_plan。不要让用户去点底部菜单。", "- 光线、构图、镜头这些专业判断是你的活,不要反过来问用户。", ] if not context.is_video: lines.extend([ "", "【出图】", "- 决定出图时必须调用 generate_image,不要只口头说「我这就出图」。", "- 一次用户消息只出一轮;张数用会话已定参数,不要自己加张。", ]) 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( "以上素材的参考图会自动附给生成模型锁人锁物,你在 prompt 里不需要重复描述它们的外观。" ) 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 build_messages(context: AgentContext) -> list[dict]: """会话历史 → 模型消息。只喂对模型有意义的:文字、追问和它的答案、生成过什么。 策略卡/方案卡这类结构化产物压成一句话,原样塞 JSON 只会挤爆上下文。 长会话只喂最近 KEEP_RECENT_MESSAGES 条原文,更早的靠 system 里的【前情提要】 ——摘要在 compress_memory() 里生成,不在这里现算。 """ messages = [{"role": "system", "content": build_system_prompt(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: answers = (message.payload or {}).get("answers") or {} labels = {f["key"]: f["label"] for f in (message.payload or {}).get("fields", [])} if answers: joined = ";".join(f"{labels.get(k, k)} → {v}" for k, v in answers.items()) messages.append({"role": "assistant", "content": f"(我问了用户几个问题)"}) messages.append({"role": "user", "content": f"(用户回答){joined}"}) 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})"}) 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: try: parsed = json.loads(raw or "{}") except ValueError: return {} return parsed if isinstance(parsed, dict) else {} def stream_creation_agent( *, conversation: CreationConversation, user, text: str, refs: list[dict] | None = None, model_config: ModelConfig | None = None, ) -> Iterator[str]: """一条用户消息 → SSE 流。生成器,由 StreamingHttpResponse 逐帧下发。""" refs = refs or [] model_config = model_config or get_default_model(ModelConfig.Capability.TEXT) if model_config is None: yield _sse({"type": "error", "detail": "没有可用的文本模型,请先在模型库配置"}) return context = AgentContext(conversation=conversation, user=user, model_config=model_config) try: with transaction.atomic(): pin_refs(conversation, refs) user_message = append_message(conversation, role="user", text=text, refs=refs) yield _sse({"type": "message", "message": _message_payload(user_message)}) 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 _sse({"type": "message", "message": _message_payload(note)}) # 压缩放在**建消息之前**:摘要要进这一轮的 system 提示词才有意义。 # 用户消息已经先回显了,所以这一小段等待不会看起来像卡住。 compress_memory(context) provider = build_provider(model_config) messages = build_messages(context) tools = tool_schemas(context) for _round in range(MAX_TOOL_ROUNDS): 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 _sse({"type": "reasoning", "text": chunk.get("text", "")}) elif kind == "delta": piece = chunk.get("text", "") text_buffer.append(piece) yield _sse({"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")] if said: bubble = append_message(conversation, role="assistant", text=said) yield _sse({"type": "message", "message": _message_payload(bubble)}) if not calls: pick = wanted_asset_pick(text, refs) param_keys = wanted_param_keys(text, is_video=context.is_video) fields = None if pick: fields = [{ "key": pick, "label": _ASSET_PICK_LABEL[pick], "type": "asset", "required": True, "asset_types": [pick], }] elif param_keys: fields = session_param_fields(param_keys, context.is_video) if fields: result, _stop = _dispatch_tool(context, "ask_user", {"fields": fields}) for event in result.get("_events", []): yield _sse(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): name = call["name"] args = _parse_arguments(call["arguments"]) yield _sse({"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 _sse({"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 _sse({"type": "message", "message": _message_payload(failure)}) yield _sse({"type": "done"}) return yield _sse({"type": "tool", "id": name, "label": _TOOL_LABELS.get(name, name), "status": "done"}) for event in result.get("_events", []): yield _sse(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 _sse({"type": "done"}) except Exception as exc: # noqa: BLE001 — SSE 里任何未捕获异常都会变成前端「白屏卡死」 logger.exception("creation agent stream failed: %s", exc) yield _sse({"type": "error", "detail": "生成过程出错了,请再试一次"}) _TOOL_LABELS = { "ask_user": "向你确认", "search_library": "查找素材", "generate_image": "生成图片", "write_strategy": "梳理创作策略", "write_plan": "编排视频方案", } def _dispatch_tool(context: AgentContext, name: str, args: dict) -> tuple[dict, bool]: """执行一个工具。返回 (结果, 是否中断循环)。 结果里的 `_events` 会原样转发给前端,`payload` 回喂给模型。 """ if name == "ask_user": fields = _coerce_fields(args.get("fields")) if not fields: return {"payload": {"error": "fields 不合法,请重新组织问题"}}, False message = append_message( context.conversation, role="assistant", kind=CreationMessage.Kind.ELICIT, payload={"fields": fields, "submitted": False, "answers": {}}, ) # 反问一旦发出就必须停,等人回答。继续跑等于自问自答。 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": message = append_message( context.conversation, role="assistant", kind=CreationMessage.Kind.STRATEGY, payload={key: str(args.get(key) or "") for key in ("target", "trust", "belief", "direction")}, ) # 策略卡只是「我理解对了吗」,不打断 —— 模型接着就该写方案 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 plan_payload = { "usp": str(args.get("usp") or ""), "points": [str(p) for p in (args.get("points") or [])][:3], "timeline": args.get("timeline") or [], "matrix": args.get("matrix") or {}, "voice_chars": args.get("voice_chars") or [], "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={"label": "开始生成", "estimated_credits": credits, "video_prompt": video_prompt, "submitted": False}, ) 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 payload, tasks = _run_generate_image(context, args) events = [] for task in tasks: message = append_message( context.conversation, role="assistant", kind=CreationMessage.Kind.GENERATING, payload={"task_id": str(task.id), "kind": "image", "prompt": args.get("prompt", "")}, task=task, ) events.append({"type": "message", "message": _message_payload(message)}) events.append({"type": "task", "task_id": str(task.id), "kind": "image"}) _remember_artifact(context.conversation, args.get("prompt", ""), "image") return {"payload": payload, "_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