执行时序与生命周期 (Execution and Lifecycle)

本篇文档系统梳理了算子节点可重写以深度介入工作流生命周期的全套回调钩子，以及处理长耗时异步任务（例如带轮询的第三方外部 API 集成）的标准工程架构范式。

---

节点生命周期回调钩子 (Lifecycle Callbacks)

所有生命周期回调均可在自定义算子中按需重写：

• allow_incoming_connection, allow_outgoing_connection：返回布尔值 bool，在物理连线即将建立时执行自定义类型与逻辑校验；
• after_incoming_connection, after_outgoing_connection：连线成功建立后的后置逻辑处理（例如根据连入端类型动态派生新端口）；
• after_incoming_connection_removed, after_outgoing_connection_removed：连线被用户拔除或注销后的清理响应；
• before_value_set：在参数数值被正式写入前触发，可返回修改后的数值（用于弃用参数迁移、数据清洗截断等）；
• after_value_set：响应参数数值变动（用于实现联动显隐、动态更新子参数下拉选项等）；
• validate_before_workflow_run, validate_before_node_run：在工作流或单节点执行前触发静态检查，返回 list[Exception] | None；
• on_griptape_event：实时捕获并处理底层工作流抛出的各类 Griptape 事件；
• initialize_spotlight：配置并初始化节点聚光灯 (Spotlight) 专属交互功能；
• get_next_control_output：返回 Parameter | None，用于动态决定控制流下一步的执行出口。

常用状态操纵辅助方法 (Helper Methods)

• hide_parameter_by_name(), show_parameter_by_name()：动态隐藏或唤醒指定名称的参数；
• append_value_to_parameter()：向列表等集合类型参数末尾追加数据项；
• publish_update_to_parameter()：向前端实时推送参数属性更新事件；
• show_message_by_name(), hide_message_by_name(), get_message_by_name_or_element_id()：动态控制 ParameterMessage 提示信息横幅的显隐与内容。

---

异步 API 集成工程范式 (Asynchronous API Integration)

对于长耗时计算任务，算子提供了两种防阻塞引擎的优雅并发路径：

1. 重写 async def aprocess()（官方强烈推荐）：引擎直接在自身的全局事件循环 (Event Loop) 上 await 调度它。原生异步 I/O 操作（例如 httpx 异步请求、await asyncio.sleep() 轮询）会与引擎其他任务无缝并发，不占用额外物理线程；
2. 重写 process() 并通过 yield 返回可调用对象 (AsyncResult)：每一个通过 yield 抛出的 Callable 会被引擎自动分发至后台多线程池中同步阻塞执行——引擎界面保持流畅响应，但计算代码本身依然保持同步阻塞与顺序执行。

若你正在从零开发全新集成，请始终无条件优先选用 aprocess() 与异步 I/O 架构；只有当你依赖现成老旧的同步第三方库（如 requests 或阻塞型官方 SDK）且无重构预算时，才考虑采用 yield 线程池范式。

---

1. 基于 aprocess() 的现代原生异步处理（推荐）

import asyncio

import httpx
from griptape_nodes.exe_types.node_types import ControlNode

POLLING_INTERVAL = 10  # 轮询间隔秒数 (建议参考供应商官方文档推荐值)
MAX_POLLING_ATTEMPTS = 60  # 最大轮询上限次数 (例如 10 分钟超时)


