AI Agents · 강의

비동기 작업 흐름 및 백그라운드 작업

장시간 실행되는 에이전트에는 큐(Celery, RQ, Temporal)가 필요합니다. 작업 ID를 반환하고 상태를 조회하게 해보세요.

비동기 작업 흐름 및 백그라운드 작업은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 2번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

이 강의의 일부는 아직 번역되지 않았으며 영어로 표시됩니다.

요청 주기가 너무 느릴 때

일부 에이전트 작업은 30초, 5분 또는 몇 시간이 걸립니다. HTTP 연결을 계속 열어 둘 수는 없습니다. 이러한 작업을 비동기 백그라운드 작업으로 옮기십시오.

패턴

클라이언트가 작업을 POST하면 서버가 작업을 생성하고 작업 ID를 반환합니다
작업자가 큐에서 작업을 가져옵니다
클라이언트가 GET /jobs/{id}를 폴링하여 상태를 확인합니다
완료되면 서버가 결과를 반환합니다

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로 제출하고 ID를 반환한 뒤, 작업자가 큐에서 처리하며 클라이언트가 폴링합니다. 간단한 경우에는 RQ/Celery를, 상태 유지 워크플로에는 Temporal을 사용합니다. 보존 기간, 재시도, 할당량, 추적을 적용합니다.

자주 묻는 질문

“비동기 작업 흐름 및 백그라운드 작업” 강의는 무료인가요?

네 — “비동기 작업 흐름 및 백그라운드 작업” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

“비동기 작업 흐름 및 백그라운드 작업”에서 뭘 배우나요?

장시간 실행되는 에이전트에는 큐(Celery, RQ, Temporal)가 필요합니다. 작업 ID를 반환하고 상태를 조회하게 해보세요. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

AI Agents을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 2번째 강의입니다.

“비동기 작업 흐름 및 백그라운드 작업” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

← AI Agents(으)로 돌아가기