physical-ai-image-attribute-augmentation

PAIDF オーケストレーション — 画像属性の拡張

画像属性拡張 DAG をエンドツーエンドで実行します。人物クロップ入力の準備、cosmos による画像編集拡張、cosmos の後処理、イベントおよび人物属性の検索、拡張データセットの生成、結果の取得を行います。

DAG の選択

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

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

マニフェストがチェックインされているプラットフォームは Kubernetes のみであるため、このリポジトリが登録する DAG は image_attribute_augmentation_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	直下のサブディレクトリが人物 ID フォルダーになっている 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.image_edit_service_url	画像編集推論エンドポイントの完全な HTTPS URL
任意	max_imgs	処理する人物 ID フォルダーの数（既定値：1、0 または負の値：すべて）
任意	cosmos.num_augmentation	人物ごとの服装のバリエーション数（既定値：1）
任意	cosmos.variable_distribution	服装属性の分布ファイルのパス（payload-contract.md を参照）

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

実行をトリガーする前に、必ず以下の準備状況チェックを行ってください。 チェックは最初の失敗時に終了します。最初の失敗があった時点で、直ちに環境セットアップスキルへ案内してください。

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

1. コントローラーの Pod — Airflow コントローラーの Pod（DAG タスクの Pod ではない）が Running であることを確認します。Pending または Failed 状態の DAG タスク Pod は正常であり、コントローラーの障害と誤認してはいけません:

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

   release=sdg-workflow-controller ラベルに一致するすべての Pod が Running でなければなりません。namespace が存在しない場合は初回インストールの状態です。environment-setup スキルにルーティングし、それ以上診断しないでください。

2. Airflow API — チェック 1 が成功した場合にのみ到達可能です。まず Kubernetes ClusterIP から AIRFLOW_URL を確立します（ホストから常にルーティング可能で、port-forward は不要です）:

   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 に到達できない場合は、environment-setup スキルにルーティングします。

3. プール — チェック 2 が成功した場合にのみ実施します。空きスロットのある必須プール: k8s_gpu_1、default_pool、および選択したモードの拡張プール（external の場合は external_image_edit_service_pool、internal の場合は iaa_internal_image_edit_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 は allocatable だけでなく、空き対合計として報告してください。external モードでは推論に GPU は不要です — すべてのタスク Pod（augmentation、cosmos_post_processing、event_and_person_attribute_search）は CPU プロファイルで実行されます。これは EVG の k8s_gpu_task プロファイルの自動ラベリングステージとは異なります。internal モードでは、サービスレプリカごとに少なくとも 1 つの GPU が必要です（VLM、LLM、image-edit ＝ 最低 3 つ）。

5. 古い失敗 Pod — トリガーする前に、コンピュート namespace に蓄積された失敗 Pod を確認し、報告します。これらは設計上保持されるもので、実行の正確性には影響しませんが、namespace のクォータを消費し、ログ検索を煩雑にします:

   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 のみをクリーンアップしてください。

各チェック結果を明示的に記録してください。

いずれかのチェックが失敗した場合: environment-setup スキルを自動的に呼び出してください — ユーザーが「セットアップ」と言うのを待ったり、スキル名を尋ねたりしないでください。

ユーザーのリクエストが初回または明示的なデプロイを示唆する場合（「デプロイ」「インストール」「セットアップ」「再インストール」「再デプロイ」「フルセットアップ」）: すべてのチェックが成功していても environment-setup スキルを呼び出し、計画されたコマンドを先に確認してください。

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

同梱ツール

• scripts/upload_images.py: ローカルの <person_id>/<image>.(jpg|jpeg|png) データを検証/アップロードします。
• scripts/payload.py: スタンドアロンの ImageAttributeAugmentationDagPayloadConfig 互換 JSON をレンダリングまたは検証します。
• scripts/summarize_results.py: ダウンロードした augmented_data.json データセットを要約します。
• scripts/workflow.py: SDG webserver API を操作します — 実行の送信、ステータスのポーリング、結果の取得、または ID で指定した単一の実行のキャンセル（その実行のみをキャンセルし、クラスターリソースや他の実行には触れません）。WEBSERVER_ENDPOINT と NGC_API_KEY が必要です。通常の操作では、以下の Airflow API パスを優先してください。

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

手順

1. 入力ソースを決定します。

   • ローカルデータの場合、アップロード前に検証します:

     python scripts/upload_images.py --path /path/to/crops --validate-only
     

   • 次に、階層を保持したままアップロードします:

     python scripts/upload_images.py \
       --path /path/to/crops --destination-path image-attribute-augmentation/my-run
     

   • 既存のストレージ URL については、person-ID のサブディレクトリが含まれていることを確認したうえで、そのまま使用してください。input_path の各直下のサブディレクトリは 1 人の person ID として扱われ、その画像は person ごとに 1 つの横長ストリップに結合されます。

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

   • external では、VLM、LLM、画像編集エンドポイントの URL を明示的に指定する必要があります。
   • internal では、DAG のサービスライフサイクルによって 3 つのサービスすべてがクラスタ内にデプロイされます。
   • サービスモードはコントローラーの配置とは独立して選択してください。ローカルコントローラーが外部推論エンドポイントを使用してもかまいません。
   • ネストされたサービスモードと出力ディレクトリは、トップレベルと一致させてください。
   • Kubernetes では、デプロイされた各エンドポイントが k8s_gpu_1 から GPU を 1 つ確保するため、internal モードには少なくとも 3 つの割り当て可能な GPU が必要です (replicas の値を増やした場合はさらに必要)。external モードでは、推論用の GPU は不要です。

3. payload-contract.md を読み、上記で収集した値からペイロードをレンダリングします。チェックインされている dev 用または CI 用のペイロードをコピーしないでください。それらにはデプロイ固有のエンドポイント URL とバケットパスが含まれており、ユーザーの実行に継承してはなりません。

   External:

   python scripts/payload.py render \
     --input-path s3://bucket/input/person-crops/ \
     --output-directory s3://bucket/output/image-attribute-augmentation/ \
     --service-mode external \
     --vlm-url https://vlm.example/v1 \
     --llm-url https://llm.example/v1 \
     --image-edit-url https://image-edit.example/v1 \
     --max-imgs 10 --num-augmentation 3 \
     --variable-distribution assets/variable-distribution.json \
     --output /tmp/iaa-payload.json
   

   Internal:

   python scripts/payload.py render \
     --input-path s3://bucket/input/person-crops/ \
     --output-directory s3://bucket/output/image-attribute-augmentation/ \
     --service-mode internal \
     --max-imgs 10 --num-augmentation 3 \
     --output /tmp/iaa-payload.json
   

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

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

5. DAG 実行をちょうど 1 回送信します。ステップ 3 のペイロードを conf.payload として渡します — リクエストの完全な形式については airflow-direct-api.md#trigger-a-run を参照してください。dag_run_id、入力パス、出力ディレクトリ、サービスモードを記録して返します。

6. トリガー直後に — 求められるのを待たずに — 実行が終端状態 (success または failed) に達するまで監視します。Airflow API を 60〜120 秒ごとにポーリングします:

   # 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 (別のマシンからは意味がありません) を報告するのではなく、ホストの実際のアドレスを解決してください:

   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 ポッドがクリーンアップされます。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 です。最終データセットは augmented_dataset/ にあります：

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

   代わりに中間的な拡張画像を確認するには、<output_directory>/<run_id>/cosmos/ を同期し、 各 <person_id>/<augmentation_index>/ フォルダーから output_metadata.json を読み取ってください。

   ファイルを解釈する前に outputs.md をお読みください。

ガードレール

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

リファレンス

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