RAG로 사내 데이터를 LLM에 연결하기: 임베딩과 벡터 DB, 청크 전략, Python 구현

작성 0xmote · 2026년 5월 22일 · 25분 읽기 · 수정 2026년 9월 25일 중급 가이드

이 글의 핵심

LLM은 학습 이후의 사내 문서를 모르기 때문에 그럴듯한 오답을 내놓기 쉽고, RAG는 검색한 근거를 함께 넣어 이 문제를 줄입니다. Pinecone과 ChromaDB 같은 벡터 DB 선택, 청크가 너무 크거나 작을 때 생기는 검색 품질 저하, 출처를 포함해 답하게 만드는 프롬프트와 캐싱 최적화를 다룹니다.

들어가며

RAG (Retrieval-Augmented Generation)는 LLM(대규모 언어 모델)의 한계를 극복하는 핵심 기술입니다. LLM은 학습 데이터 시점까지의 정보만 알고 있으며, 기업 내부 문서나 최신 정보는 모릅니다. RAG는 외부 데이터베이스에서 관련 정보를 검색하여 프롬프트에 추가함으로써, LLM이 최신·정확한 답변을 생성하도록 합니다. 이 글은 RAG의 동작 원리, 벡터 데이터베이스 선택, 청크 전략, 하이브리드 검색, Python 구현 예제, 실무 최적화를 단계별로 다룹니다.

RAG란?

기본 개념

전통적인 LLM 사용:

사용자 질문 → LLM → 답변 (학습 데이터 기반)

한계: 최신 정보 없음, 기업 내부 데이터 모름 RAG 사용:

사용자 질문 → 관련 문서 검색 → 문서 + 질문 → LLM → 답변

장점: 최신 정보, 기업 데이터 활용, 출처 제공 가능

RAG가 해결하는 문제

1. 지식 시점 문제

# ❌ LLM만 사용
질문: "우리 회사 2026년 1분기 매출은?"
답변: "죄송합니다. 해당 정보를 알 수 없습니다."
# ✅ RAG 사용
질문: "우리 회사 2026년 1분기 매출은?"
검색: [2026 Q1 재무 보고서] → "매출 150억원"
답변: "2026년 1분기 매출은 150억원입니다. (출처: Q1 재무보고서)"

2. 환각(Hallucination) 문제

# ❌ LLM만 사용 (지어낸 답변)
질문: "제품 X의 보증 기간은?"
답변: "일반적으로 1년입니다." (실제는 2년)
# ✅ RAG 사용 (문서 기반)
질문: "제품 X의 보증 기간은?"
검색: [제품 X 매뉴얼] → "보증 기간: 2년"
답변: "제품 X의 보증 기간은 2년입니다."

3. 도메인 지식 부족

# ❌ LLM만 사용
질문: "우리 회사 휴가 정책은?"
답변: "일반적인 기업은..." (일반론)
# ✅ RAG 사용
질문: "우리 회사 휴가 정책은?"
검색: [인사 규정 문서]
답변: "연차 15일, 병가 10일, 경조사 휴가 3일입니다."

RAG 아키텍처

전체 흐름

┌─────────────────────────────────────────────┐
│           1. 데이터 준비 (Offline)           │
└─────────────────────────────────────────────┘
                     │
        ┌────────────┼────────────┐
        ▼            ▼            ▼
   [문서 수집]  [청크 분할]  [임베딩 생성]
        │            │            │
        └────────────┼────────────┘
                     ▼
            [벡터 DB 저장]
┌─────────────────────────────────────────────┐
│           2. 검색 (Online)                   │
└─────────────────────────────────────────────┘
                     │
        사용자 질문 → [임베딩 변환]
                     │
                     ▼
            [벡터 유사도 검색]
                     │
                     ▼
            [상위 K개 문서 반환]
┌─────────────────────────────────────────────┐
│           3. 생성 (Online)                   │
└─────────────────────────────────────────────┘
                     │
        [질문 + 검색된 문서] → [LLM]
                     │
                     ▼
                  [답변 생성]

핵심 구성 요소

1. 문서 로더 (Document Loader)

PDF, Word, HTML, Markdown 등 다양한 포맷 지원
텍스트 추출 및 전처리 2. 텍스트 스플리터 (Text Splitter)
긴 문서를 작은 청크로 분할
맥락 유지를 위한 오버랩 설정 3. 임베딩 모델 (Embedding Model)
텍스트를 벡터로 변환
OpenAI, Cohere, Sentence-BERT 등 4. 벡터 데이터베이스 (Vector Database)
고차원 벡터 저장 및 검색
Pinecone, Weaviate, Qdrant, ChromaDB 등 5. LLM (Large Language Model)
검색된 문서를 기반으로 답변 생성
GPT-4, Claude, Llama 등

