LLM 요청이 30초를 넘기기 시작하면: Celery+Redis로 “비동기 추론 파이프라인”을 안전하게 굴리는 법 (2026년 7월 기준)

들어가며

LLM을 백엔드에 붙이면 금방 마주치는 문제가 있습니다.

• 요청 시간이 길다: RAG(검색+리랭크+생성), tool-calling, 멀티스텝 agent는 10초를 훌쩍 넘기고, 외부 API까지 붙으면 1~5분도 흔합니다.
• 동시성/격리: Web API 서버(FastAPI 등)의 worker는 “웹 요청 처리”가 본업인데, LLM 파이프라인을 같은 프로세스에서 돌리면 타임아웃/메모리/커넥션 풀이 같이 죽습니다.
• 재시도/중복 실행/관측성: 네트워크 찰나의 장애로 재시도할 때 “LLM은 돈”이라 중복 호출이 바로 비용 폭탄이 됩니다.

이때 가장 현실적인 선택지가 queue + worker 아키텍처이고, Python 생태계에서는 여전히 Celery + Redis 조합이 많이 쓰입니다. 다만 “그냥 Celery 붙이면 되겠지”로 시작하면, 2026년에도 여전히 ACK/visibility_timeout/prefetch/연결 끊김 같은 고전 함정에 걸립니다(실제로 Redis 연결 순간 끊김 이후 worker가 조용히 consume을 멈추는 사례가 최근에도 공유됨). (reddit.com)

언제 쓰면 좋은가

• HTTP request timeout(예: 30~60s)을 넘는 LLM 작업을 백그라운드로 넘기고, polling/SSE/WebSocket으로 진행률을 보여주고 싶을 때
• RAG indexing, embedding batch, 문서 요약/분할/저장 같은 CPU/IO 혼합 장기 작업
• “모델 호출/툴 호출”을 단계별로 기록/재시도하고 싶을 때

언제 쓰면 안 되는가

• 정확히 한 번(exactly-once) 실행이 비즈니스적으로 절대 조건인데, 인프라/설계로 그 보장을 끝까지 챙길 여력이 없을 때(최소한 idempotency를 전제로 설계해야 함)
• 작업이 본질적으로 “workflow engine” 수준(사람 승인, 장기간 대기, 풍부한 상태머신)이면 Celery만으로는 결국 한계가 와서 Temporal/Prefect/Argo류를 검토하는 편이 낫습니다.
• 브로커로 Redis를 쓸 때는 특히 visibility_timeout과 연결 안정성 튜닝을 못하면 “가끔씩 유실/중복/지연”이 생깁니다. Celery도 Redis에서 visibility timeout(ACK 전 재전달)을 명시적으로 설명합니다. (docs.celeryq.dev)

---

🔧 핵심 개념

1) LLM 비동기 처리의 본질: “요청-응답”이 아니라 “잡 실행(run)” 모델

LLM 파이프라인을 안정적으로 운영하려면 HTTP 요청을 “작업 생성”으로 바꾸고, 결과는 별도 채널로 조회/스트리밍해야 합니다.

• API 서버 역할: job을 enqueue, job_id 반환, 상태 조회/스트리밍 endpoint 제공
• Worker 역할: job을 가져와 실행, 중간 상태/로그/partial output 저장, 완료/실패 마킹

이 구조는 LangGraph Server도 유사하게 설명합니다. “API server가 run을 enqueue하고 queue worker가 실행하며, Redis는 signaling/cancellation/streaming pub/sub에 사용된다”는 점을 명확히 합니다. (langchain-ai.github.io)
(즉, Redis는 ‘데이터 영속 저장소’가 아니라 런타임 신호/큐 통신에 가깝게 쓰는 게 안전합니다.)

2) Celery+Redis에서 꼭 알아야 하는 내부 흐름(ACK/visibility_timeout)

Redis를 broker로 쓸 때 Celery는 “worker가 task를 ACK하기 전까지” 메시지가 안전하게 확정된 것이 아닙니다.

