[TeamBlog] 면접 모드 3가지 구현 흐름 - 100-hours-a-week/9-team-Devths-WIKI GitHub Wiki
면접 모드 3가지 구현 흐름
1. 개요
AI 취업 도우미 챗봇은 인성 면접, 기술 면접, 일반 질의 응답 3가지 모드를 제공합니다. 각 모드는 초기 단순 구현에서 시작하여 문제점을 발견하고 개선하는 과정을 거쳤습니다.
핵심 성과
| 모드 | 초기 문제 | 해결 방안 | 개선 효과 |
|---|---|---|---|
| 인성 면접 | 일반적인 질문만 생성 | 면접 데이터셋 + VectorDB | 기업별 맞춤 질문 |
| 기술 면접 | 같은 질문 반복 | 임베딩 유사도 중복 방지 | 질문 다양성 확보 |
| 일반 Q&A | 관련 없는 답변 | RAG 파이프라인 고도화 | 정확도 대폭 향상 |
2. 공통 기반 — 초기 구현 및 문제 해결
2.1 첫 배포: Gemini API + 프롬프트
┌─────────────────────────────────────────────────────────────────┐
│ 초기 아키텍처 (v1) │
├─────────────────────────────────────────────────────────────────┤
│ │
│ [사용자] → [FastAPI] → [Gemini API] → [응답] │
│ │
│ - 3가지 면접 모드를 프롬프트로만 구분 │
│ - VectorDB 없음, 사용자 데이터 미활용 │
│ - 단순하지만 한계 명확 │
│ │
└─────────────────────────────────────────────────────────────────┘
2.2 프롬프트 인젝션 문제 발생
문제: 서비스 배포 후 악의적 사용자가 프롬프트를 조작하는 시도 발견
해결: 2단계 정규식 필터링 도입
# 1단계: 위험 패턴 탐지
DANGEROUS_PATTERNS = [
r"ignore\s+(previous|above|all)\s+instructions?",
r"system\s*prompt",
r"you\s+are\s+now",
r"act\s+as\s+if",
r"pretend\s+(to\s+be|you\s+are)",
]
# 2단계: 입력 정제
def sanitize_user_input(text: str) -> str:
# 제어 문자 제거, 길이 제한, 특수 토큰 이스케이프
...
2.3 SSE 스트리밍 도입
결정: 챗봇 특성상 WebSocket 대신 SSE(Server-Sent Events) 선택
| 항목 | SSE | WebSocket |
|---|---|---|
| 통신 방향 | 단방향 (서버 → 클라이언트) | 양방향 |
| 구현 복잡도 | 낮음 | 높음 |
| 인프라 호환성 | HTTP 그대로 사용 | 별도 설정 필요 |
| 업계 표준 | OpenAI, Anthropic, Google 등 | - |
선택 이유:
- LLM 응답은 서버 → 클라이언트 단방향
- 사용자 메시지는 일반 HTTP POST로 충분
- 기존 인프라 변경 없이 적용 가능
2.4 SSE 504 타임아웃 해결
문제: 면접 질문 생성 시 60~120초 소요 → Nginx proxy_read_timeout 초과
해결: SSE Keepalive 이벤트 25초 주기 전송
async def stream_interview_response():
task = asyncio.create_task(generate_questions())
while not task.done():
try:
await asyncio.wait_for(asyncio.shield(task), timeout=25)
except asyncio.TimeoutError:
yield ": keepalive\n\n" # Nginx idle timeout 방지
result = task.result()
yield f"data: {json.dumps(result)}\n\n"
3. 인성 면접 모드
3.1 문제 인식
- 기업별로 정해진 인성 면접 주제들이 많음
- LLM만으로는 기업 특화 질문 생성 어려움
- 일반적이고 뻔한 질문만 반복
3.2 해결: 면접 데이터셋 + VectorDB 도입
┌─────────────────────────────────────────────────────────────────┐
│ 인성 면접 아키텍처 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ [HuggingFace InterView_Datasets] │
│ ↓ │
│ [임베딩 (ko-sroberta)] → [ChromaDB interview_feedback] │
│ ↓ │
│ [사용자 질문] → [유사 Q&A 검색] → [Few-shot 프롬프트 주입] │
│ ↓ │
│ [Gemini API] │
│ ↓ │
│ [맞춤 질문 생성] │
│ │
└─────────────────────────────────────────────────────────────────┘
3.3 데이터셋 자동 임베딩 파이프라인
데이터 소스: HuggingFace InterView_Datasets
자동화: /data 디렉토리에 JSON 추가 시 스크립트 실행으로 VectorDB 임베딩
# 면접 데이터셋 임베딩
docker exec -it ai-service poetry run python scripts/auto_embed_data.py
# scripts/embed_interview_dataset.py
async def embed_interview_data():
with open("/app/data/interview_dataset_valid.json") as f:
data = json.load(f)
for item in data:
await vectordb.add_document(
document_id=f"interview_{item['id']}",
text=f"질문: {item['question']}\n답변: {item['answer']}",
collection_type="interview_feedback",
metadata={
"interview_type": "personality",
"category": item.get("category", "general"),
}
)
3.4 Few-shot 프롬프팅 (SemanticSimilarityExampleSelector)
# app/services/example_selector.py
async def get_few_shot_for_personality(
vectordb: Any,
query_text: str,
k: int = 2,
interview_type_filter: str = "personality",
max_distance: float = 1.5,
) -> str:
"""인성 면접용: interview_feedback에서 유사 Q&A k개 조회"""
results = await vectordb.query(
query_text=query_text[:500],
collection_type="interview_feedback",
n_results=k,
where={"interview_type": interview_type_filter},
max_distance=max_distance,
)
lines = []
for r in results:
q = r.get("metadata", {}).get("question_only", "")
a = r.get("metadata", {}).get("answer_only", "")
lines.append(f"질문: {q}\n답변: {a}")
return "\n\n".join(lines)
3.5 시간 가중 리트리버
목적: 최신 면접 피드백 우선 반영
# 시간 감쇠 공식
score = semantic_similarity + (1.0 - decay_rate) ^ hours_passed
# 인성 면접: 시간 거의 무관 (반감기 ~69시간)
rag_decay_rate_personality = 0.01
# 기술 면접: 최신 정보 우선 (반감기 ~6.6시간)
rag_decay_rate_technical = 0.1
3.6 면접 스타일 다양화
| 스타일 | 설명 | 특징 |
|---|---|---|
standard |
표준 면접관 | 중립적, 객관적 |
pressure |
압박 면접관 | 날카로운 질문, 꼬리물기 |
friendly |
친근한 면접관 | 편안한 분위기, 격려 |
# app/prompts/persona_styles.py
def get_personality_style_prompt(style: str) -> str:
styles = {
"standard": "당신은 공정하고 객관적인 인성 면접관입니다...",
"pressure": "당신은 지원자의 진정성을 검증하는 압박 면접관입니다...",
"friendly": "당신은 지원자가 편하게 이야기할 수 있도록 돕는 면접관입니다...",
}
return styles.get(style, styles["standard"])
4. 기술 면접 모드
4.1 문제 인식
- 초기: API 호출 + 프롬프트로만 구현
- 문제: 같은 질문이 반복적으로 나옴
- 사용자 경험 저하, 면접 연습 효과 감소
4.2 해결: 질문 중복 방지 시스템
┌─────────────────────────────────────────────────────────────────┐
│ 기술 면접 질문 중복 방지 흐름 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ [새 질문 생성 요청] │
│ ↓ │
│ [LLM이 질문 후보 생성] │
│ ↓ │
│ [임베딩 생성 (ko-sroberta)] │
│ ↓ │
│ [mastered_questions와 cosine similarity 비교] │
│ ↓ │
│ ┌─────────────────────────────────────────┐ │
│ │ similarity >= 0.85 → 유사 질문 필터링 │ │
│ │ similarity < 0.85 → 새 질문으로 채택 │ │
│ └─────────────────────────────────────────┘ │
│ ↓ │
│ [채택된 질문 출제] │
│ ↓ │
│ [답변 품질 평가 → "mastered" 시 mastered_questions에 추가] │
│ │
└─────────────────────────────────────────────────────────────────┘
4.3 중복 방지 구현
# app/services/interview_dedup.py
def cosine_similarity(vec_a: list[float], vec_b: list[float]) -> float:
"""두 벡터의 코사인 유사도 계산"""
a = np.array(vec_a, dtype=np.float32)
b = np.array(vec_b, dtype=np.float32)
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
async def filter_similar_questions(
new_questions: list[dict],
mastered_questions: list[dict],
vectordb_service: Any,
threshold: float = 0.85,
) -> list[dict]:
"""완벽히 답변한 질문과 유사한 새 질문 필터링"""
if not mastered_questions:
return new_questions
mastered_embeddings = [m["embedding"] for m in mastered_questions]
filtered = []
for q in new_questions:
q_embedding = await vectordb_service.create_embedding(q["question"])
max_sim = max(cosine_similarity(q_embedding, m_emb) for m_emb in mastered_embeddings)
if max_sim < threshold:
filtered.append(q) # 유사도 낮음 → 새 질문으로 채택
return filtered
4.4 mastered_questions 관리
# app/schemas/chat.py
class InterviewSession(BaseModel):
session_id: str
interview_type: str
questions: list[InterviewQuestionState]
current_index: int = 0
# 완벽히 답변된 질문 (유사 질문 중복 방지)
mastered_questions: list[dict] = Field(
default=[],
description="질문 텍스트 + 임베딩 저장"
)
4.5 하이브리드 질문 시스템
| 구성 요소 | 역할 | 비율 |
|---|---|---|
| 템플릿 질문 3개 | 검증된 기본 질문 | 60% |
| LLM 생성 질문 2개 | 맥락 기반 동적 질문 | 40% |
# 하이브리드 질문 생성
def generate_hybrid_questions(resume_text, job_posting_text, count=5):
# 1. 템플릿에서 3개 선택
template_questions = select_from_templates(
job_posting_text,
count=3,
asked_questions=session.previous_questions
)
# 2. LLM으로 2개 생성
llm_questions = await llm.generate_interview_questions(
resume_text=resume_text,
posting_text=job_posting_text,
interview_type="technical",
count=2,
)
return template_questions + llm_questions
4.6 꼬리 질문 로직
# 답변 품질 평가 후 꼬리 질문 생성
async def evaluate_and_followup(answer: str, question: str):
# 1. 답변 품질 평가
evaluation = await llm.evaluate_answer(question, answer)
answer_quality = evaluation.get("answer_quality", "good")
# 2. 품질에 따른 분기
if answer_quality == "excellent":
# 완벽한 답변 → mastered_questions에 추가
session.mastered_questions.append({
"question": question,
"embedding": await vectordb.create_embedding(question)
})
return generate_next_question()
elif answer_quality in ("good", "needs_improvement"):
# 꼬리 질문 생성
followup = await llm.generate_followup_question(
original_question=question,
user_answer=answer,
evaluation=evaluation,
)
return followup
else: # poor
# 힌트 제공 후 재시도 유도
return generate_hint(question, answer)
4.7 면접 결과 VectorDB 자동 적재
# app/services/interview_ingestion_service.py
class InterviewIngestionService:
@staticmethod
def _make_id(user_id: str, session_id: str, question_index: int) -> str:
"""중복 방지를 위한 문서 ID 생성 (동일 세션 동일 질문 → upsert)"""
return f"interview_{user_id}_{session_id}_{question_index}"
async def ingest_interview_qa(
self,
user_id: str,
session_id: str,
question: str,
answer: str,
evaluation: dict,
):
doc_id = self._make_id(user_id, session_id, question_index)
await self.vectordb.add_document(
document_id=doc_id,
text=f"질문: {question}\n답변: {answer}",
collection_type="interview_feedback",
metadata={
"user_id": user_id,
"session_id": session_id,
"interview_type": "technical",
"answer_quality": evaluation.get("answer_quality"),
"created_at": datetime.now().isoformat(),
}
)
5. 일반 질의 응답 Q&A
5.1 문제 인식
- 초기: API 호출만 구현
- 문제: 전혀 관련 없는 이야기가 나옴
- 사용자 데이터(이력서, 채용공고)를 활용하지 못함
5.2 해결: RAG 파이프라인 고도화
┌─────────────────────────────────────────────────────────────────┐
│ RAG 파이프라인 진화 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ Phase 1: VectorDB + 임베딩 도입 │
│ ───────────────────────────────── │
│ [사용자 질문] → [임베딩] → [ChromaDB 검색] → [컨텍스트 + LLM] │
│ │
│ Phase 2: WebBaseLoader 도입 │
│ ───────────────────────────── │
│ [URL 입력] → [WebBaseLoader] → [텍스트 추출] → [VectorDB 저장] │
│ │
│ Phase 3: Tavily 자동 검색 도입 │
│ ───────────────────────────── │
│ [Celery Beat] → [Tavily Search] → [채용 트렌드 크롤링] │
│ ↓ │
│ [VectorDB 자동 임베딩] │
│ │
│ Phase 4: 리트리버 6종+ 확장 │
│ ───────────────────────────── │
│ BM25 + MMR 앙상블, 부모 문서, 다중 쿼리, 셀프 쿼리, 시간 가중 │
│ │
└─────────────────────────────────────────────────────────────────┘
5.3 Phase 1: VectorDB + 임베딩 도입
# 사용자 데이터 기반 답변
async def retrieve_context(user_id: str, query: str):
results = await vectordb.query(
query_text=query,
collection_type="resume", # 이력서
n_results=5,
where={"user_id": user_id},
)
# 채용공고도 검색
posting_results = await vectordb.query(
query_text=query,
collection_type="job_posting",
n_results=3,
where={"user_id": user_id},
)
return results + posting_results
5.4 Phase 2: WebBaseLoader 도입
한계: 사용자가 직접 URL을 입력한 곳에서만 데이터 가져옴
# app/services/web_loader_service.py
class WebLoaderService:
@staticmethod
async def extract_text_from_url(url: str) -> str:
"""URL에서 텍스트 추출 (LangChain WebBaseLoader)"""
loader = WebBaseLoader(url)
docs = loader.load()
return "\n".join(doc.page_content for doc in docs)
5.5 Phase 3: Tavily 자동 검색 도입
목적: 최신 채용 트렌드를 주기적으로 업데이트
# app/tasks/celery_app.py — Celery Beat 스케줄
celery_app.conf.beat_schedule = {
"search-trend-weekly": {
"task": "app.tasks.trend_tasks.search_trend_queries_task",
"schedule": crontab(
minute="30",
hour=9,
day_of_week=1, # 매주 월요일 오전 9시 30분
),
},
}
# app/services/tavily_search_service.py
class TavilySearchService:
async def search_and_embed(self, queries: list[str]) -> TavilySearchStats:
"""Tavily 검색 후 VectorDB 임베딩"""
all_results = []
for query in queries:
results = await self.client.search(
query=query,
search_depth="advanced",
max_results=10,
)
all_results.extend(results)
# 중복 제거 후 VectorDB 저장
deduplicated = self._deduplicate(all_results)
await self._embed_to_vectordb(deduplicated)
return TavilySearchStats(
total_queries=len(queries),
total_results=len(all_results),
deduplicated_results=len(deduplicated),
)
5.6 Phase 4: RAG 리트리버 6종+ 확장
| 리트리버 | 역할 | |----------|-----|------| | BM25 + MMR 앙상블 | 희소 + 밀집 검색 결합 (RRF) | | 부모 문서 리트리버 | 청크 검색 → 부모 문서 반환 | | 다중 쿼리 리트리버 | 쿼리 다각화 (3~5개 관점) | | 셀프 쿼리 리트리버 | 메타데이터 기반 구조화 검색 | | 시간 가중 리트리버 | 최신성 반영 (감쇠율 적용) | | FlashRank 리랭커 | BERT Cross-Encoder 재정렬 |
# app/services/rag_service.py — 앙상블 리트리버
async def retrieve_context(self, query: str, user_id: str):
# 1. BM25 (희소 검색)
bm25_results = await self._bm25_search(query)
# 2. MMR (밀집 검색 + 다양성)
mmr_results = await self._mmr_search(query, user_id)
# 3. RRF (Reciprocal Rank Fusion)
combined = self._rrf_merge(bm25_results, mmr_results)
# 4. FlashRank 리랭킹
reranked = await self._flashrank_rerank(query, combined)
return reranked[:10]
5.7 대화 맥락 관리
# Redis 기반 채팅 히스토리 반영
async def generate_response_with_history(
user_message: str,
user_id: str,
chat_id: str,
):
# 1. Redis에서 이전 대화 조회
history = await redis_session.get_history(user_id, chat_id)
# 2. 컨텍스트 검색
context = await rag.retrieve_context(user_message, user_id)
# 3. LLM 호출 (히스토리 + 컨텍스트)
response = await llm.generate_response(
user_message=user_message,
context=context,
history=history, # 이전 대화 포함
)
# 4. 새 대화 Redis에 저장
await redis_session.append_history(user_id, chat_id, user_message, response)
return response
5.8 Few-shot 프롬프팅 (일반 Q&A)
# app/services/example_selector.py
GENERAL_CHAT_EXAMPLES = [
{
"input": "자기소개서 첨삭 부탁해도 될까요?",
"output": "네, 물론이에요. 먼저 작성하신 자기소개서를 붙여 주시면..."
},
{
"input": "면접에서 떨리는데 어떻게 하면 좋을까요?",
"output": "면접 전에는 회사와 직무를 미리 파악하고..."
},
# ...
]
async def get_few_shot_for_general(
vectordb: Any,
user_message: str,
k: int = 2,
min_similarity: float = 0.3,
) -> str:
"""사용자 질문과 유사한 예제 k개 선택 후 프롬프트 주입"""
query_emb = await vectordb.create_embedding(user_message)
sims = [cosine_similarity(query_emb, emb) for emb in cached_example_embeddings]
top_indices = [idx for idx, sim in sorted(enumerate(sims), key=lambda x: -x[1])[:k]
if sim >= min_similarity]
lines = []
for idx in top_indices:
ex = GENERAL_CHAT_EXAMPLES[idx]
lines.append(f"사용자: {ex['input']}\n어시스턴트: {ex['output']}")
return "\n\n".join(lines)
6. 향후 확장: SageMaker 도입
6.1 문제 인식
- 크롤링 + VectorDB 업데이트만으로는 데이터 증가 시 관리 어려움
- 도메인 특화 모델 필요성 증가
6.2 해결: SageMaker 기반 ML 파이프라인
┌─────────────────────────────────────────────────────────────────┐
│ SageMaker ML 파이프라인 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ [Tavily 크롤링 데이터] │
│ ↓ │
│ [S3 데이터 레이크] │
│ ↓ │
│ [SageMaker Training Job] ← GPU 인스턴스 (L4/A100) │
│ ↓ │
│ [모델 아티팩트 저장] │
│ ↓ │
│ [SageMaker Endpoint 배포] │
│ ↓ │
│ [실시간 추론 서비스] │
│ │
└─────────────────────────────────────────────────────────────────┘
7. 기술 스택 요약
| 영역 | 기술 | 역할 |
|---|---|---|
| LLM | Gemini API (3.1-flash, 3.1-pro) | 질문 생성, 답변 평가 |
| VectorDB | ChromaDB (Server Mode) | 임베딩 저장, 유사도 검색 |
| 임베딩 | ko-sroberta-multitask | 한국어 특화 임베딩 |
| 리랭커 | FlashRank (BERT Cross-Encoder) | 검색 결과 재정렬 |
| 크롤링 | Tavily Search API | 채용 트렌드 자동 수집 |
| 스케줄링 | Celery Beat + Redis | 주기적 크롤링 실행 |
| 세션 관리 | Redis (ElastiCache) | 대화 히스토리 저장 |
| ML 서빙 | SageMaker | 모델 학습 및 배포 |
8. 결론
핵심 성과
- 인성 면접: 면접 데이터셋 + Few-shot 프롬프팅으로 기업별 맞춤 질문 생성
- 기술 면접: 임베딩 유사도 기반 중복 방지로 질문 다양성 확보
- 일반 Q&A: RAG 6종+ 리트리버 + FlashRank로 답변 정확도 대폭 향상
배운 점
- LLM만으로는 한계가 있음 → VectorDB + RAG 필수
- 질문 중복 방지는 임베딩 유사도로 해결 가능
- 주기적 데이터 업데이트는 Celery Beat + Tavily 조합이 효과적