physical-ai-event-video-generation

PAIDF 编排 — 事件视频生成

端到端运行事件视频生成 DAG：准备种子图像输入、使用 Cosmos3 进行图像转视频异常增强、自动标注（检测与跟踪、字幕生成、异常视觉 QA、人物属性视觉 QA、人物属性搜索）、生成异常数据集并检索结果。

DAG 选择

工作流会根据计算平台，从 airflow/dags/workflows/event_video_generation_dag/ 中为每个平台构建一个 DAG：

平台	DAG ID	清单文件
Kubernetes	event_video_generation_dag_k8s	event_video_generation_k8s_manifest.yaml

Kubernetes 是唯一已提交清单文件的平台，因此 event_video_generation_dag_k8s 是此仓库注册的唯一 DAG。只有清单文件存在时才会注册 DAG；清单文件缺失意味着该 DAG 不会出现在 Airflow 中，而不是已损坏。触发前，请先列出 Airflow 实际加载的 DAG，切勿使用列表中不存在的 DAG ID。

只有一个端到端流水线——没有仅生成或仅标注的 DAG 变体。如果用户要求生成视频但不进行自动标注，请告知他们已提交的 DAG 不支持这种流程，不要编造 DAG ID。

在 Airflow UI 中手动输入负载

如果用户想直接在 Airflow UI 中输入自己的负载，而不是让你构造并触发负载，你的任务仅限于引导他们进入 UI：确认控制器已就绪，确保 make port-forward 正在运行（请参阅 airflow-direct-api.md），并报告可访问的 URL。在这种情况下，不要生成负载、运行预检或自行触发运行——用户会在 UI 中完成这些操作。用户告知你已触发运行后，继续监控（见下文第 6 步）；你可以通过 Airflow API 查找该运行，无需知道他们使用的负载。

范围

构造任何负载之前，请向用户收集以下所有信息。 不要回退到仓库默认值、CI 负载，或任何硬编码的端点 URL 或存储桶路径。

必填项	字段	询问内容
始终需要	input_path	单张种子图像或种子图像目录的 S3（或 HTTP/HTTPS）URL
始终需要	output_directory	用于写入结果的可写 S3 URL
始终需要	服务模式	external（用户提供端点 URL）或 internal（DAG 在集群内部署服务）
外部模式	cosmos.vlm_service_url	VLM 推理端点的完整 HTTPS URL
外部模式	cosmos.llm_service_url	LLM 推理端点的完整 HTTPS URL
外部模式	cosmos.image2video_service_url	Cosmos3 图像转视频推理端点的完整 HTTPS URL
可选	max_images	从目录中处理的图像数量（默认值：10；0 或负数表示全部）
可选	cosmos.num_augmentation	每张图像生成的异常视频数（默认值：1）
可选	cosmos.variable_distribution	anomaly_type / env_type 抽样分布（请参阅 payload-contract.md）

如果用户未提供必填值，请在继续之前明确询问。不要编造或复用先前运行或已提交文件中的值。

触发运行前，务必执行以下就绪检查。 检查会在首次失败时立即停止；一旦失败，立刻转到环境设置技能。

执行任何检查之前，先建立集群连接。只能通过用户提供的凭据访问集群——仓库中绝不包含这些凭据。检查 shell 环境中是否已导出集群凭据文件路径；如果没有，请先向用户索取绝对路径，然后再运行任何集群命令。切勿假定路径或回退到磁盘上的任何默认值——完整流程请参阅 setup-and-preflight.md。 除非配置了不同的远程集群连接，否则控制器（Airflow）和 DAG 计算任务会在同一集群上运行。GPU 容量也在此集群上检查。

1. 控制器 Pod — 检查 Airflow 控制器 Pod（而非 DAG 任务 Pod）是否处于 Running 状态。 DAG 任务 Pod 处于 Pending 或 Failed 状态是正常的，不要误认为是控制器故障：

   kubectl get pods -n sdg-workflow -l "release=sdg-workflow-controller"
   

   所有匹配 release=sdg-workflow-controller 标签的 Pod 都必须处于 Running 状态。如果命名空间不存在，说明这是首次安装，应转到环境设置技能，不要继续诊断。

