physical-ai-image-attribute-augmentation

PAIDF 编排 — 图像属性增强

端到端运行图像属性增强 DAG：准备人物裁剪输入、使用 cosmos 编辑图像进行增强、cosmos 后处理、搜索事件和人物属性、生成增强数据集，以及检索结果。

DAG 选择

该工作流从 airflow/dags/workflows/image_attribute_augmentation_dag/ 为每个计算平台构建一个 DAG：

平台	DAG ID	清单
Kubernetes	image_attribute_augmentation_dag_k8s	image_attribute_augmentation_k8s_manifest.yaml

Kubernetes 是唯一已检入清单的平台，因此 image_attribute_augmentation_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，其直接子目录为人物 ID 文件夹
始终必填	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）

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

触发运行前，始终执行以下就绪检查。 检查会在首次失败时立即停止——一旦失败，马上转至环境设置技能。

执行任何检查之前，先建立集群连接。只能使用用户提供的凭据访问集群——这些凭据不会包含在代码库中。检查 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. 池 — 只有检查 2 通过后才进行。所需池必须有空闲槽位：k8s_gpu_1、 default_pool，以及所选模式对应的增强池（外部模式使用 external_image_edit_service_pool，内部模式使用 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 的空闲数和总数，而不只是可分配数。外部模式的推理无需 GPU — 每个任务 Pod （augmentation、cosmos_post_processing、event_and_person_attribute_search）都运行在 CPU 配置上，不同于 EVG 中使用 k8s_gpu_task 配置的自动标注阶段。内部模式每个服务副本至少需要一个 GPU（VLM、LLM、图像编辑服务 — 最少三个）。

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，并在清理前向用户确认。

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

如果任何检查失败：自动调用环境设置技能 — 不要等用户说“设置”，也不要让用户指定技能名称。

如果用户的请求意味着首次部署或明确要求部署（“部署”“安装”“设置”“重新安装”“重新部署”“完整设置”）：即使所有检查都通过，也要调用环境设置技能，并先确认计划执行的命令。

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

捆绑工具

• 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 路径。

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

流程

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 子目录后，保持 URL 不变。input_path 的每个直接子目录都视为一个 person ID，其图像会合并为每个人一条水平拼接图。

2. 选择服务模式。

   • external 需要明确提供 VLM、LLM 和图像编辑端点 URL。
   • internal 会由 DAG 的服务生命周期在集群内部署这三项服务。
   • 服务模式应独立于控制器部署位置选择。本地控制器可以使用外部推理端点。
   • 嵌套的服务模式和输出目录必须与顶层保持一致。
   • 在 Kubernetes 上，每个已部署的端点都会从 k8s_gpu_1 申请一块 GPU，因此 internal 模式至少需要三块可分配 GPU（如果提高任何 replicas 值，则需要更多）；external 模式的推理不需要 GPU。

3. 阅读 payload-contract.md，然后根据上述收集的值生成 payload。不要复制已纳入版本控制的开发或 CI payload——其中包含特定于部署的端点 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
   

   向用户展示生成的 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，因为其他机器无法通过它访问：

   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（任务菜单 → 标记为 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。最终数据集位于 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。

防护措施

• 绝不要使用代码库或已签入的 payload 中的默认端点 URL、存储桶路径或输入路径。 构建 payload 前，务必向用户询问所有部署特定的值。如果缺少必需值，请停止并询问——不要猜测 后自行替代。
• 在整个会话中保留用户明确指定的输入以及端点／模型选择。
• 如果 payload 验证、本地数据集验证或 Airflow 预检失败，请勿提交。
• 不要显示 AWS 凭证、Airflow bearer token 或 S3 签名 URL。
• 除非用户明确要求，否则不要启动多个运行。
• 如果用户未提供数据集位置，请询问；此工作流没有隐式的演示数据集。
• 不要编造仅用于增强或仅用于标注的 DAG ID——上方列出的 DAG 是唯一存在的 DAG。
• 只提供清单文件存在且其 DAG 已在 Airflow 中加载的平台。

参考资料

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