[AI] 04. ADR 021‐025 ‐ 최적화 - 100-hours-a-week/9-team-Devths-WIKI GitHub Wiki
[AI] ADR 021-025
Architecture Decision Records (ADR) 021-025
AI 서버 및 시스템 아키텍처 관련 주요 기술 결정 사항
📑 목차
- ADR-021: Embedding 모델 통일 + BM25 Hybrid Search
- ADR-022: n8n 도입 (AI 파이프라인 테스트)
- ADR-023: 면접 질문 Pool 캐싱 전략
- ADR-024: VectorDB 메타데이터 필터링 vs RDB + VectorDB
- ADR-025: RDB ↔ VectorDB 데이터 동기화 전략
ADR-021: Embedding 모델 통일 + BM25 Hybrid Search
📋 메타데이터
| 항목 | 내용 |
|---|---|
| 상태 | ✅ 승인됨 (Accepted) |
| 작성일 | 2026-01-12 |
| 결정자 | AI팀 |
| 관련 기능 | RAG, VectorDB, Embedding |
🎯 컨텍스트 (Context)
중요한 발견: 임베딩 모델은 하나만 써야 합니다!
문제점:
- ❌ SBERT로 저장한 벡터 ≠ Gemini로 검색 불가
- ❌ 서로 다른 임베딩 모델은 벡터 공간이 다름
- ❌ 같은 차원(768)이어도 호환 불가
현재 구조 (잘못됨):
Embedding 모델:
- Primary: SBERT (snunlp/KR-SBERT-V40K)
- Fallback 1: Gemini (gemini-embedding-001)
- Fallback 2: OpenAI (text-embedding-3-small)
→ ❌ 문제: 서로 다른 벡터 공간!
추가 요구사항:
- BM25 + Embedding Hybrid Search 도입
- 키워드 기반 + 의미 기반 검색 결합
✅ 결정 (Decision)
Gemini Embedding 단일 모델 + BM25 Hybrid Search
| 버전 | Embedding 모델 | Retriever |
|---|---|---|
| V1~V2 | Gemini만 | Vector Search |
| V3 | Gemini만 | BM25 + Vector Hybrid |
| V4 | Gemini만 | BM25 + Vector + Reranker |
📝 근거 (Rationale)
1. 임베딩 모델은 하나만 사용해야 함!
잘못된 예:
1. 이력서 저장: SBERT (768차원) → VectorDB
2. 질문 검색: Gemini (768차원) → VectorDB
→ ❌ 검색 안 됨! (벡터 공간이 다름)
올바른 예:
1. 이력서 저장: Gemini (768차원) → VectorDB
2. 질문 검색: Gemini (768차원) → VectorDB
→ ✅ 정상 검색!
2. BM25 + Embedding Hybrid Search
BM25 (키워드 30%) + Gemini Embedding (의미 70%)
↓
Hybrid Search (20개)
↓
Reranker (3개)
↓
LLM
📅 이력
| 날짜 | 변경 내용 |
|---|---|
| 2026-01-12 | 초기 결정 (Gemini 단일 모델 + BM25 Hybrid) |
ADR-022: n8n 도입 (AI 파이프라인 테스트)
📋 메타데이터
| 항목 | 내용 |
|---|---|
| 상태 | ✅ 승인됨 (Accepted) |
| 작성일 | 2026-01-12 |
| 결정자 | AI팀 |
| 관련 기능 | AI 파이프라인 테스트, 워크플로우 검증 |
🎯 컨텍스트 (Context)
AI 파이프라인을 빠르게 테스트하고 검증해야 합니다.
요구사항:
- 이력서 분석 플로우 테스트
- RAG 파이프라인 검증
- 프롬프트 실험 (빠른 반복)
- 팀 데모 (시각적 설명)
문제점:
- 코드로 테스트하면 시간 소요
- 프롬프트 변경마다 재배포
- 비개발자 팀원 이해 어려움
🔍 선택지 분석 (Options)
Option 1: 코드 기반 테스트 (FastAPI + LangChain)
# 테스트마다 코드 수정 필요
from langchain_google_genai import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(model="gemini-3-flash")
result = llm.invoke("이력서 분석해줘")
print(result)
| 장점 | 단점 |
|---|---|
| 프로덕션과 동일 | 테스트 느림 |
| 버전 관리 쉬움 | 프롬프트 변경 어려움 |
| 비개발자 이해 어려움 |
Option 2: n8n (시각적 워크플로우) ⭐
┌─────────────────────────────────────────────────────────────────┐
│ n8n 워크플로우 (드래그 앤 드롭) │
├─────────────────────────────────────────────────────────────────┤
│ │
│ [Manual Trigger] │
│ ↓ │
│ [Gemini VLM] ← 이미지 업로드 │
│ ↓ │
│ [Gemini Embedding] ← 텍스트 → 벡터 │
│ ↓ │
│ [ChromaDB] ← 벡터 저장 │
│ ↓ │
│ [Gemini LLM] ← RAG 질문 │
│ ↓ │
│ [Output] ← 결과 확인 │
│ │
└─────────────────────────────────────────────────────────────────┘
| 장점 | 단점 |
|---|---|
| 시각적 워크플로우 | 프로덕션 부적합 |
| 빠른 테스트 | 복잡한 로직 어려움 |
| AI 네이티브 지원 | Git 관리 어려움 |
| 팀 협업 |
Option 3: Airflow (배치 처리)
| 장점 | 단점 |
|---|---|
| 프로덕션 가능 | 테스트에 과함 |
| DAG 시각화 | 설정 복잡 |
| 빠른 반복 어려움 |
✅ 결정 (Decision)
n8n 도입 (테스트 전용)
| 환경 | 도구 | 용도 |
|---|---|---|
| 개발/테스트 | n8n | AI 파이프라인 테스트 |
| 프로덕션 | FastAPI + LangChain | 실제 서비스 |
| 배치 | Celery Beat / Airflow | 정기 작업 |
📝 근거 (Rationale)
1. n8n의 장점
빠른 AI 파이프라인 테스트:
기존 (코드):
1. 코드 작성
2. 테스트 실행
3. 수정
4. 재실행
→ 1회 반복: 5~10분
n8n:
1. 노드 드래그
2. 연결
3. 실행
→ 1회 반복: 30초!
AI 도구 네이티브 지원:
- ✅ LangChain
- ✅ OpenAI
- ✅ Google Gemini
- ✅ Pinecone, ChromaDB
- ✅ Webhook, HTTP Request
2. 프로젝트 적용 예시
이력서 분석 플로우 테스트:
[Manual Trigger] (이력서 PDF)
↓
[Gemini VLM] (텍스트 추출)
↓
[Gemini Embedding] (벡터 변환)
↓
[ChromaDB] (저장)
↓
[Gemini LLM] (분석 - RAG)
↓
[Output] (결과 확인)
면접 모드 테스트:
[Webhook] (사용자 질문)
↓
[ChromaDB] (이력서 검색 - RAG)
↓
[Gemini LLM] (질문 생성)
↓
[Output] (면접 질문 확인)
BM25 Hybrid Search 테스트:
[Manual Trigger] (질문)
↓
[HTTP Request] (BM25 검색)
↓
[Gemini Embedding] (벡터 검색)
↓
[Merge] (Hybrid 결합)
↓
[Cohere Rerank] (재정렬)
↓
[Output] (결과 비교)
3. 프로덕션과 분리
n8n은 테스트 전용:
- ✅ 개발 단계에서 빠른 검증
- ✅ 프롬프트 실험
- ✅ 팀 데모
프로덕션은 코드 기반:
- ✅ FastAPI + LangChain
- ✅ 버전 관리 (Git)
- ✅ 성능, 확장성
🔧 구현 예시
Docker로 n8n 설치
# n8n 실행
docker run -it --rm \
--name n8n \
-p 5678:5678 \
-v ~/.n8n:/home/node/.n8n \
n8nio/n8n
# 접속
# http://localhost:5678
n8n 워크플로우 예시
1. 이력서 분석 테스트:
노드 구성:
1. Manual Trigger
2. HTTP Request (이력서 업로드)
3. Gemini VLM (텍스트 추출)
4. Gemini Embedding (벡터 변환)
5. HTTP Request (ChromaDB 저장)
6. Gemini LLM (분석)
7. Output
실행: 버튼 클릭 → 즉시 결과 확인!
2. 프롬프트 A/B 테스트:
노드 구성:
1. Manual Trigger
2. Split In Batches (2개 프롬프트)
3. Gemini LLM (프롬프트 A)
4. Gemini LLM (프롬프트 B)
5. Merge (결과 비교)
6. Output
실행: 두 프롬프트 결과 동시 비교!
📊 비교표
| 항목 | n8n | FastAPI + LangChain | Airflow |
|---|---|---|---|
| 테스트 속도 | ⭐⭐⭐⭐⭐ | ⭐⭐ | ⭐ |
| 시각화 | ✅ | ❌ | ⚠️ |
| AI 통합 | ✅ | ✅ | ⚠️ |
| 프로덕션 | ❌ | ✅ | ✅ |
| 학습 곡선 | 🔴 낮음 | 🔴🔴 | 🔴🔴🔴 |
| 팀 협업 | ✅ | ⚠️ | ⚠️ |
| 용도 | 테스트 | 프로덕션 | 배치 |
🎯 최종 추천
개발 단계 (V1~V3)
n8n: AI 파이프라인 테스트 ✅
- 이력서 분석 플로우 검증
- RAG 파이프라인 테스트
- 프롬프트 A/B 테스트
- BM25 Hybrid Search 검증
- Reranker 성능 비교
프로덕션 (V1~V4)
FastAPI + LangChain: 실제 서비스 ✅
- 코드 기반 (Git 관리)
- 성능, 확장성
- 안정적 운영
배치 처리 (V3~V4)
Celery Beat / Airflow: 정기 작업 ✅
- 매일 새벽 패턴 분석
- 질문 풀 생성
💡 핵심 정리
n8n 도입 이유:
- ✅ 빠른 AI 파이프라인 테스트 (30초 vs 5분)
- ✅ 시각적 워크플로우 (팀 협업)
- ✅ AI 네이티브 지원 (LangChain, Gemini, OpenAI)
- ✅ 프롬프트 실험 (빠른 반복)
용도:
- ✅ 개발/테스트: n8n
- ✅ 프로덕션: FastAPI + LangChain
- ✅ 배치: Celery Beat / Airflow
결론:
- ✅ n8n으로 빠르게 테스트
- ✅ 검증 완료 후 코드로 프로덕션 구현
📅 이력
| 날짜 | 변경 내용 |
|---|---|
| 2026-01-12 | 초기 결정 (n8n 테스트 도구 도입) |
ADR-023: 면접 질문 Pool 캐싱 전략
📋 메타데이터
| 항목 | 내용 |
|---|---|
| 상태 | ✅ 승인됨 (Accepted) |
| 작성일 | 2026-01-12 |
| 결정자 | AI팀 |
| 관련 기능 | 면접 모드, 질문 생성, 캐싱 |
🎯 컨텍스트 (Context)
면접 질문 생성 시 LLM 호출 비용을 절감해야 합니다.
문제점:
- 매번 LLM으로 질문 생성 → 비용 증가
- 유사한 질문 반복 생성 → 비효율
- 캐시 미스 시 응답 지연
요구사항:
- 자주 사용되는 질문은 미리 생성 (Pool)
- Pool에 없으면 LLM으로 생성 (Fallback)
- 비용 절감 + 응답 속도 향상
🔍 선택지 분석 (Options)
Option 1: 매번 LLM 생성 (기본)
# 매번 LLM 호출
def generate_question(resume, job_posting):
prompt = f"Resume: {resume}\nJob: {job_posting}\n질문 생성해줘"
question = llm.invoke(prompt)
return question
| 장점 | 단점 |
|---|---|
| 항상 최신 질문 | 비용 높음 |
| 구현 간단 | 응답 느림 (1~3초) |
| 동일 질문 반복 생성 |
비용:
V3 예상:
- 면접 질문: 5개/세션
- 세션: 50개/일
= 250 questions/일 × 30일 = 7,500 questions/월
× $0.001/question = $7.5/월
Option 2: Question Pool + RAG Fallback ⭐
┌─────────────────────────────────────────────────────────────────┐
│ Question Pool + RAG Fallback │
├─────────────────────────────────────────────────────────────────┤
│ │
│ [사용자 요청] (면접 질문 생성) │
│ ↓ │
│ [1. Question Pool 검색] (VectorDB) │
│ - 미리 생성된 질문 Pool (1,000개) │
│ - 임베딩 유사도 검색 │
│ ↓ │
│ [2. 유사도 체크] │
│ - 유사도 > 0.85 → Pool에서 반환 (캐시 히트) ✅ │
│ - 유사도 < 0.85 → LLM 생성 (캐시 미스) ⚠️ │
│ ↓ │
│ [3-A. 캐시 히트] (80% 케이스) │
│ - Pool에서 질문 반환 (즉시) │
│ - 비용: $0 │
│ - 응답: < 100ms │
│ ↓ │
│ [3-B. 캐시 미스] (20% 케이스) │
│ - LLM으로 질문 생성 │
│ - 비용: $0.001 │
│ - 응답: 1~3초 │
│ - 생성된 질문 → Pool에 추가 (학습) │
│ │
└─────────────────────────────────────────────────────────────────┘
| 장점 | 단점 |
|---|---|
| 비용 80% 절감 | 초기 Pool 구축 필요 |
| 응답 빠름 (< 100ms) | Pool 관리 필요 |
| 자동 학습 (Pool 확장) | 유사도 임계값 조정 |
비용:
V3 예상:
- 캐시 히트: 80% (6,000 questions) → $0
- 캐시 미스: 20% (1,500 questions) → $1.5/월
총 비용: $1.5/월 (기존 $7.5/월 대비 80% 절감!)
Option 3: Redis 캐싱
# Redis 캐싱
import redis
redis_client = redis.Redis()
def generate_question(resume, job_posting):
cache_key = f"question:{hash(resume)}:{hash(job_posting)}"
# 캐시 확인
cached = redis_client.get(cache_key)
if cached:
return cached # 캐시 히트
# 캐시 미스 → LLM 생성
question = llm.invoke(prompt)
redis_client.setex(cache_key, 3600, question)
return question
| 장점 | 단점 |
|---|---|
| 빠른 캐시 | 정확한 매칭만 가능 |
| 간단한 구현 | 유사 질문 재사용 불가 |
| TTL 관리 필요 |
✅ 결정 (Decision)
Option 2: Question Pool + RAG Fallback 채택
| 버전 | 전략 | 비고 |
|---|---|---|
| V1~V2 | 매번 LLM 생성 | MVP, 간단 |
| V3 | Question Pool + RAG Fallback | 비용 절감 |
| V4 | Pool + Fallback + 배치 확장 | 자동 Pool 확장 |
📝 근거 (Rationale)
1. Question Pool 구축
초기 Pool (1,000개):
카테고리별 질문:
- 기술 질문: 400개 (Python, Django, FastAPI, DB 등)
- 인성 질문: 300개 (팀워크, 문제 해결, 커뮤니케이션 등)
- 경험 질문: 300개 (프로젝트, 성과, 실패 경험 등)
생성 방법:
1. 배치 처리로 미리 생성 (Celery Beat)
2. VectorDB에 저장 (Gemini Embedding)
3. 메타데이터: {category, difficulty, keywords}
2. RAG Fallback 로직
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from langchain_community.vectorstores import Chroma
# Question Pool (VectorDB)
embedding = GoogleGenerativeAIEmbeddings(model="models/gemini-embedding-001")
question_pool = Chroma(
collection_name="interview_questions",
embedding_function=embedding
)
def generate_interview_question(resume, job_posting):
"""면접 질문 생성 (Pool + Fallback)"""
# 1. 쿼리 생성
query = f"Resume: {resume[:200]}\nJob: {job_posting[:200]}"
# 2. Question Pool 검색
results = question_pool.similarity_search_with_score(query, k=3)
# 3. 유사도 체크
if results and results[0][1] > 0.85: # 유사도 > 0.85
# 캐시 히트 ✅
question = results[0][0].page_content
print(f"✅ Cache Hit! (유사도: {results[0][1]:.2f})")
return {
"question": question,
"source": "pool",
"cost": 0,
"latency_ms": 50
}
# 4. 캐시 미스 → LLM 생성 ⚠️
print(f"⚠️ Cache Miss! (유사도: {results[0][1]:.2f if results else 0})")
from langchain_google_genai import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(model="gemini-3-flash")
prompt = f"""
Resume: {resume}
Job Posting: {job_posting}
면접 질문을 생성하세요.
"""
question = llm.invoke(prompt).content
# 5. Pool에 추가 (학습)
question_pool.add_texts(
texts=[question],
metadatas=[{
"category": "auto_generated",
"timestamp": datetime.now().isoformat()
}]
)
return {
"question": question,
"source": "llm",
"cost": 0.001,
"latency_ms": 2000
}
3. 비용 절감 효과
시나리오 분석:
V3 예상 트래픽:
- 면접 세션: 50개/일
- 질문/세션: 5개
= 250 questions/일
캐시 히트율: 80% (Pool 성숙 후)
- 캐시 히트: 200 questions/일 → $0
- 캐시 미스: 50 questions/일 → $0.05/일
월 비용:
- 기존 (Option 1): $7.5/월
- 개선 (Option 2): $1.5/월
→ 80% 절감! ✅
4. 응답 속도 향상
| 케이스 | 응답 시간 | 비율 |
|---|---|---|
| 캐시 히트 | < 100ms | 80% |
| 캐시 미스 | 1~3초 | 20% |
| 평균 | ~500ms | 100% |
기존 (매번 LLM):
- 평균 응답: 1~3초
개선 (Pool + Fallback):
- 평균 응답: ~500ms (60% 향상!)
🔧 구현 예시
V3: Question Pool 초기화 (배치)
from celery import Celery
from langchain_google_genai import ChatGoogleGenerativeAI
app = Celery('tasks', broker='redis://localhost:6379')
@app.task
def initialize_question_pool():
"""Question Pool 초기화 (배치)"""
llm = ChatGoogleGenerativeAI(model="gemini-3-flash")
categories = [
{"name": "Python", "count": 50},
{"name": "Django", "count": 50},
{"name": "FastAPI", "count": 50},
{"name": "팀워크", "count": 50},
# ... 총 1,000개
]
for category in categories:
for i in range(category["count"]):
prompt = f"{category['name']} 관련 면접 질문 생성"
question = llm.invoke(prompt).content
# VectorDB에 저장
question_pool.add_texts(
texts=[question],
metadatas=[{"category": category["name"]}]
)
print("✅ Question Pool 초기화 완료! (1,000개)")
V4: 자동 Pool 확장 (배치)
@app.task
def expand_question_pool():
"""Question Pool 자동 확장 (매일 새벽)"""
# 1. 캐시 미스 로그 분석
cache_misses = db.get_cache_misses(last_7_days=True)
# 2. 자주 미스되는 패턴 추출
patterns = analyze_patterns(cache_misses)
# 3. 패턴별 질문 생성
for pattern in patterns:
questions = llm.batch_generate(pattern, count=10)
question_pool.add_texts(questions)
print(f"✅ Pool 확장 완료! (+{len(patterns) * 10}개)")
📊 비교표
| 항목 | 매번 LLM | Redis 캐싱 | Question Pool + RAG |
|---|---|---|---|
| 비용 | $7.5/월 | $3/월 | $1.5/월 ✅ |
| 응답 속도 | 1~3초 | < 100ms | ~500ms ✅ |
| 캐시 히트율 | 0% | 30% | 80% ✅ |
| 유사 질문 재사용 | ❌ | ❌ | ✅ |
| 자동 학습 | ❌ | ❌ | ✅ |
| 복잡도 | 🔴 | 🔴🔴 | 🔴🔴🔴 |
🎯 최종 추천
V1~V2 (MVP)
# 매번 LLM 생성
question = llm.invoke(prompt)
V3 (Question Pool)
# Pool + RAG Fallback
results = question_pool.similarity_search_with_score(query, k=3)
if results[0][1] > 0.85:
question = results[0][0] # 캐시 히트 ✅
else:
question = llm.invoke(prompt) # 캐시 미스
question_pool.add_texts([question]) # Pool에 추가
V4 (자동 확장)
# 배치로 Pool 자동 확장 (매일 새벽)
@celery.task
def expand_pool():
cache_misses = analyze_cache_misses()
new_questions = llm.batch_generate(cache_misses)
question_pool.add_texts(new_questions)
💡 핵심 정리
Question Pool + RAG Fallback:
- ✅ 비용 80% 절감 ($7.5 → $1.5/월)
- ✅ 응답 60% 향상 (1~3초 → ~500ms)
- ✅ 캐시 히트율 80%
- ✅ 자동 학습 (Pool 확장)
동작 방식:
1. Question Pool 검색 (VectorDB)
2. 유사도 > 0.85 → Pool에서 반환 (캐시 히트)
3. 유사도 < 0.85 → LLM 생성 (캐시 미스)
4. 생성된 질문 → Pool에 추가 (학습)
결론:
- ✅ V3부터 Question Pool 도입
- ✅ 비용 절감 + 응답 속도 향상
- ✅ 자동 학습으로 Pool 확장
📅 이력
| 날짜 | 변경 내용 |
|---|---|
| 2026-01-12 | 초기 결정 (Question Pool + RAG Fallback) |
ADR-024: VectorDB 메타데이터 필터링 vs RDB + VectorDB
📋 메타데이터
| 항목 | 내용 |
|---|---|
| 상태 | ✅ 승인됨 (Accepted) |
| 작성일 | 2026-01-12 |
| 결정자 | AI팀 |
| 관련 기능 | RAG, 데이터 조회, 필터링 |
🎯 컨텍스트 (Context)
RAG 시스템에서 특정 조건으로 데이터를 조회해야 합니다.
요구사항:
- 사용자별 이력서 조회 (
user_id) - 특정 카테고리 질문 조회 (
category) - 날짜 범위 조회 (
created_at) - 벡터 유사도 + 조건 필터링
문제점:
- VectorDB만 사용? RDB도 필요?
- 메타데이터 필터링 성능?
- 데이터 일관성 관리?
🔍 선택지 분석 (Options)
Option 1: VectorDB 메타데이터 필터링 ⭐
┌─────────────────────────────────────────────────────────────────┐
│ VectorDB 메타데이터 필터링 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ [VectorDB] (ChromaDB) │
│ ├─ 벡터 데이터 (임베딩) │
│ ├─ 텍스트 데이터 (원본) │
│ └─ 메타데이터 (user_id, category, created_at 등) │
│ │
│ [조회 예시] │
│ 질문: "Python 개발자 면접 질문" │
│ 필터: user_id = "user123", category = "기술" │
│ ↓ │
│ VectorDB.search( │
│ query="Python 개발자", │
│ filter={"user_id": "user123", "category": "기술"} │
│ ) │
│ ↓ │
│ 결과: 벡터 유사도 + 메타데이터 필터링 동시 적용 │
│ │
└─────────────────────────────────────────────────────────────────┘
구현 예시:
from langchain_community.vectorstores import Chroma
# VectorDB에 저장 (메타데이터 포함)
vectordb.add_texts(
texts=["Python 백엔드 개발 경험이 있나요?"],
metadatas=[{
"user_id": "user123",
"category": "기술",
"difficulty": "medium",
"created_at": "2026-01-12"
}]
)
# 메타데이터 필터링 조회
results = vectordb.similarity_search(
query="Python 개발자",
filter={"user_id": "user123", "category": "기술"},
k=5
)
| 장점 | 단점 |
|---|---|
| 단순한 구조 (VectorDB만) | 복잡한 쿼리 어려움 |
| 빠른 조회 (단일 DB) | JOIN 불가 |
| 일관성 보장 | 메타데이터 업데이트 어려움 |
| 관리 포인트 적음 | RDB 기능 제한적 |
Option 2: RDB + VectorDB (Hybrid)
┌─────────────────────────────────────────────────────────────────┐
│ RDB + VectorDB Hybrid │
├─────────────────────────────────────────────────────────────────┤
│ │
│ [RDB] (PostgreSQL) │
│ ├─ users (id, name, email, created_at) │
│ ├─ resumes (id, user_id, title, created_at) │
│ ├─ questions (id, category, difficulty, created_at) │
│ └─ vector_id (RDB ↔ VectorDB 연결) │
│ │
│ [VectorDB] (ChromaDB) │
│ ├─ 벡터 데이터 (임베딩) │
│ ├─ 텍스트 데이터 (원본) │
│ └─ 메타데이터 (최소한: id, vector_id) │
│ │
│ [조회 예시] │
│ 1. RDB에서 필터링 │
│ SELECT id FROM questions │
│ WHERE user_id = 'user123' AND category = '기술' │
│ → [q1, q2, q3] │
│ ↓ │
│ 2. VectorDB에서 벡터 검색 │
│ VectorDB.search( │
│ query="Python 개발자", │
│ filter={"id": {"$in": [q1, q2, q3]}} │
│ ) │
│ ↓ │
│ 3. RDB에서 추가 정보 조회 (JOIN) │
│ │
└─────────────────────────────────────────────────────────────────┘
구현 예시:
# 1. RDB에서 필터링
question_ids = db.execute("""
SELECT id FROM questions
WHERE user_id = :user_id AND category = :category
""", {"user_id": "user123", "category": "기술"})
# 2. VectorDB에서 벡터 검색
results = vectordb.similarity_search(
query="Python 개발자",
filter={"id": {"$in": question_ids}},
k=5
)
# 3. RDB에서 추가 정보 조회
for result in results:
question = db.get_question(result.metadata["id"])
# user, category, created_at 등 추가 정보
| 장점 | 단점 |
|---|---|
| 복잡한 쿼리 (JOIN, GROUP BY) | 복잡한 구조 |
| 메타데이터 관리 쉬움 | 두 DB 동기화 필요 |
| RDB 기능 활용 | 조회 느림 (2단계) |
| 확장성 좋음 | 관리 포인트 증가 |
Option 3: PostgreSQL + pgvector (통합)
# PostgreSQL + pgvector 확장
CREATE EXTENSION vector;
CREATE TABLE questions (
id SERIAL PRIMARY KEY,
user_id VARCHAR(50),
category VARCHAR(50),
content TEXT,
embedding vector(768), -- 벡터 컬럼
created_at TIMESTAMP
);
-- 벡터 유사도 + 메타데이터 필터링 (단일 쿼리)
SELECT * FROM questions
WHERE user_id = 'user123' AND category = '기술'
ORDER BY embedding <-> '[0.1, 0.5, ...]'::vector
LIMIT 5;
| 장점 | 단점 |
|---|---|
| 단일 DB (통합) | PostgreSQL 의존 |
| SQL 쿼리 가능 | VectorDB 전문 기능 부족 |
| 일관성 보장 | 성능 (대규모 벡터) |
✅ 결정 (Decision)
단계적 접근: VectorDB 메타데이터 → RDB + VectorDB
| 버전 | 선택 | 이유 |
|---|---|---|
| V1~V2 | VectorDB 메타데이터 | 단순, 빠름 |
| V3 | VectorDB 메타데이터 유지 | 충분히 동작 |
| V4 | RDB + VectorDB (검토) | 복잡한 쿼리 필요 시 |
📝 근거 (Rationale)
1. VectorDB 메타데이터 필터링 (V1~V3)
충분한 경우:
- ✅ 단순 필터링 (
user_id,category) - ✅ 벡터 검색이 주 목적
- ✅ JOIN 불필요
- ✅ 빠른 조회 필요
ChromaDB 메타데이터 필터링:
# 사용자별 이력서 조회
vectordb.similarity_search(
query="Python 백엔드 개발",
filter={"user_id": "user123"},
k=5
)
# 카테고리별 질문 조회
vectordb.similarity_search(
query="면접 질문",
filter={"category": "기술", "difficulty": "medium"},
k=10
)
# 날짜 범위 조회
vectordb.similarity_search(
query="최근 질문",
filter={
"created_at": {"$gte": "2026-01-01"}
},
k=5
)
장점:
- ✅ 단일 DB (관리 간단)
- ✅ 빠른 조회 (< 100ms)
- ✅ 일관성 보장
2. RDB + VectorDB가 필요한 경우 (V4)
복잡한 쿼리 필요:
- ⚠️ JOIN (users + resumes + questions)
- ⚠️ GROUP BY, HAVING
- ⚠️ 트랜잭션 필요
- ⚠️ 복잡한 비즈니스 로직
예시:
-- RDB에서만 가능한 쿼리
SELECT
u.name,
COUNT(q.id) as question_count,
AVG(q.difficulty) as avg_difficulty
FROM users u
JOIN questions q ON u.id = q.user_id
WHERE q.created_at > '2026-01-01'
GROUP BY u.name
HAVING COUNT(q.id) > 10;
하지만 프로젝트는 V3까지 불필요!
3. 프로젝트 적용 (V1~V3)
데이터 구조:
# VectorDB에 저장 (메타데이터 포함)
vectordb.add_texts(
texts=[
"Python 백엔드 개발 경험이 있나요?",
"Django ORM을 사용해본 적이 있나요?",
"팀 프로젝트에서 어떤 역할을 했나요?"
],
metadatas=[
{
"id": "q1",
"user_id": "user123",
"category": "기술",
"difficulty": "medium",
"created_at": "2026-01-12"
},
{
"id": "q2",
"user_id": "user123",
"category": "기술",
"difficulty": "hard",
"created_at": "2026-01-12"
},
{
"id": "q3",
"user_id": "user123",
"category": "인성",
"difficulty": "easy",
"created_at": "2026-01-12"
}
]
)
조회 예시:
# 1. 사용자별 기술 질문 조회
results = vectordb.similarity_search(
query="Python 개발",
filter={
"user_id": "user123",
"category": "기술"
},
k=5
)
# 2. 난이도별 질문 조회
results = vectordb.similarity_search(
query="면접 질문",
filter={
"difficulty": {"$in": ["medium", "hard"]}
},
k=10
)
# 3. 최근 질문 조회
results = vectordb.similarity_search(
query="최근 질문",
filter={
"created_at": {"$gte": "2026-01-01"}
},
k=5
)
📊 비교표
| 항목 | VectorDB 메타데이터 | RDB + VectorDB | PostgreSQL + pgvector |
|---|---|---|---|
| 구조 | 단순 | 복잡 | 중간 |
| 조회 속도 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐ |
| 복잡한 쿼리 | ❌ | ✅ | ✅ |
| JOIN | ❌ | ✅ | ✅ |
| 일관성 | ✅ | ⚠️ (동기화) | ✅ |
| 관리 | ✅ 쉬움 | ❌ 어려움 | ⚠️ 중간 |
| V3 적합성 | ✅ 추천 | ❌ 과함 | ⚠️ 가능 |
🎯 최종 추천
V1~V3 (VectorDB 메타데이터)
# 단순 필터링으로 충분
vectordb.similarity_search(
query="Python 개발자",
filter={"user_id": "user123", "category": "기술"},
k=5
)
이유:
- ✅ 단순한 구조
- ✅ 빠른 조회
- ✅ 관리 간단
- ✅ 프로젝트 요구사항 충족
V4 (RDB + VectorDB 검토)
필요한 경우:
- ⚠️ JOIN 필요 (users + resumes + questions)
- ⚠️ 복잡한 집계 (GROUP BY, HAVING)
- ⚠️ 트랜잭션 필요
- ⚠️ 비즈니스 로직 복잡
구현:
# 1. RDB 필터링
question_ids = db.execute("""
SELECT q.id FROM questions q
JOIN users u ON q.user_id = u.id
WHERE u.email = :email AND q.category = :category
""", {"email": "[email protected]", "category": "기술"})
# 2. VectorDB 검색
results = vectordb.similarity_search(
query="Python 개발자",
filter={"id": {"$in": question_ids}},
k=5
)
💡 핵심 정리
VectorDB 메타데이터 필터링:
- ✅ V1~V3 추천 (단순, 빠름)
- ✅ 단일 DB (관리 간단)
- ✅ 벡터 검색 + 필터링 동시 적용
- ✅ 프로젝트 요구사항 충족
RDB + VectorDB:
- ⚠️ V4 검토 (복잡한 쿼리 필요 시)
- ⚠️ JOIN, GROUP BY 필요
- ⚠️ 관리 복잡도 증가
결론:
- ✅ V1~V3: VectorDB 메타데이터 필터링
- ⚠️ V4: 복잡한 쿼리 필요 시 RDB + VectorDB 검토
🔧 구현 예시
VectorDB 메타데이터 필터링 (V3)
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from langchain_community.vectorstores import Chroma
# Embedding
embedding = GoogleGenerativeAIEmbeddings(model="models/gemini-embedding-001")
# VectorDB
vectordb = Chroma(
collection_name="interview_questions",
embedding_function=embedding
)
# 저장 (메타데이터 포함)
vectordb.add_texts(
texts=["Python 백엔드 개발 경험이 있나요?"],
metadatas=[{
"id": "q1",
"user_id": "user123",
"category": "기술",
"difficulty": "medium",
"created_at": "2026-01-12T10:00:00"
}]
)
# 조회 (메타데이터 필터링)
def search_questions(user_id, category, query):
"""사용자별 카테고리별 질문 검색"""
results = vectordb.similarity_search(
query=query,
filter={
"user_id": user_id,
"category": category
},
k=5
)
return results
# 사용
questions = search_questions(
user_id="user123",
category="기술",
query="Python 개발자"
)
📅 이력
| 날짜 | 변경 내용 |
|---|---|
| 2026-01-12 | 초기 결정 (VectorDB 메타데이터 필터링 우선) |
ADR-025: RDB ↔ VectorDB 데이터 동기화 전략
📋 메타데이터
| 항목 | 내용 |
|---|---|
| 상태 | ✅ 승인됨 (Accepted) |
| 작성일 | 2026-01-12 |
| 결정자 | AI팀 |
| 관련 기능 | 데이터 동기화, RAG, VectorDB |
🎯 컨텍스트 (Context)
RDB와 VectorDB 간 데이터 일관성을 유지해야 합니다.
데이터 소스:
- RDB (Supabase): 사용자 이력서, 채용공고, 면접 결과
- VectorDB (ChromaDB): 임베딩된 데이터, RAG용 벡터
문제점:
- RDB에 새 이력서 저장 → VectorDB에 언제 반영?
- 이력서 수정 시 VectorDB 업데이트?
- 데이터 일관성 보장?
요구사항:
- RDB와 VectorDB 동기화 전략
- 실시간 vs 배치 동기화
- 데이터 일관성 보장
🔍 선택지 분석 (Options)
Option 1: 실시간 동기화 (Sync)
┌─────────────────────────────────────────────────────────────────┐
│ 실시간 동기화 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ [사용자 요청] (이력서 업로드) │
│ ↓ │
│ [1. RDB 저장] (Supabase) │
│ - users, resumes 테이블 │
│ ↓ │
│ [2. 즉시 Embedding] (Gemini API) │
│ - 텍스트 → 벡터 변환 │
│ ↓ │
│ [3. VectorDB 저장] (ChromaDB) │
│ - 벡터 + 메타데이터 │
│ ↓ │
│ [응답] (완료) │
│ │
└─────────────────────────────────────────────────────────────────┘
구현:
@app.post("/ai/resume/upload")
async def upload_resume(file: UploadFile, user_id: str):
# 1. RDB 저장
resume_text = extract_text(file)
resume_id = db.insert_resume(user_id, resume_text)
# 2. 즉시 Embedding
embedding = gemini_embedding.embed_query(resume_text)
# 3. VectorDB 저장
vectordb.add_texts(
texts=[resume_text],
metadatas=[{
"id": resume_id,
"user_id": user_id,
"type": "resume"
}]
)
return {"status": "success", "resume_id": resume_id}
| 장점 | 단점 |
|---|---|
| 즉시 반영 | 응답 느림 (Embedding 시간) |
| 일관성 보장 | API 비용 증가 |
| 구현 간단 | 동시 요청 시 부하 |
Option 2: 배치 동기화 (Async) ⭐
┌─────────────────────────────────────────────────────────────────┐
│ 배치 동기화 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ [사용자 요청] (이력서 업로드) │
│ ↓ │
│ [1. RDB 저장] (Supabase) │
│ - users, resumes 테이블 │
│ - sync_status = 'pending' │
│ ↓ │
│ [응답] (즉시 완료) ✅ │
│ │
│ ───────────────────────────────────────────────────────────── │
│ │
│ [배치 처리] (매일 새벽 or 1시간마다) │
│ ↓ │
│ [1. RDB 조회] (sync_status = 'pending') │
│ ↓ │
│ [2. Batch Embedding] (Gemini API) │
│ - 여러 텍스트 한 번에 처리 │
│ ↓ │
│ [3. VectorDB 저장] (ChromaDB) │
│ ↓ │
│ [4. RDB 업데이트] (sync_status = 'synced') │
│ │
└─────────────────────────────────────────────────────────────────┘
구현:
# 1. 이력서 업로드 (즉시 응답)
@app.post("/ai/resume/upload")
async def upload_resume(file: UploadFile, user_id: str):
resume_text = extract_text(file)
resume_id = db.insert_resume(
user_id=user_id,
text=resume_text,
sync_status='pending' # 동기화 대기
)
return {"status": "success", "resume_id": resume_id}
# 2. 배치 동기화 (Celery Beat)
@celery.task
def sync_rdb_to_vectordb():
"""RDB → VectorDB 배치 동기화"""
# 1. 동기화 대기 중인 데이터 조회
pending_resumes = db.query("""
SELECT id, user_id, text
FROM resumes
WHERE sync_status = 'pending'
LIMIT 100
""")
if not pending_resumes:
return {"status": "no_data"}
# 2. Batch Embedding
texts = [r['text'] for r in pending_resumes]
embeddings = gemini_embedding.embed_documents(texts)
# 3. VectorDB 저장
vectordb.add_texts(
texts=texts,
embeddings=embeddings,
metadatas=[{
"id": r['id'],
"user_id": r['user_id'],
"type": "resume"
} for r in pending_resumes]
)
# 4. RDB 업데이트
resume_ids = [r['id'] for r in pending_resumes]
db.execute("""
UPDATE resumes
SET sync_status = 'synced', synced_at = NOW()
WHERE id = ANY(:ids)
""", {"ids": resume_ids})
return {"status": "success", "count": len(pending_resumes)}
| 장점 | 단점 |
|---|---|
| 빠른 응답 | 즉시 반영 안 됨 |
| Batch 처리 (비용 절감) | 동기화 지연 (최대 1시간) |
| 부하 분산 | 구현 복잡 |
Option 3: CDC (Change Data Capture)
# Supabase Realtime 활용
supabase.channel('resumes')
.on('INSERT', handle_insert)
.on('UPDATE', handle_update)
.subscribe()
async def handle_insert(payload):
"""RDB INSERT 시 자동 동기화"""
resume = payload['new']
# Embedding
embedding = await gemini_embedding.aembed_query(resume['text'])
# VectorDB 저장
await vectordb.aadd_texts(
texts=[resume['text']],
metadatas=[{"id": resume['id'], "user_id": resume['user_id']}]
)
| 장점 | 단점 |
|---|---|
| 실시간 동기화 | Supabase 의존 |
| 자동화 | 복잡도 증가 |
| 비용 (Realtime API) |
✅ 결정 (Decision)
단계적 접근: 실시간 → 배치 → CDC
| 버전 | 전략 | 이유 |
|---|---|---|
| V1~V2 | 실시간 동기화 | 간단, 즉시 반영 |
| V3 | 배치 동기화 | 비용 절감, 부하 분산 |
| V4+ | CDC 검토 | 실시간 + 자동화 |
📝 근거 (Rationale)
1. V1~V2: 실시간 동기화
충분한 경우:
- ✅ 트래픽 적음 (< 100 uploads/일)
- ✅ 즉시 RAG 필요
- ✅ 구현 간단
구현:
# 이력서 업로드 → 즉시 Embedding → VectorDB
@app.post("/ai/resume/upload")
async def upload_resume(file: UploadFile, user_id: str):
# 1. 텍스트 추출
resume_text = extract_text(file)
# 2. RDB 저장
resume_id = db.insert_resume(user_id, resume_text)
# 3. 즉시 Embedding
embedding = gemini_embedding.embed_query(resume_text)
# 4. VectorDB 저장
vectordb.add_texts(
texts=[resume_text],
metadatas=[{"id": resume_id, "user_id": user_id}]
)
return {"resume_id": resume_id}
2. V3: 배치 동기화
필요한 경우:
- ⚠️ 트래픽 증가 (> 100 uploads/일)
- ⚠️ Embedding 비용 절감 필요
- ⚠️ 응답 속도 중요
RDB 스키마:
CREATE TABLE resumes (
id UUID PRIMARY KEY,
user_id UUID NOT NULL,
text TEXT NOT NULL,
sync_status VARCHAR(20) DEFAULT 'pending', -- pending, synced, failed
synced_at TIMESTAMP,
created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_sync_status ON resumes(sync_status);
배치 스케줄:
# Celery Beat 설정
CELERYBEAT_SCHEDULE = {
'sync-rdb-to-vectordb': {
'task': 'tasks.sync_rdb_to_vectordb',
'schedule': crontab(minute='*/30'), # 30분마다
},
}
장점:
- ✅ 빠른 응답 (Embedding 비동기)
- ✅ Batch 처리로 비용 절감
- ✅ 부하 분산
3. 데이터 일관성 보장
동기화 상태 관리:
# 동기화 상태 확인
@app.get("/ai/resume/{resume_id}/sync-status")
async def get_sync_status(resume_id: str):
resume = db.get_resume(resume_id)
return {
"resume_id": resume_id,
"sync_status": resume['sync_status'],
"synced_at": resume['synced_at'],
"rag_ready": resume['sync_status'] == 'synced'
}
재시도 로직:
@celery.task(bind=True, max_retries=3)
def sync_rdb_to_vectordb(self):
try:
# 동기화 로직
...
except Exception as exc:
# 실패 시 재시도
db.execute("""
UPDATE resumes
SET sync_status = 'failed'
WHERE sync_status = 'pending'
""")
raise self.retry(exc=exc, countdown=300)
🔧 구현 예시
V3: 배치 동기화 (전체 흐름)
from celery import Celery
from langchain_google_genai import GoogleGenerativeAIEmbeddings
app = Celery('tasks', broker='redis://localhost:6379')
# Gemini Embedding
embedding = GoogleGenerativeAIEmbeddings(model="models/gemini-embedding-001")
@celery.task
def sync_rdb_to_vectordb():
"""RDB → VectorDB 배치 동기화 (30분마다)"""
# 1. 동기화 대기 중인 데이터 조회
pending_data = db.query("""
SELECT id, user_id, text, type
FROM (
SELECT id, user_id, text, 'resume' as type
FROM resumes
WHERE sync_status = 'pending'
UNION ALL
SELECT id, user_id, text, 'job_posting' as type
FROM job_postings
WHERE sync_status = 'pending'
) AS pending
LIMIT 100
""")
if not pending_data:
return {"status": "no_data"}
# 2. Batch Embedding (비용 절감)
texts = [d['text'] for d in pending_data]
embeddings = embedding.embed_documents(texts)
# 3. VectorDB 저장
vectordb.add_texts(
texts=texts,
embeddings=embeddings,
metadatas=[{
"id": d['id'],
"user_id": d['user_id'],
"type": d['type']
} for d in pending_data]
)
# 4. RDB 업데이트 (sync_status = 'synced')
for data in pending_data:
table = 'resumes' if data['type'] == 'resume' else 'job_postings'
db.execute(f"""
UPDATE {table}
SET sync_status = 'synced', synced_at = NOW()
WHERE id = :id
""", {"id": data['id']})
return {
"status": "success",
"count": len(pending_data),
"types": {
"resume": sum(1 for d in pending_data if d['type'] == 'resume'),
"job_posting": sum(1 for d in pending_data if d['type'] == 'job_posting')
}
}
📊 비교표
| 항목 | 실시간 동기화 | 배치 동기화 | CDC |
|---|---|---|---|
| 응답 속도 | 느림 (1~3초) | ⭐⭐⭐⭐⭐ | 빠름 |
| 동기화 지연 | 없음 | 최대 30분 | 없음 |
| 비용 | 높음 | 낮음 ✅ | 중간 |
| 복잡도 | 🔴 낮음 | 🔴🔴 | 🔴🔴🔴 |
| 일관성 | ✅ | ⚠️ (지연) | ✅ |
| V3 적합성 | ⚠️ | ✅ 추천 | ❌ 과함 |
🎯 최종 추천
V1~V2 (실시간)
# 즉시 동기화
@app.post("/ai/resume/upload")
async def upload_resume(file, user_id):
text = extract_text(file)
resume_id = db.insert_resume(user_id, text)
# 즉시 Embedding
embedding = gemini_embedding.embed_query(text)
vectordb.add_texts([text], metadatas=[{"id": resume_id}])
return {"resume_id": resume_id}
V3 (배치)
# 1. 빠른 응답
@app.post("/ai/resume/upload")
async def upload_resume(file, user_id):
text = extract_text(file)
resume_id = db.insert_resume(user_id, text, sync_status='pending')
return {"resume_id": resume_id} # 즉시 응답 ✅
# 2. 배치 동기화 (30분마다)
@celery.task
def sync_rdb_to_vectordb():
pending = db.get_pending_data()
embeddings = gemini_embedding.embed_documents([d['text'] for d in pending])
vectordb.add_texts(...)
db.update_sync_status('synced')
💡 핵심 정리
RDB ↔ VectorDB 동기화:
- ✅ V1~V2: 실시간 동기화 (간단)
- ✅ V3: 배치 동기화 (비용 절감)
- ⚠️ V4+: CDC 검토 (실시간 + 자동화)
배치 동기화 장점:
- ✅ 빠른 응답 (Embedding 비동기)
- ✅ Batch 처리로 비용 절감
- ✅ 부하 분산 (30분마다)
데이터 일관성:
- ✅ sync_status 필드로 상태 관리
- ✅ 재시도 로직 (최대 3회)
- ✅ 동기화 상태 API 제공
📅 이력
| 날짜 | 변경 내용 |
|---|---|
| 2026-01-12 | 초기 결정 (실시간 → 배치 → CDC 단계적 접근) |