Merge pull request #56 from Team-StackUp/feature/analysis-sse

feat(analysis): 분석 단계별 진행 상황 실시간 표시 (AI→RealTime 직접)

Author: csh1668

ai/src/ai_server/analyzer/_progress.py

@@ -0,0 +1,8 @@
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+
+# analyzer 가 단계마다 호출하는 진행 emitter. (phase, message) 만 받으며,
+# 실제 발행(user/target/trace 바인딩 + RealTime 직접 publish)은 messaging 레이어가 주입한다.
+# analyzer 가 messaging 을 import 하지 않도록(레이어 규칙) 타입만 여기 둔다.
+ProgressEmitter = Callable[[str, str], Awaitable[None]]

ai/src/ai_server/analyzer/repository_analyzer.py

@@ -8,6 +8,7 @@
     EmbeddingStepError,
     chunk_embed_and_upsert,
 )
+from ai_server.analyzer._progress import ProgressEmitter
 from ai_server.analyzer.sources.github_repo import (
     GitHubRepoSourceExtractor,
     RepositoryFetchError,
@@ -66,7 +67,12 @@ async def analyze(
         default_branch: str = "main",
         user_id: int | None,
         analyzed_document_id: int,
+        progress: ProgressEmitter | None = None,
     ) -> RepositoryAnalysisResult:
+        async def emit(phase: str, message: str) -> None:
+            if progress is not None:
+                await progress(phase, message)
+
         if user_id is None:
             raise RepositoryAnalyzeError(
                 code="MISSING_USER_ID",
@@ -86,6 +92,7 @@ async def analyze(
                 code=err.code, message=err.message, retriable=err.retriable
             ) from err
 
+        await emit("EXTRACTING", "레포지토리 코드를 가져오는 중…")
         log.info(
             "repository.extract.start",
             repository_id=repository_id,
@@ -109,6 +116,7 @@ async def analyze(
                 retriable=False,
             )
 
+        await emit("SUMMARIZING", "AI가 코드를 분석·요약하는 중…")
         log.info(
             "repository.llm.start",
             repository_id=repository_id,
@@ -128,6 +136,7 @@ async def analyze(
             md_chars=len(analysis.markdown),
         )
 
+        await emit("EMBEDDING", "분석 결과를 임베딩하는 중…")
         try:
             chunk_count = await chunk_embed_and_upsert(
                 document_id=analyzed_document_id,

ai/src/ai_server/analyzer/resume_analyzer.py

@@ -8,6 +8,7 @@
     EmbeddingStepError,
     chunk_embed_and_upsert,
 )
+from ai_server.analyzer._progress import ProgressEmitter
 from ai_server.analyzer.sources.base import SourceExtractor
 from ai_server.chain.document_analysis_chain import DocumentAnalyzer
 from ai_server.core.client import CoreClient
@@ -61,7 +62,13 @@ async def analyze(
         resume_id: int,
         file_path: str,
         analyzed_document_id: int,
+        progress: ProgressEmitter | None = None,
     ) -> ResumeAnalysisResult:
+        async def emit(phase: str, message: str) -> None:
+            if progress is not None:
+                await progress(phase, message)
+
+        await emit("EXTRACTING", "이력서 텍스트를 추출하는 중…")
         log.info(
             "resume.extract.start",
             resume_id=resume_id,
@@ -76,6 +83,7 @@ async def analyze(
                 retriable=False,
             )
 
+        await emit("SUMMARIZING", "AI가 이력서를 분석·요약하는 중…")
         log.info(
             "resume.llm.start",
             resume_id=resume_id,
@@ -96,6 +104,7 @@ async def analyze(
             md_chars=len(analysis.markdown),
         )
 
+        await emit("EMBEDDING", "분석 결과를 임베딩하는 중…")
         try:
             chunk_count = await chunk_embed_and_upsert(
                 document_id=analyzed_document_id,

ai/src/ai_server/config/settings.py

@@ -32,6 +32,9 @@ class Settings(BaseSettings):
     ai_callback_routing_questions: str = "callback.questions"
     ai_callback_routing_feedback: str = "callback.feedback"
     ai_callback_routing_voice: str = "callback.voice"
+    # AI -> RealTime 직접 발행 (분석 단계 진행 상황). Core 를 거치지 않는 휘발성 알림.
+    ai_realtime_exchange: str = "stackup.realtime"
+    ai_realtime_routing_user: str = "realtime.user.notify"
     feedback_rag_top_k: int = 5
     # 질문 풀 초기 크기. Core 의 applyPool 이 questions[0] 만 INSERT 하므로 1 로 고정해 토큰 낭비 차단.
     # 후속 작업에서 풀 저장 도입 시 늘리기 (예: 5).

ai/src/ai_server/messaging/consumers/repository_consumer.py

@@ -8,6 +8,7 @@
     RepositoryAnalyzer,
 )
 from ai_server.messaging.idempotency import LruIdempotencyStore
+from ai_server.messaging.progress import AnalysisProgressNotifier
 from ai_server.messaging.publisher import CallbackPublisher
 from ai_server.model.envelope import Envelope
 from ai_server.model.messages.analyze import (
@@ -26,11 +27,13 @@ def __init__(
         publisher: CallbackPublisher,
         idempotency: LruIdempotencyStore,
         callback_routing_key: str,
+        progress_notifier: AnalysisProgressNotifier | None = None,
     ) -> None:
         self._analyzer = analyzer
         self._publisher = publisher
         self._idempotency = idempotency
         self._callback_routing_key = callback_routing_key
+        self._progress = progress_notifier
 
     async def handle(self, message: AbstractIncomingMessage) -> None:
         async with message.process(requeue=False):
@@ -92,13 +95,24 @@ async def _run_and_build_payload(
         user_id: int | None,
         trace_id: str,
     ) -> AnalysisCallbackPayload:
+        progress = (
+            self._progress.emitter_for(
+                user_id=user_id,
+                target_type="REPOSITORY",
+                target_id=req.repository_id,
+                trace_id=trace_id,
+            )
+            if self._progress is not None
+            else None
+        )
         try:
             result = await self._analyzer.analyze(
                 repository_id=req.repository_id,
                 repo_full_name=req.repo_full_name,
                 default_branch=req.default_branch,
                 user_id=user_id,
                 analyzed_document_id=req.analyzed_document_id,
+                progress=progress,
             )
         except RepositoryAnalyzeError as err:
             log.warning(

ai/src/ai_server/messaging/consumers/resume_consumer.py

@@ -5,6 +5,7 @@
 
 from ai_server.analyzer.resume_analyzer import ResumeAnalyzeError, ResumeAnalyzer
 from ai_server.messaging.idempotency import LruIdempotencyStore
+from ai_server.messaging.progress import AnalysisProgressNotifier
 from ai_server.messaging.publisher import CallbackPublisher
 from ai_server.model.envelope import Envelope
 from ai_server.model.messages.analyze import (
@@ -23,11 +24,13 @@ def __init__(
         publisher: CallbackPublisher,
         idempotency: LruIdempotencyStore,
         callback_routing_key: str,
+        progress_notifier: AnalysisProgressNotifier | None = None,
     ) -> None:
         self._analyzer = analyzer
         self._publisher = publisher
         self._idempotency = idempotency
         self._callback_routing_key = callback_routing_key
+        self._progress = progress_notifier
 
     async def handle(self, message: AbstractIncomingMessage) -> None:
         async with message.process(requeue=False):
@@ -59,7 +62,9 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
                 trace_id=envelope.trace_id,
             )
 
-            payload = await self._run_and_build_payload(req, envelope.trace_id)
+            payload = await self._run_and_build_payload(
+                req, envelope.trace_id, user_id=envelope.context.user_id
+            )
 
             await self._publisher.publish(
                 routing_key=self._callback_routing_key,
@@ -81,12 +86,25 @@ async def _run_and_build_payload(
         self,
         req: ResumeAnalyzeRequest,
         trace_id: str,
+        *,
+        user_id: int | None,
     ) -> AnalysisCallbackPayload:
+        progress = (
+            self._progress.emitter_for(
+                user_id=user_id,
+                target_type="RESUME",
+                target_id=req.resume_id,
+                trace_id=trace_id,
+            )
+            if self._progress is not None
+            else None
+        )
         try:
             result = await self._analyzer.analyze(
                 resume_id=req.resume_id,
                 file_path=req.file_path,
                 analyzed_document_id=req.analyzed_document_id,
+                progress=progress,
             )
         except ResumeAnalyzeError as err:
             log.warning(

ai/src/ai_server/messaging/progress.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+import structlog
+
+from ai_server.analyzer._progress import ProgressEmitter
+from ai_server.messaging.publisher import CallbackPublisher
+from ai_server.model.envelope import MessageContext
+from ai_server.model.messages.realtime import (
+    AnalysisProgressData,
+    RealtimeNotifyPayload,
+)
+
+log = structlog.get_logger(__name__)
+
+ANALYSIS_PROGRESS_EVENT = "ANALYSIS_PROGRESS"
+
+
+class AnalysisProgressNotifier:
+    """AI -> RealTime 직접 발행. 분석 단계 진행 상황을 user 채널 SSE 로 흘린다.
+
+    진행 정보는 휘발성이라 발행 실패가 분석을 막아선 안 된다 — 모든 예외를 삼키고 경고만 남긴다.
+    """
+
+    def __init__(self, *, publisher: CallbackPublisher, routing_key: str) -> None:
+        self._publisher = publisher
+        self._routing_key = routing_key
+
+    async def emit(
+        self,
+        *,
+        user_id: int | None,
+        target_type: str,
+        target_id: int,
+        phase: str,
+        message: str,
+        trace_id: str,
+    ) -> None:
+        if user_id is None:
+            return
+
+        payload = RealtimeNotifyPayload(
+            event_type=ANALYSIS_PROGRESS_EVENT,
+            data=AnalysisProgressData(
+                target_type=target_type,
+                target_id=target_id,
+                phase=phase,
+                message=message,
+            ),
+        )
+        try:
+            await self._publisher.publish(
+                routing_key=self._routing_key,
+                message_type=self._routing_key,
+                payload=payload,
+                trace_id=trace_id,
+                correlation_id=f"progress-{target_type}-{target_id}-{phase}",
+                context=MessageContext(user_id=user_id),
+            )
+        except Exception:
+            log.warning(
+                "analysis.progress.publish_failed",
+                target_type=target_type,
+                target_id=target_id,
+                phase=phase,
+                trace_id=trace_id,
+            )
+
+    def emitter_for(
+        self,
+        *,
+        user_id: int | None,
+        target_type: str,
+        target_id: int,
+        trace_id: str,
+    ) -> ProgressEmitter:
+        async def _emit(phase: str, message: str) -> None:
+            await self.emit(
+                user_id=user_id,
+                target_type=target_type,
+                target_id=target_id,
+                phase=phase,
+                message=message,
+                trace_id=trace_id,
+            )
+
+        return _emit

ai/src/ai_server/messaging/runner.py

@@ -39,6 +39,7 @@
 from ai_server.messaging.consumers.resume_consumer import ResumeConsumer
 from ai_server.messaging.consumers.web_consumer import WebResumeConsumer
 from ai_server.messaging.idempotency import LruIdempotencyStore
+from ai_server.messaging.progress import AnalysisProgressNotifier
 from ai_server.messaging.publisher import CallbackPublisher
 from ai_server.storage.factory import build_storage
 
@@ -59,6 +60,16 @@ def __init__(self, settings: Settings) -> None:
             exchange_name=settings.ai_callback_exchange,
             publisher_name=settings.ai_publisher_name,
         )
+        # AI -> RealTime 직접 발행용 (분석 단계 진행 상황). Core 콜백 익스체인지와 별개.
+        self._realtime_publisher = CallbackPublisher(
+            connection=self._connection,
+            exchange_name=settings.ai_realtime_exchange,
+            publisher_name=settings.ai_publisher_name,
+        )
+        self._progress_notifier = AnalysisProgressNotifier(
+            publisher=self._realtime_publisher,
+            routing_key=settings.ai_realtime_routing_user,
+        )
         self._idempotency = LruIdempotencyStore(
             max_size=settings.ai_idempotency_lru_size,
         )
@@ -130,12 +141,14 @@ def __init__(self, settings: Settings) -> None:
             publisher=self._publisher,
             idempotency=self._idempotency,
             callback_routing_key=settings.ai_callback_routing_analysis,
+            progress_notifier=self._progress_notifier,
         )
         self._repository_consumer = RepositoryConsumer(
             analyzer=repo_analyzer,
             publisher=self._publisher,
             idempotency=self._idempotency,
             callback_routing_key=settings.ai_callback_routing_analysis,
+            progress_notifier=self._progress_notifier,
         )
         self._web_consumer = WebResumeConsumer(
             analyzer=web_analyzer,
@@ -202,6 +215,7 @@ def __init__(self, settings: Settings) -> None:
     async def start(self) -> None:
         await self._connection.open()
         await self._publisher.open()
+        await self._realtime_publisher.open()
 
         channel = self._connection.channel
 

ai/src/ai_server/model/messages/realtime.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel
+
+from ai_server.model._config import camel_config
+
+
+# AI -> RealTime 직접 발행되는 분석 진행 상황. RealTime 이 user 채널 SSE 로 fan-out 한다.
+# 휘발성(진행 표시용)이라 Core/DB 를 거치지 않는다. 종료 상태(ANALYZED/FAILED)는
+# 기존대로 Core 콜백 -> REPO_STATE/DOC_STATE 로 전달된다.
+class AnalysisProgressData(BaseModel):
+    model_config = camel_config()
+
+    target_type: str  # "REPOSITORY" | "RESUME"
+    target_id: int
+    phase: str  # "EXTRACTING" | "SUMMARIZING" | "EMBEDDING"
+    message: str
+
+
+# RealTime Envelope.payload 형태 ({eventType, data}). Core 의 RealtimeNotifyPayload 와 동일.
+class RealtimeNotifyPayload(BaseModel):
+    model_config = camel_config()
+
+    event_type: str
+    data: AnalysisProgressData