Gerenciamento de tarefas assíncronas

Dois padrões assíncronos ativados QwenCloud: baseado em tarefas para geração de mídia, lote para processamento de texto de alto volume

QwenCloud oferece dois sistemas de processamento assíncrono. Escolha com base no que você está construindo:

API de tarefas

A geração de imagens, a geração de vídeos e a transcrição de arquivos usam a API de tarefas: você envia uma única solicitação, recebe um ID de tarefa e pesquisa o resultado.

Como funciona

Cada tarefa segue o mesmo padrão de duas etapas:

1. Crie uma tarefa — Envie uma solicitação POST e receba um task_id
2. Consulte o resultado — Enquete GET /api/v1/tasks/{task_id} até que a tarefa atinja um estado terminal

POST (create) ──→ task_id
                     │
       ┌─────────────┘
       ▼
  ┌─────────┐    ┌─────────┐    ┌───────────┐
  │ PENDING │───→│ RUNNING │───→│ SUCCEEDED │
  └─────────┘    └─────────┘    └───────────┘
       │              │
       ▼              ▼
  ┌──────────┐   ┌────────┐
  │ CANCELED │   │ FAILED │
  └──────────┘   └────────┘

Estados da tarefa

Crie uma tarefa

Algumas APIs (como geração de imagens) suportam modos síncronos e assíncronos. Adicione o X-DashScope-Async cabeçalho para forçar a execução assíncrona:

curl -X POST https://maas.qwencloudapi.com/api/v1/services/aigc/image-generation/generation \
  -H "Authorization: Bearer $DASHSCOPE_API_KEY" \
  -H "Content-Type: application/json" \
  -H "X-DashScope-Async: enable" \
  -d '{
    "model": "wan2.6-t2i",
    "input": {
      "messages": [
        {
          "role": "user",
          "content": [
            {"text": "A cat sitting on a windowsill at sunset"}
          ]
        }
      ]
    }
  }'

A resposta retorna um task_id:

{
  "request_id": "e5d5af82-9a08-xxx",
  "output": {
    "task_id": "86ecf553-d340-xxx",
    "task_status": "PENDING"
  }
}

APIs de geração de vídeo são sempre assíncronas — você não precisa do X-DashScope-Async cabeçalho para estes.

Consulte o resultado

Pesquise o endpoint da tarefa com o retornado task_id:

curl

```bash
curl -X GET "https://maas.qwencloudapi.com/api/v1/tasks/86ecf553-d340-xxx" \
  -H "Authorization: Bearer $DASHSCOPE_API_KEY"
```

Python

```python
import requests
import os

task_id = "86ecf553-d340-xxx"
url = f"https://maas.qwencloudapi.com/api/v1/tasks/{task_id}"
headers = {"Authorization": f"Bearer {os.getenv('DASHSCOPE_API_KEY')}"}

response = requests.get(url, headers=headers)
result = response.json()
print(result["output"]["task_status"])
```

Uma tarefa concluída retorna o resultado no output campo:

{
  "request_id": "e5d5af82-9a08-xxx",
  "output": {
    "task_id": "86ecf553-d340-xxx",
    "task_status": "SUCCEEDED",
    "submit_time": "2025-01-15 10:30:00.000",
    "scheduled_time": "2025-01-15 10:30:01.000",
    "end_time": "2025-01-15 10:30:12.000",
    "finished": true,
    "choices": [
      {
        "finish_reason": "stop",
        "message": {
          "role": "assistant",
          "content": [
            {
              "image": "https://dashscope-result-xxx.oss-xxx.aliyuncs.com/...",
              "type": "image"
            }
          ]
        }
      }
    ]
  },
  "usage": {
    "image_count": 1,
    "input_tokens": 0,
    "output_tokens": 0
  }
}

Modelos mais antigos (Wan 2.5 e anteriores) retornam uma estrutura de resposta diferente com results e task_metrics campos em vez de choices. Consulte a referência da API de texto para imagem para obter detalhes sobre ambos os formatos.

Uma tarefa com falha inclui um código de erro e uma mensagem:

{
  "output": {
    "task_id": "86ecf553-d340-xxx",
    "task_status": "FAILED",
    "code": "DataInspectionFailed",
    "message": "Input or output data may contain inappropriate content."
  }
}

Estratégias de votação

A pesquisa ingênua de intervalo fixo desperdiça chamadas de API e pode atingir limites de taxa. Em vez disso, use a espera exponencial.

Comece com um intervalo curto e aumente gradativamente. Diferentes modalidades têm diferentes tempos típicos de conclusão:

Python

```python
import time
import requests
import os

def poll_task(task_id, initial_interval=3, max_interval=15, timeout=120):
  """Poll a task with exponential backoff."""
  url = f"https://maas.qwencloudapi.com/api/v1/tasks/{task_id}"
  headers = {"Authorization": f"Bearer {os.getenv('DASHSCOPE_API_KEY')}"}

  interval = initial_interval
  elapsed = 0

  while elapsed < timeout:
    response = requests.get(url, headers=headers)
    result = response.json()
    status = result["output"]["task_status"]

    if status == "SUCCEEDED":
      return result
    elif status == "FAILED":
      raise Exception(
        f"Task failed: {result['output'].get('code', 'Unknown')} - "
        f"{result['output'].get('message', '')}"
      )

    time.sleep(interval)
    elapsed += interval
    interval = min(interval * 1.5, max_interval)

  raise TimeoutError(f"Task {task_id} did not complete within {timeout}s")

# Image generation: poll every 3s, escalate to 15s, timeout at 2min
result = poll_task("86ecf553-d340-xxx", initial_interval=3, max_interval=15, timeout=120)

# Video generation: poll every 15s, escalate to 60s, timeout at 5min
result = poll_task("86ecf553-d340-xxx", initial_interval=15, max_interval=60, timeout=300)
```

