[LLMOps] Day 1: RAG 파이프라인의 이해 - 엔지니어링 관점에서 본 아키텍처

서론: LLM의 한계와 RAG의 등장

대규모 언어 모델(LLM)은 강력하지만, 근본적인 한계가 있다.

• 지식의 고정성: 학습 데이터의 시점에서 지식이 고정되어, 최신 정보를 알지 못한다.
• 환각(Hallucination): 모르는 내용을 그럴듯하게 지어낸다.
• 도메인 지식 부족: 기업 내부 문서나 특정 도메인의 전문 지식을 학습하지 못했다.

RAG(Retrieval-Augmented Generation)는 이 문제를 우아하게 해결한다. LLM에게 답변을 생성하기 전, 관련 문서를 검색해서 제공하는 것이다. 마치 시험 문제를 풀 때 참고서를 옆에 두는 것과 같다.

1. RAG 파이프라인의 구조

RAG는 크게 두 개의 독립적인 파이프라인으로 구성된다.

1.1 Indexing Pipeline (오프라인)

Document → Chunking → Embedding → Vector DB 저장

1. 문서 수집: PDF, HTML, Markdown 등 다양한 소스에서 문서를 수집
2. 청킹(Chunking): 문서를 작은 단위로 분할 (보통 200~1000 토큰)
3. 임베딩(Embedding): 텍스트를 벡터로 변환 (예: OpenAI ada-002, 1536차원)
4. 저장: 벡터 데이터베이스에 인덱싱 (Pinecone, Weaviate, Qdrant 등)

1.2 Retrieval Pipeline (온라인)

Query → Embedding → Vector Search → Reranking → LLM → Response

1. 쿼리 임베딩: 사용자 질문을 같은 임베딩 모델로 벡터화
2. 유사도 검색: 벡터 DB에서 코사인 유사도 기반 Top-K 검색
3. 리랭킹(선택): 검색 결과를 재정렬하여 정확도 향상
4. 프롬프트 구성: 검색된 문서를 컨텍스트로 LLM에 전달
5. 응답 생성: LLM이 컨텍스트 기반으로 답변 생성

2. 최소 구현: Python으로 보는 RAG 파이프라인

복잡한 프레임워크 없이, 핵심 개념만으로 구현해보자.

2.1 의존성 설치

pip install openai chromadb langchain-text-splitters

2.2 Indexing Pipeline 구현

import chromadb
from openai import OpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 1. 문서 청킹
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", " ", ""]
)

documents = [
    "Rust는 메모리 안전성을 보장하는 시스템 프로그래밍 언어이다.",
    "PyO3를 사용하면 Rust로 Python 확장을 쉽게 작성할 수 있다."
]

chunks = []
for doc in documents:
    chunks.extend(text_splitter.split_text(doc))

# 2. 임베딩 생성
client = OpenAI()
embeddings = []
for chunk in chunks:
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=chunk
    )
    embeddings.append(response.data[0].embedding)

# 3. 벡터 DB 저장
chroma_client = chromadb.Client()
collection = chroma_client.create_collection("tech_docs")
collection.add(
    embeddings=embeddings,
    documents=chunks,
    ids=[f"doc_{i}" for i in range(len(chunks))]
)

2.3 Retrieval Pipeline 구현

def rag_query(question: str, top_k: int = 3):
    # 1. 쿼리 임베딩
    query_embedding = client.embeddings.create(
        model="text-embedding-3-small",
        input=question
    ).data[0].embedding

    # 2. 벡터 검색
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k
    )

    # 3. 컨텍스트 구성
    context = "\n\n".join(results['documents'][0])

    # 4. LLM 호출
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "다음 컨텍스트를 기반으로 답변하세요."},
            {"role": "user", "content": f"컨텍스트:\n{context}\n\n질문: {question}"}
        ]
    )

    return response.choices[0].message.content

# 사용 예시
answer = rag_query("Rust로 Python 확장을 만들 수 있나요?")
print(answer)

이 코드는 약 50줄로 동작하는 RAG 시스템이다. 프로덕션에서는 더 많은 고려사항이 있지만, 핵심 원리는 동일하다.

3. 엔지니어링 관점의 설계 결정

3.1 동기 vs 비동기 처리

# 동기 처리 (Simple but Slow)
for doc in documents:
    embedding = get_embedding(doc)  # 200ms
    store(embedding)                # 50ms
# Total: 250ms × N

# 비동기 처리 (Complex but Fast)
import asyncio

async def process_batch(batch):
    tasks = [get_embedding_async(doc) for doc in batch]
    embeddings = await asyncio.gather(*tasks)
    await store_batch(embeddings)

# Total: 200ms + 50ms (병렬 처리)

Trade-off: 문서가 1,000개 이하라면 동기 처리로 충분하다. 하지만 10,000개 이상이라면 비동기 배치 처리가 필수다.

3.2 청킹 전략의 성능 영향

전략	장점	단점
고정 크기 (500토큰)	구현 간단, 속도 빠름	문맥 단절 가능
문단 기반	의미 보존	크기 불균형
의미 기반 (Semantic)	정확도 최고	처리 시간 10배 증가

추천: 첫 구현은 고정 크기로 시작하고, 정확도가 부족하면 하이브리드 전략으로 전환한다.

4. 다음 단계: 프로덕션으로 가는 길

현재 구현은 프로토타입 수준이다. 프로덕션에서는 다음 문제를 해결해야 한다.

1. 확장성: 수백만 개의 문서를 어떻게 인덱싱할 것인가?
2. 레이턴시: 99 percentile 응답 시간을 300ms 이하로 유지하려면?
3. 정확도: 검색 품질을 정량적으로 측정하고 개선하려면?
4. 비용: 임베딩 API 호출 비용을 줄이려면?

이어지는 시리즈에서 각 문제의 실전 해결책을 다룰 것이다.

---