Study/RAG(Retrieval-Augmented Generation)

1. RAG 파이프라인 기초 아키텍처

bluebamus 2026. 2. 17.

1. RAG(Retrieval-Augmented Generation) 개요

- RAG는 LLM이 학습하지 않은 외부 지식을 검색(Retrieval)하여 생성(Generation)에 활용하는 패턴이다. LLM의 환각(Hallucination)을 줄이고, 최신 정보나 도메인 특화 지식을 활용할 수 있게 한다.

1.1. RAG가 필요한 이유

문제	RAG의 해결 방식
LLM의 지식 한계 (학습 데이터 컷오프)	외부 문서에서 최신 정보 검색
환각 (Hallucination)	검색된 문서를 근거로 답변 생성
도메인 특화 지식 부재	기업/도메인 전용 문서를 벡터화하여 활용
답변의 투명성 부족	출처(source) 추적 가능
모델 재학습 비용	문서 업데이트만으로 지식 갱신

1.2. RAG 유형 분류

유형	설명	특징
Naive RAG	기본 검색 + 생성	단순 파이프라인, 질의 → 검색 → 생성
Advanced RAG	전처리/후처리 강화	질의 변환, Re-ranking, 답변 검증 등 추가
Modular RAG	모듈화된 구조	각 단계를 독립 모듈로 구성, 유연한 조합/교체 가능

1.2.1. Naive RAG

- 문서 로딩→청킹→임베딩→검색→생성의 기본 흐름으로, 빠르게 프로토타입을 만들 때 적합하다. 사용자 질의를 그대로 임베딩하여 벡터 DB에서 유사 문서를 검색하고, 검색된 문서를 프롬프트에 삽입하여 LLM이 답변을 생성하는 가장 단순한 구조이다. 추가적인 전처리나 후처리 없이 동작하므로 구현이 간단하지만, 복잡한 질의나 노이즈가 많은 문서에서는 품질이 떨어질 수 있다.

1.2.2. Advanced RAG

- Naive RAG의 한계를 보완하기 위해 파이프라인의 전처리/후처리 단계를 강화한 방식이다. 주요 기법은 다음과 같다:

1.2.2.1. 질의 변환 (Query Transformation)

- 사용자의 원본 질의가 검색에 최적화되어 있지 않은 경우가 많다. 질의 변환은 검색 성능을 높이기 위해 질의를 재구성하는 전처리 단계이다.

1.2.2.1.1. HyDE (Hypothetical Document Embedding):

- 사용자 질의에 대한 "가상의 답변 문서"를 LLM으로 먼저 생성한 뒤, 그 가상 문서를 임베딩하여 벡터 검색에 사용한다. 질의(질문)와 문서(답변)의 임베딩 공간 차이를 줄여 검색 정확도를 높이는 기법이다. 예를 들어 "소득세 계산법은?"이라는 질문 대신, LLM이 생성한 "소득세는 과세표준에 세율을 곱하여 계산하며..."라는 가상 답변을 임베딩하면, 실제 문서와 임베딩 공간에서 더 가까워진다.

1.2.2.1.2. 질의 분해 (Query Decomposition):

- 복합적인 질의를 여러 개의 하위 질의로 분해하여 각각 검색한 뒤 결과를 종합한다. "소득세와 법인세의 차이점과 각각의 계산 방법은?"이라는 복합 질의를 "소득세의 계산 방법은?", "법인세의 계산 방법은?", "소득세와 법인세의 차이점은?" 세 개의 하위 질의로 분해하여 각각에 대해 더 정확한 검색 결과를 얻는다.

1.2.2.2. Re-ranking (재순위화)

- 벡터 검색으로 후보 문서를 넓게 가져온 뒤(예: 상위 20개), Cross-encoder 모델이나 별도의 Re-ranker 모델(Cohere Rerank, bge-reranker 등)로 질의-문서 쌍의 관련성을 정밀하게 재평가하여 최종 상위 K개를 선별한다. 벡터 검색의 Bi-encoder 방식보다 느리지만 정확도가 높아, 1차 검색(recall) 후 2차 정밀 순위(precision)를 매기는 2단계 전략에 사용된다.

1.2.2.3. Self-check (자기 검증)

- LLM이 생성한 답변을 다시 LLM에게 검증시키는 후처리 단계이다. 생성된 답변이 검색된 문서의 내용과 일치하는지(충실성), 질의에 적절히 답변하고 있는지(관련성)를 별도의 프롬프트로 평가한다. 점수가 낮으면 재생성하거나 "답변할 수 없음"으로 처리하여 환각을 줄인다.

1.2.2.4. 하이브리드 검색 (Hybrid Search)

- 벡터 유사도 검색(의미 기반)과 키워드 검색(BM25, TF-IDF 기반)을 결합한 방식이다. 의미적으로 유사한 문서를 놓치지 않으면서도, 특정 키워드가 정확히 포함된 문서도 함께 검색할 수 있다. 예를 들어 "제127조의2"와 같은 법률 조항 번호는 벡터 검색으로는 찾기 어렵지만 키워드 검색으로는 정확히 매칭된다. 두 검색 결과를 Reciprocal Rank Fusion(RRF) 등으로 합산하여 최종 순위를 결정한다.

1.2.3. Modular RAG

- 각 컴포넌트(로더, 청커, 임베딩, 검색기, 생성기)를 독립적으로 교체 가능한 모듈로 설계한다. 도메인별 최적 조합을 실험하기 용이하며, 특정 단계만 교체하거나 추가할 수 있다. 예를 들어 임베딩 모델만 OpenAI에서 Upstage로 교체하거나, 검색기에 Re-ranker 모듈을 추가하는 것이 모듈 교체만으로 가능하다.

1.3 RAG vs Fine-tuning

기준	RAG	Fine-tuning
지식 업데이트	문서 교체만으로 즉시 반영	재학습 필요 (수 시간~수 일)
비용	검색/생성 시 API 비용	학습 시 대규모 GPU 비용
환각 통제	출처 기반 답변으로 통제 용이	학습 데이터에 의존
외부 지식	외부 문서 활용 가능	학습 데이터 내 지식만 사용
도메인 특화	문서 준비만으로 가능	도메인 데이터 수집+학습 필요
모델 크기	기존 모델 그대로 사용	학습된 모델 별도 배포
적합 상황	최신 정보, 출처 추적, 빈번한 업데이트	고유한 스타일/톤, 특수 태스크

- 권장: 대부분의 도메인 지식 활용 시나리오에서는 RAG가 비용-효과적이며, Fine-tuning은 RAG로 해결이 어려운 특수한 출력 형식이나 스타일이 필요할 때 고려한다. 두 기법을 병행(RAG + Fine-tuned 모델)하는 것도 가능하다.

1.4 RAG 파이프라인 전체 흐름

[오프라인: 인덱싱 단계]
[원본 문서] → [문서 로딩] → [청킹(Chunking)] → [임베딩(Embedding)] → [벡터 DB 저장]

[온라인: 검색 및 생성 단계]
[사용자 질의] → [질의 변환(선택)] → [질의 임베딩] → [벡터 DB 유사도 검색] → [문서 청크 반환]
                                                                                    ↓
                                                              [컨텍스트 구성] → [LLM 생성] → [응답]

- 파이프라인은 크게 두 단계로 나뉜다:

1) 인덱싱 단계 (오프라인): 원본 문서를 로딩→청킹→임베딩하여 벡터 DB에 미리 저장해두는 준비 과정이다. 이 과정은 문서가 변경될 때만 수행한다.

2) 검색 및 생성 단계 (온라인): 사용자가 질의를 입력하면, 해당 질의를 임베딩한 뒤 벡터 DB에서 유사도 검색을 수행하여 관련 문서 청크를 가져온다. 검색된 문서들을 컨텍스트로 구성하여 LLM에 전달하면 최종 답변이 생성된다.

- 핵심 포인트: 질의 임베딩 이후 벡터 DB에서 수행하는 것이 "유사도 검색(Similarity Search)"이다. 사용자 질의의 임베딩 벡터와 벡터 DB에 저장된 문서 청크의 임베딩 벡터 간 코사인 유사도(또는 유클리디안 거리 등)를 계산하여 가장 가까운 K개의 문서 청크를 반환한다. 이것이 RAG에서의 문서 검색 핵심 메커니즘이다.

2. RAG 파이프라인 단계별 상세

2.1. 단계 1: 문서 로딩 (Document Loading)

- 원본 데이터를 시스템이 처리할 수 있는 형태로 변환하는 첫 단계이다.

2.1.1. 지원 포맷과 로더

포맷	LangChain 로더	설명
DOCX	Docx2txtLoader	Word 문서 로딩
PDF	PyPDFLoader, PDFMinerLoader	PDF 페이지별 로딩
HTML	WebBaseLoader, BSHTMLLoader	웹 페이지 로딩
CSV	CSVLoader	구조화 데이터 로딩
JSON	JSONLoader	JSON 구조 로딩
Markdown	UnstructuredMarkdownLoader	마크다운 파싱
Text	TextLoader	일반 텍스트
Directory	DirectoryLoader	디렉토리 전체 로딩

2.1.2. 기본 사용 패턴

from langchain_community.document_loaders import Docx2txtLoader

loader = Docx2txtLoader('./documents/tax_law.docx')
documents = loader.load()

# 각 Document 객체의 구조
# Document(page_content="문서 내용...", metadata={"source": "./documents/tax_law.docx"})

2.1.3. 핵심 개념: Document 객체

from langchain_core.documents import Document

doc = Document(
    page_content="실제 텍스트 내용",
    metadata={
        "source": "파일 경로 또는 URL",
        "page": 0,
        "category": "세법",
        # 임의의 메타데이터 추가 가능
    }
)

2.1.4. LangChain 로더 vs Document 직접 생성의 차이

- LangChain이 제공하는 로더(`Docx2txtLoader`, `PyPDFLoader` 등)와 `Document` 객체를 직접 생성하는 방식은 최종적으로 동일한 `Document` 객체를 만들지만, 용도와 적합한 상황이 다르다.

2.1.4.1. LangChain 로더 사용 (자동화된 파싱)

- 로더는 파일 포맷에 맞는 파싱 로직을 내장하고 있어, 파일 경로만 지정하면 자동으로 텍스트를 추출하고 메타데이터(출처, 페이지 번호 등)를 설정한 `Document` 객체를 반환한다.

# 로더가 파싱, 메타데이터 생성을 자동 처리
loader = Docx2txtLoader('./tax_law.docx')
documents = loader.load()
# → [Document(page_content="파싱된 텍스트...", metadata={"source": "./tax_law.docx"})]

- 파일에서 텍스트를 추출하는 로직을 직접 작성할 필요 없음

- 메타데이터(source, page 등)가 자동 생성됨

- 파일 포맷이 명확하고 표준 구조를 따를 때 적합

2.1.4.2. Document 직접 생성 (수동 제어)

- 이미 가공된 텍스트 데이터가 있거나, 로더가 지원하지 않는 커스텀 데이터 소스를 사용할 때 `Document` 객체를 직접 생성한다.

from langchain_core.documents import Document

# DB에서 가져온 데이터를 Document로 변환
records = fetch_from_database()
documents = [
    Document(
        page_content=record["content"],
        metadata={
            "source": "internal_db",
            "doc_id": record["id"],
            "category": record["category"],
            "updated_at": record["updated_at"]
        }
    )
    for record in records
]

# API 응답을 Document로 변환
api_data = call_external_api()
documents = [
    Document(
        page_content=item["text"],
        metadata={"source": "api", "endpoint": "/v1/data", "timestamp": item["ts"]}
    )
    for item in api_data
]

2.1.4.3. Document 직접 생성을 선택하는 기준

상황	설명
로더가 지원하지 않는 데이터 소스	내부 데이터베이스, 커스텀 API, 크롤링 결과 등
전처리가 이미 완료된 텍스트	이미 정제된 텍스트를 직접 Document로 래핑
메타데이터를 세밀하게 제어	로더의 기본 메타데이터 외에 커스텀 필드(카테고리, 버전, 권한 등)가 필요
여러 소스를 통합	서로 다른 소스의 데이터를 하나의 Document 리스트로 병합
로더의 파싱 결과가 불만족	로더의 자동 파싱이 특정 문서 구조에 맞지 않을 때 직접 파싱 후 Document 생성
테스트/프로토타입	간단한 테스트용 데이터를 빠르게 만들 때

- 실무 기준: 파일 기반 데이터는 LangChain 로더를 우선 사용하고, 로더가 지원하지 않거나 이미 가공된 데이터는 `Document`를 직접 생성한다. 두 방식을 혼합하여 사용하는 것도 일반적이다.

2.2. 단계 2: 청킹 (Text Chunking/Splitting)

- 문서를 LLM의 컨텍스트 윈도우와 검색 효율에 맞게 작은 단위로 분할한다. 청킹은 RAG 파이프라인의 품질을 결정하는 핵심 단계로, 청크가 너무 크면 검색 시 관련 없는 내용이 포함되고, 너무 작으면 문맥이 손실된다.

2.2.1. 청킹 전략 비교

전략	클래스	특징	적합한 경우
재귀적 문자 분할	RecursiveCharacterTextSplitter	구분자 우선순위로 분할 (\n\n → \n → → ``)	일반 텍스트 (가장 범용적)
토큰 기반 분할	TokenTextSplitter	토큰 수 기준 분할	정확한 토큰 제어 필요 시
마크다운 헤더 분할	MarkdownHeaderTextSplitter	헤더 계층 기준 분할	마크다운 문서
문자 분할	CharacterTextSplitter	단일 구분자 기준 분할	단순 분할
시맨틱 분할	SemanticChunker	의미 유사도 기반 분할	의미 단위 보존 필요 시
HTML 분할	HTMLHeaderTextSplitter	HTML 태그 기준 분할	웹 문서

2.2.2. 청킹 전략별 상세 설명, 장단점, 사용 조건

2.2.2.1. RecursiveCharacterTextSplitter (재귀적 문자 분할)

- 가장 범용적으로 사용되는 전략이다. 여러 구분자를 우선순위 순서대로 시도하여 가능한 한 자연스러운 단위(문단→줄→단어)로 분할한다.

   - 목적: 문맥을 최대한 보존하면서 일정 크기 이하로 분할
   - 사용 조건: 문서 구조가 특별하지 않은 일반 텍스트, 법률 문서, 기술 문서 등 대부분의 경우
   - 장점: 자연스러운 분할 경계, 높은 범용성, 안정적인 성능
   - 단점: 의미 단위를 보장하지는 않음, 구분자가 없는 텍스트에서는 강제 분할 발생

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1500,       # 각 청크의 최대 문자 수
    chunk_overlap=200,     # 인접 청크 간 중복 문자 수 (문맥 연속성 보장)
    length_function=len,   # 길이 측정 함수 (커스텀 가능, 예: 토큰 수 기준)
    separators=["\n\n", "\n", ". ", " ", ""]  # 분할 우선순위 (문단→줄→문장→단어→문자)
)

chunks = splitter.split_documents(documents)

- 파라미터 정의 기준: `chunk_size`는 임베딩 모델의 최대 토큰 제한과 검색 정밀도를 고려하여 설정한다. `chunk_overlap`은 일반적으로 `chunk_size`의 10~15%로 설정하면 문맥 연속성과 저장 효율의 균형이 맞다.

2.2.2.2. TokenTextSplitter (토큰 기반 분할)

- 문자 수가 아닌 토큰 수를 기준으로 분할한다. LLM의 토큰 제한에 정확히 맞추어야 할 때 사용한다.

- 목적: LLM 토큰 제한에 정확히 맞추는 분할

- 사용 조건: 임베딩 모델이나 LLM의 토큰 제한을 정밀하게 관리해야 할 때, 다국어 텍스트에서 문자 수와 토큰 수 차이가 클 때

- 장점: 토큰 수 정밀 제어, LLM 입력 크기 보장

- 단점: 문맥 경계를 무시할 수 있음, 처리 속도가 약간 느림 (토큰화 과정 필요)

from langchain_text_splitters import TokenTextSplitter

splitter = TokenTextSplitter(
    chunk_size=500,        # 각 청크의 최대 토큰 수
    chunk_overlap=50,      # 인접 청크 간 중복 토큰 수
    encoding_name="cl100k_base"  # 토큰화 인코딩 (OpenAI 모델 기준)
    # model_name="gpt-4"  # 또는 모델명으로 인코딩을 자동 선택
)

