From 380896f8a364592b6132827e8e518b4399cd60e0 Mon Sep 17 00:00:00 2001 From: johnnyzhang-eng Date: Fri, 14 Aug 2026 17:51:17 +0800 Subject: [PATCH] =?UTF-8?q?feat(prompt):=20=E7=94=A8=E6=88=B7=E6=8F=8F?= =?UTF-8?q?=E8=BF=B0=E5=85=88=E8=BF=87=E6=8E=AA=E8=BE=9E=E9=97=A8=E7=A6=81?= =?UTF-8?q?=E4=B8=8E=E9=80=82=E9=85=8D=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/windup_ai_engine/ports/__init__.py | 105 ++++- .../src/windup_ai_engine/prompt/adapter.py | 119 ++++++ .../src/windup_ai_engine/prompt/custom.py | 41 +- .../src/windup_ai_engine/prompt/lint.py | 182 +++++++++ .../src/windup_ai_engine/strategy/concrete.py | 57 ++- .../src/windup_common/models/__init__.py | 2 + .../src/windup_common/models/character.py | 25 +- backend/tests/test_character_contract.py | 21 + backend/tests/test_prompt_adapter.py | 384 ++++++++++++++++++ 9 files changed, 901 insertions(+), 35 deletions(-) create mode 100644 backend/packages/ai_engine/src/windup_ai_engine/prompt/adapter.py create mode 100644 backend/packages/ai_engine/src/windup_ai_engine/prompt/lint.py create mode 100644 backend/tests/test_prompt_adapter.py diff --git a/backend/packages/ai_engine/src/windup_ai_engine/ports/__init__.py b/backend/packages/ai_engine/src/windup_ai_engine/ports/__init__.py index c3eb1684..4fd23fd2 100644 --- a/backend/packages/ai_engine/src/windup_ai_engine/ports/__init__.py +++ b/backend/packages/ai_engine/src/windup_ai_engine/ports/__init__.py @@ -13,7 +13,9 @@ from enum import Enum from typing import Protocol, runtime_checkable -from windup_common.models import ActionSpec, CharacterCard +from windup_common.models import ActionSpec, CharacterCard, CharacterStance, Facing + +from windup_ai_engine.prompt.lint import Kind, LintIssue # ---- server 实现、注入给 ai_engine 的进度回调 port ---- @@ -41,8 +43,9 @@ class MasterRejected(ValueError): """母版不具备可生成性,在**调用付费模型之前**拒绝。 与 ai_engine 其他异常的分工(这条分工是给 server 用的): - - ``MasterRejected`` = **调用方的输入不行**,同一张母版重试多少次都一样。 - server 应映射成 4xx、把 ``code`` 翻成"请换一张母版"类文案,**不要重试**。 + - ``MasterRejected`` / ``PromptRejected`` = **调用方的输入不行**,同一份输入 + 重试多少次都一样。server 应映射成 4xx、按 ``code`` 选"换一张母版" / + "改一下这句描述"的文案,**不要重试**。 - ``NotImplementedError`` / 其他 ``ValueError`` = 引擎侧装配或产出出了问题 (路线没注入、strategy 吐空帧、帧数对不上),属于 5xx、要人介入, 让用户换母版是把锅甩错地方。 @@ -54,6 +57,88 @@ def __init__(self, code: MasterRejectCode, detail: str) -> None: self.detail = detail +class PromptRejectCode(str, Enum): + """用户那句动作描述被拒的原因 —— server 据此选文案,别 parse 异常消息做分支。 + + 每个取值对应一条**模型侧的机制**(见 :mod:`windup_ai_engine.prompt.lint`), + 不是文风偏好;判定全部本地零成本,发生在付费调用之前。 + """ + + EMPTY = "empty" # 没写动作:模型照跑,回来一段站着不动的视频 + TOO_LONG = "too_long" # 越长越容易夹带外观,而外观由母版承载 + NEGATION = "negation" # 无 negative_prompt,"不要 X"把 X 送进画面 + HAZARD_NOUN = "hazard_noun" # 特效名词盖住轮廓,抠图留脏边 + SHAPE_PRIOR = "shape_prior" # 断言母版里没有的装备形状,焊到角色身上 + SUBTHRESHOLD = "subthreshold" # 幅度低于模型可控分辨率 → 逐帧随机抖 + UNANCHORED_PROP = "unanchored_prop" # 没交代身体整体怎么动,手里的东西自行漂移 + MULTI_STAGE = "multi_stage" # 静态模型没有时间轴,多阶段摊成分解姿势图 + STANCE_MISMATCH = "stance_mismatch" # 非双足角色写人体部位 → 凭空长出人的上肢 + + +class PromptRejected(ValueError): + """这段描述送进模型必然出坏产物,在**调用付费模型之前**拒绝。 + + 形状与 :class:`MasterRejected` 一致、分工同一条:它是**调用方输入不行**那一类, + server 映射 4xx 让用户改那句话,而不是 5xx 报"系统出问题了"——用户改得动的东西 + 被报成服务器故障,他只会重试同一句话。 + + 多条机制同时命中时 ``code`` 取报告序里的第一条,``detail`` 仍把每条都列出来: + 只讲一条会让用户改完再被下一条拦一次。 + """ + + def __init__(self, code: PromptRejectCode, detail: str) -> None: + super().__init__(f"这段描述跑不出可用产物({code.value}):{detail}") + self.code = code + self.detail = detail + + +# ---- 用户大白话 → 正式提示词(实现在别处,见下)---- +@dataclass(frozen=True) +class AdaptedPrompt: + """一次**成功**适配的结果 —— 拿到它就等于可以往下送。 + + 不可适配走 :class:`PromptRejected`,不在这里留一个"拒了"的字段:同一件事两条返回 + 路径,调用方得写两套处理,而漏写返回值那条是静默的 —— 空文本照样进付费调用。 + """ + + text: str + """正式提示词。 + + ``kind="i2v"`` 时它**不含**循环性尾句:循环与否是请求的属性(``ActionSpec.cyclic``), + 适配器的入参里没有,替调用方猜一条会把一次性动作首尾闭环,而帧数 / 时长 / 成色全正常。 + 调用方按自己声明的循环性追加 ``prompt.custom`` 的两条尾句之一。 + """ + + issues: tuple[LintIssue, ...] = () + """确定性改写做不到、但不足以拦下的问题(warn 级)。error 级都走拒绝,不会到这里。""" + + +class PromptAdapterPort(Protocol): + """把用户那句大白话改写进已验证的骨架。 + + 引擎侧只定协议:确定性规则之外的改写要调模型,那属于 provider 那一层。 + + Args: + user_text: 用户自述的动作,只讲做什么动作。 + kind: 目标模型类型 —— 决定哪些规则成立(见 ``prompt.lint`` 的 ``kind``)。 + facing: 母版朝向。**必须与母版一致**。 + stance: 角色体型(``CharacterCard.stance``)。非双足时"手臂"一类词会让模型 + 凭空长出人的上肢,故它参与判定,不只是记录。 + + Raises: + PromptRejected: 这段描述送进模型必然出坏产物 → 4xx,让用户改这句话。 + """ + + def adapt( + self, + user_text: str, + *, + kind: Kind, + facing: Facing, + stance: CharacterStance, + ) -> AdaptedPrompt: ... + + # ---- ai_engine 出参(不含存储引用:上传 / 落库在 server 侧)---- @dataclass(frozen=True) class ActionQuality: @@ -132,12 +217,12 @@ class CharacterGeneratorPort(Protocol): 不关心租户 / 配额 / 任务状态 / 存储(那些在 app.server)。 Args: - card: 角色卡。**当前唯一实现的视频路线一个字段都不读**——``git grep 'card\\.'`` - 在 ai_engine 下零命中(2026-08-08 复核)。这不是遗漏:i2v 的角色身份完全由 - ``master`` 这张母版图像承载,身份描述再写一遍反而会和母版打架。本参数是给 - 未实现路线预留的入参:逐帧图生图(#53)要靠 ``name`` / ``desc`` 在每帧提示词里 - 锁一致性,渲染出帧(#81 #122)要靠 ``master_ref`` / ``version`` 定位 3D 资产。 - **调用方不要指望改 card 能影响视频路线的产出。** + card: 角色卡。视频路线**只读 ``stance`` 一个字段**,且它不进提示词、只决定用户那句 + 描述里的人体部位词(手臂 / 手肘)放不放行 —— 非双足角色放行了,模型会给它接上 + 一对人的上肢。**角色身份不读 card**:i2v 的一致性完全由 ``master`` 这张母版图像 + 承载,身份描述再写一遍反而会和母版打架。其余字段是给未实现路线预留的:逐帧图生图 + (#53)要靠 ``name`` / ``desc`` 在每帧提示词里锁一致性,渲染出帧(#81 #122)要靠 + ``master_ref`` / ``version`` 定位 3D 资产。**改 name / desc 影响不了视频路线的产出。** action: 动作规格(类型 / 帧数 / 风格化 / 朝向)。视频路线的实际入参在这里: ``action``、``n_frames``、``facing``、``stylize`` 等。 master: 定妆母版图 bytes(server 从 reference_image_url 取)。**视频路线的 @@ -154,6 +239,8 @@ class CharacterGeneratorPort(Protocol): Raises: MasterRejected: 母版形态不可生成(见 :class:`MasterRejectCode`)。**在花钱 之前抛**,同一张母版重试无意义 → server 映射 4xx、请用户换母版。 + PromptRejected: ``action=custom`` 时用户那句描述必然出坏产物(见 + :class:`PromptRejectCode`)。同样在花钱之前抛 → 4xx、请用户改那句话。 NotImplementedError: 该动作分流到的路线没有实现或没注入 strategy。 ValueError: 产出对不上契约(空帧 / 帧数不足)。钱已经花了,但错产物不放行。 diff --git a/backend/packages/ai_engine/src/windup_ai_engine/prompt/adapter.py b/backend/packages/ai_engine/src/windup_ai_engine/prompt/adapter.py new file mode 100644 index 00000000..dc775457 --- /dev/null +++ b/backend/packages/ai_engine/src/windup_ai_engine/prompt/adapter.py @@ -0,0 +1,119 @@ +"""零模型的 :class:`~windup_ai_engine.ports.PromptAdapterPort` 实现。 + +先做规则版而不是直接上 LLM:它不花钱、确定性、可测,且换成 LLM 版之后它仍然是兜底 +(模型不可用时的降级)与对照组(判断 LLM 改写到底有没有比规则更好)。 + +它只做确定性做得到的三件事:跑门禁并在 error 级上拒掉、把用户那句话嵌进已验证的骨架、 +追加统一的单主体与构图后缀。**翻译、改写措辞、把"轻微"换成一个具体幅度,规则做不到** +—— 那些是 LLM 版的活,这里只负责讲清楚拦在哪、为什么。 + +放在 ai_engine 而不是 framework:分层门禁(``lint-imports`` 的"包分层链")规定 +framework 在 ai_engine 之下,framework 里的模块 import 不到本层的门禁与骨架。 +将来的 LLM 版同样住这一层,按 ``VideoFrameStrategy`` 与 ``VideoProvider`` 的成例, +把模型调用作为 framework 的 provider 注入进来。 +""" +from __future__ import annotations + +from windup_common.models import CharacterStance, Facing + +from windup_ai_engine.ports import AdaptedPrompt, PromptRejectCode, PromptRejected +from windup_ai_engine.prompt.custom import MAX_ACTION_CHARS, build_custom_body +from windup_ai_engine.prompt.lint import Kind, lint + +__all__ = ["RuleBasedPromptAdapter"] + +# 统一后缀。全是正向措辞:这条通路没有 negative_prompt,"背景里没有别人"会把别人请进来。 +_COMPOSITION = ( + "One single character alone in the frame, the whole body inside the frame, " + "on one plain flat background." +) + +# 静态模型没有时间轴,一段多阶段描述会被摊平成并排的分解姿势图 —— 一张图里好几个身位, +# 而它对切片来说是废的。故给静态模型的必须是单一瞬间。 +_SINGLE_INSTANT = "ONE single frozen instant of that motion, one single pose." + +_STAGE_MARKERS = ( + "then", "after that", "afterwards", "followed by", "next,", "and finally", + "然后", "接着", "紧接着", "之后", "再", "最后", "先", "收势", +) +_ARM_WORDS = ("arm", "arms", "elbow", "hand", "hands", "手臂", "胳膊", "手肘") + +# 每个非双足体型自带一套可替换的部位说法:拒绝理由要给得出改法,"这个词不行"给不了。 +# 缺一支就是拒了却说不出改哪儿,故 :class:`CharacterStance` 加成员必须同时加这里。 +_STANCE_PARTS = { + CharacterStance.QUADRUPED: "前肢 / 头颈 / 尾", + CharacterStance.SERPENTINE: "躯干起伏 / 尾 / 头颈", +} + +# 门禁类别 → 拒绝码。直查不 get:漏配一条是引擎侧的装配缺口(该 5xx 让人介入), +# 兜个通用码会把它伪装成用户的输入问题,而用户按那条文案改多少遍都过不了。 +_CODE_BY_CATEGORY = { + "negation": PromptRejectCode.NEGATION, + "hazard_noun": PromptRejectCode.HAZARD_NOUN, + "shape_prior": PromptRejectCode.SHAPE_PRIOR, + "subthreshold": PromptRejectCode.SUBTHRESHOLD, + "unanchored_prop": PromptRejectCode.UNANCHORED_PROP, +} + + +class RuleBasedPromptAdapter: + """确定性适配:能判的当场判,判不了的照原样嵌进骨架。""" + + def adapt( + self, + user_text: str, + *, + kind: Kind = "i2v", + facing: Facing = Facing.SIDE, + stance: CharacterStance | str = CharacterStance.BIPED, + ) -> AdaptedPrompt: + """Raises ``PromptRejected``:这段描述送进模型必然出坏产物,理由带 code 与机制。""" + stance = CharacterStance(stance) # 非法体型要炸,不静默按双足放行 + clause = (user_text or "").strip() + if not clause: + raise PromptRejected( + PromptRejectCode.EMPTY, + "没写动作内容。空描述不会报错,只会拿回一段站着不动的视频," + "而帧数和时长全对、看不出描述丢了。", + ) + + if len(clause) > MAX_ACTION_CHARS: + raise PromptRejected( + PromptRejectCode.TOO_LONG, + f"描述有 {len(clause)} 字,超过上限 {MAX_ACTION_CHARS}。描述越长越容易" + f"夹带角色外观,而外观由母版承载,写两遍会打架。只留动作本身。", + ) + + issues = lint(clause, kind=kind) + blockers = [ + (_CODE_BY_CATEGORY[i.category], i.message) for i in issues if i.level == "error" + ] + low = clause.lower() + + if kind == "still": + marker = next((m for m in _STAGE_MARKERS if m in low), None) + if marker: + blockers.append(( + PromptRejectCode.MULTI_STAGE, + f"「{marker}」把这段描述分成了好几个阶段,而静态模型没有时间轴:" + f"它会把各阶段并排画成一张分解姿势图,一张图里好几个身位。" + f"只描述其中一个瞬间。", + )) + + if stance is not CharacterStance.BIPED: + hit = next((w for w in _ARM_WORDS if w in low), None) + if hit: + blockers.append(( + PromptRejectCode.STANCE_MISMATCH, + f"这个角色的体型是 {stance.value},不是双足,而「{hit}」会让模型给它凭空" + f"接上人的上肢。改写成发力的那个部位({_STANCE_PARTS[stance]})。", + )) + + if blockers: + raise PromptRejected( + blockers[0][0], "\n".join(f"· {m}" for _, m in blockers) + ) + + body = build_custom_body(clause, facing=facing) + parts = [body, _SINGLE_INSTANT, _COMPOSITION] if kind == "still" else [body, _COMPOSITION] + return AdaptedPrompt(text=" ".join(parts), issues=tuple(issues)) diff --git a/backend/packages/ai_engine/src/windup_ai_engine/prompt/custom.py b/backend/packages/ai_engine/src/windup_ai_engine/prompt/custom.py index 5e6b937a..c97f8fad 100644 --- a/backend/packages/ai_engine/src/windup_ai_engine/prompt/custom.py +++ b/backend/packages/ai_engine/src/windup_ai_engine/prompt/custom.py @@ -9,7 +9,13 @@ from windup_common.models import Facing -__all__ = ["build_custom_prompt", "MAX_ACTION_CHARS"] +__all__ = [ + "build_custom_prompt", + "build_custom_body", + "CYCLIC_TAIL", + "ONESHOT_TAIL", + "MAX_ACTION_CHARS", +] # 不是接口限制,是产品判断:描述越长越容易夹带角色外观,而外观由母版承载,写两遍会打架。 MAX_ACTION_CHARS = 200 @@ -33,16 +39,34 @@ # 两条尾句都刻意不写"在地面上 / 双脚可见 / 回到直立站姿"——那些是着地直立类动作的前提, # 游泳、飞行、攀爬、倒地都不成立,而文字与动作矛盾时模型会自己找辙调和。 -_CYCLIC_TAIL = ( +# 两条尾句公开:循环性是**请求**的属性(ActionSpec.cyclic),不是文本的属性,所以任何 +# 只拿到文本的组件(如 prompt.adapter)都不该替调用方选一条 —— 猜错是静默的。 +CYCLIC_TAIL = ( "The motion is one smooth repeating cycle that returns to its starting pose, " "and the character stays centered in the same spot in frame." ) -_ONESHOT_TAIL = ( +ONESHOT_TAIL = ( "The character performs this ONCE as one single committed motion, " "then holds the final pose and stays still." ) +def build_custom_body(action: str, *, facing: Facing | str = Facing.SIDE) -> str: + """朝向锁 + 用户那句话 + 装备存在无关句,**不含**循环性尾句。 + + Raises: + ValueError: 理由同 :func:`build_custom_prompt`。 + """ + text = (action or "").strip() + if not text: + raise ValueError("自定义动作的描述不能为空") + if len(text) > MAX_ACTION_CHARS: + raise ValueError(f"自定义动作描述 {len(text)} 字,超过上限 {MAX_ACTION_CHARS}") + lock = _FACING_LOCK[Facing(facing)] # 非法朝向要炸,不静默落到某一支 + # 朝向放最前:最强的约束先钉。 + return f"The character {lock}: {text}, {_KEEP_WHAT_IT_HAS}." + + def build_custom_prompt( action: str, *, @@ -60,12 +84,5 @@ def build_custom_prompt( ValueError: 描述为空或超长。空描述不兜底默认动作——那会付一次 i2v 的钱拿到一段 站着不动的视频,而帧数时长全对、看不出描述丢了。 """ - text = (action or "").strip() - if not text: - raise ValueError("自定义动作的描述不能为空") - if len(text) > MAX_ACTION_CHARS: - raise ValueError(f"自定义动作描述 {len(text)} 字,超过上限 {MAX_ACTION_CHARS}") - lock = _FACING_LOCK[Facing(facing)] # 非法朝向要炸,不静默落到某一支 - tail = _CYCLIC_TAIL if cyclic else _ONESHOT_TAIL - # 朝向放最前:最强的约束先钉。 - return f"The character {lock}: {text}, {_KEEP_WHAT_IT_HAS}. {tail}" + tail = CYCLIC_TAIL if cyclic else ONESHOT_TAIL + return f"{build_custom_body(action, facing=facing)} {tail}" diff --git a/backend/packages/ai_engine/src/windup_ai_engine/prompt/lint.py b/backend/packages/ai_engine/src/windup_ai_engine/prompt/lint.py new file mode 100644 index 00000000..e3dd69a8 --- /dev/null +++ b/backend/packages/ai_engine/src/windup_ai_engine/prompt/lint.py @@ -0,0 +1,182 @@ +"""提示词措辞门禁 —— 纯函数,只依赖标准库。 + +这里每条规则约束的都是**模型侧的机制**,不是团队的文风偏好;报错信息也按机制写, +因为收到它的人要据此改自己的描述,"命中禁词"这种话给不了任何改法。 + +规则与机制: + - 否定式:该 i2v 接口没有 negative_prompt,模型不处理否定极性、只 latch 到名词上, + 于是"不要 X"把 X 送进画面。任何否定都要改成正面描述。 + - 特效名词(烟尘/火花/火焰…):提到即被勾成画面里的效果,它盖住角色轮廓, + 抠图与像素化会在边缘留脏。 + - 冲击词:不直接生成特效,但强相关,常连带勾出烟尘碎屑 —— 故只降级提示。 + - 亚阈值微动:要求的位移小于模型可控的分辨率,得到的不是小幅动作而是逐帧随机抖动。 + - 持物无身体锚:i2v 强跟身体、弱跟手持物;身体整体怎么动没被指定时,持物自行漂移。 + - 装备形状先验:母版是角色身份的唯一来源。断言刃面朝向 / 弧线 / 前手这类形状, + 等于把母版里不存在的形状焊到角色身上,并跟着该角色去到所有动作。 + 装备名词只降级提示 —— 用户看着自己的母版写"剑"可能属实,而形状先验用户无从判断。 + +``kind`` 的适用范围按机制划,不按方便:亚阈值微动与持物锚定都靠"帧与帧之间"才成立, +静态图没有帧间(前者退成弱指令、后者整条不成立,单件道具素材图里根本没有身体)。 +""" +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Literal + +__all__ = ["Kind", "Level", "LintIssue", "lint"] + +Kind = Literal["i2v", "still"] +Level = Literal["error", "warn"] + + +@dataclass(frozen=True) +class LintIssue: + """一条命中。``term`` 给的是原文里真实出现的那段字,便于调用方定位。""" + + level: Level + category: str + term: str + message: str + + +# 汉字没有词边界,ASCII 才加 \b;分开存是为了让两类词用各自正确的匹配方式, +# 而不是让中文词退化成"整词匹配"从而永不命中。 +_NEGATION = ( + "no", "not", "never", "without", "avoid", "avoids", "avoiding", + "don't", "doesn't", "isn't", "aren't", "won't", "cannot", + "不要", "不能", "不许", "禁止", "避免", "勿", +) +_HAZARD = ( + "dust", "dusty", "smoke", "smoky", "smoking", "spark", "sparks", "sparking", + "debris", "flame", "flames", "flaming", "fire", "fires", "fiery", + "ember", "embers", "dirt", "haze", "hazy", + "烟雾", "扬尘", "火花", "灰尘", "火焰", "碎屑", +) +_IMPACT = ( + "impact", "impacts", "strike", "strikes", "striking", "struck", + "hammer", "hammers", "hammering", "slam", "slams", "slamming", + "smash", "smashes", "smashing", "explosive", "explosion", + "burst", "bursts", "bursting", "slash", "slashes", "slashing", + "冲击", "命中", "劈砍", +) +_SUBTHRESHOLD = ( + "slightly", "slight", "subtle", "subtly", "tiny", "barely", "faint", + "faintly", "micro", "minuscule", + "轻微", "细微", "几个像素", "一点点", "微微", +) +# 持物动作词。刻意不收"挥/swing":它们同样用于空手动作(挥手),而 2b 的机制只在 +# 手里真有东西时成立,收进来会把"挥手"判成漂移风险。 +_PROP_ACTION = ( + "hold", "holds", "holding", "held", "grip", "grips", "gripping", + "wield", "wields", "wielding", "brandish", "brandishes", "brandishing", + "握", "持", "拿着", "举着", "挥剑", "挥刀", "挥舞", +) +# 只收"整个身体往哪儿去"的词。肩、胸这类上肢局部不算锚:它们与手持物一起漂, +# 定不住整体,收进来等于让 2b 形同虚设。 +_BODY_ANCHOR = ( + "whole body", "body", "torso", "hips", "waist", + "lunge", "lunges", "step", "steps", "stride", "strides", + "身体", "躯干", "重心", "整体", "腰", +) +_SHAPE_PRIOR = ( + "broad side", "broadside", "flat side", "flat of the blade", "blade side", + "crescent", "crescent arc", "leading arm", "leading hand", "leading edge", + "刃面", "新月", "前手", "前臂朝", +) +_EQUIPMENT = ( + "sword", "blade", "weapon", "staff", "wand", "axe", "spear", "shield", + "dagger", "bow", "gun", "rifle", + "剑", "刀", "武器", "盾", "法杖", "长矛", "弓", +) + +_ASCII_WORD = re.compile(r"^[a-z' ]+$") + + +def _compile(terms: tuple[str, ...]) -> re.Pattern[str]: + parts = [ + rf"\b{re.escape(t)}\b" if _ASCII_WORD.match(t) else re.escape(t) + for t in terms + ] + return re.compile("|".join(parts), re.IGNORECASE) + + +_PATTERNS = { + name: _compile(terms) + for name, terms in ( + ("negation", _NEGATION), ("hazard", _HAZARD), ("impact", _IMPACT), + ("subthreshold", _SUBTHRESHOLD), ("prop", _PROP_ACTION), + ("body", _BODY_ANCHOR), ("shape_prior", _SHAPE_PRIOR), + ("equipment", _EQUIPMENT), + ) +} + + +def _hits(text: str, name: str) -> list[str]: + """按出现顺序去重,免得同一个词在长句里刷出十条一模一样的问题。""" + seen: dict[str, None] = {} + for m in _PATTERNS[name].finditer(text): + seen.setdefault(m.group(0).lower(), None) + return list(seen) + + +def lint(text: str, *, kind: Kind = "i2v") -> list[LintIssue]: + """按目标模型类型查一段提示词。 + + ``kind`` 只影响 2a / 2b 两条:它们的机制都是帧与帧之间的,静态图没有帧间。 + """ + issues: list[LintIssue] = [] + + for term in _hits(text, "negation"): + issues.append(LintIssue( + "error", "negation", term, + f"「{term}」是否定式。这条通路没有 negative_prompt,模型不处理否定极性、" + f"只把名词 latch 进画面,写「不要 X」等于点名要 X。改成正面说你要的样子。", + )) + for term in _hits(text, "hazard"): + issues.append(LintIssue( + "error", "hazard_noun", term, + f"「{term}」是特效名词,提到即被勾成画面里的效果;它盖住角色轮廓," + f"后面的抠图和像素化会在边缘留下脏边。只写身体在做什么。", + )) + for term in _hits(text, "shape_prior"): + issues.append(LintIssue( + "error", "shape_prior", term, + f"「{term}」是装备的形状先验。角色身份的唯一来源是母版,提示词里断言刃面 / " + f"弧线 / 前手这类形状,会把母版里没有的形状焊到这个角色身上,并跟着它去到" + f"所有动作。只描述身体怎么发力。", + )) + for term in _hits(text, "impact"): + issues.append(LintIssue( + "warn", "impact_verb", term, + f"「{term}」是冲击词,模型常连带画出烟尘碎屑。换成具体的身体动作" + f"(哪个部位先动、往哪个方向发力)更稳。", + )) + for term in _hits(text, "equipment"): + issues.append(LintIssue( + "warn", "equipment_noun", term, + f"「{term}」在断言这个角色带着某件装备。带没带由母版说了算,写在提示词里" + f"与母版不一致时,模型会自己造一件出来。", + )) + + for term in _hits(text, "subthreshold"): + # 静态图没有帧间,抖不起来;但"轻微"对单张图同样给不出可执行的幅度,故仍报。 + level: Level = "error" if kind == "i2v" else "warn" + issues.append(LintIssue( + level, "subthreshold", term, + f"「{term}」要求的幅度低于模型可控的分辨率。" + + ("视频侧拿到的不是小幅动作,而是逐帧随机抖动。" + if kind == "i2v" else "静态图没有帧间抖动,但它仍是一条模型执行不了的弱指令。") + + "给一个看得见的幅度(动到哪儿、动多远)。", + )) + + if kind == "i2v": + props = _hits(text, "prop") + if props and not _hits(text, "body"): + issues.append(LintIssue( + "error", "unanchored_prop", props[0], + f"写了持物动作「{props[0]}」,却没交代身体整体怎么动。i2v 强跟身体、" + f"弱跟手持物:身体没被指定时,手里的东西会自己漂。补一句躯干 / 重心 / " + f"整体位移。", + )) + return issues diff --git a/backend/packages/ai_engine/src/windup_ai_engine/strategy/concrete.py b/backend/packages/ai_engine/src/windup_ai_engine/strategy/concrete.py index 4274ddbd..f1076b9b 100644 --- a/backend/packages/ai_engine/src/windup_ai_engine/strategy/concrete.py +++ b/backend/packages/ai_engine/src/windup_ai_engine/strategy/concrete.py @@ -12,13 +12,20 @@ import numpy as np -from windup_common.models import ActionSpec, ActionType, CharacterCard, GenRoute, Stylize +from windup_common.models import ( + ActionSpec, + ActionType, + CharacterCard, + CharacterStance, + GenRoute, + Stylize, +) from windup_framework.providers import ImageProvider, MatteProvider, VideoProvider from windup_ai_engine._imgio import from_png as _img from windup_ai_engine._imgio import to_png as _png from windup_ai_engine.master_prep import prepare_master -from windup_ai_engine.ports import ProgressPort +from windup_ai_engine.ports import ProgressPort, PromptAdapterPort, PromptRejected from windup_ai_engine.postprocess import master_pixel_spec, pixelate_frames from windup_ai_engine.slicing import extract_all_frames_bytes, pick_cycle, pick_oneshot from windup_ai_engine.prompt import ( @@ -28,6 +35,8 @@ build_jump_prompt, build_walk_prompt, ) +from windup_ai_engine.prompt.adapter import RuleBasedPromptAdapter +from windup_ai_engine.prompt.custom import CYCLIC_TAIL, ONESHOT_TAIL from windup_ai_engine.strategy.base import DerivationStrategy, is_cyclic @@ -41,21 +50,23 @@ class VideoFrameStrategy(DerivationStrategy): route = GenRoute.VIDEO_I2V - def __init__(self, video: VideoProvider, matte: MatteProvider) -> None: + def __init__( + self, + video: VideoProvider, + matte: MatteProvider, + adapter: PromptAdapterPort | None = None, + ) -> None: self._video = video self._matte = matte + self._adapter = RuleBasedPromptAdapter() if adapter is None else adapter - def _build_prompt(self, action: ActionSpec) -> str: + def _build_prompt(self, action: ActionSpec, stance: CharacterStance) -> str: """按动作类型选提示词;朝向随 ActionSpec.facing。""" - # custom 单独一支:它的动作内容来自用户,只能由 build_custom_prompt 把那句话 - # 嵌进机制骨架(朝向锁 / 正向措辞 / 装备存在无关 / 一次性的单次+终态保持)。 + # custom 单独一支:它的动作内容来自用户,只能由骨架把那句话嵌进机制约束 + # (朝向锁 / 正向措辞 / 装备存在无关 / 一次性的单次+终态保持)。 # 不能塞进下面那张表 —— 那张表里的 builder 只接 facing。 if action.action is ActionType.CUSTOM: - return build_custom_prompt( - action.custom_action or "", - facing=action.facing, - cyclic=bool(action.cyclic), - ) + return self._custom_prompt(action, stance) builders = { ActionType.JUMP: build_jump_prompt, ActionType.IDLE: build_idle_prompt, @@ -64,6 +75,28 @@ def _build_prompt(self, action: ActionSpec) -> str: build = builders.get(action.action, build_walk_prompt) return build(facing=action.facing) + def _custom_prompt(self, action: ActionSpec, stance: CharacterStance) -> str: + """用户那句话先过适配器,再按声明的循环性收尾。 + + Raises: + PromptRejected: 这段描述送进模型必然出坏产物 → server 映射 4xx 让用户改。 + """ + clause = action.custom_action or "" + cyclic = bool(action.cyclic) + try: + adapted = self._adapter.adapt( + clause, kind="i2v", facing=action.facing, stance=stance, + ) + except PromptRejected: + # 与下面那条分得很清:这不是组件不可用,是这段描述本身跑不出可用产物, + # 而下一步就是付费调用 —— 回退等于照样把钱花掉换一段废视频。 + # 顺序也是约束:它是 ValueError 的子类,放到宽兜底后面就永远轮不上。 + raise + except Exception: + # 适配器坏掉只该丢掉那层改写,不该把整条生成打死:骨架本身不依赖它。 + return build_custom_prompt(clause, facing=action.facing, cyclic=cyclic) + return f"{adapted.text} {CYCLIC_TAIL if cyclic else ONESHOT_TAIL}" + def derive( self, card: CharacterCard, @@ -80,7 +113,7 @@ def derive( progress.step("derive", 0, 3, f"{action.action.value}: i2v 生成视频") # 母版按动作预处理:jump 要在顶部补空间,否则角色腾空时头顶顶出视频画面被裁 framed = prepare_master(master, action.action.value) - video = self._video.i2v(framed, self._build_prompt(action), seconds=5) + video = self._video.i2v(framed, self._build_prompt(action, card.stance), seconds=5) dense = extract_all_frames_bytes(video) # 跨动作一致性:用视频首帧(=母版姿态)的角色高当共同定标基准。各动作都从同一母版 diff --git a/backend/packages/common/src/windup_common/models/__init__.py b/backend/packages/common/src/windup_common/models/__init__.py index 52434b34..d3d2b39a 100644 --- a/backend/packages/common/src/windup_common/models/__init__.py +++ b/backend/packages/common/src/windup_common/models/__init__.py @@ -3,6 +3,7 @@ ActionSpec, ActionType, CharacterCard, + CharacterStance, CharacterView, Facing, GenRoute, @@ -14,6 +15,7 @@ "GenRoute", "Facing", "CharacterView", + "CharacterStance", "Stylize", "DEFAULT_N_FRAMES", "CharacterCard", diff --git a/backend/packages/common/src/windup_common/models/character.py b/backend/packages/common/src/windup_common/models/character.py index 07cdbf3d..2655946f 100644 --- a/backend/packages/common/src/windup_common/models/character.py +++ b/backend/packages/common/src/windup_common/models/character.py @@ -58,6 +58,20 @@ class ActionType(str, Enum): CUSTOM = "custom" +class CharacterStance(str, Enum): + """角色体型 —— 决定"手臂 / 手肘"这类人体部位词能不能进提示词。 + + 非双足角色的描述里出现"手臂",模型会给它凭空接上一对人的上肢来调和文图矛盾, + 而帧数、时长、成色全部正常。成员按"发力部位不同"分,每个非双足成员都要在 + ``ai_engine.prompt.adapter`` 的替换建议表里有自己那套部位说法,否则拒绝理由 + 只能告诉用户"不行"、给不出改法。 + """ + + BIPED = "biped" # 双腿站立 + 一对上肢(人形 / 类人怪) + QUADRUPED = "quadruped" # 四肢着地,前肢兼作手 + SERPENTINE = "serpentine" # 无肢,靠躯干起伏推进(蛇 / 鳗 / 触手) + + class GenRoute(str, Enum): """生成路线 —— 实测挣得的分流依据(见 ai_engine.strategy 层 docstring)。 @@ -116,8 +130,9 @@ class Stylize(str, Enum): class CharacterCard(BaseModel): """角色卡 —— 一致性主键 + 资产库基础(产品核心实体)。 - 注意:视频路线**不读本模型的任何字段**,角色身份由母版图像承载。 - 详见 ``windup_ai_engine.ports.CharacterGeneratorPort`` 的 docstring。 + 注意:视频路线的**角色身份**由母版图像承载,不读 name / desc。它只读 ``stance``, + 且只用来判用户那句描述能不能进提示词。详见 + ``windup_ai_engine.ports.CharacterGeneratorPort`` 的 docstring。 """ model_config = _STRICT @@ -128,6 +143,12 @@ class CharacterCard(BaseModel): master_ref: str = "" # 定妆母版的存储 ref(对象存储,非本地路径) version: str = "v1" + # 体型。默认值等于对每个没声明体型的角色做一次断言,所以取断言最少的那个:双足这一支 + # 不往提示词里加任何部位词、也不要求用户改写措辞。反过来把默认设成非双足,会让占多数的 + # 人形角色被要求把"手臂"改写成"前肢 / 尾",那些词进提示词就是让模型凭空长出对应部位。 + # 不自动识别:那要调模型,而认错是静默的 —— 错误的体型只在下一次付费生成的画面上显形。 + stance: CharacterStance = CharacterStance.BIPED + # 注:曾有 `palette: str = ""`。2026-08-08 删除,理由是它会变成"看起来生效、实则被 # 忽略"的第二真相源:真正锁色的色板由 postprocess.master_pixel_spec 从母版像素里量出来 # (ndarray,喂给 _snap_to_palette),而这个字段零消费方、无格式约定。调用方填了 diff --git a/backend/tests/test_character_contract.py b/backend/tests/test_character_contract.py index 7f2cf01d..304ac866 100644 --- a/backend/tests/test_character_contract.py +++ b/backend/tests/test_character_contract.py @@ -28,6 +28,7 @@ ActionSpec, ActionType, CharacterCard, + CharacterStance, CharacterView, Facing, Stylize, @@ -88,6 +89,26 @@ def test_character_card_default_view_is_a_legal_value(): assert CharacterCard(name="n", desc="d").view in set(CharacterView) +def test_character_stance_rejects_typos(): + """体型拼错要当场炸:裸 str 时 "Quadruped" 会被当成非双足以外的第 N 种值一路放行, + 而它唯一的消费方是"手臂一类词放不放行"的判定 —— 判错的代价是模型给四足角色接上 + 一对人的上肢,一次付费生成之后才在画面上看出来。 + """ + with pytest.raises(ValidationError): + CharacterCard(name="n", desc="d", stance="Quadruped") + card = CharacterCard(name="n", desc="d", stance="quadruped") + assert card.stance is CharacterStance.QUADRUPED + + +def test_character_card_defaults_to_the_least_asserting_stance(): + """默认值是对每个没声明体型的角色做的断言,所以取断言最少的那支。 + + 双足这一支不往提示词里加任何部位词;反过来把默认设成非双足,会让占多数的人形角色 + 被要求把"手臂"改写成"前肢 / 尾",而那些词进了提示词就是让模型凭空长出对应部位。 + """ + assert CharacterCard(name="n", desc="d").stance is CharacterStance.BIPED + + def test_unknown_field_name_is_rejected_not_silently_dropped(): """字段名打错要炸。pydantic 默认 extra="ignore" 会静默吞掉。 diff --git a/backend/tests/test_prompt_adapter.py b/backend/tests/test_prompt_adapter.py new file mode 100644 index 00000000..66e1b214 --- /dev/null +++ b/backend/tests/test_prompt_adapter.py @@ -0,0 +1,384 @@ +"""提示词门禁与适配器。 + +锁的是两类**静默**损失,不是"函数能跑": + ① 措辞规则只写在文档里 → 随包发的提示词自己违反它,而生成照常成功、只是产物没法用; + ② 用户那句大白话原样送进模型 → 报不了错,只在付完钱之后从画面上看出来。 +所以门禁要真的跑在随包发的 md 上,适配器要在**调用付费模型之前**给出可执行的拒绝理由。 +""" +from __future__ import annotations + +import io + +import pytest +from PIL import Image + +from windup_ai_engine.ports import AdaptedPrompt, PromptRejectCode, PromptRejected +from windup_ai_engine.prompt._md import load_doc +from windup_ai_engine.prompt.adapter import _STANCE_PARTS, RuleBasedPromptAdapter +from windup_ai_engine.prompt.custom import CYCLIC_TAIL, MAX_ACTION_CHARS, ONESHOT_TAIL +from windup_ai_engine.prompt.lint import lint +from windup_ai_engine.strategy.concrete import VideoFrameStrategy +from windup_common.models import ( + ActionSpec, + ActionType, + CharacterCard, + CharacterStance, + Facing, + Stylize, +) + +# master_poses 是喂静态模型的母版姿态,其余四份是 i2v 正文 —— 规则的适用范围按机制分, +# 用错 kind 会让 2a/2b 对静态图误报。 +SHIPPED = { + "walk.md": "i2v", "jump.md": "i2v", "idle.md": "i2v", + "attack.md": "i2v", "master_poses.md": "still", +} + + +def _errors(text: str, kind: str) -> list: + return [i for i in lint(text, kind=kind) if i.level == "error"] + + +# ── ① 门禁跑在随包发的提示词上 ──────────────────────────────────────────── + + +@pytest.mark.parametrize("doc", [d for d in SHIPPED if d != "attack.md"]) +def test_shipped_prompts_pass_the_lint(doc: str): + """除 attack 外,随包发的每一节都不许有 error 级问题。""" + for section, text in load_doc(doc).items(): + if not text: # master_poses 里有意留空的节 + continue + bad = _errors(text, SHIPPED[doc]) + assert not bad, f"{doc} `## {section}`: {[(i.category, i.term) for i in bad]}" + + +def test_attack_prompt_is_caught_on_equipment_shape_priors(): + """attack 正文里焊着刃面 / 弧线 / 前手 —— 母版里没有的形状会跟着这个角色走。 + + 这条是门禁的真实性检验:词表写不对,它就报不出这三个。 + """ + caught: set[str] = set() + for section, text in load_doc("attack.md").items(): + caught |= {i.term for i in _errors(text, "i2v") if i.category == "shape_prior"} + assert {"broad side", "crescent", "leading arm"} <= caught, caught + + +def test_lint_reports_the_mechanism_not_just_the_word(): + """报错要能让人照着改;"命中禁词"给不了任何改法。""" + (issue,) = [i for i in lint("kicks up dust", kind="i2v") if i.category == "hazard_noun"] + assert "dust" in issue.message + assert len(issue.message) > 30, issue.message + + +# ── ② 适用范围按机制分 ─────────────────────────────────────────────────── + + +def test_subthreshold_is_fatal_for_video_and_only_a_warning_for_stills(): + """低于可控分辨率的位移会逐帧随机抖 —— 静态图没有帧间,抖不起来。""" + (v,) = [i for i in lint("slightly tilts the head", kind="i2v") if i.category == "subthreshold"] + (s,) = [i for i in lint("slightly tilts the head", kind="still") if i.category == "subthreshold"] + assert v.level == "error" + assert s.level == "warn" + + +def test_unanchored_prop_is_not_checked_for_stills(): + """单件道具的素材图里根本没有身体,拿"身体锚"去要求它是无中生有。""" + text = "holds the lantern out at arm length" + assert [i for i in lint(text, kind="i2v") if i.category == "unanchored_prop"] + assert not [i for i in lint(text, kind="still") if i.category == "unanchored_prop"] + + +def test_a_body_anchor_clears_the_prop_rule(): + """反向:交代了身体整体怎么动就不该再报 —— 否则这条规则等于禁止一切持物动作。""" + text = "holds the lantern out while the torso leans forward" + assert not [i for i in lint(text, kind="i2v") if i.category == "unanchored_prop"] + + +def test_upper_body_parts_do_not_count_as_a_body_anchor(): + """肩 / 肘随手持物一起漂,定不住整体;把它们算作锚会让 2b 形同虚设。""" + assert [i for i in lint("holds the sword steady at the shoulder", kind="i2v") + if i.category == "unanchored_prop"] + + +# ── ③ 适配器:拒绝要讲机制,且发生在花钱之前 ────────────────────────────── + + +def _adapt( + text: str, + *, + kind: str = "i2v", + facing: Facing = Facing.SIDE, + stance: CharacterStance = CharacterStance.BIPED, +): + return RuleBasedPromptAdapter().adapt(text, kind=kind, facing=facing, stance=stance) + + +def _reject(text: str, **kw) -> PromptRejected: + with pytest.raises(PromptRejected) as e: + _adapt(text, **kw) + return e.value + + +# 每条拒绝理由 → 它的 code。server 据 code 选文案,所以"拒了"不够,拒对**哪一条**才算。 +_REJECTIONS = [ + ("", "i2v", PromptRejectCode.EMPTY, "站着不动"), + ("x" * (MAX_ACTION_CHARS + 1), "i2v", PromptRejectCode.TOO_LONG, str(MAX_ACTION_CHARS)), + ("不要扬尘", "i2v", PromptRejectCode.NEGATION, "negative_prompt"), + ("kicks up dust", "i2v", PromptRejectCode.HAZARD_NOUN, "轮廓"), + ("swings with the broad side forward", "i2v", PromptRejectCode.SHAPE_PRIOR, "母版"), + ("轻微抖动一下", "i2v", PromptRejectCode.SUBTHRESHOLD, "抖"), + ("holds the sword steady at the shoulder", "i2v", PromptRejectCode.UNANCHORED_PROP, "漂"), + ("蓄力后攻击再收势", "still", PromptRejectCode.MULTI_STAGE, "分解姿势"), +] + + +@pytest.mark.parametrize(("text", "kind", "code", "mechanism"), _REJECTIONS) +def test_each_rejection_carries_its_own_code_and_the_mechanism(text, kind, code, mechanism): + """拒绝理由必须讲机制:用户拿到"提示词不合规"改不动,拿到"否定词只会被 latch 进画面"才改得动。""" + rejected = _reject(text, kind=kind) + assert rejected.code is code + assert mechanism in rejected.detail, rejected.detail + assert len(rejected.detail) > 30, rejected.detail + + +def test_every_reject_code_is_reachable(): + """枚举值不许只存在于定义里 —— 没有拒绝路径能产出的 code 是死值,server 却要为它写文案。""" + covered = {c for _, _, c, _ in _REJECTIONS} | {PromptRejectCode.STANCE_MISMATCH} + assert covered == set(PromptRejectCode) + + +def test_the_same_multi_stage_description_is_fine_for_video(): + """同一句话对 i2v 成立 —— 规则跟机制走,不跟文本走。""" + assert _adapt("蓄力后攻击再收势", kind="i2v").text + + +def test_both_mechanisms_are_listed_when_two_rules_fire(): + """两条都命中时要把两条都讲出来,改完一条又被拦一次是最烦的;code 取报告序的第一条。""" + rejected = _reject("不要扬尘") + assert rejected.code is PromptRejectCode.NEGATION + assert "不要" in rejected.detail and "扬尘" in rejected.detail + + +def test_impact_words_only_warn_and_still_produce_a_prompt(): + """冲击词是连带风险不是必然,拦下来的代价比放过去大。""" + got = _adapt("slams the whole body onto the floor") + assert [i for i in got.issues if i.category == "impact_verb"] + + +@pytest.mark.parametrize("stance", [s for s in CharacterStance if s is not CharacterStance.BIPED]) +def test_non_biped_stance_rejects_human_limb_wording(stance): + rejected = _reject("raises the left arm high", stance=stance) + assert rejected.code is PromptRejectCode.STANCE_MISMATCH + assert stance.value in rejected.detail + assert _STANCE_PARTS[stance] in rejected.detail, "拒了却没告诉他改成哪个部位" + + +def test_every_non_biped_stance_brings_its_own_replacement_parts(): + """加一个体型却不给替换部位,拒绝理由就退回"这个词不行"。""" + assert set(_STANCE_PARTS) == set(CharacterStance) - {CharacterStance.BIPED} + + +def test_biped_stance_keeps_the_same_wording(): + assert _adapt("raises the left arm high", stance=CharacterStance.BIPED).text + + +@pytest.mark.parametrize("blank", ["", " ", "\n"]) +def test_blank_text_is_rejected_instead_of_silently_producing_a_skeleton(blank): + """空描述只会拿回一段站着不动的视频,而帧数和时长全对。""" + assert _reject(blank).code is PromptRejectCode.EMPTY + + +def test_a_rejection_never_hands_back_a_prompt(): + """拒绝走异常而不是返回值:漏读一个 rejected 字段是静默的,漏 catch 不是。""" + assert "rejected_reason" not in AdaptedPrompt.__dataclass_fields__ + + +# ── ④ 适配器的产物本身要过得了门禁 ─────────────────────────────────────── + + +@pytest.mark.parametrize("facing", [Facing.SIDE, Facing.FRONT]) +@pytest.mark.parametrize("kind", ["i2v", "still"]) +def test_adapted_output_carries_the_skeleton_and_passes_its_own_lint(facing, kind): + got = _adapt("来回走动", kind=kind, facing=facing) + assert "来回走动" in got.text, "用户那句话没进提示词" + assert "whatever the character already wears" in got.text # 装备存在无关 + assert "One single character alone in the frame" in got.text # 单主体 + 构图 + assert not _errors(got.text, kind), _errors(got.text, kind) + + +def test_adapted_output_names_a_whole_body_anchor(): + """"来回走动"这类位移动作,骨架必须自己带上整体位移词,否则持物会漂。""" + text = _adapt("来回走动").text + assert any(w in text.lower() for w in ("whole body", "body", "torso", "hips")) + + +def test_still_output_demands_a_single_instant_and_video_output_does_not(): + """静态图要单一瞬间;i2v 反过来,写"一个瞬间"等于要一段静止视频。""" + assert "ONE single frozen instant" in _adapt("来回走动", kind="still").text + assert "frozen instant" not in _adapt("来回走动", kind="i2v").text + + +def test_video_output_leaves_the_loop_tail_to_the_caller(): + """循环性是请求的属性,适配器的入参里没有 —— 它替调用方猜一条就是静默出错。""" + text = _adapt("来回走动", kind="i2v").text + assert CYCLIC_TAIL not in text and ONESHOT_TAIL not in text + + +# ── ⑤ 接进管线:回退、拒绝、循环性 ─────────────────────────────────────── + + +def _png() -> bytes: + im = Image.new("RGBA", (64, 96), (0, 0, 0, 0)) + for y in range(20, 80): + for x in range(24, 40): + im.putpixel((x, y), (200, 60, 60, 255)) + buf = io.BytesIO() + im.save(buf, "PNG") + return buf.getvalue() + + +class _NullProgress: + def step(self, stage: str, i: int, total: int, note: str = "") -> None: + pass + + +class _SpyVideo: + def __init__(self) -> None: + self.prompts: list[str] = [] + + def i2v(self, first_frame, prompt, seconds=5, size="1280x720"): + self.prompts.append(prompt) + return b"fake-mp4" + + +class _Matte: + def cutout(self, frame): + return frame + + +def _spec(action: str, *, cyclic: bool = False) -> ActionSpec: + return ActionSpec( + action=ActionType.CUSTOM, custom_action=action, cyclic=cyclic, + n_frames=8, stylize=Stylize.NONE, + ) + + +def _offline(monkeypatch, adapter=None) -> tuple[VideoFrameStrategy, _SpyVideo]: + dense = [Image.open(io.BytesIO(_png())).convert("RGBA") for _ in range(24)] + monkeypatch.setattr( + "windup_ai_engine.strategy.concrete.extract_all_frames_bytes", + lambda video, cap=150: dense, + ) + video = _SpyVideo() + return VideoFrameStrategy(video, _Matte(), adapter), video + + +def test_the_prompt_that_reaches_the_model_went_through_the_adapter(monkeypatch): + strat, video = _offline(monkeypatch) + strat.derive(CharacterCard(name="t", desc="t"), + _spec("waves the right hand above the head"), _png(), _NullProgress()) + (sent,) = video.prompts + assert "waves the right hand above the head" in sent + assert "One single character alone in the frame" in sent, "适配器没接上" + assert sent.endswith(ONESHOT_TAIL), "一次性动作丢了单次 + 终态保持" + + +def test_the_declared_loop_flag_still_picks_the_tail(monkeypatch): + """适配器不碰循环性,但循环性必须仍然生效。""" + strat, video = _offline(monkeypatch) + card = CharacterCard(name="t", desc="t") + strat.derive(card, _spec("来回走动", cyclic=True), _png(), _NullProgress()) + assert video.prompts[-1].endswith(CYCLIC_TAIL) + + +def test_a_rejected_description_never_reaches_the_paid_call(monkeypatch): + """拒绝的全部价值在这一条:错在调用之前抛,而不是花完钱看画面。""" + strat, video = _offline(monkeypatch) + with pytest.raises(PromptRejected) as e: + strat.derive(CharacterCard(name="t", desc="t"), + _spec("轻微抖动一下"), _png(), _NullProgress()) + # 类型和 code 都要穿到管线外:server 靠它们判 4xx 与选文案,靠 parse 消息的话 + # 改一次措辞就分支失效,而失效的表现是把用户的输入问题报成"系统出问题了"。 + assert e.value.code is PromptRejectCode.SUBTHRESHOLD + assert video.prompts == [], "拒了还是把请求发了出去" + + +# ── ⑥ 体型从角色卡进到判定 ─────────────────────────────────────────────── + + +class _StanceSpy: + """记下真实收到的入参 —— 断言桩返回了什么证明不了管线传对了东西。""" + + def __init__(self) -> None: + self.seen: list = [] + + def adapt(self, user_text, *, kind, facing, stance): + self.seen.append(stance) + return AdaptedPrompt(text="ZZREWRITTEN") + + +@pytest.mark.parametrize("stance", list(CharacterStance)) +def test_the_card_stance_is_what_the_adapter_receives(monkeypatch, stance): + """写死 biped 时这条变红:非双足规则在真实管线里一次都不会生效。""" + spy = _StanceSpy() + strat, _ = _offline(monkeypatch, spy) + strat.derive(CharacterCard(name="t", desc="t", stance=stance), + _spec("走两步"), _png(), _NullProgress()) + assert spy.seen == [stance] + + +def test_a_quadruped_card_rejects_human_limb_wording_end_to_end(monkeypatch): + """规则 5 在真实管线里生效的唯一证据:走 derive、用真适配器、不碰付费调用。""" + strat, video = _offline(monkeypatch) + card = CharacterCard(name="t", desc="t", stance=CharacterStance.QUADRUPED) + with pytest.raises(PromptRejected) as e: + strat.derive(card, _spec("raises the left arm high"), _png(), _NullProgress()) + assert e.value.code is PromptRejectCode.STANCE_MISMATCH + assert video.prompts == [] + + +def test_the_same_wording_passes_for_the_default_biped_card(monkeypatch): + """反向:默认体型不拦人体部位词,否则占多数的人形角色被逼着写"前肢"。""" + strat, video = _offline(monkeypatch) + strat.derive(CharacterCard(name="t", desc="t"), + _spec("raises the left arm high"), _png(), _NullProgress()) + assert "raises the left arm high" in video.prompts[0] + + +def test_a_broken_adapter_falls_back_to_the_existing_skeleton(monkeypatch): + """新组件坏掉不许把整条生成打死 —— 骨架不依赖它。""" + + class _Boom: + def adapt(self, user_text, *, kind, facing, stance): + raise RuntimeError("模型服务挂了") + + strat, video = _offline(monkeypatch, _Boom()) + strat.derive(CharacterCard(name="t", desc="t"), + _spec("waves the right hand"), _png(), _NullProgress()) + (sent,) = video.prompts + assert "waves the right hand" in sent + assert "SIDE VIEW facing right" in sent # 朝向锁还在 + assert sent.endswith(ONESHOT_TAIL) + assert "One single character alone" not in sent, "坏掉的适配器不该还有产物" + + +def test_a_custom_adapter_can_replace_the_rule_based_one(monkeypatch): + """协议是真的可替换的:换实现只换注入,管线零改动。""" + + class _Fixed: + def adapt(self, user_text, *, kind, facing, stance): + return AdaptedPrompt(text="ZZREWRITTEN") + + strat, video = _offline(monkeypatch, _Fixed()) + strat.derive(CharacterCard(name="t", desc="t"), + _spec("waves"), _png(), _NullProgress()) + assert video.prompts[0].startswith("ZZREWRITTEN") + + +def test_fixed_actions_are_untouched_by_the_adapter(monkeypatch): + """只有 custom 的文本来自用户;walk 一类的正文是校准过的,不许被改写。""" + strat, video = _offline(monkeypatch) + strat.derive(CharacterCard(name="t", desc="t"), + ActionSpec(action=ActionType.WALK, n_frames=8, stylize=Stylize.NONE), + _png(), _NullProgress()) + assert "One single character alone" not in video.prompts[0]