PricingLoginStart free
AI Agents · Aula

Fluxos de trabalho assíncronos e tarefas em segundo plano

Agentes de longa duração precisam de filas (Celery, RQ, Temporal) — retorne um identificador de tarefa e consulte o status.

Aula 2 de 415 etapas

Fluxos de trabalho assíncronos e tarefas em segundo plano é uma aula grátis de AI Agents no CoddyKit. Esta é a aula 2 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de AI Agents, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de AI Agents inclui 4 aulas no total.

Partes desta aula ainda não foram traduzidas e aparecem em inglês.

Quando o Ciclo de Requisição Está Lento Demais

Algumas tarefas de agentes levam 30s, 5min ou horas. Você não pode manter a conexão HTTP aberta. Mova-as para tarefas assíncronas em segundo plano.

O Padrão

  1. O cliente envia POST da tarefa -> o servidor cria a tarefa e retorna job_id
  2. O trabalhador busca a tarefa em uma fila
  3. O cliente consulta GET /jobs/{id} para obter o status
  4. Quando termina, o servidor retorna o resultado

Submit Endpoint

from uuid import uuid4

@app.post('/jobs')
def submit(req: AgentRequest):
    job_id = str(uuid4())
    redis.hset(f'job:{job_id}', mapping={'status': 'queued', 'user_id': req.user_id})
    queue.enqueue('run_agent_job', job_id, req.query)
    return {'job_id': job_id, 'status': 'queued'}

Status Endpoint

@app.get('/jobs/{job_id}')
def status(job_id: str):
    data = redis.hgetall(f'job:{job_id}')
    if not data:
        raise HTTPException(404)
    return {
        'status': data['status'],
        'result': data.get('result'),
        'error': data.get('error')
    }

Trabalhador

Um processo separado consome a fila:

def run_agent_job(job_id, query):
    redis.hset(f'job:{job_id}', 'status', 'running')
    try:
        result = run_agent(query)
        redis.hset(f'job:{job_id}', mapping={'status': 'done', 'result': result})
    except Exception as e:
        redis.hset(f'job:{job_id}', mapping={'status': 'failed', 'error': str(e)})

Opções de Filas

  • RQ (Fila Redis) — Python minimalista
  • Celery — Python clássico, muitos recursos
  • Temporal — fluxos de trabalho duráveis, novas tentativas, observabilidade
  • Dramatiq — alternativa ao Celery
  • Nativo da nuvem — Cloud Tasks, SQS, Pub/Sub

Temporal para Fluxos de Trabalho Duráveis

Os fluxos de trabalho de agentes mantêm estado. O modelo de execução durável do Temporal é ideal:

import temporalio

@temporalio.workflow.defn
class AgentWorkflow:
    @temporalio.workflow.run
    async def run(self, query: str) -> str:
        plan = await workflow.execute_activity(plan_step, query, schedule_to_close_timeout=timedelta(minutes=2))
        results = await workflow.execute_activity(execute_step, plan, schedule_to_close_timeout=timedelta(minutes=10))
        return await workflow.execute_activity(synthesise_step, results)

Transmissão de Atualizações Parciais

Tarefas de longa duração se beneficiam de atualizações de progresso. Use Eventos Enviados pelo Servidor ou WebSockets:

@app.get('/jobs/{job_id}/stream')
def stream_progress(job_id):
    def gen():
        while True:
            update = redis.brpop(f'updates:{job_id}', timeout=30)
            if not update:
                yield 'data: {"status": "timeout"}\n\n'
                break
            yield f'data: {update[1].decode()}\n\n'
            if 'done' in update[1].decode():
                break
    return StreamingResponse(gen(), media_type='text/event-stream')

Tempos de Expiração das Tarefas

Não mantenha os registros das tarefas para sempre:

redis.expire(f'job:{job_id}', 86400)   # 1 day

Novas Tentativas

Falhas transitórias são submetidas automaticamente a novas tentativas; falhas permanentes vão para uma DLQ para análise humana:

import time

def retry(retries=3, retry_backoff=True):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for attempt in range(1, retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f'attempt {attempt} failed: {e}')
                    if attempt == retries:
                        raise
            return None
        return wrapper
    return decorator

attempts = {'n': 0}

@retry(retries=3, retry_backoff=True)
def run_agent_job(job_id):
    attempts['n'] += 1
    if attempts['n'] < 3:
        raise RuntimeError('transient error')
    return f'job {job_id} done'

print(run_agent_job('job-1'))

Limites de Concorrência

Para controlar os custos de GPU/API, limite o número de trabalhadores simultâneos:

rq worker --burst --max-jobs 1000 --queue agent

# Or per-queue concurrency in Temporal worker config.

Cotas por Usuário

Acompanhe as tarefas em andamento por usuário:

key = f'inflight:{user_id}'
if redis.scard(key) >= 5:
    raise HTTPException(429, 'Too many in-flight jobs')
redis.sadd(key, job_id)

Observabilidade

Rastreie cada tarefa de ponta a ponta. Marque os intervalos com job_id e user_id. Tarefas com falha devem abrir automaticamente um chamado ou gerar um alerta.

Padrão de Consulta de Status

Por que usar o padrão de envio e consulta de status para agentes de longa duração?

Responda na aula interativa

Recapitulação

Envie a tarefa com POST, retorne o identificador, o trabalhador a processa a partir da fila e o cliente consulta. RQ/Celery para casos simples; Temporal para fluxos de trabalho com estado. Tempos de expiração, novas tentativas, cotas e rastreamentos.

Grátis para começar

Aprenda AI Agents com um tutor de IA — grátis

Escreva e execute código real no seu navegador, obtenha ajuda instantânea de um tutor de IA 24/7 e continue de onde parou na web ou no app.

Cursos
60
Aulas
239
Começar a aprender grátisVer o curso completo

Perguntas Frequentes

A aula “Fluxos de trabalho assíncronos e tarefas em segundo plano” é grátis?

Sim — o texto completo de “Fluxos de trabalho assíncronos e tarefas em segundo plano” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de AI Agents, atualize para CoddyKit PRO. O curso de AI Agents inclui 4 aulas no total.

O que vou aprender em “Fluxos de trabalho assíncronos e tarefas em segundo plano”?

Agentes de longa duração precisam de filas (Celery, RQ, Temporal) — retorne um identificador de tarefa e consulte o status. Você pratica AI Agents com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.

Preciso ter experiência prévia para começar AI Agents?

Nenhuma experiência prévia é necessária. AI Agents no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 2 de 4.

Quanto tempo leva a aula “Fluxos de trabalho assíncronos e tarefas em segundo plano”?

A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.

Posso escrever e executar código nesta aula de AI Agents?

Sim. Cada aula de AI Agents inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.

Todas as aulas deste curso

  1. Disponibilizando agentes por trás de uma API
  2. Fluxos de trabalho assíncronos e tarefas em segundo plano
  3. Limitação de taxa e gerenciamento de cotas
  4. Implantações blue-green e canário para agentes
← Voltar para AI Agents