영수증을 읽는 것과 재고 후보를 만드는 것은 다른 작업입니다. 입력 유형을 구분하고, 문자를 읽고, 식품 후보를 고르고, 이름·측정값·날짜를 검증한 뒤 사용자에게 확인받는 순서는 유지합니다.

단계별 외부 큐를 사용하지 않습니다. 단일 FastAPI 프로세스가 인메모리 asyncio.Queue에 분석 작업을 넣고, 소비자 코루틴 2개(ANALYSIS_CONSUMER_COUNT=2)가 큐를 나눠 각자 작업 하나의 전체 파이프라인을 실행합니다. 모델 추론은 원격 GPU Worker에 한 번의 HTTP 호출(POST /internal/v1/infer)로 맡기고, Worker 안에서 OCR과 Gemma는 각각 세마포어 1개로 직렬 실행합니다. 파이프라인은 LangGraph 상태 그래프로 감싸 재시도 가능한 실패(retryable=true)만 1초 뒤 1회 재실행합니다. 멀티스텝 에이전트가 아니라 재시도 래퍼로만 사용합니다.

flowchart TD
    A[POST analyses] --> B[인메모리 큐 등록]
    B --> C[소비자 코루틴]
    C --> D[품질·입력 유형 판별]
    D -- 기타 또는 판독 불가 --> X[미인식·확인 필요 결과]
    D -- 영수증 --> E[OCR·좌표 기반 상품행 구성]
    E --> F[PostgreSQL 상품·별칭·보관기간 조회]
    F --> G{미확정 행이 있는가}
    G -- 예 --> H[Gemma 원문 행 선택]
    G -- 아니오 --> J[후보 결합·검증]
    H --> J
    D -- 실물 --> I[YOLO-World 영역 분리·OCR 힌트·Gemma 비전]
    I --> J
    J --> L[인메모리 완료 결과]
    X --> L
    L --> CB[AI가 백엔드 고정 주소로<br/>완료 결과 콜백]
    CB --> O[사용자 확인 후 백엔드가 MySQL 재고 저장]

한 작업 안에서는 단계가 순서대로 실행됩니다. 소비자 2개가 동시에 Worker를 호출할 수 있지만 Worker의 OCR·Gemma가 각각 1개씩 직렬이므로, 실측상 동시성 2를 넘기면 처리량은 늘지 않고 대기시간만 늘었습니다.

어떤 AI 상태도 사용자 확인 전에 백엔드 MySQL의 ingredients에 저장하지 않습니다. expiration.date가 비어 있는 분석 후보도 반환할 수 있지만, 백엔드는 expiration_date NOT NULL 조건 때문에 사용자가 날짜를 정하기 전에는 재고를 생성하지 않습니다.

작업 객체는 최소한 analysisId, requestId, schemaVersion, pipelineVersion, 현재 stage, 상태, 생성 시각, 이미지 참조와 단계 결과를 가집니다. 전체 OCR 원문과 이미지 바이트를 일반 로그나 큐 항목에 복제하지 않습니다.

동일 프로세스 안에서는 requestId와 요청 내용 해시로 중복 접수를 막습니다. 소비자 하나가 작업을 한 번 꺼내 전체 단계를 실행하므로 메시지 가시성 시간, lease, 단계별 claim, 실패 대기열은 없습니다. 제한된 자체 모델 재시도는 현재 작업의 함수 호출 안에서만 수행합니다.

프로세스가 종료되면 작업과 결과는 복구되지 않습니다. 재시작 후 같은 analysisId를 조회하면 ANALYSIS_NOT_FOUND가 될 수 있습니다. 정상 종료에서도 정해진 유예시간 안에 끝나지 않은 작업은 유실될 수 있으며, 유예시간은 실측 후 확정합니다.

분석 완료 후 AI는 환경설정에 등록된 백엔드 내부 POST /api/v1/internal/image-analyses/callback으로 결과를 전달합니다. 임의 callbackUrl은 받지 않습니다. 백엔드는 analysisId를 기준으로 콜백을 멱등 처리하고 자기 사용자 요청과의 임시 매핑을 유지합니다. GET /ai/v1/analyses/{analysisId}는 정상 폴링이 아니라 예상 완료 제한시간이 지났는데도 콜백이 없는 작업의 제한된 확인용으로만 사용합니다. 완료 제한시간·확인 횟수와 백엔드 임시 매핑 저장 방식·TTL은 별도 확정 대상입니다. 콜백 재시도 기록도 AI 메모리에 있으므로 프로세스 재시작 이후 전달을 보장하지 않습니다.

아래 코드는 구성요소의 책임과 실행 순서를 설명합니다.

# 소비자 settings.analysis_consumer_count개
analysis_queue = asyncio.Queue(maxsize=settings.analysis_queue_maxsize)
analysis_store = InMemoryAnalysisStore()

async def create_analysis(request):
    task = analysis_store.create_or_get(request)
    if task.is_new:
        try:
            analysis_queue.put_nowait(task.analysis_id)
        except asyncio.QueueFull:
            analysis_store.remove(task.analysis_id)
            raise AnalysisQueueFull()
    return task.accepted_response()