이 구성 요소 가운데 답변 품질을 가장 크게 좌우하는 것은 대개 LLM이 아니라 검색 단계입니다. 검색이 엉뚱한 청크를 가져오면 아무리 좋은 모델도 그 청크를 근거로 그럴듯한 오답을 만들고, 반대로 정답이 담긴 청크가 상위에 오면 작은 모델도 제대로 답합니다. 그래서 RAG 품질을 올릴 때는 모델을 바꾸기 전에 “질문별로 정답 청크가 top-k 안에 들어오는가”부터 확인하는 편이 효율적입니다. 처음 RAG를 만들 때 흔히 겪는 일이, 답이 틀려서 프롬프트를 몇 번이나 고쳤는데 알고 보니 해당 문서가 PDF 추출 단계에서 표가 깨져 텍스트로 아예 들어가지 않았던 경우입니다. 파이프라인 앞단(로더·청크)의 출력을 직접 눈으로 확인하는 습관이 디버깅 시간을 크게 줄여 줍니다.

벡터 임베딩

임베딩이란?

텍스트를 고차원 벡터로 변환하여 의미적 유사도를 계산 가능하게 합니다:

"강아지는 귀엽다" → [0.2, 0.8, 0.1, ..., 0.4]  (1536차원)
"개는 사랑스럽다" → [0.3, 0.7, 0.2, ..., 0.5]  (유사도 높음)
"자동차는 빠르다" → [0.9, 0.1, 0.8, ..., 0.2]  (유사도 낮음)

주요 임베딩 모델

모델	차원	성능	비용	특징
OpenAI text-embedding-3-large	3072	⭐⭐⭐⭐⭐	$$$	최고 성능
OpenAI text-embedding-3-small	1536	⭐⭐⭐⭐	$$	가성비 좋음
Cohere embed-multilingual-v3	1024	⭐⭐⭐⭐	$$	다국어 강점
Sentence-BERT (오픈소스)	768	⭐⭐⭐	Free	자체 호스팅

Python 예제

from openai import OpenAI
client = OpenAI(api_key="your-api-key")
def get_embedding(text: str) -> list[float]:
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding
# 사용 예
text = "C++26의 Static Reflection은 메타프로그래밍을 혁신합니다."
embedding = get_embedding(text)
print(f"차원: {len(embedding)}")  # 1536
print(f"첫 5개 값: {embedding[:5]}")

유사도 계산

import numpy as np
def cosine_similarity(vec1: list[float], vec2: list[float]) -> float:
    """코사인 유사도 계산 (-1~1, 높을수록 유사)"""
    a = np.array(vec1)
    b = np.array(vec2)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
# 예제
query = "C++ 리플렉션"
doc1 = "C++26 Static Reflection 가이드"
doc2 = "Python 웹 크롤링 튜토리얼"
emb_query = get_embedding(query)
emb_doc1 = get_embedding(doc1)
emb_doc2 = get_embedding(doc2)
print(f"query vs doc1: {cosine_similarity(emb_query, emb_doc1):.3f}")  # doc1 쪽이 더 높게 나옴
print(f"query vs doc2: {cosine_similarity(emb_query, emb_doc2):.3f}")

코사인 유사도의 절대값에 의미를 두지 않는 것이 중요합니다. 모델마다 점수 분포가 달라서, 어떤 모델은 무관한 문장끼리도 0.7 안팎이 나오고 어떤 모델은 관련 문장도 0.5를 잘 넘지 않습니다. “0.8 이상이면 관련 문서”처럼 고정 임계값을 박아 두면 임베딩 모델을 바꾸는 순간 검색이 조용히 망가집니다. 임계값이 필요하다면 실제 질문·정답 쌍으로 분포를 확인한 뒤 모델별로 정해야 합니다. 또 OpenAI 임베딩은 길이가 1로 정규화되어 나오므로 내적만으로 코사인 유사도와 같은 값을 얻을 수 있고, 벡터 DB에서 dotproduct 메트릭을 써도 결과가 같습니다.

문서와 질문은 반드시 같은 임베딩 모델로 변환해야 합니다. 모델을 text-embedding-3-small에서 -large로 바꾸면 차원(1536→3072)부터 달라서 벡터 DB가 dimension mismatch 류의 에러를 내고, 차원이 같은 다른 모델로 바꾸면 에러 없이 검색 품질만 무너지므로 더 위험합니다. 모델을 바꿀 때는 전체 문서를 다시 임베딩해 새 인덱스를 만들어야 하므로, 어떤 모델로 임베딩했는지를 인덱스 이름이나 메타데이터에 기록해 두는 편이 안전합니다.

벡터 데이터베이스

주요 벡터 DB 비교

DB	타입	특징	추천 용도
Pinecone	관리형	쉬운 시작, 자동 스케일링	프로토타입, 빠른 출시
Weaviate	오픈소스/관리형	하이브리드 검색 강력	복잡한 검색 요구사항
Qdrant	오픈소스	Rust 기반, 빠른 성능	자체 호스팅, 대용량
ChromaDB	오픈소스	간단한 API, 로컬 개발	개발/테스트 환경
Milvus	오픈소스	엔터프라이즈급	대규모 프로덕션

Pinecone 예제