2. Airflow API — 仅当检查 1 通过时才检查。首先根据 Kubernetes ClusterIP 设置 AIRFLOW_URL（从主机始终可路由，无需端口转发）：

   AIRFLOW_URL="http://$(kubectl get svc -n sdg-workflow \
     sdg-workflow-controller-api-server \
     -o jsonpath='{.spec.clusterIP}'):8080"
   

   然后确认目标 DAG 已加载，且 is_paused: False。完整的身份验证和检查流程请参阅 airflow-direct-api.md。 如果 API 无法访问，请转到环境设置技能。

3. Pools — 仅当检查 2 通过时才检查。所需的池必须有空闲槽位：k8s_gpu_1、 default_pool，以及所选模式对应的 image2video 池 （外部模式使用 external_image2video_service_pool，内部模式使用 internal_image2video_service_pool）。

4. 计算集群 GPU — 检查集群（使用上文建立的集群连接）：

   kubectl get nodes \
     -o custom-columns='NAME:.metadata.name,GPU_ALLOC:.status.allocatable.nvidia\.com/gpu'
   # Also check pods already consuming GPUs — capacity ≠ availability on a shared cluster
   kubectl get pods -n sdg-workflow \
     --field-selector=status.phase=Running -o wide
   

   计算集群是共享的——其他用户的运行可能正在进行。报告 GPU 的空闲数和总数，而不只是可分配数。有两个相互独立的 GPU 来源，其中只有一个取决于模式：

   • 无论服务模式如何，都会运行本地模型推理的任务 Pod： detection_and_tracking、captioning 和 visual_qa 都使用 k8s_gpu_task 配置文件运行（各需 1 个 GPU）——外部模式和内部模式都需要这部分资源，因为这些任务会在 Pod 内推理，而不是调用端点。event_and_person_attribute_search 和 augmentation 使用 CPU 配置文件，不消耗 GPU。因此，外部模式至少需要三个 GPU，而不是零个。
   • 内部部署的端点（external_services: false）：每个 VLM/LLM 副本需要 1 个 GPU，每个 image2video 副本需要两个 GPU（gpu_count: 2、host_ipc: true）——每种服务各部署一个副本时，共需四个 GPU。
   • 内部模式的总需求 = 两种来源相加：三个任务 Pod 所需的 GPU 加上四个端点所需的 GPU——至少需要七个 GPU，而不是四个。

5. 残留的失败 Pod — 触发运行前，检查计算命名空间中累积的失败 Pod，并报告结果。这些 Pod 按设计会保留，不影响运行正确性，但会占用命名空间配额，并使日志搜索更杂乱：

   kubectl get pods -n sdg-workflow \
     --field-selector=status.phase=Failed \
     -o custom-columns='NAME:.metadata.name,AGE:.metadata.creationTimestamp,DAG:.metadata.labels.dag_id'
   

   只有在与用户确认后，才清理 dag_id 标签匹配用户所拥有运行的 Pod。

明确记录每项检查的结果。

如果任一检查失败：自动调用环境设置技能——不要等用户说“设置”，也不要询问他们要调用哪个技能。

如果用户的请求意味着首次部署或明确要求部署（“deploy”“install”“set up”“reinstall”“redeploy”“full setup”）：即使所有检查都通过，也要调用环境设置技能，并先确认计划执行的命令。

如果所有检查都通过且用户只想运行工作流：直接继续处理负载并触发运行。

随附工具

• scripts/upload_images.py：验证并上传本地种子图像或包含种子图像的平铺目录。
• scripts/payload.py：生成或验证独立的、兼容 EventVideoGenerationDagPayloadConfig 的 JSON。
• scripts/summarize_results.py：汇总下载的 anomaly_dataset/dataset.json 数据集。

