用 Python 创建并消费图片识别 Job

从手工操作进阶到 API 调用

上一教程已经通过控制台创建了第一个 Job。本教程保留同一个 mock-image-detector · 1.0.0 算法和模拟目标检测 Runner，只把“上传图片并创建任务”替换成 Python 业务端调用。完成后，你会得到一条可嵌入业务系统的最小自动化链路。

开始前请先完成“跑通第一个图片识别 Job”中的启动平台和注册算法步骤，并保持 Django 服务运行。无需重新注册算法，也不需要 Celery。

• 业务端 Python：申请短时上传 URL、PUT 图片、用 object_id 创建 Job。
• 平台：校验上传文件并把 Job 放入队列。
• Python Runner：轮询队列、领取任务、下载图片并回传 detections。
• 业务端或用户：通过 API 或详情页读取最终状态和结果。

Python 业务端 → 预签名上传图片 → POST /api/v1/jobs/ → Queued
                                                        ↓
结果页 / GET Job ← Succeeded ← complete + detections ← Python Runner

认识创建任务脚本

仓库内置 examples/clients/create_image_detection_job.py，只使用 Python 标准库。脚本不会把图片直接塞进 Job JSON，而是先创建 StoredObject 并取得短时 PUT URL，上传成功后再把 object_id 放进创建任务请求。这样同一套代码可以兼容本地私有存储以及正式环境的 OSS、COS、OBS。

• 第一步：POST /api/v1/objects/presigned-put-url/，登记文件名、大小、MIME 和 SHA-256。
• 第二步：按平台返回的 method、url 和 headers 上传原始图片；短时 URL 不写入日志。
• 第三步：POST /api/v1/jobs/，指定算法标识、版本、object_id 和 threshold。
• 脚本默认读取 examples/assets/sample-street-scene.svg，也可通过 --image 换成本地图片。

signed = client.post_json(
    "/api/v1/objects/presigned-put-url/",
    {"role": "input", "original_name": image_path.name, "size": len(image)},
)
client.upload(signed["url"], signed["method"], signed["headers"], image)

job = client.post_json(
    "/api/v1/jobs/",
    {
        "name": "Python 创建的模拟图片识别",
        "algorithm_key": "mock-image-detector",
        "algorithm_version": "1.0.0",
        "input": {"object_id": signed["object"]["id"]},
        "parameters": {"threshold": 0.5},
    },
)

配置业务端和 Runner 身份

本地演示可以让两个 Python 进程都暂时使用 admin 的 Basic Authentication。创建脚本使用 SI_API_*，Runner 使用 SI_RUNNER_*，分开命名是为了明确生产环境中“任务创建者”和“任务执行者”应是两个不同身份。

export SI_API_USERNAME=admin
export SI_API_PASSWORD='与 ADMIN_PASSWORD 相同的本地测试密码'
export SI_RUNNER_USERNAME=admin
export SI_RUNNER_PASSWORD='与 ADMIN_PASSWORD 相同的本地测试密码'

让 Python Runner 持续等待任务

打开第二个终端，在项目根目录启动 Runner。这次不传 --once，让它每 5 秒轮询一次队列；队列暂时为空是正常现象。

export SI_RUNNER_USERNAME=admin
export SI_RUNNER_PASSWORD='与 ADMIN_PASSWORD 相同的本地测试密码'
conda run -n SovereignInference python examples/runners/mock_image_detection_runner.py

用 Python 上传图片并创建任务

再打开第三个终端，在项目根目录设置业务端凭据并运行创建脚本。正常情况下会打印 Job ID、queued 状态和详情页地址；持续运行的 Runner 随后会自动领取它。

export SI_API_USERNAME=admin
export SI_API_PASSWORD='与 ADMIN_PASSWORD 相同的本地测试密码'
conda run -n SovereignInference python examples/clients/create_image_detection_job.py \
  --image examples/assets/sample-street-scene.svg \
  --threshold 0.5

核对消费结果

观察两个终端：创建脚本先输出任务已创建，Runner 随后输出“任务 … 已完成”。打开创建脚本打印的详情页并刷新，状态应从“排队中”依次变成“运行中”和“已成功”。

• 输入对象的原文件名是 sample-street-scene.svg，并带有 size 与 sha256。
• 参数区域包含 threshold: 0.5，Runner 会据此过滤模拟检测框。
• 结果包含 mock: true、图片尺寸、object_count 和 person / vehicle detections。
• 事件时间线包含创建、领取、两次心跳和完成，Worker 节点同步刷新最后心跳。

创建端：任务 <job-id> 已创建，当前状态：queued。
消费端：任务 <job-id> 已完成，模拟识别到 2 个目标。

迁移到真实业务代码

接入实际业务时，可以直接复用示例中的三段请求，但应把认证、HTTP 超时、重试和 Job ID 持久化纳入业务客户端。Runner 侧则将 build_mock_result 替换成真实模型推理，领取、心跳、完成和失败协议保持不变。

• 为上传和创建请求设置有限次数的指数退避；只对安全或带幂等键的操作重试。
• 保存业务记录与 Job ID 的映射，使用 GET /api/v1/jobs/<id>/ 查询状态。
• 大文件避免一次性读入内存，并根据对象存储能力实现分片上传。
• Runner 执行耗时推理时应周期性发送 heartbeat，并在收到 should_cancel 后尽快停止。
• 生产环境使用 HTTPS、最小权限 Token、私有 Bucket 与受控网络出口。