async def consume_analyses(pipeline, executor):
    while True:
        analysis_id = await analysis_queue.get()
        try:
            analysis_store.mark_processing(analysis_id)
            result = await pipeline.run(analysis_id, executor)
            analysis_store.complete(analysis_id, result)
            await callback_delivery.try_send(analysis_id, result)  # 콜백 오류를 분석 실패로 바꾸지 않음
        except ExpectedAnalysisError as error:
            analysis_store.fail(analysis_id, error.public_code)
        finally:
            analysis_queue.task_done()

async def run_receipt_path(task, executor, catalog, classifier):
    document = await executor.extract_text(task.object_key)
    safe_document = redact_sensitive_data(document)
    resolved, unresolved = catalog.resolve(build_rows(safe_document))
    if unresolved:
        supplemented = await executor.select_food_rows(classifier, safe_document, unresolved)
        resolved = merge_and_validate_original_rows(resolved, supplemented)
    return validate_and_format(resolved)

원격 GPU Worker 호출(RemoteInferenceClient.infer)이 OCR·Gemma 결과를 한 번에 돌려주고, EC2는 그 결과에 카탈로그·규칙·상태 판정을 적용합니다. 로컬 추론 모드에서는 OCR을 asyncio.to_thread와 잠금으로 한 번에 하나씩 실행합니다. asyncio.Queue를 별도 프로세스가 직접 공유하지 않습니다.

analysis_queue_maxsize는 대기 작업 100개로 제한합니다. 큐가 가득 차면 접수를 기다리게 두지 않고 429 AI_QUEUE_FULL과 “현재 분석 작업량이 너무 많습니다. 잠시 후 다시 시도해 주세요.”를 반환하여 백엔드가 retryAfterMs 이후 같은 requestId로 재시도하거나 직접 입력을 안내하게 합니다. callback_delivery.try_send는 콜백 실패를 내부 전달 상태로 기록하고 제한적으로 재시도하지만, 이미 완료된 분석을 FAILED로 되돌리지 않습니다.

flowchart TD
    A[백엔드 MySQL 재료 조회] --> B[추천 API에 재고 스냅샷 전달]
    B --> C[원본 구성 보존·재료명 정규화<br/>재고·제외 재료 fingerprint 생성]
    C --> D{추천 캐시 hit?}
    D -- 예 --> F[추천 결과 반환]
    D -- 아니오 --> E{AI PostgreSQL에<br/>사용 가능한 레시피가 있는가}
    E -- 예 --> R[레시피 추출·필터·순위]
    E -- 아니오 --> FS[식품안전나라 검색]
    FS --> FM{후보 있음?}
    FM -- 없음 --> TM[TheMealDB 검색]
    TM --> TMM{후보 있음?}
    TMM -- 없음 --> TV[Tavily 최종 검색]
    FM -- 있음 --> V[출처·형식·중복 검증]
    TMM -- 있음 --> V
    TV --> TVQ{저장·재사용 가능한 후보?}
    TVQ -- 있음 --> V
    TVQ -- 없음 --> Z[빈 추천 결과 반환]
    V --> PG[(AI PostgreSQL 레시피 원본)]
    PG --> R
    R --> CS[인메모리 추천 캐시 저장]
    CS --> F
    PG --> G[변경 레시피 동기화 API]
    G --> H[백엔드 주기 작업]
    H --> I[(백엔드 MySQL 레시피 사본)]

재료 데이터 흐름은 백엔드에서 AI로만 향합니다. AI는 전달받은 재고를 검색 입력으로 사용하며 백엔드 재료를 수정하지 않습니다. 레시피 데이터 흐름은 AI PostgreSQL에서 백엔드 MySQL로만 향합니다. 백엔드는 동기화된 레시피 사본을 임의 편집하지 않습니다.

AI PostgreSQL과 백엔드 MySQL의 대리키는 공유하지 않습니다. 레시피는 source + sourceRecipeId로 식별하고 sourceRevision으로 같은 버전인지 확인합니다. 추천 직후 아직 MySQL에 동기화되지 않은 레시피가 있을 수 있으므로 추천 응답은 백엔드 recipeId가 아니라 안정 원천 식별자와 표시용 스냅샷을 반환합니다.

백엔드는 동기화 API를 커서로 읽고 자기 MySQL에 트랜잭션으로 반영합니다. AI가 백엔드 DB에 직접 쓰지 않습니다. 같은 페이지 재처리는 멱등하게 처리하고 원천 삭제는 물리 삭제 대신 INACTIVE로 전달합니다. 검색 공급자는 식품안전나라, TheMealDB, Tavily 순서이며 Tavily는 앞선 두 API에 후보가 없을 때만 호출합니다. 검색·동기화 주기와 PostgreSQL 스키마는 운영 설정으로 관리합니다.

인메모리 큐 도입이 정확도나 처리속도를 개선했다고 주장하지 않습니다. 별도 평가와 실제 EC2 부하 측정이 필요합니다.