
RAG 파이프라인 구축: 벡터 DB + LLM으로 문서 검색
LLM은 학습 데이터 밖의 지식을 모른다. RAG가 이 문제를 어떻게 해결하는지, 문서 수집부터 청킹, 임베딩, 벡터 저장, 검색, 생성까지 전체 파이프라인을 Python과 TypeScript 예제로 구축한다.

LLM은 학습 데이터 밖의 지식을 모른다. RAG가 이 문제를 어떻게 해결하는지, 문서 수집부터 청킹, 임베딩, 벡터 저장, 검색, 생성까지 전체 파이프라인을 Python과 TypeScript 예제로 구축한다.
AI Agent를 만들 때 반복적으로 등장하는 핵심 패턴 세 가지—Tool Use, ReAct, Chain of Thought—를 실제 TypeScript 코드와 함께 정리했다. 이 패턴을 이해하면 Agent 설계가 훨씬 명확해진다.

ChatGPT는 질문에 답하지만, AI Agent는 스스로 계획하고 도구를 사용해 작업을 완료한다. 이 차이가 왜 중요한지 정리했다.

둘 다 같은 Transformer 자식인데 왜 다를까? '빈칸 채우기'와 '이어 쓰기' 비유로 알아보는 BERT와 GPT의 결정적 차이. 프로젝트에서 겪은 시행착오와 선택 가이드.

단어와 문장을 숫자 벡터로 바꾸면 '의미'를 수학으로 계산할 수 있다. 코사인 유사도, ANN 알고리즘, OpenAI 임베딩 API까지 원리부터 실전까지 한번에 정리했다.