class MyAsyncNode(ControlNode):
    async def aprocess(self) -> None:
        """原生异步处理 API 请求流水线。"""
        try:
            # 1. 设置安全兜底默认值
            self._set_safe_defaults()

            # 2. 校验提取 API 密钥凭证
            api_key = self._validate_api_key()

            async with httpx.AsyncClient(timeout=60) as client:
                # 3. 异步提交计算任务
                task_id = await self._submit_task(client, api_key)

                # 4. 异步轮询任务执行成败
                result = await self._poll_for_completion(client, task_id, api_key)

            # 5. 将计算成果写入输出端口
            self.parameter_output_values["output"] = result

        except Exception as e:
            self._set_safe_defaults()
            self._log(f"任务执行失败: {e}")
            raise RuntimeError(f"{self.name}: {e}") from e

    async def _submit_task(self, client: httpx.AsyncClient, api_key: str) -> str:
        response = await client.post(
            "https://api.example.com/v1/tasks",
            json=self._build_payload(),
            headers={"Authorization": f"Bearer {api_key}"},
        )
        response.raise_for_status()
        return response.json()["task_id"]

    async def _poll_for_completion(self, client: httpx.AsyncClient, task_id: str, api_key: str) -> str:
        for attempt in range(MAX_POLLING_ATTEMPTS):
            await asyncio.sleep(POLLING_INTERVAL)  # 严禁在 aprocess 中调用阻塞的 time.sleep()！

            response = await client.get(
                "https://api.example.com/v1/query/task",
                params={"task_id": task_id},
                headers={"Authorization": f"Bearer {api_key}"},
            )
            response.raise_for_status()
            status_data = response.json()

            if status_data["status"] == "Success":
                return status_data["result"]
            if status_data["status"] == "Fail":
                error_msg = status_data.get("error_message", "未知错误")
                raise RuntimeError(f"异步任务失败: {error_msg}")
            # 状态为 "Processing", "Pending" 时继续在循环中休眠轮询

        raise RuntimeError(f"任务在 {MAX_POLLING_ATTEMPTS * POLLING_INTERVAL} 秒内未完成，触发超时中断")

核心设计准则：

• 重写 async def aprocess() 而非 process()——引擎底层会直接对其执行 await；
• 全程采用异步 I/O：使用 httpx.AsyncClient 发起网络请求，使用 await asyncio.sleep() 替代同步休眠；
• 切勿在 aprocess() 内部调用任何阻塞型函数（如 requests.*、time.sleep()），否则会导致整个引擎事件循环发生致命卡死——若不可避免必须调用第三方阻塞库，请通过 await asyncio.to_thread(blocking_fn) 将其委托至线程池；
• 基类默认的 aprocess() 内部封装了对 process() 的兼容调用，因此自定义节点仅需在 aprocess() 与 process() 中二选一重写。

---

2. 基于后台线程池的阻塞处理 (process() + yield)

针对使用同步阻塞库构建的旧逻辑，重写 process() 并 yield 一个 Callable 对象。引擎会在后台多线程池中同步运行该函数，并在完成后将返回值注入回生成器恢复执行——引擎主进程保持完全响应，但这本质上依然是同步阻塞执行：

from griptape_nodes.exe_types.node_types import ControlNode, AsyncResult


class MyBlockingNode(ControlNode):
    def process(self) -> AsyncResult | None:
        """将阻塞型繁重工作让渡给后台工作线程池执行。"""
        yield lambda: self._process()

    def _process(self) -> None:
        """主计算方法 (在后台物理线程中同步阻塞运行)。"""
        try:
            # 1. 设置安全兜底默认值
            self._set_safe_defaults()

            # 2. 校验 API 凭证
            api_key = self._validate_api_key()

            # 3. 阻塞提交任务
            task_id = self._submit_task(api_key)

            # 4. 阻塞轮询直至结束
            result = self._poll_for_completion(task_id, api_key)

            # 5. 写入输出端口
            self.parameter_output_values["output"] = result

        except Exception as e:
            self._set_safe_defaults()
            self._log(f"处理失败: {e}")
            raise RuntimeError(f"{self.name}: {str(e)}") from e

核心特征：

• process() 返回类型声明为 AsyncResult | None，并通过 yield 抛出无参 Callable；
• 抛出的每一个 Callable 在后台线程池中串行执行，生成器会按序接收其返回值；
• 适用于遗留同步代码资产（如 requests），新算子请一律转向 aprocess()。

---

3. 长周期异步任务的标准轮询模式 (Polling Pattern)

在接入长耗时生成类服务（如视频扩散模型生成、模型微调训练等）时，推荐采用标准“三步走”设计模式。以下以同步 requests 结合后台线程池为例（若使用 aprocess()，只需等价替换为 httpx.AsyncClient 与 await asyncio.sleep()）：

步骤 1：任务提交 (Task Submission)

def _submit_task(self, params: dict[str, Any], headers: dict[str, str]) -> dict[str, Any]:
    """提交任务并提取响应中的 task_id。"""
    payload = self._build_payload(params)

    response = requests.post(self.API_BASE_URL, json=payload, headers=headers, timeout=DEFAULT_TIMEOUT)
    response.raise_for_status()

    response_data = response.json()
    task_id = response_data.get("task_id")
    return response_data

步骤 2：状态轮询 (Status Polling)

POLLING_INTERVAL = 10  # 秒 (根据服务商 API 规范配置)
MAX_POLLING_ATTEMPTS = 60  # 最大轮询次数


