From 562b8ebb12db4138ad9559b7acd8f5d328343a40 Mon Sep 17 00:00:00 2001 From: xyh202131 Date: Thu, 27 Aug 2026 14:53:03 +0800 Subject: [PATCH 1/5] =?UTF-8?q?feat(video):=20=E6=8E=A5=E5=85=A5=20Agnes?= =?UTF-8?q?=20Video=202.5=20=E4=B8=BB=E6=A8=A1=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 3 + .../src/windup_framework/config/provider.py | 4 + .../src/windup_framework/gateway/registry.py | 1 + .../src/windup_framework/gateway/types.py | 2 + .../src/windup_framework/gateway/video.py | 29 +- .../providers/protocol/__init__.py | 3 + .../providers/protocol/agnes_video.py | 186 +++++++++++ .../src/windup_framework/providers/sufy.py | 50 ++- backend/tests/test_agnes_video_protocol.py | 290 ++++++++++++++++++ backend/tests/test_gateway_registry.py | 18 ++ backend/tests/test_gateway_video.py | 64 +++- 11 files changed, 631 insertions(+), 19 deletions(-) create mode 100644 backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py create mode 100644 backend/tests/test_agnes_video_protocol.py diff --git a/.env.example b/.env.example index f6430376..a3d307de 100644 --- a/.env.example +++ b/.env.example @@ -49,6 +49,9 @@ AI_VIDEO_FALLBACKS=kling-v2-6 # veo3.1 只对逗号分隔的这些用户 id 开放(组内做高质量素材用,不面向客户)。 # 空 = 谁都不能用。它按秒计费且比 kling 贵,且**永远不进兜底链**,只能由请求显式指定。 AI_VIDEO_VEO_USER_IDS= +# Agnes Video 2.5 走独立国际站；密钥仅配置在服务端环境变量中。 +AI_VIDEO_AGNES_BASE_URL=https://apihub.agnes-ai.com/v1 +AI_VIDEO_AGNES_API_KEY= # 留空即按型号查上游牌价（美元）；配了它就整条链共用这一个价，跨型号时会算错。 AI_IMAGE_UNIT_COST= AI_VIDEO_UNIT_COST_PER_SECOND= diff --git a/backend/packages/framework/src/windup_framework/config/provider.py b/backend/packages/framework/src/windup_framework/config/provider.py index 7eeb8f28..6ae0444c 100644 --- a/backend/packages/framework/src/windup_framework/config/provider.py +++ b/backend/packages/framework/src/windup_framework/config/provider.py @@ -49,6 +49,10 @@ class AIProviderSettings(BaseSettings): # 用途是组内做高质量素材,不面向客户。空值 = 没有任何人可用(默认)。 # 逗号分隔的用户 id,如 "1,7,12"。 video_veo_user_ids: str = "" + # Agnes 使用独立国际站凭证，不复用 Modelink 的全局 base_url/api_key。 + # Key 只从部署环境读取；为空时 provider 会在上传首帧和发请求之前拒绝建单。 + video_agnes_base_url: str = "https://apihub.agnes-ai.com/v1" + video_agnes_api_key: str = "" chat_fallbacks: str = "" image_fallbacks: str = "gemini-3.1-flash-image-preview" diff --git a/backend/packages/framework/src/windup_framework/gateway/registry.py b/backend/packages/framework/src/windup_framework/gateway/registry.py index 0ffe7238..6237040c 100644 --- a/backend/packages/framework/src/windup_framework/gateway/registry.py +++ b/backend/packages/framework/src/windup_framework/gateway/registry.py @@ -12,6 +12,7 @@ "kling-v2-6": Family.VIDEO_INPUT_REFERENCE, "kling-video-o1": Family.VIDEO_IMAGE_LIST, # 登记但不允许进 chain "veo3.1": Family.VIDEO_FAL_QUEUE, # 登记≠放行,还要在 USER_GATED_MODELS 的白名单里 + "agnes-video-2.5": Family.VIDEO_AGNES, } diff --git a/backend/packages/framework/src/windup_framework/gateway/types.py b/backend/packages/framework/src/windup_framework/gateway/types.py index faf290c4..4fb09960 100644 --- a/backend/packages/framework/src/windup_framework/gateway/types.py +++ b/backend/packages/framework/src/windup_framework/gateway/types.py @@ -20,6 +20,7 @@ class Family(str, Enum): VIDEO_INPUT_REFERENCE = "video.input_reference" VIDEO_IMAGE_LIST = "video.image_list" VIDEO_FAL_QUEUE = "video.fal_queue" + VIDEO_AGNES = "video.agnes" #: 每个 scene 允许出现哪些 family。链上混不同 family 是合法的 —— 兜底型号与主型号 @@ -34,6 +35,7 @@ class Family(str, Enum): Scene.CHARACTER_ACTION: frozenset({ Family.VIDEO_INPUT_REFERENCE, Family.VIDEO_FAL_QUEUE, + Family.VIDEO_AGNES, }), } diff --git a/backend/packages/framework/src/windup_framework/gateway/video.py b/backend/packages/framework/src/windup_framework/gateway/video.py index 0971d371..bcb54bf9 100644 --- a/backend/packages/framework/src/windup_framework/gateway/video.py +++ b/backend/packages/framework/src/windup_framework/gateway/video.py @@ -29,7 +29,7 @@ hash_bytes, hash_image_input, ) -from windup_framework.gateway.types import AdapterResult, NextStep, Scene +from windup_framework.gateway.types import AdapterResult, Family, NextStep, Scene _DEFAULT_RETRY_AFTER_S = 2.0 _SLEEP_CAP_S = 30.0 @@ -95,7 +95,9 @@ def poll_i2v( snap = adapter.inspect_job(job_id, model=model) if snap.ok and snap.job_status == "completed" and snap.edge_fingerprint: if hasattr(adapter, "download_completed"): - return adapter.download_completed(job_id, snap.edge_fingerprint) + return adapter.download_completed( + job_id, snap.edge_fingerprint, model=model + ) return snap return adapter.follow_job(job_id, model=model) @@ -331,6 +333,17 @@ def fail(http_status: int | None) -> None: has_job_id=has_job_id, ) has_next_route = route_index + 1 < len(routes) + has_next_model = i + 1 < len(models) + uses_isolated_model_credentials = ( + self._registry.family_of(model) is Family.VIDEO_AGNES + ) + if ( + uses_isolated_model_credentials + and bound_job_id is None + and has_next_model + and step in (NextStep.OPEN_AGGREGATOR, NextStep.FALLBACK_KEY) + ): + step = NextStep.FALLBACK if step is NextStep.FAIL and bound_job_id is None: tier_step = budget.tier_b_escalation( error_type, @@ -418,13 +431,15 @@ def fail(http_status: int | None) -> None: and error_type is not ModelErrorType.UPSTREAM_FAILED ): fail(last_http_status) + allowed_pre_submit_fallbacks = { + ModelErrorType.RATE_LIMIT, + ModelErrorType.MODEL_NOT_FOUND, + } + if uses_isolated_model_credentials: + allowed_pre_submit_fallbacks.add(ModelErrorType.UNREACHED) if ( bound_job_id is None - and error_type - not in ( - ModelErrorType.RATE_LIMIT, - ModelErrorType.MODEL_NOT_FOUND, - ) + and error_type not in allowed_pre_submit_fallbacks ): fail(last_http_status) fallback_used = True diff --git a/backend/packages/framework/src/windup_framework/providers/protocol/__init__.py b/backend/packages/framework/src/windup_framework/providers/protocol/__init__.py index e836f5d3..e0d7da0c 100644 --- a/backend/packages/framework/src/windup_framework/providers/protocol/__init__.py +++ b/backend/packages/framework/src/windup_framework/providers/protocol/__init__.py @@ -1,3 +1,4 @@ +from .agnes_video import AGNES_VIDEO_25, AgnesVideoProtocol from .fal_queue import ( FAL_I2V_ENDPOINTS, FalQueueVideoProtocol, @@ -15,6 +16,8 @@ from .types import HttpCall, JobProtocol, VideoRequest __all__ = [ + "AGNES_VIDEO_25", + "AgnesVideoProtocol", "FAL_I2V_ENDPOINTS", "FAL_IMAGE_ENDPOINTS", "IMAGE_LIST_MODELS", diff --git a/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py b/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py new file mode 100644 index 00000000..f8c608aa --- /dev/null +++ b/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py @@ -0,0 +1,186 @@ +"""Agnes Video 2.5 的 OpenAI Videos 兼容协议面。""" + +from __future__ import annotations + +import json +import re +from urllib.parse import urlencode, urlparse + +import httpx + +from windup_common.enums.model import ModelErrorType +from windup_framework.gateway.types import AdapterResult + +from .openai_video import http_error, json_object +from .types import HttpCall, VideoRequest + +AGNES_VIDEO_25 = "agnes-video-2.5" +AGNES_OUTPUT_SIZE = "720P" +_RATIO_BY_REDUCED = { + (7, 3): "21:9", # 文档给出的 720P 像素是 1680x720，约分后为 7:3。 + (16, 9): "16:9", + (4, 3): "4:3", + (1, 1): "1:1", + (3, 4): "3:4", + (9, 16): "9:16", +} + + +def agnes_aspect_ratio(size: str) -> str: + """内部 ``WIDTHxHEIGHT`` 画布转 Agnes 白名单比例；不支持的比例在建单前拒绝。""" + matched = re.fullmatch(r"(\d+)x(\d+)", size.strip()) + if matched is None: + raise ValueError(f"视频画布必须是 WIDTHxHEIGHT，收到 {size!r}") + width, height = (int(part) for part in matched.groups()) + if width <= 0 or height <= 0: + raise ValueError(f"视频画布宽高必须为正数，收到 {size!r}") + import math + + divisor = math.gcd(width, height) + reduced = (width // divisor, height // divisor) + ratio = _RATIO_BY_REDUCED.get(reduced) + if ratio is None: + allowed = ", ".join(sorted(_RATIO_BY_REDUCED.values())) + raise ValueError( + f"Agnes Video 2.5 不支持画幅 {reduced[0]}:{reduced[1]}；可用值：{allowed}" + ) + return ratio + + +def _public_url(value: str | None) -> str: + parsed = urlparse(value or "") + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + raise ValueError("Agnes 首帧必须是可公开访问的 http(s) 公网 URL") + return value or "" + + +def _error_text(value: object) -> str: + if isinstance(value, dict): + message = value.get("message") + if message: + return str(message) + return json.dumps(value, ensure_ascii=False, sort_keys=True) + return str(value or "") + + +class AgnesVideoProtocol: + """首帧 URL 建单，随后用 ``video_id + model_name`` 轮询结果。""" + + def __init__( + self, + api_key: str, + base_url: str = "https://apihub.agnes-ai.com/v1", + ) -> None: + self._key = api_key + parsed = urlparse(base_url) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError(f"Agnes Base URL 无效：{base_url!r}") + self._poll_root = f"{parsed.scheme}://{parsed.netloc}" + + @property + def _headers(self) -> dict[str, str]: + return {"Authorization": f"Bearer {self._key}"} + + def build_submit(self, req: VideoRequest) -> HttpCall: + if req.model != AGNES_VIDEO_25: + raise ValueError( + f"Agnes 协议只接受模型 {AGNES_VIDEO_25}，收到 {req.model!r}" + ) + if not 4 <= req.seconds <= 12: + raise ValueError(f"Agnes 视频时长必须在 4 到 12 秒之间，收到 {req.seconds}") + body = { + "model": AGNES_VIDEO_25, + "prompt": req.prompt, + "seconds": str(req.seconds), + "mode": "keyframe", + "size": AGNES_OUTPUT_SIZE, + "aspect_ratio": agnes_aspect_ratio(req.size), + "first_frame": _public_url(req.first_frame_url), + "n": 1, + } + return HttpCall(method="POST", path="/videos", headers=self._headers, body=body) + + def parse_submit(self, resp: httpx.Response) -> AdapterResult: + if not 200 <= resp.status_code < 300: + return http_error(resp) + payload = json_object(resp) + if payload is None: + return AdapterResult( + ok=False, + error_type=ModelErrorType.INVALID_RESPONSE, + http_status=resp.status_code, + edge_fingerprint="响应不是 JSON 对象", + ) + video_id = payload.get("video_id") + if not video_id: + return AdapterResult( + ok=False, + error_type=ModelErrorType.INVALID_RESPONSE, + http_status=resp.status_code, + edge_fingerprint="响应没有 video_id", + ) + return AdapterResult( + ok=True, + job_id=str(video_id), + maybe_billed=True, + http_status=resp.status_code, + job_status=str(payload.get("status") or "queued").lower(), + ) + + def build_poll(self, job_id: str) -> HttpCall: + query = urlencode({"video_id": job_id, "model_name": AGNES_VIDEO_25}) + return HttpCall( + method="GET", + path=f"{self._poll_root}/agnesapi?{query}", + headers=self._headers, + ) + + def parse_poll(self, resp: httpx.Response, job_id: str) -> AdapterResult: + if not 200 <= resp.status_code < 300: + return http_error(resp, job_id=job_id, phase="follow") + payload = json_object(resp) + if payload is None: + return AdapterResult( + ok=False, + error_type=ModelErrorType.INVALID_RESPONSE, + http_status=resp.status_code, + job_id=job_id, + maybe_billed=True, + edge_fingerprint="轮询响应不是 JSON 对象", + ) + status = str(payload.get("status") or "").lower() + if status == "completed": + metadata = payload.get("metadata") + url = metadata.get("url") if isinstance(metadata, dict) else None + return AdapterResult( + ok=True, + job_id=job_id, + maybe_billed=True, + http_status=resp.status_code, + job_status=status, + result_url=str(url) if url else None, + ) + if status in {"failed", "cancelled"}: + return AdapterResult( + ok=False, + error_type=ModelErrorType.UPSTREAM_FAILED, + job_id=job_id, + maybe_billed=True, + http_status=resp.status_code, + job_status=status, + edge_fingerprint=_error_text(payload.get("error")), + ) + return AdapterResult( + ok=False, + job_id=job_id, + maybe_billed=True, + http_status=resp.status_code, + job_status=status or "in_progress", + ) + + def build_fetch(self, job_id: str) -> HttpCall | None: + del job_id + return None + + def parse_fetch(self, resp: httpx.Response, job_id: str) -> AdapterResult: + return self.parse_poll(resp, job_id) diff --git a/backend/packages/framework/src/windup_framework/providers/sufy.py b/backend/packages/framework/src/windup_framework/providers/sufy.py index 13584931..d763a4d2 100644 --- a/backend/packages/framework/src/windup_framework/providers/sufy.py +++ b/backend/packages/framework/src/windup_framework/providers/sufy.py @@ -44,6 +44,7 @@ from .interfaces import FirstFrameUploader, ImageProvider, VideoProvider from .protocol import HttpCall, VideoRequest +from .protocol.agnes_video import AGNES_VIDEO_25, AgnesVideoProtocol from .protocol.fal_queue import VeoQueueVideoProtocol from .protocol.image_faces import FalQueueImageFace, OpenAIImagesFace from .protocol.openai_video import OpenAIVideoProtocol, fit_first_frame @@ -123,6 +124,11 @@ def _protocol_for(self, model: str | None): return VeoQueueVideoProtocol( self._cfg.api_key, base_url=self._cfg.normalized_base_url ) + if FAMILIES.get(model or "") is Family.VIDEO_AGNES: + return AgnesVideoProtocol( + self._cfg.video_agnes_api_key, + base_url=self._cfg.video_agnes_base_url, + ) return OpenAIVideoProtocol(self._cfg.api_key) def _first_frame_url(self, first_frame: bytes, size: str) -> str: @@ -133,18 +139,33 @@ def _first_frame_url(self, first_frame: bytes, size: str) -> str: """ if self._uploader is None: raise RuntimeError( - "veo 首帧只吃公网 URL,但本 provider 没有注入 uploader;" + "当前视频模型首帧只吃公网 URL,但本 provider 没有注入 uploader;" "组装层要传 FirstFrameUploader 进来" ) return self._uploader.upload(fit_first_frame(first_frame, size), "image/jpeg") - def _client(self) -> httpx.Client: + def _client(self, model: str | None = None) -> httpx.Client: + from windup_framework.gateway.registry import FAMILIES + from windup_framework.gateway.types import Family + + if FAMILIES.get(model or "") is Family.VIDEO_AGNES: + base_url = self._cfg.video_agnes_base_url.rstrip("/") + api_key = self._cfg.video_agnes_api_key + else: + base_url = self._cfg.normalized_base_url + api_key = self._cfg.api_key return httpx.Client( - base_url=self._cfg.normalized_base_url, - headers={"Authorization": f"Bearer {self._cfg.api_key}"}, + base_url=base_url, + headers={"Authorization": f"Bearer {api_key}"}, timeout=self._cfg.timeout, ) + def _client_for(self, model: str | None): + """Agnes 取专属凭证；其余型号保留既有无参 ``_client()`` 调用契约。""" + if model == AGNES_VIDEO_25: + return self._client(model) + return self._client() + def submit_video( self, first_frame: bytes, @@ -154,6 +175,13 @@ def submit_video( model: str, ) -> AdapterResult: """一次 POST 建单。成功: ok=True, job_id, body=b"", maybe_billed=True。""" + if model == AGNES_VIDEO_25 and not self._cfg.video_agnes_api_key.strip(): + return AdapterResult( + ok=False, + error_type=ModelErrorType.UNREACHED, + maybe_billed=False, + edge_fingerprint="建单前被拒: AI_VIDEO_AGNES_API_KEY 未配置", + ) protocol = self._protocol_for(model) req = VideoRequest( model=model, @@ -163,7 +191,7 @@ def submit_video( mode=self._mode, first_frame=first_frame, ) - if isinstance(protocol, VeoQueueVideoProtocol): + if isinstance(protocol, (VeoQueueVideoProtocol, AgnesVideoProtocol)): # 只有这一面需要先上传、且有一组建单前的硬断言,所以 try 只包住它 —— # 包住 kling 那条会顺手改掉一条跑在线上的路径的失败语义,而那不是本次要动的东西。 # 失败一律记 maybe_billed=False:这一步还没碰上游,和"发出去了但不知道结果" @@ -181,7 +209,7 @@ def submit_video( ) else: call = protocol.build_submit(req) - with self._client() as client: + with self._client_for(model) as client: try: resp = client.request( call.method, call.path, json=call.body, headers=dict(call.headers) @@ -200,7 +228,7 @@ def inspect_job(self, job_id: str, model: str | None = None) -> AdapterResult: veo 的单据在那条路径上是 404,所以异步轮询那条链必须把型号一路带下来。 """ protocol = self._protocol_for(model) - with self._client() as client: + with self._client_for(model) as client: resp = _poll_get(client, protocol.build_poll(job_id)) parsed = protocol.parse_poll(resp, job_id) if parsed.error_type is not None or not parsed.ok: @@ -281,14 +309,16 @@ def with_poll( poll_ms=poll_ms, poll_count=poll_count, ) - downloaded = self.download_completed(job_id, url) + downloaded = self.download_completed(job_id, url, model=model) return with_poll(downloaded, download_ms=downloaded.download_ms) - def download_completed(self, job_id: str, url: str) -> AdapterResult: + def download_completed( + self, job_id: str, url: str, model: str | None = None + ) -> AdapterResult: """按 inspect 拿到的 URL 下载 mp4,不轮询。""" try: download_t0 = time.monotonic() - with self._client() as client: + with self._client_for(model) as client: body = _download(client, url) download_ms = int((time.monotonic() - download_t0) * 1000) except RuntimeError as exc: diff --git a/backend/tests/test_agnes_video_protocol.py b/backend/tests/test_agnes_video_protocol.py new file mode 100644 index 00000000..518a0e0f --- /dev/null +++ b/backend/tests/test_agnes_video_protocol.py @@ -0,0 +1,290 @@ +import io +import json + +import httpx +import pytest + +from windup_common.enums.model import ModelErrorType +from windup_framework.config.provider import AIProviderSettings +from windup_framework.providers.protocol.agnes_video import AgnesVideoProtocol +from windup_framework.providers.protocol.types import VideoRequest +from windup_framework.providers.sufy import SufyVideoProvider + + +MODEL = "agnes-video-2.5" +PUBLIC_FRAME = "https://media.windup.xin/i2v/frame.jpg" + + +def _request(**overrides) -> VideoRequest: + values = { + "model": MODEL, + "prompt": "角色向右自然行走，固定镜头", + "seconds": 5, + "size": "1280x720", + "mode": "std", + "first_frame": b"jpeg", + "first_frame_url": PUBLIC_FRAME, + } + values.update(overrides) + return VideoRequest(**values) + + +def _protocol() -> AgnesVideoProtocol: + return AgnesVideoProtocol("agnes-secret") + + +def test_agnes_submit_uses_keyframe_contract_and_public_first_frame(): + """防止把现有 Kling 的 input_reference/std 请求形状发给 Agnes。""" + call = _protocol().build_submit(_request()) + + assert call.method == "POST" + assert call.path == "/videos" + assert call.headers == {"Authorization": "Bearer agnes-secret"} + assert call.body == { + "model": MODEL, + "prompt": "角色向右自然行走，固定镜头", + "seconds": "5", + "mode": "keyframe", + "size": "720P", + "aspect_ratio": "16:9", + "first_frame": PUBLIC_FRAME, + "n": 1, + } + + +@pytest.mark.parametrize( + ("size", "aspect_ratio"), + [ + ("720x720", "1:1"), + ("960x720", "4:3"), + ("720x960", "3:4"), + ("720x1280", "9:16"), + ("1680x720", "21:9"), + ], +) +def test_agnes_submit_maps_supported_canvas_ratios(size, aspect_ratio): + assert ( + _protocol().build_submit(_request(size=size)).body["aspect_ratio"] + == aspect_ratio + ) + + +def test_agnes_rejects_non_public_first_frame_before_spending(): + with pytest.raises(ValueError, match="公网 URL"): + _protocol().build_submit(_request(first_frame_url="data:image/jpeg;base64,abc")) + + +def test_agnes_rejects_unsupported_duration_before_spending(): + with pytest.raises(ValueError, match="4.*12"): + _protocol().build_submit(_request(seconds=13)) + + +def test_agnes_submit_tracks_video_id_instead_of_task_id(): + response = httpx.Response( + 200, + json={ + "id": "task-1", + "task_id": "task-1", + "video_id": "video-1", + "status": "queued", + }, + ) + + result = _protocol().parse_submit(response) + + assert result.ok + assert result.job_id == "video-1" + assert result.job_status == "queued" + + +def test_agnes_poll_always_includes_model_name(): + call = _protocol().build_poll("video-1") + + assert call.method == "GET" + assert call.path == ( + "https://apihub.agnes-ai.com/agnesapi" + "?video_id=video-1&model_name=agnes-video-2.5" + ) + + +def test_agnes_completed_result_reads_metadata_url(): + response = httpx.Response( + 200, + json={ + "video_id": "video-1", + "status": "completed", + "progress": 100, + "metadata": {"url": "https://cdn.agnes-ai.com/out.mp4"}, + }, + ) + + result = _protocol().parse_poll(response, "video-1") + + assert result.ok + assert result.job_status == "completed" + assert result.result_url == "https://cdn.agnes-ai.com/out.mp4" + + +def test_agnes_failed_result_is_upstream_failed_and_keeps_message(): + response = httpx.Response( + 200, + json={ + "video_id": "video-1", + "status": "failed", + "error": {"message": "Invalid reference media"}, + }, + ) + + result = _protocol().parse_poll(response, "video-1") + + assert result.error_type is ModelErrorType.UPSTREAM_FAILED + assert result.job_status == "failed" + assert "Invalid reference media" in result.edge_fingerprint + + +class _Uploader: + def __init__(self) -> None: + self.seen: list[tuple[bytes, str]] = [] + + def upload(self, first_frame: bytes, content_type: str) -> str: + self.seen.append((first_frame, content_type)) + return PUBLIC_FRAME + + +def _jpeg() -> bytes: + from PIL import Image + + buf = io.BytesIO() + Image.new("RGB", (32, 32), (80, 120, 160)).save(buf, "JPEG") + return buf.getvalue() + + +def _provider(handler, *, api_key="agnes-secret", uploader=None) -> SufyVideoProvider: + cfg = AIProviderSettings( + base_url="https://api.modelink.ai/v1", + api_key="modelink-secret", + video_model=MODEL, + video_agnes_base_url="https://apihub.agnes-ai.com/v1", + video_agnes_api_key=api_key, + ) + provider = SufyVideoProvider( + config=cfg, + model=MODEL, + uploader=uploader, + poll_interval=0.01, + first_poll_after=0.01, + ) + + def client(model=None): + agnes = model == MODEL + return httpx.Client( + base_url=( + "https://apihub.agnes-ai.com/v1" + if agnes + else "https://api.modelink.ai/v1" + ), + headers={ + "Authorization": f"Bearer {api_key if agnes else 'modelink-secret'}" + }, + transport=httpx.MockTransport(handler), + ) + + provider._client = client # type: ignore[method-assign] + return provider + + +def test_provider_uses_agnes_credentials_and_uploads_fitted_first_frame(): + seen: list[httpx.Request] = [] + uploader = _Uploader() + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response( + 200, + json={"video_id": "video-1", "task_id": "task-1", "status": "queued"}, + ) + + result = _provider(handler, uploader=uploader).submit_video( + _jpeg(), "向右走", 5, "1280x720", MODEL + ) + + assert result.ok and result.job_id == "video-1" + assert str(seen[0].url) == "https://apihub.agnes-ai.com/v1/videos" + assert seen[0].headers["Authorization"] == "Bearer agnes-secret" + assert json.loads(seen[0].content)["first_frame"] == PUBLIC_FRAME + assert len(uploader.seen) == 1 + assert uploader.seen[0][1] == "image/jpeg" + + +def test_provider_keeps_kling_on_modelink_credentials_and_data_uri(): + seen: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(200, json={"id": "kling-job"}) + + result = _provider(handler, uploader=_Uploader()).submit_video( + _jpeg(), "向右走", 5, "1280x720", "kling-v2-5-turbo" + ) + + assert result.ok and result.job_id == "kling-job" + assert str(seen[0].url) == "https://api.modelink.ai/v1/videos" + assert seen[0].headers["Authorization"] == "Bearer modelink-secret" + assert json.loads(seen[0].content)["input_reference"].startswith( + "data:image/jpeg;base64," + ) + + +def test_missing_agnes_key_is_rejected_before_upload_or_network(): + sent: list[httpx.Request] = [] + uploader = _Uploader() + + result = _provider( + lambda request: sent.append(request) or httpx.Response(500), + api_key="", + uploader=uploader, + ).submit_video(_jpeg(), "向右走", 5, "1280x720", MODEL) + + assert sent == [] + assert uploader.seen == [] + assert result.error_type is ModelErrorType.UNREACHED + assert result.maybe_billed is False + + +def test_agnes_inspect_uses_root_poll_endpoint_and_agnes_key(): + seen: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response( + 200, + json={ + "video_id": "video-1", + "status": "completed", + "metadata": {"url": "https://cdn.agnes-ai.com/out.mp4"}, + }, + ) + + result = _provider(handler, uploader=_Uploader()).inspect_job("video-1", MODEL) + + assert result.ok and result.job_status == "completed" + assert str(seen[0].url) == ( + "https://apihub.agnes-ai.com/agnesapi" + "?video_id=video-1&model_name=agnes-video-2.5" + ) + assert seen[0].headers["Authorization"] == "Bearer agnes-secret" + + +def test_same_origin_agnes_download_uses_agnes_key(): + seen: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(200, content=b"mp4") + + result = _provider(handler, uploader=_Uploader()).download_completed( + "video-1", "https://apihub.agnes-ai.com/files/out.mp4", model=MODEL + ) + + assert result.ok and result.body == b"mp4" + assert seen[0].headers["Authorization"] == "Bearer agnes-secret" diff --git a/backend/tests/test_gateway_registry.py b/backend/tests/test_gateway_registry.py index 7a0d16f5..c9ede03a 100644 --- a/backend/tests/test_gateway_registry.py +++ b/backend/tests/test_gateway_registry.py @@ -17,6 +17,24 @@ def test_default_chains(): assert r.family_of("kling-v2-6") is Family.VIDEO_INPUT_REFERENCE +def test_agnes_primary_can_fallback_to_kling(): + r = ModelRegistry.from_settings( + AIProviderSettings( + image_model="gemini-2.5-flash-image", + image_fallbacks="", + video_model="agnes-video-2.5", + video_fallbacks="kling-v2-5-turbo,kling-v2-6", + ) + ) + + assert r.chain(Scene.CHARACTER_ACTION) == ( + "agnes-video-2.5", + "kling-v2-5-turbo", + "kling-v2-6", + ) + assert r.family_of("agnes-video-2.5") is Family.VIDEO_AGNES + + def test_shipped_image_chain_is_gpt_image_2_then_gemini_flash(): """出厂默认必须是这两个型号,且它们分属两个协议面。 diff --git a/backend/tests/test_gateway_video.py b/backend/tests/test_gateway_video.py index 43798286..5127bb99 100644 --- a/backend/tests/test_gateway_video.py +++ b/backend/tests/test_gateway_video.py @@ -41,6 +41,64 @@ def _video_gw(adapter, circuit=None) -> VideoGateway: settings=cfg, ) + +def _agnes_gw(adapter, circuit=None) -> VideoGateway: + cfg = AIProviderSettings( + video_model="agnes-video-2.5", + video_fallbacks="kling-v2-5-turbo,kling-v2-6", + ) + return VideoGateway( + registry=ModelRegistry.from_settings(cfg), + adapter=adapter, + circuit=circuit or CircuitBreaker(cooldown_s=60), + settings=cfg, + ) + + +def test_agnes_rate_limit_falls_back_to_kling_after_retries(monkeypatch): + """Agnes 使用独立 key，耗尽重试后不能误走 Modelink 的 key 路由。""" + monkeypatch.setattr("windup_framework.gateway.video.time.sleep", lambda _: None) + rate = AdapterResult( + ok=False, + error_type=ModelErrorType.RATE_LIMIT, + http_status=429, + retry_after_s=0, + ) + adapter = FakeVideoAdapter( + submits={ + "agnes-video-2.5": [rate, rate, rate], + "kling-v2-5-turbo": [ + AdapterResult(ok=True, job_id="kling-job", maybe_billed=True) + ], + "kling-v2-6": [], + }, + follows={"kling-job": MP4}, + ) + + assert _agnes_gw(adapter).i2v(b"frame", "walk") == MP4.body + assert adapter.submit_models == ["agnes-video-2.5"] * 3 + ["kling-v2-5-turbo"] + + +def test_agnes_unreached_falls_back_to_kling_after_safe_retry(): + """Agnes 请求未到上游时安全重试一次，仍失败则启用 Kling。""" + adapter = FakeVideoAdapter( + submits={ + "agnes-video-2.5": [UNREACHED, UNREACHED], + "kling-v2-5-turbo": [ + AdapterResult(ok=True, job_id="kling-job", maybe_billed=True) + ], + "kling-v2-6": [], + }, + follows={"kling-job": MP4}, + ) + + assert _agnes_gw(adapter).i2v(b"frame", "walk") == MP4.body + assert adapter.submit_models == [ + "agnes-video-2.5", + "agnes-video-2.5", + "kling-v2-5-turbo", + ] + def test_submit_522_retries_once_does_not_open_second_job_on_fallback_model(): ad = FakeVideoAdapter( submits={"kling-v2-5-turbo": [UNREACHED, UNREACHED], "kling-v2-6": [ @@ -314,13 +372,15 @@ def inspect_job(self, job_id, model=None): edge_fingerprint="https://cdn.example.com/out.mp4", ) - def download_completed(self, job_id, url): + def download_completed(self, job_id, url, model=None): + self.followed.append(f"download:{model}") return AdapterResult(ok=True, body=b"mp4-bytes", job_id=job_id, job_status="completed") ad = _PollAdapter(submits={}, follows={}) - result = _video_gw(ad).poll_i2v("j-start") + result = _video_gw(ad).poll_i2v("j-start", model="kling-v2-5-turbo") assert result.ok assert result.body == b"mp4-bytes" + assert ad.followed == ["inspect:j-start", "download:kling-v2-5-turbo"] def test_poll_i2v_still_pending(): From c9129979dace91fc43f19ff9ed2aa96cf4fc51f9 Mon Sep 17 00:00:00 2001 From: xyh202131 Date: Thu, 27 Aug 2026 15:12:21 +0800 Subject: [PATCH 2/5] =?UTF-8?q?feat(video):=20=E6=94=AF=E6=8C=81=20Agnes?= =?UTF-8?q?=202.5=20Flash=20Token=20Plan?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/windup_framework/gateway/registry.py | 1 + .../providers/protocol/__init__.py | 9 ++- .../providers/protocol/agnes_video.py | 15 +++-- .../src/windup_framework/providers/sufy.py | 7 ++- backend/tests/test_agnes_video_protocol.py | 56 ++++++++++++++++--- backend/tests/test_gateway_registry.py | 18 ++++++ 6 files changed, 91 insertions(+), 15 deletions(-) diff --git a/backend/packages/framework/src/windup_framework/gateway/registry.py b/backend/packages/framework/src/windup_framework/gateway/registry.py index 6237040c..d0e73348 100644 --- a/backend/packages/framework/src/windup_framework/gateway/registry.py +++ b/backend/packages/framework/src/windup_framework/gateway/registry.py @@ -13,6 +13,7 @@ "kling-video-o1": Family.VIDEO_IMAGE_LIST, # 登记但不允许进 chain "veo3.1": Family.VIDEO_FAL_QUEUE, # 登记≠放行,还要在 USER_GATED_MODELS 的白名单里 "agnes-video-2.5": Family.VIDEO_AGNES, + "agnes-video-2.5-flash": Family.VIDEO_AGNES, } diff --git a/backend/packages/framework/src/windup_framework/providers/protocol/__init__.py b/backend/packages/framework/src/windup_framework/providers/protocol/__init__.py index e0d7da0c..93b7daeb 100644 --- a/backend/packages/framework/src/windup_framework/providers/protocol/__init__.py +++ b/backend/packages/framework/src/windup_framework/providers/protocol/__init__.py @@ -1,4 +1,9 @@ -from .agnes_video import AGNES_VIDEO_25, AgnesVideoProtocol +from .agnes_video import ( + AGNES_VIDEO_25, + AGNES_VIDEO_25_FLASH, + AGNES_VIDEO_MODELS, + AgnesVideoProtocol, +) from .fal_queue import ( FAL_I2V_ENDPOINTS, FalQueueVideoProtocol, @@ -17,6 +22,8 @@ __all__ = [ "AGNES_VIDEO_25", + "AGNES_VIDEO_25_FLASH", + "AGNES_VIDEO_MODELS", "AgnesVideoProtocol", "FAL_I2V_ENDPOINTS", "FAL_IMAGE_ENDPOINTS", diff --git a/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py b/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py index f8c608aa..f500d3bd 100644 --- a/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py +++ b/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py @@ -15,6 +15,8 @@ from .types import HttpCall, VideoRequest AGNES_VIDEO_25 = "agnes-video-2.5" +AGNES_VIDEO_25_FLASH = "agnes-video-2.5-flash" +AGNES_VIDEO_MODELS = frozenset({AGNES_VIDEO_25, AGNES_VIDEO_25_FLASH}) AGNES_OUTPUT_SIZE = "720P" _RATIO_BY_REDUCED = { (7, 3): "21:9", # 文档给出的 720P 像素是 1680x720，约分后为 7:3。 @@ -70,8 +72,13 @@ def __init__( self, api_key: str, base_url: str = "https://apihub.agnes-ai.com/v1", + *, + model: str = AGNES_VIDEO_25, ) -> None: + if model not in AGNES_VIDEO_MODELS: + raise ValueError(f"Agnes 视频型号无效：{model!r}") self._key = api_key + self._model = model parsed = urlparse(base_url) if parsed.scheme not in {"http", "https"} or not parsed.netloc: raise ValueError(f"Agnes Base URL 无效：{base_url!r}") @@ -82,14 +89,14 @@ def _headers(self) -> dict[str, str]: return {"Authorization": f"Bearer {self._key}"} def build_submit(self, req: VideoRequest) -> HttpCall: - if req.model != AGNES_VIDEO_25: + if req.model != self._model: raise ValueError( - f"Agnes 协议只接受模型 {AGNES_VIDEO_25}，收到 {req.model!r}" + f"Agnes 协议当前绑定模型 {self._model}，收到 {req.model!r}" ) if not 4 <= req.seconds <= 12: raise ValueError(f"Agnes 视频时长必须在 4 到 12 秒之间，收到 {req.seconds}") body = { - "model": AGNES_VIDEO_25, + "model": self._model, "prompt": req.prompt, "seconds": str(req.seconds), "mode": "keyframe", @@ -128,7 +135,7 @@ def parse_submit(self, resp: httpx.Response) -> AdapterResult: ) def build_poll(self, job_id: str) -> HttpCall: - query = urlencode({"video_id": job_id, "model_name": AGNES_VIDEO_25}) + query = urlencode({"video_id": job_id, "model_name": self._model}) return HttpCall( method="GET", path=f"{self._poll_root}/agnesapi?{query}", diff --git a/backend/packages/framework/src/windup_framework/providers/sufy.py b/backend/packages/framework/src/windup_framework/providers/sufy.py index d763a4d2..f540c54d 100644 --- a/backend/packages/framework/src/windup_framework/providers/sufy.py +++ b/backend/packages/framework/src/windup_framework/providers/sufy.py @@ -44,7 +44,7 @@ from .interfaces import FirstFrameUploader, ImageProvider, VideoProvider from .protocol import HttpCall, VideoRequest -from .protocol.agnes_video import AGNES_VIDEO_25, AgnesVideoProtocol +from .protocol.agnes_video import AGNES_VIDEO_MODELS, AgnesVideoProtocol from .protocol.fal_queue import VeoQueueVideoProtocol from .protocol.image_faces import FalQueueImageFace, OpenAIImagesFace from .protocol.openai_video import OpenAIVideoProtocol, fit_first_frame @@ -128,6 +128,7 @@ def _protocol_for(self, model: str | None): return AgnesVideoProtocol( self._cfg.video_agnes_api_key, base_url=self._cfg.video_agnes_base_url, + model=model or "", ) return OpenAIVideoProtocol(self._cfg.api_key) @@ -162,7 +163,7 @@ def _client(self, model: str | None = None) -> httpx.Client: def _client_for(self, model: str | None): """Agnes 取专属凭证；其余型号保留既有无参 ``_client()`` 调用契约。""" - if model == AGNES_VIDEO_25: + if model in AGNES_VIDEO_MODELS: return self._client(model) return self._client() @@ -175,7 +176,7 @@ def submit_video( model: str, ) -> AdapterResult: """一次 POST 建单。成功: ok=True, job_id, body=b"", maybe_billed=True。""" - if model == AGNES_VIDEO_25 and not self._cfg.video_agnes_api_key.strip(): + if model in AGNES_VIDEO_MODELS and not self._cfg.video_agnes_api_key.strip(): return AdapterResult( ok=False, error_type=ModelErrorType.UNREACHED, diff --git a/backend/tests/test_agnes_video_protocol.py b/backend/tests/test_agnes_video_protocol.py index 518a0e0f..331b3051 100644 --- a/backend/tests/test_agnes_video_protocol.py +++ b/backend/tests/test_agnes_video_protocol.py @@ -6,12 +6,16 @@ from windup_common.enums.model import ModelErrorType from windup_framework.config.provider import AIProviderSettings -from windup_framework.providers.protocol.agnes_video import AgnesVideoProtocol +from windup_framework.providers.protocol.agnes_video import ( + AGNES_VIDEO_25_FLASH, + AgnesVideoProtocol, +) from windup_framework.providers.protocol.types import VideoRequest from windup_framework.providers.sufy import SufyVideoProvider MODEL = "agnes-video-2.5" +FLASH_MODEL = "agnes-video-2.5-flash" PUBLIC_FRAME = "https://media.windup.xin/i2v/frame.jpg" @@ -107,6 +111,18 @@ def test_agnes_poll_always_includes_model_name(): ) +def test_token_plan_flash_uses_flash_model_for_submit_and_poll(): + protocol = AgnesVideoProtocol("agnes-secret", model=AGNES_VIDEO_25_FLASH) + + submit = protocol.build_submit(_request(model=FLASH_MODEL)) + poll = protocol.build_poll("video-flash-1") + + assert submit.body["model"] == FLASH_MODEL + assert poll.path.endswith( + "?video_id=video-flash-1&model_name=agnes-video-2.5-flash" + ) + + def test_agnes_completed_result_reads_metadata_url(): response = httpx.Response( 200, @@ -159,24 +175,26 @@ def _jpeg() -> bytes: return buf.getvalue() -def _provider(handler, *, api_key="agnes-secret", uploader=None) -> SufyVideoProvider: +def _provider( + handler, *, api_key="agnes-secret", uploader=None, model=MODEL +) -> SufyVideoProvider: cfg = AIProviderSettings( base_url="https://api.modelink.ai/v1", api_key="modelink-secret", - video_model=MODEL, + video_model=model, video_agnes_base_url="https://apihub.agnes-ai.com/v1", video_agnes_api_key=api_key, ) provider = SufyVideoProvider( config=cfg, - model=MODEL, + model=model, uploader=uploader, poll_interval=0.01, first_poll_after=0.01, ) def client(model=None): - agnes = model == MODEL + agnes = model in {MODEL, FLASH_MODEL} return httpx.Client( base_url=( "https://apihub.agnes-ai.com/v1" @@ -216,6 +234,28 @@ def handler(request: httpx.Request) -> httpx.Response: assert uploader.seen[0][1] == "image/jpeg" +def test_token_plan_flash_provider_uses_agnes_credentials(): + seen: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response( + 200, + json={"video_id": "video-flash-1", "status": "queued"}, + ) + + result = _provider( + handler, + uploader=_Uploader(), + model=FLASH_MODEL, + ).submit_video(_jpeg(), "向右走", 5, "1280x720", FLASH_MODEL) + + assert result.ok and result.job_id == "video-flash-1" + assert str(seen[0].url) == "https://apihub.agnes-ai.com/v1/videos" + assert seen[0].headers["Authorization"] == "Bearer agnes-secret" + assert json.loads(seen[0].content)["model"] == FLASH_MODEL + + def test_provider_keeps_kling_on_modelink_credentials_and_data_uri(): seen: list[httpx.Request] = [] @@ -235,7 +275,8 @@ def handler(request: httpx.Request) -> httpx.Response: ) -def test_missing_agnes_key_is_rejected_before_upload_or_network(): +@pytest.mark.parametrize("model", [MODEL, FLASH_MODEL]) +def test_missing_agnes_key_is_rejected_before_upload_or_network(model): sent: list[httpx.Request] = [] uploader = _Uploader() @@ -243,7 +284,8 @@ def test_missing_agnes_key_is_rejected_before_upload_or_network(): lambda request: sent.append(request) or httpx.Response(500), api_key="", uploader=uploader, - ).submit_video(_jpeg(), "向右走", 5, "1280x720", MODEL) + model=model, + ).submit_video(_jpeg(), "向右走", 5, "1280x720", model) assert sent == [] assert uploader.seen == [] diff --git a/backend/tests/test_gateway_registry.py b/backend/tests/test_gateway_registry.py index c9ede03a..68702d2f 100644 --- a/backend/tests/test_gateway_registry.py +++ b/backend/tests/test_gateway_registry.py @@ -35,6 +35,24 @@ def test_agnes_primary_can_fallback_to_kling(): assert r.family_of("agnes-video-2.5") is Family.VIDEO_AGNES +def test_token_plan_agnes_flash_primary_can_fallback_to_kling(): + r = ModelRegistry.from_settings( + AIProviderSettings( + image_model="gemini-2.5-flash-image", + image_fallbacks="", + video_model="agnes-video-2.5-flash", + video_fallbacks="kling-v2-5-turbo,kling-v2-6", + ) + ) + + assert r.chain(Scene.CHARACTER_ACTION) == ( + "agnes-video-2.5-flash", + "kling-v2-5-turbo", + "kling-v2-6", + ) + assert r.family_of("agnes-video-2.5-flash") is Family.VIDEO_AGNES + + def test_shipped_image_chain_is_gpt_image_2_then_gemini_flash(): """出厂默认必须是这两个型号,且它们分属两个协议面。 From 32932969685e8bf57b9941bc0f34935cafb9a89f Mon Sep 17 00:00:00 2001 From: xyh202131 Date: Thu, 27 Aug 2026 15:23:47 +0800 Subject: [PATCH 3/5] =?UTF-8?q?fix(video):=20=E7=AD=89=E5=BE=85=20Agnes=20?= =?UTF-8?q?=E6=88=90=E7=89=87=E5=9C=B0=E5=9D=80=E5=9B=9E=E5=A1=AB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../providers/protocol/agnes_video.py | 13 ++++++++++++- backend/tests/test_agnes_video_protocol.py | 19 +++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py b/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py index f500d3bd..c6603151 100644 --- a/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py +++ b/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py @@ -159,13 +159,24 @@ def parse_poll(self, resp: httpx.Response, job_id: str) -> AdapterResult: if status == "completed": metadata = payload.get("metadata") url = metadata.get("url") if isinstance(metadata, dict) else None + if not url: + # 线上偶尔先把状态推进到 completed，随后才回填 metadata.url。 + # 官方交付条件要求两者同时存在，因此这里继续轮询同一任务；若立即 + # 判 INVALID_RESPONSE，已经生成成功的任务将无法恢复且兜底会重复建单。 + return AdapterResult( + ok=False, + job_id=job_id, + maybe_billed=True, + http_status=resp.status_code, + job_status="in_progress", + ) return AdapterResult( ok=True, job_id=job_id, maybe_billed=True, http_status=resp.status_code, job_status=status, - result_url=str(url) if url else None, + result_url=str(url), ) if status in {"failed", "cancelled"}: return AdapterResult( diff --git a/backend/tests/test_agnes_video_protocol.py b/backend/tests/test_agnes_video_protocol.py index 331b3051..f9909081 100644 --- a/backend/tests/test_agnes_video_protocol.py +++ b/backend/tests/test_agnes_video_protocol.py @@ -141,6 +141,25 @@ def test_agnes_completed_result_reads_metadata_url(): assert result.result_url == "https://cdn.agnes-ai.com/out.mp4" +def test_agnes_completed_without_metadata_url_remains_pollable(): + response = httpx.Response( + 200, + json={ + "video_id": "video-1", + "status": "completed", + "progress": 100, + "metadata": None, + }, + ) + + result = _protocol().parse_poll(response, "video-1") + + assert not result.ok + assert result.error_type is None + assert result.job_status == "in_progress" + assert result.result_url is None + + def test_agnes_failed_result_is_upstream_failed_and_keeps_message(): response = httpx.Response( 200, From da4240caf2888fccf0597ce75db72b9d6c9dfc8f Mon Sep 17 00:00:00 2001 From: xyh202131 Date: Thu, 27 Aug 2026 16:00:55 +0800 Subject: [PATCH 4/5] =?UTF-8?q?fix(video):=20=E5=85=BC=E5=AE=B9=20Agnes=20?= =?UTF-8?q?Flash=20=E9=A1=B6=E5=B1=82=E6=88=90=E7=89=87=E5=9C=B0=E5=9D=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../providers/protocol/agnes_video.py | 7 +++++-- backend/tests/test_agnes_video_protocol.py | 20 +++++++++++++++++++ 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py b/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py index c6603151..73715942 100644 --- a/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py +++ b/backend/packages/framework/src/windup_framework/providers/protocol/agnes_video.py @@ -158,9 +158,12 @@ def parse_poll(self, resp: httpx.Response, job_id: str) -> AdapterResult: status = str(payload.get("status") or "").lower() if status == "completed": metadata = payload.get("metadata") - url = metadata.get("url") if isinstance(metadata, dict) else None + metadata_url = metadata.get("url") if isinstance(metadata, dict) else None + # Agnes Video 2.5 Flash 的线上完成响应会把成片地址直接放在 + # 顶层 url；标准模型仍使用文档约定的 metadata.url。 + url = metadata_url or payload.get("url") if not url: - # 线上偶尔先把状态推进到 completed，随后才回填 metadata.url。 + # 线上偶尔先把状态推进到 completed，随后才回填成片地址。 # 官方交付条件要求两者同时存在，因此这里继续轮询同一任务；若立即 # 判 INVALID_RESPONSE，已经生成成功的任务将无法恢复且兜底会重复建单。 return AdapterResult( diff --git a/backend/tests/test_agnes_video_protocol.py b/backend/tests/test_agnes_video_protocol.py index f9909081..ce811696 100644 --- a/backend/tests/test_agnes_video_protocol.py +++ b/backend/tests/test_agnes_video_protocol.py @@ -141,6 +141,26 @@ def test_agnes_completed_result_reads_metadata_url(): assert result.result_url == "https://cdn.agnes-ai.com/out.mp4" +def test_agnes_flash_completed_result_reads_top_level_url(): + response = httpx.Response( + 200, + json={ + "id": "video-flash-1", + "status": "completed", + "progress": 100, + "url": "https://cdn.agnes-ai.com/flash-out.mp4", + }, + ) + + result = AgnesVideoProtocol( + "agnes-secret", model=AGNES_VIDEO_25_FLASH + ).parse_poll(response, "video-flash-1") + + assert result.ok + assert result.job_status == "completed" + assert result.result_url == "https://cdn.agnes-ai.com/flash-out.mp4" + + def test_agnes_completed_without_metadata_url_remains_pollable(): response = httpx.Response( 200, From 1a116e70268b0ba140d69401f6616eff1f19ac90 Mon Sep 17 00:00:00 2001 From: xyh202131 Date: Thu, 27 Aug 2026 16:21:02 +0800 Subject: [PATCH 5/5] =?UTF-8?q?fix(video):=20=E9=9A=94=E7=A6=BB=20Agnes=20?= =?UTF-8?q?=E5=85=B1=E4=BA=AB=E7=86=94=E6=96=AD=E8=8C=83=E5=9B=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/windup_framework/gateway/video.py | 61 ++++++------------- backend/tests/test_gateway_video.py | 46 ++++++++++++++ 2 files changed, 66 insertions(+), 41 deletions(-) diff --git a/backend/packages/framework/src/windup_framework/gateway/video.py b/backend/packages/framework/src/windup_framework/gateway/video.py index bcb54bf9..60c3018c 100644 --- a/backend/packages/framework/src/windup_framework/gateway/video.py +++ b/backend/packages/framework/src/windup_framework/gateway/video.py @@ -60,6 +60,20 @@ def __init__(self, registry, adapter, circuit, settings, route_adapters=None) -> def _adapter_for(self, route: GatewayRoute): return lookup_adapter(self._route_adapters, route, self._adapter) + def _open_circuit_scope(self, route: GatewayRoute, model: str) -> str | None: + """返回会阻止本次型号提交的熔断层；Agnes 不受 Modelink 共享层影响。""" + isolated = self._registry.family_of(model) is Family.VIDEO_AGNES + if not isolated: + if self._circuit.is_open("aggregator"): + return "aggregator" + if self._circuit.is_open("base_url:" + route.base_url_id): + return "base_url" + if self._circuit.is_open(key_circuit_id(route)): + return "key" + if self._circuit.is_open("model:" + model): + return "model" + return None + def start_i2v( self, first_frame: bytes, @@ -141,50 +155,13 @@ def fail(http_status: int | None) -> None: f"http_status={http_status} error_type={err}" ) - if self._circuit.is_open("aggregator"): - model = models[0] if models else "" - route = routes[0] - self._emit( - AttemptTrace( - request_id=request_id, - scene=Scene.CHARACTER_ACTION, - model=model, - route=route, - attempt_index=seq.next_index(), - retry_count=0, - route_reason="skip_circuit_open", - outcome="failed", - circuit_scope="aggregator", - total_latency_ms=total_ms(), - maybe_billed=False, - detail=AttemptDetail( - input_hash=input_hash, - policy_next_step="fail", - upstream_reached="false", - ), - ) - ) - fail(None) - for route_index, route in enumerate(routes): - if self._circuit.is_open("base_url:" + route.base_url_id): - if route_index + 1 < len(routes): - fallback_used = True - route_reason_override = "base_url_unreached" - continue - fail(last_http_status) - if self._circuit.is_open(key_circuit_id(route)): - if route_index + 1 < len(routes): - fallback_used = True - route_reason_override = "key_rate_limit" - continue - fail(last_http_status) - adapter = self._adapter_for(route) switch_to_next_route = False for i, model in enumerate(models): model_index = start_i + i - if self._circuit.is_open("model:" + model): + open_scope = self._open_circuit_scope(route, model) + if open_scope is not None: fallback_used = True fallback_reason = "skip" self._emit( @@ -197,13 +174,15 @@ def fail(http_status: int | None) -> None: retry_count=0, route_reason="skip_circuit_open", outcome="failed", - circuit_scope="model", + circuit_scope=open_scope, fallback_used=fallback_used, total_latency_ms=total_ms(), maybe_billed=False, detail=AttemptDetail( input_hash=input_hash, - policy_next_step="fallback", + policy_next_step=( + "fallback" if open_scope == "model" else "fail" + ), upstream_reached="false", model_index=model_index, ), diff --git a/backend/tests/test_gateway_video.py b/backend/tests/test_gateway_video.py index 5127bb99..1f17e72f 100644 --- a/backend/tests/test_gateway_video.py +++ b/backend/tests/test_gateway_video.py @@ -99,6 +99,52 @@ def test_agnes_unreached_falls_back_to_kling_after_safe_retry(): "kling-v2-5-turbo", ] + +@pytest.mark.parametrize( + "shared_scope", + [ + "aggregator", + "base_url:primary", + "key:primary:primary.key0", + ], +) +def test_agnes_bypasses_shared_modelink_circuits(shared_scope): + circuit = CircuitBreaker(cooldown_s=60) + circuit.open(shared_scope) + adapter = FakeVideoAdapter( + submits={ + "agnes-video-2.5": [ + AdapterResult(ok=True, job_id="agnes-job", maybe_billed=True) + ], + "kling-v2-5-turbo": [], + "kling-v2-6": [], + }, + follows={"agnes-job": MP4}, + ) + + assert _agnes_gw(adapter, circuit=circuit).i2v(b"frame", "walk") == MP4.body + assert adapter.submit_models == ["agnes-video-2.5"] + + +def test_open_aggregator_still_blocks_kling_after_agnes_fails(): + circuit = CircuitBreaker(cooldown_s=60) + circuit.open("aggregator") + adapter = FakeVideoAdapter( + submits={ + "agnes-video-2.5": [UNREACHED, UNREACHED], + "kling-v2-5-turbo": [ + AdapterResult(ok=True, job_id="must-not-submit", maybe_billed=True) + ], + "kling-v2-6": [], + }, + follows={}, + ) + + with pytest.raises(RuntimeError, match="unreached"): + _agnes_gw(adapter, circuit=circuit).i2v(b"frame", "walk") + assert adapter.submit_models == ["agnes-video-2.5", "agnes-video-2.5"] + + def test_submit_522_retries_once_does_not_open_second_job_on_fallback_model(): ad = FakeVideoAdapter( submits={"kling-v2-5-turbo": [UNREACHED, UNREACHED], "kling-v2-6": [