Skip to main content
GET
/
v1
/
tasks
/
{task_id}
获取任务状态
curl --request GET \
  --url https://ai.kaiho.cc/v1/tasks/{task_id}

​
概述

许多 AI 生成任务（如视频生成）是异步处理的。使用任务管理 API 可以查询任务状态、进度和结果。

​
任务生命周期

1

创建任务

调用生成 API 创建任务，获得 task_id
2

排队中 (queued)

任务在队列中等待处理
3

处理中 (processing)

任务正在生成中，可查看进度
4

完成 (completed) / 失败 (failed)

任务完成或失败，可获取结果或错误信息

​
请求参数

​
task_id
string
required
任务的唯一标识符，在创建任务时返回。

​
请求示例

curl https://ai.kaiho.cc/v1/tasks/task_abc123 \
  -H "Authorization: Bearer YOUR_API_KEY"

​
响应格式

​
排队中

{
  "id": "task_abc123",
  "type": "video_generation",
  "status": "queued",
  "progress": 0,
  "queue_position": 5,
  "estimated_time": 180,
  "created_at": "2024-01-15T10:30:00Z"
}

​
处理中

{
  "id": "task_abc123",
  "type": "video_generation",
  "status": "processing",
  "progress": 45,
  "current_step": "rendering",
  "estimated_time_remaining": 90,
  "created_at": "2024-01-15T10:30:00Z",
  "started_at": "2024-01-15T10:31:00Z"
}

​
已完成

{
  "id": "task_abc123",
  "type": "video_generation",
  "status": "completed",
  "progress": 100,
  "result": {
    "video_url": "https://storage.kaiho.cc/videos/video-123.mp4",
    "thumbnail_url": "https://storage.kaiho.cc/videos/thumb-123.jpg",
    "duration": 10,
    "resolution": "1920x1080",
    "size_bytes": 15728640
  },
  "created_at": "2024-01-15T10:30:00Z",
  "started_at": "2024-01-15T10:31:00Z",
  "completed_at": "2024-01-15T10:33:30Z",
  "processing_time": 150
}

​
失败

{
  "id": "task_abc123",
  "type": "video_generation",
  "status": "failed",
  "progress": 65,
  "error": {
    "code": "content_violation",
    "message": "生成的内容违反了我们的内容政策",
    "details": "检测到不适当的内容"
  },
  "created_at": "2024-01-15T10:30:00Z",
  "started_at": "2024-01-15T10:31:00Z",
  "failed_at": "2024-01-15T10:32:45Z"
}

​
任务状态说明

状态说明可用字段
queued任务已创建，等待处理queue_position, estimated_time
processing任务正在处理中progress, current_step, estimated_time_remaining
completed任务成功完成result, processing_time
failed任务处理失败error
cancelled任务被取消-

​
轮询最佳实践

import time

def poll_task(task_id):
    while True:
        status = get_task_status(task_id)
        
        if status['status'] in ['completed', 'failed']:
            return status
        
        # 根据进度调整轮询间隔
        if status['progress'] < 20:
            time.sleep(10)  # 初期较慢，10秒轮询一次
        else:
            time.sleep(5)   # 后期较快，5秒轮询一次

import time

def poll_with_backoff(task_id, max_attempts=100):
    interval = 2
    max_interval = 60
    
    for attempt in range(max_attempts):
        status = get_task_status(task_id)
        
        if status['status'] in ['completed', 'failed']:
            return status
        
        time.sleep(interval)
        interval = min(interval * 1.5, max_interval)  # 指数增长，最大60秒

def smart_poll(task_id):
    status = get_task_status(task_id)
    
    # 根据预估剩余时间智能等待
    if 'estimated_time_remaining' in status:
        wait_time = max(5, status['estimated_time_remaining'] / 5)
        time.sleep(wait_time)
    else:
        time.sleep(10)

​
使用 Webhook（推荐）

避免频繁轮询，使用 webhook 接收完成通知： 
# 创建任务时指定 webhook
response = requests.post(
    "https://ai.kaiho.cc/v1/videos/generations",
    headers={"Authorization": "Bearer YOUR_API_KEY"},
    json={
        "model": "sora-2",
        "prompt": "视频描述...",
        "webhook_url": "https://your-server.com/webhook",
        "webhook_events": ["completed", "failed"]  # 指定要接收的事件
    }
)
Webhook 回调示例： 
{
  "event": "task.completed",
  "task_id": "task_abc123",
  "type": "video_generation",
  "status": "completed",
  "result": {
    "video_url": "https://storage.kaiho.cc/videos/video-123.mp4",
    "duration": 10
  },
  "completed_at": "2024-01-15T10:33:30Z"
}

​
批量查询任务

查询多个任务的状态： 
curl https://ai.kaiho.cc/v1/tasks \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -G \
  -d "ids=task_abc123,task_def456,task_ghi789"

​
取消任务

取消正在处理的任务： 
import requests

response = requests.delete(
    f"https://ai.kaiho.cc/v1/tasks/task_abc123",
    headers={"Authorization": "Bearer YOUR_API_KEY"}
)

print(response.json())
# {"id": "task_abc123", "status": "cancelled"}
注意： 已经开始处理的任务可能无法立即取消，仍会产生部分费用。

​
任务历史查询

获取您的任务历史记录： 
response = requests.get(
    "https://ai.kaiho.cc/v1/tasks",
    headers={"Authorization": "Bearer YOUR_API_KEY"},
    params={
        "type": "video_generation",  # 筛选任务类型
        "status": "completed",        # 筛选任务状态
        "limit": 50,                  # 每页数量
        "offset": 0                   # 偏移量
    }
)

tasks = response.json()

​
错误码说明

错误码说明处理建议
task_not_found任务不存在检查 task_id 是否正确
content_violation内容违规修改提示词，避免违规内容
insufficient_credits余额不足充值账户
processing_timeout处理超时简化任务或联系支持
system_error系统错误稍后重试或联系支持

​
完整示例

综合使用任务管理的完整示例： 
import requests
import time

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ai.kaiho.cc/v1"

def create_video_task(prompt):
    """创建视频生成任务"""
    response = requests.post(
        f"{BASE_URL}/videos/generations",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": "sora-2",
            "prompt": prompt,
            "duration": 10,
            "resolution": "1920x1080"
        }
    )
    return response.json()

def get_task_status(task_id):
    """获取任务状态"""
    response = requests.get(
        f"{BASE_URL}/tasks/{task_id}",
        headers={"Authorization": f"Bearer {API_KEY}"}
    )
    return response.json()

def wait_for_completion(task_id, timeout=600):
    """等待任务完成"""
    start_time = time.time()
    
    while True:
        if time.time() - start_time > timeout:
            raise TimeoutError(f"任务超时: {task_id}")
        
        status = get_task_status(task_id)
        
        print(f"状态: {status['status']}, 进度: {status.get('progress', 0)}%")
        
        if status['status'] == 'completed':
            return status['result']
        elif status['status'] == 'failed':
            raise Exception(f"任务失败: {status['error']}")
        
        # 智能等待
        time.sleep(5)

# 使用示例
task = create_video_task("一只猫在弹钢琴")
print(f"任务已创建: {task['id']}")

result = wait_for_completion(task['id'])
print(f"视频已生成: {result['video_url']}")
性能优化： 使用 webhook 代替轮询可以显著减少API调用次数，降低延迟并节省成本。
⌘I