physical-ai-event-video-generation

PAIDF オーケストレーション — イベント動画生成

イベント動画生成 DAG をエンドツーエンドで実行します。シード画像の入力準備、Cosmos3 による画像から動画への変換と異常の拡張、自動ラベリング（検出と追跡、キャプション生成、異常のビジュアル QA、人物属性のビジュアル QA、人物属性検索）、異常データセットの生成、結果の取得を行います。

DAG の選択

このワークフローでは、airflow/dags/workflows/event_video_generation_dag/ からコンピューティングプラットフォームごとに 1 つの DAG を構築します。

プラットフォーム	DAG ID	マニフェスト
Kubernetes	event_video_generation_dag_k8s	event_video_generation_k8s_manifest.yaml

マニフェストがチェックインされているプラットフォームは Kubernetes のみであるため、このリポジトリが登録する DAG は event_video_generation_dag_k8s だけです。DAG はマニフェストが存在する場合にのみ登録されます。マニフェストがない場合、DAG は壊れているのではなく、Airflow に存在しません。実行をトリガーする前に、Airflow が実際に読み込んだ DAG を一覧表示し、その一覧にない DAG ID は決して指定しないでください。

エンドツーエンドのパイプラインは 1 つだけです。生成のみ、またはラベリングのみの 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	1 枚のシード画像、またはシード画像のディレクトリを指す S3（または HTTP/HTTPS）URL
常に必須	output_directory	結果の書き込み先となる、書き込み可能な S3 URL
常に必須	サービスモード	external（ユーザーがエンドポイント URL を指定）または internal（DAG がクラスター内にサービスをデプロイ）
外部モード	cosmos.vlm_service_url	VLM 推論エンドポイントの HTTPS URL（完全な URL）
外部モード	cosmos.llm_service_url	LLM 推論エンドポイントの HTTPS URL（完全な URL）
外部モード	cosmos.image2video_service_url	Cosmos3 の画像から動画への推論エンドポイントの HTTPS URL（完全な URL）
任意	max_images	ディレクトリから処理する画像の数（デフォルト：10、0 以下の場合はすべて）
任意	cosmos.num_augmentation	画像ごとに生成する異常動画の数（デフォルト：1）
任意	cosmos.variable_distribution	anomaly_type / env_type のサンプリング分布（payload-contract.md を参照）

必須項目がユーザーから提示されていない場合は、先に進む前に明示的に確認してください。以前の実行やチェックイン済みファイルから値を捏造したり、再利用したりしてはいけません。

実行をトリガーする前に、以下の準備状況チェックを必ず実施してください。 チェックは失敗した時点で打ち切り、最初の失敗が発生したら直ちに環境セットアップスキルに引き継いでください。

チェックを始める前に、クラスターへの接続を確立してください。クラスターへの接続にはユーザーが提供する認証情報のみを使用します。認証情報がリポジトリに含まれることはありません。シェル環境ですでにクラスター認証情報ファイルのパスがエクスポートされているか確認してください。エクスポートされていない場合は、クラスターのコマンドを実行する前に、絶対パスをユーザーに確認してください。パスを決めつけたり、ディスク上のデフォルトにフォールバックしたりしてはいけません。手順全体については、setup-and-preflight.md を参照してください。 コントローラー（Airflow）とDAGの計算タスクは、別のリモートクラスター接続が設定されていない限り、同じクラスターで実行されます。GPUの容量はこのクラスターで確認します。

