이코에코(Eco²) Clean Architecture #13: Scan Worker 마이그레이션 로드맵
작성일: 2026-01-05
상태: Implemented
1. 마이그레이션 배경
1.1 기존 구조의 한계
domains/scan과 domains/_shared/waste_pipeline은 빠른 프로토타이핑에 적합했으나, 시스템이 성장하면서 다음과 같은 문제가 드러났다.
레이어 경계가 모호했다. Vision API 호출, 프롬프트 로딩, 파이프라인 조합이 모두 한 디렉토리에 뒤섞여 있어 모델 확장이 어려웠다. OpenAI만 사용하던 초기와 달리, Gemini 등 멀티모델을 지원해 보다 에이전틱스럽게 작동하길 바랐다.
1.2 마이그레이션 목표
| Clean Architecture 적용 | Application/Infrastructure 레이어 분리 |
| LLM DI | 모델명 기반 런타임 어댑터 선택 |
| Stateless Reducer | Step을 순수 함수로 설계 |
| 체크포인팅 | 실패 시 마지막 성공 지점부터 재시작 |
| SSE 동등성 | 기존 Redis Streams + SSE Gateway 연동 유지 |
2. 핵심 의사결정
2.1 Pipeline은 Application Layer
마이그레이션 초기, waste_pipeline을 어느 레이어에 배치할지 Opus와 논의가 있었다.
Domain Layer 지지측은 "분리배출 규칙은 비즈니스 핵심"이라고 주장했다. 그러나 실제 코드를 분석해보니, pipeline.py의 대부분은 순서 조합, fallback, retry 정책이었다. "플라스틱은 재활용"이라는 도메인 규칙이 아니라, "Vision 실패 시 2초 후 재시도"라는 운영 정책이 주를 이뤘다. waste_pipeline은 유스케이스 오케스트레이션이므로 Application Layer에 배치한다.
Domain Layer: "플라스틱 → 재활용" (불변식, 거의 변경 없음)
Application Layer: "Vision → Rule → Answer 순서, retry 3회" (자주 변경)2.2 Port/Adapter로 LLM 추상화
OpenAI SDK를 직접 호출하던 코드를 VisionModelPort, LLMPort 인터페이스로 추상화했다.
이제 Gemini 외에도 Claude, Grok 등 일정 수준 이상의 Vision과 Strctured Json 아웃풋을 지원한다면 새 어댑터만 구현해 파이프라인 코드 수정 없이 모델을 교체할 수 있다.
# 클라이언트 요청
POST /scan { "model": "gemini-2.5-flash" }
# DI Factory에서 자동 선택
provider = MODEL_PROVIDER_MAP["gemini-2.5-flash"] # → "gemini"
adapter = GeminiVisionAdapter(model="gemini-2.5-flash")명시적 매핑 정책을 채택했다. prefix 기반 추론(gemini-* → gemini)은 오분류 위험이 있어, 지원 모델을 딕셔너리로 명시했다.
| 모델 | Provider |
|---|---|
gpt-5.2, gpt-5.1, gpt-5-mini 등 |
gpt |
gemini-2.5-pro, gemini-2.0-flash 등 |
gemini |
3. 최종 디렉토리 구조
apps/scan_worker/
├── domain/enums/ # 불변 열거형 (WasteCategory, PipelineStage)
│
├── application/
│ ├── classify/
│ │ ├── commands/ # 오케스트레이션
│ │ │ └── execute_pipeline.py # ClassifyPipeline, CheckpointingStepRunner
│ │ ├── steps/ # 순수 Step (Port만 의존)
│ │ │ ├── vision_step.py
│ │ │ ├── rule_step.py
│ │ │ ├── answer_step.py
│ │ │ └── reward_step.py
│ │ ├── ports/ # ABC 인터페이스
│ │ └── dto/ # ClassifyContext
│ └── common/
│ └── step_interface.py # Step ABC
│
├── infrastructure/
│ ├── llm/ # LLM Adapter (모델 패밀리별)
│ │ ├── gpt/
│ │ │ ├── vision.py # GPTVisionAdapter
│ │ │ └── llm.py # GPTLLMAdapter
│ │ └── gemini/
│ │ ├── vision.py # GeminiVisionAdapter
│ │ └── llm.py # GeminiLLMAdapter
│ ├── assets/ # 정적 에셋
│ │ ├── prompts/ # 프롬프트 템플릿 (.txt)
│ │ └── data/ # 분류체계, 규정 JSON
│ ├── asset_loader/ # 에셋 로딩 캡슐화
│ ├── retrievers/ # RAG (JsonRegulationRetriever)
│ └── persistence_redis/ # Redis (Event, Cache, Checkpoint)
│
├── presentation/tasks/ # Celery Task 진입점
└── setup/ # DI Factory, Config4. Port/Adapter 설계
4.1 Port 목록
인프라 의존성을 추상화한 7개 Port를 정의했다.
| Port | 역할 | 구현체 |
|---|---|---|
VisionModelPort |
이미지 분석 | GPTVisionAdapter, GeminiVisionAdapter |
LLMPort |
텍스트 생성 | GPTLLMAdapter, GeminiLLMAdapter |
RetrieverPort |
규정 검색 (RAG) | JsonRegulationRetriever |
PromptRepositoryPort |
프롬프트/스키마 로딩 | FilePromptRepository |
EventPublisherPort |
Redis Streams 발행 | RedisEventPublisher |
ResultCachePort |
결과 캐시 | RedisResultCache |
ContextStorePort |
체크포인팅 | RedisContextStore |
4.2 DI Factory
setup/dependencies.py가 Composition Root 역할을 한다. Celery Task에서는 Factory 함수만 호출하면 된다.
# presentation/tasks/vision_task.py
step = get_vision_step(model) # DI Factory가 적절한 어댑터 주입
runner = get_checkpointing_step_runner()
ctx = runner.run_step(step, "vision", ctx)싱글톤 인스턴스(@lru_cache)와 요청별 인스턴스를 구분했다.
- 싱글톤:
PromptRepository,Retriever,EventPublisher(상태 없음) - 요청별:
VisionModel,LLM(모델명에 따라 다른 인스턴스)
5. Stateless Reducer + 체크포인팅
Stateless Reducer는 각 Step을 순수 함수로 설계하는 패턴이다.
Side effect(이벤트 발행, 체크포인트 저장)는 Runner가 외부에서 처리한다.
체크포인팅은 Step 완료 시 Context를 Redis에 저장하여, 실패 시 마지막 성공 지점부터 재시작할 수 있다. LLM 재호출 비용을 절감한다.
vision ──✅──▶ rule ──✅──▶ answer ──❌(실패)
↓
resume_from_checkpoint("task-123")
↓
rule 체크포인트에서 ctx 복원
↓
answer부터 재시작 (vision, rule 스킵)6. 큐 라우팅
태스크명과 큐명을 1:1로 통일했다. 기존 my.save_character는 제거하고 users.save_character로 대체했다.
| 태스크 | 큐 | 비고 |
|---|---|---|
scan.vision |
scan.vision |
Vision 분석 |
scan.rule |
scan.rule |
규정 검색 |
scan.answer |
scan.answer |
답변 생성 |
scan.reward |
scan.reward |
보상 처리 |
character.match |
character.match |
캐릭터 매칭 (동기 대기) |
character.save_ownership |
character.save_ownership |
소유권 저장 |
users.save_character |
users.save_character |
사용자 캐릭터 저장 |
my.save_character |
- | 제거됨 |
7. SSE + Event Relay 동등성
기존 레거시(domains/scan)의 SSE + Event Relay Layer를 그대로 유지했다.
Step 시작/완료 시 Redis Streams에 이벤트를 발행하고, Event Router가 이를 Pub/Sub으로 브로드캐스트한다.
┌─────────────────────────────────────────────────────────────────────────────┐
│ 이벤트 흐름 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ CheckpointingStepRunner │
│ │ │
│ ├── publish_stage_event("vision", "started") │
│ │ ↓ │
│ │ Redis Streams: scan:events:{shard} │
│ │ ↓ │
│ │ Event Router (consumer) │
│ │ ↓ │
│ │ Redis Pub/Sub: scan:events:{job_id} │
│ │ ↓ │
│ │ SSE Gateway → Client │
│ │ │
│ ├── step.run(ctx) │
│ │ │
│ ├── save_checkpoint("vision", ctx) │
│ │ │
│ └── publish_stage_event("vision", "completed") │
│ │
└─────────────────────────────────────────────────────────────────────────────┘8. 최종 아키텍처
┌─────────────────────────────────────────────────────────────────────────────┐
│ Scan Worker 아키텍처 │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ [Presentation] Celery Task │
│ │ │
│ │ model: "gpt-5.2" │
│ ▼ │
│ [Application] CheckpointingStepRunner │
│ │ │
│ ├── VisionStep ──▶ VisionModelPort ──▶ GPT/Gemini Vision │
│ │ ↓ │
│ │ ✅ checkpoint:vision │
│ │ │
│ ├── RuleStep ────▶ RetrieverPort ────▶ JsonRegulationRetriever │
│ │ ↓ │
│ │ ✅ checkpoint:rule │
│ │ │
│ ├── AnswerStep ──▶ LLMPort ──────────▶ GPT/Gemini LLM │
│ │ ↓ │
│ │ ✅ checkpoint:answer │
│ │ │
│ └── RewardStep ──▶ Celery Tasks │
│ ├── character.save_ownership │
│ └── users.save_character │
│ │
│ ───────────────────────────────────────────────────────────────────────── │
│ │
│ EventPublisherPort → Redis Streams → Event Router → SSE Gateway │
│ │
└─────────────────────────────────────────────────────────────────────────────┘9. 설정 외부화
| 항목 | 권장 방식 | 예시 |
|---|---|---|
| 지원 모델 목록 | MODEL_PROVIDER_MAP 코드 상수 |
모델 추가 시 코드 수정 필요 |
| 내부 서비스 주소 | env/ConfigMap | dev: localhost, prod: k8s-service.namespace |
| CORS, allowed hosts | env/ConfigMap | 운영 정책 변경 시 재배포 없이 반영 |
| API Key | SecretStr + K8s Secret |
로깅에서 마스킹 |