diff --git a/ai/src/ai_server/chain/followup_generation_chain.py b/ai/src/ai_server/chain/followup_generation_chain.py index a787b5aa..132d8fa3 100644 --- a/ai/src/ai_server/chain/followup_generation_chain.py +++ b/ai/src/ai_server/chain/followup_generation_chain.py @@ -27,6 +27,7 @@ async def generate( mode: str, previous_question: str, answer_text: str, + context: str = "(none)", ) -> FollowupResult: ... @@ -41,6 +42,7 @@ async def generate( mode: str, previous_question: str, answer_text: str, + context: str = "(none)", ) -> FollowupResult: result = await self._chain.ainvoke( { @@ -48,6 +50,7 @@ async def generate( "mode": mode, "previous_question": previous_question, "answer_text": answer_text, + "context": context, } ) if not isinstance(result, FollowupResult): diff --git a/ai/src/ai_server/chain/prompts/followup_generation.py b/ai/src/ai_server/chain/prompts/followup_generation.py index e8c10d17..e863776d 100644 --- a/ai/src/ai_server/chain/prompts/followup_generation.py +++ b/ai/src/ai_server/chain/prompts/followup_generation.py @@ -18,7 +18,12 @@ HUMAN_PROMPT = ( "직군: {job_category}\n" "면접 모드: {mode}\n\n" + "모드별 지침:\n" + "- TECHNICAL: 기술 역량, 프로젝트 경험, 문제 해결을 중심으로 파고듭니다.\n" + "- PERSONALITY: 협업, 갈등 해결, 성장 경험을 중심으로 파고듭니다.\n" + "- INTEGRATED: 기술 질문과 인성 질문의 관점을 균형 있게 반영합니다.\n\n" "직전 질문:\n{previous_question}\n\n" "지원자 답변:\n{answer_text}\n\n" + "검색 문서 컨텍스트:\n---\n{context}\n---\n\n" "{format_instructions}" ) diff --git a/ai/src/ai_server/messaging/consumers/followup_consumer.py b/ai/src/ai_server/messaging/consumers/followup_consumer.py index dfd40dcd..4334cdfa 100644 --- a/ai/src/ai_server/messaging/consumers/followup_consumer.py +++ b/ai/src/ai_server/messaging/consumers/followup_consumer.py @@ -4,6 +4,7 @@ from aio_pika.abc import AbstractIncomingMessage from ai_server.chain.followup_generation_chain import FollowupGenerator +from ai_server.core.client import CoreClient from ai_server.messaging.idempotency import LruIdempotencyStore from ai_server.messaging.publisher import CallbackPublisher from ai_server.model.envelope import Envelope @@ -11,6 +12,7 @@ FollowupCallbackPayload, GenerateFollowupRequest, ) +from ai_server.rag.embedder import EmbeddingProvider log = structlog.get_logger(__name__) @@ -23,11 +25,17 @@ def __init__( publisher: CallbackPublisher, idempotency: LruIdempotencyStore, callback_routing_key: str, + core_client: CoreClient | None = None, + embedder: EmbeddingProvider | None = None, + rag_top_k: int = 5, ) -> None: self._generator = generator self._publisher = publisher self._idempotency = idempotency self._callback_routing_key = callback_routing_key + self._core = core_client + self._embedder = embedder + self._rag_top_k = rag_top_k async def handle(self, message: AbstractIncomingMessage) -> None: async with message.process(requeue=False): @@ -65,6 +73,7 @@ async def handle(self, message: AbstractIncomingMessage) -> None: mode=req.mode, previous_question=req.previous_question, answer_text=req.answer_text, + context=await self._build_rag_context(req), ) payload = FollowupCallbackPayload( @@ -89,3 +98,25 @@ async def handle(self, message: AbstractIncomingMessage) -> None: session_id=req.session_id, trace_id=envelope.trace_id, ) + + 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)" + + query = f"{req.previous_question}\n\n{req.answer_text}" + try: + query_vec = (await self._embedder.embed([query]))[0] + hits = await self._core.search_embeddings( + query_embedding=query_vec, + document_ids=req.context_document_ids, + top_k=self._rag_top_k, + ) + except Exception as exc: + log.warn("followup.rag.failed", error=str(exc), session_id=req.session_id) + return "(none)" + + if not hits: + return "(none)" + 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 72d97053..08081684 100644 --- a/ai/src/ai_server/messaging/consumers/questions_consumer.py +++ b/ai/src/ai_server/messaging/consumers/questions_consumer.py @@ -4,6 +4,7 @@ from aio_pika.abc import AbstractIncomingMessage from ai_server.chain.question_generation_chain import QuestionGenerator +from ai_server.core.client import CoreClient from ai_server.messaging.idempotency import LruIdempotencyStore from ai_server.messaging.publisher import CallbackPublisher from ai_server.model.envelope import Envelope @@ -12,6 +13,7 @@ GenerateQuestionsRequest, QuestionPoolCallbackPayload, ) +from ai_server.rag.embedder import EmbeddingProvider log = structlog.get_logger(__name__) @@ -25,14 +27,20 @@ def __init__( idempotency: LruIdempotencyStore, callback_routing_key: str, initial_pool_size: int = 1, + core_client: CoreClient | None = None, + embedder: EmbeddingProvider | None = None, + rag_top_k: int = 5, ) -> None: self._generator = generator self._publisher = publisher self._idempotency = idempotency self._callback_routing_key = callback_routing_key - # Core 의 QuestionsCallbackService.applyPool 은 questions[0] 만 INSERT 하고 나머지는 폐기. - # 토큰 낭비를 줄이기 위해 envelope.max_questions 대신 풀 크기를 강제. 후속 작업에서 풀 저장 도입 시 늘리기 쉬움. + # Core compatibility keeps callback.kind=POOL, but this is the initial + # question result. maxQuestions remains the full session limit. self._initial_pool_size = max(1, initial_pool_size) + self._core = core_client + self._embedder = embedder + self._rag_top_k = rag_top_k async def handle(self, message: AbstractIncomingMessage) -> None: async with message.process(requeue=False): @@ -57,7 +65,10 @@ async def handle(self, message: AbstractIncomingMessage) -> None: return req = envelope.payload - effective_pool_size = self._initial_pool_size # envelope.max_questions 무시 + effective_pool_size = max( + 1, + req.initial_question_count, + ) log.info( "questions.generate.start", message_id=envelope.message_id, @@ -68,7 +79,7 @@ async def handle(self, message: AbstractIncomingMessage) -> None: trace_id=envelope.trace_id, ) - context_text = _build_context(req.documents) + context_text = await self._build_context(req) pool = await self._generator.generate( job_category=req.job_category, mode=req.mode, @@ -98,6 +109,34 @@ async def handle(self, message: AbstractIncomingMessage) -> None: trace_id=envelope.trace_id, ) + async def _build_context(self, req: GenerateQuestionsRequest) -> str: + base_context = _build_context(req.documents) + if not self._core or not self._embedder: + return base_context + + document_ids = [d.document_id for d in req.documents] + if not document_ids: + return base_context + + query = _build_initial_rag_query(req) + try: + query_vec = (await self._embedder.embed([query]))[0] + hits = await self._core.search_embeddings( + query_embedding=query_vec, + document_ids=document_ids, + top_k=self._rag_top_k, + ) + except Exception as exc: + log.warn("questions.rag.failed", error=str(exc), session_id=req.session_id) + return base_context + + if not hits: + return base_context + rag_context = "\n---\n".join( + f"[doc#{h.document_id} chunk#{h.chunk_index}] {h.chunk_text}" for h in hits + ) + return f"{base_context}\n\n## Retrieved document chunks\n{rag_context}" + def _build_context(documents: list[DocumentContext]) -> str: parts: list[str] = [] @@ -112,3 +151,20 @@ def _build_context(documents: list[DocumentContext]) -> str: block.append(d.markdown) parts.append("\n".join(block)) return "\n\n".join(parts) if parts else "(no documents)" + + +def _build_initial_rag_query(req: GenerateQuestionsRequest) -> str: + parts = [ + f"mode: {req.mode}", + f"job category: {req.job_category}", + ] + for d in req.documents: + doc_parts = [f"document #{d.document_id} {d.source_type}"] + if d.summary: + doc_parts.append(d.summary) + if d.tech_stack: + doc_parts.append("tech stack: " + ", ".join(d.tech_stack)) + if d.markdown: + doc_parts.append(d.markdown[:1000]) + parts.append("\n".join(doc_parts)) + return "\n\n".join(parts) diff --git a/ai/src/ai_server/messaging/runner.py b/ai/src/ai_server/messaging/runner.py index ec372fbd..4edcc3ca 100644 --- a/ai/src/ai_server/messaging/runner.py +++ b/ai/src/ai_server/messaging/runner.py @@ -154,6 +154,8 @@ def __init__(self, settings: Settings) -> None: idempotency=self._idempotency, callback_routing_key=settings.ai_callback_routing_questions, initial_pool_size=settings.questions_initial_pool_size, + core_client=core_client, + embedder=embedder, ) # 꼬리질문 생성 (US-19) @@ -165,6 +167,8 @@ def __init__(self, settings: Settings) -> None: publisher=self._publisher, idempotency=self._idempotency, callback_routing_key=settings.ai_callback_routing_questions, + core_client=core_client, + embedder=embedder, ) # 종합 피드백 생성 (US-24) diff --git a/ai/src/ai_server/model/messages/followup.py b/ai/src/ai_server/model/messages/followup.py index bfc118b0..c2fb7b1b 100644 --- a/ai/src/ai_server/model/messages/followup.py +++ b/ai/src/ai_server/model/messages/followup.py @@ -1,6 +1,6 @@ from typing import Literal -from pydantic import BaseModel +from pydantic import BaseModel, Field from ai_server.model._config import camel_config @@ -18,6 +18,7 @@ class GenerateFollowupRequest(BaseModel): answer_text: str mode: InterviewMode job_category: Literal["FRONTEND", "BACKEND", "INFRA", "DBA"] + context_document_ids: list[int] = Field(default_factory=list) class AnswerEvaluation(BaseModel): diff --git a/ai/src/ai_server/model/messages/questions.py b/ai/src/ai_server/model/messages/questions.py index 12d83f0e..c0efaa39 100644 --- a/ai/src/ai_server/model/messages/questions.py +++ b/ai/src/ai_server/model/messages/questions.py @@ -33,6 +33,7 @@ class GenerateQuestionsRequest(BaseModel): mode: InterviewMode job_category: JobCategory documents: list[DocumentContext] = [] + initial_question_count: int = 1 max_questions: int = 10 diff --git a/ai/tests/test_core_client.py b/ai/tests/test_core_client.py index 9f71227e..d50842bf 100644 --- a/ai/tests/test_core_client.py +++ b/ai/tests/test_core_client.py @@ -7,6 +7,7 @@ CoreEmbeddingUpsertError, CoreTokenError, EmbeddingChunkPayload, + EmbeddingSearchHit, HttpCoreClient, ) @@ -241,3 +242,86 @@ async def test_upsert_embeddings_httpx_error_retriable() -> None: ) assert exc_info.value.code == "CORE_UNAVAILABLE" assert exc_info.value.retriable is True + + +# ----------- search_embeddings ----------- + + +def _make_post_client( + *, + status: int = 200, + json_body: dict | None = None, + text: str = "", + raise_exc: Exception | None = None, +) -> MagicMock: + client = MagicMock() + resp = MagicMock(spec=httpx.Response) + resp.status_code = status + resp.text = text + resp.json = ( + MagicMock(return_value=json_body) + if json_body is not None + else MagicMock(side_effect=ValueError("no json")) + ) + if raise_exc is not None: + client.post = AsyncMock(side_effect=raise_exc) + else: + client.post = AsyncMock(return_value=resp) + return client + + +@pytest.mark.asyncio +async def test_search_embeddings_uses_latest_core_contract() -> None: + client = _make_post_client( + json_body={ + "hits": [ + { + "documentId": 7, + "chunkIndex": 2, + "chunkText": "Spring transaction boundaries", + "distance": 0.17, + } + ] + } + ) + core = HttpCoreClient(base_url="http://core:38010", api_key="k", client=client) + + hits = await core.search_embeddings( + query_embedding=[0.1, 0.2], + document_ids=[7, 8], + top_k=3, + ) + + assert hits == [ + EmbeddingSearchHit( + document_id=7, + chunk_index=2, + chunk_text="Spring transaction boundaries", + distance=0.17, + ) + ] + client.post.assert_awaited_once_with( + "/api/internal/embeddings/search", + json={ + "queryEmbedding": [0.1, 0.2], + "documentIds": [7, 8], + "topK": 3, + }, + ) + + +@pytest.mark.parametrize("status", [400, 401, 403, 404, 500]) +@pytest.mark.asyncio +async def test_search_embeddings_non_2xx_returns_empty(status: int) -> None: + client = _make_post_client(status=status, text="bad") + core = HttpCoreClient(base_url="http://core:38010", api_key="k", client=client) + + assert await core.search_embeddings(query_embedding=[0.1]) == [] + + +@pytest.mark.asyncio +async def test_search_embeddings_http_error_returns_empty() -> None: + client = _make_post_client(raise_exc=httpx.ConnectError("dns fail")) + core = HttpCoreClient(base_url="http://core:38010", api_key="k", client=client) + + assert await core.search_embeddings(query_embedding=[0.1]) == [] diff --git a/ai/tests/test_followup_consumer.py b/ai/tests/test_followup_consumer.py index 661629c8..be80ce01 100644 --- a/ai/tests/test_followup_consumer.py +++ b/ai/tests/test_followup_consumer.py @@ -6,6 +6,7 @@ import pytest from ai_server.chain.followup_generation_chain import FollowupResult +from ai_server.core.client import EmbeddingSearchHit from ai_server.messaging.consumers.followup_consumer import FollowupConsumer from ai_server.messaging.idempotency import LruIdempotencyStore from ai_server.model.messages.followup import ( @@ -47,6 +48,7 @@ def _envelope() -> bytes: "answerText": "RabbitMQ로 보냈습니다.", "mode": "TECHNICAL", "jobCategory": "BACKEND", + "contextDocumentIds": [7], }, "context": {"userId": 42, "sessionId": 99}, } @@ -86,6 +88,85 @@ async def test_consumer_generates_followup_and_publishes_callback(): assert publisher.publish.await_args.kwargs["message_type"] == "callback.questions" +@pytest.mark.asyncio +async def test_consumer_injects_followup_rag_context_when_available(): + generator = MagicMock() + generator.generate = AsyncMock( + return_value=FollowupResult( + followup_question="outbox 저장과 발행의 원자성은 어떻게 보장했나요?", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=3.0, structure="PARTIAL_STAR" + ), + ) + ) + publisher = MagicMock() + publisher.publish = AsyncMock() + core = MagicMock() + core.search_embeddings = AsyncMock( + return_value=[ + EmbeddingSearchHit( + document_id=7, + chunk_index=2, + chunk_text="Outbox rows are inserted in the same transaction", + distance=0.12, + ) + ] + ) + embedder = MagicMock() + embedder.embed = AsyncMock(return_value=[[0.1, 0.2, 0.3]]) + + consumer = FollowupConsumer( + generator=generator, + publisher=publisher, + idempotency=LruIdempotencyStore(max_size=10), + callback_routing_key="callback.questions", + core_client=core, + embedder=embedder, + rag_top_k=3, + ) + await consumer.handle(_StubMessage(_envelope())) + + embedder.embed.assert_awaited_once() + core.search_embeddings.assert_awaited_once_with( + query_embedding=[0.1, 0.2, 0.3], + document_ids=[7], + top_k=3, + ) + context = generator.generate.await_args.kwargs["context"] + assert "Outbox rows are inserted in the same transaction" in context + + +@pytest.mark.asyncio +async def test_consumer_falls_back_when_followup_rag_fails(): + generator = MagicMock() + generator.generate = AsyncMock( + return_value=FollowupResult( + followup_question="실패 재처리는 어떻게 했나요?", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=3.0, structure="PARTIAL_STAR" + ), + ) + ) + publisher = MagicMock() + publisher.publish = AsyncMock() + core = MagicMock() + core.search_embeddings = AsyncMock(side_effect=RuntimeError("core down")) + embedder = MagicMock() + embedder.embed = AsyncMock(return_value=[[0.1, 0.2, 0.3]]) + + consumer = FollowupConsumer( + generator=generator, + publisher=publisher, + idempotency=LruIdempotencyStore(max_size=10), + callback_routing_key="callback.questions", + core_client=core, + embedder=embedder, + ) + await consumer.handle(_StubMessage(_envelope())) + + assert generator.generate.await_args.kwargs["context"] == "(none)" + + @pytest.mark.asyncio async def test_consumer_idempotent_skip(): generator = MagicMock() diff --git a/ai/tests/test_questions_consumer.py b/ai/tests/test_questions_consumer.py index 7f495288..aeb63b5c 100644 --- a/ai/tests/test_questions_consumer.py +++ b/ai/tests/test_questions_consumer.py @@ -6,6 +6,7 @@ import pytest from ai_server.chain.question_generation_chain import GeneratedQuestionPool +from ai_server.core.client import EmbeddingSearchHit from ai_server.messaging.consumers.questions_consumer import ( QuestionsConsumer, _build_context, @@ -85,6 +86,7 @@ async def test_consumer_generates_questions_and_publishes_callback(): } ], "maxQuestions": 5, + "initialQuestionCount": 2, } ) await consumer.handle(_StubMessage(body)) @@ -93,8 +95,8 @@ async def test_consumer_generates_questions_and_publishes_callback(): call = generator.generate.await_args assert call.kwargs["job_category"] == "BACKEND" assert call.kwargs["mode"] == "TECHNICAL" - # envelope.max_questions(=5) 무시하고 initial_pool_size(default 1) 로 강제. Core 가 첫 질문만 사용. - assert call.kwargs["max_questions"] == 1 + # maxQuestions is the session limit; initialQuestionCount controls this result. + assert call.kwargs["max_questions"] == 2 assert "Java" in call.kwargs["context"] publisher.publish.assert_awaited_once() @@ -137,6 +139,166 @@ async def test_consumer_skips_when_message_id_already_seen(): publisher.publish.assert_not_awaited() +@pytest.mark.asyncio +async def test_consumer_defaults_initial_question_count_to_one(): + generator = MagicMock() + generator.generate = AsyncMock(return_value=GeneratedQuestionPool(questions=[])) + publisher = MagicMock() + publisher.publish = AsyncMock() + + consumer = QuestionsConsumer( + generator=generator, + publisher=publisher, + idempotency=LruIdempotencyStore(max_size=10), + callback_routing_key="callback.questions", + initial_pool_size=3, + ) + body = _envelope( + { + "sessionId": 99, + "mode": "TECHNICAL", + "jobCategory": "BACKEND", + "documents": [], + "maxQuestions": 5, + } + ) + await consumer.handle(_StubMessage(body)) + + assert generator.generate.await_args.kwargs["max_questions"] == 1 + + +@pytest.mark.asyncio +async def test_consumer_clamps_initial_question_count_to_at_least_one(): + generator = MagicMock() + generator.generate = AsyncMock(return_value=GeneratedQuestionPool(questions=[])) + publisher = MagicMock() + publisher.publish = AsyncMock() + + consumer = QuestionsConsumer( + generator=generator, + publisher=publisher, + idempotency=LruIdempotencyStore(max_size=10), + callback_routing_key="callback.questions", + initial_pool_size=3, + ) + body = _envelope( + { + "sessionId": 99, + "mode": "TECHNICAL", + "jobCategory": "BACKEND", + "documents": [], + "initialQuestionCount": 0, + "maxQuestions": 5, + } + ) + await consumer.handle(_StubMessage(body)) + + assert generator.generate.await_args.kwargs["max_questions"] == 1 + + +@pytest.mark.asyncio +async def test_consumer_injects_initial_rag_chunks_when_available(): + generator = MagicMock() + generator.generate = AsyncMock(return_value=GeneratedQuestionPool(questions=[])) + publisher = MagicMock() + publisher.publish = AsyncMock() + core = MagicMock() + core.search_embeddings = AsyncMock( + return_value=[ + EmbeddingSearchHit( + document_id=1, + chunk_index=4, + chunk_text="Outbox table uses status and retry count", + distance=0.11, + ) + ] + ) + embedder = MagicMock() + embedder.embed = AsyncMock(return_value=[[0.1, 0.2]]) + + consumer = QuestionsConsumer( + generator=generator, + publisher=publisher, + idempotency=LruIdempotencyStore(max_size=10), + callback_routing_key="callback.questions", + core_client=core, + embedder=embedder, + rag_top_k=2, + ) + body = _envelope( + { + "sessionId": 99, + "mode": "TECHNICAL", + "jobCategory": "BACKEND", + "documents": [ + { + "documentId": 1, + "sourceType": "REPOSITORY", + "summary": "outbox 구현", + "techStack": ["Spring"], + "markdown": "transactional publisher", + } + ], + "initialQuestionCount": 1, + "maxQuestions": 5, + } + ) + await consumer.handle(_StubMessage(body)) + + embedder.embed.assert_awaited_once() + core.search_embeddings.assert_awaited_once_with( + query_embedding=[0.1, 0.2], + document_ids=[1], + top_k=2, + ) + context = generator.generate.await_args.kwargs["context"] + assert "Outbox table uses status and retry count" in context + assert "outbox 구현" in context + + +@pytest.mark.asyncio +async def test_consumer_falls_back_to_document_context_when_rag_fails(): + generator = MagicMock() + generator.generate = AsyncMock(return_value=GeneratedQuestionPool(questions=[])) + publisher = MagicMock() + publisher.publish = AsyncMock() + core = MagicMock() + core.search_embeddings = AsyncMock(side_effect=RuntimeError("core down")) + embedder = MagicMock() + embedder.embed = AsyncMock(return_value=[[0.1, 0.2]]) + + consumer = QuestionsConsumer( + generator=generator, + publisher=publisher, + idempotency=LruIdempotencyStore(max_size=10), + callback_routing_key="callback.questions", + core_client=core, + embedder=embedder, + ) + body = _envelope( + { + "sessionId": 99, + "mode": "TECHNICAL", + "jobCategory": "BACKEND", + "documents": [ + { + "documentId": 1, + "sourceType": "RESUME", + "summary": "Java/Spring backend", + "techStack": ["Java"], + "markdown": "payment service", + } + ], + "maxQuestions": 5, + } + ) + await consumer.handle(_StubMessage(body)) + + context = generator.generate.await_args.kwargs["context"] + assert "Java/Spring backend" in context + assert "Retrieved document chunks" not in context + + def test_build_context_handles_empty_documents(): assert _build_context([]) == "(no documents)"