서론: 왜 하이브리드 RAG가 중요한가
프로덕션 환경에서 RAG 시스템을 구축하다 보면 대부분 단일 데이터 소스만 다루는 단순한 시나리오에서 시작합니다. 그러나 실제 비즈니스 데이터는 훨씬 복잡합니다. 저는去年某 제조 기업에서 3개월간 하이브리드 RAG 시스템을 구축하면서 이 문제를 깊이 다뤄야 한다는 것을 깨달았습니다.
해당 기업은:
- 구조화 데이터: ERP 시스템의 제품 테이블, 재고 현황, 거래 내역 (PostgreSQL, BigQuery)
- 반구조화 데이터: QC 보고서, 검사 로그 (JSON, XML)
- 비구조화 데이터: 기술 문서, 메뉴얼, 이메일 (PDF, Markdown)
이 세 가지 데이터 소스를 통합 검색해야 하는 요구사항이 있었습니다. 이 글에서는 HolySheep AI의 단일 API 키로 이 세 가지数据类型을 효율적으로 검색하는 아키텍처를 설명드리겠습니다.
아키텍처 설계
전체 시스템 구조
┌─────────────────────────────────────────────────────────────────────┐
│ Hybrid Table Data RAG │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Structured │ │ Semi- │ │ Unstructured │ │
│ │ Data │ │ Structured │ │ Data │ │
│ │ (PostgreSQL) │ │ (JSON Logs) │ │ (Documents) │ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Table │ │ JSON Schema │ │ Document │ │
│ │ Indexer │ │ Extractor │ │ Chunker │ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │
│ └───────────────────┼───────────────────┘ │
│ ▼ │
│ ┌──────────────────┐ │
│ │ Vector Store │ │
│ │ (Unified Index) │ │
│ └────────┬─────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────┐ │
│ │ Query Router │ │
│ │ (Hybrid Search) │ │
│ └────────┬─────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────┐ │
│ │ Reranker + LLM │ │
│ │ (HolySheep AI) │ │
│ └──────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘핵심 컴포넌트 설명
1. Table Indexer (구조화 데이터): SQL 쿼리 결과를 스키마 인식 방식으로 벡터화합니다. 컬럼명, 데이터 타입, 관계 정보를 메타데이터로 보존합니다.
2. JSON Schema Extractor (반구조화 데이터): 로그와 설정 파일에서 스키마를 추출하고 계층 구조를 유지하면서 임베딩합니다.
3. Document Chunker (비구조화 데이터): 문서를 의미 단위로 분할합니다. 테이블 데이터의 경우 행 단위, 문서는 제목-내용 관계를 고려합니다.
4. Unified Index: 세 소스의 임베딩을 통합 벡터 DB에 저장합니다. 각 임베딩에 데이터 소스 타입과 스키마 정보를 태깅합니다.
5. Query Router: 사용자 질문을 분석하여 적절한 데이터 소스에 검색하고, 결과를 합성합니다.
구현 코드
1단계: 데이터 소스 설정 및 인덱싱
import json
import sqlite3
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime
import asyncio
import aiohttp
from openai import AsyncOpenAI
HolySheep AI 클라이언트 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
client = AsyncOpenAI(
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL
)
@dataclass
class TableRecord:
"""구조화된 테이블 레코드"""
table_name: str
columns: List[str]
values: List[Any]
primary_key: Optional[str] = None
foreign_keys: Dict[str, str] = field(default_factory=dict)
@dataclass
class DocumentChunk:
"""비구조화 문서 청크"""
content: str
source: str
chunk_type: str # 'table_row', 'paragraph', 'list_item'
metadata: Dict[str, Any] = field(default_factory=dict)
class HybridIndexer:
"""
하이브리드 RAG 인덱서
구조화, 반구조화, 비구조화 데이터를 통합 인덱싱합니다.
실제 프로덕션에서는 PostgreSQL, S3, Elasticsearch 등을 연결합니다.
"""
def __init__(self):
self.embeddings = []
self.metadata = []
async def embed_text(self, text: str, metadata: Dict[str, Any]) -> List[float]:
"""HolySheep AI를 사용한 임베딩 생성"""
response = await client.embeddings.create(
model="text-embedding-3-small",
input=text
)
embedding = response.data[0].embedding
self.embeddings.append(embedding)
self.metadata.append({
**metadata,
"embedded_at": datetime.now().isoformat()
})
return embedding
async def index_structured_table(
self,
table_name: str,
columns: List[str],
rows: List[tuple],
primary_key: str = None
) -> int:
"""
구조화된 테이블 데이터 인덱싱
Args:
table_name: 테이블 이름
columns: 컬럼명 리스트
rows: 데이터 행 리스트
primary_key: 기본키 컬럼명
Returns:
인덱싱된 레코드 수
실제 환경에서는:
- SELECT * FROM products WHERE updated_at > last_sync_time
- 배치 인덱싱으로 대용량 데이터 처리
- 증분 동기화机制 구현
"""
indexed_count = 0
for row in rows:
# 테이블 데이터를 자연어로 변환
# 예: ["iPhone 15", "Apple", 1299, 50] ->
# "제품명: iPhone 15, 제조사: Apple, 가격: 1299美元, 재고: 50개"
row_dict = dict(zip(columns, row))
natural_text = self._table_row_to_text(table_name, row_dict)
# 메타데이터에 스키마 정보 포함
metadata = {
"data_type": "structured",
"table_name": table_name,
"columns": columns,
"primary_key": primary_key,
"row_data": row_dict
}
await self.embed_text(natural_text, metadata)
indexed_count += 1
print(f"✓ {table_name}: {indexed_count}개 레코드 인덱싱 완료")
return indexed_count
def _table_row_to_text(self, table_name: str, row: Dict[str, Any]) -> str:
"""테이블 행을 검색 가능한 자연어 텍스트로 변환"""
parts = [f"테이블: {table_name}"]
for key, value in row.items():
if value is not None:
parts.append(f"{key}: {value}")
return ", ".join(parts)
async def index_json_logs(self, logs: List[Dict], source: str) -> int:
"""
반구조화 JSON 로그 인덱싱
스키마 정보를 추출하여 검색 가능성을 높입니다.
실제 환경에서는:
- CloudWatch Logs, Elasticsearch에서 수집
- 스키마 진화 처리 (新增 필드 호환)
"""
indexed_count = 0
for log in logs:
# 계층 구조를 플랫 텍스트로 변환
text_parts = [f"로그 소스: {source}"]
# 최상위 필드
for key, value in log.items():
if isinstance(value, (str, int, float, bool)):
text_parts.append(f"{key}: {value}")
elif isinstance(value, list):
text_parts.append(f"{key}: {len(value)}개 항목")
# 중첩 객체 처리
nested = {k: v for k, v in log.items() if isinstance(v, dict)}
if nested:
text_parts.append(f"세부 정보: {json.dumps(nested, ensure_ascii=False)}")
metadata = {
"data_type": "semi_structured",
"source": source,
"schema_keys": list(log.keys())
}
await self.embed_text(" | ".join(text_parts), metadata)
indexed_count += 1
print(f"✓ JSON 로그 ({source}): {indexed_count}개 인덱싱 완료")
return indexed_count
사용 예시
async def main():
indexer = HybridIndexer()
# 1. 구조화된 제품 테이블 인덱싱
product_columns = ["product_id", "name", "brand", "price_usd", "stock"]
product_rows = [
(1, "iPhone 15 Pro", "Apple", 999, 150),
(2, "Galaxy S24", "Samsung", 899, 200),
(3, "Pixel 8", "Google", 699, 75),
]
await indexer.index_structured_table(
"products",
product_columns,
product_rows,
primary_key="product_id"
)
# 2. 반구조화 로그 인덱싱
logs = [
{"level": "ERROR", "service": "order-api", "message": "Payment failed", "error_code": 5001},
{"level": "WARN", "service": "inventory", "message": "Low stock alert", "sku": "IPH15PRO"},
{"level": "INFO", "service": "shipping", "message": "Order shipped", "tracking": "TRK123456"}
]
await indexer.index_json_logs(logs, "microservices")
print(f"\n총 {len(indexer.embeddings)}개 임베딩 생성 완료")
asyncio.run(main())2단계: 하이브리드 검색 및 쿼리 라우팅
import numpy as np
from enum import Enum
from typing import List, Tuple, Optional
class DataSourceType(Enum):
STRUCTURED = "structured"
SEMI_STRUCTURED = "semi_structured"
UNSTRUCTURED = "unstructured"
@dataclass
class SearchResult:
content: str
score: float
data_type: DataSourceType
metadata: Dict[str, Any]
source_table: Optional[str] = None
class HybridQueryRouter:
"""
하이브리드 쿼리 라우터
사용자 질문의 의도를 분석하여 적절한 데이터 소스에 검색합니다.
검색 전략:
1. 질문 분석 -> 관련 데이터 소스 식별
2. 각 소스에 병렬 검색
3. 결과 재순위화 (Cross-Encoder Reranking)
4. 컨텍스트 합성
"""
# 데이터 소스 키워드 매핑
SOURCE_KEYWORDS = {
DataSourceType.STRUCTURED: [
"가격", "재고", "수량", "판매", "매출", "고객", "주문",
"price", "stock", "quantity", "sales", "revenue"
],
DataSourceType.SEMI_STRUCTURED: [
"로그", "에러", "경고", "오류", "실패", "시스템",
"log", "error", "warning", "alert", "exception"
],
DataSourceType.UNSTRUCTURED: [
"문서", "메뉴얼", "가이드", "설명서", "정책",
"document", "manual", "guide", "policy", "how to"
]
}
def __init__(self, indexer: HybridIndexer, top_k: int = 10):
self.indexer = indexer
self.top_k = top_k
def classify_query_intent(self, query: str) -> List[Tuple[DataSourceType, float]]:
"""
질문의 의도를 분석하여 관련 데이터 소스를 점수화
Returns:
[(데이터소스, 관련성 점수), ...] 점수 내림차순 정렬
실제 환경에서는:
- LLM을 사용한 제로샷 분류
- Few-shot 학습으로 정확도 향상
- 사용자 피드백 기반 점진적 개선
"""
query_lower = query.lower()
scores = {}
for source_type, keywords in self.SOURCE_KEYWORDS.items():
# 키워드 매칭 점수
keyword_matches = sum(1 for kw in keywords if kw in query_lower)
# BM25 스코어 (간단한 버전)
doc_length = len(query.split())
bm25_score = keyword_matches / doc_length if doc_length > 0 else 0
scores[source_type] = bm25_score
# 점수 정규화 및 정렬
total = sum(scores.values()) + 0.001 # 0除防止
sorted_scores = sorted(
[(k, v / total) for k, v in scores.items()],
key=lambda x: x[1],
reverse=True
)
return sorted_scores
async def hybrid_search(
self,
query: str,
data_types: Optional[List[DataSourceType]] = None,
top_k: Optional[int] = None
) -> List[SearchResult]:
"""
하이브리드 검색 실행
Args:
query: 검색 질의
data_types: 검색할 데이터 소스 타입 (None이면 자동 탐지)
top_k: 반환할 결과 수
Returns:
검색 결과 리스트 (점수 순)
성능 최적화:
- 각 데이터 소스별 병렬 검색
- early stopping으로 불필요한 연산 제거
- 결과 캐싱 (Redis 활용)
"""
k = top_k or self.top_k
# 1. 쿼리 의도 분류
if data_types is None:
intent_scores = self.classify_query_intent(query)
data_types = [ds for ds, score in intent_scores if score > 0.1]
# 관련 소스가 없으면 전체 검색
if not data_types:
data_types = list(DataSourceType)
# 2. 쿼리 임베딩
query_embedding = await self._embed_query(query)
# 3. 각 데이터 소스별 검색 (병렬 실행)
search_tasks = []
for source_type in data_types:
task = self._search_by_source(
query_embedding,
source_type,
k=3 # 소스당 상위 3개
)
search_tasks.append(task)
all_results = await asyncio.gather(*search_tasks)
# 4. 결과 병합 및 재순위화
merged = []
for results in all_results:
merged.extend(results)
# 코사인 유사도로 최종 순위 결정
merged = sorted(merged, key=lambda x: x.score, reverse=True)
return merged[:k]
async def _embed_query(self, query: str) -> List[float]:
"""쿼리 임베딩 생성"""
response = await client.embeddings.create(
model="text-embedding-3-small",
input=query
)
return response.data[0].embedding
async def _search_by_source(
self,
query_embedding: List[float],
data_type: DataSourceType,
k: int
) -> List[SearchResult]:
"""특정 데이터 소스에서 검색"""
results = []
query_vec = np.array(query_embedding)
for idx, emb in enumerate(self.indexer.embeddings):
metadata = self.indexer.metadata[idx]
# 데이터 소스 필터링
if metadata.get("data_type") != data_type.value:
continue
# 코사인 유사도 계산
emb_vec = np.array(emb)
similarity = np.dot(query_vec, emb_vec) / (
np.linalg.norm(query_vec) * np.linalg.norm(emb_vec) + 1e-8
)
# 컨텐츠 재구성
content = self._reconstruct_content(metadata)
results.append(SearchResult(
content=content,
score=float(similarity),
data_type=data_type,
metadata=metadata,
source_table=metadata.get("table_name")
))
# 소스별 상위 k개 반환
results.sort(key=lambda x: x.score, reverse=True)
return results[:k]
def _reconstruct_content(self, metadata: Dict) -> str:
"""메타데이터에서 검색 가능한 콘텐츠 재구성"""
data_type = metadata.get("data_type")
if data_type == "structured":
# 테이블 데이터 포맷
row_data = metadata.get("row_data", {})
parts = []
for k, v in row_data.items():
parts.append(f"{k}: {v}")
return f"[{metadata['table_name']}] " + " | ".join(parts)
elif data_type == "semi_structured":
# 로그 데이터 포맷
return metadata.get("content", str(metadata))
else:
return metadata.get("content", "")
검색 실행 예시
async def search_example():
router = HybridQueryRouter(indexer)
# 구조화 데이터 의도가 강한 질문
query1 = "Apple 제품 중 1000달러 이하인 것有哪些?"
results1 = await router.hybrid_search(query1)
print(f"\n검색어: {query1}")
print(f"결과: {len(results1)}개")
for r in results1:
print(f" [{r.data_type.value}] {r.content[:80]}... (score: {r.score:.3f})")
# 로그 검색
query2 = "최근 에러 로그 보여줘"
results2 = await router.hybrid_search(query2)
print(f"\n검색어: {query2}")
print(f"결과: {len(results2)}개")
asyncio.run(search_example())3단계: HolySheep AI를 활용한 결과 합성
from openai import OpenAI
class TableDataSynthesizer:
"""
테이블 RAG 결과 합성기
HolySheep AI의 다중 모델 지원을 활용하여:
- DeepSeek V3 (저렴): 구조화된 테이블 요약
- Claude Sonnet (고품질): 복잡한 분석 및 추론
- Gemini 2.5 Flash (빠름): 실시간 검색 결과 병합
비용 최적화 전략:
1. Cache-aware prompting으로 토큰 절약
2. Streaming response로 TTFT 개선
3. 배치 처리로 API 호출 횟수 최소화
"""
def __init__(self):
self.client = OpenAI(
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL
)
async def synthesize_response(
self,
query: str,
search_results: List[SearchResult],
model: str = "deepseek/deepseek-chat-v3-0324"