def _poll_for_completion(self, task_id: str, headers: dict[str, str]) -> str | None:
    """持续轮询任务执行状态，并在成功后返回最终结果标识符。"""
    query_url = "https://api.example.com/v1/query/task"

    for attempt in range(MAX_POLLING_ATTEMPTS):
        time.sleep(POLLING_INTERVAL)  # 每次轮询前的安全休眠

        response = requests.get(
            query_url,
            headers=headers,
            params={"task_id": task_id},  # 优先采用 Query 参数传递 task_id
            timeout=DEFAULT_TIMEOUT,
        )
        response.raise_for_status()

        status_data = response.json()
        status = status_data.get("status")

        self._log(f"轮询第 {attempt + 1} 次: 当前状态 = {status}")

        if status == "Success":
            file_id = status_data.get("file_id")
            return file_id
        elif status == "Fail":
            error_msg = status_data.get("error_message", "未知错误")
            raise RuntimeError(f"任务执行失败: {error_msg}")
        # 若为 "Processing"、"Pending" 等过渡状态，继续循环休眠

    raise RuntimeError(f"任务在 {MAX_POLLING_ATTEMPTS * POLLING_INTERVAL} 秒内未完成，触发超时")

步骤 3：资产提取与下载 (Result Retrieval)

def _retrieve_result(self, file_id: str, headers: dict[str, str]) -> str:
    """根据任务产出的结果 ID 获取真实文件可下载 URL。"""
    retrieve_url = "https://api.example.com/v1/files/retrieve"

    response = requests.get(retrieve_url, headers=headers, params={"file_id": file_id}, timeout=DEFAULT_TIMEOUT)
    response.raise_for_status()

    response_data = response.json()
    download_url = response_data.get("file", {}).get("download_url")

    return download_url

关键设计审查项：

• 严格遵循服务商官方建议的轮询频率（通常为 5~10 秒/次）；
• 设置合理的最大轮询重试上限，杜绝死循环挂死；
• 明确区分 Query 查询参数与 Path 路径参数（以官方文档规范为准）；
• 完整覆盖所有业务状态：Success、Fail、Processing、Pending；
• 输出详尽的轮询日志以便用户在画布上洞悉进度；
• 在捕获到任何异常时，主动将节点参数重置为安全默认值。

---

4. 基于连线状态动态切换 API 端点 (Dynamic Endpoint Selection)

当算子支持根据连线输入动态适配工作模式时（例如：接入了前置图像端口则切换为图生视频 Image-to-Video，未接图像端口则降级为纯文生视频 Text-to-Video），应在计算方法内部动态裁决目标端点 URL，而非僵化写死：

IMAGE2VIDEO_URL = "https://api.example.com/v1/videos/image2video"
TEXT2VIDEO_URL = "https://api.example.com/v1/videos/text2video"


def _process(self):
    image_data = self._get_image_data("start_frame")
    has_images = image_data is not None

    # 根据连线入参动态裁决端点
    if has_images:
        api_url = IMAGE2VIDEO_URL
    else:
        api_url = TEXT2VIDEO_URL

    payload = self._build_payload()
    if image_data:
        payload["image"] = image_data

    response = requests.post(api_url, headers=headers, json=payload, timeout=30)
    # 状态轮询亦应采用与之对应的 API 基地址
    poll_url = f"{api_url}/{task_id}"

这种模式确保了用户在只想进行文生图时无需强行连接多余的虚假图像输入，并保证了精准调用底层供应商的专用模型端点。

---

5. 图像资产转 Base64 编码规范 (Image Artifact to Base64)

下方的 ImageArtifact 兼容分支主要用于向下兼容来自旧版本工作流或未升级节点的历史数据。在开发新算子时，请一律采用 ParameterImage 声明轻量化的 ImageUrlArtifact——详见 参数负载大小优化 (Parameter Payload Size)。

⚠️ 致命陷阱：Localhost 本地网址处理 (Localhost URL Handling)

当向第三方公网云端 API 发送图像素材时，来自本地静态文件服务器的 ImageUrlArtifact 常常包含 localhost 或 127.0.0.1 协议头。外部云端服务器绝对无法访问开发者的本地回环网络！必须精准嗅探并在发起请求前将其无损转换为 Base64 编码的 Data URI：

import base64