chunks = splitter.split_documents(documents)

- 파라미터 정의 기준: `encoding_name`은 사용하는 LLM/임베딩 모델의 토크나이저와 일치시켜야 한다. OpenAI 모델은 `cl100k_base`, 이전 모델은 `p50k_base`를 사용한다.

2.2.2.3. MarkdownHeaderTextSplitter (마크다운 헤더 분할)

- 마크다운 문서의 헤더 계층(#, ##, ### 등)을 기준으로 분할하여, 각 섹션의 구조적 의미를 보존한다.

- 목적: 마크다운 문서의 논리적 구조(섹션)를 보존하면서 분할
    - 사용 조건: 마크다운 형식의 기술 문서, API 문서, 위키 등
    - 장점: 섹션별 의미 단위 보존, 헤더 정보가 메타데이터로 자동 추가
    - 단점: 마크다운 형식이 아닌 문서에는 사용 불가, 헤더 없는 긴 섹션은 추가 분할

from langchain_text_splitters import MarkdownHeaderTextSplitter

headers_to_split_on = [
    ("#", "Header 1"),     # H1 기준 분할
    ("##", "Header 2"),    # H2 기준 분할
    ("###", "Header 3"),   # H3 기준 분할
]

splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on,
    strip_headers=False  # True면 헤더 텍스트를 본문에서 제거하고 메타데이터에만 저장
)

chunks = splitter.split_text(markdown_text)
# 각 청크의 metadata에 {"Header 1": "제목", "Header 2": "소제목"} 형태로 저장

- 파라미터 정의 기준: `headers_to_split_on`에 어떤 레벨의 헤더까지 분할 기준으로 사용할지를 정의한다. 문서의 계층 깊이에 따라 조절한다. 보통 H1~H3까지 사용한다.

2.2.2.4. CharacterTextSplitter (문자 분할)

- 하나의 단일 구분자만으로 텍스트를 분할하는 가장 단순한 전략이다.

- 목적: 명확한 단일 구분자가 있는 데이터를 빠르고 간단하게 분할
    - 사용 조건: CSV, 로그 파일, 줄바꿈으로 구분된 레코드 등 구분자가 일정한 데이터
    - 장점: 구현이 가장 단순, 예측 가능한 분할 결과, 빠른 처리
    - 단점: 자연어 텍스트에는 부적합, 단일 구분자가 없으면 품질 저하

from langchain_text_splitters import CharacterTextSplitter

splitter = CharacterTextSplitter(
    separator="\n\n",     # 단일 구분자 (이 구분자로만 분할)
    chunk_size=1000,
    chunk_overlap=100,
    length_function=len
)

chunks = splitter.split_documents(documents)

2.2.2.5. SemanticChunker (시맨틱 분할)

- 텍스트의 의미적 유사도를 기반으로 분할 지점을 결정한다. 인접 문장들의 임베딩 유사도를 비교하여 의미가 크게 바뀌는 지점에서 분할한다.
- 목적: 의미 단위를 최대한 보존하는 분할

- 사용 조건: 주제가 자주 바뀌는 문서, 의미 단위 보존이 품질에 큰 영향을 미치는 경우

- 장점: 의미적으로 일관된 청크 생성, 문맥 손실 최소화

- 단점: 임베딩 모델 호출이 필요하여 비용/시간 증가, 청크 크기 불균일

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",  # 분할 임계값 유형
    # "percentile": 유사도 차이의 상위 N% 지점에서 분할 (기본값, 가장 안정적)
    # "standard_deviation": 평균에서 N 표준편차 이상 벗어나면 분할
    # "interquartile": 사분위 범위 기반 분할
    breakpoint_threshold_amount=90  # percentile 사용 시: 상위 90% 지점에서 분할
)

chunks = splitter.split_documents(documents)

- 파라미터 정의 기준: `breakpoint_threshold_amount`를 높이면(예: 95) 큰 주제 변화에서만 분할하여 청크가 커지고, 낮추면(예: 70) 작은 변화에서도 분할하여 청크가 작아진다. `percentile` 방식이 가장 안정적이며 실무에서 권장된다.

2.2.2.6. HTMLHeaderTextSplitter (HTML 헤더 분할)

- HTML 문서의 태그 계층(h1, h2, h3 등)을 기준으로 분할한다.

    - 목적: 웹 문서의 구조적 계층을 보존하면서 분할
    - 사용 조건: 웹 크롤링 결과, HTML 문서
    - 장점: HTML 구조 보존, 헤더 정보가 메타데이터에 자동 추가
    - 단점: HTML 형식이 아닌 문서에는 사용 불가

from langchain_text_splitters import HTMLHeaderTextSplitter

headers_to_split_on = [
    ("h1", "Header 1"),
    ("h2", "Header 2"),
    ("h3", "Header 3"),
]

splitter = HTMLHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
chunks = splitter.split_text(html_text)

2.2.3. chunk_size와 chunk_overlap의 관계

chunk_size = 1500, chunk_overlap = 200 인 경우:

청크 1: [===========1500자===========]
청크 2:                    [==200==[===========1500자===========]
청크 3:                                          [==200==[===========1500자===========]

→ 중복 영역(overlap)이 문맥 연속성을 보장

2.2.4. 복합 청킹 전략 (전략 조합)

- 실무에서는 하나의 청킹 전략만 사용하는 것이 아니라, 여러 전략을 조합하여 사용하는 경우가 많다.

2.2.4.1. 구조 기반 분할 + 크기 기반 재분할

- 가장 많이 사용되는 복합 전략이다. 먼저 문서의 구조(마크다운 헤더, HTML 태그)로 논리적 섹션을 나눈 뒤, 각 섹션이 너무 크면 `RecursiveCharacterTextSplitter`로 재분할한다.

from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

# 1단계: 마크다운 헤더로 논리적 섹션 분할
md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[("#", "H1"), ("##", "H2"), ("###", "H3")]
)
md_chunks = md_splitter.split_text(markdown_text)

# 2단계: 큰 섹션을 크기 기반으로 재분할
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=150
)
final_chunks = text_splitter.split_documents(md_chunks)

- 목적: 논리적 구조를 보존하면서도 균일한 크기의 청크를 생성

- 사용 조건: 구조화된 문서(마크다운, HTML)에서 섹션별 크기가 불균일할 때

2.2.4.2. 시맨틱 분할 + 크기 제한

- SemanticChunker로 의미 단위를 나눈 뒤, 크기가 과도한 청크만 추가 분할한다.

from langchain_experimental.text_splitter import SemanticChunker
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 1단계: 의미 단위 분할
semantic_splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=85
)
semantic_chunks = semantic_splitter.split_documents(documents)

# 2단계: 너무 큰 청크만 재분할
size_splitter = RecursiveCharacterTextSplitter(chunk_size=1500, chunk_overlap=200)
final_chunks = []
for chunk in semantic_chunks:
    if len(chunk.page_content) > 1500:
        final_chunks.extend(size_splitter.split_documents([chunk]))
    else:
        final_chunks.append(chunk)

- 목적: 의미 보존을 우선하면서 LLM 토큰 제한도 준수

- 사용 조건: 의미 단위가 중요하지만 과도하게 큰 청크가 발생할 수 있는 경우

2.2.5. 실무에서의 청킹 전략 선택 가이드

문서 유형	권장 전략	이유
일반 텍스트/보고서	RecursiveCharacterTextSplitter	범용성, 안정적 성능
마크다운/기술 문서	MarkdownHeaderTextSplitter + RecursiveCharacterTextSplitter	구조 보존 + 크기 제어
웹 크롤링 HTML	HTMLHeaderTextSplitter + RecursiveCharacterTextSplitter	구조 보존 + 크기 제어
법률/계약서	RecursiveCharacterTextSplitter (chunk_size 1000~2000)	문맥 보존 중요
FAQ/정의집	RecursiveCharacterTextSplitter (chunk_size 200~500)	개별 항목 단위
토큰 정밀 제어	TokenTextSplitter	정확한 토큰 수 관리
주제 변화가 잦은 문서	SemanticChunker	의미 단위 보존
로그/CSV	CharacterTextSplitter (separator=\n)	줄 단위 분할

- 성능 및 효율성 실무 권장: 대부분의 프로덕션 환경에서 `RecursiveCharacterTextSplitter`가 기본 선택이다. chunk_size 500~1000, chunk_overlap 50~150으로 시작한 뒤, 검색 품질 평가(Retrieval Evaluation)를 통해 최적값을 튜닝한다. 구조화된 문서라면 구조 기반 분할과 크기 기반 재분할의 복합 전략이 가장 좋은 성능을 보인다.

2.2.6. CharacterTextSplitter vs RecursiveCharacterTextSplitter 비교

- `CharacterTextSplitter`는 단일 구분자로 분할하고, `RecursiveCharacterTextSplitter`는 여러 구분자를 순차적으로 시도하여 더 자연스러운 청크를 생성한다.

from langchain_text_splitters import CharacterTextSplitter, RecursiveCharacterTextSplitter

sample_text = "첫 번째 문단입니다.\n\n두 번째 문단입니다. 이 문단은 조금 더 깁니다.\n\n세 번째 문단입니다."

# CharacterTextSplitter: 단일 구분자(\n\n)로만 분할
char_splitter = CharacterTextSplitter(
    separator="\n\n",        # 이 구분자로만 분할
    chunk_size=50,
    chunk_overlap=0
)
char_chunks = char_splitter.split_text(sample_text)
print(f"CharacterTextSplitter: {len(char_chunks)}개 청크")
for i, chunk in enumerate(char_chunks):
    print(f"  [{i}] ({len(chunk)}자) {chunk[:50]}")

# RecursiveCharacterTextSplitter: 여러 구분자를 순차적으로 시도
recursive_splitter = RecursiveCharacterTextSplitter(
    separators=["\n\n", "\n", ". ", " ", ""],  # 우선순위 순서
    chunk_size=50,
    chunk_overlap=0
)
recursive_chunks = recursive_splitter.split_text(sample_text)
print(f"\nRecursiveCharacterTextSplitter: {len(recursive_chunks)}개 청크")
for i, chunk in enumerate(recursive_chunks):
    print(f"  [{i}] ({len(chunk)}자) {chunk[:50]}")

- 실무 권장: 대부분의 경우 `RecursiveCharacterTextSplitter`를 사용한다. `CharacterTextSplitter`는 CSV나 로그 파일처럼 명확한 단일 구분자가 있는 데이터에 적합하다.

2.3. 단계 3: 임베딩 (Embedding)

- 텍스트를 고차원 벡터 공간에 매핑하여 의미적 유사도 계산을 가능하게 한다.

2.3.1. 임베딩 모델 비교

모델	제공자	차원	특징
text-embedding-3-large	OpenAI	3072	최고 품질, 다국어, 차원 축소 지원
text-embedding-3-small	OpenAI	1536	비용 최적화, 빠른 속도
text-embedding-ada-002	OpenAI	1536	레거시, 안정적
text-embedding-004	Google (Gemini)	768	다국어 지원, 무료 티어 제공
embedding-001	Google (Gemini)	768	레거시, 안정적
voyage-3	Voyage AI (Anthropic 권장)	1024	Claude 생태계 권장, 높은 품질
voyage-3-lite	Voyage AI	512	경량, 비용 최적화
solar-embedding-1-large	Upstage	4096	한국어 최적화
HuggingFace 모델	오픈소스	다양	무료, 로컬 실행

- 참고: Anthropic(Claude)은 자체 임베딩 API를 제공하지 않으며, 임베딩 용도로 Voyage AI를 공식 파트너로 권장한다. Claude API는 텍스트 생성에 특화되어 있으며, 임베딩은 별도 모델을 사용해야 한다.

2.3.2. 주요 제공자별 유사 수준 모델 비교

수준	OpenAI	Google Gemini	Voyage AI (Claude 권장)
최고 품질	text-embedding-3-large (3072d)	text-embedding-004 (768d)	voyage-3 (1024d)
비용 최적화	text-embedding-3-small (1536d)	embedding-001 (768d)	voyage-3-lite (512d)
레거시	text-embedding-ada-002 (1536d)	-	voyage-2 (1024d)

2.3.3. 임베딩 동작 원리

- 시나리오 1: "세법 쿼리 벡터 재사용" (OpenAI)

from langchain_openai import OpenAIEmbeddings
import pickle

# 세법 담당자가 자주 묻는 질문 벡터 미리 생성
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

# 자주 검색되는 쿼리 벡터 생성 & 저장
frequent_queries = {
    "소득세 환급": embeddings.embed_query("소득세 환급"),
    "법인세 마감": embeddings.embed_query("법인세 마감일")
}

# 저장 (매일 1000회 반복 검색 시 0.1초 단축)
with open("sebeop_queries.pkl", "wb") as f:
    pickle.dump(frequent_queries, f)

print("✅ 자주 묻는 세법 질문 벡터 저장 완료")
# 사용: pickle.load() → 즉시 vectorstore.similarity_search_by_vector()

- 시나리오 2: "빠른 검색" (Gemini)

from langchain_google_genai import GoogleGenerativeAIEmbeddings

emb = GoogleGenerativeAIEmbeddings(model="models/text-embedding-004")

# 실시간 검색어
search = "운동화 러닝화 나이키"
query_vector = emb.embed_query(search)

shoes = ["아디다스 축구화", "나이키 에어맥스", "뉴발란스 러닝화"]
best_shoe = shoes[np.argmax(emb.embed_documents(shoes))]

print(f"'{search}' → {best_shoe}")
# 결과: "나이키 에어맥스"

- 시나리오 3: "직접 유사도 계산" (Voyage AI)

from langchain_voyageai import VoyageAIEmbeddings
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

voyage_emb = VoyageAIEmbeddings(model="voyage-3")

# 세법 문서들
sebeop_texts = [
    "소득세 환급: 다음달 10일",
    "법인세: 3월 31일 마감", 
    "부가세: 분기별 신고"
]

query = "환급 언제"
query_vector = voyage_emb.embed_query(query)
doc_vectors = voyage_emb.embed_documents(sebeop_texts)

# 직접 유사도 계산 (VectorStore 우회)
scores = cosine_similarity([query_vector], doc_vectors)[0]
best_match = np.argmax(scores)

print(f"질문: '{query}'")
print(f"최고 일치 문서: '{sebeop_texts[best_match]}'")
print(f"유사도: {scores[best_match]:.3f}")
# 결과: "소득세 환급: 다음달 10일" (0.912)

- 시나리오 4: "사용자 선호도 캐시" (Voyage)

from langchain_voyageai import VoyageAIEmbeddings
import pickle

emb = VoyageAIEmbeddings(model="voyage-3")

# VIP 고객 선호 상품 벡터 저장
vip_like = "셔츠 청바지 가죽벨트"
vip_vector = emb.embed_query(vip_like)

with open("vip_preference.pkl", "wb") as f:
    pickle.dump(vip_vector, f)

print("VIP 선호도 벡터 저장 완료")
# 나중에: 새 상품과 vip_vector 비교 → 개인화 추천

2.3.3.1. embed_documents를 사용해야 하는 1% 특수 사용 사례

2.3.3.1.1. 대용량 배치 처리 (메모리/비용 최적화)

- 문제: 10만 문서 → from_documents는 메모리 폭발

# ❌ 비효율: 한 번에 모든 문서 처리
# vectorstore = FAISS.from_documents(100000_docs, embeddings)  # OOM!

# ✅ 배치 처리
batch_size = 1000
all_vectors = []

for i in range(0, len(all_texts), batch_size):
    batch = all_texts[i:i+batch_size]
    batch_vectors = embeddings.embed_documents(batch)  # 배치별 임베딩
    all_vectors.extend(batch_vectors)
    print(f"Batch {i//batch_size + 1} 완료")

# 나중에 VectorStore 생성
vectorstore = FAISS.from_embeddings(zip(all_texts, all_vectors), embeddings)
vectorstore.save_local("large_index")

2.3.3.1.2. 멀티벡터 인덱싱 (Multi-Vector)

- 문제: 원문 + 요약 + 키워드 → 각기 다른 벡터 필요

docs = [doc1, doc2]
summaries = ["요약1", "요약2"]  # LLM으로 생성
keywords = [["rag", "llm"], ["검색", "인덱스"]]

# 각 레이어 별도 임베딩
full_vectors = embeddings.embed_documents([d.page_content for d in docs])
summary_vectors = embeddings.embed_documents(summaries)
keyword_vectors = embeddings.embed_documents([" ".join(k) for k in keywords])

