From 669fd1141949e31a229554309c105320faabb10e Mon Sep 17 00:00:00 2001 From: SeoHyeon Date: Fri, 5 Jun 2026 00:24:03 +0900 Subject: [PATCH 1/2] =?UTF-8?q?perf(interview):=20=EA=BC=AC=EB=A6=AC?= =?UTF-8?q?=EC=A7=88=EB=AC=B8=20=EC=B2=AB=20=ED=86=A0=ED=81=B0=20=EC=A7=80?= =?UTF-8?q?=EC=97=B0=20=EB=8B=A8=EC=B6=95=20+=20=ED=83=80=EC=9E=90?= =?UTF-8?q?=EA=B8=B0=20=EC=97=B0=EC=B6=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1(AI): followup RAG에서 실패·저속한 LLM 리랭커를 기본 비활성화(followup_rerank_enabled=false)하고 candidate 확장 없이 top_k 직접 검색. _build_rag_context 전체에 타임아웃(followup_rag_timeout_sec=1.5s) 폴백 추가 → 첫 토큰까지 ~3s에서 ~1.5s 목표. P2(FE): SESSION_MESSAGE_DELTA가 짧은 질문에서 0.3s에 몰려 와도 보이도록 useTypewriter로 누적 텍스트를 점진 표시(StageQuestion/QuestionBubble). 긴 스트림은 남은 양 비례 step으로 따라잡아 지연 없음. --- ai/src/ai_server/config/settings.py | 4 + .../messaging/consumers/followup_consumer.py | 27 ++- ai/src/ai_server/messaging/runner.py | 2 + ai/tests/test_followup_consumer.py | 160 ++++++++++++++++++ .../interview/lib/useTypewriter.test.ts | 22 +++ .../features/interview/lib/useTypewriter.ts | 29 ++++ .../interview/model/useLiveInterview.ts | 4 +- .../interview/ui/live/ConversationThread.tsx | 2 +- .../interview/ui/live/InterviewStage.tsx | 2 +- .../interview/ui/live/QuestionBubble.tsx | 9 +- .../interview/ui/live/StageQuestion.tsx | 6 +- 11 files changed, 255 insertions(+), 12 deletions(-) create mode 100644 frontend/src/features/interview/lib/useTypewriter.test.ts create mode 100644 frontend/src/features/interview/lib/useTypewriter.ts diff --git a/ai/src/ai_server/config/settings.py b/ai/src/ai_server/config/settings.py index 092ef5ed..bb56f542 100644 --- a/ai/src/ai_server/config/settings.py +++ b/ai/src/ai_server/config/settings.py @@ -41,6 +41,10 @@ class Settings(BaseSettings): # 리랭킹: 하이브리드 검색으로 후보 N개(candidate_k)를 가져와 LLM 으로 재정렬 후 top_k 주입. rerank_enabled: bool = True rerank_candidate_k: int = 20 + # 꼬리질문 RAG 저지연: 리랭커(LLM 호출, 종종 실패)를 건너뛰고 검색 순서를 그대로 쓴다. + followup_rerank_enabled: bool = False + # RAG 컨텍스트 구성(임베딩+검색+리랭크) 전체 상한. 초과 시 (none) 으로 폴백해 첫 토큰을 막지 않는다. + followup_rag_timeout_sec: float = 1.5 # 질문 풀 초기 크기. Core 의 applyPool 이 questions[0] 만 INSERT 하므로 1 로 고정해 토큰 낭비 차단. # 후속 작업에서 풀 저장 도입 시 늘리기 (예: 5). questions_initial_pool_size: int = 1 diff --git a/ai/src/ai_server/messaging/consumers/followup_consumer.py b/ai/src/ai_server/messaging/consumers/followup_consumer.py index 4dc6606f..1ef8dae5 100644 --- a/ai/src/ai_server/messaging/consumers/followup_consumer.py +++ b/ai/src/ai_server/messaging/consumers/followup_consumer.py @@ -50,6 +50,8 @@ def __init__( tts=None, storage=None, tts_voice: str = "", + rerank_enabled: bool = True, + rag_timeout_sec: float = 1.5, ) -> None: self._generator = generator self._publisher = publisher @@ -65,6 +67,8 @@ def __init__( self._tts = tts self._storage = storage self._tts_voice = tts_voice + self._rerank_enabled = rerank_enabled + self._rag_timeout_sec = rag_timeout_sec async def handle(self, message: AbstractIncomingMessage) -> None: async with message.process(requeue=False): @@ -233,8 +237,22 @@ async def _synth_segment( async def _build_rag_context(self, req: GenerateFollowupRequest) -> str: if not self._core or not self._embedder or not req.context_document_ids: return "(none)" + try: + return await asyncio.wait_for( + self._do_build_rag_context(req), timeout=self._rag_timeout_sec + ) + except asyncio.TimeoutError: + log.warning( + "followup.rag.timeout", + session_id=req.session_id, + timeout_sec=self._rag_timeout_sec, + ) + return "(none)" + async def _do_build_rag_context(self, req: GenerateFollowupRequest) -> str: query = f"{req.previous_question}\n\n{req.answer_text}" + # 리랭크 비활성 시 candidate 확장 없이 top_k 만 검색(LLM 리랭커 호출 회피). + search_k = self._candidate_k if self._rerank_enabled else self._rag_top_k try: query_vec = ( await self._embedder.embed([query], task_type="RETRIEVAL_QUERY") @@ -243,7 +261,7 @@ async def _build_rag_context(self, req: GenerateFollowupRequest) -> str: query_embedding=query_vec, query_text=query, document_ids=req.context_document_ids, - top_k=self._candidate_k, + top_k=search_k, ) except Exception as exc: log.warn("followup.rag.failed", error=str(exc), session_id=req.session_id) @@ -251,9 +269,10 @@ async def _build_rag_context(self, req: GenerateFollowupRequest) -> str: if not hits: return "(none)" - hits = await rerank_hits( - self._reranker, query=query, hits=hits, top_k=self._rag_top_k - ) + if self._rerank_enabled: + hits = await rerank_hits( + self._reranker, query=query, hits=hits, top_k=self._rag_top_k + ) return "\n---\n".join( f"[doc#{h.document_id} chunk#{h.chunk_index}] {h.chunk_text}" for h in hits ) diff --git a/ai/src/ai_server/messaging/runner.py b/ai/src/ai_server/messaging/runner.py index 5bbb368f..b613c624 100644 --- a/ai/src/ai_server/messaging/runner.py +++ b/ai/src/ai_server/messaging/runner.py @@ -213,6 +213,8 @@ def __init__(self, settings: Settings) -> None: tts=tts, storage=storage, tts_voice=settings.openai_tts_voice, + rerank_enabled=settings.followup_rerank_enabled, + rag_timeout_sec=settings.followup_rag_timeout_sec, ) # 종합 피드백 생성 (US-24) diff --git a/ai/tests/test_followup_consumer.py b/ai/tests/test_followup_consumer.py index 2abebacd..1e6837ab 100644 --- a/ai/tests/test_followup_consumer.py +++ b/ai/tests/test_followup_consumer.py @@ -480,3 +480,163 @@ async def test_callback_includes_answer_message_id(): payload = publisher.publish.await_args.kwargs["payload"] assert payload.answer_message_id == 502 # _envelope 의 answerMessageId assert payload.followup_message_id == 503 # _envelope 의 followupMessageId + + +# --------------------------------------------------------------------------- +# 꼬리질문 RAG 저지연 픽스: rerank_enabled=False, timeout 폴백 +# --------------------------------------------------------------------------- + + +def _make_req() -> "GenerateFollowupRequest": + from ai_server.model.messages.followup import GenerateFollowupRequest + + return GenerateFollowupRequest( + session_id=99, + parent_message_id=501, + answer_message_id=502, + followup_message_id=503, + previous_question="결제 outbox 어떻게 구현?", + answer_text="RabbitMQ로 보냈습니다.", + mode="TECHNICAL", + job_category="BACKEND", + context_document_ids=[1], + ) + + +def _make_hit(document_id: int = 1, chunk_index: int = 0, chunk_text: str = "x"): + from types import SimpleNamespace + + return SimpleNamespace( + document_id=document_id, chunk_index=chunk_index, chunk_text=chunk_text + ) + + +@pytest.mark.asyncio +async def test_rag_rerank_skipped_when_disabled(): + """rerank_enabled=False: search_embeddings 는 top_k 만 요청, reranker.rerank 는 호출 안 됨.""" + import asyncio + + from ai_server.messaging.consumers.followup_consumer import FollowupConsumer + from ai_server.messaging.idempotency import LruIdempotencyStore + + hit = _make_hit(chunk_text="이 청크가 반환돼야 한다") + + embedder = MagicMock() + embedder.embed = AsyncMock(return_value=[[0.0]]) + + core = MagicMock() + core.search_embeddings = AsyncMock(return_value=[hit]) + + fake_reranker = MagicMock() + fake_reranker.rerank = AsyncMock(return_value=[0]) + + generator = MagicMock() + generator.generate = AsyncMock() + publisher = MagicMock() + publisher.publish = AsyncMock() + + consumer = FollowupConsumer( + generator=generator, + publisher=publisher, + idempotency=LruIdempotencyStore(max_size=10), + callback_routing_key="callback.questions", + core_client=core, + embedder=embedder, + rag_top_k=5, + reranker=fake_reranker, + candidate_k=20, + rerank_enabled=False, + ) + + req = _make_req() + result = await consumer._build_rag_context(req) + + # 청크 텍스트가 결과에 포함돼야 한다 + assert "이 청크가 반환돼야 한다" in result + # rerank 비활성 시 candidate 확장 없이 rag_top_k 만 검색 + call_kwargs = core.search_embeddings.await_args.kwargs + assert call_kwargs["top_k"] == 5 # rag_top_k, NOT candidate_k(20) + # 리랭커 rerank 는 절대 호출되지 않음 + fake_reranker.rerank.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_rag_rerank_used_when_enabled(): + """rerank_enabled=True: search_embeddings 는 candidate_k, reranker.rerank 는 호출됨.""" + hit = _make_hit(chunk_text="후보 청크") + + embedder = MagicMock() + embedder.embed = AsyncMock(return_value=[[0.0]]) + + core = MagicMock() + core.search_embeddings = AsyncMock(return_value=[hit]) + + fake_reranker = MagicMock() + fake_reranker.rerank = AsyncMock(return_value=[0]) + + generator = MagicMock() + generator.generate = AsyncMock() + publisher = MagicMock() + publisher.publish = AsyncMock() + + consumer = FollowupConsumer( + generator=generator, + publisher=publisher, + idempotency=LruIdempotencyStore(max_size=10), + callback_routing_key="callback.questions", + core_client=core, + embedder=embedder, + rag_top_k=5, + reranker=fake_reranker, + candidate_k=20, + rerank_enabled=True, + ) + + req = _make_req() + result = await consumer._build_rag_context(req) + + assert "후보 청크" in result + call_kwargs = core.search_embeddings.await_args.kwargs + assert call_kwargs["top_k"] == 20 # candidate_k + # 리랭커 rerank 는 호출돼야 한다 + fake_reranker.rerank.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_rag_timeout_returns_none(): + """rag_timeout_sec 초과 시 _build_rag_context 가 '(none)' 을 반환한다.""" + import asyncio + + from ai_server.messaging.consumers.followup_consumer import FollowupConsumer + from ai_server.messaging.idempotency import LruIdempotencyStore + + async def _slow_embed(texts, *, task_type=""): + await asyncio.sleep(0.1) # 타임아웃(0.01s) 보다 훨씬 길다 + return [[0.0]] + + embedder = MagicMock() + embedder.embed = _slow_embed + + core = MagicMock() + core.search_embeddings = AsyncMock(return_value=[]) + + generator = MagicMock() + generator.generate = AsyncMock() + publisher = MagicMock() + publisher.publish = AsyncMock() + + consumer = FollowupConsumer( + generator=generator, + publisher=publisher, + idempotency=LruIdempotencyStore(max_size=10), + callback_routing_key="callback.questions", + core_client=core, + embedder=embedder, + rag_top_k=5, + rag_timeout_sec=0.01, + ) + + req = _make_req() + result = await consumer._build_rag_context(req) + + assert result == "(none)" diff --git a/frontend/src/features/interview/lib/useTypewriter.test.ts b/frontend/src/features/interview/lib/useTypewriter.test.ts new file mode 100644 index 00000000..a40c4e16 --- /dev/null +++ b/frontend/src/features/interview/lib/useTypewriter.test.ts @@ -0,0 +1,22 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest' +import { renderHook, act } from '@testing-library/react' +import { useTypewriter } from './useTypewriter' + +describe('useTypewriter', () => { + beforeEach(() => vi.useFakeTimers()) + afterEach(() => vi.useRealTimers()) + + it('enabled=false 면 전체 텍스트 즉시', () => { + const { result } = renderHook(() => useTypewriter('안녕하세요', false)) + expect(result.current).toBe('안녕하세요') + }) + + it('enabled 면 점진적으로 드러나 결국 전체에 도달', () => { + const { result } = renderHook(() => useTypewriter('가나다라마바사아자차', true)) + // 처음엔 일부만(또는 비어있음) + const initial = result.current.length + act(() => { vi.advanceTimersByTime(35 * 12) }) + expect(result.current).toBe('가나다라마바사아자차') + expect(initial).toBeLessThan('가나다라마바사아자차'.length) + }) +}) diff --git a/frontend/src/features/interview/lib/useTypewriter.ts b/frontend/src/features/interview/lib/useTypewriter.ts new file mode 100644 index 00000000..c6ee7509 --- /dev/null +++ b/frontend/src/features/interview/lib/useTypewriter.ts @@ -0,0 +1,29 @@ +import { useEffect, useRef, useState } from 'react' + +// 스트리밍 텍스트를 일정 속도로 점진 표시(타자기). enabled=false면 전체 즉시 반환. +// fullText 가 늘어나면 따라가고, 남은 양에 비례해 step 을 키워 길고 빠른 스트림은 지연 없이 따라잡는다. +export function useTypewriter(fullText: string, enabled: boolean): string { + const [count, setCount] = useState(enabled ? 0 : fullText.length) + const targetRef = useRef(fullText) + targetRef.current = fullText + + useEffect(() => { + if (!enabled) { + setCount(targetRef.current.length) + return + } + const id = setInterval(() => { + setCount((c) => { + const target = targetRef.current.length + if (c >= target) return c + const remaining = target - c + const step = Math.max(2, Math.ceil(remaining / 6)) + return Math.min(target, c + step) + }) + }, 35) + return () => clearInterval(id) + }, [enabled]) + + const shown = enabled ? Math.min(count, fullText.length) : fullText.length + return fullText.slice(0, shown) +} diff --git a/frontend/src/features/interview/model/useLiveInterview.ts b/frontend/src/features/interview/model/useLiveInterview.ts index 2d70a915..14ba87f0 100644 --- a/frontend/src/features/interview/model/useLiveInterview.ts +++ b/frontend/src/features/interview/model/useLiveInterview.ts @@ -15,7 +15,7 @@ import type { OptimisticAnswer } from './optimistic' import { applyDelta, isStreamingMessage, FOLLOWUP_GENERATING_TEXT } from './streamingBuffer' import type { DeltaPayload } from './streamingBuffer' -export type ThreadItem = Message & { key: string } +export type ThreadItem = Message & { key: string; streaming?: boolean } export type ConnectionStatus = 'connecting' | 'open' | 'closed' export function useLiveInterview(sessionId: number) { @@ -58,7 +58,7 @@ export function useLiveInterview(sessionId: number) { const mergedMessages = serverMessages.map((m) => { const buffered = deltaBuffer[m.id ?? -1] if (buffered !== undefined && isStreamingMessage(m, buffered)) { - return { ...m, content: buffered } + return { ...m, content: buffered, streaming: true as const } } return m }) diff --git a/frontend/src/features/interview/ui/live/ConversationThread.tsx b/frontend/src/features/interview/ui/live/ConversationThread.tsx index 93b2120c..9317ff3c 100644 --- a/frontend/src/features/interview/ui/live/ConversationThread.tsx +++ b/frontend/src/features/interview/ui/live/ConversationThread.tsx @@ -27,7 +27,7 @@ export function ConversationThread({
{items.map((item) => isQuestion(item) ? ( - + ) : ( ), diff --git a/frontend/src/features/interview/ui/live/InterviewStage.tsx b/frontend/src/features/interview/ui/live/InterviewStage.tsx index 5082d4bf..9450562f 100644 --- a/frontend/src/features/interview/ui/live/InterviewStage.tsx +++ b/frontend/src/features/interview/ui/live/InterviewStage.tsx @@ -119,7 +119,7 @@ export function InterviewStage({ {awaitingQuestion || !currentQuestion ? ( ) : ( - + )}
diff --git a/frontend/src/features/interview/ui/live/QuestionBubble.tsx b/frontend/src/features/interview/ui/live/QuestionBubble.tsx index f543659f..f77b6a6c 100644 --- a/frontend/src/features/interview/ui/live/QuestionBubble.tsx +++ b/frontend/src/features/interview/ui/live/QuestionBubble.tsx @@ -3,6 +3,7 @@ import { StatusBadge } from '@/shared/ui/StatusBadge' import { categoryLabel } from '../../lib/categoryLabel' import { useTtsPlayback } from '../../lib/media/useTtsPlayback' import { FOLLOWUP_GENERATING_TEXT } from '../../model/streamingBuffer' +import { useTypewriter } from '../../lib/useTypewriter' function PlayIcon({ playing }: { playing: boolean }) { return ( @@ -15,13 +16,17 @@ function PlayIcon({ playing }: { playing: boolean }) { export function QuestionBubble({ message, autoPlay = false, + streaming = false, }: { message: Message autoPlay?: boolean + streaming?: boolean }) { const label = categoryLabel(message.category) const hasMeta = Boolean(label || message.targetEvidence) const ttsReady = message.ttsStatus === 'SUCCEEDED' + const isSentinel = message.content === FOLLOWUP_GENERATING_TEXT + const shownText = useTypewriter(message.content ?? '', !!streaming && !isSentinel) const { playing, toggle, audioNode } = useTtsPlayback({ sessionId: message.sessionId, @@ -42,14 +47,14 @@ export function QuestionBubble({ )}
- {message.content === FOLLOWUP_GENERATING_TEXT ? ( + {isSentinel ? ( ● ● ● ) : ( -

{message.content}

+

{shownText}

)} {ttsReady && (
diff --git a/frontend/src/features/interview/ui/live/StageQuestion.tsx b/frontend/src/features/interview/ui/live/StageQuestion.tsx index c9ad8886..153b5bb6 100644 --- a/frontend/src/features/interview/ui/live/StageQuestion.tsx +++ b/frontend/src/features/interview/ui/live/StageQuestion.tsx @@ -1,6 +1,7 @@ import type { Message } from '@/domain/session' import { categoryLabel } from '../../lib/categoryLabel' import { useTtsPlayback } from '../../lib/media/useTtsPlayback' +import { useTypewriter } from '../../lib/useTypewriter' function PlayIcon({ playing }: { playing: boolean }) { return ( @@ -11,9 +12,10 @@ function PlayIcon({ playing }: { playing: boolean }) { } // 면접관이 지금 막 던진 한 질문에만 집중시키는 카드. -export function StageQuestion({ question, segmented = false }: { question: Message; segmented?: boolean }) { +export function StageQuestion({ question, segmented = false, streaming = false }: { question: Message; segmented?: boolean; streaming?: boolean }) { const label = categoryLabel(question.category) const ttsReady = question.ttsStatus === 'SUCCEEDED' + const shownText = useTypewriter(question.content ?? '', !!streaming) const { playing, toggle, audioNode } = useTtsPlayback({ sessionId: question.sessionId, @@ -40,7 +42,7 @@ export function StageQuestion({ question, segmented = false }: { question: Messa

- {question.content} + {shownText}

{ttsReady && ( From 08998f490e855112cbc80df561d7e909e22bba83 Mon Sep 17 00:00:00 2001 From: SeoHyeon Date: Fri, 5 Jun 2026 00:42:22 +0900 Subject: [PATCH 2/2] =?UTF-8?q?refactor(ai):=20=EC=8B=A4=ED=8C=A8=C2=B7?= =?UTF-8?q?=EC=A0=80=EC=86=8D=ED=95=9C=20LLM=20=EB=A6=AC=EB=9E=AD=EC=BB=A4?= =?UTF-8?q?=20=EC=A0=9C=EA=B1=B0,=20RAG=EB=A5=BC=20top=5Fk=20=EC=A7=81?= =?UTF-8?q?=EC=A0=91=20=EA=B2=80=EC=83=89=EC=9C=BC=EB=A1=9C=20=EB=8B=A8?= =?UTF-8?q?=EC=88=9C=ED=99=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 리랭커가 프로덕션에서 항상 실패→검색순서 폴백 중이라 가치 0에 ~1-2s만 소모. 모든 RAG 경로(질문풀·꼬리질문·피드백)가 이미 검색순서를 쓰고 있어 제거는 동작 무변경이며 지연·비용만 감소. - rag/reranker.py, chain/prompts/rerank.py, test_reranker.py 삭제 - 3개 consumer: candidate_k 후보 fetch + rerank_hits 제거 → rag_top_k 직접 검색 - settings: rerank_enabled/rerank_candidate_k/followup_rerank_enabled 제거 (followup_rag_timeout_sec 타임아웃 안전망은 유지) --- ai/src/ai_server/chain/prompts/rerank.py | 22 ---- ai/src/ai_server/config/settings.py | 7 +- .../messaging/consumers/feedback_consumer.py | 10 +- .../messaging/consumers/followup_consumer.py | 15 +-- .../messaging/consumers/questions_consumer.py | 10 +- ai/src/ai_server/messaging/runner.py | 9 -- ai/src/ai_server/rag/reranker.py | 119 ------------------ ai/tests/test_followup_consumer.py | 64 +--------- ai/tests/test_questions_consumer.py | 2 +- ai/tests/test_reranker.py | 76 ----------- 10 files changed, 11 insertions(+), 323 deletions(-) delete mode 100644 ai/src/ai_server/chain/prompts/rerank.py delete mode 100644 ai/src/ai_server/rag/reranker.py delete mode 100644 ai/tests/test_reranker.py diff --git a/ai/src/ai_server/chain/prompts/rerank.py b/ai/src/ai_server/chain/prompts/rerank.py deleted file mode 100644 index b5df409b..00000000 --- a/ai/src/ai_server/chain/prompts/rerank.py +++ /dev/null @@ -1,22 +0,0 @@ -# 검색 후보 리랭킹 프롬프트. -# 하이브리드 검색이 가져온 후보 청크들을 쿼리와 함께 한 번에 모델에 넣어, -# 관련도 높은 순으로 인덱스를 재정렬한다 (cross-encoder 대용, 호출 1회). - -SYSTEM_PROMPT = ( - "당신은 검색 결과 재정렬기(reranker)입니다. " - "주어진 쿼리에 대해 후보 청크들의 관련도를 평가하고, " - "가장 관련 높은 순서로 인덱스를 정렬하세요.\n" - "- 관련도는 쿼리의 의도에 대한 답이 청크에 실제로 담겨 있는 정도로 판단합니다.\n" - "- 관련 없는 청크는 결과에서 제외해도 됩니다.\n" - "- 응답은 반드시 지정된 JSON 스키마(ranked_indices)를 따릅니다." -) - -HUMAN_PROMPT = ( - "쿼리:\n{query}\n\n" - "후보 청크 (각 [i] 인덱스):\n" - "---\n" - "{candidates}\n" - "---\n\n" - "관련도가 높은 순으로 최대 {top_k}개의 인덱스를 ranked_indices 에 담아 반환하세요.\n" - "{format_instructions}" -) diff --git a/ai/src/ai_server/config/settings.py b/ai/src/ai_server/config/settings.py index bb56f542..7f51c3a7 100644 --- a/ai/src/ai_server/config/settings.py +++ b/ai/src/ai_server/config/settings.py @@ -38,12 +38,7 @@ class Settings(BaseSettings): ai_realtime_exchange: str = "stackup.realtime" ai_realtime_routing_user: str = "realtime.user.notify" feedback_rag_top_k: int = 5 - # 리랭킹: 하이브리드 검색으로 후보 N개(candidate_k)를 가져와 LLM 으로 재정렬 후 top_k 주입. - rerank_enabled: bool = True - rerank_candidate_k: int = 20 - # 꼬리질문 RAG 저지연: 리랭커(LLM 호출, 종종 실패)를 건너뛰고 검색 순서를 그대로 쓴다. - followup_rerank_enabled: bool = False - # RAG 컨텍스트 구성(임베딩+검색+리랭크) 전체 상한. 초과 시 (none) 으로 폴백해 첫 토큰을 막지 않는다. + # RAG 컨텍스트 구성(임베딩+검색) 전체 상한. 초과 시 (none) 으로 폴백해 첫 토큰을 막지 않는다. followup_rag_timeout_sec: float = 1.5 # 질문 풀 초기 크기. Core 의 applyPool 이 questions[0] 만 INSERT 하므로 1 로 고정해 토큰 낭비 차단. # 후속 작업에서 풀 저장 도입 시 늘리기 (예: 5). diff --git a/ai/src/ai_server/messaging/consumers/feedback_consumer.py b/ai/src/ai_server/messaging/consumers/feedback_consumer.py index 24be44d7..d61dd1cc 100644 --- a/ai/src/ai_server/messaging/consumers/feedback_consumer.py +++ b/ai/src/ai_server/messaging/consumers/feedback_consumer.py @@ -15,7 +15,6 @@ VoiceAnalysisSummary, ) from ai_server.rag.embedder import EmbeddingProvider -from ai_server.rag.reranker import NoopReranker, Reranker, rerank_hits log = structlog.get_logger(__name__) @@ -41,8 +40,6 @@ def __init__( core_client: CoreClient, embedder: EmbeddingProvider | None = None, rag_top_k: int = 5, - reranker: Reranker | None = None, - candidate_k: int = 20, ) -> None: self._generator = generator self._publisher = publisher @@ -51,8 +48,6 @@ def __init__( self._core = core_client self._embedder = embedder self._rag_top_k = rag_top_k - self._reranker = reranker or NoopReranker() - self._candidate_k = max(candidate_k, rag_top_k) async def handle(self, message: AbstractIncomingMessage) -> None: async with message.process(requeue=False): @@ -148,16 +143,13 @@ async def _build_rag_context(self, req: GenerateFeedbackRequest) -> str: query_embedding=query_vec, query_text=last_answer, document_ids=req.context_document_ids, - top_k=self._candidate_k, + top_k=self._rag_top_k, ) except Exception as exc: log.warn("feedback.rag.failed", error=str(exc), session_id=req.session_id) return "(none)" if not hits: return "(none)" - hits = await rerank_hits( - self._reranker, query=last_answer, hits=hits, top_k=self._rag_top_k - ) return "\n---\n".join( f"[doc#{h.document_id} chunk#{h.chunk_index}] {h.chunk_text}" for h in hits ) diff --git a/ai/src/ai_server/messaging/consumers/followup_consumer.py b/ai/src/ai_server/messaging/consumers/followup_consumer.py index 1ef8dae5..cd40f094 100644 --- a/ai/src/ai_server/messaging/consumers/followup_consumer.py +++ b/ai/src/ai_server/messaging/consumers/followup_consumer.py @@ -20,7 +20,6 @@ GenerateFollowupRequest, ) from ai_server.rag.embedder import EmbeddingProvider -from ai_server.rag.reranker import NoopReranker, Reranker, rerank_hits log = structlog.get_logger(__name__) @@ -43,14 +42,11 @@ def __init__( core_client: CoreClient | None = None, embedder: EmbeddingProvider | None = None, rag_top_k: int = 5, - reranker: Reranker | None = None, - candidate_k: int = 20, streaming_generator: StreamingFollowupGenerator | None = None, session_notifier: SessionRealtimeNotifier | None = None, tts=None, storage=None, tts_voice: str = "", - rerank_enabled: bool = True, rag_timeout_sec: float = 1.5, ) -> None: self._generator = generator @@ -60,14 +56,11 @@ def __init__( self._core = core_client self._embedder = embedder self._rag_top_k = rag_top_k - self._reranker = reranker or NoopReranker() - self._candidate_k = max(candidate_k, rag_top_k) self._streaming = streaming_generator self._notifier = session_notifier self._tts = tts self._storage = storage self._tts_voice = tts_voice - self._rerank_enabled = rerank_enabled self._rag_timeout_sec = rag_timeout_sec async def handle(self, message: AbstractIncomingMessage) -> None: @@ -251,8 +244,6 @@ async def _build_rag_context(self, req: GenerateFollowupRequest) -> str: async def _do_build_rag_context(self, req: GenerateFollowupRequest) -> str: query = f"{req.previous_question}\n\n{req.answer_text}" - # 리랭크 비활성 시 candidate 확장 없이 top_k 만 검색(LLM 리랭커 호출 회피). - search_k = self._candidate_k if self._rerank_enabled else self._rag_top_k try: query_vec = ( await self._embedder.embed([query], task_type="RETRIEVAL_QUERY") @@ -261,7 +252,7 @@ async def _do_build_rag_context(self, req: GenerateFollowupRequest) -> str: query_embedding=query_vec, query_text=query, document_ids=req.context_document_ids, - top_k=search_k, + top_k=self._rag_top_k, ) except Exception as exc: log.warn("followup.rag.failed", error=str(exc), session_id=req.session_id) @@ -269,10 +260,6 @@ async def _do_build_rag_context(self, req: GenerateFollowupRequest) -> str: if not hits: return "(none)" - if self._rerank_enabled: - hits = await rerank_hits( - self._reranker, query=query, hits=hits, top_k=self._rag_top_k - ) return "\n---\n".join( f"[doc#{h.document_id} chunk#{h.chunk_index}] {h.chunk_text}" for h in hits ) diff --git a/ai/src/ai_server/messaging/consumers/questions_consumer.py b/ai/src/ai_server/messaging/consumers/questions_consumer.py index caad4fcd..82875b14 100644 --- a/ai/src/ai_server/messaging/consumers/questions_consumer.py +++ b/ai/src/ai_server/messaging/consumers/questions_consumer.py @@ -14,7 +14,6 @@ QuestionPoolCallbackPayload, ) from ai_server.rag.embedder import EmbeddingProvider -from ai_server.rag.reranker import NoopReranker, Reranker, rerank_hits log = structlog.get_logger(__name__) @@ -31,8 +30,6 @@ def __init__( core_client: CoreClient | None = None, embedder: EmbeddingProvider | None = None, rag_top_k: int = 5, - reranker: Reranker | None = None, - candidate_k: int = 20, ) -> None: self._generator = generator self._publisher = publisher @@ -44,8 +41,6 @@ def __init__( self._core = core_client self._embedder = embedder self._rag_top_k = rag_top_k - self._reranker = reranker or NoopReranker() - self._candidate_k = max(candidate_k, rag_top_k) async def handle(self, message: AbstractIncomingMessage) -> None: async with message.process(requeue=False): @@ -138,7 +133,7 @@ async def _build_context(self, req: GenerateQuestionsRequest) -> str: query_embedding=query_vec, query_text=query, document_ids=document_ids, - top_k=self._candidate_k, + top_k=self._rag_top_k, ) except Exception as exc: log.warn("questions.rag.failed", error=str(exc), session_id=req.session_id) @@ -146,9 +141,6 @@ async def _build_context(self, req: GenerateQuestionsRequest) -> str: if not hits: return base_context - hits = await rerank_hits( - self._reranker, query=query, hits=hits, top_k=self._rag_top_k - ) rag_context = "\n---\n".join( f"[doc#{h.document_id} chunk#{h.chunk_index}] {h.chunk_text}" for h in hits ) diff --git a/ai/src/ai_server/messaging/runner.py b/ai/src/ai_server/messaging/runner.py index b613c624..9217323d 100644 --- a/ai/src/ai_server/messaging/runner.py +++ b/ai/src/ai_server/messaging/runner.py @@ -32,7 +32,6 @@ from ai_server.messaging.connection import RabbitConnection from ai_server.rag.chunker import MarkdownChunker from ai_server.rag.embedder import build_embedding_provider -from ai_server.rag.reranker import build_reranker from ai_server.messaging.consumers.feedback_consumer import FeedbackConsumer from ai_server.messaging.consumers.followup_consumer import FollowupConsumer from ai_server.messaging.consumers.questions_consumer import QuestionsConsumer @@ -102,7 +101,6 @@ def __init__(self, settings: Settings) -> None: max_retries=settings.embedding_max_retries, retry_base_delay_sec=settings.embedding_retry_base_delay_sec, ) - reranker = build_reranker(settings, core_client=core_client) vision_pdf_reader = build_vision_pdf_reader(settings, core_client=core_client) # 이력서 PDF @@ -182,8 +180,6 @@ def __init__(self, settings: Settings) -> None: initial_pool_size=settings.questions_initial_pool_size, core_client=core_client, embedder=embedder, - reranker=reranker, - candidate_k=settings.rerank_candidate_k, ) # 꼬리질문 생성 (US-19) @@ -206,14 +202,11 @@ def __init__(self, settings: Settings) -> None: callback_routing_key=settings.ai_callback_routing_questions, core_client=core_client, embedder=embedder, - reranker=reranker, - candidate_k=settings.rerank_candidate_k, streaming_generator=streaming_followup_generator, session_notifier=session_notifier, tts=tts, storage=storage, tts_voice=settings.openai_tts_voice, - rerank_enabled=settings.followup_rerank_enabled, rag_timeout_sec=settings.followup_rag_timeout_sec, ) @@ -228,8 +221,6 @@ def __init__(self, settings: Settings) -> None: callback_routing_key=settings.ai_callback_routing_feedback, core_client=core_client, embedder=embedder, - reranker=reranker, - candidate_k=settings.rerank_candidate_k, rag_top_k=settings.feedback_rag_top_k, ) diff --git a/ai/src/ai_server/rag/reranker.py b/ai/src/ai_server/rag/reranker.py deleted file mode 100644 index 2668705e..00000000 --- a/ai/src/ai_server/rag/reranker.py +++ /dev/null @@ -1,119 +0,0 @@ -from __future__ import annotations - -from typing import Protocol - -import structlog -from langchain_core.output_parsers import PydanticOutputParser -from langchain_core.prompts import ChatPromptTemplate -from langchain_core.runnables import Runnable -from pydantic import BaseModel, Field - -from ai_server.chain.prompts.rerank import HUMAN_PROMPT, SYSTEM_PROMPT -from ai_server.config.settings import Settings -from ai_server.core.client import CoreClient -from ai_server.observability.llm_logging_callback import CoreAiLogCallback - -log = structlog.get_logger(__name__) - - -class RerankResult(BaseModel): - ranked_indices: list[int] = Field( - default_factory=list, - description="relevant candidate indices, most relevant first", - ) - - -class Reranker(Protocol): - async def rerank( - self, *, query: str, candidates: list[str], top_k: int - ) -> list[int]: ... - - -async def rerank_hits( - reranker: Reranker, *, query: str, hits: list, top_k: int -) -> list: - """검색 hit 리스트(.chunk_text 보유)를 query 기준으로 재정렬해 top_k 반환.""" - if not hits: - return list(hits) - indices = await reranker.rerank( - query=query, candidates=[h.chunk_text for h in hits], top_k=top_k - ) - return [hits[i] for i in indices] - - -def _identity(n: int, top_k: int) -> list[int]: - return list(range(min(top_k, n))) - - -# 리랭킹 비활성/미주입 시 검색 순서를 그대로 사용하는 passthrough. -class NoopReranker: - async def rerank( - self, *, query: str, candidates: list[str], top_k: int - ) -> list[int]: - return _identity(len(candidates), top_k) - - -# LLM 으로 후보를 한 번에 재정렬한다. 실패 시 검색 순서로 graceful degrade. -class LlmReranker: - def __init__(self, chain: Runnable) -> None: - self._chain = chain - - async def rerank( - self, *, query: str, candidates: list[str], top_k: int - ) -> list[int]: - if not candidates: - return [] - numbered = "\n\n".join(f"[{i}] {c}" for i, c in enumerate(candidates)) - try: - result = await self._chain.ainvoke( - {"query": query, "candidates": numbered, "top_k": top_k} - ) - seen: set[int] = set() - ordered: list[int] = [] - for i in result.ranked_indices: - if 0 <= i < len(candidates) and i not in seen: - seen.add(i) - ordered.append(i) - if not ordered: - raise ValueError("리랭커가 유효한 인덱스를 반환하지 않음") - return ordered[:top_k] - except Exception as exc: # noqa: BLE001 - log.warning("rerank.failed_fallback_to_search_order", error=str(exc)) - return _identity(len(candidates), top_k) - - -def build_reranker( - settings: Settings, core_client: CoreClient | None = None -) -> Reranker: - if not settings.rerank_enabled: - return NoopReranker() - - from langchain_openai import ChatOpenAI - - parser = PydanticOutputParser(pydantic_object=RerankResult) - prompt = ChatPromptTemplate.from_messages( - [ - ("system", SYSTEM_PROMPT), - ("human", HUMAN_PROMPT), - ] - ).partial(format_instructions=parser.get_format_instructions()) - - callbacks = [] - if core_client is not None: - callbacks.append( - CoreAiLogCallback( - core_client=core_client, - request_type="rerank", - default_model=settings.llm_flash_model, - ) - ) - - llm = ChatOpenAI( - model=settings.llm_flash_model, - temperature=0.0, - api_key=settings.llm_api_key or None, - base_url=settings.llm_base_url, - max_tokens=settings.llm_flash_max_tokens, - callbacks=callbacks, - ) - return LlmReranker(prompt | llm | parser) diff --git a/ai/tests/test_followup_consumer.py b/ai/tests/test_followup_consumer.py index 1e6837ab..9873a6f3 100644 --- a/ai/tests/test_followup_consumer.py +++ b/ai/tests/test_followup_consumer.py @@ -168,7 +168,7 @@ async def _stream(*, on_question_token, **kwargs): call = core.search_embeddings.await_args assert call.kwargs["query_embedding"] == [0.1, 0.2, 0.3] assert call.kwargs["document_ids"] == [7] - assert call.kwargs["top_k"] == 20 # candidate_k (리랭크 후보 수) + assert call.kwargs["top_k"] == 3 # rag_top_k (direct vector search) assert call.kwargs["query_text"] # 하이브리드 검색: 쿼리 텍스트 동봉 assert ( "Outbox rows are inserted in the same transaction" in received_kwargs["context"] @@ -483,7 +483,7 @@ async def test_callback_includes_answer_message_id(): # --------------------------------------------------------------------------- -# 꼬리질문 RAG 저지연 픽스: rerank_enabled=False, timeout 폴백 +# 꼬리질문 RAG 저지연: top_k 직접 검색, timeout 폴백 # --------------------------------------------------------------------------- @@ -512,10 +512,8 @@ def _make_hit(document_id: int = 1, chunk_index: int = 0, chunk_text: str = "x") @pytest.mark.asyncio -async def test_rag_rerank_skipped_when_disabled(): - """rerank_enabled=False: search_embeddings 는 top_k 만 요청, reranker.rerank 는 호출 안 됨.""" - import asyncio - +async def test_rag_searches_top_k_directly(): + """search_embeddings 는 rag_top_k 로 직접 검색하고, 청크 텍스트가 결과에 포함된다.""" from ai_server.messaging.consumers.followup_consumer import FollowupConsumer from ai_server.messaging.idempotency import LruIdempotencyStore @@ -527,9 +525,6 @@ async def test_rag_rerank_skipped_when_disabled(): core = MagicMock() core.search_embeddings = AsyncMock(return_value=[hit]) - fake_reranker = MagicMock() - fake_reranker.rerank = AsyncMock(return_value=[0]) - generator = MagicMock() generator.generate = AsyncMock() publisher = MagicMock() @@ -543,9 +538,6 @@ async def test_rag_rerank_skipped_when_disabled(): core_client=core, embedder=embedder, rag_top_k=5, - reranker=fake_reranker, - candidate_k=20, - rerank_enabled=False, ) req = _make_req() @@ -553,53 +545,9 @@ async def test_rag_rerank_skipped_when_disabled(): # 청크 텍스트가 결과에 포함돼야 한다 assert "이 청크가 반환돼야 한다" in result - # rerank 비활성 시 candidate 확장 없이 rag_top_k 만 검색 - call_kwargs = core.search_embeddings.await_args.kwargs - assert call_kwargs["top_k"] == 5 # rag_top_k, NOT candidate_k(20) - # 리랭커 rerank 는 절대 호출되지 않음 - fake_reranker.rerank.assert_not_awaited() - - -@pytest.mark.asyncio -async def test_rag_rerank_used_when_enabled(): - """rerank_enabled=True: search_embeddings 는 candidate_k, reranker.rerank 는 호출됨.""" - hit = _make_hit(chunk_text="후보 청크") - - embedder = MagicMock() - embedder.embed = AsyncMock(return_value=[[0.0]]) - - core = MagicMock() - core.search_embeddings = AsyncMock(return_value=[hit]) - - fake_reranker = MagicMock() - fake_reranker.rerank = AsyncMock(return_value=[0]) - - generator = MagicMock() - generator.generate = AsyncMock() - publisher = MagicMock() - publisher.publish = AsyncMock() - - consumer = FollowupConsumer( - generator=generator, - publisher=publisher, - idempotency=LruIdempotencyStore(max_size=10), - callback_routing_key="callback.questions", - core_client=core, - embedder=embedder, - rag_top_k=5, - reranker=fake_reranker, - candidate_k=20, - rerank_enabled=True, - ) - - req = _make_req() - result = await consumer._build_rag_context(req) - - assert "후보 청크" in result + # rag_top_k 로 직접 검색 call_kwargs = core.search_embeddings.await_args.kwargs - assert call_kwargs["top_k"] == 20 # candidate_k - # 리랭커 rerank 는 호출돼야 한다 - fake_reranker.rerank.assert_awaited_once() + assert call_kwargs["top_k"] == 5 # rag_top_k @pytest.mark.asyncio diff --git a/ai/tests/test_questions_consumer.py b/ai/tests/test_questions_consumer.py index 275c126a..cd54d0b1 100644 --- a/ai/tests/test_questions_consumer.py +++ b/ai/tests/test_questions_consumer.py @@ -261,7 +261,7 @@ async def test_consumer_injects_initial_rag_chunks_when_available(): call = core.search_embeddings.await_args assert call.kwargs["query_embedding"] == [0.1, 0.2] assert call.kwargs["document_ids"] == [1, 2] - assert call.kwargs["top_k"] == 20 # candidate_k (리랭크 후보 수) + assert call.kwargs["top_k"] == 2 # rag_top_k (direct vector search) assert call.kwargs["query_text"] # 하이브리드 검색: 쿼리 텍스트 동봉 context = generator.generate.await_args.kwargs["context"] assert "Outbox table uses status and retry count" in context diff --git a/ai/tests/test_reranker.py b/ai/tests/test_reranker.py deleted file mode 100644 index f5094417..00000000 --- a/ai/tests/test_reranker.py +++ /dev/null @@ -1,76 +0,0 @@ -from dataclasses import dataclass - -import pytest - -from ai_server.rag.reranker import ( - LlmReranker, - NoopReranker, - RerankResult, - rerank_hits, -) - - -@dataclass -class _Hit: - chunk_text: str - - -class _FakeChain: - def __init__(self, result=None, *, raises: Exception | None = None) -> None: - self._result = result - self._raises = raises - - async def ainvoke(self, _inputs): - if self._raises is not None: - raise self._raises - return self._result - - -@pytest.mark.asyncio -async def test_noop_reranker_returns_identity_truncated() -> None: - r = NoopReranker() - assert await r.rerank(query="q", candidates=["a", "b", "c"], top_k=2) == [0, 1] - - -@pytest.mark.asyncio -async def test_llm_reranker_reorders_by_returned_indices() -> None: - chain = _FakeChain(RerankResult(ranked_indices=[2, 0])) - r = LlmReranker(chain) - assert await r.rerank(query="q", candidates=["a", "b", "c"], top_k=2) == [2, 0] - - -@pytest.mark.asyncio -async def test_llm_reranker_filters_out_of_range_and_dupes() -> None: - chain = _FakeChain(RerankResult(ranked_indices=[1, 1, 9, -1, 0])) - r = LlmReranker(chain) - # 중복/범위밖 제거 후 순서 유지 - assert await r.rerank(query="q", candidates=["a", "b", "c"], top_k=5) == [1, 0] - - -@pytest.mark.asyncio -async def test_llm_reranker_falls_back_to_search_order_on_failure() -> None: - chain = _FakeChain(raises=RuntimeError("llm down")) - r = LlmReranker(chain) - assert await r.rerank(query="q", candidates=["a", "b", "c"], top_k=2) == [0, 1] - - -@pytest.mark.asyncio -async def test_llm_reranker_empty_indices_falls_back() -> None: - chain = _FakeChain(RerankResult(ranked_indices=[])) - r = LlmReranker(chain) - assert await r.rerank(query="q", candidates=["a", "b"], top_k=2) == [0, 1] - - -@pytest.mark.asyncio -async def test_rerank_hits_maps_indices_back_to_hits() -> None: - chain = _FakeChain(RerankResult(ranked_indices=[2, 0])) - r = LlmReranker(chain) - hits = [_Hit("zero"), _Hit("one"), _Hit("two")] - out = await rerank_hits(r, query="q", hits=hits, top_k=2) - assert [h.chunk_text for h in out] == ["two", "zero"] - - -@pytest.mark.asyncio -async def test_rerank_hits_empty_returns_empty() -> None: - out = await rerank_hits(NoopReranker(), query="q", hits=[], top_k=5) - assert out == []