def _get_image_data(self, image_artifact: ImageArtifact | ImageUrlArtifact) -> str:
    """将图像资产规整为可直接被云端 API 消费的公网 URL 或 Base64 Data URI。"""

    # 1. 处理现代的 ImageUrlArtifact
    if isinstance(image_artifact, ImageUrlArtifact):
        url = image_artifact.value

        # 若属于本地回环地址，外部公网云端无法直接读取，必须在本地下载并编码为 Base64
        if url.startswith(("http://localhost", "http://127.0.0.1", "https://localhost", "https://127.0.0.1")):
            self._log(f"检测到本地回环网络图像，正在转换为 Base64: {url[:100]}...")
            response = requests.get(url, timeout=30)
            response.raise_for_status()
            image_bytes = response.content

            # 探测真实的 MIME 类型
            mime_type = response.headers.get("content-type", "image/jpeg")
            if not mime_type.startswith("image/"):
                mime_type = "image/jpeg"

            base64_data = base64.b64encode(image_bytes).decode("utf-8")
            return f"data:{mime_type};base64,{base64_data}"

        # 真正可公网访问的外部 URL 直接放行直通
        self._log(f"直接使用外部公网 URL: {url[:100]}...")
        return url

    # 2. 向下兼容老旧的 ImageArtifact
    if isinstance(image_artifact, ImageArtifact):
        # 优先读取内置属性
        if hasattr(image_artifact, "base64") and hasattr(image_artifact, "mime_type"):
            base64_data = image_artifact.base64  # 原始 Base64 字符串 (无前缀)
            mime_type = image_artifact.mime_type  # 例如 'image/jpeg'

            # 检查是否已包含 data URI 前缀
            if base64_data.startswith("data:"):
                self._log("使用 ImageArtifact.base64 (已包含 data URI 前缀)")
                return base64_data

            # 手动拼装标准 Data URI
            self._log(f"格式化 ImageArtifact.base64，MIME 类型: {mime_type}")
            return f"data:{mime_type};base64,{base64_data}"

        # 兜底：手动字节流提取
        self._log("兜底降级：手动提取字节流并进行 Base64 编码")
        if hasattr(image_artifact, "value") and hasattr(image_artifact.value, "read"):
            image_artifact.value.seek(0)
            image_bytes = image_artifact.value.read()
        elif hasattr(image_artifact, "data"):
            if isinstance(image_artifact.data, bytes):
                image_bytes = image_artifact.data
            elif hasattr(image_artifact.data, "read"):
                image_artifact.data.seek(0)
                image_bytes = image_artifact.data.read()
            else:
                raise ValueError("不支持的 ImageArtifact 数据格式")
        else:
            raise ValueError("不支持的 ImageArtifact 数据格式")

        # 利用 PIL 探测图像 MIME 类型
        mime_type = "image/jpeg"
        try:
            from PIL import Image
            from io import BytesIO

            img = Image.open(BytesIO(image_bytes))
            format_to_mime = {"JPEG": "image/jpeg", "PNG": "image/png", "WEBP": "image/webp"}
            mime_type = format_to_mime.get(img.format, "image/jpeg")
        except Exception:
            pass

        base64_data = base64.b64encode(image_bytes).decode("utf-8")
        return f"data:{mime_type};base64,{base64_data}"

    raise ValueError("不支持的未知资产类型")

标准参数定义：

Parameter(
    name="image_input",
    input_types=["ImageUrlArtifact", "ImageArtifact"],  # ImageArtifact 仅保留用于兼容旧版工作流输入
    type="ImageUrlArtifact",
    tooltip="输入图像 (本地文件路径或 URL)",
    ui_options={"clickable_file_browser": True},  # 开启点击弹出文件拾取器
)

---

6. 多图输入聚合校验 (Multi-Image Input Validation)

当算子接收多个图像参数时，编写一个结构严谨、指明出错端口的通用校验器：