# MultiVectorRetriever로 결합 (고급 RAG)
from langchain.retrievers.multi_vector import MultiVectorRetriever

2.3.3.1.3. 캐싱 및 재사용 (변경 감지)

- 문제: 문서 변경 시 전체 재임베딩 비효율

import hashlib, pickle

def smart_embed(texts, cache_file="embed_cache.pkl"):
    cache = {}
    new_vectors = []
    
    try:
        cache = pickle.load(open(cache_file, "rb"))
    except: pass
    
    for text in texts:
        text_hash = hashlib.md5(text.encode()).hexdigest()
        if text_hash in cache:
            new_vectors.append(cache[text_hash])
        else:
            vec = embeddings.embed_documents([text])[0]
            cache[text_hash] = vec
            new_vectors.append(vec)
    
    pickle.dump(cache, open(cache_file, "wb"))
    return new_vectors

2.3.3.1.4. 디버깅 및 벡터 분석

# 벡터 품질 직접 확인
test_docs = ["세법 문서1", "세법 문서2", "무관 문서"]
vectors = embeddings.embed_documents(test_docs)

# 코사인 유사도 직접 계산
from sklearn.metrics.pairwise import cosine_similarity
sim_matrix = cosine_similarity(vectors)
print("세법 문서들 간 유사도:", sim_matrix[0][1])  # 0.85 → 양호

2.3.3.1.5. 커스텀/외부 임베딩 통합

# HuggingFace 로컬 모델 + LangChain 호환
class CustomEmbeddings:
    def embed_documents(self, texts):
        # 외부 API 또는 로컬 모델 호출
        return hf_model.encode(texts).tolist()
    
    def embed_query(self, text):
        return self.embed_documents([text])[0]

embeddings = CustomEmbeddings()
vectors = embeddings.embed_documents(your_texts)  # 직접 사용

2.3.3.1.6. HyDE (Hypothetical Document Embeddings)

- HyDE는 사용자의 짧은 질문을 LLM이 가상의 답변 문서로 확장한 뒤, 그 가상 문서를 임베딩하여 검색하는 기법입니다. 짧은 쿼리보다 가상 문서가 실제 저장된 문서와 임베딩 공간에서 더 가까워지므로 검색 품질이 향상됩니다.

"""
HyDE (Hypothetical Document Embeddings)
- 짧은 쿼리를 LLM으로 가상 답변 문서로 확장
- 가상 문서의 임베딩으로 검색하여 품질 향상
- 일반 쿼리 검색 vs HyDE 검색 비교
"""
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document

# ============================================================
# 1. 설정
# ============================================================
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# ============================================================
# 2. 벡터 DB 준비 (세법 문서)
# ============================================================
docs = [
    Document(page_content="소득세법 제46조에 따르면 근로소득공제는 총급여액에서 일정 금액을 공제한다."),
    Document(page_content="연말정산 결과 기납부세액이 결정세액보다 많으면 차액을 환급받는다."),
    Document(page_content="환급금은 원천징수의무자를 통해 다음달 10일까지 지급된다."),
    Document(page_content="부가가치세 신고는 매 분기 다음달 25일까지 해야 한다."),
]

db = FAISS.from_documents(documents=docs, embedding=embeddings)

# ============================================================
# 3. 일반 쿼리 검색 (비교 기준)
# ============================================================
query = "소득세 환급 기준"

print("=" * 60)
print("[일반 검색] 쿼리를 그대로 임베딩하여 검색")
print("=" * 60)

normal_results = db.similarity_search_with_relevance_scores(query, k=3)
for i, (doc, score) in enumerate(normal_results, 1):
    print(f"  #{i} | Score: {score:.4f} | {doc.page_content[:60]}...")

# ============================================================
# 4. HyDE: LLM으로 가상 답변 문서 생성
#    - 쿼리에 대한 "가상의 정답 문서"를 LLM이 생성
#    - 실제 정확할 필요 없음 (임베딩 유사도만 활용)
# ============================================================
hyde_prompt = ChatPromptTemplate.from_template(
    "다음 질문에 대해 전문적인 답변 문서를 작성하세요. "
    "실제 법령이나 규정을 인용하는 형식으로 작성하세요.\n\n"
    "질문: {question}\n\n"
    "답변 문서:"
)

hyde_chain = hyde_prompt | llm
hyde_response = hyde_chain.invoke({"question": query})
hypothetical_doc = hyde_response.content

print(f"\n{'=' * 60}")
print("[HyDE] LLM이 생성한 가상 문서:")
print("=" * 60)
print(f"  {hypothetical_doc[:200]}...")

# ============================================================
# 5. HyDE 검색: 가상 문서를 임베딩하여 검색
#    - 가상 문서의 벡터가 실제 문서와 더 가까움
# ============================================================
hyde_vector = embeddings.embed_query(hypothetical_doc)

# FAISS에서 벡터로 직접 검색
hyde_results = db.similarity_search_by_vector(hyde_vector, k=3)

print(f"\n{'=' * 60}")
print("[HyDE 검색] 가상 문서 임베딩으로 검색")
print("=" * 60)

# 점수 비교를 위해 수동 계산
hyde_doc_vectors = embeddings.embed_documents([d.page_content for d in docs])
hyde_scores = cosine_similarity([hyde_vector], hyde_doc_vectors)[0]

for i, doc in enumerate(hyde_results):
    # 해당 문서의 점수 찾기
    doc_idx = next(j for j, d in enumerate(docs) if d.page_content == doc.page_content)
    score = hyde_scores[doc_idx]
    print(f"  #{i+1} | Score: {score:.4f} | {doc.page_content[:60]}...")

# ============================================================
# 6. 결과 비교 요약
# ============================================================
normal_vector = embeddings.embed_query(query)
normal_scores = cosine_similarity([normal_vector], hyde_doc_vectors)[0]

print(f"\n{'=' * 60}")
print("[비교] 일반 vs HyDE 유사도 점수")
print("=" * 60)
print(f"{'문서':<40} {'일반':>8} {'HyDE':>8} {'차이':>8}")
print("-" * 60)
for i, doc in enumerate(docs):
    diff = hyde_scores[i] - normal_scores[i]
    marker = "↑" if diff > 0 else "↓"
    print(f"  {doc.page_content[:35]:<38} {normal_scores[i]:>7.4f} {hyde_scores[i]:>7.4f} {marker}{abs(diff):.4f}")

- 동작 원리

사용자 쿼리: "소득세 환급 기준"
        ↓
LLM이 가상 답변 생성: "소득세 환급은 연말정산 결과 기납부세액이 
                      결정세액보다 많은 경우 차액을 돌려받는 것으로..."
        ↓
가상 답변을 임베딩 → 이 벡터로 벡터 DB 검색
        ↓
실제 관련 문서 반환

2.3.3.1.7. 세법 RAG에 적용할 만한 경우

from langchain_openai import OpenAIEmbeddings
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from langchain_voyageai import VoyageAIEmbeddings
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

# ===== 1. 세법 문서 준비 =====
sebeop_query = "소득세 환급 기준"
sebeop_docs = [
    "소득세법 제46조: 근로소득공제 대상 확인",
    "국세청 환급 시기: 다음달 10일까지",
    "무관: 부가가치세 신고"
]

print("=== 세법 임베딩 모델 비교 ===")

# ===== 2. OpenAI (최고 품질) =====
openai_emb = OpenAIEmbeddings(model="text-embedding-3-large")
openai_query = openai_emb.embed_query(sebeop_query)
openai_docs = openai_emb.embed_documents(sebeop_docs)
print(f"OpenAI: 쿼리 {len(openai_query)}차원, 문서 {len(openai_docs)}개")

# ===== 3. Google Gemini (비용 효율) =====
gemini_emb = GoogleGenerativeAIEmbeddings(model="models/text-embedding-004")
gemini_query = gemini_emb.embed_query(sebeop_query)
print(f"Gemini: 쿼리 {len(gemini_query)}차원")

# ===== 4. Voyage AI (한국어/Claude 최적) =====
voyage_emb = VoyageAIEmbeddings(model="voyage-3")
voyage_query = voyage_emb.embed_query(sebeop_query)
print(f"Voyage: 쿼리 {len(voyage_query)}차원")

# ===== 5. 성능 비교 (OpenAI 기준) =====
sim_scores = cosine_similarity([openai_query], openai_docs)[0]
print("\n세법 문서 유사도 (OpenAI):")
for i, score in enumerate(sim_scores):
    print(f"  '{sebeop_docs[i]}': {score:.3f}")

2.3.4. 임베딩의 핵심 속성

- 의미 보존: "소득세"와 "인컴택스"는 벡터 공간에서 가까운 위치
- 차원 축소: 고차원 텍스트를 고정 길이 벡터로 변환

- 코사인 유사도: 두 벡터 간 유사도를 -1~1 범위로 측정

2.3.5. 실무에서 청킹 모델과 임베딩 모델의 선택 기준

- 청킹 전략과 임베딩 모델은 독립적으로 선택할 수 있다. 청킹은 텍스트를 분할하는 전처리 과정이고, 임베딩은 분할된 텍스트를 벡터로 변환하는 과정이므로 서로 다른 기준으로 선택해도 문제없다.

- 단, 다음 사항을 고려해야 한다:

고려 사항	설명
임베딩 모델의 최대 토큰 제한	OpenAI text-embedding-3: 8191 토큰/청크, 300K 토큰/요청 초과 시 400 에러. Gemini: 2048 토큰/입력, 20K 토큰/요청
토큰 기반 분할 시 인코딩 일치	TokenTextSplitter(encoding_name="cl100k_base") → OpenAI와 정확히 일치해야 토큰 수 오차 없음
언어 특화	한국어 RAG → Upstage Solar (세법 문서 최고), multilingual-e5-large, Voyage-3
비용 vs 품질	text-embedding-3-small ($0.00002/1K 토큰) 프로토, text-embedding-3-large ($0.00013/1K 토큰) 프로덕션
차원 수와 저장 비용	3072d(OpenAI-large): 품질↑ 저장비↑, 768d(Gemini): 비용↓ 속도↑

- 실무 권장: 먼저 `RecursiveCharacterTextSplitter`(chunk_size=1000) + `text-embedding-3-small`로 프로토타입을 구성한 뒤, 검색 품질을 평가하고 임베딩 모델이나 chunk_size를 조절한다. 한국어 특화가 필요하면 Upstage Solar를 고려한다.

2.3.6. 전체 인덱스 재구축이 필요한 경우

- 벡터 DB에 저장된 인덱스를 전체적으로 다시 구축해야 하는 상황이 있다:

상황	이유
임베딩 모델 변경	다른 벡터 공간 사용 → 기존 벡터와 비교 불가. OpenAI→Upstage Solar 전환 시 100% 재인덱싱 필수
chunk_size 또는 청킹 전략 변경	문맥 경계 불일치 → 검색 품질 저하. chunk_size 1000→2000 시 재청킹+재임베딩
원본 문서의 대규모 변경	20% 이상 문서 변경 시 증분 업데이트보다 전체 재인덱싱이 정확. 세법 개정 시 해당
벡터 DB 마이그레이션	Chroma→Pinecone/PGVector 시 데이터 형식/인덱스 구조 변경
메타데이터 스키마 변경	필터링 조건 변경(year, law_type 추가) 시 기존 메타데이터 불일치

- 전체 인덱스 재구축 과정

"""
Chroma 전체 인덱스 재구축
- 임베딩 모델 변경, chunk 설정 변경 등으로 재구축이 필요한 경우
- 기존 데이터 백업 → 삭제 → 새 설정으로 재생성
"""
import shutil
from pathlib import Path
import chromadb
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import Docx2txtLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings

CHROMA_DIR = "./chroma_db"
COLLECTION_NAME = "tax_collection"

# ============================================================
# 1. 기존 컬렉션 백업 (선택)
#    - dirs_exist_ok=True로 반복 실행 시 덮어쓰기 허용
# ============================================================
backup_path = Path("./chroma_db_backup")
if Path(CHROMA_DIR).exists():
    shutil.copytree(CHROMA_DIR, backup_path, dirs_exist_ok=True)
    print(f"[백업 완료] {backup_path}")

# ============================================================
# 2. 기존 컬렉션 삭제
#    - chromadb 클라이언트로 직접 삭제
#    - 컬렉션이 없으면 에러 발생하므로 예외 처리
# ============================================================
client = chromadb.PersistentClient(path=CHROMA_DIR)
try:
    client.delete_collection(COLLECTION_NAME)
    print(f"[삭제 완료] {COLLECTION_NAME}")
except ValueError:
    print(f"[삭제 스킵] {COLLECTION_NAME} 컬렉션이 존재하지 않음")

# ============================================================
# 3. 원본 문서를 새 설정으로 재처리
#    - chunk_size, chunk_overlap 등을 변경하여 재분할
# ============================================================
loader = Docx2txtLoader("./tax_doc.docx")
documents = loader.load()

new_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,    # 변경된 chunk_size
    chunk_overlap=150
)
new_chunks = new_splitter.split_documents(documents)

print(f"[문서 재처리] {len(documents)}건 → {len(new_chunks)}건")

# ============================================================
# 4. 새 임베딩 모델로 인덱스 재생성
#    - 모델 변경 시 차원이 달라지므로 반드시 전체 재구축 필요
#    - text-embedding-3-small: 1536차원
#    - text-embedding-3-large: 3072차원
#    - 이후 로드 시에도 동일한 모델을 사용해야 함
# ============================================================
new_embedding = OpenAIEmbeddings(model="text-embedding-3-large")

db = Chroma.from_documents(
    documents=new_chunks,
    embedding=new_embedding,
    persist_directory=CHROMA_DIR,
    collection_name=COLLECTION_NAME
)

print(f"[재구축 완료] {COLLECTION_NAME}")

# ============================================================
# 5. 검색 테스트로 품질 확인
# ============================================================
test_results = db.similarity_search("테스트 질의", k=3)
print(f"\n[품질 확인] {len(test_results)}건 검색됨")
for i, doc in enumerate(test_results, 1):
    print(f"  #{i}: {doc.page_content[:100]}...")

2.4. 단계 4: 벡터 데이터베이스 (Vector Database)

- 임베딩된 벡터를 저장하고 유사도 검색을 수행하는 저장소이다.

2.4.1. 벡터 DB 비교

데이터베이스	유형	장점	단점	적합 환경
Chroma	로컬/오픈소스	초간단 설정, LangChain 완벽 통합, 무료	대규모(>10M) 제한, 분산 미지원	개발/프로토타입, 소규모 RAG
Pinecone	관리형 클라우드	무한 확장, 실시간 업데이트, 서버리스	유료($0.1/100K 벡터), 벤더 락인	프로덕션 대규모, 스타트업
FAISS	로컬 라이브러리	GPU 가속 초고속, Facebook 품질	메타데이터 약함, 로컬 전용	대용량 로컬, 연구/테스트
Weaviate	하이브리드/오픈소스	GraphQL+하이브리드 검색, 모듈화	설정 복잡, 자원 소모	지식그래프, 복합 검색
Qdrant	하이브리드/오픈소스	Rust 초고속, 필터링 최강, Docker 쉬움	학습곡선 있음	고급 필터링, 중형 프로덕션
Milvus	클라우드/엔터프라이즈	10억 벡터+, 샤딩/HA 내장	운영 복잡, 설정 어려움	대기업, 미션크리티컬

2.4.2. Chroma 사용 패턴

from langchain_community.vectorstores import Chroma

# 최초 생성 및 저장
db = Chroma.from_documents(
    documents=chunks,
    embedding=embedding,
    persist_directory="./chroma_db",
    collection_name="tax_collection"
)

# 기존 컬렉션 로딩
db = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embedding,
    collection_name="tax_collection"
)

2.4.3. Pinecone 사용 패턴

"""
Pinecone 벡터스토어 사용 패턴 요약
- 클라우드 관리형 벡터 DB (별도 서버 불필요)
- 자동 영속화 (save 호출 불필요)
- 대규모 벡터 검색에 적합
"""
import os
from pinecone import Pinecone, ServerlessSpec
from langchain_pinecone import PineconeVectorStore
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document

# ============================================================
# 1. 설정
# ============================================================
os.environ["PINECONE_API_KEY"] = "your-pinecone-api-key"
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

INDEX_NAME = "tax-index"
embedding = OpenAIEmbeddings(model="text-embedding-3-small")