from pinecone import Pinecone, ServerlessSpec
from openai import OpenAI
# 초기화 (pinecone 패키지 v3 이상: pinecone.init()은 제거됨)
pc = Pinecone(api_key="your-pinecone-key")
# 인덱스 생성
index_name = "my-rag-index"
if index_name not in pc.list_indexes().names():
    pc.create_index(
        name=index_name,
        dimension=1536,  # OpenAI embedding 차원
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )
index = pc.Index(index_name)
# 문서 추가
openai_client = OpenAI(api_key="your-openai-key")
def add_document(doc_id: str, text: str, metadata: dict):
    embedding = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    ).data[0].embedding
    
    index.upsert(vectors=[{
        "id": doc_id,
        "values": embedding,
        "metadata": {"text": text, **metadata}
    }])
# 검색
def search(query: str, top_k: int = 3):
    query_embedding = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=query
    ).data[0].embedding
    
    results = index.query(
        vector=query_embedding,
        top_k=top_k,
        include_metadata=True
    )
    
    return [(match.metadata["text"], match.score) 
            for match in results.matches]
# 사용
add_document("doc1", "C++26은 Static Reflection을 도입했습니다.", 
             {"category": "cpp", "date": "2026-03-31"})
add_document("doc2", "Python은 동적 타이핑 언어입니다.", 
             {"category": "python", "date": "2026-03-30"})
results = search("C++ 리플렉션")
for text, score in results:
    print(f"[{score:.3f}] {text}")

ChromaDB 예제 (로컬 개발)

Pinecone 예제에서 메타데이터에 원문 text를 함께 넣은 이유는, 벡터 DB가 돌려주는 것이 ID와 점수뿐이면 원문을 다시 다른 저장소에서 조회해야 하기 때문입니다. 대신 Pinecone은 레코드당 메타데이터 크기에 제한(40KB)이 있으므로, 긴 청크를 통째로 넣다가 upsert가 거부되는 일이 생깁니다. 청크가 크거나 원문 관리가 따로 필요하다면 원문은 일반 DB나 오브젝트 스토리지에 두고 벡터 DB에는 ID만 두는 구조가 낫습니다.

import chromadb
# 클라이언트 생성 (Chroma 0.4 이상: 디스크에 영속 저장)
# 예전 Settings(chroma_db_impl="duckdb+parquet") 방식은 제거되어 에러가 납니다
client = chromadb.PersistentClient(path="./chroma_db")
# 컬렉션 생성
collection = client.get_or_create_collection(
    name="my_documents",
    metadata={"description": "RAG document store"}
)
# 문서 추가 (자동 임베딩)
collection.add(
    documents=[
        "C++26 Static Reflection은 컴파일 타임 타입 정보를 제공합니다.",
        "Rust는 메모리 안전성을 보장하는 시스템 프로그래밍 언어입니다.",
        "WebAssembly는 브라우저에서 네이티브 수준 성능을 제공합니다."
    ],
    metadatas=[
        {"category": "cpp", "level": "advanced"},
        {"category": "rust", "level": "intermediate"},
        {"category": "web", "level": "intermediate"}
    ],
    ids=["doc1", "doc2", "doc3"]
)
# 검색
results = collection.query(
    query_texts=["C++ 타입 정보"],
    n_results=2
)
print(results["documents"])
print(results["distances"])

ChromaDB의 collection.add(documents=...)는 임베딩 함수를 지정하지 않으면 기본 내장 모델(all-MiniLM-L6-v2 계열)로 임베딩합니다. 영어 위주로 학습된 작은 모델이라 한국어 문서에서는 검색 품질이 눈에 띄게 떨어지므로, 한국어 자료라면 embedding_function에 다국어 모델이나 OpenAI 임베딩을 명시하는 편이 좋습니다. 또 results["distances"]는 유사도가 아니라 거리라서 값이 작을수록 가깝습니다. Pinecone의 score(클수록 가까움)와 방향이 반대라 두 DB를 오가며 정렬 조건을 뒤집어 쓰는 실수가 흔합니다.

청크 전략

청크 크기 결정

고려 사항:

LLM 컨텍스트 윈도우: GPT-4는 128K 토큰, Claude는 200K 토큰
검색 정확도: 너무 작으면 맥락 부족, 너무 크면 노이즈 증가
비용: 토큰 수에 비례한 API 비용 권장 크기:
일반 문서: 500-1000 토큰 (약 2000-4000자)
코드: 함수/클래스 단위 (100-500 토큰)
FAQ: 질문-답변 쌍 단위

청크 분할 전략

1. 고정 크기 분할 (Fixed-size)

def chunk_by_tokens(text: str, chunk_size: int = 500, overlap: int = 50):
    # tiktoken으로 토큰 계산
    import tiktoken
    enc = tiktoken.get_encoding("cl100k_base")
    
    tokens = enc.encode(text)
    chunks = []
    
    for i in range(0, len(tokens), chunk_size - overlap):
        chunk_tokens = tokens[i:i + chunk_size]
        chunk_text = enc.decode(chunk_tokens)
        chunks.append(chunk_text)
    
    return chunks

2. 문장 경계 분할 (Sentence-based)