def _validate_image(self, image_artifact: ImageArtifact | ImageUrlArtifact, param_name: str) -> list[Exception]:
    """校验单张图像，并在异常消息中清晰标明具体的参数名称。"""
    exceptions = []

    if isinstance(image_artifact, ImageArtifact):
        # 提取图像原始二进制字节流
        if hasattr(image_artifact, "value") and hasattr(image_artifact.value, "read"):
            image_artifact.value.seek(0)
            image_bytes = image_artifact.value.read()
            image_artifact.value.seek(0)
        else:
            return exceptions

        # 校验文件大小 (限制小于 20MB)
        size_mb = len(image_bytes) / (1024 * 1024)
        if size_mb >= 20:
            exceptions.append(ValueError(f"{self.name}: 参数 {param_name} 体积超出限制，必须 < 20MB (当前: {size_mb:.1f}MB)"))

        # 校验格式与物理几何分辨率
        try:
            from PIL import Image
            from io import BytesIO

            img = Image.open(BytesIO(image_bytes))

            if img.format not in ["JPEG", "PNG", "WEBP"]:
                exceptions.append(
                    ValueError(f"{self.name}: 参数 {param_name} 格式不合法，必须为 JPG, PNG 或 WebP (当前: {img.format})")
                )

            width, height = img.size
            short_edge = min(width, height)
            if short_edge <= 300:
                exceptions.append(
                    ValueError(f"{self.name}: 参数 {param_name} 图像分辨率过小，短边必须 > 300px (当前: {short_edge}px)")
                )
        except ImportError:
            self._log("未检测到 PIL 库，跳过图像深度几何校验")
        except Exception as e:
            self._log(f"校验参数 {param_name} 时发生解析异常: {e}")

    return exceptions


def validate_before_node_run(self) -> list[Exception] | None:
    """在节点运算前批量实施参数静态健壮性校验。"""
    exceptions = []

    # 独立校验首帧参数
    first_frame = self.get_parameter_value("first_frame_image")
    if first_frame:
        exceptions.extend(self._validate_image(first_frame, "first_frame_image"))

    # 独立校验尾帧参数
    last_frame = self.get_parameter_value("last_frame_image")
    if last_frame:
        exceptions.extend(self._validate_image(last_frame, "last_frame_image"))

    return exceptions if exceptions else None

---

7. 模型动态依赖联动管理 (Model-Dependent Parameter Management)

当算子内建了多种模型选项，且不同模型所支持的下游参数（分辨率、时长等）差异巨大时：

def after_value_set(self, parameter: Parameter, value: Any) -> None:
    """处理模型选择变动时的参数联动显隐与选项热更新。"""
    if parameter.name == "model":
        if value == "AdvancedModel":
            # 唤醒高级模型特有的参数
            self.show_parameter_by_name("advanced_option")

            # 动态替换分辨率下拉菜单为高级模型支持列表
            resolution_param = self.get_parameter_by_name("resolution")
            if resolution_param:
                for child in resolution_param.children:
                    if hasattr(child, "choices"):
                        child.choices = ADVANCED_MODEL_RESOLUTIONS
                        break
        else:
            # 切换为基础模型时隐藏并重置高级参数
            self.hide_parameter_by_name("advanced_option")

            resolution_param = self.get_parameter_by_name("resolution")
            if resolution_param:
                for child in resolution_param.children:
                    if hasattr(child, "choices"):
                        child.choices = STANDARD_RESOLUTIONS
                        break
                self.set_parameter_value("resolution", "720P")

    return super().after_value_set(parameter, value)

针对模型维度的运行前强校验：

def validate_before_node_run(self) -> list[Exception] | None:
    """在执行前严格拦截非法模型与参数组合。"""
    exceptions = []

    model = self.get_parameter_value("model")
    duration = self.get_parameter_value("duration")
    resolution = self.get_parameter_value("resolution")

    # 例如：10 秒超长生成仅限 AdvancedModel 且禁止搭配 4K
    if duration == 10:
        if model != "AdvancedModel":
            exceptions.append(ValueError(f"{self.name}: 10秒时长生成仅受 AdvancedModel 支持"))
        elif resolution == "4K":
            exceptions.append(ValueError(f"{self.name}: 暂不支持 10秒 与 4K 分辨率同时开启"))

    # 必填联动字段检查
    if model in ["ModelB", "ModelC"]:
        required_param = self.get_parameter_value("required_for_model_b_c")
        if not required_param:
            exceptions.append(ValueError(f"{self.name}: 当前选中的 {model} 必须填写相关附属参数"))

    return exceptions if exceptions else None

---

8. 废弃模型自动平滑迁移与横幅公告 (Deprecated Model Migration)

当上游 AI 模型供应商下架旧模型（例如预览版 Preview 模型被正式 GA 商业版取代）时，算子应能够自动静默迁移已存盘的历史工作流，同时以醒目横幅通知用户保存变更。此架构由三个组件协作构成：