• task_acks_late=True: 작업이 끝난 뒤 ACK → 작업 도중 worker가 죽으면 재전달 가능(중복 실행 가능성은 증가)
• visibility_timeout: ACK를 기다리는 최대 시간. 이 시간을 넘기면 “작업이 죽었다”고 보고 다른 worker에게 재전달될 수 있습니다. Celery 문서가 이 값을 직접 설명합니다. (docs.celeryq.dev)
  • LLM 작업이 20분 걸리는데 visibility_timeout이 10분이면? → 같은 job이 중복 실행될 수 있습니다(비용/데이터 중복).

또 하나의 실무 포인트: Redis 연결이 잠깐 흔들린 뒤 “worker 프로세스는 살아있는데 consume이 멈춤” 같은 형태가 관측됩니다. 이때 에러가 안 튀어 알람이 조용히 지나가는 게 문제고, socket_keepalive, health_check_interval 같은 transport 옵션이 거론됩니다. (reddit.com)

3) “LLM 백엔드 비동기”에서 Celery만으로 부족해지는 지점

• 중복 실행 방지: Celery는 “적어도 한 번(at-least-once)”에 가깝습니다. 따라서 LLM 호출은 idempotency key(작업 키) 기반 dedup를 앱 레벨에서 반드시 설계해야 합니다.
• 결과 저장/스트리밍: Redis result backend에 장문 결과를 쌓는 건 비용/메모리 관점에서 위험합니다. “결과는 DB/S3에, Redis는 상태/락/진행률”로 나누는 편이 운영이 쉽습니다.
• 장시간 작업의 취소(cancellation): async 코드에서만 안전하게 cancel 가능한 프레임워크들도 있고(예: LangGraph의 async task timeout/cancel 관련 언급), sync 작업은 안전한 취소가 어렵습니다. (reference.langchain.com)

---

💻 실전 코드

현실 시나리오: RAG 인덱싱(문서 업로드 → chunking → embedding → 벡터DB upsert) + 진행률 조회

• API 서버는 즉시 job_id 반환
• Worker는 단계별 진행률을 Redis에 기록
• 결과(메타데이터/에러)는 Postgres(예시로 SQLite로 대체) 같은 영속 DB에 저장
• LLM/embedding 호출은 “중복 실행” 대비해 idempotency key를 사용(여기서는 job_id 기반으로 단순화)

0) 의존성 / 실행

python -m venv .venv && source .venv/bin/activate
pip install fastapi uvicorn celery redis sqlalchemy pydantic

# Redis 실행(로컬)
docker run --rm -p 6379:6379 redis:7

# 터미널 1) Celery worker
celery -A app.celery_app worker -l INFO --concurrency=4

# 터미널 2) API 서버
uvicorn app:api --reload --port 8000

1) app.py (FastAPI + Celery + Redis + DB)

# app.py
import json
import os
import time
import hashlib
from datetime import datetime
from typing import Optional, List

import redis
from celery import Celery
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from sqlalchemy import create_engine, Column, String, DateTime, Text
from sqlalchemy.orm import declarative_base, sessionmaker

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
DB_URL = os.getenv("DB_URL", "sqlite:///./jobs.db")

r = redis.Redis.from_url(REDIS_URL, decode_responses=True)