从此技能目录运行命令。凭据必须从启动 Agent 的 shell 环境继承；绝不要要求用户将机密值粘贴到提示词中。

流程

1. 确定输入来源。

   • 对于本地数据，上传前先验证：

     python scripts/upload_images.py --path /path/to/seed-images --validate-only
     

   • 然后上传：

     python scripts/upload_images.py \
       --path /path/to/seed-images --destination-path event-video-generation/my-run
     

   • 对于现有的存储 URL，在确认它指向单张图片或扁平的图片目录（.jpg、.jpeg、.png、.bmp、.gif、.tiff、 .webp）后，按原样使用。与人物裁剪工作流不同，这里没有子目录约定——input_path 直属目录中的每个匹配文件都是一张输入图片。当 input_path 指向一个目录时，DAG 会对匹配文件排序，并取前 max_images 个。

2. 选择服务模式。

   • external 要求显式提供 VLM、LLM 和 image2video 端点 URL。
   • internal 允许 DAG 的服务生命周期在集群内部部署全部三个服务。
   • 服务模式与控制器部署位置应分别选择。本地控制器可以使用外部推理端点。
   • 确保嵌套的服务模式和输出目录与顶层设置一致。
   • 在 Kubernetes 上，VLM 和 LLM 各自从 k8s_gpu_1 申请一块 GPU，而 image2video 每个副本申请 两 块 GPU（gpu_count: 2、host_ipc: true）——仅内部部署这些端点就需要四块 GPU。这是在 detection_and_tracking/captioning/visual_qa 始终从 k8s_gpu_task 申请的三块 GPU 之外额外需要的，无论服务模式如何（参见上文的就绪检查 GPU 需求明细）：external 模式至少需要三块可分配 GPU，internal 模式至少需要七块——不是零块和四块。

3. 阅读 payload-contract.md，然后根据上文收集的值生成 payload。不要复制已检入的 dev 或 CI payload——其中包含特定于部署的端点 URL 和存储桶路径，不能沿用到用户运行中。

   External：

   python scripts/payload.py render \
     --input-path s3://bucket/input/seed-images/ \
     --output-directory s3://bucket/output/event-video-generation/ \
     --service-mode external \
     --vlm-url https://vlm.example/v1 \
     --llm-url https://llm.example/v1 \
     --image2video-url https://image2video.example/v1 \
     --max-images 10 --num-augmentation 3 \
     --variable-distribution assets/variable-distribution.json \
     --output /tmp/evg-payload.json
   

   Internal：

   python scripts/payload.py render \
     --input-path s3://bucket/input/seed-images/ \
     --output-directory s3://bucket/output/event-video-generation/ \
     --service-mode internal \
     --max-images 10 --num-augmentation 3 \
     --output /tmp/evg-payload.json
   

   向用户展示生成的 payload（或其已验证的内容），并在继续之前征得明确确认。只有用户确认后，才能继续进行预检和触发；如果用户想要修改，则重新生成并再次确认。

4. 通过 Airflow API 对 DAG 执行预检。检查 DAG 已加载、所需资源池有可用槽位，且控制器 Pod 健康——参见 airflow-direct-api.md#preflight-direct-path。 只确认凭据存在；绝不打印凭据值。

5. 提交且仅提交一个 DAG 运行。将第 3 步的 payload 作为 conf.payload 传入——完整请求格式参见 airflow-direct-api.md#trigger-a-run。记录并返回 dag_run_id、输入路径、输出目录和服务模式。