# ============================================================
# 2. 인덱스 생성 (최초 1회)
#    - dimension은 임베딩 모델 차원과 반드시 일치해야 함
# ============================================================
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])

if INDEX_NAME not in pc.list_indexes().names():
    pc.create_index(
        name=INDEX_NAME,
        dimension=1536,
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )

# ============================================================
# 3. 문서 업로드 (최초 생성)
# ============================================================
docs = [
    Document(page_content="근로소득세는 급여에 부과되는 세금이다.", metadata={"source": "tax.pdf"}),
    Document(page_content="종합소득세는 모든 소득을 합산하여 과세한다.", metadata={"source": "tax.pdf"}),
]

db = PineconeVectorStore.from_documents(
    documents=docs,
    embedding=embedding,
    index_name=INDEX_NAME
)

# ============================================================
# 4. 기존 인덱스 연결 (이후 재접속 시)
# ============================================================
db = PineconeVectorStore.from_existing_index(
    index_name=INDEX_NAME,
    embedding=embedding
)

# ============================================================
# 5. 검색
# ============================================================
results = db.similarity_search("근로소득세 계산 방법은?", k=3)
for doc in results:
    print(doc.page_content[:80])

# ============================================================
# 6. 증분 추가 (자동 저장, save 불필요)
# ============================================================
db.add_documents([
    Document(page_content="2024년 세법 개정 내용", metadata={"year": 2024})
])

# ============================================================
# 7. Retriever 변환 (RAG 파이프라인 연결용)
# ============================================================
retriever = db.as_retriever(search_type="similarity", search_kwargs={"k": 4})
docs = retriever.invoke("2024년 세법 개정 내용은?")

- FAISS와의 핵심 차이는 인덱스 사전 생성이 필요하다는 점과, 클라우드 자동 영속화로 save_local() 호출이 불필요하다는 점이다.

2.4.4. FAISS (로컬 고성능 벡터 검색)

- FAISS(Facebook AI Similarity Search)는 대규모 벡터에 대해 빠른 유사도 검색을 수행하는 로컬 벡터 스토어다. 서버 설치 없이 인메모리로 동작하며, 디스크 저장/로드를 지원한다.

"""
FAISS 벡터스토어 사용 패턴 요약
- 로컬 인메모리 고성능 벡터 검색 (서버 불필요)
- save_local() / load_local()로 디스크 저장/로드
- 증분 추가 후 반드시 save_local() 필요 (자동 저장 안됨)
"""
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document

# ============================================================
# 1. 설정
# ============================================================
FAISS_INDEX_PATH = "./faiss_tax_index"
embedding = OpenAIEmbeddings(model="text-embedding-3-small")

# ============================================================
# 2. 인덱스 생성 + 문서 업로드
# ============================================================
docs = [
    Document(page_content="근로소득세는 급여에 부과되는 세금이다.", metadata={"source": "tax.pdf"}),
    Document(page_content="종합소득세는 모든 소득을 합산하여 과세한다.", metadata={"source": "tax.pdf"}),
]

faiss_db = FAISS.from_documents(documents=docs, embedding=embedding)

# ============================================================
# 3. 디스크 저장 (index.faiss + index.pkl 생성)
# ============================================================
faiss_db.save_local(FAISS_INDEX_PATH)

# ============================================================
# 4. 저장된 인덱스 로드 (동일한 임베딩 모델 필수)
# ============================================================
loaded_db = FAISS.load_local(
    FAISS_INDEX_PATH,
    embedding,
    allow_dangerous_deserialization=True  # pickle 역직렬화 허용 (신뢰할 수 있는 파일만)
)

# ============================================================
# 5. 검색
# ============================================================
results = loaded_db.similarity_search("근로소득세 계산 방법은?", k=3)
for doc in results:
    print(doc.page_content[:80])

# ============================================================
# 6. 증분 추가 → 반드시 save_local() (자동 저장 안됨)
# ============================================================
loaded_db.add_documents([
    Document(page_content="2024년 세법 개정 내용", metadata={"year": 2024})
])
loaded_db.save_local(FAISS_INDEX_PATH)  # 이거 빠지면 추가분 유실

# ============================================================
# 7. Retriever 변환 (RAG 파이프라인 연결용)
# ============================================================
retriever = loaded_db.as_retriever(search_type="similarity", search_kwargs={"k": 4})
docs = retriever.invoke("2024년 세법 개정 내용은?")

단계	DB	이유
개발	Chroma	personal_law 컬렉션, 로컬 persist ./chroma_db 최고
테스트	FAISS	세법 문서 대량 테스트 시 속도 5배
운영	Pinecone	고객 1000명 동시 세법 검색

2.5 단계 5: 검색 (Retrieval)

- 사용자 질의와 가장 유사한 문서 청크를 벡터 DB에서 검색한다. 검색(Retrieval)은 RAG 파이프라인에서 답변 품질에 가장 직접적인 영향을 미치는 단계이다. 관련 문서를 정확히 찾아야 LLM이 올바른 답변을 생성할 수 있다.

2.5.1. 검색 방식 비교

방식	설명	장점	단점
유사도 검색	코사인/내적 기반 Top-K 반환	간단 빠름, 의미 검색	중복 결과, 키워드 약함
MMR	유사도 + 다양성 균형 (lambda_mult=0.5)	중복 제거, 다양한 세법 관점	약간 느림
유사도 임계값	score_threshold=0.8 이상만 반환	쓰레기 결과 제거	결과 0개 위험
BM25	키워드(TF-IDF) 매칭	"소득세법 제46조" 정확 검색	의미 이해 없음
하이브리드	벡터 + BM25 가중 평균	키워드+의미 모두	설정 복잡
Multi-Query	LLM으로 쿼리 변형 3~5개 생성	검색 누락↓ (환급/공제 동시)	LLM 비용 3배
Contextual Compression	검색 후 LLM으로 압축	정밀 추출, 토큰 절약	LLM 호출 추가
Parent Document	작은 청크 검색 → 원본 큰 문서 반환	완전 문맥, 세법 전체 보기	저장소 2배
Self-Query	쿼리에서 메타데이터 자동 추출 (year=2026)	세법 연도/유형 필터링	LLM 필요
Ensemble	MMR + BM25 + SelfQuery 결합	최고 품질	복잡도 최상

2.5.2. 검색 방식별 상세 설명 및 코드

2.5.2.1. 유사도 검색 (Similarity Search)

- 가장 기본적인 검색 방식으로, 질의 벡터와 문서 벡터 간 코사인 유사도가 높은 순으로 K개를 반환한다.

retriever = db.as_retriever(
    search_type="similarity",
    search_kwargs={
        "k": 4  # 반환할 문서 수
    }
)

# 검색 실행
docs = retriever.invoke("근로소득세 계산 방법은?")

- k 값 설정 기준: 단순 Q&A는 3~4, 복합 분석은 5~10. 너무 크면 노이즈 증가, 너무 작으면 정보 부족

- 실무 권장 설정: `k=4` (일반적인 Q&A에 가장 균형 잡힌 값)

2.5.2.2. MMR (Maximal Marginal Relevance)

- 유사도가 높으면서도 서로 다양한(중복이 적은) 문서를 선택한다. 먼저 `fetch_k`개를 유사도순으로 가져온 뒤, 그 중에서 `lambda_mult` 파라미터로 관련성과 다양성의 균형을 맞춰 `k`개를 선택한다.

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# ============================================================
# 1. 벡터스토어 초기화
# ============================================================
embedding = OpenAIEmbeddings()

db = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embedding
)

# ============================================================
# 2. lambda_mult 값에 따른 결과 비교 (디버깅/튜닝용)
#    - 운영 시에는 제거하거나 로깅으로 전환
#    - 0.0: 최대 다양성 / 0.5: 균형 / 1.0: 순수 유사도
# ============================================================
test_query = "테스트 쿼리를 입력하세요"

print("=" * 70)
print("[lambda_mult 값에 따른 결과 비교]")
print("=" * 70)

for lam in [0.3, 0.5, 0.7, 1.0]:
    docs = db.max_marginal_relevance_search(
        query=test_query,
        k=4,
        fetch_k=20,
        lambda_mult=lam
    )
    print(f"\n[lambda_mult={lam}]")
    for i, doc in enumerate(docs, 1):
        print(f"  #{i}: {doc.page_content[:80]}...")

print("=" * 70)

# ============================================================
# 3. Retriever 생성 (MMR 방식)
#    - search_type="mmr":
#      관련성(relevance)과 다양성(diversity)의 균형을 맞추는 검색
#    - fetch_k: 1차로 유사도 기반 후보 문서를 가져오는 수
#      → k의 3~10배가 권장 (너무 크면 성능 저하)
#    - k: fetch_k 후보 중 MMR 알고리즘으로 최종 선택하는 문서 수
#      → 반드시 fetch_k > k 이어야 다양성 선택이 의미 있음
#    - lambda_mult: 유사도와 다양성의 비율 조절 (0~1)
#      → 위 비교 결과를 기반으로 적절한 값 설정
# ============================================================
retriever = db.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 4,             # 최종 반환 문서 수
        "fetch_k": 20,      # 1차 후보 문서 수 (k의 5배)
        "lambda_mult": 0.5  # 유사도와 다양성 균형
    }
)

# ============================================================
# 4. Retriever 실행 및 결과 처리
#    - MMR은 fetch_k 후보에서 k개를 반드시 선택하므로
#      DB에 문서가 충분하면 빈 결과가 나오지 않음
#    - 단, 유사도가 낮은 문서도 다양성 때문에 포함될 수 있으므로
#      필요 시 후처리 품질 필터링 적용
# ============================================================
query = "실제 사용자 질문을 입력하세요"
retrieved_docs = retriever.invoke(query)

print(f"\n[MMR 검색 결과] {len(retrieved_docs)}건\n")

for i, doc in enumerate(retrieved_docs, 1):
    print(f"--- 문서 #{i} ---")
    print(f"내용: {doc.page_content[:200]}...")
    if doc.metadata:
        print(f"출처: {doc.metadata}")
    print()

# ============================================================
# 5. (선택) 품질 하한선 후처리
#    - MMR 자체에는 score_threshold 기능이 없으므로
#      relevance score를 별도로 조회하여 낮은 품질 문서 제거
#    - 엄격한 품질 관리가 필요한 경우에만 적용
# ============================================================
SCORE_THRESHOLD = 0.7

# MMR 결과 문서들의 실제 유사도 점수 확인
docs_with_scores = db.similarity_search_with_relevance_scores(
    query=query,
    k=20  # MMR 결과 문서들이 포함될 만큼 충분히 설정
)

# {문서내용: 점수} 매핑 생성
score_map = {doc.page_content: score for doc, score in docs_with_scores}

# MMR 결과에서 threshold 미달 문서 필터링
filtered_docs = []
removed_count = 0

for doc in retrieved_docs:
    score = score_map.get(doc.page_content, 0.0)
    if score >= SCORE_THRESHOLD:
        filtered_docs.append(doc)
    else:
        removed_count += 1
        print(f"[필터링 제거] Score: {score:.4f} | {doc.page_content[:60]}...")

if removed_count > 0:
    print(f"\n→ {removed_count}건의 저품질 문서가 제거되었습니다.")

print(f"\n[최종 결과] {len(filtered_docs)}건\n")
for i, doc in enumerate(filtered_docs, 1):
    print(f"--- 최종 문서 #{i} ---")
    print(f"내용: {doc.page_content[:200]}...")
    print()

- lambda_mult 튜닝 → MMR retriever 생성 → 결과 반환 → 품질 후처리 필터링 4단계입니다.

- 5단계의 품질 하한선 후처리는 MMR의 약점(낮은 유사도 문서 포함 가능성)을 보완하는 선택적 단계로, 엄격한 품질 관리가 필요한 경우에만 적용하시면 됩니다.
- `fetch_k` 설정 기준: k의 3~5배. 후보 풀이 충분해야 다양성 확보 가능
- `lambda_mult` 설정 기준: 0.5가 기본 균형. 유사한 문서가 많이 검색되는 환경(법률, 규정 등)에서는 0.3~0.4로 낮추어 다양성을 높임

- 실무 권장 설정: `k=4, fetch_k=20, lambda_mult=0.5`

2.5.2.3. 유사도 점수 임계값 (Similarity Score Threshold)

- 최소 유사도 점수를 넘는 문서만 반환한다. 관련 없는 문서가 포함되는 것을 방지한다.

from langchain_community.vectorstores import Chroma  # 또는 FAISS 등 사용 중인 벡터스토어
from langchain_openai import OpenAIEmbeddings

# ============================================================
# 1. 벡터스토어 초기화 (이미 생성된 DB를 로드하는 경우)
# ============================================================
embedding = OpenAIEmbeddings()

db = Chroma(
    persist_directory="./chroma_db",  # 저장된 벡터스토어 경로
    embedding_function=embedding
)

# ============================================================
# 2. 점수 분포 사전 확인 (threshold 튜닝을 위한 디버깅용)
#    - 실제 운영 시에는 제거하거나 로깅으로 전환
#    - 이 단계를 통해 적절한 score_threshold 값을 결정
# ============================================================
test_query = "테스트 쿼리를 입력하세요"

docs_with_scores = db.similarity_search_with_relevance_scores(
    query=test_query,
    k=10  # 충분히 넉넉하게 조회하여 점수 분포 파악
)

print("=" * 60)
print("[점수 분포 확인]")
print("=" * 60)
for i, (doc, score) in enumerate(docs_with_scores, 1):
    print(f"  #{i} | Score: {score:.4f} | {doc.page_content[:80]}...")
print("=" * 60)

# ============================================================
# 3. Retriever 생성
#    - search_type="similarity_score_threshold":
#      유사도 점수가 threshold 이상인 문서만 반환
#    - score_threshold: 최소 유사도 점수 (0~1)
#      → 위 점수 분포 확인 결과를 기반으로 조정
#    - k: 내부적으로 먼저 k개 후보를 검색한 뒤 threshold 필터링
#      → threshold를 넘는 문서가 k개보다 많아도 k개까지만 반환
# ============================================================
retriever = db.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={
        "score_threshold": 0.7,  # 최소 유사도 점수 (점수 분포 확인 후 조정)
        "k": 4                   # 최대 반환 문서 수
    }
)

# ============================================================
# 4. Retriever 실행 및 결과 처리
#    - threshold 미달 시 빈 리스트 반환 → fallback 처리 필수
# ============================================================
query = "실제 사용자 질문을 입력하세요"
retrieved_docs = retriever.invoke(query)

if retrieved_docs:
    # 정상적으로 관련 문서를 찾은 경우
    print(f"\n[검색 결과] {len(retrieved_docs)}건의 관련 문서를 찾았습니다.\n")
    for i, doc in enumerate(retrieved_docs, 1):
        print(f"--- 문서 #{i} ---")
        print(f"내용: {doc.page_content[:200]}...")
        # metadata가 있는 경우 출처 정보 표시
        if doc.metadata:
            print(f"출처: {doc.metadata}")
        print()
else:
    # threshold를 넘는 문서가 없는 경우 → fallback 로직
    print("\n[경고] 관련 문서를 찾지 못했습니다.")
    print("→ score_threshold를 낮추거나, 질문을 다시 작성해보세요.")

    # fallback 예시: threshold 없이 top-k로 재검색
    fallback_retriever = db.as_retriever(
        search_type="similarity",  # 단순 유사도 top-k
        search_kwargs={"k": 2}
    )
    fallback_docs = fallback_retriever.invoke(query)

    if fallback_docs:
        print(f"\n[Fallback] 유사도 상위 {len(fallback_docs)}건을 대신 반환합니다.\n")
        for i, doc in enumerate(fallback_docs, 1):
            print(f"--- Fallback 문서 #{i} ---")
            print(f"내용: {doc.page_content[:200]}...")
            print()

- 처리 흐름 (4단계)

- 1단계 — 벡터스토어 초기화

- 사전에 임베딩되어 저장된 Chroma DB를 로드합니다. 이 DB에는 문서들이 벡터(숫자 배열)로 변환되어 저장되어 있으며, 질문이 들어오면 질문도 같은 임베딩 모델로 벡터화하여 유사도를 비교합니다.

- 2단계 — 점수 분포 사전 확인 (디버깅)