Node.js

```javascript
async function pollTask(taskId, { initialInterval = 3000, maxInterval = 15000, timeout = 120000 } = {}) {
  const url = `https://maas.qwencloudapi.com/api/v1/tasks/${taskId}`;
  const headers = { Authorization: `Bearer ${process.env.DASHSCOPE_API_KEY}` };

  let interval = initialInterval;
  const deadline = Date.now() + timeout;

  while (Date.now() < deadline) {
    const res = await fetch(url, { headers });
    const result = await res.json();
    const { task_status, code, message } = result.output;

    if (task_status === "SUCCEEDED") return result;
    if (task_status === "FAILED") throw new Error(`Task failed: ${code} - ${message}`);

    await new Promise(resolve => setTimeout(resolve, interval));
    interval = Math.min(interval * 1.5, maxInterval);
  }

  throw new Error(`Task ${taskId} did not complete within ${timeout}ms`);
}
```

abstração em nível de SDK

O DashScope Python SDK encapsula o loop de pesquisa. Usar call() para uma experiência síncrona, ou async_call() + wait() para controle explícito.

import dashscope
from dashscope.aigc.image_generation import ImageGeneration

dashscope.base_http_api_url = 'https://maas.qwencloudapi.com/api/v1'

message = {"role": "user", "content": [{"text": "A cat sitting on a windowsill at sunset"}]}

# Option 1: Synchronous — SDK polls internally
result = ImageGeneration.call(
  model="wan2.7-image-pro",
  messages=[message],
  n=1
)
print(result.output.choices[0].message.content[0]["image"])

# Option 2: Explicit async — you control when to wait
task = ImageGeneration.async_call(
  model="wan2.7-image-pro",
  messages=[message],
  n=1
)
# Do other work here...
result = ImageGeneration.wait(task)
print(result.output.choices[0].message.content[0]["image"])

O mesmo padrão se aplica a todas as modalidades:

from dashscope import VideoSynthesis

task = VideoSynthesis.async_call(model="wan2.6-t2v", prompt="...")
result = VideoSynthesis.wait(task)

Gerencie tarefas em escala

Listar tarefas

Consulte tarefas por intervalo de tempo, status ou modelo:

curl -X GET "https://maas.qwencloudapi.com/api/v1/tasks/?start_time=20250115100000&end_time=20250115120000&status=FAILED&model_name=wan2.6-t2i&page_no=1&page_size=20" \
  -H "Authorization: Bearer $DASHSCOPE_API_KEY"

Filtros suportados:

O intervalo de tempo não pode exceder 24 horas. Se ambos start_time e end_time são omitidos, a API retorna tarefas das últimas 24 horas.

Cancelar uma tarefa

Cancelar uma tarefa que ainda está em PENDING estado:

curl -X POST "https://maas.qwencloudapi.com/api/v1/tasks/86ecf553-d340-xxx/cancel" \
  -H "Authorization: Bearer $DASHSCOPE_API_KEY"

Somente tarefas em PENDING estado pode ser cancelado. Quando uma tarefa passa para RUNNING, ele continuará até a conclusão.

Limites de taxa

Os endpoints de gerenciamento de tarefas (consulta, lista, cancelamento) compartilham um limite de taxa de 20 QPS por conta. Se você gerencia muitas tarefas simultâneas, faça verificações de status em lote, em vez de pesquisar cada tarefa individualmente em alta frequência.

Práticas recomendadas da API de tarefas

Baixe os resultados imediatamente — Os dados da tarefa e URLs de resultados (imagens, vídeos) são retidos apenas por 24 horas. Depois disso, os metadados da tarefa e os arquivos gerados são excluídos automaticamente. Sempre baixe e salve os resultados no armazenamento persistente assim que a tarefa for bem-sucedida.

Lidar com tarefas de múltiplas saídas — Quando você solicita múltiplas saídas (como n=4 imagens), a tarefa está marcada SUCCEEDED se pelo menos uma saída for gerada. Iterar sobre o choices array e verifique finish_reason para lidar com o sucesso parcial:

result = poll_task(task_id)
choices = result["output"]["choices"]
print(f"{len(choices)} outputs generated")

for choice in choices:
  if choice["finish_reason"] == "stop":
    for item in choice["message"]["content"]:
      if "image" in item:
        download(item["image"])
  else:
    print(f"Output skipped: {choice['finish_reason']}")

Evite tarefas duplicadas — Cada solicitação POST cria uma nova tarefa, mesmo com parâmetros idênticos. Se o seu aplicativo tentar novamente um envio com falha, rastreie os IDs das tarefas para evitar trabalho duplicado e custos desnecessários.

API em lote

A API Batch processa grandes volumes de geração de texto ou solicitações de incorporação por 50% do preço padrão. Você carrega um arquivo JSONL de solicitações, aguarda a conclusão do processamento e baixa os resultados.

A API Batch usa o endpoint e o SDK compatíveis com OpenAI. Para obter o guia completo com exemplos, formato de entrada, estados de tarefas e operações de gerenciamento, consulte API Batch.

Erros comuns

Esses erros se aplicam à API Task e à API Batch:

##Próximas etapas

• Gerenciar tarefas assíncronas — Referência da API de tarefas (consulta, lista, cancelamento)
• API de lote — Guia completo da API Batch com todas as linguagens
• Texto para imagem — Geração de imagens com exemplos de tarefas assíncronas
• Texto para vídeo — Geração de vídeo com exemplos de tarefas assíncronas
• Limites de taxa — Detalhes do limite de taxa no nível da conta