# 필요한 모듈 import
import nltk
nltk.download('punkt')
def chunk_by_sentences(text: str, max_sentences: int = 5):
    sentences = nltk.sent_tokenize(text)
    chunks = []
    
    for i in range(0, len(sentences), max_sentences):
        chunk = ' '.join(sentences[i:i + max_sentences])
        chunks.append(chunk)
    
    return chunks

3. 의미 단위 분할 (Semantic)

# 필요한 모듈 import
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""]  # 우선순위
)
chunks = splitter.split_text(long_document)

4. 마크다운 구조 기반

def chunk_by_markdown_sections(markdown_text: str):
    import re
    
    # ## 헤딩으로 분할
    sections = re.split(r'\n## ', markdown_text)
    chunks = []
    
    for i, section in enumerate(sections):
        if i > 0:
            section = "## " + section
        
        # 너무 긴 섹션은 추가 분할
        if len(section) > 2000:
            sub_chunks = chunk_by_sentences(section, max_sentences=10)
            chunks.extend(sub_chunks)
        else:
            chunks.append(section)
    
    return chunks

오버랩 전략

# 오버랩 없음: 맥락 손실 위험
chunks = ["A B C", "D E F", "G H I"]
# 오버랩 있음: 맥락 유지
chunks = ["A B C D", "C D E F", "E F G H I"]
#              ^^^      ^^^

권장 오버랩: 청크 크기의 10-20%

오버랩은 경계에서 문장이 잘려 의미가 반쪽씩 두 청크로 흩어지는 문제를 줄여 주지만, 비용도 있습니다. 오버랩만큼 같은 텍스트가 중복 임베딩되어 저장 공간과 임베딩 비용이 늘고, 인접한 두 청크가 거의 같은 내용으로 top-k를 함께 차지해 실제로 LLM에 들어가는 정보의 다양성이 줄어듭니다. 앞에서 본 MMR이나 인접 청크 병합을 함께 쓰는 이유가 여기에 있습니다. 헤딩·문단 경계로 자르는 구조 기반 분할을 쓰면 애초에 문장이 잘리는 일이 적어서 오버랩을 작게 잡아도 되는 경우가 많습니다.

검색 최적화

기본 검색 (Semantic Search)

from langchain.vectorstores import Pinecone
from langchain.embeddings import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Pinecone.from_existing_index(
    index_name="my-index",
    embedding=embeddings
)
# 검색
query = "C++ 메모리 관리 방법"
docs = vectorstore.similarity_search(query, k=3)
for doc in docs:
    print(doc.page_content)
    print(doc.metadata)

메타데이터 필터링

# 특정 카테고리만 검색
docs = vectorstore.similarity_search(
    query,
    k=3,
    filter={"category": "cpp", "level": "advanced"}
)
# 날짜 범위 필터
docs = vectorstore.similarity_search(
    query,
    k=3,
    filter={
        "date": {"$gte": "2026-01-01", "$lte": "2026-03-31"}
    }
)

MMR (Maximal Marginal Relevance)

문제: 유사한 문서만 반환되어 다양성 부족 해결: MMR로 관련성과 다양성 균형

docs = vectorstore.max_marginal_relevance_search(
    query,
    k=5,
    fetch_k=20,  # 후보 20개 중
    lambda_mult=0.5  # 0=다양성 우선, 1=관련성 우선
)

Python 구현

아래 예제는 흐름을 보여 주기 위해 LangChain의 예전 import 경로(langchain.vectorstores, langchain.embeddings 등)와 run()·get_relevant_documents() 호출을 그대로 둡니다. 최신 LangChain에서는 이 경로들이 langchain_community·langchain_openai 같은 별도 패키지로 옮겨졌고, 호출도 invoke()로 통일되어 옛 코드를 실행하면 LangChainDeprecationWarning이 뜨거나 import 에러가 납니다. 예를 들어 from langchain_openai import OpenAIEmbeddings, ChatOpenAI, from langchain_community.vectorstores import Chroma, retriever.invoke(query)처럼 바꾸면 됩니다. LangChain은 버전마다 API 변동이 커서, 운영 코드에서는 버전을 고정하고 업그레이드는 따로 검증하는 편이 안전합니다.

기본 RAG 시스템

from langchain.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA
# 1. 문서 로드
loader = DirectoryLoader('./docs', glob="**/*.md", loader_cls=TextLoader)
documents = loader.load()
# 2. 청크 분할
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = text_splitter.split_documents(documents)
# 3. 임베딩 및 벡터 DB 저장
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)
# 4. RAG 체인 생성
llm = ChatOpenAI(model="gpt-4", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # 모든 문서를 하나의 프롬프트에
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3})
)
# 5. 질문
query = "C++26 Static Reflection의 주요 장점은?"
answer = qa_chain.run(query)
print(answer)

커스텀 프롬프트

from langchain.prompts import PromptTemplate
template = """다음 문서를 참고하여 질문에 답변하세요.
답변은 한국어로 작성하며, 출처를 명시하세요.
문서:
{context}
질문: {question}
답변:"""
prompt = PromptTemplate(
    template=template,
    input_variables=["context", "question"]
)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(),
    chain_type_kwargs={"prompt": prompt}
)