- similarity_search_with_relevance_scores()로 테스트 쿼리의 점수 분포를 확인합니다. 이 단계가 중요한 이유는 임베딩 모델과 데이터 특성에 따라 점수 분포가 크게 달라지기 때문입니다. 예를 들어 OpenAI 임베딩은 대부분 0.50.85 사이에 분포하는 반면, 다른 모델은 0.20.6에 몰릴 수 있습니다. 이 확인 없이 threshold를 0.7로 고정하면 결과가 0건이 되는 상황이 발생할 수 있습니다.

- 3단계 — Retriever 생성

- similarity_score_threshold 모드로 retriever를 생성합니다. 내부적으로 k개 후보를 먼저 검색한 뒤 threshold 필터링을 적용하므로, threshold를 넘는 문서만 최종 반환됩니다.

- 4단계 — 결과 처리 + Fallback

- 검색 결과가 있으면 정상 처리하고, 없으면 threshold 없는 단순 top-k 검색으로 fallback하여 사용자에게 최소한의 응답을 보장합니다.

- `score_threshold` 설정 기준: 0.7~0.8이 일반적. 높이면 정밀도 상승하지만 결과가 적어질 수 있음. 도메인과 임베딩 모델에 따라 튜닝 필요

- 실무 권장 설정: `score_threshold=0.7` (시작점), 품질 평가 후 0.65~0.85 범위에서 조정

2.5.2.4. BM25 키워드 검색

- TF-IDF 기반의 전통적 키워드 매칭 검색이다. 정확한 키워드(고유명사, 법률 조항 번호 등)가 포함된 문서를 찾는 데 유리하다.

from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document

# ── 문서셋 ──
chunks = [
    Document(page_content="근로소득세는 소득세법 제46조에 따라 계산됩니다."),
    Document(page_content="제127조의2는 법인세 관련 규정입니다."),
    Document(page_content="소득세 환급은 다음달 10일입니다."),
    Document(page_content="근로소득공제 기준은 연 5천만원입니다."),
]

# ── BM25 검색기 생성 ──
bm25_retriever = BM25Retriever.from_documents(
    documents=chunks,
    k=4,
)

# ── 검색 실행 ──
docs = bm25_retriever.invoke("제127조의2 소득세법")

for i, doc in enumerate(docs, 1):
    print(f"[{i}] {doc.page_content}")

- 실무 권장 설정: `k=4`. BM25는 키워드 매칭이므로 도메인 용어가 정확한 질의에 강함
- 사용 시기: 법률 조항 번호, 제품명, 고유명사 등 정확한 키워드 매칭이 중요한 경우

2.5.2.5. 하이브리드 검색 (Ensemble Retriever)

- 벡터 검색(의미 기반)과 BM25 검색(키워드 기반)을 결합하여 두 방식의 장점을 모두 활용한다. 내부적으로 Reciprocal Rank Fusion(RRF)을 사용하여 두 검색기의 결과를 합산한다.

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# ── 벡터스토어 설정 ──
db = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(),
)

# ── 테스트 문서셋 ──
chunks = [
    Document(page_content="근로소득세는 소득세법 제46조에 따라 계산됩니다."),
    Document(page_content="제127조의2는 법인세 관련 규정입니다."),
    Document(page_content="소득세 환급은 다음달 10일입니다."),
    Document(page_content="근로소득공제 기준은 연 5천만원입니다."),
]

# ── 벡터 의미 검색기 ──
vector_retriever = db.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4},
)

# ── BM25 키워드 검색기 ──
bm25_retriever = BM25Retriever.from_documents(chunks, k=4)

# ── 하이브리드 앙상블 검색기 (RRF 융합) ──
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.7, 0.3],
)

# ── 실행 ──
query = "제127조의2 소득세 계산 방법"
docs = ensemble_retriever.invoke(query)

print(f"하이브리드 검색 결과 ({len(docs)}개):")
for i, doc in enumerate(docs, 1):
    print(f"[{i}] {doc.page_content[:60]}...")

Query: "제127조의2 소득세 계산 방법"

VectorRetriever 결과 (의미 기반):
1. "근로소득세는 소득세법 제46조..." (소득세 관련)
2. "근로소득공제 기준..." (계산 관련)

BM25Retriever 결과 (키워드 기반):  
1. "제127조의2는 법인세..." (정확 키워드 매칭)

앙상블 최종 (RRF 융합):
1. 제127조의2 (BM25 강점)
2. 소득세 제46조 (Vector 강점) 
→ **의미 + 키워드 완벽 보완**

- `weights` 설정 기준: [0.5, 0.5]가 기본. 의미 검색이 더 중요하면 [0.7, 0.3], 키워드 매칭이 중요하면 [0.3, 0.7]로 조정

- 실무 권장 설정: `weights=[0.5, 0.5]` (시작점). 법률/기술 문서처럼 정확한 용어가 중요한 도메인에서는 BM25 가중치를 높임

2.5.2.6. Multi-Query Retriever

- 하나의 질의를 LLM을 사용하여 여러 관점의 변형 질의로 확장한 뒤, 각 질의에 대해 검색을 수행하고 결과를 합산한다. 원본 질의가 모호하거나 다양한 관점이 필요할 때 검색 재현율(recall)을 크게 높인다.

- 실행 과정 (invoke() 호출 시):

- 원본 질의에 대해 LLM이 3개의 변형 질의를 생성합니다.

- 각 변형 질의마다 base_retriever를 병렬 호출하여 문서를 검색합니다.

- 검색된 모든 문서에서 중복을 제거하고 Reciprocal Rank Fusion(RRF) 알고리즘으로 재정렬합니다.

- 최종적으로 고유한 상위 문서들을 반환합니다.

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma

# ── LLM 설정 ──
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 벡터스토어 설정 ──
db = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(),
)

# ── MultiQueryRetriever 생성 ──
multi_retriever = MultiQueryRetriever.from_llm(
    retriever=db.as_retriever(search_kwargs={"k": 4}),
    llm=llm,
)

# ── 실행 ──
query = "소득세와 법인세의 차이는?"
docs = multi_retriever.invoke(query)

print(f"검색된 문서: {len(docs)}개")
for i, doc in enumerate(docs, 1):
    print(f"[{i}] {doc.page_content[:60]}...")

입력: "소득세와 법인세의 차이는?"

1단계: LLM 쿼리 변형 (자동)
LLM 프롬프트 → 3~5개 변형 생성:
- "소득세 법인세 차이점"
- "소득세 vs 법인세 구분" 
- "개인 소득세 법인세 비교"

2단계: 각 변형별 검색 (k=4)
Query1 → 4개 문서
Query2 → 4개 문서  
Query3 → 4개 문서

3단계: 중복 제거 + Reciprocal Rank Fusion (RRF)
최종 → **고유 Top 문서** 반환 (보통 6~10개)

INFO: Generated queries:
1. "소득세와 법인세의 주요 차이점은?"
2. "개인 소득세 법인세 구분 기준" 
3. "소득세 법인세 적용 대상 비교"

하이브리드 검색 결과 (8개):
✅ 소득세법 제1조: 개인 소득 과세...
✅ 법인세법: 법인 과세 대상...
✅ 근로소득 vs 배당소득 구분...

- 실무 권장 설정: 기본 설정 사용. LLM이 자동으로 3개의 변형 질의를 생성

- 사용 시기: 질의가 모호하거나 복합적인 경우, 검색 재현율을 높이고 싶은 경우

2.5.2.7. Contextual Compression Retriever

- 1차 검색으로 가져온 문서에서 질의와 관련된 핵심 부분만 추출(압축)하여 반환한다. 불필요한 내용을 제거하여 LLM에 전달되는 컨텍스트의 정밀도를 높인다.

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

# ── LLM 설정 ──
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 벡터스토어 설정 ──
db = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(),
)

# ── 압축기: LLM이 각 문서에서 질의 관련 부분만 추출 ──
compressor = LLMChainExtractor.from_llm(llm)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=db.as_retriever(search_kwargs={"k": 6}),
)

# ── 실행 ──
docs = compression_retriever.invoke("근로소득세 공제 항목은?")

for i, doc in enumerate(docs, 1):
    print(f"[{i}] {doc.page_content[:150]}...")

1. db.as_retriever(k=6) → 6개 후보 문서 검색
2. LLMChainExtractor 각 문서에 적용:
   프롬프트: "다음 문서에서 '근로소득세 공제 항목' 관련 부분만 추출"
3. 무관 문서 → 빈 문자열 (삭제)
4. 관련 문서 → 핵심 문장만 추출
5. 최종 2~4개 압축 문서 반환

- 실무 권장 설정: base_retriever의 k를 평소보다 높게(6~10) 설정하여 넓게 검색한 뒤 압축

- 사용 시기: 청크가 크고 관련 없는 내용이 많이 포함될 때, LLM 컨텍스트를 효율적으로 활용하고 싶을 때

2.5.2.8. Parent Document Retriever

- 작은 청크로 정밀하게 검색하되, 실제로는 더 큰 부모 청크(또는 전체 문서)를 반환하여 풍부한 문맥을 제공한다. 검색 정밀도와 문맥 풍부함을 동시에 확보하는 전략이다.

from langchain.retrievers import ParentDocumentRetriever
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.storage import InMemoryStore
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

# ── 임베딩 & 벡터스토어 ──
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
db = Chroma(
    collection_name="parent_child_law",
    embedding_function=embeddings,
    persist_directory="./chroma_db",
)

# ── 테스트 세법 문서 ──
documents = [
    Document(
        page_content=(
            "소득세법 제46조 근로소득세\n"
            "근로소득세는 총급여액에서 근로소득공제 등을 차감하여 계산됩니다.\n"
            "공제 기준: 연 5천만원 한도, 70세 이상은 별도 공제 적용됩니다.\n"
            "제127조의2는 법인세 관련 규정입니다.\n"
        ) * 5,  # 충분한 길이로 parent/child 분할 테스트
        metadata={"source": "소득세법", "year": 2024},
    ),
]

# ── 분할기 ──
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=50)

# ── 부모 문서 저장소 (InMemory) ──
store = InMemoryStore()

# ── ParentDocumentRetriever ──
parent_retriever = ParentDocumentRetriever(
    vectorstore=db,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

# ── 문서 인덱싱 (add_documents가 자동으로 parent/child 분할 처리) ──
parent_retriever.add_documents(documents)

# ── 검색 ──
docs = parent_retriever.invoke("근로소득세 계산 방법")

print(f"Parent Document 검색 결과 ({len(docs)}개):")
for i, doc in enumerate(docs, 1):
    print(f"\n[{i}] {len(doc.page_content)}자")
    print(f"    내용: {doc.page_content[:120]}...")
    print(f"    메타: {doc.metadata}")

1. add_documents() 인덱싱:
   원본 문서 → parent_splitter(2000자) → 부모 청크들
   각 부모 → child_splitter(400자) → 자식 청크들
   
   자식들 → Chroma DB 임베딩 저장 (검색용)
   부모들 → docstore에 ID별 저장 (문맥용)

2. invoke() 검색:
   query → Chroma에서 자식 청크 Top-K 검색
   각 자식의 parent_id → docstore에서 부모 청크 조회
   중복 제거 → 부모 청크 리스트 반환

- 실무 권장 설정: 부모 chunk_size 1500~2000, 자식 chunk_size 300~500

- 사용 시기: 검색 정밀도와 문맥 풍부함을 동시에 원할 때, 긴 문서에서 특정 부분을 찾되 주변 맥락도 필요할 때

2.5.2.9. Self-Query Retriever

- 사용자 질의를 LLM이 분석하여 의미 검색 부분과 메타데이터 필터 조건을 자동으로 분리한다. "2024년 소득세 관련 문서"라는 질의에서 "소득세"는 의미 검색, "2024년"은 메타데이터 필터로 자동 변환한다.

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.schema import AttributeInfo

# ── LLM 설정 ──
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 벡터스토어 설정 ──
db = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(),
)

# ── 메타데이터 필드 정의 ──
metadata_field_info = [
    AttributeInfo(
        name="year",
        description="문서의 연도 (예: 2024, 2025)",
        type="integer",
    ),
    AttributeInfo(
        name="category",
        description="문서 카테고리 (소득세, 법인세, 부가세)",
        type="string",
    ),
]

# ── SelfQueryRetriever 생성 ──
self_query_retriever = SelfQueryRetriever.from_llm(
    llm=llm,
    vectorstore=db,
    document_contents="세법 관련 법령 및 해설 문서",
    metadata_field_info=metadata_field_info,
    verbose=True,
    enable_limit=True,
    search_type="similarity",
)

# ══════════════════════════════════════════
# 내부 동작 흐름 확인
# ══════════════════════════════════════════

# ── 1단계: LLM 질의 분석 (Query Decomposition) ──
# SelfQueryRetriever 내부의 query_constructor를 직접 호출하여
# LLM이 사용자 질문을 어떻게 분석하는지 확인
question = "2024년 소득세 관련 문서를 찾아줘"

structured_query = self_query_retriever.query_constructor.invoke(question)
print("=" * 60)
print("1단계: LLM 질의 분석 결과")
print("=" * 60)
print(f"  검색 질의 (query): {structured_query.query}")
print(f"  메타데이터 필터 (filter): {structured_query.filter}")
print()

# ── 2단계: 내부 쿼리 변환 ──
# StructuredQuery → 벡터스토어 네이티브 필터로 변환
# (Chroma의 경우 $eq, $gt 등의 where 조건으로 변환됨)
print("=" * 60)
print("2단계: 내부 쿼리 변환")
print("=" * 60)
print(f"  원본 질문: {question}")
print(f"  → 시맨틱 검색어: \"{structured_query.query}\"")
print(f"  → 메타데이터 필터: {structured_query.filter}")
print(f"  예상 Chroma 필터: year == 2024 AND category == '소득세'")
print()

# ── 3단계: 벡터스토어 실행 (사용자에게 보이지 않음) ──
# 실제로는 self_query_retriever.invoke() 한 번으로 1~3단계가 모두 실행됨
print("=" * 60)
print("3단계: 벡터스토어 실행")
print("=" * 60)

docs = self_query_retriever.invoke(question)

print(f"  검색된 문서 수: {len(docs)}")
for i, doc in enumerate(docs, 1):
    meta = doc.metadata
    print(f"\n  [{i}] 메타데이터: {meta}")
    print(f"      내용: {doc.page_content[:100]}...")
```

**실행 시 출력 예시:**
```
============================================================
1단계: LLM 질의 분석 결과
============================================================
  검색 질의 (query): 소득세
  메타데이터 필터 (filter): and(eq("year", 2024), eq("category", "소득세"))

============================================================
2단계: 내부 쿼리 변환
============================================================
  원본 질문: 2024년 소득세 관련 문서를 찾아줘
  → 시맨틱 검색어: "소득세"
  → 메타데이터 필터: and(eq("year", 2024), eq("category", "소득세"))
  예상 Chroma 필터: year == 2024 AND category == '소득세'

============================================================
3단계: 벡터스토어 실행
============================================================
  검색된 문서 수: 4

  [1] 메타데이터: {'year': 2024, 'category': '소득세'}
      내용: 근로소득세는 총급여액에서 근로소득공제를 차감하여...

- 실무 권장 설정: 메타데이터 필드를 상세히 정의할수록 정확한 필터 추출 가능

- 사용 시기: 메타데이터(날짜, 카테고리, 작성자 등)로 필터링이 필요한 경우

2.5.2.10. k 값 선정 기준

k 값	특징	적합한 경우	세법 RAG 예시 github+1
1~2	가장 관련성 높은 결과만	단순 질의, 정확한 답	"제46조 공제액 계산법?"
3~4	관련성과 다양성 균형	일반 Q&A (가장 보편적)	"2024 소득세 주요 변경점?"
5~10	넓은 맥락 제공	복합 질의, 분석	"소득세 vs 법인세 비교"
10+	최대 맥락	종합 보고서 생성	"2025 세법 전체 개정안 분석"

2.5.2.11. 실무에서 가장 많이 사용하는 검색 조합

순위	검색 방식	적합한 상황	설정 예시 python.langchain+1
1	하이브리드 (Ensemble)	대부분의 프로덕션 환경	EnsembleRetriever(retrievers=[vector(k=4), BM25(k=4)], weights=[0.5, 0.5])
2	MMR	유사 문서 많은 도메인 (세법 조문 중복)	k=4, fetch_k=20, lambda_mult=0.5
3	유사도 검색	단순 Q&A, 빠른 프로토타입	similarity_search(k=4)
4	Parent Document + 하이브리드	긴 문서, 전체 문맥 필요	부모 2000자, 자식 400자 + Ensemble

