0Pricing
AI Agents · Урок

Асинхронные процессы и фоновые задачи

Долго работающим агентам нужны очереди (Celery, RQ, Temporal): возвращайте идентификатор задачи и проверяйте её статус.

«Асинхронные процессы и фоновые задачи» — бесплатный урок AI Agents на CoddyKit. Это урок 2 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения AI Agents, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс AI Agents содержит 4 уроков всего.

Части этого урока еще не переведены и отображаются на английском.

Когда цикл обработки запроса слишком медленный

Некоторые задачи агентов выполняются 30 с, 5 мин или несколько часов. Нельзя оставлять HTTP-соединение открытым. Перенесите их в асинхронные фоновые задания.

Шаблон

  1. Клиент отправляет POST-запрос с заданием → сервер создаёт задание и возвращает job_id
  2. Обработчик извлекает задание из очереди
  3. Клиент запрашивает состояние через GET /jobs/{id}
  4. После завершения сервер возвращает результат

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')
    }

Обработчик

Отдельный процесс извлекает задания из очереди:

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)})

Варианты очередей

  • RQ (Redis Queue) — минималистичный Python
  • Celery — классический Python-инструмент с множеством функций
  • Temporal — надёжные рабочие процессы, повторные попытки, наблюдаемость
  • Dramatiq — альтернатива Celery
  • Облачные сервисы — Cloud Tasks, SQS, Pub/Sub

Temporal для надёжных рабочих процессов

Рабочие процессы агентов хранят состояние. Модель надёжного выполнения Temporal отлично подходит:

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)

Потоковая передача частичных обновлений

Долго выполняющиеся задания выигрывают от обновлений о ходе выполнения. Используйте события, отправляемые сервером, или 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')

Сроки хранения заданий

Не храните записи о заданиях вечно:

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

Повторные попытки

Временные сбои устраняются автоматически повторными попытками, а постоянные ошибки отправляются в DLQ для проверки человеком:

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'))

Ограничения параллелизма

Чтобы контролировать расходы на GPU/API, ограничьте число одновременно работающих обработчиков:

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

# Or per-queue concurrency in Temporal worker config.

Квоты для каждого пользователя

Отслеживайте выполняющиеся задания для каждого пользователя:

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

Наблюдаемость

Трассируйте каждое задание от начала до конца. Помечайте интервалы трассировки значениями job_id и user_id. Для заданий с ошибками автоматически создавайте заявку или отправляйте оповещение.

Шаблон опроса состояния

Зачем использовать шаблон отправки и опроса состояния для долго выполняющихся агентов?

Итоги

Отправьте задание через POST, верните идентификатор, обработчик выполнит его из очереди, а клиент будет опрашивать состояние. RQ/Celery подходят для простых случаев, Temporal — для рабочих процессов с состоянием. Сроки хранения, повторные попытки, квоты и трассировка.

Часто задаваемые вопросы

Урок «Асинхронные процессы и фоновые задачи» бесплатный?

Да — полный текст урока «Асинхронные процессы и фоновые задачи» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс AI Agents, подпишись на CoddyKit PRO. Курс AI Agents содержит 4 уроков всего.

Чему я научусь в уроке «Асинхронные процессы и фоновые задачи»?

Долго работающим агентам нужны очереди (Celery, RQ, Temporal): возвращайте идентификатор задачи и проверяйте её статус. Ты практикуешь AI Agents с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.

Нужен ли мне опыт, чтобы начать AI Agents?

Предыдущий опыт не требуется. AI Agents на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 2 из 4.

Сколько времени занимает урок «Асинхронные процессы и фоновые задачи»?

Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.

Можно ли писать и запускать код в этом уроке AI Agents?

Да. Каждый урок AI Agents включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.

Все уроки этого курса

  1. Предоставление агентов через API
  2. Асинхронные процессы и фоновые задачи
  3. Ограничение частоты и управление квотами
  4. Сине-зелёные и канареечные развёртывания агентов
← Назад к AI Agents