출처 포함 답변

from langchain.chains import RetrievalQAWithSourcesChain
qa_chain = RetrievalQAWithSourcesChain.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever()
)
result = qa_chain({"question": query})
print("답변:", result["answer"])
print("출처:", result["sources"])

하이브리드 검색

Semantic + Keyword 검색

문제: 벡터 검색만으로는 정확한 키워드 매칭 부족 해결: BM25(키워드 검색) + 벡터 검색 결합

from langchain.retrievers import BM25Retriever, EnsembleRetriever
# BM25 검색기 (키워드 기반)
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 3
# 벡터 검색기
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
# 하이브리드 검색기
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # BM25: 40%, Vector: 60%
)
# 검색
docs = ensemble_retriever.get_relevant_documents(query)

Reranking

문제: 초기 검색 결과의 순서가 최적이 아닐 수 있음 해결: Reranker 모델로 재정렬

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank
# Cohere Rerank 사용
compressor = CohereRerank(
    model="rerank-english-v2.0",
    top_n=3
)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vector_retriever
)
# 검색 + 재정렬
docs = compression_retriever.get_relevant_documents(query)

성능 최적화

캐싱

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache
# LLM 응답 캐싱
set_llm_cache(InMemoryCache())
# 같은 질문 반복 시 캐시 사용 (비용 절감)
answer1 = qa_chain.run("C++ 메모리 관리")
answer2 = qa_chain.run("C++ 메모리 관리")  # 캐시에서 반환

배치 임베딩

# ❌ 비효율적: 하나씩 임베딩
for text in texts:
    embedding = get_embedding(text)
# ✅ 효율적: 배치 임베딩
embeddings = client.embeddings.create(
    model="text-embedding-3-small",
    input=texts  # 리스트 전체
).data
# 요청 1회에 여러 입력을 보내므로 왕복(RTT) 횟수와 레이트 리밋 소모가 줄어듦
# 단, 요청당 입력 개수·총 토큰 수 상한이 있어 큰 코퍼스는 적당한 크기로 나눠 보내야 함

인덱스 최적화

# Pinecone: 파티션 활용
index.upsert(
    vectors=vectors,
    namespace="cpp-docs"  # 네임스페이스로 분리
)
# 특정 네임스페이스만 검색
results = index.query(
    vector=query_embedding,
    top_k=3,
    namespace="cpp-docs"
)

청크 메타데이터 활용

# 메타데이터에 요약 추가
chunks_with_summary = []
for chunk in chunks:
    summary = llm.predict(f"다음 텍스트를 한 문장으로 요약:\n{chunk}")
    chunks_with_summary.append({
        "text": chunk,
        "summary": summary,
        "char_count": len(chunk)
    })
# 검색 시 요약으로 먼저 필터링

실무 사례

고객 지원 챗봇

요구사항:

제품 매뉴얼, FAQ, 정책 문서 기반 답변
실시간 업데이트 반영
출처 제공 구현:

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)
qa_chain = ConversationalRetrievalChain.from_llm(
    llm=ChatOpenAI(model="gpt-4"),
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    memory=memory,
    return_source_documents=True
)
# 대화
result = qa_chain({"question": "제품 보증 기간은?"})
print(result["answer"])
print("출처:", [doc.metadata["source"] for doc in result["source_documents"]])
# 후속 질문 (이전 대화 맥락 유지)
result = qa_chain({"question": "연장 가능한가요?"})

사내 문서 검색

요구사항:

수천 개의 내부 문서 검색
부서별, 프로젝트별 필터링
권한 관리 구현:

def search_with_permissions(query: str, user_id: str, department: str):
    # 사용자 권한에 따른 필터
    filter_dict = {
        "$or": [
            {"department": department},
            {"public": True}
        ]
    }
    
    docs = vectorstore.similarity_search(
        query,
        k=5,
        filter=filter_dict
    )
    
    return docs
# 사용 (user_id는 감사 로그 등에 활용하고, 필터 자체는 서버가 확정한 부서 값으로 구성)
docs = search_with_permissions(
    "2026년 예산 계획",
    user_id="user123",
    department="finance"
)

권한 필터는 사내 RAG에서 가장 사고가 나기 쉬운 부분입니다. 필터를 검색 이후에 적용하면(top-k를 먼저 뽑고 권한 없는 문서를 걸러내면) 결과가 k개보다 적게 남거나 0개가 되는 문제가 생기므로, 위처럼 벡터 DB 쿼리의 필터로 넣어 검색과 동시에 적용해야 합니다. 또 부서 값은 반드시 인증된 세션에서 서버가 꺼내야 하며, 사용자가 보낸 요청 파라미터나 LLM이 질문에서 추출한 값을 그대로 필터에 쓰면 “재무팀 문서도 같이 찾아줘” 같은 질문으로 권한을 우회할 수 있습니다. 아래 Self-Query 기법을 권한 필터와 함께 쓸 때 특히 주의해야 하는 지점입니다.

코드 검색 및 설명

요구사항:

대규모 코드베이스에서 관련 코드 찾기
함수 설명 자동 생성
API 사용 예제 제공 구현:

