diff --git a/.env.example b/.env.example index b986b308..423b7b93 100644 --- a/.env.example +++ b/.env.example @@ -37,6 +37,15 @@ AI_API_KEY=your-ai-api-key # 取值即默认值——不写这两行时跑的就是它们。 AI_IMAGE_MODEL=gemini-2.5-flash-image AI_VIDEO_MODEL=kling-v2-5-turbo +# 判官(看图问答)。必须填一个**能读图的聊天模型**,不是出图模型:它要回 JSON 不是回图。 +AI_JUDGE_MODEL=gemini-2.5-flash + +# ── 判官闸口 ── +# 开了才会调判官,每交付一个动作多一次付费调用。 +QUALITY_GATE_ENABLED=false +# 判官说有问题就不交付。攒够 shadow 数据、定出判据之前别开: +# 误杀掉的是用户已付过钱的产物,退不回来。 +QUALITY_GATE_ENFORCE=false # ── 积分定价 ── QUOTA_REGISTER_GIFT_AMOUNT=100 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..d4834449 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,7 @@ from enum import Enum from typing import Protocol, runtime_checkable -from windup_common.models import ActionSpec, CharacterCard +from windup_common.models import ActionSpec, CharacterCard, JudgeVerdict # ---- server 实现、注入给 ai_engine 的进度回调 port ---- @@ -124,6 +124,21 @@ class GeneratedAction: quality: ActionQuality = field(kw_only=True) +# ---- 出门那道闸的仪器:判官(server 注入实现,framework 层有一个)---- +# 与 ``ActionQuality`` 分工不同:那三个数由本地像素算出来,零成本、量的是帧**之间**的 +# 关系;判官量的是一帧画面**里**有什么,要花一次付费调用,且本地算不出来 —— 像素统计 +# 分不出"两个角色"和"一个角色 + 一件道具"。 +@runtime_checkable +class JudgePort(Protocol): + """交付帧 + 母版 → 四个可数读数(:class:`JudgeVerdict`)。 + + 读不出结论必须抛错,不得兜底成"通过":静默放行会让"没判"与"判了没问题"在下游 + 长得一样。``master`` 必填 —— "有没有母版里没有的物体"离开母版无从回答。 + """ + + def judge(self, frame: bytes, master: bytes, action: str) -> JudgeVerdict: ... + + # ---- ai_engine 暴露给 server(server 调用的唯一入口)---- @runtime_checkable class CharacterGeneratorPort(Protocol): diff --git a/backend/packages/app/src/windup_app/server/orchestrator/executor.py b/backend/packages/app/src/windup_app/server/orchestrator/executor.py index 7459e6f7..03941a46 100644 --- a/backend/packages/app/src/windup_app/server/orchestrator/executor.py +++ b/backend/packages/app/src/windup_app/server/orchestrator/executor.py @@ -22,8 +22,9 @@ from sqlalchemy.orm import Session from windup_common.models import ActionSpec, ActionType as EngineActionType, CharacterCard +from windup_framework.config.quality_gate import settings as gate_settings -from windup_app.server.orchestrator import task_repo +from windup_app.server.orchestrator import quality_gate, task_repo from windup_app.server.orchestrator._fetch import fetch_own_media from windup_app.server.orchestrator.model import ( CharacterActionInput, @@ -32,7 +33,7 @@ ) if TYPE_CHECKING: - from windup_ai_engine.ports import CharacterGeneratorPort, ProgressPort + from windup_ai_engine.ports import CharacterGeneratorPort, JudgePort, ProgressPort from windup_framework.providers import ImageProvider, MatteProvider logger = logging.getLogger("windup.generation.executor") @@ -183,6 +184,7 @@ def __init__( self, *, generator: CharacterGeneratorPort | None = None, + judge: JudgePort | None = None, upload: Callable[[bytes], str] | None = None, fetch_master: Callable[[CharacterActionInput], bytes] | None = None, fetch_constraints: Callable[[Session, int | None], ProjectConstraints] | None = None, @@ -195,6 +197,9 @@ def __init__( # 各自惰性加载一份 ONNX 会话,按桶各建等于把同一个模型在进程里装多次。 self._matte: MatteProvider | None = None self._image: ImageProvider | None = None + # 判官同样与视频模型无关,故不分桶。缺省 None 时**不建**实例:建了就意味着每个 + # 任务多一次付费调用,那要由 QUALITY_GATE_ENABLED 显式打开,见 _get_judge。 + self._judge: JudgePort | None = judge # 本执行器是进程级单例,而每个请求起一个线程跑 run_action_task,上面几个缓存 # 都是跨线程共用的可变状态。缺锁时并发首请求会各装一套(见 _get_generator)。 self._assembly_lock = threading.Lock() @@ -286,13 +291,37 @@ def _produce_action(self, input: CharacterActionInput, cons: ProjectConstraints) ) upload = self._upload or self._upload_frame + checked = [_require_size(png, cons.sprite_w, cons.sprite_h) for png in generated.frames] frames = [ - {"index": i, - "image_url": upload(_require_size(png, cons.sprite_w, cons.sprite_h)), - "duration_ms": dur} - for i, (png, dur) in enumerate(zip(generated.frames, generated.durations)) + {"index": i, "image_url": upload(png), "duration_ms": dur} + for i, (png, dur) in enumerate(zip(checked, generated.durations)) ] - return {"type": "character_action", "action_type": input.action_type.value, "frames": frames} + result = { + "type": "character_action", + "action_type": input.action_type.value, + "frames": frames, + } + decision = quality_gate.review( + self._get_judge(), checked, master, input.action_type.value + ) + if decision is not None: + result["judge"] = decision.as_payload() + if decision.blocked: + # 帧已经生成、已经上传,钱早就花完了。拦在这里的意义只剩"不把坏产物当成 + # 交付物交出去";这也正是拦截档默认关着的原因。 + raise quality_gate.QualityBlocked(decision.problems) + return result + + def _get_judge(self) -> JudgePort | None: + """闸口启用时懒建判官;未启用返回 ``None``,一次调用都不发。""" + if self._judge is not None or not gate_settings.enabled: + return self._judge + with self._assembly_lock: + if self._judge is None: + from windup_framework.providers import SufyJudgeProvider + + self._judge = SufyJudgeProvider() + return self._judge def _get_generator(self, video_model: str | None = None) -> CharacterGeneratorPort: """懒装配 CharacterGenerator,按模型名分桶。 diff --git a/backend/packages/app/src/windup_app/server/orchestrator/model.py b/backend/packages/app/src/windup_app/server/orchestrator/model.py index 3c4aab8e..161decb9 100644 --- a/backend/packages/app/src/windup_app/server/orchestrator/model.py +++ b/backend/packages/app/src/windup_app/server/orchestrator/model.py @@ -116,6 +116,13 @@ class CharacterActionOutput: type: str = "character_action" action_type: str = "" frames: list[CharacterActionFrame] = field(default_factory=list) + # 判官读数(``quality_gate.GateDecision.as_payload``)。``None`` = **没判**,不是 + # "判了没问题" —— 闸口默认不启用,把缺省读成"干净"会让 shadow 期的统计凭空多出一批 + # 从未判读过的样本。形状留 dict 而不是拆成字段:shadow 期正是要观察该记哪些东西, + # 每加一个读数就改一次 ORM 反序列化的话,数据还没攒够就先僵住了。 + # 字段名不叫 quality:引擎那份本地像素成色(``ports.ActionQuality``)也要落到同一个 + # payload 里,两者来源与代价都不同,共用一个键会让先写的那份被后写的悄悄盖掉。 + judge: dict | None = None # -- 任务记录 ------------------------------------------------------------ diff --git a/backend/packages/app/src/windup_app/server/orchestrator/quality_gate.py b/backend/packages/app/src/windup_app/server/orchestrator/quality_gate.py new file mode 100644 index 00000000..0d91b0ca --- /dev/null +++ b/backend/packages/app/src/windup_app/server/orchestrator/quality_gate.py @@ -0,0 +1,123 @@ +"""出门那道闸:把判官读数翻成"交不交付"。 + +在 server 而不是 ai_engine,因为判出问题之后的每一种处置(退款、重跑、换母版)都要 +再花一次钱,那是产品决策;引擎的分工是如实报数,判决归调用方。 +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass + +from windup_ai_engine.ports import JudgePort, JudgeVerdict +from windup_framework.config.quality_gate import QualityGateSettings, settings + +logger = logging.getLogger("windup.generation.quality_gate") + +# 稳定的机器可读标签。用槽位名而不是句子,是因为下游要按它分桶统计; +# 给人看的解释在 :func:`_problems` 的判据里,不在这几个字符串里。 +PROBLEM_MULTIPLE_SUBJECTS = "multiple_subjects" +PROBLEM_NO_SUBJECT = "no_subject" +PROBLEM_FOREIGN_OBJECTS = "foreign_objects" +PROBLEM_ACTION_MISMATCH = "action_mismatch" +PROBLEM_CLIPPED = "clipped" + +_EXPECTED_SUBJECTS = 1 + + +@dataclass(frozen=True) +class GateDecision: + """一次判读的结论。 + + ``verdict`` 有值 = 判过了,``problems`` 可能为空;``error`` 有值 = 判官坏了、 + 什么都没判出来,那不是"通过"。压成一个布尔的话这两种状态就没法分开了。 + """ + + frame_index: int + problems: tuple[str, ...] = () + blocked: bool = False + verdict: JudgeVerdict | None = None + error: str | None = None + + def as_payload(self) -> dict: + """写进任务结果的形状 —— 复核要用的东西一样不少(含模型原话)。""" + data: dict = { + "frame_index": self.frame_index, + "problems": list(self.problems), + "blocked": self.blocked, + } + if self.error is not None: + data["error"] = self.error + if self.verdict is not None: + data["subject_count"] = self.verdict.subject_count + data["foreign_objects"] = list(self.verdict.foreign_objects) + data["action_matches"] = self.verdict.action_matches + data["clipped"] = self.verdict.clipped + data["raw"] = self.verdict.raw + return data + + +class QualityBlocked(ValueError): + """判官判出问题且闸口处于拦截档 —— 不交付。""" + + def __init__(self, problems: tuple[str, ...]) -> None: + super().__init__(f"交付被判官拦下:{', '.join(problems)}") + self.problems = problems + + +def pick_frame(count: int) -> int: + """只判一帧 —— 判满一段等于把成本乘上帧数。 + + 取中间那帧而不是首帧:首帧最接近母版,正是"动作对不对"最看不出来的一帧。 + """ + return count // 2 + + +def review( + judge: JudgePort | None, + frames: list[bytes], + master: bytes, + action: str, + *, + config: QualityGateSettings = settings, +) -> GateDecision | None: + """判一段交付物;``None`` = 没判(没注入判官或闸口未启用)。 + + "没判"要与"判了没问题"分得开:只有后者能支持"这批产物是干净的"这句话。 + """ + if judge is None or not config.enabled or not frames: + return None + + index = pick_frame(len(frames)) + try: + verdict = judge.judge(frames[index], master, action) + except Exception as exc: # noqa: BLE001 —— 判官的任何故障都归"仪器坏了" + # 仪器故障绝不拦截:拦下去等于因为我们自己的判官挂了,把用户已付费的产物扣住。 + logger.warning("判官判读失败(第 %d 帧,动作 %s):%s", index, action, exc) + return GateDecision(frame_index=index, error=str(exc)) + + problems = _problems(verdict) + # shadow 档(``enforce`` 默认 false)只记不拦:阈值要拿 shadow 数据定,反过来先开 + # 拦截就是拍脑袋定判据,而误杀掉的是用户已付费、退不回来的产物。 + blocked = bool(problems) and config.enforce + if problems: + logger.info("判官在第 %d 帧读出 %s(拦截=%s)", index, problems, blocked) + return GateDecision( + frame_index=index, problems=problems, blocked=blocked, verdict=verdict, + ) + + +def _problems(verdict: JudgeVerdict) -> tuple[str, ...]: + """四问 → 问题标签。每一条都有唯一答案,不含任何阈值。""" + found: list[str] = [] + if verdict.subject_count > _EXPECTED_SUBJECTS: + found.append(PROBLEM_MULTIPLE_SUBJECTS) + elif verdict.subject_count < _EXPECTED_SUBJECTS: + found.append(PROBLEM_NO_SUBJECT) + if verdict.foreign_objects: + found.append(PROBLEM_FOREIGN_OBJECTS) + if not verdict.action_matches: + found.append(PROBLEM_ACTION_MISMATCH) + if verdict.clipped: + found.append(PROBLEM_CLIPPED) + return tuple(found) diff --git a/backend/packages/app/src/windup_app/server/orchestrator/task_repo.py b/backend/packages/app/src/windup_app/server/orchestrator/task_repo.py index d2b60b5c..a93f2337 100644 --- a/backend/packages/app/src/windup_app/server/orchestrator/task_repo.py +++ b/backend/packages/app/src/windup_app/server/orchestrator/task_repo.py @@ -227,5 +227,6 @@ def _deserialize_result( type=raw.get("type", "character_action"), action_type=raw.get("action_type", ""), frames=frames, + judge=raw.get("judge"), ) return None diff --git a/backend/packages/common/src/windup_common/models/__init__.py b/backend/packages/common/src/windup_common/models/__init__.py index 52434b34..c8a4551d 100644 --- a/backend/packages/common/src/windup_common/models/__init__.py +++ b/backend/packages/common/src/windup_common/models/__init__.py @@ -8,6 +8,7 @@ GenRoute, Stylize, ) +from windup_common.models.quality import JudgeVerdict __all__ = [ "ActionType", @@ -18,4 +19,5 @@ "DEFAULT_N_FRAMES", "CharacterCard", "ActionSpec", + "JudgeVerdict", ] diff --git a/backend/packages/common/src/windup_common/models/quality.py b/backend/packages/common/src/windup_common/models/quality.py new file mode 100644 index 00000000..b5541ec3 --- /dev/null +++ b/backend/packages/common/src/windup_common/models/quality.py @@ -0,0 +1,33 @@ +"""判官读数的数据契约。 + +落在 common 而不是 ai_engine.ports,因为构造它的是 framework 层的 provider,而分层 +门禁禁止 framework 依赖 ai_engine;调用方仍从 ``windup_ai_engine.ports`` 取。 +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +# 刻意**没有** score 字段:主观评分的噪声大到能盖过真实差异,而出参里一旦有个分数,迟早 +# 有人拿它卡阈值 —— 那时卡掉的是噪声,每一次误杀都是用户已付费、不可退的产物。下面四项 +# 各有唯一答案,人眼复核一遍就能确认对错。"好看"由输入端(母版规格 + 提示词骨架)保证。 +@dataclass(frozen=True) +class JudgeVerdict: + """一帧交付物的四个可数、可复核读数 —— 不含"好不好看"的判断。""" + + subject_count: int + """画面里出现了几个角色主体。期望 1;≥2 通常是 i2v 把角色分裂成了两个。""" + + foreign_objects: tuple[str, ...] + """母版里没有、生成帧里却出现的物体名。空元组 = 没有多出来的东西。""" + + action_matches: bool + """这一帧的姿态是否属于所要求的动作类别 —— 判类别,不判动作做得好不好。""" + + clipped: bool + """角色是否被画面边缘裁到。""" + + raw: str + """模型原话。留着是为了让人能复核判读本身对不对 —— 四个读数都是模型给的, + 判官自己出错时,没有原话就无从分辨"产物真有问题"和"判官读错了"。""" diff --git a/backend/packages/framework/src/windup_framework/config/provider.py b/backend/packages/framework/src/windup_framework/config/provider.py index faeede24..be4df4c9 100644 --- a/backend/packages/framework/src/windup_framework/config/provider.py +++ b/backend/packages/framework/src/windup_framework/config/provider.py @@ -33,6 +33,11 @@ class AIProviderSettings(BaseSettings): # 而费用可能已经产生(2026-07-29 实测)。 video_model: str = "kling-v2-5-turbo" image_model: str = "gemini-2.5-flash-image" + # 判官是**看图的聊天模型**,不是图像生成模型:它要读一张图然后回一段 JSON,而 + # ``image_model`` 那个型号只会回图;共用一个字段的话,换判官会连带把出图换掉。 + # 本默认值未在本仓实测过 —— 网关目录里没有它时,``_post`` 的 400/404 分支会指到 + # ``GET /models`` 去核对。 + judge_model: str = "gemini-2.5-flash" @property def normalized_base_url(self) -> str: diff --git a/backend/packages/framework/src/windup_framework/config/quality_gate.py b/backend/packages/framework/src/windup_framework/config/quality_gate.py new file mode 100644 index 00000000..8eecf9b0 --- /dev/null +++ b/backend/packages/framework/src/windup_framework/config/quality_gate.py @@ -0,0 +1,27 @@ +"""判官闸口配置。""" + +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class QualityGateSettings(BaseSettings): + """判官闸口开关。环境变量前缀 ``QUALITY_GATE_``。 + + 两个开关分开,是因为它们各自的代价不同,不该被一个 flag 绑在一起。 + """ + + model_config = SettingsConfigDict( + env_prefix="QUALITY_GATE_", + env_file=("../.env", ".env"), + env_file_encoding="utf-8", + extra="ignore", + ) + + # 每交付一个动作多打一次付费模型调用,所以默认不开:开它是一次花钱的决定。 + enabled: bool = False + + # 判官说有问题就不交付。默认关,而且在积够 shadow 数据、定出判据之前不该开: + # 误杀掉的是用户**已经付过钱**的产物,退不回来;而漏放一个坏产物,用户可以重试。 + enforce: bool = False + + +settings = QualityGateSettings() diff --git a/backend/packages/framework/src/windup_framework/providers/__init__.py b/backend/packages/framework/src/windup_framework/providers/__init__.py index fd1f448e..6408c8cf 100644 --- a/backend/packages/framework/src/windup_framework/providers/__init__.py +++ b/backend/packages/framework/src/windup_framework/providers/__init__.py @@ -3,6 +3,7 @@ from windup_framework.config.provider import AIProviderSettings from windup_framework.providers.chat import create_chat_model from windup_framework.providers.image import create_image_client +from windup_framework.providers.judge import JudgeResponseError, SufyJudgeProvider from windup_framework.providers.interfaces import ( ImageProvider, MatteProvider, @@ -29,4 +30,7 @@ # FAL 队列面的 i2v(现役接口形态);首帧要公网 URL,故与 uploader 成对出现 "SufyImageProvider", "OnnxU2NetMatteProvider", + # 判官:出参是结构化读数而不是 bytes,故不在 interfaces 的三个 Protocol 之列 + "SufyJudgeProvider", + "JudgeResponseError", ] diff --git a/backend/packages/framework/src/windup_framework/providers/judge.py b/backend/packages/framework/src/windup_framework/providers/judge.py new file mode 100644 index 00000000..cb518422 --- /dev/null +++ b/backend/packages/framework/src/windup_framework/providers/judge.py @@ -0,0 +1,146 @@ +"""判官 provider —— 问四个有唯一答案的问题,不问"好不好看"。 + +读不出结论一律抛错:兜底成"通过"会让判官坏掉与产物没问题在下游长得一样,而这两种 +情形要做的事相反 —— 一个是去修判官,一个是照常交付。 +""" + +from __future__ import annotations + +import base64 +import json +import re + +from windup_common.models import JudgeVerdict +from windup_framework.config.provider import AIProviderSettings, settings + +from .sufy import ChatCompletionsFace + +# 四问是**可数、可复核**的,所以提示词把答案形状写死成一个对象、并逐字段说清判据。 +# 不要求模型解释理由:理由是自由文本,读它就等于又回到主观判断。 +_PROMPT = """You are a strict visual inspector for 2D game sprite frames. + +Image 1 is the MASTER reference of the character. +Image 2 is one GENERATED frame to inspect. + +Only report what can be counted or verified against image 1. Do NOT rate quality, +style, beauty, anatomy or appeal. Do not explain. + +Reply with exactly one JSON object and nothing else: +{{"subject_count": , "foreign_objects": [, ...], \ +"action_matches": , "clipped": }} + +- subject_count: how many distinct character bodies are visible in image 2. +- foreign_objects: short names of objects visible in image 2 but absent from + image 1; [] when there are none. +- action_matches: true when the pose in image 2 belongs to the action "{action}". +- clipped: true when any part of the character is cut off by the image border. +""" + +# 模型常把 JSON 裹进 markdown 代码围栏。剥围栏是解析的一部分,不是兜底 —— +# 剥完仍不是合法 JSON 照样抛。 +_FENCE = re.compile(r"^\s*```(?:json)?\s*(.*?)\s*```\s*$", re.DOTALL) + +_REQUIRED = ("subject_count", "foreign_objects", "action_matches", "clipped") + + +class JudgeResponseError(RuntimeError): + """判官的回答读不出四个读数 —— 仪器故障,不是产物有问题。 + + 单列一个类型,好让上层把它与"产物被判有问题"分开:后者可以据以拦截,前者拦谁都不对。 + """ + + +class SufyJudgeProvider(ChatCompletionsFace): + """看图问答判官:一帧 + 母版 → :class:`JudgeVerdict`。""" + + def __init__( + self, + config: AIProviderSettings = settings, + model: str | None = None, + ) -> None: + super().__init__(config, model or config.judge_model) + + def judge(self, frame: bytes, master: bytes, action: str) -> JudgeVerdict: + body = { + "model": self._model, + "messages": [{ + "role": "user", + "content": [ + {"type": "text", "text": _PROMPT.format(action=action)}, + _image_part(master), + _image_part(frame), + ], + }], + # 判读要的是同一张图每次给同一个答案;温度一高,阈值卡的就成了采样噪声。 + "temperature": 0, + # 提示词里已经要过 JSON,这里再要一次是因为两者的强度不同:提示词到哪个模型 + # 都生效但可以被无视,response_format 在支持它的网关上是硬约束。 + "response_format": {"type": "json_object"}, + } + with self._client() as client: + payload = self._post(client, body) + return _parse_verdict(_content(payload)) + + +def _image_part(raw: bytes) -> dict: + return { + "type": "image_url", + "image_url": {"url": "data:image/png;base64," + base64.b64encode(raw).decode()}, + } + + +def _content(payload: dict) -> str: + """取出模型正文;``content`` 在不同网关下是字符串或 parts 数组,别的形状抛错。""" + try: + content = payload["choices"][0]["message"]["content"] + except (KeyError, IndexError, TypeError) as exc: + raise JudgeResponseError(f"判官响应里没有 message.content:{json.dumps(payload)[:300]}") from exc + if isinstance(content, str): + return content + if isinstance(content, list): + text = "".join(p.get("text", "") for p in content if isinstance(p, dict)) + if text: + return text + raise JudgeResponseError(f"判官响应的 content 形状读不了:{str(content)[:300]}") + + +def _parse_verdict(text: str) -> JudgeVerdict: + """模型正文 → :class:`JudgeVerdict`;任何一项读不出来就抛 :class:`JudgeResponseError`。""" + stripped = _FENCE.sub(r"\1", text) + try: + data = json.loads(stripped) + except json.JSONDecodeError as exc: + raise JudgeResponseError(f"判官没有返回 JSON:{text[:300]!r}") from exc + if not isinstance(data, dict): + raise JudgeResponseError(f"判官返回的不是 JSON 对象:{text[:300]!r}") + missing = [k for k in _REQUIRED if k not in data] + if missing: + raise JudgeResponseError(f"判官回答缺字段 {missing}:{text[:300]!r}") + + return JudgeVerdict( + subject_count=_as_count(data["subject_count"], text), + foreign_objects=_as_names(data["foreign_objects"], text), + action_matches=_as_bool(data["action_matches"], "action_matches", text), + clipped=_as_bool(data["clipped"], "clipped", text), + raw=text, + ) + + +def _as_count(value: object, text: str) -> int: + # bool 是 int 的子类,不排掉的话 ``true`` 会被读成 1 个主体 —— 一个凭空捏造的读数。 + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise JudgeResponseError(f"subject_count 不是非负整数({value!r}):{text[:300]!r}") + return value + + +def _as_names(value: object, text: str) -> tuple[str, ...]: + if not isinstance(value, list) or any(not isinstance(v, str) for v in value): + raise JudgeResponseError(f"foreign_objects 不是字符串数组({value!r}):{text[:300]!r}") + return tuple(value) + + +def _as_bool(value: object, field: str, text: str) -> bool: + # 不接 "true" / 1:字符串与数字要靠一套约定才能变成布尔,而约定错了没人会发现。 + if not isinstance(value, bool): + raise JudgeResponseError(f"{field} 不是布尔值({value!r}):{text[:300]!r}") + return value diff --git a/backend/packages/framework/src/windup_framework/providers/sufy.py b/backend/packages/framework/src/windup_framework/providers/sufy.py index e881feca..f2bc197f 100644 --- a/backend/packages/framework/src/windup_framework/providers/sufy.py +++ b/backend/packages/framework/src/windup_framework/providers/sufy.py @@ -340,31 +340,25 @@ def _retry_exhausted_message(status: int, tries: int) -> str: _DATA_URI = re.compile(r"data:image/[^;]+;base64,([A-Za-z0-9+/=]{100,})") -class SufyImageProvider(ImageProvider): - """文生图 / 图生图 provider(OpenAI 兼容的 ``/chat/completions`` 面)。 +class ChatCompletionsFace: + """网关 ``/chat/completions`` 面的共用管道:建 client、发请求、判哪些失败可以重发。 - 调用形状与 i2v 那两个 provider 完全不同:图像走 chat 接口、参考图以 data URI 塞进 - ``content`` 数组,没有提交-轮询-下载三段式。 - - 2026-08-10 修:此前 ``gen_image`` 直接抛 NotImplementedError,而 - ``POST /generation/image`` 端点是可达的、``ImageTaskExecutor`` 又默认实例化本类 —— - 于是每个图像任务都稳定走到 FAILED。端点看着可用、实际必失败,正是本仓最忌讳的形态 - (机器审逮到)。实现取自管线仓已跑通的通路(同日用它出过三张角色母版)。 + 出图与判官共用一份 —— 同一网关同一把 key,限流与 52x 的语义一样;各写一份的话, + 改一次重试判据要记得改两处,漏掉的那处的代价是重复计费。 """ - def __init__( - self, - config: AIProviderSettings = settings, - model: str | None = None, - ) -> None: + # 出图比一次问答慢得多,所以超时按能力放大;判官用基准超时。 + _timeout_multiplier: float = 1.0 + + def __init__(self, config: AIProviderSettings, model: str) -> None: self._cfg = config - self._model = model or config.image_model + self._model = model def _client(self) -> httpx.Client: return httpx.Client( base_url=self._cfg.normalized_base_url, headers={"Authorization": f"Bearer {self._cfg.api_key}"}, - timeout=self._cfg.timeout * _IMAGE_TIMEOUT_MULTIPLIER, + timeout=self._cfg.timeout * self._timeout_multiplier, # retries 只覆盖建连阶段的失败(SSL 握手、连接被重置)。本机走代理时这类抖动 # 常见,已跑通的管线实现正是靠一层网络重试扛住的;不加会在人家能恢复的地方 # 放弃。它不重试读超时与 5xx —— 那两种请求可能已达上游,重发会重复计费。 @@ -394,7 +388,7 @@ def _post(self, client: httpx.Client, body: dict) -> dict: # 上限同样兜住指数退避:上游挂掉时不该把一个图像任务堵成长时间阻塞。 delay = min(float(2**attempt), _MAX_RETRY_WAIT) logger.warning( - "图像服务返回 %d，第 %d/%d 次请求，%.2f 秒后重试", + "模型服务返回 %d，第 %d/%d 次请求，%.2f 秒后重试", code, attempt, _POST_TRIES, @@ -410,6 +404,28 @@ def _post(self, client: httpx.Client, body: dict) -> dict: ) return resp.raise_for_status().json() + +class SufyImageProvider(ChatCompletionsFace, ImageProvider): + """文生图 / 图生图 provider(OpenAI 兼容的 ``/chat/completions`` 面)。 + + 调用形状与 i2v 那两个 provider 完全不同:图像走 chat 接口、参考图以 data URI 塞进 + ``content`` 数组,没有提交-轮询-下载三段式。 + + 2026-08-10 修:此前 ``gen_image`` 直接抛 NotImplementedError,而 + ``POST /generation/image`` 端点是可达的、``ImageTaskExecutor`` 又默认实例化本类 —— + 于是每个图像任务都稳定走到 FAILED。端点看着可用、实际必失败,正是本仓最忌讳的形态 + (机器审逮到)。实现取自管线仓已跑通的通路(同日用它出过三张角色母版)。 + """ + + _timeout_multiplier = _IMAGE_TIMEOUT_MULTIPLIER + + def __init__( + self, + config: AIProviderSettings = settings, + model: str | None = None, + ) -> None: + super().__init__(config, model or config.image_model) + def gen_image(self, prompt: str, refs: list[bytes]) -> bytes: """提示词 + 参考图 → 一张 PNG bytes。拿不到有效图就抛,不返回空 bytes。 diff --git a/backend/tests/test_quality_judge.py b/backend/tests/test_quality_judge.py new file mode 100644 index 00000000..0230ec10 --- /dev/null +++ b/backend/tests/test_quality_judge.py @@ -0,0 +1,465 @@ +"""判官 provider 与出门闸口(不联网:httpx MockTransport,不产生任何费用)。 + +桩是个真去数提交上来的图里有几个主体的假模型,不是一条写死的回答:写死回答只能证明 +"我们会解析 JSON",而请求形状、base64、参考图顺序错了它一条都发现不了。 +""" + +from __future__ import annotations + +import base64 +import io +import json +import re + +import httpx +import pytest +from PIL import Image + +from windup_ai_engine.ports import JudgePort +from windup_app.server.orchestrator import quality_gate +from windup_common.models import JudgeVerdict +from windup_framework.config.provider import AIProviderSettings +from windup_framework.config.quality_gate import QualityGateSettings +from windup_framework.providers.judge import ( + JudgeResponseError, + SufyJudgeProvider, + _parse_verdict, +) + +GATEWAY = "https://gw.invalid/v1" + + +def _cfg() -> AIProviderSettings: + return AIProviderSettings(base_url=GATEWAY, api_key="test-key", judge_model="judge-x") + + +def _png(subjects: int, size: int = 96) -> bytes: + """白底 + N 个不相接的黑块 —— "已知有几个主体"的合成图。""" + im = Image.new("RGB", (size, size), (255, 255, 255)) + px = im.load() + for n in range(subjects): + left = 8 + n * 40 + for x in range(left, left + 20): + for y in range(20, size - 20): + px[x, y] = (0, 0, 0) + buf = io.BytesIO() + im.save(buf, "PNG") + return buf.getvalue() + + +def _count_subjects(png: bytes) -> int: + """数图里有几段互不相接的暗列 —— 假模型的"视觉"。""" + im = Image.open(io.BytesIO(png)).convert("L") + w, h = im.size + px = im.load() + dark = [any(px[x, y] < 128 for y in range(h)) for x in range(w)] + return sum(1 for x in range(w) if dark[x] and not (x and dark[x - 1])) + + +def _oracle_handler(seen: list[dict]): + """假模型:记下请求体,数**第二张**图(生成帧)的主体数,按契约回一段 JSON。""" + + def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + seen.append(body) + parts = body["messages"][0]["content"] + images = [p for p in parts if p["type"] == "image_url"] + frame = base64.b64decode(images[1]["image_url"]["url"].split(",", 1)[1]) + answer = { + "subject_count": _count_subjects(frame), + "foreign_objects": [], + "action_matches": True, + "clipped": False, + } + return httpx.Response(200, json={ + "choices": [{"message": {"content": json.dumps(answer)}}] + }) + + return handler + + +def _provider(monkeypatch, handler) -> SufyJudgeProvider: + """把 provider 的 client 换成走 MockTransport 的,保留它自己组的 headers / base_url。""" + provider = SufyJudgeProvider(config=_cfg()) + real = httpx.Client + + def factory(**kwargs): + kwargs["transport"] = httpx.MockTransport(handler) + return real(**kwargs) + + monkeypatch.setattr("windup_framework.providers.sufy.httpx.Client", factory) + return provider + + +# ── 控制样本:已知答案的合成图 ────────────────────────────────────────────── + + +def test_control_sample_two_subjects(monkeypatch): + """喂已知双主体的合成图 → subject_count ≥ 2。""" + seen: list[dict] = [] + provider = _provider(monkeypatch, _oracle_handler(seen)) + verdict = provider.judge(_png(2), _png(1), "walk") + assert verdict.subject_count >= 2 + + +def test_control_sample_single_subject(monkeypatch): + """喂已知单主体的合成图 → subject_count == 1。""" + seen: list[dict] = [] + provider = _provider(monkeypatch, _oracle_handler(seen)) + verdict = provider.judge(_png(1), _png(1), "walk") + assert verdict.subject_count == 1 + + +def test_control_sample_fixture_itself_is_discriminating(): + """先验仪器:合成图本身若数不出差别,上面两条就是同一个断言跑了两遍。""" + assert _count_subjects(_png(1)) == 1 + assert _count_subjects(_png(2)) == 2 + + +# ── 请求形状 ──────────────────────────────────────────────────────────────── + + +def test_request_carries_both_images_and_demands_json(monkeypatch): + """母版在前、生成帧在后,两张都以 image_url 送出,且明确要了 JSON。""" + seen: list[dict] = [] + provider = _provider(monkeypatch, _oracle_handler(seen)) + master, frame = _png(1), _png(2) + provider.judge(frame, master, "attack") + + body = seen[0] + assert body["model"] == "judge-x" + parts = body["messages"][0]["content"] + images = [p for p in parts if p["type"] == "image_url"] + assert len(images) == 2, "母版与生成帧都要送,少一张就答不了'多出来的物体'" + urls = [p["image_url"]["url"] for p in images] + assert all(u.startswith("data:image/png;base64,") for u in urls) + assert base64.b64decode(urls[0].split(",", 1)[1]) == master + assert base64.b64decode(urls[1].split(",", 1)[1]) == frame + + assert body["response_format"] == {"type": "json_object"} + prompt = next(p["text"] for p in parts if p["type"] == "text") + assert "JSON" in prompt + assert "attack" in prompt, "动作类别要进提示词,否则第三问无从判起" + assert body["temperature"] == 0 + + +def test_request_hits_chat_completions_with_bearer(monkeypatch): + seen_requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(request) + return httpx.Response(200, json={"choices": [{"message": {"content": json.dumps({ + "subject_count": 1, "foreign_objects": [], + "action_matches": True, "clipped": False, + })}}]}) + + provider = _provider(monkeypatch, handler) + provider.judge(_png(1), _png(1), "idle") + + request = seen_requests[0] + assert str(request.url) == f"{GATEWAY}/chat/completions" + assert request.headers["Authorization"] == "Bearer test-key" + + +# ── 读不出结论必须抛错,不得兜底成"通过" ──────────────────────────────────── + + +@pytest.mark.parametrize("text", [ + "sure, the frame looks great!", + "", + "[1, 2, 3]", + '{"subject_count": 1, "foreign_objects": []}', + '{"subject_count": "one", "foreign_objects": [], "action_matches": true, "clipped": false}', + '{"subject_count": true, "foreign_objects": [], "action_matches": true, "clipped": false}', + '{"subject_count": -1, "foreign_objects": [], "action_matches": true, "clipped": false}', + '{"subject_count": 1, "foreign_objects": "none", "action_matches": true, "clipped": false}', + '{"subject_count": 1, "foreign_objects": [7], "action_matches": true, "clipped": false}', + '{"subject_count": 1, "foreign_objects": [], "action_matches": "yes", "clipped": false}', + '{"subject_count": 1, "foreign_objects": [], "action_matches": true, "clipped": 0}', +]) +def test_unreadable_answer_raises(text): + with pytest.raises(JudgeResponseError): + _parse_verdict(text) + + +def test_provider_propagates_parse_failure(monkeypatch): + """整条通路上也不能被吞掉 —— 静默放行比拦错更糟。""" + provider = _provider(monkeypatch, lambda req: httpx.Response( + 200, json={"choices": [{"message": {"content": "looks fine to me"}}]} + )) + with pytest.raises(JudgeResponseError): + provider.judge(_png(1), _png(1), "walk") + + +def test_content_as_parts_array_is_accepted(monkeypatch): + """有的网关把正文包成 parts 数组;两种形状都要读得出来。""" + answer = json.dumps({ + "subject_count": 1, "foreign_objects": [], + "action_matches": True, "clipped": False, + }) + provider = _provider(monkeypatch, lambda req: httpx.Response(200, json={ + "choices": [{"message": {"content": [{"type": "text", "text": answer}]}}] + })) + assert provider.judge(_png(1), _png(1), "walk").subject_count == 1 + + +def test_unreadable_content_shape_raises(monkeypatch): + provider = _provider(monkeypatch, lambda req: httpx.Response(200, json={ + "choices": [{"message": {"content": {"unexpected": "shape"}}}] + })) + with pytest.raises(JudgeResponseError): + provider.judge(_png(1), _png(1), "walk") + + +def test_missing_content_raises(monkeypatch): + provider = _provider(monkeypatch, lambda req: httpx.Response(200, json={"choices": []})) + with pytest.raises(JudgeResponseError): + provider.judge(_png(1), _png(1), "walk") + + +def test_code_fence_is_stripped_not_a_fallback(): + verdict = _parse_verdict( + '```json\n{"subject_count": 2, "foreign_objects": ["sword"], ' + '"action_matches": false, "clipped": true}\n```' + ) + assert verdict.subject_count == 2 + assert verdict.foreign_objects == ("sword",) + assert verdict.action_matches is False + assert verdict.clipped is True + assert "sword" in verdict.raw, "原话要原样留着,复核判读对不对全靠它" + + +def test_verdict_has_no_score_field(): + """出参里不能有分数:一有分数就会有人拿它卡阈值,而那卡的是噪声。""" + assert not any( + re.search(r"score|rating|grade", name) + for name in JudgeVerdict.__dataclass_fields__ + ) + + +def test_provider_satisfies_judge_port(): + assert isinstance(SufyJudgeProvider(config=_cfg()), JudgePort) + + +def test_shared_face_keeps_each_capability_own_timeout(): + """共用管道不能抹掉各能力自己的超时:出图比一次问答慢得多。""" + from windup_framework.providers.sufy import _IMAGE_TIMEOUT_MULTIPLIER, SufyImageProvider + + config = _cfg() + for provider, want in ( + (SufyJudgeProvider(config=config), config.timeout), + (SufyImageProvider(config=config), config.timeout * _IMAGE_TIMEOUT_MULTIPLIER), + ): + client = provider._client() + try: + assert client.timeout.read == want + finally: + client.close() + + +# ── 闸口:shadow 记录、不拦截 ─────────────────────────────────────────────── + + +class _StubJudge: + def __init__(self, verdict: JudgeVerdict | Exception) -> None: + self._verdict = verdict + self.calls: list[tuple[bytes, bytes, str]] = [] + + def judge(self, frame: bytes, master: bytes, action: str) -> JudgeVerdict: + self.calls.append((frame, master, action)) + if isinstance(self._verdict, Exception): + raise self._verdict + return self._verdict + + +def _verdict(**kw) -> JudgeVerdict: + base = { + "subject_count": 1, "foreign_objects": (), "action_matches": True, + "clipped": False, "raw": "{}", + } + return JudgeVerdict(**{**base, **kw}) + + +_FRAMES = [b"f0", b"f1", b"f2"] +_MASTER = b"m" +_SHADOW = QualityGateSettings(enabled=True, enforce=False) +_ENFORCING = QualityGateSettings(enabled=True, enforce=True) + + +def test_gate_off_never_calls_the_judge(): + """默认关 = 一次付费调用都不发。""" + judge = _StubJudge(_verdict()) + decision = quality_gate.review( + judge, _FRAMES, _MASTER, "walk", config=QualityGateSettings(), + ) + assert decision is None + assert judge.calls == [] + + +def test_no_judge_injected_returns_none(): + assert quality_gate.review(None, _FRAMES, _MASTER, "walk", config=_SHADOW) is None + + +def test_shadow_records_problems_without_blocking(): + judge = _StubJudge(_verdict(subject_count=2, clipped=True)) + decision = quality_gate.review(judge, _FRAMES, _MASTER, "walk", config=_SHADOW) + assert decision.problems == ( + quality_gate.PROBLEM_MULTIPLE_SUBJECTS, quality_gate.PROBLEM_CLIPPED, + ) + assert decision.blocked is False + payload = decision.as_payload() + assert payload["subject_count"] == 2 and payload["blocked"] is False + + +def test_clean_verdict_has_no_problems(): + decision = quality_gate.review( + _StubJudge(_verdict()), _FRAMES, _MASTER, "walk", config=_SHADOW, + ) + assert decision.problems == () + assert decision.blocked is False + + +@pytest.mark.parametrize("kw,expected", [ + ({"subject_count": 0}, quality_gate.PROBLEM_NO_SUBJECT), + ({"subject_count": 3}, quality_gate.PROBLEM_MULTIPLE_SUBJECTS), + ({"foreign_objects": ("chair",)}, quality_gate.PROBLEM_FOREIGN_OBJECTS), + ({"action_matches": False}, quality_gate.PROBLEM_ACTION_MISMATCH), + ({"clipped": True}, quality_gate.PROBLEM_CLIPPED), +]) +def test_each_question_maps_to_its_own_problem(kw, expected): + decision = quality_gate.review( + _StubJudge(_verdict(**kw)), _FRAMES, _MASTER, "walk", config=_SHADOW, + ) + assert decision.problems == (expected,) + + +def test_enforce_blocks_only_when_switched_on(): + judge = _StubJudge(_verdict(action_matches=False)) + assert quality_gate.review( + judge, _FRAMES, _MASTER, "walk", config=_ENFORCING, + ).blocked is True + + +def test_judge_failure_never_blocks_and_is_not_a_pass(): + """仪器坏了不许拦 —— 拦的是用户已付费的产物;但也不能记成"判了没问题"。""" + judge = _StubJudge(JudgeResponseError("判官没有返回 JSON")) + decision = quality_gate.review(judge, _FRAMES, _MASTER, "walk", config=_ENFORCING) + assert decision.blocked is False + assert decision.verdict is None + assert "JSON" in decision.error + payload = decision.as_payload() + assert payload["problems"] == [] and "error" in payload + assert "subject_count" not in payload, "没判出来就不能在结果里留下任何读数" + + +def test_middle_frame_is_judged(): + """只判一帧,而且不是首帧 —— 首帧最像母版,动作对不对在那里最看不出来。""" + judge = _StubJudge(_verdict()) + decision = quality_gate.review(judge, _FRAMES, _MASTER, "walk", config=_SHADOW) + assert decision.frame_index == 1 + assert len(judge.calls) == 1 + assert judge.calls[0] == (b"f1", _MASTER, "walk") + + +def test_empty_frames_is_not_judged(): + judge = _StubJudge(_verdict()) + assert quality_gate.review(judge, [], _MASTER, "walk", config=_SHADOW) is None + assert judge.calls == [] + + +# ── 接进编排:shadow 结论落进任务结果,交付不受影响 ──────────────────────── + + +@pytest.fixture +def session_factory(): + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + from sqlalchemy.pool import StaticPool + + from windup_framework.db import Base + + engine = create_engine( + "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool, + ) + Base.metadata.create_all(engine) + return sessionmaker(bind=engine) + + +class _SpyGenerator: + """出两帧 256×256 的桩引擎(不联网、不解码视频)。""" + + def generate(self, card, action, master, progress, canvas=None): + from windup_ai_engine.ports import ActionQuality, GeneratedAction + + w, h = canvas or (256, 256) + frame = _blank_png(w, h) + return GeneratedAction( + frames=[frame, frame], + durations=[100, 100], + quality=ActionQuality(motion_scale=1.0, dead_frames=(), loop_seam=None), + ) + + +def _blank_png(w: int, h: int) -> bytes: + buf = io.BytesIO() + Image.new("RGBA", (w, h), (0, 0, 0, 0)).save(buf, "PNG") + return buf.getvalue() + + +def _run_task(session_factory, judge, gate_enabled: bool, enforce: bool, monkeypatch): + from windup_app.server.orchestrator.executor import ActionTaskExecutor + from windup_app.server.orchestrator.model import ActionType, CharacterActionInput + from windup_app.server.orchestrator.service import AiGenerationService + + live = quality_gate.settings + monkeypatch.setattr(live, "enabled", gate_enabled) + monkeypatch.setattr(live, "enforce", enforce) + + executor = ActionTaskExecutor( + generator=_SpyGenerator(), + judge=judge, + upload=lambda _png: "https://cdn.example.invalid/f.png", + fetch_master=lambda _input: b"master-bytes", + session_factory=session_factory, + ) + action_input = CharacterActionInput( + character_id=1, action_type=ActionType.WALK, num_frames=2, + ) + service = AiGenerationService() + with session_factory() as s: + task_id = service.generate_character_action(s, user_id=1, input=action_input).id + s.commit() + executor.run_action_task(task_id, action_input) + with session_factory() as s: + return service.get_task(s, project_id=None, task_id=task_id) + + +def test_shadow_verdict_lands_in_task_result(session_factory, monkeypatch): + judge = _StubJudge(_verdict(subject_count=2)) + task = _run_task(session_factory, judge, True, False, monkeypatch) + + from windup_app.server.orchestrator.model import TaskStatus + + assert task.status is TaskStatus.COMPLETED, "shadow 期判官说有问题也照常交付" + assert len(task.result.frames) == 2 + assert task.result.judge["problems"] == [quality_gate.PROBLEM_MULTIPLE_SUBJECTS] + assert task.result.judge["blocked"] is False + assert judge.calls[0][1] == b"master-bytes", "判官要拿到母版才答得了'多出来的物体'" + + +def test_gate_disabled_leaves_no_reading_and_costs_nothing(session_factory, monkeypatch): + judge = _StubJudge(_verdict()) + task = _run_task(session_factory, judge, False, False, monkeypatch) + assert task.result.judge is None, "没判就该是 None,不能看起来像'判了没问题'" + assert judge.calls == [] + + +def test_enforce_fails_the_task(session_factory, monkeypatch): + task = _run_task( + session_factory, _StubJudge(_verdict(clipped=True)), True, True, monkeypatch, + ) + + from windup_app.server.orchestrator.model import TaskStatus + + assert task.status is TaskStatus.FAILED + assert quality_gate.PROBLEM_CLIPPED in task.error_message