# --- DB (영속 저장소) ---
engine = create_engine(DB_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

class Job(Base):
    __tablename__ = "jobs"
    id = Column(String, primary_key=True)
    status = Column(String, nullable=False)         # PENDING/RUNNING/SUCCEEDED/FAILED
    created_at = Column(DateTime, nullable=False)
    updated_at = Column(DateTime, nullable=False)
    result_json = Column(Text, nullable=True)
    error = Column(Text, nullable=True)

Base.metadata.create_all(bind=engine)

# --- Celery ---
celery_app = Celery(
    "llm_async",
    broker=REDIS_URL,
    backend=REDIS_URL,  # 결과 자체는 DB에 두고, Celery backend는 최소 사용 권장
)

# Redis broker 안정성/장기 작업 튜닝 (핵심)
celery_app.conf.update(
    task_acks_late=True,                 # 작업 끝나고 ACK
    task_reject_on_worker_lost=True,     # worker 죽으면 재큐잉
    worker_prefetch_multiplier=1,        # LLM 장기 작업에서 prefetch로 인한 head-of-line 방지
    broker_transport_options={
        # LLM 작업 최장 시간을 고려해서 설정(예: 1시간)
        # Redis broker에서 ACK 지연 시 재전달 판단 기준
        "visibility_timeout": 3600,      # Celery Redis 문서에서 설명하는 옵션 ([docs.celeryq.dev](https://docs.celeryq.dev/en/v5.6.2/getting-started/backends-and-brokers/redis.html?utm_source=openai))
        # 연결 끊김/idle timeout 완화에 자주 언급되는 옵션들(환경별 검증 필요) ([reddit.com](https://www.reddit.com/r/django/comments/1ummhtw/your_celery_worker_can_show_up_while_it_quietly/?utm_source=openai))
        "health_check_interval": 25,
        "socket_keepalive": True,
    },
)

# --- 진행률/상태 키 규칙 ---
def progress_key(job_id: str) -> str:
    return f"job:{job_id}:progress"

def lock_key(job_id: str) -> str:
    return f"job:{job_id}:lock"

def job_id_from_payload(tenant_id: str, doc_url: str) -> str:
    # 같은 문서/테넌트는 같은 작업으로 dedup하고 싶을 때(예시)
    raw = f"{tenant_id}:{doc_url}".encode("utf-8")
    return hashlib.sha256(raw).hexdigest()[:24]

def set_progress(job_id: str, step: str, percent: int, detail: str = ""):
    r.set(progress_key(job_id), json.dumps({
        "job_id": job_id,
        "step": step,
        "percent": percent,
        "detail": detail,
        "ts": datetime.utcnow().isoformat() + "Z",
    }), ex=3600)  # 1시간 후 만료(필요 시 조정)

# --- API ---
api = FastAPI()

class IndexRequest(BaseModel):
    tenant_id: str
    doc_url: str
    force: bool = False

class JobStatus(BaseModel):
    job_id: str
    status: str
    progress: Optional[dict] = None
    result: Optional[dict] = None
    error: Optional[str] = None

@api.post("/v1/rag/index", response_model=JobStatus)
def enqueue_index(req: IndexRequest):
    job_id = job_id_from_payload(req.tenant_id, req.doc_url)

    db = SessionLocal()
    job = db.get(Job, job_id)

    if job and not req.force and job.status in ("PENDING", "RUNNING", "SUCCEEDED"):
        # 이미 돌고 있거나 완료된 작업 재사용
        return JobStatus(
            job_id=job_id,
            status=job.status,
            progress=json.loads(r.get(progress_key(job_id)) or "null"),
            result=json.loads(job.result_json) if job.result_json else None,
            error=job.error,
        )

    now = datetime.utcnow()
    if not job:
        job = Job(id=job_id, status="PENDING", created_at=now, updated_at=now)
        db.add(job)
    else:
        job.status = "PENDING"
        job.updated_at = now
        job.result_json = None
        job.error = None

    db.commit()
    set_progress(job_id, step="ENQUEUED", percent=0, detail="queued")

    # 실제 작업 enqueue
    index_document.delay(job_id, req.tenant_id, req.doc_url)
    return JobStatus(job_id=job_id, status="PENDING", progress=json.loads(r.get(progress_key(job_id))))

@api.get("/v1/jobs/{job_id}", response_model=JobStatus)
def get_job(job_id: str):
    db = SessionLocal()
    job = db.get(Job, job_id)
    if not job:
        raise HTTPException(404, "job not found")

    prog = r.get(progress_key(job_id))
    return JobStatus(
        job_id=job_id,
        status=job.status,
        progress=json.loads(prog) if prog else None,
        result=json.loads(job.result_json) if job.result_json else None,
        error=job.error,
    )

# --- Worker task ---
@celery_app.task(bind=True, name="index_document", max_retries=3, default_retry_delay=30)
def index_document(self, job_id: str, tenant_id: str, doc_url: str):
    # 중복 실행/동시 실행 방지: Redis lock(간단 버전)
    # (더 강하게 하려면 DB unique + 상태 전이, 또는 분산락 라이브러리/Redis Redlock 등 검토)
    if not r.set(lock_key(job_id), "1", nx=True, ex=3600):
        return  # 이미 누군가 실행 중

    db = SessionLocal()
    job = db.get(Job, job_id)
    try:
        job.status = "RUNNING"
        job.updated_at = datetime.utcnow()
        db.commit()

        set_progress(job_id, "FETCH_DOC", 5, f"fetching {doc_url}")
        text = fetch_document(doc_url)

        set_progress(job_id, "CHUNK", 20, "splitting text")
        chunks = chunk_text(text)

        set_progress(job_id, "EMBED", 40, f"embedding {len(chunks)} chunks")
        vectors = embed_chunks(chunks)  # 여기서 실제 embedding provider 호출한다고 가정

        set_progress(job_id, "UPSERT", 80, "upserting to vector store")
        upsert_vectors(tenant_id, vectors)

        result = {
            "tenant_id": tenant_id,
            "doc_url": doc_url,
            "chunks": len(chunks),
            "indexed_at": datetime.utcnow().isoformat() + "Z",
        }
        job.status = "SUCCEEDED"
        job.result_json = json.dumps(result)
        job.updated_at = datetime.utcnow()
        db.commit()

        set_progress(job_id, "DONE", 100, "completed")

    except Exception as e:
        job.status = "FAILED"
        job.error = repr(e)
        job.updated_at = datetime.utcnow()
        db.commit()
        set_progress(job_id, "FAILED", 100, "error")

        # 네트워크/외부 API 계열이면 retry 고려(에러 타입 분기 추천)
        raise self.retry(exc=e)

    finally:
        r.delete(lock_key(job_id))

# --- 아래는 예시 구현(실서비스에서는 실제 구현으로 교체) ---
def fetch_document(url: str) -> str:
    # 실제로는 S3/HTTP/MinIO 등에서 다운로드
    time.sleep(0.5)
    return ("This is a long document. " * 2000)

def chunk_text(text: str) -> List[str]:
    # 현실에서는 tokenizer 기반 chunking + overlap
    # 예시는 단순 분할
    size = 2000
    return [text[i:i+size] for i in range(0, len(text), size)]

def embed_chunks(chunks: List[str]) -> List[dict]:
    # 여기서 OpenAI embedding 등 호출한다고 가정
    # 비용/재시도/timeout이 핵심이라서, 실제로는 provider별 timeout, backoff, idempotency 설계 필요
    time.sleep(1.0)
    return [{"id": i, "vector": [0.0, 0.1, 0.2]} for i in range(len(chunks))]

def upsert_vectors(tenant_id: str, vectors: List[dict]):
    # Pinecone/Qdrant/pgvector 등으로 upsert
    time.sleep(0.3)

2) 예상 동작(요청/조회)