2.6. 단계 6: 질의 변환 (Query Transformation)

- 사용자의 원본 질의를 검색에 최적화된 형태로 변환하는 전처리 단계이다. 사용자가 입력하는 자연어 질의는 일상적 표현, 모호한 용어, 복합적인 의도를 포함하는 경우가 많아 그대로 벡터 검색에 사용하면 최적의 결과를 얻기 어렵다. 질의 변환은 이러한 간극을 줄여 검색 품질을 높이는 핵심 기법이다.

- 예를 들어 사용자가 "직장인이 내는 세금 줄이는 법"이라고 질문했을 때, 벡터 DB에 저장된 문서에는 "근로소득자의 소득공제 방법"이라는 표현이 있을 수 있다. 질의 변환은 이러한 표현의 차이를 보완하여 관련 문서를 찾을 확률을 높인다.

2.6.1. 질의 변환 기법 상세

기법	설명	구현 방식 (LangChain) langchain+1
키워드 사전 변환	일상 용어→도메인 용어 ("세금"→"소득세")	KeywordDictionary(llm, dictionary={"세금":"소득세 제46조"})
질의 확장 (Query Expansion)	동의어/관련어 추가 ("공제"→"감면,환급")	LLM으로 3~5개 변형 생성 후 병합
질의 분해 (Decomposition)	복합 질의→하위 질의 ("소득세 vs 법인세"→2개 질의)	DecomposingRetriever(llm, base_retriever)
HyDE	가상 답변 생성 후 검색 (질의 임베딩 대신 답변 임베딩)	HyDERetriever(retriever=base_retriever)
Step-back Prompting	추상화 ("제46조 공제액"→"소득세 공제 원리")	LLM 프롬프트 + 원질의 병렬 검색
Multi-Query	다양한 관점 질의 ("소득세 변경"→5개 관점)	MultiQueryRetriever.from_llm(llm, retriever)

2.6.1.1. 키워드 사전 변환

- 가장 단순하면서도 효과적인 방법이다. 사용자가 자주 쓰는 일상 용어를 도메인 전문 용어로 치환하는 사전을 미리 정의해두고, 질의를 변환한다. LLM 호출 없이 규칙 기반으로 동작하므로 빠르고 비용이 들지 않는다.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# ── LLM 설정 ──
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 키워드 사전 ──
keyword_dict = """
직장인 → 거주자, 근로소득자
월급쟁이 → 근로소득자
세금 → 소득세, 법인세
연말정산 → 근로소득 연말정산, 소득공제
"""

# ── 사전 기반 질의 변환 체인 ──
dictionary_prompt = ChatPromptTemplate.from_template(
    "다음 사전을 참고하여 사용자의 질문을 변환하세요.\n"
    "사전: {dictionary}\n"
    "질문: {question}\n"
    "변환된 질문:"
)

dictionary_chain = dictionary_prompt | llm | StrOutputParser()

# ── 실행 ──
result = dictionary_chain.invoke({
    "question": "직장인 연말정산 세금 줄이는 법",
    "dictionary": keyword_dict,
})
print(result)
# 예상: "근로소득자 근로소득 연말정산 소득세 절세 방법"

2.6.1.2. 질의 확장 (Query Expansion)

- 원본 질의에 동의어, 관련어, 상위/하위 개념을 추가하여 검색 범위를 넓힌다. 질의를 완전히 바꾸는 것이 아니라 보강하는 방식이다.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# ── LLM 설정 ──
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 질의 확장 체인 ──
expansion_prompt = ChatPromptTemplate.from_template(
    "다음 질문에 대해 검색 성능을 높이기 위한 관련 키워드와 동의어를 포함하여 "
    "확장된 검색 질의를 생성하세요. 원래 의도를 유지하면서 관련 용어를 추가하세요.\n\n"
    "원본 질문: {question}\n"
    "확장된 검색 질의:"
)

expansion_chain = expansion_prompt | llm | StrOutputParser()

# ── 실행 ──
result = expansion_chain.invoke({"question": "소득세 절세 방법"})
print(result)
# 예상: "소득세 절세 방법 소득공제 세액공제 연말정산 근로소득 비과세"

2.6.1.3. 질의 분해 (Query Decomposition)

- 복합적인 질의를 독립적인 하위 질의로 분해하여 각각 검색한다. "소득세와 법인세의 차이점과 각각의 세율은?"처럼 여러 정보를 동시에 묻는 질의를 분해하면, 각 하위 질의에 대해 더 정확한 문서를 검색할 수 있다.

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_chroma import Chroma

# ── LLM 설정 ──
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 벡터스토어 / Retriever 설정 ──
db = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(),
)
retriever = db.as_retriever(search_kwargs={"k": 4})

# ── 질의 분해 체인 ──
decomposition_prompt = ChatPromptTemplate.from_template(
    "다음 복합 질문을 독립적으로 검색할 수 있는 하위 질문들로 분해하세요.\n"
    "각 하위 질문은 한 가지 정보만 묻도록 하세요.\n\n"
    "복합 질문: {question}\n"
    "하위 질문들 (각 줄에 하나씩):"
)

decomposition_chain = decomposition_prompt | llm | StrOutputParser()

sub_questions = decomposition_chain.invoke({
    "question": "소득세와 법인세의 차이점과 각각의 세율은?"
})
print(f"하위 질문:\n{sub_questions}\n")

# ── 각 하위 질의로 검색 + 중복 제거 ──
all_docs = []
seen_ids = set()

for sub_q in sub_questions.strip().split("\n"):
    if sub_q.strip():
        docs = retriever.invoke(sub_q.strip())
        for doc in docs:
            doc_id = doc.metadata.get("id", hash(doc.page_content))
            if doc_id not in seen_ids:
                all_docs.append(doc)
                seen_ids.add(doc_id)

# Top-8 선택 (과도한 컨텍스트 방지)
final_docs = all_docs[:8]

for i, doc in enumerate(final_docs, 1):
    print(f"[{i}] {doc.page_content[:100]}...")

2.6.1.4. HyDE (Hypothetical Document Embedding)

- 질의를 직접 임베딩하는 대신, LLM에게 "이 질문에 대한 답변을 담은 가상의 문서"를 먼저 생성시키고, 그 가상 문서를 임베딩하여 벡터 검색에 사용한다. 질문과 문서는 임베딩 공간에서 서로 다른 영역에 위치하는 경향이 있는데(질문은 의문형, 문서는 서술형), HyDE는 가상 답변 문서를 만들어 문서 영역에서 검색함으로써 이 차이를 줄인다.

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_chroma import Chroma

# ── LLM 설정 ──
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 벡터스토어 설정 ──
db = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(),
)

# ── 1단계: HyDE - 가상 문서 생성 ──
hyde_prompt = ChatPromptTemplate.from_template(
    "다음 질문에 대한 답변을 포함하는 문서의 관련 단락을 작성하세요.\n"
    "정확하지 않아도 되며, 실제 문서에 있을 법한 내용을 작성하면 됩니다.\n\n"
    "질문: {question}\n"
    "관련 문서 단락:"
)

hyde_chain = hyde_prompt | llm | StrOutputParser()

hypothetical_doc = hyde_chain.invoke({
    "question": "근로소득세 계산 방법은?"
})
print(f"가상 문서:\n{hypothetical_doc}\n")

# ── 2단계: 가상 문서를 임베딩하여 유사 문서 검색 ──
docs = db.similarity_search(hypothetical_doc, k=4)

for i, doc in enumerate(docs, 1):
    print(f"[{i}] {doc.page_content[:100]}...")

2.6.1.5. Step-back Prompting

- 구체적인 질의를 한 단계 추상화하여 더 넓은 범위의 관련 문서를 검색한다. "2024년 근로소득세율 6% 구간의 과세표준은?"처럼 매우 구체적인 질의는 벡터 검색에서 관련 문서를 놓칠 수 있으므로, "근로소득세율 구간과 과세표준"처럼 추상화된 질의로 변환하여 검색한다.

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_chroma import Chroma

# ── LLM 설정 ──
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 벡터스토어 / Retriever 설정 ──
db = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(),
)
retriever = db.as_retriever(search_kwargs={"k": 3})

# ── Step-Back Prompting 체인 ──
stepback_prompt = ChatPromptTemplate.from_template(
    "다음 질문에서 한 단계 뒤로 물러나서, 이 질문에 답하기 위해 필요한 "
    "더 넓은 범위의 배경 지식을 묻는 질문을 생성하세요.\n\n"
    "구체적 질문: {question}\n"
    "추상화된 질문:"
)

stepback_chain = stepback_prompt | llm | StrOutputParser()

# ── 실행 ──
question = "2024년 근로소득세율 6% 구간의 과세표준은?"

# 1. 추상화된 질문 생성
abstract_question = stepback_chain.invoke({"question": question})
print(f"추상화된 질문: {abstract_question}")

# 2. 원본 + 추상화 질의 모두로 검색하여 합산
docs_specific = retriever.invoke(question)
docs_abstract = retriever.invoke(abstract_question)
all_docs = docs_specific + docs_abstract

# 3. 중복 제거
seen = set()
unique_docs = []
for doc in all_docs:
    if doc.page_content not in seen:
        seen.add(doc.page_content)
        unique_docs.append(doc)

for i, doc in enumerate(unique_docs, 1):
    print(f"[{i}] {doc.page_content[:100]}...")

2.6.1.6. Multi-Query (다관점 질의 생성)

- 하나의 질의를 여러 다른 관점에서 재구성하여 검색 재현율을 높인다. 각 변형 질의가 원본과 같은 의도를 가지지만 다른 표현을 사용하므로, 다양한 관련 문서를 찾을 수 있다.

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.retrievers.multi_query import MultiQueryRetriever

# ── LLM 설정 ──
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 벡터스토어 설정 ──
db = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(),
)

# ── MultiQueryRetriever ──
# 자동으로 3개의 변형 질의를 생성하여 검색
multi_retriever = MultiQueryRetriever.from_llm(
    retriever=db.as_retriever(search_kwargs={"k": 4}),
    llm=llm,
)

# ── 실행 ──
# 예: "소득세 절세 방법"
# → 변형1: "근로소득세를 줄이기 위한 공제 항목은?"
# → 변형2: "소득세 납부액을 절약하는 전략은?"
# → 변형3: "세금 부담을 줄이는 합법적 방법은?"
# → 세 질의의 검색 결과를 중복 제거 후 합산
docs = multi_retriever.invoke("소득세 절세 방법")

for i, doc in enumerate(docs, 1):
    print(f"[{i}] {doc.page_content[:100]}...")

"소득세 절세 방법"
↓
LLM 3개 변형 생성:
1. "근로소득세 줄이는 공제 항목?"
2. "소득세 절약 전략은?"
3. "세금 감면 방법들?"
↓
각각 retriever.invoke(k=4) → 12개 문서
↓
RRF 병합 → 중복 제거 → Top-6~8 반환

2.6.2. 질의 변환 활용 예시 모음

2.6.2.1. 예시 1: 키워드 사전 + 질의 확장 조합

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 용어 변환 사전 ──
tax_dictionary = """
- 직장인 → 근로소득자
- 세금 줄이는 법 → 절세 방법, 소득공제, 세액공제
- 월급 → 근로소득
- 퇴직금 → 퇴직소득
- 프리랜서 → 사업소득자
- 집 살 때 세금 → 취득세, 양도소득세
"""

# ── 사전 변환 + 질의 확장 체인 ──
combined_prompt = ChatPromptTemplate.from_template(
    "1단계: 다음 사전을 참고하여 질문의 용어를 전문 용어로 변환하세요.\n"
    "사전: {dictionary}\n\n"
    "2단계: 변환된 질문에 관련 동의어와 키워드를 추가하여 검색 질의를 확장하세요.\n\n"
    "원본 질문: {question}\n"
    "최종 검색 질의:"
)

combined_chain = combined_prompt | llm | StrOutputParser()

# ── 실행 ──
result = combined_chain.invoke({
    "question": "직장인 세금 줄이는 법",
    "dictionary": tax_dictionary,
})

print(result)
# 예상 출력: "근로소득자 소득세 절세 방법 소득공제 세액공제 연말정산"

2.6.2.2. 예시 2: HyDE + Retriever 통합

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# ── LLM 설정 ──
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── HyDE 프롬프트: 가상의 답변 문서를 생성 ──
hyde_prompt = ChatPromptTemplate.from_template(
    """다음 질문에 대해 답변하는 문서를 작성하세요.
실제 문서처럼 상세하고 사실적으로 작성하세요.

질문: {question}
문서:"""
)

# ── Retriever 설정 ──
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# ── HyDE 체인: 질문 → 가상 문서 생성 → 가상 문서로 검색 ──
hyde_retrieval_chain = (
    {"question": RunnablePassthrough()}
    | hyde_prompt
    | llm
    | StrOutputParser()
    | retriever
)

docs = hyde_retrieval_chain.invoke("퇴직금에 대한 세금은 어떻게 계산하나요?")

2.6.2.3. 예시 3: 질의 분해 + 결과 종합

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.documents import Document

# ── LLM 설정 ──
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 질의 분해 체인 ──
decomposition_prompt = ChatPromptTemplate.from_template(
    """다음 복합 질문을 검색에 적합한 단순 하위 질문 2~4개로 분해하세요.
각 질문은 번호를 붙여 줄바꿈으로 구분하세요.

질문: {question}
하위 질문:"""
)

decomposition_chain = decomposition_prompt | llm | StrOutputParser()

# ── Retriever (예시: Chroma 벡터스토어) ──
# 실제 환경에 맞게 교체하세요
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# ── 복합 질의 분해 + 검색 + 중복 제거 ──
def decompose_and_search(question: str) -> list[Document]:
    # 1. 질의 분해
    sub_questions_raw = decomposition_chain.invoke({"question": question})

    # 2. 각 하위 질의로 검색
    all_docs = []
    for line in sub_questions_raw.strip().split("\n"):
        sub_q = line.strip().lstrip("0123456789. ")
        if sub_q:
            docs = retriever.invoke(sub_q)
            all_docs.extend(docs)

    # 3. 중복 제거
    seen = set()
    unique_docs = []
    for doc in all_docs:
        if doc.page_content not in seen:
            seen.add(doc.page_content)
            unique_docs.append(doc)

    return unique_docs

# ── 최종 종합 답변 체인 ──
from langchain.chains.combine_documents import create_stuff_documents_chain

answer_prompt = ChatPromptTemplate.from_template(
    """다음 문서를 참고하여 질문에 상세하게 답하세요.

문서:
{context}

질문: {input}
답변:"""
)

stuff_chain = create_stuff_documents_chain(llm, answer_prompt)

# ── 실행 ──
question = "소득세와 법인세의 차이점과 각각의 세율은?"
docs = decompose_and_search(question)

result = stuff_chain.invoke({
    "input": question,
    "context": docs,
})

print(result)

2.7. 단계 7: 컨텍스트 구성과 생성 (Generation)

- 검색된 문서를 LLM에 전달하여 최종 답변을 생성한다. 이 단계에서는 검색된 여러 문서를 어떻게 결합하여 LLM에 전달할지(문서 결합 전략)와, 어떤 체인 구조로 검색-생성 파이프라인을 구성할지가 핵심이다.

2.7.1. 문서 결합 전략

- 검색된 문서 청크가 여러 개일 때, 이를 LLM에 전달하는 방식이 문서 결합 전략이다. 문서의 양, 토큰 제한, 응답 품질 요구 수준에 따라 적절한 전략을 선택한다.

전략	설명	장점	단점	세법 추천도
Stuff	모든 문서를 하나의 프롬프트에 삽입	구현 간단, 문맥 완전	토큰 제한 초과 가능	⭐⭐⭐⭐⭐ (k=4 이하)
Map-Reduce	각 문서별 요약 후 종합	대량 문서 처리 가능	속도 느림, 비용 증가	⭐⭐⭐ (복합질의)
Refine	순차적으로 답변 정제	상세한 답변	순서 의존성, 느림	⭐⭐⭐⭐ (법률 해석)
Map-Rerank	각 문서별 답변+점수 후 최고 선택	최적 답변 선택	비용 높음	⭐⭐⭐⭐ (고정밀)

- Stuff (한번에 넣기)

[문서1 + 문서2 + 문서3] → LLM 1회 호출 → 최종 답변

