PricingLoginStart free
AI Agents · Lekcja

Asynchroniczne przepływy pracy i zadania w tle

Długotrwałe agenty potrzebują kolejek (Celery, RQ, Temporal) — zwracaj identyfikator zadania i odpytywanie o jego status.

Lekcja 2 z 415 kroki

Asynchroniczne przepływy pracy i zadania w tle to bezpłatna lekcja AI Agents na CoddyKit. To lekcja 2 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Agents, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Agents zawiera 4 lekcji w sumie.

Części tej lekcji nie zostały jeszcze przetłumaczone i są wyświetlane po angielsku.

Gdy cykl obsługi żądania trwa zbyt długo

Niektóre zadania agentów trwają 30 s, 5 min lub kilka godzin. Nie można utrzymywać otwartego połączenia HTTP. Należy przenieść je do asynchronicznych zadań w tle.

Wzorzec

  1. Klient wysyła zadanie metodą POST -> serwer tworzy zadanie i zwraca job_id
  2. Worker pobiera zadanie z kolejki
  3. Klient odpytuje GET /jobs/{id}, aby sprawdzić status
  4. Po zakończeniu serwer zwraca wynik

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

Worker

Osobny proces pobiera zadania z kolejki:

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

Wybór kolejki

  • RQ (Redis Queue) — minimalistyczne rozwiązanie dla Pythona
  • Celery — klasyczne rozwiązanie dla Pythona z wieloma funkcjami
  • Temporal — trwałe przepływy pracy, ponawianie prób i obserwowalność
  • Dramatiq — alternatywa dla Celery
  • Cloud-native — Cloud Tasks, SQS, Pub/Sub

Temporal dla trwałych przepływów pracy

Przepływy pracy agentów mają stan. Model trwałego wykonywania z Temporal doskonale tu pasuje:

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)

Strumieniowanie częściowych aktualizacji

Długotrwałe zadania zyskują na aktualizacjach postępu. Należy użyć Server-Sent Events lub 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')

TTL zadań

Nie należy przechowywać rekordów zadań bezterminowo:

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

Ponawianie prób

Błędy przejściowe są automatycznie ponawiane, a błędy trwałe trafiają do DLQ w celu weryfikacji przez człowieka:

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

Limity współbieżności

Aby kontrolować koszty GPU/API, należy ograniczyć liczbę współbieżnych workerów:

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

# Or per-queue concurrency in Temporal worker config.

Limity na użytkownika

Należy śledzić liczbę zadań będących w toku dla każdego użytkownika:

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

Obserwowalność

Należy śledzić każde zadanie od początku do końca. Spany należy oznaczać za pomocą job_id i user_id. W przypadku nieudanych zadań powinien automatycznie powstać ticket lub alert.

Wzorzec odpytywania statusu

Dlaczego w przypadku długotrwałych agentów warto używać wzorca przesyłania zadania i odpytywania jego statusu?

Odpowiedz w interaktywnej lekcji

Podsumowanie

Wysłanie zadania metodą POST, zwrócenie identyfikatora, przetworzenie zadania przez workera z kolejki i odpytywanie przez klienta. RQ/Celery do prostych zadań; Temporal do przepływów pracy ze stanem. TTL, ponawianie prób, limity i ślady.

Bezpłatny start

Ucz się AI Agents dzięki korepetycjom AI — za darmo

Pisz i uruchamiaj kod w przeglądarce, otrzymuj natychmiastową pomoc od korepetytora AI dostępnego 24/7 i kontynuuj naukę w sieci lub w aplikacji.

Kursy
60
Lekcje
239
Zacznij naukę bezpłatnieWyświetl pełny kurs

Często zadawane pytania

Czy lekcja „Asynchroniczne przepływy pracy i zadania w tle” jest bezpłatna?

Tak — pełny tekst „Asynchroniczne przepływy pracy i zadania w tle” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Agents, przejdź na CoddyKit PRO. Kurs AI Agents zawiera 4 lekcji w sumie.

Co nauczysz się w „Asynchroniczne przepływy pracy i zadania w tle”?

Długotrwałe agenty potrzebują kolejek (Celery, RQ, Temporal) — zwracaj identyfikator zadania i odpytywanie o jego status. Ćwiczysz AI Agents z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć AI Agents?

Nie wymagamy żadnego doświadczenia. AI Agents w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 2 z 4.

Ile czasu zajmuje lekcja „Asynchroniczne przepływy pracy i zadania w tle”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji AI Agents?

Tak. Każda lekcja AI Agents zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

  1. Udostępnianie agentów za API
  2. Asynchroniczne przepływy pracy i zadania w tle
  3. Ograniczanie częstotliwości i zarządzanie limitami
  4. Wdrożenia blue-green i canary dla agentów
← Powrót do AI Agents