# 작업 생성
curl -s -X POST localhost:8000/v1/rag/index \
  -H 'content-type: application/json' \
  -d '{"tenant_id":"t1","doc_url":"s3://bucket/doc1.pdf"}' | jq

# 상태 조회
curl -s localhost:8000/v1/jobs/<job_id> | jq

예상 출력 흐름(대략):

• ENQUEUED → FETCH_DOC → CHUNK → EMBED → UPSERT → DONE
• 실패 시 FAILED + error 필드

---

⚡ 실전 팁 & 함정

Best Practice 1) Redis broker를 쓸 거면 visibility_timeout을 “최장 작업 시간”으로 잡아라

Celery Redis 문서가 말하는 그대로, worker가 ACK를 못 하면 메시지가 재전달될 수 있습니다. (docs.celeryq.dev)
LLM 작업은 tail latency가 길어서 평균이 아니라 P99 기준으로 잡아야 합니다.

• 작업이 5~20분까지 튀면 visibility_timeout을 1시간으로 잡고,
• 대신 “실제로 죽은 작업”의 재전달이 늦어지는 트레이드오프를 받아들입니다(운영 알람/헬스체크로 보완).

Best Practice 2) worker_prefetch_multiplier=1로 head-of-line blocking을 막아라

LLM 작업은 긴 작업/짧은 작업이 섞일 때가 많습니다. prefetch가 크면 한 worker가 여러 개를 “미리 잡아” 다른 worker가 놀게 됩니다.
장기 작업 워커는 보통 prefetch=1 + concurrency 적당히가 운영 난이도가 낮습니다.