from langchain.document_loaders import GitLoader
# Git 저장소 로드
loader = GitLoader(
    repo_path="./my-repo",
    file_filter=lambda x: x.endswith(('.cpp', '.h', '.py'))
)
documents = loader.load()
# 함수 단위로 청크
def chunk_by_functions(code: str, language: str):
    # tree-sitter로 AST 파싱
    import tree_sitter
    # ... 함수 단위 추출
    return function_chunks
# 코드 검색
query = "HTTP 요청을 보내는 함수"
docs = vectorstore.similarity_search(query, k=3)
# LLM으로 설명 생성
for doc in docs:
    explanation = llm.predict(
        f"다음 코드를 설명하세요:\n\n{doc.page_content}"
    )
    print(explanation)

논문/연구 자료 QA

요구사항:

수백 페이지 PDF 논문 분석
수식, 그래프 처리
인용 추적 구현:

from langchain.document_loaders import PyPDFLoader
# PDF 로드
loader = PyPDFLoader("research_paper.pdf")
pages = loader.load()
# 페이지별 메타데이터 추가
for i, page in enumerate(pages):
    page.metadata["page"] = i + 1
    page.metadata["source"] = "research_paper.pdf"
# 벡터 DB 저장
vectorstore = Chroma.from_documents(pages, embeddings)
# 질문 (페이지 번호 포함)
result = qa_chain({"question": "Transformer 아키텍처의 핵심은?"})
print(result["answer"])
print("페이지:", [doc.metadata["page"] for doc in result["source_documents"]])

고급 기법

Query Expansion (질문 확장)

def expand_query(query: str) -> list[str]:
    """LLM으로 질문을 여러 형태로 확장"""
    prompt = f"""다음 질문을 3가지 다른 방식으로 다시 작성하세요:
    
    원본: {query}
    
    1.
    2.
    3.
    """
    
    response = llm.predict(prompt)
    expanded = [query] + response.strip().split('\n')
    return expanded
# 확장된 질문으로 검색
query = "C++ 성능 최적화"
expanded_queries = expand_query(query)
all_docs = []
for q in expanded_queries:
    docs = vectorstore.similarity_search(q, k=2)
    all_docs.extend(docs)
# 중복 제거 후 사용
unique_docs = list({doc.page_content: doc for doc in all_docs}.values())

HyDE (Hypothetical Document Embeddings)

def hyde_search(query: str, k: int = 3):
    """가상 답변을 생성하여 검색"""
    # 1. LLM으로 가상 답변 생성
    hypothetical_answer = llm.predict(
        f"다음 질문에 대한 답변을 작성하세요:\n{query}"
    )
    
    # 2. 가상 답변으로 검색 (질문이 아닌 답변으로 검색!)
    docs = vectorstore.similarity_search(hypothetical_answer, k=k)
    
    # 3. 실제 문서로 최종 답변 생성
    context = "\n\n".join([doc.page_content for doc in docs])
    final_answer = llm.predict(
        f"문서:\n{context}\n\n질문: {query}\n\n답변:"
    )
    
    return final_answer
# 사용
answer = hyde_search("C++ 컴파일 타임 최적화 방법")

Self-Query (자동 필터 추출)

from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo
# 메타데이터 스키마 정의
metadata_field_info = [
    AttributeInfo(
        name="category",
        description="문서 카테고리 (cpp, python, java 등)",
        type="string"
    ),
    AttributeInfo(
        name="level",
        description="난이도 (beginner, intermediate, advanced)",
        type="string"
    ),
    AttributeInfo(
        name="date",
        description="작성 날짜",
        type="date"
    )
]
retriever = SelfQueryRetriever.from_llm(
    llm=llm,
    vectorstore=vectorstore,
    document_contents="프로그래밍 튜토리얼 및 가이드",
    metadata_field_info=metadata_field_info
)
# 자연어 질문에서 자동으로 필터 추출
docs = retriever.get_relevant_documents(
    "2026년 3월에 작성된 고급 C++ 가이드를 찾아줘"
)
# 자동으로 filter={"category": "cpp", "level": "advanced", "date": "2026-03"} 생성

Parent Document Retriever (작게 찾고 크게 넘기기)

청크 크기에는 근본적인 긴장이 있습니다. 검색 정확도는 청크가 작을수록 좋아집니다. 한 청크에 주제가 하나만 담겨 임베딩이 질문과 선명하게 맞기 때문입니다. 반면 LLM이 답을 쓰려면 앞뒤 맥락이 필요해서, 200자짜리 조각만 넘기면 “그래서 그 설정이 어느 버전 기준인지” 같은 정보가 빠진 채 답하게 됩니다. Parent Document Retriever는 이 둘을 분리합니다. 문서를 큰 부모 청크(예: 2,000자)로 나누고, 각 부모를 다시 작은 자식 청크(예: 400자)로 나눠 자식만 임베딩해서 검색한 뒤, 검색된 자식이 속한 부모 청크를 LLM에 넘깁니다. 매뉴얼이나 정책 문서처럼 섹션 단위로 의미가 완결되는 자료에서 효과가 크고, 대신 컨텍스트에 들어가는 토큰이 늘어나므로 가져오는 부모 개수(k)를 함께 줄여야 합니다.

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain.text_splitter import RecursiveCharacterTextSplitter

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,                     # 자식 청크 임베딩 저장
    docstore=InMemoryStore(),                    # 부모 청크 원문 저장 (운영에선 영속 저장소)
    child_splitter=RecursiveCharacterTextSplitter(chunk_size=400),
    parent_splitter=RecursiveCharacterTextSplitter(chunk_size=2000),
)
retriever.add_documents(docs)
parents = retriever.invoke("환불 요청은 며칠 안에 해야 하나요?")

