RAG: 기업용 지능형 문서 검색의 혁신
RAG는 벡터 검색과 생성형 AI를 결합해 사내 문서 저장소에서 정확한 답변을 제공합니다. 시스템은 PDF, 엑셀 파일, 코드 등 다양한 형식의 문서에서 관련 내용을 추출한 후, 원본 자료에만 기반해 설명을 생성하므로 환상적 오류가 크게 줄어듭니다.
프로세스는 전처리 단계에서 시작됩니다. 문서는 파싱되어 콘텐츠 유형을 고려한 지능적인 청크로 분할되며, 벡터 데이터베이스에 저장된 임베딩으로 변환됩니다.
파일 형식에서 텍스트 추출
시스템은 데이터 손실 없이 여러 형식을 지원하며, 각 형식마다 전용 추출기가 사용됩니다.
SUPPORTED = {
'.pdf': lambda f: ''.join(p.extract_text() or '' for p in PyPDF2.PdfReader(f).pages),
'.docx': lambda f: '\n'.join(p.text for p in Document(f).paragraphs),
'.pptx': lambda f: '\n'.join(
shape.text
for slide in Presentation(f).slides
for shape in slide.shapes
if hasattr(shape, "text") and shape.text.strip()
),
'.xlsx': extract_xlsx,
'.csv': extract_csv,
# .txt, .json, .py, .js 등
}출력은 깨끗하고 처리 가능한 텍스트입니다.
콘텐츠 유형별 지능형 청크 분할
텍스트는 고정 크기로 나누지 않고, 의미적으로 세분화됩니다. 먼저 콘텐츠 유형이 감지됩니다:
def detect_content_type(text):
if re.search(r'<table|</td>|<tr>', text, re.IGNORECASE):
return 'table'
tab_lines = sum(1 for line in text.splitlines() if '\t' in line)
if tab_lines > max(3, len(text.splitlines()) * 0.4):
return 'table'
if re.search(r'function\s+\w+\s*\(|\bdef\s+\w+\s*\(|\bclass\s+\w+\b', text):
return 'code'
return 'prose'- 표: 각 청크에 반복되는 헤더 포함
- 코드: 함수/클래스 단위로 분할하면서 논리 구조 유지
- 일반 텍스트: 문장 임베딩을 활용한 의미적 경계 적용
표 처리 예시:
def split_table_by_rows(text, chunk_size):
lines = [l for l in text.splitlines() if l.strip()]
header = lines[0]
chunks = []
cur_lines = [header]
for line in lines[1:]:
if sum(len(l) for l in cur_lines) + len(line) >= chunk_size:
chunks.append('\n'.join(cur_lines))
cur_lines = [header, line]
else:
cur_lines.append(line)
if cur_lines:
chunks.append('\n'.join(cur_lines))
return chunks의미적 청크 분할은 인접 문장 간 코사인 유사도를 기반으로 하며, 적응형 임계값(25퍼센트 백분위수)을 사용합니다.
임베딩 생성
청크는 text-embedding-3-small(1536차원) 같은 모델을 사용해 벡터화되며, 벡터와 메타데이터는 데이터베이스에 저장됩니다.
def get_embeddings(texts, batch_size=100):
embeddings = []
for i in range(0, len(texts), batch_size):
resp = requests.post(
"https://openrouter.ai/api/v1/embeddings",
headers=HEADERS,
json={
"model": "openai/text-embedding-3-small",
"input": texts[i:i + batch_size],
"encoding_format": "float"
},
).json()
embeddings.extend(d["embedding"] for d in resp.get("data", []))
return np.array(embeddings)다단계 검색
쿼리는 벡터로 변환되며, LLM을 통해 세 가지 형태로 재구성된 후 하이브리드 검색을 거칩니다.
하이브리드 검색: 벡터 + BM25
- 벡터 검색: 임베딩 공간 내 코사인 유사도
- BM25: k1=1.5, b=0.75 설정의 TF-IDF 기반 정확한 매칭
결과는 RRF(Reciprocal Rank Fusion) 방식으로 융합됩니다:
# RRF 융합
for rank, idx in enumerate(sorted_by_vector):
rrf_scores[idx] += 1.0 / (k_rrf + rank)
for rank, idx in enumerate(sorted_by_bm25):
rrf_scores[idx] += bm25_normalized[idx] * (1.0 / (k_rrf + rank))재정렬 및 필터링
상위 50개 후보를 LLM이 평가하여 최적의 5~10개를 선택합니다. 강제 포함 규칙도 적용됩니다:
- 따옴표 안의 문장("..." 또는 «...»)
- 숫자/아이디 (정규식: \b(№|N|number)\s*\d+\b)
quote_matches = re.findall(r'«([^»]+)»|"([^"+])"', query)
for match in quote_matches:
quote = (match[0] or match[1]).strip()
if len(quote) > 5:
for i, chunk in enumerate(chunks_data):
if quote in chunk['content']:
forced_indices.append(i)
break성능 최적화
계층적 검색
먼저 주요 문서를 주석 기반으로 식별한 후, 해당 문서 내 청크를 검색합니다.
doc_summaries = [chunks_data_by_doc[doc_id][0]['content'][:500] for doc_id in all_doc_ids]
doc_embs = get_embeddings(doc_summaries)
sims = np.dot(doc_embs_normalized, query_embedding_normalized)
top_doc_ids = [all_doc_ids[i] for i in np.argsort(sims)[::-1][:5]]
chunks_data = [ch for ch in chunks_data if ch['doc_id'] in top_doc_ids]반복 검색
복잡한 쿼리에는 LLM 피드백을 활용해 여러 라운드를 거쳐 결과를 점진적으로 개선합니다.
핵심 요약
- RAG는 관련 청크에서만 답변을 생성함으로써 환상적 오류를 최소화합니다.
- 하이브리드 검색(벡터 + BM25 + RRF)은 순수 벡터 검색보다 정확도를 20~30% 향상시킵니다.
- 의미적 청크 분할은 표와 코드의 맥락을 보존해 노이즈를 줄입니다.
- 계층적 접근은 대규모 문서 집합(1만 건 이상)에서 검색 속도를 높입니다.
- 강제 필터는 숫자나 인용문 포함 사실의 검색을 보장합니다.
— Editorial Team
아직 댓글이 없습니다.