1. コントローラーポッド — Airflowのコントローラーポッド（DAGタスクポッドではありません）がRunning状態であることを確認します。 DAGタスクポッドがPendingまたはFailed状態になるのは正常です。コントローラーの障害と誤認しないでください。

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

   release=sdg-workflow-controllerラベルに一致するすべてのポッドが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. プール — 手順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数を報告してください。独立したGPU使用元は2つあり、そのうちモードに依存するのは1つだけです。

   • サービスモードにかかわらず、ローカルモデル推論を実行するタスクポッド： detection_and_tracking、captioning、visual_qaはすべてk8s_gpu_taskプロファイルで実行されます（それぞれGPUを1つ使用します）。これらはエンドポイントを呼び出すのではなく、ポッド内で推論を行うため、この使用量は外部モードと内部モードの両方に適用されます。event_and_person_attribute_searchとaugmentationはCPUプロファイルで実行されるため、GPUを使用しません。そのため、外部モードでも最低3つのGPUが必要であり、0ではありません。
   • 内部デプロイされたエンドポイント（external_services: false）：VLM/LLMレプリカごとにGPUを1つ、image2videoレプリカごとにGPUを2つ使用します（gpu_count: 2、host_ipc: true）。各サービスのレプリカが1つの場合、GPUは合計4つ必要です。
   • 内部モードの合計 = 2つの使用元の合計：タスクポッド用のGPU 3つとエンドポイント用のGPU 4つで、最低7つのGPUが必要です。4つではありません。

5. 古い失敗ポッド — トリガーする前に、計算名前空間に蓄積した失敗ポッドを確認し、その数を報告します。これらは設計上保持されており、実行の正しさには影響しませんが、名前空間のクォータを消費し、ログ検索を煩雑にします。

   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ラベルに一致するポッドのみを削除してください。

各確認結果を明示的に記録してください。

いずれかの確認に失敗した場合：ユーザーが「セットアップして」と言うのを待ったり、スキル名を尋ねたりせず、環境セットアップスキルを自動的に呼び出してください。

ユーザーの依頼が初回または明示的なデプロイを示す場合（「デプロイ」「インストール」「セットアップ」「再インストール」「再デプロイ」「完全なセットアップ」）：すべての確認に成功していても環境セットアップスキルを呼び出し、まず実行予定のコマンドを確認してください。

すべての確認に成功し、ユーザーがワークフローの実行のみを希望している場合：ペイロードの作成とトリガーに直接進んでください。

同梱ツール

• scripts/upload_images.py：ローカルのシード画像またはシード画像を含むフラットディレクトリを検証してアップロードします。
• scripts/payload.py：EventVideoGenerationDagPayloadConfig互換のスタンドアロンJSONをレンダリングまたは検証します。
• scripts/summarize_results.py：ダウンロードしたanomaly_dataset/dataset.jsonデータセットを要約します。

このスキルのディレクトリからコマンドを実行してください。認証情報はエージェントを起動したシェルから継承する必要があります。ユーザーに秘密の値をプロンプトへ貼り付けるよう求めてはいけません。

手順

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）を指していることを確認したうえで、そのまま使用してください。person-crop ワークフローとは異なり、サブディレクトリの規約はありません。input_path 直下にある一致するすべてのファイルが 1 つの入力画像となります。input_path がディレクトリを指す場合、DAG は一致するファイルをソートし、そのうち最初の max_images 個を取得します。

2. サービスモードを選択します。

   • external では、VLM、LLM、image2video のエンドポイント URL を明示的に指定する必要があります。
   • internal では、DAG のサービスライフサイクルが 3 つのサービスすべてをクラスター内にデプロイできます。
   • サービスモードはコントローラーの配置とは独立して選択してください。ローカルコントローラーが外部の推論エンドポイントを使用する場合もあります。
   • ネストされたサービスモードと出力ディレクトリをトップレベルと一致させてください。
   • Kubernetes では、VLM と LLM はそれぞれ k8s_gpu_1 から 1 つの GPU を要求しますが、image2video はレプリカごとに 2 つ の GPU を要求します（gpu_count: 2、host_ipc: true）— 内部にデプロイされるエンドポイントだけで 4 つの GPU が必要です。これは、サービスモードに関係なく detection_and_tracking/captioning/visual_qa が常に k8s_gpu_task から要求する 3 つの GPU に加えてのものであり、代わりではありません（上記のレディネスチェックの GPU 内訳を参照）: external モードでは最低 3 つの割り当て可能な GPU、internal モードでは最低 7 つが必要です — ゼロと 4 ではありません。