트러블슈팅

문제 1: 검색 결과가 관련 없음

원인:

청크 크기가 너무 작거나 큼
임베딩 모델이 도메인에 맞지 않음
쿼리가 너무 모호함 해결:

# 1. 청크 크기 조정
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1500,  # 1000에서 증가
    chunk_overlap=300  # 200에서 증가
)
# 2. 쿼리 확장
expanded_queries = expand_query(original_query)
# 3. 하이브리드 검색 사용
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.5, 0.5]
)

문제 2: LLM이 문서를 무시하고 일반 지식으로 답변

원인:

프롬프트가 약함
문서가 질문과 관련 없음 해결:

template = """당신은 제공된 문서만을 기반으로 답변하는 어시스턴트입니다.
문서에 정보가 없으면 "제공된 문서에서 해당 정보를 찾을 수 없습니다"라고 답변하세요.
절대로 문서 외의 지식을 사용하지 마세요.
문서:
{context}
질문: {question}
답변 (문서 기반):"""
# temperature를 0으로 설정 (창의성 최소화)
llm = ChatOpenAI(model="gpt-4", temperature=0)

문제 3: 토큰 한도 초과

원인:

검색된 문서가 너무 많거나 큼
컨텍스트 윈도우 초과 해결:

# 1. Map-Reduce 체인 사용
from langchain.chains.question_answering import load_qa_chain
chain = load_qa_chain(
    llm,
    chain_type="map_reduce"  # 각 문서 개별 처리 후 병합
)
# 2. Refine 체인 사용
chain = load_qa_chain(
    llm,
    chain_type="refine"  # 순차적으로 답변 개선
)
# 3. 검색 결과 제한
retriever = vectorstore.as_retriever(
    search_kwargs={"k": 3}  # 5에서 3으로 감소
)

문제 4: 느린 검색 속도

원인:

벡터 DB 인덱스 최적화 부족
임베딩 생성 병목 해결:

# 1. 인덱스 타입 변경 (Pinecone pod 기반 인덱스)
from pinecone import PodSpec
pc.create_index(
    name=index_name,
    dimension=1536,
    metric="cosine",
    spec=PodSpec(environment="us-east-1-aws", pod_type="p2.x1")  # 저지연 (비용 증가)
)
# 2. 로컬 캐시
import diskcache
cache = diskcache.Cache('./embedding_cache')
def get_embedding_cached(text: str):
    if text in cache:
        return cache[text]
    
    embedding = get_embedding(text)
    cache[text] = embedding
    return embedding
# 3. 비동기 처리
import asyncio
async def search_async(queries: list[str]):
    tasks = [vectorstore.asimilarity_search(q, k=3) for q in queries]
    results = await asyncio.gather(*tasks)
    return results

평가 지표

Retrieval 평가

Precision@K: 상위 K개 중 관련 문서 비율

def precision_at_k(retrieved_docs, relevant_docs, k):
    retrieved_k = retrieved_docs[:k]
    relevant_count = sum(1 for doc in retrieved_k if doc in relevant_docs)
    return relevant_count / k

Recall@K: 전체 관련 문서 중 검색된 비율

# 함수 정의 및 구현
def recall_at_k(retrieved_docs, relevant_docs, k):
    retrieved_k = retrieved_docs[:k]
    relevant_count = sum(1 for doc in retrieved_k if doc in relevant_docs)
    return relevant_count / len(relevant_docs)

MRR (Mean Reciprocal Rank): 첫 관련 문서의 순위

def mrr(retrieved_docs, relevant_docs):
    for i, doc in enumerate(retrieved_docs):
        if doc in relevant_docs:
            return 1 / (i + 1)
    return 0

Generation 평가

자동 평가:

from langchain.evaluation import load_evaluator
# 정확성 평가
evaluator = load_evaluator("qa")
result = evaluator.evaluate_strings(
    prediction=answer,
    reference=ground_truth,
    input=query
)
# 관련성 평가
evaluator = load_evaluator("criteria", criteria="relevance")
result = evaluator.evaluate_strings(
    prediction=answer,
    input=query
)

사람 평가:

정확성: 답변이 사실과 일치하는가?
완전성: 필요한 정보를 모두 포함하는가?
간결성: 불필요한 정보가 없는가?
출처: 출처가 정확한가?

비용 최적화

비용 구조

