이전 글 Async, Thread, Process에서는 작업 부하의 병목에 따라 실행 도구를 나눴다.
- Non-blocking I/O에는 async
- Blocking I/O에는 제한된 thread 위임
- CPU-bound 계산에는 process나 별도 연산 서비스
그다음 흔히 이런 질문이 나온다.
그러면 worker는 process인가? Thread pool의 thread 하나를 worker라고 부르는 것인가?
둘 다 가능하지만 핵심을 놓친 질문이다. Thread와 process는 코드가 어떻게 실행되는가를 설명한다. Worker는 누가 작업을 가져가고, 진행 상태와 실패를 관리하며, 언제 책임을 내려놓는가를 설명한다.
실행 방식 작업 부하의 책임
─────────────────────────────── ─────────────────────────────
async task / thread / process API / queue / worker
어떻게 동시에 진행·계산하는가? 누가 수명 주기를 책임지는가?이 둘을 같은 축의 선택지로 놓으면 Agent backend의 중요한 실패 양상을 놓치게 된다.
가장 짧은 구현: API에서 백그라운드 task를 만든다
Agent run을 비동기로 시작하는 가장 간단한 코드는 다음과 같다.
import asyncio
from fastapi import FastAPI
app = FastAPI()
@app.post("/analysis", status_code=202)
async def start_analysis(request: AnalysisRequest):
run_id = new_run_id()
asyncio.create_task(run_analysis(run_id, request))
return {"run_id": run_id, "status": "started"}엔드포인트는 즉시 202 Accepted를 반환하고 coroutine은 같은 event loop에서 계속 실행된다. 시연에서는 잘 작동한다. 하지만 HTTP 응답을 먼저 반환했다고 작업이 지속 가능해진 것은 아니다.
이 task를 실제로 책임지는 것은 API process다.
API process
├─ HTTP 요청 처리기 ───────> 202 Accepted
└─ asyncio task ──────────> long-running analysis
API process 종료
└─ task도 함께 종료순차 배포, 메모리 부족 종료, 애플리케이션 충돌, 컨테이너 재시작이 발생하면 메모리 내 task도 함께 사라질 수 있다. 사용자는 이미 run_id를 받았지만 그 run을 다시 발견하거나 이어갈 영속적 책임 주체가 없다.
Python의 asyncio.create_task()는 현재 event loop에 coroutine을 등록하는 API다. 공식 문서는 백그라운드 task가 실행 중 사라지지 않도록 참조를 보관하고, 완료된 task의 예외도 회수해야 한다고 명시한다. (Coroutines and Tasks) 이 규칙을 지켜도 process 실패를 넘어서는 지속성까지 생기지는 않는다.
background_tasks: set[asyncio.Task] = set()
def start_local_task(coroutine) -> None:
task = asyncio.create_task(coroutine)
background_tasks.add(task)
task.add_done_callback(background_tasks.discard)이 코드는 task 참조 관리 문제를 해결할 뿐이다. API process가 종료되면 집합과 task가 모두 사라진다.
API와 worker를 분리한다는 것은 무엇인가
“API와 worker를 분리한다”는 말에는 논리적 책임과 배포 수명 주기 두 층이 있다. 개발 환경에서는 API와 worker를 한 process나 한 Pod에 둘 수 있다. 그래도 HTTP 접수와 장시간 run 실행은 코드 경계로 나눌 수 있다.
운영 환경에서는 보통 API Deployment와 Worker Deployment를 서로 다른 Pod 집합으로 둔다. API는 요청량·연결 수 기준으로, worker는 queue 길이·작업 시간·CPU·GPU·외부 API 부하 기준으로 확장한다. API Pod가 재시작되어도 queue와 PostgreSQL에 남은 run은 worker가 계속 처리하거나 다른 worker가 회수할 수 있어야 한다.
개발
API와 worker를 한 process 또는 한 Pod에서 실행할 수 있음
- 코드 책임은 분리
- 배포·장애 수명은 공유
운영
별도 Deployment · Pod 집합 — 확장·배포·장애 수명이 분리됨
별도 Pod는 목적이 아니라 결과다. 서로 다른 책임·자원·확장·실패 수명을 독립적으로 관리하려다 보니 Deployment가 갈라진다. 전체 지도는 Agent Backend 한 장 지도 2편을 참고한다.
Process 내부의 백그라운드 task가 항상 나쁜 것은 아니다
API process에서 응답 이후 작업을 실행하는 것 자체가 잘못은 아니다. FastAPI도 BackgroundTasks를 제공하며, 작은 응답 후 작업에는 유용하다고 설명한다. 같은 공식 문서는 여러 process나 서버에서 실행할 무거운 백그라운드 계산이라면 queue를 사용하는 더 큰 도구가 적합할 수 있다고 구분한다. (FastAPI Background Tasks)
Process 내부 task는 다음 조건에서 합리적일 수 있다.
- 실행 시간이 짧다.
- 실패하면 다음 request에서 다시 계산해도 된다.
- 결과를 지속적으로 추적할 필요가 없다.
- 중간 진행 상황이나 취소 기능을 제공하지 않는다.
- 배포 중 유실되어도 사업상 영향이 작다.
예를 들어 가능한 범위에서 수행하는 캐시 예열이나 중요하지 않은 원격 측정 정보 전송은 이 범주에 들어갈 수 있다. 반대로 다음 조건이 하나라도 중요하면 책임을 process 밖으로 옮기는 편이 자연스럽다.
- 사용자가
run_id로 상태와 결과를 다시 조회해야 한다. - 수십 초에서 수분 이상 실행될 수 있다.
- 실패 시 재시도하거나 다른 worker가 회수해야 한다.
- 부수 효과가 있어 중복 실행을 제어해야 한다.
- API와 다른 CPU, 메모리, GPU 자원 특성에 맞춰 확장해야 한다.
- 클라이언트 연결 종료와 무관하게 실행을 계속해야 한다.
경계는 특정 시간 숫자 하나로 정해지지 않는다. 5초짜리 결제 작업은 지속적인 책임 주체가 필요할 수 있고, 30초짜리 재생성 가능한 미리보기는 요청 안에서 기다려도 될 수 있다. 핵심은 지연 시간이 아니라 유실과 중복 실행을 시스템이 어떻게 다뤄야 하는가다.
Queue는 실행의 접수 기록을 process 밖에 남긴다
API와 worker 사이에 queue를 두면 실행 요청이 특정 API process의 메모리에서 벗어난다.
Client
│ POST /runs
▼
API ── 검증 / 권한 확인 / run 생성 ──> Queue
│ │
└────────── 202 + run_id ▼
Worker
│
├─ 작업 확보
├─ Agent 실행
├─ 진행 상황 갱신
├─ 결과 저장
└─ 처리 완료 통지(ack)API process가 응답 직후 재시작되어도 queue에 제출된 작업은 다른 소비자가 가져갈 수 있다. 물론 queue라는 상자를 그렸다고 모든 실패가 해결되는 것은 아니다. 영속적 제출, 작업 확보와 임대, 처리 완료 통지(ack), 재시도, 처리 불가 메시지 정책이 실제로 설계되어야 한다.
API의 책임은 실행 자체보다 안전하게 접수하는 것에 가까워진다.
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.post("/runs", status_code=202)
async def create_run(request: RunRequest, user: User = Depends(current_user)):
authorize(user, request)
validated = validate_run_request(request)
run_id = new_run_id()
await run_store.create(
run_id=run_id,
user_id=user.id,
status="submitting",
input=validated,
)
try:
await job_queue.publish(
{"run_id": run_id, "input": validated}
)
except Exception:
await run_store.mark_submission_failed(run_id)
raise HTTPException(503, "Could not submit run")
await run_store.mark_queued(run_id)
return {"run_id": run_id, "status": "queued"}이 예시는 책임 이관을 보여주기 위한 최소 구조다. 데이터베이스 쓰기에는 성공했지만 queue 게시 직전에 process가 죽는 이중 쓰기 실패까지 원자적으로 해결하지는 않는다. 운영 환경에서는 트랜잭션 아웃박스처럼 접수 기록과 게시를 연결하는 방법이나, queue 자체를 기준 정보로 삼는 방법이 필요하다. 중요한 것은 202를 반환하기 전에 시스템이 나중에 다시 발견할 수 있는 실행 의도를 남기는 것이다.
Worker는 현재 시도의 수명 주기를 책임진다
Worker는 queue에서 메시지를 읽는 process라는 구현 형태보다 그 메시지를 받은 뒤 수행하는 책임으로 정의하는 편이 정확하다.
대기 중
│ 확보
▼
확보됨 ──> 실행 중 ──> 결과 저장됨 ──> 처리 완료 통지(ack)
│
└────> 실패 / 재시도 가능 / 최종 종료일반적인 worker의 책임은 다음과 같다.
- 실행 가능한 작업을 확보한다.
- Run과 attempt 상태를
running으로 전환한다. - Agent graph와 tool orchestration을 실행한다.
- 진행 상황과 이벤트를 외부 저장소에 기록한다.
- 결과 또는 최종 실패를 지속적으로 보존한다.
- 저장이 끝난 뒤 queue 메시지의 완료를 확인한다.
Plain Python으로 표현하면 핵심 순서는 다음과 같다.
async def handle_message(message: JobMessage) -> None:
run_id = message.run_id
attempt_id = await run_store.start_attempt(run_id)
try:
result = await execute_agent(
run_id=run_id,
payload=message.payload,
)
await run_store.complete_attempt(
run_id=run_id,
attempt_id=attempt_id,
result=result,
)
except RetryableError as error:
await run_store.record_retryable_failure(
run_id, attempt_id, error
)
await message.retry()
return
except Exception as error:
await run_store.fail_attempt(
run_id, attempt_id, error
)
finally:
await release_run_resources(run_id)
# Terminal result가 durable하게 기록된 뒤에만 완료 처리한다.
await message.ack()이 코드도 queue 제품별 의미를 생략한 개념 예시다. 일부 queue에서는 재시도가 부정 확인 응답(NACK)이고, 일부에서는 가시성 시간 초과나 대기 항목 회수로 구현된다. 그 차이는 10편 Worker가 죽을 때에서 다룬다.
여기서 중요한 순서는 실행 → 최종 상태 저장 → 처리 완료 통지(ack)다. 먼저 처리 완료를 통지(ack)한 뒤 결과 저장에 실패하면 queue에는 완료된 것으로 보이지만 사용자는 결과를 찾을 수 없다. 반대로 결과를 저장하고 처리 완료 통지(ack) 직전에 worker가 죽으면 같은 메시지가 다시 전달될 수 있다. 그래서 worker 아키텍처는 최소 한 번 전달과 멱등성 문제를 없애는 것이 아니라 드러낸다. 결과 저장과 ack 사이의 중복 가능성과 멱등 부수 효과는 Agent Tool은 정확히 한 번 실행될 수 있을까에서 다룬다.
Worker 하나가 job 하나만 실행해야 하는 것은 아니다
Worker를 OS thread 하나와 동일시하면 worker process 안의 동시성을 설계하기 어렵다. I/O 중심 Agent라면 worker process 하나가 asyncio event loop를 실행하면서 여러 작업을 제한된 수만큼 진행할 수 있다.
Worker process
└─ asyncio event loop
├─ run A ── await LLM ────── resume
├─ run B ── await DB ─────── resume
└─ run C ── await search ─── resume동시 실행 수에는 반드시 상한이 있어야 한다. Queue에 메시지가 많다는 이유로 모두 task로 만들면 LLM 호출량 제한, 데이터베이스 연결 풀, 메모리가 먼저 고갈된다.
import asyncio
MAX_IN_FLIGHT = 8
async def worker_main() -> None:
in_flight: set[asyncio.Task] = set()
while True:
while len(in_flight) < MAX_IN_FLIGHT:
message = await job_queue.receive()
task = asyncio.create_task(handle_message(message))
in_flight.add(task)
done, in_flight = await asyncio.wait(
in_flight,
return_when=asyncio.FIRST_COMPLETED,
)
for task in done:
# handle_message가 예상하지 못한 오류로 끝났다면 여기서 관찰한다.
await task실제 worker는 종료 신호를 받으면 새 메시지 확보를 중단하고, 진행 중인 작업을 제한 시간 안에 마치거나 체크포인트를 남긴 뒤 종료해야 한다. MAX_IN_FLIGHT도 단순 상수가 아니라 메모리, 하위 서비스 할당량, 작업 비용에 맞춰 조정해야 한다.
CPU-bound 작업 부하라면 같은 worker 역할을 다른 방식으로 구현할 수 있다.
I/O worker process
└─ asyncio event loop + bounded job concurrency
CPU worker pool
└─ 여러 process, 보통 할당된 CPU 슬롯마다 task 하나
GPU inference worker
└─ 모델 서버 + 배치 처리 + GPU 메모리 관리즉 worker는 역할이고 async/thread/process는 그 역할을 구현하는 도구다.
API와 worker의 책임을 분리하면 무엇이 달라지는가
| 책임 | API | Worker |
|---|---|---|
| 인증 | 요청 주체 확인 | 검증된 주체의 맥락 사용 |
| 권한 부여 | Run 생성 권한 확인 | 도구 실행 시 세부 정책 재검증 가능 |
| 요청 검증 | 입력 스키마와 제한 검사 | 실행 시 필요한 불변 조건 확인 |
| Run 생성 | run_id와 접수 상태 생성 | 시도 생성과 실행 중 상태 전환 |
| Queue | 작업 제출 | 작업 확보, 처리 완료 통지(ack), 재시도 |
| Agent 실행 | 짧은 동기 실행만 선택적으로 | 장시간 그래프와 도구 오케스트레이션 |
| 진행 상황과 이벤트 | 조회와 SSE 전달 | 상태 갱신과 이벤트 게시 |
| 결과 | 조회 엔드포인트 제공 | 결과와 최종 실패 저장 |
| 확장 신호 | 요청률, 연결 수 | Queue 적재량, run 지연 시간, 자원 사용량 |
이 분리는 코드베이스를 반드시 두 저장소로 나누라는 뜻이 아니다. 같은 도메인 모델과 애플리케이션 패키지를 공유할 수 있다. 중요한 것은 process 수명 주기와 확장 단위를 분리하는 것이다.
API는 짧고 예측 가능한 요청 처리에 최적화할 수 있다. Worker는 긴 시간 초과, 도구 자격 증명, CPU/GPU 자원, 재시도 정책에 맞춰 따로 배포하고 확장할 수 있다. PDF 파싱 worker가 밀려도 상태 조회 엔드포인트와 SSE 연결까지 같이 고갈되지 않도록 실패 영역도 분리된다.
Worker가 자동으로 보장하지 않는 것
Worker를 추가하면 architecture diagram은 그럴듯해지지만 다음 성질이 저절로 생기지는 않는다.
| 필요한 성질 | 별도로 필요한 설계 |
|---|---|
| Process 재시작 이후 복구 | 영속 큐와 재전달 |
| 중복 실행 안전성 | 멱등성 키와 부수 효과 방지 장치 |
| 긴 run 재개 | 체크포인트와 재개 가능한 상태 |
| 사용자 취소 | 영속적 취소 프로토콜 |
| 진행 상황 조회 | 외부 상태와 이벤트 저장소 |
| 처리 불가 메시지 격리 | 최대 시도 횟수와 처리 불가 메시지 정책 |
| 과부하 방지 | 유입 제어와 역압 |
| 누가 실행 중인지 확인 | 임대, heartbeat 또는 대기 작업의 책임 정보 |
이것이 queue와 worker를 create_task()의 더 무거운 버전으로 보면 안 되는 이유다. Worker 경계는 복구를 설계할 자리를 만들어주지만, 복구 정책 자체는 아니다.
Kubernetes에서 worker는 특별한 객체가 아니다
Kubernetes에는 Deployment, StatefulSet, Job 같은 작업 부하 자원이 있다. 그러나 애플리케이션 아키텍처에서 말하는 worker는 Kubernetes 객체 이름이 아니다.
API와 worker가 같은 container image를 사용해도 된다.
image: my-agent:1.0
API Deployment
명령: uvicorn app.api:app
Worker Deployment
명령: python -m app.worker두 Deployment는 같은 코드와 이미지를 공유하지만 서로 다른 명령, 복제본 수, 자원 요청량, 자동 확장 신호, 비밀 정보 범위를 가질 수 있다. Kubernetes Deployment는 애플리케이션 작업 부하의 Pod 집합과 순차 배포를 관리하고, Pod는 영구적인 개체가 아니라 비교적 일시적인 실행 단위로 취급된다. (Deployments, Pod Lifecycle) 따라서 진행 상태를 worker Pod의 메모리에만 두면 Pod 교체와 함께 사라질 수 있다.
Kubernetes의 Job은 완료될 때까지 하나 이상의 Pod를 실행하는 별도 컨트롤러다. (Jobs) 애플리케이션의 모든 Agent run마다 Kubernetes Job을 만들어야 한다는 뜻은 아니다. 짧은 run이 많은 시스템은 장시간 실행되는 worker Deployment가 queue를 소비하는 편이 단순할 수 있고, 강한 자원 격리가 필요한 큰 일괄 작업은 Kubernetes Job이 맞을 수 있다. 선택은 작업 부하의 세분성, 시작 비용, 격리 요구에 달려 있다.
Kubernetes 배포와 자동 확장은 11편에서 더 자세히 다룬다. 지금 필요한 개념 모형은 하나다.
컨테이너가 worker인 것이 아니다. Queue에서 받은 실행 시도를 책임지는 애플리케이션 역할이 worker다.
요청에서 worker로 책임을 넘기는 판단 기준
최종적으로는 다음 질문으로 결정할 수 있다.
- 클라이언트 연결이 끊겨도 실행을 계속해야 하는가?
- API 배포나 충돌 뒤에도 작업을 다시 찾아야 하는가?
- 재시도, 취소, 진행 상황, 감사 기록이 필요한가?
- API와 다른 자원 특성이나 확장 정책이 필요한가?
- 같은 run이 다시 전달될 때 안전하게 처리할 수 있어야 하는가?
대부분 아니오이고 작업이 짧다면 요청 처리기가 직접 await하는 구조가 가장 단순하다. 일부만 예여도 queue와 worker 경계를 검토할 이유가 생긴다. 아키텍처는 구성 요소 수가 아니라 필요한 실패 의미에 의해 커져야 한다.
Worker를 도입하면 Agent run의 owner는 명확해진다. 하지만 다음 문제가 곧바로 나타난다.
- Worker가 실행 중인 현재 상태는 어디에 저장하는가?
- API는 진행 상황을 어떻게 읽는가?
- Worker가 죽은 뒤 새 worker는 어디에서 이어가는가?
- 최종 결과와 임시 작업 상태를 같은 저장소에 넣어도 되는가?
메모리 내 Python 객체만으로는 process 경계를 넘을 수 없다. 다음 글 Agent State는 어디에 있어야 하는가에서는 Agent state를 PostgreSQL, Redis, checkpoint로 나누고, 영속 기록과 자주 변하는 hot projection의 역할을 구분한다.
Discussion
토론
질문, 정정, 반례, 다른 해석이 있다면 남겨 주세요.