1. DEPRECATED_MODELS 映射表：建立旧版模型标识符与新版替代模型的精准键值字典；
2. 隐藏的 ParameterMessage 节点元素：充当前端可交互、可关闭的信息公告通知栏；
3. before_value_set 生命周期钩子：在反序列化数据注入生效前，无损拦截并将旧模型偷梁换柱。

步骤 1：声明废弃映射表与当前支持模型

from griptape_nodes.exe_types.core_types import Parameter, ParameterMessage
from griptape_nodes.traits.button import Button

MODELS = [
    "veo-3.1-generate-001",
    "veo-3.1-fast-generate-001",
]

# 废弃模型字典与升级目标的映射表
DEPRECATED_MODELS: dict[str, str] = {
    "veo-3.1-generate-preview": "veo-3.1-generate-001",
    "veo-3.1-fast-generate-preview": "veo-3.1-fast-generate-001",
    "veo-3.0-generate-001": "veo-3.1-generate-001",
    "veo-2.0-generate-001": "veo-3.1-generate-001",
}

步骤 2：在 __init__ 中注册默认隐藏的公告横幅

def __init__(self, **kwargs):
    super().__init__(**kwargs)

    # ... 声明 model 参数 ...

    # 注册默认隐藏的废弃模型告警通知框
    self.add_node_element(
        ParameterMessage(
            name="model_deprecation_notice",
            title="模型废弃自动迁移提醒",
            variant="info",
            value="",
            traits={
                Button(
                    full_width=True,
                    on_click=lambda _, __: self.hide_message_by_name("model_deprecation_notice"),
                )
            },
            button_text="我知道了 (Dismiss)",
            hide=True,
        )
    )

步骤 3：在 before_value_set 拦截并平滑置换

before_value_set 在参数值正式生效前触发。这是完成数据置换的最佳时机，因为后续触发的 after_value_set（以及所有依赖模型的动态逻辑）将直接感知到合法的新模型：

def before_value_set(self, parameter: Parameter, value: Any) -> Any:
    """自动将废弃模型迁移至最新版本，并向用户呼出通知横幅。"""
    if parameter.name == "model" and value in DEPRECATED_MODELS:
        replacement = DEPRECATED_MODELS[value]
        message = self.get_message_by_name_or_element_id("model_deprecation_notice")
        if message is not None:
            message.value = (
                f"原选中的 '{value}' 模型已被云端下线废弃。"
                f"系统已自动将其升级替代为最新版本 '{replacement}'。"
                "请尽快存盘工作流以固化此变更。"
            )
            self.show_message_by_name("model_deprecation_notice")
        value = replacement

    return super().before_value_set(parameter, value)

步骤 4：在用户手动选择有效模型时折叠横幅

def after_value_set(self, parameter: Parameter, value: Any) -> None:
    if parameter.name == "model":
        if value not in DEPRECATED_MODELS:
            self.hide_message_by_name("model_deprecation_notice")

    return super().after_value_set(parameter, value)

---

9. 工业级全链路调试日志 (Debug Logging)

针对涉及复杂外部 HTTP 调用的算子，打印充分详实的结构化日志是迅速排查故障的定海神针：

• 记录全量 API 交互报文（包含提交入参、轮询状态、最终返回）；
• 监控 Payload 载荷体积（尤其是 base64 图像编码长度，防止意外超限）；
• 追踪分支执行路径（明晰当前究竟命中了哪个代码分支）；
• 捕获完整的云端错误响应体（包含 status_code 以及供应商返回的详细错误文本）。

# 监控载荷体积示例
def _log_request(self, payload: dict) -> None:
    if "first_frame_image" in payload:
        img_len = len(payload.get("first_frame_image", ""))
        self._log(f"first_frame_image 字符长度: {img_len} 字符 (~{img_len / 1024:.1f}KB)")

---

10. API 协议校验严谨性 (Documentation Verification)

在对接第三方云服务时，请严格核对供应商官方文档：

1. 模型大小写拼写：如 MiniMax-Hailuo-02 绝不能误写为 video-01；
2. API 路由结构：确认是 /v1/query/video_generation 还是路径参数 /v1/video_generation/{id}；
3. 入参传递形式：准确选用 Query 字符串参数还是 JSON Body 体参数；
4. 轮询推荐周期：尊重厂商官方限流频次，避免触发 429 Too Many Requests。