RAG 비용은 크게 임베딩 비용(문서 색인 시 한 번, 질문마다 한 번)과 생성 비용(질문 + 검색 청크 + 답변 토큰)으로 나뉩니다. OpenAI 임베딩은 text-embedding-3-small이 100만 토큰당 $0.02, -large가 $0.13 수준으로 저렴해서, 대부분의 서비스에서 비용의 대부분은 생성 단계에서 나옵니다. 질문마다 top-k 청크가 입력 토큰으로 붙기 때문에 k와 청크 크기가 곧 요청당 비용이 됩니다. 생성 모델 단가는 자주 바뀌므로 각 제공사의 최신 가격표를 기준으로 “평균 입력 토큰 × 일일 질문 수”를 계산해 보는 것이 가장 정확합니다.

비용 절감 전략

1. 임베딩 모델 선택

# 프로토타입: 작은 모델
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 프로덕션: 성능 필요 시
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
# 오픈소스: 비용 제로
from langchain.embeddings import HuggingFaceEmbeddings
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

2. LLM 모델 선택

# 간단한 질문: GPT-3.5-turbo
llm_simple = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 복잡한 질문: GPT-4
llm_complex = ChatOpenAI(model="gpt-4", temperature=0)
# 라우팅
def route_query(query: str):
    if is_simple_query(query):
        return qa_chain_simple.run(query)
    else:
        return qa_chain_complex.run(query)

3. 캐싱

from langchain.cache import SQLiteCache
from langchain.globals import set_llm_cache
set_llm_cache(SQLiteCache(database_path=".langchain.db"))
# 동일 질문 반복 시 API 호출 없음

4. 검색 결과 제한

# ❌ 비용 높음: 10개 문서 전달
retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
# ✅ 비용 절감: 3개만 전달
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

마무리

RAG는 LLM을 실무에 적용하는 핵심 기술입니다:

핵심 포인트:

검색 증강: 외부 지식을 프롬프트에 추가
실시간 업데이트: 문서 변경 즉시 반영
출처 제공: 답변의 신뢰성 향상
비용 효율: Fine-tuning보다 저렴 구현 단계:

문서 수집 및 청크 분할
임베딩 생성 및 벡터 DB 저장
검색 파이프라인 구축
LLM 통합 및 프롬프트 최적화
평가 및 개선 시작 가이드:

프로토타입: ChromaDB + OpenAI + LangChain
프로덕션: Pinecone + GPT-4 + 커스텀 파이프라인
엔터프라이즈: Weaviate + Azure OpenAI + 권한 관리 다음 학습:
Python 시리즈에서 Python 기초
알고리즘 시리즈에서 검색 알고리즘
데이터베이스 비교 참고 자료:
LangChain 문서
Pinecone 가이드
OpenAI Embeddings API

자주 묻는 질문 (FAQ)

Q. 검색된 문서를 넘겼는데도 LLM이 일반 지식으로 답하면 어떻게 해야 하나요?

A. 원인은 대개 프롬프트가 약하거나 검색된 문서 자체가 질문과 관련이 없는 경우입니다. 프롬프트에 제공된 문서만 근거로 답하고 정보가 없으면 찾을 수 없다고 말하라는 지시를 명시하고, temperature를 0으로 낮춰 창의적인 보충을 줄입니다. 그래도 계속되면 검색 단계를 먼저 점검해 관련 청크가 실제로 상위에 오는지 확인해야 합니다.

이 글의 핵심

들어가며

RAG란?

기본 개념

RAG가 해결하는 문제

RAG 아키텍처

전체 흐름

핵심 구성 요소

벡터 임베딩

임베딩이란?

주요 임베딩 모델

Python 예제

유사도 계산

벡터 데이터베이스

주요 벡터 DB 비교

Pinecone 예제

ChromaDB 예제 (로컬 개발)

청크 전략

청크 크기 결정

청크 분할 전략

오버랩 전략

검색 최적화

기본 검색 (Semantic Search)

메타데이터 필터링

MMR (Maximal Marginal Relevance)

Python 구현

기본 RAG 시스템

커스텀 프롬프트

출처 포함 답변

하이브리드 검색

Semantic + Keyword 검색

Reranking

성능 최적화

캐싱

배치 임베딩

인덱스 최적화

청크 메타데이터 활용

실무 사례

고객 지원 챗봇

사내 문서 검색

코드 검색 및 설명

논문/연구 자료 QA

고급 기법

Query Expansion (질문 확장)

HyDE (Hypothetical Document Embeddings)

Self-Query (자동 필터 추출)

Parent Document Retriever (작게 찾고 크게 넘기기)

트러블슈팅

문제 1: 검색 결과가 관련 없음

문제 2: LLM이 문서를 무시하고 일반 지식으로 답변

문제 3: 토큰 한도 초과

문제 4: 느린 검색 속도

평가 지표

Retrieval 평가

Generation 평가

비용 최적화

비용 구조

비용 절감 전략

마무리

자주 묻는 질문 (FAQ)

Q. 검색된 문서를 넘겼는데도 LLM이 일반 지식으로 답하면 어떻게 해야 하나요?

같이 보면 좋은 글