Best Practice 3) idempotency/dedup은 “필수 기능”으로 취급하라

acks_late를 켰다는 건 곧 “중복 실행이 발생할 수 있다”는 뜻입니다.
LLM 호출(=비용)이라면 더더욱, 최소한 아래 중 하나는 해야 합니다.

• job_id를 입력에서 안정적으로 만들고(tenant+doc_url hash 등) 동일 작업 재요청 시 재사용
• Redis lock / DB 상태 전이로 동시 실행 차단
• 외부 LLM 호출에 request id를 심고(가능한 provider면) 중복 청구 방지

흔한 함정 1) “worker는 살아있는데 consume이 멈췄다”

최근에도 Redis 연결 순간 장애 이후 조용히 멈추는 사례가 공유되고, health_check_interval, socket_keepalive 같은 옵션이 언급됩니다. (reddit.com)
핵심은 프로세스 업/다운이 아니라 ‘진짜로 처리량이 나오고 있는지’를 봐야 한다는 겁니다.

• 알람을 “에러율”이 아니라 queue length, 처리율, oldest job age로 잡으세요.
• worker heartbeat만 보지 말고 “최근 N분 처리 건수=0”도 알람으로.

흔한 함정 2) Redis에 결과를 다 넣고 영속 저장을 안 한다

Redis는 메모리 기반이고 eviction/TTL/재시작 이슈가 있습니다. LangGraph도 Redis를 “ephemeral signaling” 성격으로 설명합니다. (langchain-ai.github.io)
결과/감사 로그/재현에 필요한 데이터는 DB/Object storage로 빼는 편이 안전합니다.

비용/성능/안정성 트레이드오프(현실적인 결론)

• 안정성(중복 방지, 재시도, 관측성)을 챙기려면 코드가 늘어납니다. “Celery로 공짜로 해결”이 아니라, 워커 아키텍처는 결국 제품 기능입니다.
• Redis broker는 셋업이 쉬운 대신, 운영 안정성을 위해 튜닝/모니터링이 필요합니다.
• 장기적으로 agent/workflow가 복잡해지면 Celery의 task primitive만으로는 표현력이 부족해지고, 별도의 orchestration 레이어가 필요해집니다.

---

🚀 마무리

정리하면, 2026년 7월에도 LLM 백엔드에서 Celery+Redis는 여전히 “가성비 좋은 비동기 처리” 옵션입니다. 다만 성공 조건은 명확합니다.

• (필수) acks_late + visibility_timeout을 작업 특성(P99)으로 튜닝할 것 (docs.celeryq.dev)
• (필수) idempotency/dedup을 앱 레벨에서 설계할 것(LLM은 중복 실행이 곧 비용)
• (강력 추천) Redis는 큐/진행률/락 중심, 결과는 DB/Object storage로 분리할 것
• (운영 핵심) “worker가 살아있다”가 아니라 “처리율이 유지된다”를 모니터링할 것(consume 정지류 이슈 대비) (reddit.com)

다음 학습/확장 추천:

• LangGraph Server처럼 “API enqueue + worker 실행 + Redis signaling” 구조를 참고해, 스트리밍/취소/체크포인팅까지 포함한 런타임 설계를 고민해보면 좋습니다. (langchain-ai.github.io)
• OpenAI Responses API의 background mode처럼 “장기 작업을 비동기로 처리하고 polling”하는 제품 패턴도 함께 비교해보면, Celery로 직접 운영할 부분과 외부 API에 맡길 부분을 나누는 데 도움이 됩니다. (openai.com)