- Map-Reduce (분할 → 종합)

- 각 문서를 개별 요약(Map)한 뒤, 요약들을 모아 종합(Reduce)합니다. Map 단계는 병렬 처리가 가능합니다. LLM 호출은 N+1회입니다.

[문서1] → LLM → 요약1 ─┐
[문서2] → LLM → 요약2 ──┼→ [요약1+요약2+요약3] → LLM → 최종 답변
[문서3] → LLM → 요약3 ─┘

- Refine (순차 정제)

- 앞 문서의 답변을 다음 문서로 계속 보완합니다. 이전 맥락이 누적되므로 정밀하지만, 순차 처리라 병렬화가 불가능합니다. LLM 호출은 N회입니다.

[문서1] → LLM → 초기 답변
                  ↓
[초기 답변 + 문서2] → LLM → 정제1
                              ↓
[정제1 + 문서3] → LLM → 정제2 (최종)

- Map-Rerank (개별 평가 → 선택)

- 각 문서에서 독립적으로 답변과 신뢰도 점수를 생성한 뒤, 최고 점수 답변 하나를 선택합니다. 문서 간 정보를 종합하지 않습니다. LLM 호출은 N회입니다.

[문서1] → LLM → 답변1 (score: 85)
[문서2] → LLM → 답변2 (score: 90) ← 최고 점수 선택
[문서3] → LLM → 답변3 (score: 15)

항목	Stuff	Map-Reduce	Refine	Map-Rerank
LLM 호출	1회	N+1회	N회	N회
병렬 처리	-	✅ 가능	❌ 불가	✅ 가능
문서 간 종합	✅	✅	✅ 누적	❌ 단일 선택
토큰 효율	낮음	높음	높음	높음
적합 상황	소량 문서	대량 요약	정밀 답변	최적 문서 선택

2.7.1.1. Stuff (전체 삽입)

- 가장 단순하고 많이 사용되는 전략이다. 검색된 모든 문서를 하나의 프롬프트에 연결하여 LLM에 한 번에 전달한다. 문서 합계가 LLM의 컨텍스트 윈도우 이내일 때 사용한다.

from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template(
    "다음 문서를 참고하여 질문에 답하세요.\n\n"
    "문서:\n{context}\n\n"
    "질문: {input}\n"
    "답변:"
)

# 모든 문서를 {context}에 연결하여 삽입
stuff_chain = create_stuff_documents_chain(llm, prompt)

# 실행 시 documents 리스트를 context로 전달
result = stuff_chain.invoke({
    "input": "근로소득세 계산 방법은?",
    "context": retrieved_docs  # 검색된 Document 리스트
})

- 동작: `[문서1] + [문서2] + [문서3] + [문서4]` → 하나의 프롬프트로 결합 → LLM 1회 호출

- 적합: 검색된 문서 합계가 LLM 컨텍스트 윈도우의 70~80% 이하일 때

2.7.1.2. Map-Reduce (맵-리듀스)

- 각 문서에 대해 개별적으로 LLM을 호출하여 중간 결과(요약 또는 부분 답변)를 생성한 뒤(Map 단계), 모든 중간 결과를 다시 LLM에 전달하여 최종 답변을 종합한다(Reduce 단계). 병렬 처리가 가능하여 대량 문서에 적합하다.

from langchain.chains import MapReduceDocumentsChain, ReduceDocumentsChain, LLMChain
from langchain.chains.combine_documents.stuff import StuffDocumentsChain
from langchain_core.prompts import ChatPromptTemplate

# Map 단계: 문서별 요약
map_prompt = ChatPromptTemplate.from_template(
    """다음 문서에서 질문과 관련된 핵심 내용을 요약하세요.
문서: {context}
질문: {question}
관련 내용 요약:"""
)

# Reduce 단계: 요약들 종합
reduce_prompt = ChatPromptTemplate.from_template(
    """다음은 여러 문서에서 추출한 관련 내용 요약입니다.
이를 종합하여 질문에 대한 최종 답변을 생성하세요.
요약 목록:
{context}
질문: {question}
최종 답변:"""
)

# 체인 구성
map_chain = LLMChain(llm=llm, prompt=map_prompt)
reduce_chain = LLMChain(llm=llm, prompt=reduce_prompt)

# Reduce 시 문서들을 stuff 방식으로 합침
combine_documents_chain = StuffDocumentsChain(
    llm_chain=reduce_chain,
    document_variable_name="context",
)

# 문서가 너무 많으면 재귀적으로 reduce
reduce_documents_chain = ReduceDocumentsChain(
    combine_documents_chain=combine_documents_chain,
    collapse_documents_chain=combine_documents_chain,  # 토큰 초과 시 재축소
    token_max=4000,
)

# Map-Reduce 체인
map_reduce_chain = MapReduceDocumentsChain(
    llm_chain=map_chain,                        # Map 단계
    reduce_documents_chain=reduce_documents_chain,  # Reduce 단계
    document_variable_name="context",
    return_intermediate_steps=False,
)

# 실행
result = map_reduce_chain.invoke({
    "input_documents": retrieved_docs,
    "question": "소득세 전체 개정사항 요약"
})

1. 입력: {"input_documents": [doc1,doc2,...,doc20], "question": "소득세 개정사항"}

2. Map 단계 (병렬 실행, N개 LLM 호출):
   doc1 → map_prompt → "2024 소득공제 한도 상향"
   doc2 → map_prompt → "세율 누진구조 변경"
   ...
   doc20 → map_prompt → "비과세 한도 조정"
   ↓
   중간결과: [요약1, 요약2, ..., 요약20]

3. Reduce 단계 (1회 LLM 호출):
   reduce_prompt + 모든 요약들 → 
   "2024 소득세 개정사항: 1) 공제 상향 2) 세율 변경 3) 비과세 조정"

4. 출력: {"answer": 최종 종합 답변}

- 동작: `[문서1]→LLM→요약1`, `[문서2]→LLM→요약2`, ... → `[요약1+요약2+...]→LLM→최종답변`

- LLM 호출 수: N(Map) + 1(Reduce) = N+1회

- 적합: 문서가 많아 Stuff로 한 번에 넣을 수 없을 때, 병렬 처리가 필요할 때

2.7.1.3. Refine (순차적 정제)

- 첫 번째 문서로 초기 답변을 생성한 뒤, 두 번째 문서를 보면서 답변을 보완하고, 세 번째 문서로 다시 보완하는 식으로 순차적으로 답변을 정제해 나간다. 문서 순서에 의존하며, 이전 문서의 맥락이 다음 단계에 누적된다.

"""
Refine 체인: 문서를 순차적으로 읽으며 답변을 정제
- 문서1 → 초기 답변 생성
- 문서2 → 기존 답변 + 새 문서로 보완
- 문서3 → 다시 보완... (반복)
"""
from langchain.chains import RefineDocumentsChain, LLMChain
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.documents import Document

# ============================================================
# 1. LLM 및 문서 준비
# ============================================================
llm = ChatOpenAI(model="gpt-4o-mini")

retrieved_docs = [
    Document(page_content="소득세법상 과세표준 1,400만원 이하는 6% 세율 적용"),
    Document(page_content="과세표준 1,400만~5,000만원은 15%, 5,000만~8,800만원은 24%"),
    Document(page_content="과세표준 8,800만~1.5억은 35%, 10억 초과는 45% 최고세율"),
]

# ============================================================
# 2. 프롬프트 구성
# ============================================================
initial_prompt = ChatPromptTemplate.from_template(
    """다음 문서를 참고하여 질문에 답하세요.
문서: {context}
질문: {question}
답변:"""
)

refine_prompt = ChatPromptTemplate.from_template(
    """기존 답변: {existing_answer}
추가 문서를 참고하여 기존 답변을 보완하세요.
추가 정보가 없으면 기존 답변을 그대로 유지하세요.
추가 문서: {context}
질문: {question}
보완된 답변:"""
)

# ============================================================
# 3. 체인 구성
# ============================================================
initial_chain = LLMChain(llm=llm, prompt=initial_prompt)
refine_chain = LLMChain(llm=llm, prompt=refine_prompt)

chain = RefineDocumentsChain(
    initial_llm_chain=initial_chain,
    refine_llm_chain=refine_chain,
    document_variable_name="context",
    initial_response_name="existing_answer",
)

# ============================================================
# 4. 실행
# ============================================================
result = chain.invoke({
    "input_documents": retrieved_docs,
    "question": "소득세 세율 구간별 한도"
})

print(result["output_text"])
```

---

## 3. 동작 흐름
```
[문서1] "1,400만원 이하 6%"
    → LLM 초기 답변: "소득세는 1,400만원 이하 6%입니다."

[문서2] "1,400만~5,000만원 15%, 5,000만~8,800만원 24%"
    → LLM 정제: "소득세는 1,400만원 이하 6%, 5,000만원까지 15%, 8,800만원까지 24%입니다."

[문서3] "8,800만~1.5억 35%, 10억 초과 45%"
    → LLM 최종 정제: "소득세는 6단계 누진세율로 6%~45%까지 적용됩니다. (전체 구간 포함)"

- LLM 호출 횟수는 문서 수와 동일(3건이면 3회)합니다. Stuff 방식(문서 전체를 한번에 넣는 것)보다 LLM 호출이 많지만, 각 호출의 컨텍스트가 작으므로 토큰 한도를 초과하는 대량 문서 처리에 적합합니다.

- 동작: `[문서1]→LLM→초기답변` → `[초기답변+문서2]→LLM→정제1` → `[정제1+문서3]→LLM→정제2(최종)`

- LLM 호출 수: N회 (초기 1회 + 정제 N-1회)

- 문서 순서: 검색 결과의 유사도 순(기본). 순서를 변경하려면 검색 결과 리스트를 직접 재정렬하면 된다

- 장점: 각 호출의 컨텍스트가 작아 토큰 한도 초과 없이 대량 문서 처리 가능

- 단점: 순차 처리로 병렬화 불가, 앞쪽 문서에 편향될 수 있음

- 적합: 여러 문서의 정보를 순차적으로 종합해야 할 때, 정밀한 답변이 필요할 때

2.7.1.4. Map-Rerank (맵-재순위)

- 각 문서에 대해 개별적으로 답변을 생성하면서 동시에 답변의 신뢰도 점수를 매긴 뒤, 가장 높은 점수의 답변을 최종 결과로 선택한다. "가장 관련성 높은 단일 문서에서 최적의 답변을 추출"하는 전략이다.

"""
Map-Rerank: 각 문서별 답변 + 신뢰도 점수 생성 → 최고 점수 답변 선택
- 동작: [문서1→LLM→답변+점수], [문서2→LLM→답변+점수], ... → 최고 점수 선택
- LLM 호출 수: N회 (문서 수만큼, 병렬화 가능)
- 적합: 가장 관련성 높은 단일 문서에서 최적 답변을 추출할 때
"""
import json
import re
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.documents import Document

# ============================================================
# 1. 설정
# ============================================================
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

retrieved_docs = [
    Document(page_content="소득세법상 과세표준 1,400만원 이하는 6% 세율 적용"),
    Document(page_content="과세표준 1,400만~5,000만원은 15%, 5,000만~8,800만원은 24%"),
    Document(page_content="부가가치세 일반과세자는 10% 세율이 적용된다"),
]

# ============================================================
# 2. Map-Rerank 프롬프트
# ============================================================
map_rerank_prompt = PromptTemplate.from_template(
    "다음 문서를 참고하여 질문에 답하세요.\n"
    "답변과 함께 이 문서가 질문에 얼마나 관련이 있는지 0~100 점수를 매기세요.\n\n"
    "문서: {context}\n"
    "질문: {question}\n\n"
    "반드시 다음 JSON 형식으로만 응답하세요 (다른 텍스트 없이):\n"
    '{{"answer": "답변 내용", "score": 점수}}\n'
)

# ============================================================
# 3. JSON 파싱 헬퍼
#    - LLM이 코드블록으로 감싸는 경우 대응
#    - ```json ... ``` 패턴 제거
# ============================================================
def parse_json_response(text: str) -> dict | None:
    # 마크다운 코드블록 제거
    cleaned = re.sub(r"```json?\s*", "", text)
    cleaned = re.sub(r"```", "", cleaned).strip()
    try:
        return json.loads(cleaned)
    except json.JSONDecodeError:
        return None

# ============================================================
# 4. Map-Rerank 실행
# ============================================================
def map_rerank_search(docs, question):
    results = []

    for i, doc in enumerate(docs, 1):
        response = llm.invoke(
            map_rerank_prompt.format(
                context=doc.page_content,
                question=question
            )
        )

        parsed = parse_json_response(response.content)
        if parsed and "answer" in parsed and "score" in parsed:
            results.append({
                "answer": parsed["answer"],
                "score": parsed["score"],
                "source": doc.page_content
            })
            print(f"  문서#{i} | Score: {parsed['score']:>3} | {doc.page_content[:50]}...")
        else:
            print(f"  문서#{i} | ⚠️ 파싱 실패: {response.content[:80]}...")

    if not results:
        return None

    results.sort(key=lambda x: x["score"], reverse=True)
    return results[0]

# ============================================================
# 5. 실행 및 결과
# ============================================================
question = "근로소득세 계산 방법은?"

print(f"[Map-Rerank] 질문: {question}\n")
best = map_rerank_search(retrieved_docs, question)

if best:
    print(f"\n[최종 선택]")
    print(f"  답변: {best['answer']}")
    print(f"  신뢰도: {best['score']}")
    print(f"  출처: {best['source'][:60]}...")
else:
    print("\n[실패] 유효한 답변을 생성하지 못했습니다.")
```

---

## 동작 흐름
```
[문서1] "1,400만원 이하 6%"     → LLM → {"answer": "...", "score": 85}
[문서2] "1,400만~8,800만원 구간" → LLM → {"answer": "...", "score": 90}  ← 최고 점수
[문서3] "부가가치세 10%"         → LLM → {"answer": "...", "score": 15}

→ Score 90인 문서2의 답변을 최종 선택

3. 파이프라인 단계별 의존 관계

문서 로딩 ─────→ 청킹 ─────→ 임베딩 ────────→ 벡터 DB
   │                │             │            │
   │ 포맷별 로더    │ 크기/전략   │ 모델 선택   │ DB 선택
   │ 선택 필요      │ 결정 필요   │ 결정 필요   │ 결정 필요
   │                │             │            │
   ↓                ↓             ↓            ↓
[입력 품질]     [청크 품질]    [벡터 품질]   [검색 성능]
                                                │
질의 변환 ←──── 사용자 질의                     │
   │                                            │
   ↓                                            ↓
[변환된 질의] ──→ 유사도 검색 ──→ 컨텍스트 구성 ──→ LLM 생성 ──→ 응답

"""
============================================================
LangChain RAG 파이프라인 — 단계별 함수 레퍼런스
============================================================
OpenAI 모델 기반, 각 단계에서 선택 가능한 함수와 추천 조건 정리

파이프라인 흐름:
  문서 로딩 → 청킹 → 임베딩 → 벡터 DB → (저장)
  사용자 질의 → 질의 변환 → 유사도 검색 → 컨텍스트 구성 → LLM 생성 → 응답
============================================================
"""

# ============================================================
# 공통 import
# ============================================================
import os
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate

os.environ["OPENAI_API_KEY"] = "your-key"


# ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
# PHASE 1: 인덱싱 파이프라인 (최초 1회)
# ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒


# ============================================================
# STEP 1. 문서 로딩 → [입력 품질] 결정
# ============================================================
# 💡 선택 기준: 원본 문서의 파일 포맷에 따라 로더를 선택
# ============================================================

# --- PDF ---
# 가장 흔한 문서 형식. 페이지 단위로 Document 객체 생성
from langchain_community.document_loaders import PyPDFLoader
pdf_docs = PyPDFLoader("./문서.pdf").load()

# --- Word (.docx) ---
# 워드 문서에서 텍스트만 추출 (서식 제거)
from langchain_community.document_loaders import Docx2txtLoader
docx_docs = Docx2txtLoader("./문서.docx").load()

# --- CSV ---
# 각 행을 하나의 Document로 변환. 표 형태 데이터에 적합
from langchain_community.document_loaders import CSVLoader
csv_docs = CSVLoader("./데이터.csv").load()

# --- 웹페이지 ---
# URL에서 HTML을 가져와 텍스트 추출
from langchain_community.document_loaders import WebBaseLoader
web_docs = WebBaseLoader("https://example.com/article").load()