ChatGPT나 Claude에게 회사 내부 문서에 대해 물어보면 어떻게 될까? 아마 이런 답이 돌아올 것이다.
"저는 그 정보에 접근할 수 없습니다."
또는 더 나쁜 경우:
"네, OOO 회사의 내부 정책은 다음과 같습니다: ..." (그리고 완전히 틀린 정보를 자신 있게 말함)
LLM에는 두 가지 결정적 한계가 있다.
1. Knowledge Cutoff (지식 한계): LLM은 학습 시점 이후의 정보를 모른다. GPT-4o의 학습 데이터는 2023년까지다. 2024년에 발표된 신기술이나 최근 뉴스는 모른다.
2. Hallucination (환각): 모르는 걸 물어보면 그럴듯한 거짓말을 한다. 자신이 틀렸다는 걸 인식하지 못하면서 확신에 차서 말한다. 이게 LLM을 기업에서 바로 쓰기 어렵게 만드는 주범이다.
RAG(Retrieval-Augmented Generation)는 이 두 문제를 동시에 해결한다.
아이디어는 간단하다. LLM에게 질문할 때, 관련 문서를 먼저 검색해서 "참고 자료"로 함께 넘겨주는 것이다. LLM은 자기 기억에서 답을 만들어내는 게 아니라, 주어진 문서를 바탕으로 답한다.
RAG 파이프라인은 크게 두 단계로 나뉜다.
[Ingestion Pipeline — 오프라인]
문서 수집 → 청킹 → 임베딩 → 벡터 DB 저장
[Query Pipeline — 온라인]
사용자 질문 → 질문 임베딩 → 벡터 DB 검색 → LLM 생성
Raw Documents (PDF, DOCX, Web, DB)
│
▼
[Document Loader] ← 파일 포맷별 파서
│
▼
[Text Splitter] ← 청킹 전략 (크기, 오버랩)
│
▼
[Embedding Model] ← text-embedding-3-small, BGE, etc.
│
▼
[Vector Store] ← Pinecone, pgvector, Weaviate, Chroma
User Question
│
▼
[Embedding Model] ← 질문을 벡터로
│
▼
[Vector Store] ← 유사 벡터 검색 (Top-K)
│
▼
[Retrieved Chunks] ← 관련 문서 조각들
│
▼
[Prompt Assembly] ← "다음 문서를 참고해서 답하시오: ..."
│
▼
[LLM] ← GPT-4, Claude, Gemini
│
▼
[Generated Answer] ← 근거 있는 답변
RAG를 이해하려면 임베딩(Embedding)부터 이해해야 한다.
임베딩은 텍스트를 숫자 배열(벡터)로 변환하는 것이다. 의미가 비슷한 텍스트는 비슷한 벡터 값을 갖는다.
# 개념적으로
embed("고양이는 귀엽다") → [0.2, 0.8, 0.1, ..., 0.4] # 1536차원
embed("猫はかわいい") → [0.21, 0.79, 0.11, ..., 0.41] # 비슷!
embed("투자 위험 고지") → [0.9, 0.1, 0.7, ..., 0.2] # 다름
벡터 간의 거리(코사인 유사도, 유클리드 거리 등)를 계산해서 "얼마나 비슷한 의미인지" 수치로 표현할 수 있다.
| 모델 | 제공사 | 차원 | 특징 |
|---|---|---|---|
| text-embedding-3-small | OpenAI | 1536 | 가성비 좋음, 다국어 |
| text-embedding-3-large | OpenAI | 3072 | 고품질, 비쌈 |
| text-embedding-ada-002 | OpenAI | 1536 | 구버전, 여전히 많이 쓰임 |
| BGE-M3 | BAAI | 1024 | 오픈소스, 다국어 최강 |
| E5-large | Microsoft | 1024 | 오픈소스, 좋은 성능 |
| Gemini Embedding | 768 | Gemini 생태계 |
한국어 문서를 많이 다룬다면 BGE-M3를 적극 추천한다. 한국어, 영어, 중국어 등 100개 이상 언어에서 강력한 성능을 보인다.
문서를 벡터 DB에 넣기 전에 적절한 크기로 잘라야 한다. 이걸 청킹(Chunking)이라고 한다.
왜 잘라야 하냐? LLM의 컨텍스트 윈도우는 제한적이고, 관련 없는 내용이 많이 들어가면 검색 품질도 떨어지기 때문이다.
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 토큰 또는 문자 수
chunk_overlap=50, # 청크 간 중복 (문맥 유지)
separators=["\n\n", "\n", ".", " "] # 우선순위대로 분리
)
가장 단순하고 빠르다. 하지만 문장 중간에 잘릴 수 있다.
2. Semantic Chunking (의미 기반)# 임베딩을 활용해서 의미가 바뀌는 지점에서 분리
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings
splitter = SemanticChunker(
OpenAIEmbeddings(),
breakpoint_threshold_type="percentile",
breakpoint_threshold_amount=95,
)
품질이 좋지만 임베딩 API 호출 비용이 든다.
3. Document-Aware Chunking (문서 구조 인식) 마크다운 헤딩, HTML 태그, PDF 섹션 등 문서 구조를 이해해서 분리한다.
from langchain.text_splitter import MarkdownHeaderTextSplitter
headers_to_split_on = [
("#", "Header 1"),
("##", "Header 2"),
("###", "Header 3"),
]
splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=headers_to_split_on
)
| 사용 사례 | 추천 청크 크기 | 이유 |
|---|---|---|
| FAQ/짧은 Q&A | 100~200 토큰 | 짧은 질문-답변 쌍 |
| 기술 문서 | 300~500 토큰 | 섹션 단위 이해 |
| 법률/계약 문서 | 500~800 토큰 | 문맥이 중요 |
| 코드 | 함수/클래스 단위 | 논리적 단위로 분리 |
오버랩(overlap)은 청크 크기의 10~20%를 권장한다. 청크 경계에서 잘린 문맥이 다음 청크에서 이어지도록.
관리형 서비스. 설정이 간단하고 안정적이다. 스타트업에서 빠르게 시작하기 좋다.
from pinecone import Pinecone, ServerlessSpec
pc = Pinecone(api_key="your-api-key")
# 인덱스 생성
pc.create_index(
name="documents",
dimension=1536, # text-embedding-3-small 차원
metric="cosine",
spec=ServerlessSpec(cloud="aws", region="us-east-1")
)
index = pc.Index("documents")
# 벡터 삽입
index.upsert(vectors=[
{
"id": "doc-1-chunk-0",
"values": embedding_vector, # List[float]
"metadata": {
"text": "청크 원문",
"source": "document.pdf",
"page": 1
}
}
])
# 검색
results = index.query(
vector=query_embedding,
top_k=5,
include_metadata=True
)
이미 PostgreSQL을 쓰고 있다면 가장 현실적인 선택이다. 별도 인프라 없이 기존 DB에 벡터 기능을 추가한다.
-- 확장 설치
CREATE EXTENSION IF NOT EXISTS vector;
-- 테이블 생성
CREATE TABLE document_chunks (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
document_id UUID REFERENCES documents(id),
content TEXT NOT NULL,
embedding VECTOR(1536), -- 차원 수
metadata JSONB,
created_at TIMESTAMP DEFAULT NOW()
);
-- 인덱스 생성 (HNSW 알고리즘 — 빠른 근사 검색)
CREATE INDEX ON document_chunks
USING hnsw (embedding vector_cosine_ops);
-- 유사도 검색 쿼리
SELECT
id,
content,
metadata,
1 - (embedding <=> $1::vector) AS similarity
FROM document_chunks
ORDER BY embedding <=> $1::vector
LIMIT 5;
// TypeScript + Supabase (pgvector)
const { data } = await supabase.rpc('match_documents', {
query_embedding: queryEmbedding,
match_threshold: 0.7,
match_count: 5,
});
// Supabase에서 제공하는 RPC 함수
// CREATE OR REPLACE FUNCTION match_documents(...)
오픈소스 벡터 DB. 자체 호스팅 또는 클라우드. 멀티모달 지원(텍스트+이미지)이 강점이다.
로컬 개발/프로토타이핑에 최고다. 인메모리 또는 로컬 파일 저장. 설치가 pip install chromadb 하나로 끝난다.
import chromadb
from chromadb.utils import embedding_functions
client = chromadb.Client() # 인메모리 (개발용)
# client = chromadb.PersistentClient(path="./chroma_db") # 영구 저장
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
api_key="your-api-key",
model_name="text-embedding-3-small"
)
collection = client.create_collection(
name="documents",
embedding_function=openai_ef
)
# 문서 추가 — 임베딩은 자동으로 생성됨
collection.add(
ids=["chunk-1", "chunk-2"],
documents=["청크 1 내용", "청크 2 내용"],
metadatas=[{"source": "doc.pdf"}, {"source": "doc.pdf"}]
)
# 검색
results = collection.query(
query_texts=["사용자 질문"],
n_results=5
)
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.document_loaders import PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
# 1. 문서 로드
loader = DirectoryLoader(
"./docs",
glob="**/*.pdf",
loader_cls=PyPDFLoader
)
documents = loader.load()
# 2. 청킹
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
)
chunks = splitter.split_documents(documents)
print(f"총 {len(chunks)}개 청크 생성")
# 3. 임베딩 + 벡터 DB 저장
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db"
)
# 4. 검색 설정
retriever = vectorstore.as_retriever(
search_type="mmr", # MMR: 다양성 보장 검색
search_kwargs={
"k": 5, # Top-5 검색
"fetch_k": 20, # MMR 후보 20개에서 5개 선택
}
)
# 5. 프롬프트 템플릿
prompt_template = """
다음 문서를 참고하여 질문에 답하세요.
문서에 없는 내용은 "문서에 해당 정보가 없습니다"라고 답하세요.
[참고 문서]
{context}
[질문]
{question}
[답변]
"""
PROMPT = PromptTemplate(
template=prompt_template,
input_variables=["context", "question"]
)
# 6. RAG 체인 구성
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff", # 검색된 문서를 그냥 이어붙임
retriever=retriever,
chain_type_kwargs={"prompt": PROMPT},
return_source_documents=True, # 출처 반환
)
# 7. 쿼리 실행
result = qa_chain.invoke({"query": "환불 정책은 어떻게 되나요?"})
print(result["result"])
print("\n[출처]")
for doc in result["source_documents"]:
print(f"- {doc.metadata['source']}, p.{doc.metadata.get('page', '?')}")
import { openai } from '@ai-sdk/openai';
import { embed, generateText } from 'ai';
import { createClient } from '@supabase/supabase-js';
const supabase = createClient(
process.env.SUPABASE_URL!,
process.env.SUPABASE_SERVICE_ROLE_KEY!
);
// 인제스천: 문서 청크를 DB에 저장
async function ingestDocument(
content: string,
metadata: Record<string, unknown>
): Promise<void> {
// 청킹 (단순 구현)
const chunks = splitIntoChunks(content, 500, 50);
for (const chunk of chunks) {
// 임베딩 생성
const { embedding } = await embed({
model: openai.embedding('text-embedding-3-small'),
value: chunk,
});
// pgvector에 저장
await supabase.from('document_chunks').insert({
content: chunk,
embedding: JSON.stringify(embedding),
metadata,
});
}
}
// 검색: 유사 청크 찾기
async function retrieveRelevantChunks(
query: string,
limit = 5
): Promise<Array<{ content: string; metadata: unknown; similarity: number }>> {
const { embedding: queryEmbedding } = await embed({
model: openai.embedding('text-embedding-3-small'),
value: query,
});
const { data, error } = await supabase.rpc('match_document_chunks', {
query_embedding: queryEmbedding,
match_threshold: 0.7,
match_count: limit,
});
if (error) throw error;
return data;
}
// RAG: 검색 + 생성
async function answerQuestion(question: string): Promise<string> {
// 1. 관련 청크 검색
const chunks = await retrieveRelevantChunks(question);
if (chunks.length === 0) {
return "관련 문서를 찾을 수 없습니다.";
}
// 2. 컨텍스트 조합
const context = chunks
.map((c, i) => `[문서 ${i + 1}]\n${c.content}`)
.join('\n\n');
// 3. LLM으로 답변 생성
const { text } = await generateText({
model: openai('gpt-4o-mini'),
system: `당신은 주어진 문서를 기반으로만 질문에 답하는 어시스턴트입니다.
문서에 없는 내용은 절대 추측하지 마세요.`,
prompt: `[참고 문서]\n${context}\n\n[질문]\n${question}`,
});
return text;
}
// 청킹 헬퍼
function splitIntoChunks(
text: string,
chunkSize: number,
overlap: number
): string[] {
const chunks: string[] = [];
let start = 0;
while (start < text.length) {
const end = Math.min(start + chunkSize, text.length);
chunks.push(text.slice(start, end));
start += chunkSize - overlap;
}
return chunks;
}
기본 RAG를 구축했다면, 이제 품질을 올릴 차례다.
벡터 검색(의미 검색) + 키워드 검색(BM25)을 결합한다. "RAG"라는 특정 용어처럼 키워드가 중요할 때는 BM25가, "변환기를 이용한 생성 모델"처럼 의미로 물을 때는 벡터 검색이 강하다.
from langchain.retrievers import EnsembleRetriever
from langchain.retrievers import BM25Retriever
# 벡터 검색기
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# BM25 키워드 검색기
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5
# 앙상블 (가중치: 벡터 60%, BM25 40%)
ensemble_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.6, 0.4]
)
사용자 질문을 LLM으로 재작성하거나 여러 버전으로 확장해서 더 많은 관련 문서를 찾는다.
from langchain.retrievers.multi_query import MultiQueryRetriever
# GPT가 질문을 3가지 버전으로 재작성
multi_query_retriever = MultiQueryRetriever.from_llm(
retriever=vectorstore.as_retriever(),
llm=ChatOpenAI(model="gpt-4o-mini")
)
# "환불 방법" → ["환불 절차는?", "반품 정책", "구매 취소 방법"]
벡터 검색 결과를 Cross-Encoder로 재순위화한다. 초기 검색보다 훨씬 정확하지만 느리다.
from langchain.retrievers.contextual_compression import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
# Cohere Rerank 모델 사용
compressor = CohereRerank(model="rerank-multilingual-v3.0")
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})
)
# 20개 검색 후 상위 5개로 재순위
RAG 파이프라인 품질을 어떻게 측정하냐?
RAGAS 프레임워크를 많이 사용한다. 4가지 지표를 자동으로 평가한다.
| 지표 | 의미 | 목표 |
|---|---|---|
| Faithfulness | 답변이 검색된 문서에 근거하는가? | 높을수록 좋음 |
| Answer Relevancy | 답변이 질문에 관련 있는가? | 높을수록 좋음 |
| Context Recall | 정답을 위한 문서가 검색됐는가? | 높을수록 좋음 |
| Context Precision | 검색된 문서가 실제로 유용한가? | 높을수록 좋음 |
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_recall
# 테스트 데이터셋
dataset = {
"question": ["환불 기간은?"],
"answer": ["14일 이내"],
"contexts": [["구매 후 14일 이내 환불 가능..."]],
"ground_truth": ["14일 이내 환불 가능합니다"]
}
result = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_recall])
print(result)
RAG는 LLM의 지식 한계와 환각 문제를 해결하는 현재 가장 실용적인 방법이다.
파이프라인을 단계별로 정리하면:
가장 흔한 실수는 "검색 품질"을 무시하는 것이다. LLM이 아무리 좋아도, 잘못된 문서가 들어가면 잘못된 답이 나온다. Garbage In, Garbage Out은 RAG에서도 유효하다. 청킹 전략과 검색 품질에 시간을 투자해라.