3. payload-contract.md を読み、上記で収集した値からペイロードをレンダリングします。チェックインされた dev 用または CI 用のペイロードをコピーしないでください — それらにはデプロイメント固有のエンドポイント 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
   

   レンダリングされたペイロード（またはその検証済みの内容）をユーザーに示し、続行する前に明示的な確認を得てください。ユーザーが確認した場合にのみ、プリフライトとトリガーに進んでください。変更を希望する場合は、再レンダリングして再確認してください。

4. Airflow API を介して DAG をプリフライトします。DAG がロードされていること、必要なプールにスロットがあること、コントローラーの Pod が正常であることを確認します — airflow-direct-api.md#preflight-direct-path を参照してください。存在の確認のみを行い、認証情報の値を決して出力しないでください。

5. 正確に 1 回の DAG 実行を送信します。ステップ 3 のペイロードを 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 になったらすぐにポーリングを停止します。ランタイムに合ったポーリングループを使用してください — シェルの while ループ、バックグラウンドプロセス、またはツールネイティブのスケジューラーです。各ポーリングを待つためにユーザーをブロックせず、状態の変化が発生したら報告してください。

   ユーザーは Airflow UI で進捗をライブで確認することもできると伝えてください。make port-forward はフォアグラウンドで実行され終了しないため、バックグラウンドジョブとして起動してください — また、エージェントが所有するフォワードはセッションとともに終了するため、ユーザー自身のターミナルで実行することを推奨します。プレースホルダーや 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 のシャットダウンパスがトリガーされ、Deployment、Service、GPU ポッドがクリーンアップされます。 DagRun や DAG を削除しないでください。クリーンアップがバイパスされ、古いクラスターリソースが 残ってしまいます。

7. ランが success または failed に到達したら、ユーザーに尋ねてください: 「結果をダウンロードして分析しますか？」 自動的にダウンロードしないでください。確認を待ちます。

   ユーザーが確認した場合は、シェル環境ですでに利用可能な AWS 認証情報（標準の AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY / AWS_DEFAULT_REGION、AWS プロファイル、 またはインスタンスロール）を使用してください。ユーザーにプロンプトへ認証情報を 貼り付けるよう決して求めないでください。 実行の成果物は <output_directory>/<run_id>/ の下にあり、<output_directory> は ペイロードの値、<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 を読んでください。

ガードレール

• コードベースやチェックイン済みペイロードのデフォルトのエンドポイント URL、バケットパス、 入力パスを決して使用しないでください。 ペイロードを構築する前に、常にデプロイ固有の すべての値をユーザーに確認してください。必須の値が欠けている場合は、停止して尋ねてください。 推測で代用しないでください。
• セッション全体を通じて、ユーザーの明示的な入力およびエンドポイント/モデルの選択を 維持してください。
• ペイロード検証、ローカルデータセット検証、または Airflow プリフライトが失敗した場合は 送信しないでください。
• AWS 認証情報、Airflow ベアラートークン、S3 署名付き URL を表示しないでください。
• ユーザーが明示的に要求しない限り、複数のランを開始しないでください。
• データセットの場所が指定されていない場合は確認してください。このワークフローには 暗黙のデモデータセットはありません。
• 生成専用やラベリング専用の DAG ID を捏造しないでください。存在するのは上記に記載された DAG のみです。
• マニフェストが存在し、DAG が Airflow にロードされているプラットフォームのみを 提供してください。
• 人物クロップの服装/属性拡張リクエストをここにルーティングしないでください。それは image-attribute-augmentation-workflow です。

リファレンス

• 環境、ストレージ、ポリシーの要件については setup-and-preflight.md を読んでください。
• ペイロードを作成または変更する場合は payload-contract.md を読んでください。
• 結果を取得または解釈する場合は outputs.md を読んでください。
• 検証、API、またはランタイムエラーの後は troubleshooting.md を読んでください。
• すべての Airflow 操作については airflow-direct-api.md を読んでください: プリフライト、トリガー、モニタリング、タスクごとの内訳、ログの取得。