# --- 텍스트 파일 ---
# 단순 .txt 파일 로드
from langchain_community.document_loaders import TextLoader
txt_docs = TextLoader("./문서.txt", encoding="utf-8").load()

# ┌─────────────────────────────────────────────────────┐
# │  로더 선택 가이드                                     │
# │  PDF        → PyPDFLoader (범용)                     │
# │  Word       → Docx2txtLoader                        │
# │  CSV/Excel  → CSVLoader                             │
# │  웹페이지    → WebBaseLoader                         │
# │  텍스트      → TextLoader                            │
# │  디렉토리    → DirectoryLoader (폴더 내 파일 일괄)     │
# └─────────────────────────────────────────────────────┘


# ============================================================
# STEP 2. 청킹 (텍스트 분할) → [청크 품질] 결정
# ============================================================
# 💡 선택 기준: 문서 특성과 정밀도 요구에 따라 분할 전략 선택
# ============================================================

docs = pdf_docs  # STEP 1에서 로드한 문서 사용

# --- 방법 A: RecursiveCharacterTextSplitter (가장 범용적, 기본 추천) ---
# 구분자 우선순위(\n\n → \n → . → 공백)로 재귀 분할
# 대부분의 일반 문서에 적합
from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter_a = RecursiveCharacterTextSplitter(
    chunk_size=500,     # 청크 최대 글자수 (짧을수록 정밀, 길수록 맥락 보존)
    chunk_overlap=50    # 앞 청크와 겹치는 글자수 (문맥 연결 유지)
)
chunks = splitter_a.split_documents(docs)

# --- 방법 B: TokenTextSplitter (토큰 수 기준) ---
# LLM 토큰 한도를 정확히 맞춰야 할 때
from langchain_text_splitters import TokenTextSplitter

splitter_b = TokenTextSplitter(
    chunk_size=200,     # 토큰 수 기준
    chunk_overlap=20
)

# --- 방법 C: SemanticChunker (의미 기반 분할) ---
# 임베딩으로 문장 간 의미 유사도를 계산하여 자연스러운 경계에서 분할
# 품질 최고, 비용 높음 (임베딩 API 호출 필요)
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

splitter_c = SemanticChunker(
    OpenAIEmbeddings(model="text-embedding-3-small")
)

# ┌──────────────────────────────────────────────────────────┐
# │  청킹 선택 가이드                                         │
# │                                                          │
# │  일반 문서 (보고서, 가이드)  → RecursiveCharacter (기본)    │
# │  토큰 한도 정밀 제어         → TokenTextSplitter          │
# │  최고 품질 (비용 허용 시)    → SemanticChunker             │
# │                                                          │
# │  chunk_size 가이드:                                       │
# │  200~500  → 짧은 QA, 정확한 검색 (정밀도 우선)             │
# │  500~1000 → 일반 문서 (균형)                               │
# │  1000+    → 긴 맥락 필요 (요약, 분석)                      │
# └──────────────────────────────────────────────────────────┘


# ============================================================
# STEP 3. 임베딩 → [벡터 품질] 결정
# ============================================================
# 💡 선택 기준: 품질, 비용, 언어(한국어) 지원에 따라 선택
# ============================================================

# --- 방법 A: OpenAI text-embedding-3-small (기본 추천) ---
# 가성비 최고. 1536차원. 대부분의 용도에 충분
from langchain_openai import OpenAIEmbeddings

embedding = OpenAIEmbeddings(model="text-embedding-3-small")

# --- 방법 B: OpenAI text-embedding-3-large (최고 품질) ---
# 3072차원. 정밀도가 중요한 프로덕션 환경
# embedding = OpenAIEmbeddings(model="text-embedding-3-large")

# ┌──────────────────────────────────────────────────────────┐
# │  임베딩 모델 선택 가이드 (OpenAI 기준)                      │
# │                                                          │
# │  text-embedding-3-small  → 가성비, 빠른 속도 (기본 추천)   │
# │  text-embedding-3-large  → 최고 품질, 프로덕션 환경         │
# │                                                          │
# │  ⚠️ 주의: 모델 변경 시 벡터 DB 전체 재구축 필요             │
# └──────────────────────────────────────────────────────────┘


# ============================================================
# STEP 4. 벡터 DB 저장 → [검색 성능] 결정
# ============================================================
# 💡 선택 기준: 인프라 환경, 데이터 규모, 운영 방식
# ============================================================

# --- 방법 A: FAISS (로컬, 고성능, 기본 추천) ---
# 서버 불필요. 인메모리 동작. save_local()로 수동 저장
from langchain_community.vectorstores import FAISS

db = FAISS.from_documents(documents=chunks, embedding=embedding)
db.save_local("./faiss_index")  # 수동 저장 필수

# 로드 시:
# db = FAISS.load_local("./faiss_index", embedding, allow_dangerous_deserialization=True)

# --- 방법 B: Chroma (로컬, 자동 저장) ---
# 메타데이터 필터 검색 지원. 자동 영속화 (save 불필요)
# from langchain_community.vectorstores import Chroma
#
# db = Chroma.from_documents(
#     documents=chunks,
#     embedding=embedding,
#     persist_directory="./chroma_db",
#     collection_name="my_collection"
# )

# --- 방법 C: Pinecone (클라우드, 대규모) ---
# 서버리스. 수억 벡터 지원. 자동 저장
# from langchain_pinecone import PineconeVectorStore
#
# db = PineconeVectorStore.from_documents(
#     documents=chunks,
#     embedding=embedding,
#     index_name="my-index"
# )

# ┌──────────────────────────────────────────────────────────┐
# │  벡터 DB 선택 가이드                                       │
# │                                                          │
# │  프로토타입 / 소규모    → FAISS (간편, 빠름)                │
# │  메타데이터 필터 필요    → Chroma (where 조건 검색)         │
# │  대규모 프로덕션        → Pinecone (클라우드, 확장성)       │
# │                                                          │
# │  저장 방식 차이:                                           │
# │  FAISS    → save_local() 수동 호출 (안 하면 유실!)         │
# │  Chroma   → 자동 영속화                                    │
# │  Pinecone → 클라우드 자동 저장                              │
# └──────────────────────────────────────────────────────────┘


# ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
# PHASE 2: 쿼리 파이프라인 (매번 실행)
# ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒


# ============================================================
# STEP 5. 사용자 질의 → 질의 변환 (선택적)
# ============================================================
# 💡 선택 기준: 검색 품질을 높이고 싶을 때 적용
#    단순 QA라면 생략하고 바로 STEP 6으로 가도 됨
# ============================================================

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
query = "소득세 환급 기준은?"

# --- 방법 A: 변환 없이 그대로 사용 (기본) ---
search_query = query

# --- 방법 B: HyDE (가상 문서 생성 후 임베딩) ---
# LLM이 가상의 답변 문서를 생성 → 그 문서로 검색
# 짧은 쿼리의 검색 품질을 크게 향상
hyde_prompt = ChatPromptTemplate.from_template(
    "다음 질문에 대한 전문적인 답변 문서를 작성하세요.\n질문: {question}\n답변:"
)
hyde_chain = hyde_prompt | llm
# hyde_doc = hyde_chain.invoke({"question": query}).content
# search_query = hyde_doc  # 가상 문서를 검색 쿼리로 사용

# --- 방법 C: Multi-Query (질문을 여러 버전으로 확장) ---
# 하나의 질문을 다양한 표현으로 변환 → 각각 검색 → 결과 합산
from langchain.retrievers.multi_query import MultiQueryRetriever

# multi_retriever = MultiQueryRetriever.from_llm(
#     retriever=db.as_retriever(),
#     llm=llm
# )
# multi_docs = multi_retriever.invoke(query)

# ┌──────────────────────────────────────────────────────────┐
# │  질의 변환 선택 가이드                                      │
# │                                                          │
# │  단순 QA              → 변환 없이 그대로 (기본)             │
# │  짧은 쿼리 품질 향상   → HyDE (가상 문서 생성)              │
# │  다양한 관점 검색      → Multi-Query (질문 확장)            │
# │                                                          │
# │  ⚠️ 변환 적용 시 LLM 호출이 추가되므로 비용/지연 증가       │
# └──────────────────────────────────────────────────────────┘


# ============================================================
# STEP 6. 유사도 검색 (Retriever)
# ============================================================
# 💡 선택 기준: 정확도 vs 다양성 vs 품질 보장
# ============================================================

# --- 방법 A: similarity (단순 유사도, 기본 추천) ---
# 가장 유사한 상위 k개 문서를 그대로 반환
retriever_a = db.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

# --- 방법 B: similarity_score_threshold (품질 보장) ---
# 유사도 점수가 threshold 이상인 문서만 반환 (0건 가능 → fallback 필요)
retriever_b = db.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={
        "score_threshold": 0.7,  # 최소 유사도 (점수 분포 확인 후 조정)
        "k": 4
    }
)

# --- 방법 C: MMR (다양성 확보) ---
# 유사도 + 다양성 균형. 중복 문서 제거
retriever_c = db.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 4,              # 최종 반환 수
        "fetch_k": 20,       # 1차 후보 수 (k의 3~10배)
        "lambda_mult": 0.5   # 0=다양성, 1=유사도
    }
)

retriever = retriever_a  # 기본 선택

# ┌──────────────────────────────────────────────────────────┐
# │  검색 전략 선택 가이드                                      │
# │                                                          │
# │  단순 QA, 빠른 검색      → similarity (기본)               │
# │  정확도 중시, 노이즈 제거  → score_threshold               │
# │  종합 분석, 보고서 생성    → MMR (다양성 확보)              │
# │                                                          │
# │  반환 문서 수:                                             │
# │  similarity       → 항상 k건                              │
# │  score_threshold  → 0~k건 (가변, fallback 필요)            │
# │  MMR              → 항상 k건                              │
# └──────────────────────────────────────────────────────────┘


# ============================================================
# STEP 7. 컨텍스트 구성 (Chain 전략)
# ============================================================
# 💡 선택 기준: 문서 수, 정밀도, 속도, 비용
# ============================================================

# --- 방법 A: Stuff (한번에 넣기, 기본 추천) ---
# 모든 문서를 프롬프트에 한번에 넣고 1회 호출. 가장 단순하고 빠름
from langchain.chains.combine_documents import create_stuff_documents_chain

stuff_prompt = ChatPromptTemplate.from_template(
    "다음 문서를 참고하여 질문에 답하세요.\n\n{context}\n\n질문: {input}\n답변:"
)

stuff_chain = create_stuff_documents_chain(llm, stuff_prompt)

# --- 방법 B: Map-Reduce (분할 요약 → 종합) ---
# 각 문서를 개별 요약(Map) → 요약들을 모아 종합(Reduce)
# LLM 호출: N+1회, Map 단계 병렬 가능
from langchain.chains import MapReduceDocumentsChain, LLMChain, ReduceDocumentsChain
from langchain.chains.combine_documents.stuff import StuffDocumentsChain
from langchain_core.prompts import PromptTemplate

# map_prompt = PromptTemplate.from_template("다음을 요약하세요:\n{context}")
# reduce_prompt = PromptTemplate.from_template("요약들을 종합하세요:\n{context}")

# --- 방법 C: Refine (순차 정제) ---
# 문서1→초기답변, 답변+문서2→정제, 답변+문서3→정제...
# LLM 호출: N회, 맥락 누적으로 정밀, 병렬 불가
from langchain.chains import RefineDocumentsChain

# initial_prompt = ...  (초기 답변용)
# refine_prompt = ...   (정제용)

# --- 방법 D: Map-Rerank (개별 평가 → 최고 점수 선택) ---
# 각 문서에서 독립 답변 + 신뢰도 점수 → 최고 점수 선택
# LLM 호출: N회, 병렬 가능, 문서 간 종합 안 함

# ┌──────────────────────────────────────────────────────────┐
# │  Chain 선택 가이드                                         │
# │                                                          │
# │  문서 소량 (1~5건)       → Stuff (1회 호출, 가장 빠름)      │
# │  문서 대량 요약           → Map-Reduce (병렬 가능)          │
# │  정밀한 종합 답변         → Refine (맥락 누적)              │
# │  최적 단일 문서 선택      → Map-Rerank (점수 기반)          │
# │                                                          │
# │  LLM 호출 횟수:                                           │
# │  Stuff: 1회 / Map-Reduce: N+1회                           │
# │  Refine: N회 / Map-Rerank: N회                            │
# └──────────────────────────────────────────────────────────┘


# ============================================================
# STEP 8. LLM 생성 → 응답
# ============================================================
# 💡 선택 기준: 품질, 비용, 속도
# ============================================================

# --- 방법 A: GPT-4o-mini (가성비, 기본 추천) ---
# 빠르고 저렴. 대부분의 RAG QA에 충분
from langchain_openai import ChatOpenAI

llm_a = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# --- 방법 B: GPT-4o (최고 품질) ---
# 복잡한 추론, 긴 문서 분석에 적합
llm_b = ChatOpenAI(model="gpt-4o", temperature=0)

# ┌──────────────────────────────────────────────────────────┐
# │  LLM 선택 가이드 (OpenAI 기준)                              │
# │                                                          │
# │  빠른 QA, 프로토타입  → gpt-4o-mini (기본 추천)             │
# │  복잡한 분석, 프로덕션 → gpt-4o (최고 품질)                  │
# │                                                          │
# │  temperature:                                            │
# │  0    → 일관된 답변 (사실 기반 QA에 추천)                   │
# │  0.7  → 창의적 답변 (브레인스토밍 등)                       │
# └──────────────────────────────────────────────────────────┘


# ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
# 전체 연결: 완전한 RAG 파이프라인 실행 예시
# ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒

from langchain.chains import create_retrieval_chain

# 1) Retriever + Stuff Chain 조합 (가장 기본적인 RAG)
rag_chain = create_retrieval_chain(retriever, stuff_chain)

# 2) 실행
result = rag_chain.invoke({"input": "소득세 환급 기준은?"})

# 3) 결과 확인
print(result["answer"])                # 최종 답변
print(len(result["context"]))          # 참조된 문서 수
for doc in result["context"]:          # 참조 문서 내용
    print(f"  - {doc.page_content[:80]}...")


# ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒
# 추천 조합 (상황별)
# ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒

# ┌──────────────────────────────────────────────────────────┐
# │  🚀 빠른 프로토타입                                        │
# │  로더: PyPDFLoader                                        │
# │  청킹: RecursiveCharacter (500/50)                        │
# │  임베딩: text-embedding-3-small                           │
# │  DB: FAISS                                               │
# │  검색: similarity (k=4)                                   │
# │  체인: Stuff                                              │
# │  LLM: gpt-4o-mini                                        │
# ├──────────────────────────────────────────────────────────┤
# │  🎯 프로덕션 (정확도 우선)                                  │
# │  로더: 문서 포맷별 선택                                     │
# │  청킹: SemanticChunker                                    │
# │  임베딩: text-embedding-3-large                           │
# │  DB: Pinecone                                            │
# │  검색: score_threshold (0.7) + fallback                   │
# │  체인: Refine                                             │
# │  LLM: gpt-4o                                             │
# ├──────────────────────────────────────────────────────────┤
# │  📊 종합 분석 / 보고서                                      │
# │  로더: 문서 포맷별 선택                                     │
# │  청킹: RecursiveCharacter (1000/150)                      │
# │  임베딩: text-embedding-3-large                           │
# │  DB: Chroma (메타데이터 필터 활용)                          │
# │  검색: MMR (다양성 확보)                                    │
# │  체인: Map-Reduce                                         │
# │  LLM: gpt-4o                                             │
# └──────────────────────────────────────────────────────────┘

저작자표시 비영리 동일조건 (새창열림)

'Study > RAG(Retrieval-Augmented Generation)' 카테고리의 다른 글

11. LangChain HuggingFace 오픈소스를 활용한 프로덕션급 RAG Pipeline 구성 (Advanced) (0)	2026.03.17
9. LangChain 실무 RAG 파이프라인 구현 가이드 (Advanced) (0)	2026.03.15
3. RAG 커스텀 최적화 - 3가지 시나리오별 맞춤 전략 (0)	2026.02.22
2. RAG 단계별 기술과 OpenAI API (0)	2026.02.18
[인프런] RAG 마스터: 기초부터 고급기법까지 (feat. LangChain) - 학습 후기 (0)	2026.01.17