6. 触发后立即开始监控运行状态，无需等待用户提出要求，直到运行进入终态（success 或 failed）。每隔 60–120 秒轮询 Airflow API：

   # Poll run state
   RESPONSE=$(curl -s -H "Authorization: Bearer $TOKEN" \
     "$AIRFLOW_URL/api/v2/dags/$DAG_ID/dagRuns/$RUN_ID")
   RESPONSE="$RESPONSE" python3 -c "import json, os; print(json.loads(os.environ['RESPONSE'])['state'])"
   

   当状态为 running 或 failed 时，如需查看各任务的状态明细，参见 airflow-direct-api.md。

   运行状态变为 success 或 failed 后立即停止轮询。使用适合当前运行环境的轮询循环——shell while 循环、后台进程或工具原生调度器。不要让用户等待每次轮询；状态发生变化时及时报告。

   告知用户也可以在 Airflow UI 中实时查看进度。make port-forward 会在前台运行且不会退出，因此应将它作为后台任务启动——最好让用户在自己的终端中运行，因为 Agent 启动的端口转发会在会话结束时终止。应解析出主机的实际地址，而不是报告占位符或 localhost；从另一台机器访问时，localhost 没有意义：

   HOST_IP=$(hostname -I | awk '{print $1}')
   echo "Airflow UI: http://$HOST_IP:8080"
   

   默认凭据为 admin/admin，定义在 deploy/values.yaml 的 airflow.createUserJob.defaultUser 下（而非 webserver.defaultUser）。投入生产使用前请更新凭据。

   有关各任务的完整细分，请参阅 airflow-direct-api.md。

   若要停止正在进行的运行：打开 Airflow UI，找到活动的 DagRun，定位正在运行的 任务，并将其标记为 Failed（任务菜单 → Mark Failed）。这会触发 DAG 的关闭流程， 清理 Deployments、Services 和 GPU Pod。不要删除 DagRun 或 DAG，否则会绕过清理流程， 留下过期的集群资源。

7. 运行状态变为 success 或 failed 后，询问用户： “您想下载并分析结果吗？” 不要自动下载，等待用户确认。

   如果用户确认，请使用 Shell 环境中已有的 AWS 凭据（标准的 AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY / AWS_DEFAULT_REGION、 AWS 配置文件或实例角色）。绝不要要求用户将凭据粘贴到提示词中。 运行产物位于 <output_directory>/<run_id>/ 下，其中 <output_directory> 是 payload 值，<run_id> 是第 5 步中的 dag_run_id。最终数据集位于 anomaly_dataset/：

   aws s3 sync "<output_directory>/<run_id>/anomaly_dataset/" /tmp/evg-results/
   python scripts/summarize_results.py --results-dir /tmp/evg-results
   

   若要检查生成的中间视频，请改为同步 <output_directory>/<run_id>/cosmos/， 并读取各个 <video_key>/<augmentation_index>/ 文件夹中的 metadata.json。

   解读文件前请阅读 outputs.md。

防护规则

• 绝不要使用代码库或已检入 payload 中的默认端点 URL、存储桶路径或输入路径。 构建 payload 前，始终向用户询问每个特定于部署的值。如果缺少必需值，请停止并询问，不要猜测替代值。
• 在整个会话中保留用户明确提供的输入以及端点/模型选择。
• 如果 payload 验证、本地数据集验证或 Airflow 预检失败，不要提交任务。
• 不要展示 AWS 凭据、Airflow Bearer 令牌或 S3 签名 URL。
• 除非用户明确要求，否则不要启动多个运行。
• 如果用户未提供数据集位置，请询问；此工作流没有默认的演示数据集。
• 不要编造仅用于生成或仅用于标注的 DAG ID——上文列出的 DAG 是唯一存在的 DAG。
• 只能提供清单文件存在且 DAG 已加载到 Airflow 中的平台。
• 不要在此处处理人物裁剪区域的服装/属性增强请求；应使用 image-attribute-augmentation-workflow。

参考资料

• 有关环境、存储和策略要求，请阅读 setup-and-preflight.md。
• 创建或修改 payload 时，请阅读 payload-contract.md。
• 获取或解读结果时，请阅读 outputs.md。
• 验证、API 或运行时错误发生后，请阅读 troubleshooting.md。
• 有关所有 Airflow 交互操作，请阅读 airflow-direct-api.md：预检、触发、监控、各任务细分和日志检索。