RAG 파이프라인 처음부터 끝까지

RAG를 처음 짜면 흔히 색인과 질의를 한 스크립트에 섞는다. 프로토타입에서는 괜찮지만, 프로덕션에서는 두 경로를 명확히 분리해야 한다. 이유는 단순하다.

• 색인(indexing)은 무겁고 가끔 돈다. 문서가 추가/수정될 때만 실행. 임베딩 비용·시간이 여기 집중된다.
• 질의(query)는 가볍고 매번 돈다. 사용자 한 번 물을 때마다 실행. 레이턴시가 곧 UX.

둘을 섞으면 "질문 한 번에 전체 문서를 다시 임베딩"하는 참사가 난다(실제로 자주 본다).

단계	색인 시점	질의 시점
로딩	O	X
청킹	O	X
임베딩	O (문서)	O (질문만)
인덱싱	O	X
검색	X	O
리랭킹	X	O
생성	X	O

임베딩 모델은 양쪽에서 반드시 동일해야 한다는 점이 함정이다. 색인할 때 text-embedding-3-large로 임베딩하고 질의할 때 다른 모델로 임베딩하면, 두 벡터가 같은 공간에 있지 않아 검색이 무의미해진다. 임베딩 모델 버전을 인덱스 메타데이터에 박아 두고, 질의 시점에 검증하는 것이 안전하다.

# 인덱스 메타에 임베딩 모델을 고정해 둔다
INDEX_META = {
    "embedding_model": "text-embedding-3-large",
    "embedding_dim": 3072,
    "chunk_strategy": "recursive-512-64",
    "created_at": "2026-06-15",
}

def assert_query_compat(query_model: str):
    assert query_model == INDEX_META["embedding_model"], (
        f"임베딩 모델 불일치: 인덱스={INDEX_META['embedding_model']}, "
        f"질의={query_model}. 인덱스를 재생성하거나 같은 모델을 쓰세요."
    )

임베딩 모델을 바꾸면 인덱스를 통째로 재생성해야 한다. 차원이 다르면 애초에 안 들어가지만, 차원이 같아도(예: 두 모델 모두 1024차원) 공간이 달라 조용히 망가진다. 이게 RAG에서 가장 흔한 "왜 갑자기 검색이 엉망이지"의 원인이다.

청킹 전에 문서를 깨끗한 텍스트로 만드는 단계다. 여기서 대충 하면 뒤 단계가 전부 오염된다.

포맷별 주의점

• PDF: 가장 골치 아프다. 2단 레이아웃, 표, 머리말/꼬리말, 페이지 번호가 텍스트에 섞인다. pypdf는 빠르지만 레이아웃을 자주 깨뜨린다. pdfplumber나 unstructured가 표/레이아웃을 더 잘 보존한다. 스캔 PDF(이미지)는 OCR이 필요하다.
• HTML: <nav>, <footer>, 광고, 사이드바를 제거해야 한다. trafilatura나 readability-lxml로 본문만 추출.
• Markdown: 비교적 깨끗하지만 코드 블록과 표를 청킹 단계에서 쪼개지 않도록 표시해 둬야 한다.

import re
import pdfplumber

def load_pdf(path: str) -> list[dict]:
    """페이지 단위로 텍스트 추출 + 페이지 번호를 메타로 보존."""
    pages = []
    with pdfplumber.open(path) as pdf:
        for i, page in enumerate(pdf.pages):
            text = page.extract_text() or ""
            text = clean_text(text)
            if text.strip():
                pages.append({"text": text, "page": i + 1, "source": path})
    return pages

def clean_text(text: str) -> str:
    # 하이픈으로 줄바꿈된 단어 복원: "inter-\nnal" -> "internal"
    text = re.sub(r"(\w)-\n(\w)", r"\1\2", text)
    # 단일 줄바꿈은 공백으로(문단 내), 이중 줄바꿈은 보존(문단 구분)
    text = re.sub(r"(?<!\n)\n(?!\n)", " ", text)
    # 반복 공백 정리
    text = re.sub(r"[ \t]+", " ", text)
    # 흔한 머리말/꼬리말 패턴 제거(예시 — 실제 문서에 맞게 조정)
    text = re.sub(r"\n?Page \d+ of \d+\n?", "", text)
    return text.strip()

메타데이터를 이 단계에서 확보하라. source(파일/URL), page, section, title, last_modified 같은 필드는 나중에 인용 표시, 필터링(예: "최근 1년 문서만"), 디버깅에 전부 쓰인다. 청킹 후에는 복원하기 어렵다.

함정: PDF에서 하이픈 줄바꿈(inter-\nnal)을 복원하지 않으면 internal이 inter와 nal로 토큰화되어 임베딩과 BM25 모두에서 매칭이 깨진다. 한국어 PDF는 줄바꿈이 공백 없이 단어 중간을 끊는 경우가 많아 더 조심해야 한다.

청킹은 RAG에서 가장 과소평가되는 단계다. "512 토큰으로 자르고 64 겹치기"를 기본값으로 쓰는 사람이 많은데, 문서 종류에 따라 최적이 크게 다르다.

왜 청크를 나누는가

1. 임베딩 모델은 입력 길이 제한이 있다(보통 8K 토큰). 긴 문서는 한 벡터로 표현할 수 없다.
2. 한 벡터가 표현하는 의미는 "평균"이다. 너무 길면 여러 주제가 섞여 검색 신호가 흐려진다.
3. LLM 컨텍스트와 비용 — 관련 조각만 넣어야 한다.

크기의 트레이드오프

청크 크기	장점	단점
작음 (128~256토큰)	정밀한 매칭, 핀포인트 검색	문맥 부족, 답변에 필요한 정보가 잘림
중간 (400~600토큰)	균형 — 대부분의 기본값	—
큼 (1000+토큰)	풍부한 문맥	검색 신호 희석, 무관 내용 다수 포함

기본 권장: 일반 산문은 400~600토큰 청크 + 10~15% 겹침. 겹침(overlap)은 청크 경계에서 문장이 잘려 정보가 손실되는 걸 막는다.

재귀적 분할(recursive splitting) — 가장 실용적인 기본 전략. 큰 구분자(\n\n)부터 시도하고, 청크가 너무 크면 점점 작은 구분자(\n, . , )로 내려간다. 의미 단위(문단 → 문장 → 단어)를 최대한 보존한다.

import tiktoken

enc = tiktoken.get_encoding("cl100k_base")  # 토큰 길이 측정용(임베딩 모델용 근사)

def token_len(text: str) -> int:
    return len(enc.encode(text))

def recursive_split(text: str, max_tokens=512, overlap=64,
                    separators=("\n\n", "\n", ". ", " ")) -> list[str]:
    """의미 단위를 보존하며 max_tokens 이하로 분할."""
    if token_len(text) <= max_tokens:
        return [text]
    sep = separators[0]
    rest = separators[1:] or (" ",)
    parts = text.split(sep)
    chunks, current = [], ""
    for part in parts:
        candidate = (current + sep + part) if current else part
        if token_len(candidate) <= max_tokens:
            current = candidate
        else:
            if current:
                chunks.append(current)
            # 한 part 자체가 너무 크면 더 작은 구분자로 재귀
            if token_len(part) > max_tokens:
                chunks.extend(recursive_split(part, max_tokens, overlap, rest))
                current = ""
            else:
                current = part
    if current:
        chunks.append(current)
    return _add_overlap(chunks, overlap)

def _add_overlap(chunks: list[str], overlap: int) -> list[str]:
    """각 청크 앞에 이전 청크의 마지막 overlap 토큰을 붙인다."""
    if overlap <= 0 or len(chunks) <= 1:
        return chunks
    out = [chunks[0]]
    for i in range(1, len(chunks)):
        prev_tokens = enc.encode(chunks[i - 1])[-overlap:]
        prefix = enc.decode(prev_tokens)
        out.append(prefix + " " + chunks[i])
    return out

구조를 깨지 마라. 코드 블록, 표, 마크다운 리스트는 중간에 자르면 의미가 망가진다. 이런 요소는 통째로 한 청크에 넣거나(크기를 초과하더라도), 표는 행 단위로 자르되 헤더를 각 청크에 반복해 넣는 식으로 처리한다.

메타데이터 청킹 트릭: 각 청크 앞에 "문서 제목 + 섹션 제목"을 붙여 임베딩하면 검색 정확도가 올라간다. "환불 정책"이라는 섹션의 청크에 본문만 있으면 "환불"이라는 단어가 없을 수도 있는데, 섹션 제목을 붙이면 신호가 명확해진다.

def contextualize(chunk: str, doc_title: str, section: str) -> str:
    return f"[문서: {doc_title} | 섹션: {section}]\n{chunk}"

함정: 겹침을 너무 크게 주면(예: 50%) 인덱스 크기가 2배가 되고 중복 청크가 검색 결과를 차지해 다양성이 떨어진다. 10~20%가 적정선이다.

임베딩은 텍스트를 고정 길이 벡터로 바꾼다. 의미가 비슷한 텍스트는 벡터 공간에서 가깝다. 모델 선택이 검색 품질의 상한을 결정한다.

선택 기준

• 차원(dimension): 클수록 표현력이 높지만 저장·검색 비용이 늘어난다. 1024~3072이 일반적. 일부 모델(예: OpenAI text-embedding-3-*)은 차원을 줄여서(Matryoshka) 쓸 수 있다 — 정확도를 약간 희생하고 비용을 크게 줄인다.
• 다국어 지원: 한국어가 핵심이면 다국어 모델이 필수다. 영어 전용 모델에 한국어를 넣으면 성능이 급락한다. BGE-M3, Cohere embed-multilingual-v3, OpenAI text-embedding-3-large 등이 한국어를 합리적으로 처리한다.
• 도메인 적합성: 법률·의료·코드처럼 특수 도메인은 범용 임베딩이 약할 수 있다. 도메인 평가셋으로 후보 모델 2~3개를 직접 비교하라(7번 섹션 평가 참고).
• 오픈소스 vs API: 오픈소스(BGE, E5, GTE)는 로컬에서 돌려 데이터를 외부로 안 보낸다. API(OpenAI, Cohere, Voyage)는 운영 부담이 적다.

비대칭 임베딩 주의: 일부 모델은 "질의"와 "문서"에 다른 프리픽스를 요구한다. 예를 들어 E5 계열은 문서에 passage: , 질의에 query: 를 붙여야 제 성능이 난다. 이걸 빼먹으면 성능이 조용히 떨어진다.

from openai import OpenAI

client = OpenAI()
EMBED_MODEL = "text-embedding-3-large"

def embed_batch(texts: list[str], batch_size=128) -> list[list[float]]:
    """문서를 배치로 임베딩. API 호출 횟수를 줄여 비용·시간 절감."""
    vectors = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        resp = client.embeddings.create(model=EMBED_MODEL, input=batch)
        # resp.data는 입력 순서를 보존한다
        vectors.extend([d.embedding for d in resp.data])
    return vectors

# E5 계열을 쓴다면 프리픽스를 반드시 구분
# 문서: embed_batch([f"passage: {c}" for c in chunks])
# 질의: embed_batch([f"query: {q}"])

배치 임베딩은 필수다. 청크 하나씩 API를 부르면 네트워크 왕복으로 수십 배 느려진다. 배치 크기는 모델 토큰 제한 안에서 최대로(보통 100~2048개). 색인은 한 번 실행이지만 문서가 수만 건이면 배치 여부가 분 단위 vs 시간 단위를 가른다.

정규화: 코사인 유사도를 쓸 거면 벡터를 L2 정규화해 두면 내적(dot product)으로 코사인을 계산할 수 있어 빠르다. 많은 벡터 DB가 이미 처리하지만, 직접 numpy로 검색한다면 확인하라.

import numpy as np

def l2_normalize(vectors: np.ndarray) -> np.ndarray:
    norms = np.linalg.norm(vectors, axis=1, keepdims=True)
    return vectors / np.clip(norms, 1e-12, None)  # 0 division 방지

함정: 빈 문자열이나 공백만 있는 청크를 임베딩하면 일부 API가 에러를 내거나 무의미한 벡터를 반환한다. 임베딩 전에 빈 청크를 거르고, 토큰 길이가 모델 한계를 넘는 청크는 잘라야 한다(8K 초과 시 대부분 에러).

임베딩한 벡터를 검색 가능한 인덱스에 넣는다. 그리고 — 이게 핵심인데 — 벡터 검색만 쓰지 마라. 키워드 검색(BM25)과 합친 하이브리드가 거의 항상 더 낫다.

왜 하이브리드인가

• 벡터(dense) 검색은 의미가 비슷한 걸 잘 찾는다. "비밀번호 재설정"으로 "패스워드 변경"을 찾아낸다. 하지만 정확한 키워드·고유명사·코드·제품번호·약어에는 약하다("ERR_429", "SKU-7781" 같은 건 의미가 없으니 벡터가 못 잡는다).
• BM25(sparse) 검색은 정확한 단어 매칭에 강하다. 드문 키워드·식별자에 강하고, 의미 일반화에는 약하다.

둘을 합치면 서로의 약점을 메운다. 정답이 둘 중 하나에는 잡힐 확률이 올라간다.

RRF(Reciprocal Rank Fusion) — 두 검색 결과의 점수 스케일이 다르므로(코사인 0~1 vs BM25 0~수십) 점수를 직접 더하면 안 된다. 대신 순위를 융합한다. 각 결과 리스트에서의 순위만 쓰므로 스케일 문제가 없다.

from collections import defaultdict

def reciprocal_rank_fusion(result_lists: list[list[str]], k: int = 60) -> list[tuple[str, float]]:
    """여러 검색 결과(문서 ID 순위 리스트)를 RRF로 융합.
    result_lists: [[벡터검색 결과 id들(순위순)], [BM25 결과 id들(순위순)]]
    k: 완충 상수(보통 60). 클수록 상위 순위의 영향이 평탄해짐.
    """
    scores = defaultdict(float)
    for results in result_lists:
        for rank, doc_id in enumerate(results):
            scores[doc_id] += 1.0 / (k + rank + 1)  # rank는 0부터
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

전체 하이브리드 검색 흐름

from rank_bm25 import BM25Okapi

class HybridRetriever:
    def __init__(self, chunks: list[dict], vectors: np.ndarray):
        self.chunks = chunks                     # [{"id":..., "text":...}, ...]
        self.vectors = l2_normalize(vectors)     # 정규화된 문서 벡터
        # BM25는 토큰화된 코퍼스가 필요. 한국어는 형태소 분석기 권장(아래 함정 참고)
        tokenized = [c["text"].lower().split() for c in chunks]
        self.bm25 = BM25Okapi(tokenized)
        self.id_list = [c["id"] for c in chunks]

    def search(self, query: str, query_vec: np.ndarray, top_k=50) -> list[dict]:
        # 1) 벡터 검색 — 내적(정규화했으므로 코사인과 동일)
        qv = l2_normalize(query_vec.reshape(1, -1))[0]
        sims = self.vectors @ qv
        dense_top = np.argsort(sims)[::-1][:top_k]
        dense_ids = [self.id_list[i] for i in dense_top]

        # 2) BM25 검색
        scores = self.bm25.get_scores(query.lower().split())
        sparse_top = np.argsort(scores)[::-1][:top_k]
        sparse_ids = [self.id_list[i] for i in sparse_top]

        # 3) RRF 융합
        fused = reciprocal_rank_fusion([dense_ids, sparse_ids])
        by_id = {c["id"]: c for c in self.chunks}
        return [by_id[doc_id] for doc_id, _ in fused[:top_k]]

프로덕션 벡터 DB: 위 numpy 예시는 수만 건까지는 괜찮지만, 수십만~수백만 건이면 ANN(근사 최근접) 인덱스가 필요하다. pgvector(Postgres 확장, 기존 DB에 붙이기 좋음), Qdrant, Weaviate, Milvus 등이 HNSW 인덱스로 빠른 검색을 제공한다. 이들 대부분은 하이브리드 검색을 내장 지원하므로 RRF를 직접 짤 필요가 줄어든다.

-- pgvector 예: HNSW 인덱스 + 코사인 거리
CREATE INDEX ON chunks USING hnsw (embedding vector_cosine_ops);
SELECT id, text, 1 - (embedding <=> :query_vec) AS similarity
FROM chunks
ORDER BY embedding <=> :query_vec  -- <=> 는 코사인 거리
LIMIT 50;

함정 — 한국어 BM25: text.split()은 공백 기준이라 한국어 조사·어미가 붙은 채로 토큰화된다("환불을", "환불은", "환불"이 전부 다른 토큰). 그러면 키워드 매칭이 깨진다. kiwipiepy나 konlpy 같은 형태소 분석기로 명사·어간을 추출해 토큰화해야 한국어 BM25가 제대로 작동한다.

from kiwipiepy import Kiwi
kiwi = Kiwi()

def tokenize_ko(text: str) -> list[str]:
    # 명사/동사/형용사 어간만 추출 — 조사·어미 제거
    return [t.form for t in kiwi.tokenize(text)
            if t.tag.startswith(("N", "V", "VA", "XR"))]

top-k는 크게 잡아라. 이 단계의 목표는 recall(정답을 놓치지 않기)이다. 50~100개를 가져온 뒤 다음 리랭킹 단계에서 정밀하게 줄인다. 여기서 5개만 가져오면 정답이 6번째에 있을 때 영원히 못 쓴다.

리랭킹은 RAG 품질을 가장 싸게 끌어올리는 단계다. 벡터 검색으로 가져온 top-50을, 더 정확한 모델로 다시 점수 매겨 top-5로 줄인다.

왜 두 단계로 나누나 — bi-encoder vs cross-encoder

• 임베딩 검색은 bi-encoder다. 질의와 문서를 각각 따로 벡터로 만들고 거리를 잰다. 문서 벡터를 미리 계산해 둘 수 있어 빠르다(수백만 건도 ms). 대신 질의와 문서가 만나서 상호작용하지 않으므로 정밀도가 떨어진다.
• 리랭커는 cross-encoder다. 질의와 문서를 함께 모델에 넣어 "이 문서가 이 질문에 얼마나 답하는가"를 직접 점수화한다. 훨씬 정확하지만 느리다 — 미리 계산이 불가능하고 (질의, 문서) 쌍마다 모델을 한 번씩 돌려야 한다.

그래서 조합한다: bi-encoder로 수백만 → 50개로 싸게 거르고, cross-encoder로 50 → 5개를 정밀하게 줄인다. 50번의 cross-encoder 호출은 감당할 만하다.

# 옵션 A: 오픈소스 cross-encoder (로컬, 데이터 외부 유출 없음)
from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")  # 다국어 지원

def rerank(query: str, candidates: list[dict], top_n=5) -> list[dict]:
    pairs = [[query, c["text"]] for c in candidates]
    scores = reranker.predict(pairs)  # (질의, 문서) 쌍마다 관련도 점수
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    out = []
    for chunk, score in ranked[:top_n]:
        chunk = {**chunk, "rerank_score": float(score)}
        out.append(chunk)
    return out

# 옵션 B: API 리랭커 (Cohere Rerank, Voyage rerank 등 — 운영 부담 적음)
import cohere
co = cohere.Client()

def rerank_api(query: str, candidates: list[dict], top_n=5) -> list[dict]:
    docs = [c["text"] for c in candidates]
    resp = co.rerank(model="rerank-multilingual-v3.0",
                     query=query, documents=docs, top_n=top_n)
    # resp.results는 relevance_score 내림차순, 원본 인덱스를 보존
    return [{**candidates[r.index], "rerank_score": r.relevance_score}
            for r in resp.results]

점수 임계값 컷오프: top_n으로 자르는 것 외에, 리랭크 점수가 낮은 청크는 아예 버리는 게 좋다. 질문과 무관한 청크를 컨텍스트에 넣으면 LLM이 헷갈린다("distraction"). 관련 문서가 0개면 솔직히 "모른다"고 답하게 하는 편이 환각보다 낫다.

def rerank_with_threshold(query, candidates, top_n=5, min_score=0.3):
    ranked = rerank(query, candidates, top_n=len(candidates))
    kept = [c for c in ranked if c["rerank_score"] >= min_score][:top_n]
    return kept  # 비어 있으면 "근거 문서 없음"으로 처리

임계값은 모델마다 다르다. cross-encoder 점수 스케일은 모델별로 천차만별(어떤 건 0~1, 어떤 건 로짓으로 음수~양수)이라, min_score는 평가셋으로 직접 튜닝해야 한다. 임의의 0.3을 그대로 믿지 마라.

비용·레이턴시 트레이드오프: 리랭킹은 검색 단계에서 가장 느린 부분이 될 수 있다. 후보 50개를 리랭킹하면 오픈소스 모델 기준 수십~수백 ms. 레이턴시가 빡빡하면 후보를 30개로 줄이거나, 작은 리랭커 모델을 쓰거나, API 리랭커의 배치 처리를 활용한다. 그래도 리랭킹을 빼는 것보다는 거의 항상 넣는 게 품질상 이득이다.

RAG에서 가장 많이 생략되는데 가장 중요한 단계다. 평가 없이 청크 크기·임베딩 모델·top-k를 바꾸면, 좋아졌는지 나빠졌는지 알 수 없다. 검색과 생성을 분리해서 평가하라 — 검색이 틀렸는데 생성을 고치는 건 헛수고다.

평가셋 만들기: 30~100개의 (질문, 정답이 들어있는 문서/청크, 이상적 답변) 셋을 만든다. 직접 만들거나, LLM에게 청크를 주고 "이 청크로 답할 수 있는 질문을 만들어라"로 합성한다. 합성 후에는 사람이 한 번 검수하라.

검색 지표 (생성 이전, retrieval만)

지표	정의	무엇을 본다
Hit Rate@k	정답 청크가 top-k에 들어왔는가 (비율)	가장 단순, recall 감각
MRR (Mean Reciprocal Rank)	정답의 첫 등장 순위의 역수 평균	정답이 "위쪽"에 오는가
Recall@k	top-k에 들어온 정답 청크 비율	정답이 여럿일 때
NDCG@k	순위 가중 관련도	순서 품질까지

def hit_rate_at_k(results_ids: list[str], gold_ids: set[str], k: int) -> int:
    return 1 if set(results_ids[:k]) & gold_ids else 0

def mrr(results_ids: list[str], gold_ids: set[str]) -> float:
    for rank, doc_id in enumerate(results_ids, start=1):
        if doc_id in gold_ids:
            return 1.0 / rank
    return 0.0

def evaluate_retrieval(eval_set, retriever, embed_fn, k=5):
    hits, mrrs = [], []
    for item in eval_set:
        qv = embed_fn(item["question"])
        results = retriever.search(item["question"], qv, top_k=k)
        ids = [r["id"] for r in results]
        gold = set(item["gold_chunk_ids"])
        hits.append(hit_rate_at_k(ids, gold, k))
        mrrs.append(mrr(ids, gold))
    return {"hit_rate@k": sum(hits)/len(hits), "mrr": sum(mrrs)/len(mrrs)}

생성 지표 (검색 결과로 답변까지 만든 뒤)

• Faithfulness(충실도): 답변이 제공된 컨텍스트에서 뒷받침되는가 — 환각 측정의 핵심.
• Answer Relevance(답변 관련성): 답변이 질문에 실제로 답하는가.
• Context Precision/Recall: 가져온 컨텍스트가 질문에 적절한가, 필요한 걸 다 가져왔는가.

이 지표들은 정답 문자열 매칭으로는 측정이 어렵다(같은 의미를 다른 말로 쓰니까). 그래서 LLM-as-judge를 쓴다 — 강한 모델에게 "이 답변이 이 컨텍스트로 뒷받침되는가"를 채점하게 한다. Ragas 같은 라이브러리가 이 지표들을 구현해 두었다.

# LLM-as-judge 충실도 채점 — 구조화 출력으로 안정적인 파싱
import anthropic
client = anthropic.Anthropic()

def judge_faithfulness(answer: str, context: str) -> dict:
    resp = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=1024,
        system=("당신은 RAG 답변 채점자입니다. 답변의 각 주장이 주어진 컨텍스트로 "
                "뒷받침되는지만 판단하세요. 외부 지식이나 상식은 근거로 인정하지 마세요."),
        messages=[{"role": "user", "content":
            f"<context>\n{context}\n</context>\n\n<answer>\n{answer}\n</answer>\n\n"
            "답변을 개별 주장으로 나누고, 각 주장이 컨텍스트로 뒷받침되면 supported, "
            "아니면 unsupported로 분류하세요."}],
        output_config={"format": {"type": "json_schema", "schema": {
            "type": "object",
            "properties": {
                "claims": {"type": "array", "items": {"type": "object",
                    "properties": {
                        "claim": {"type": "string"},
                        "verdict": {"type": "string", "enum": ["supported", "unsupported"]}
                    },
                    "required": ["claim", "verdict"], "additionalProperties": False}},
            },
            "required": ["claims"], "additionalProperties": False}}},
    )
    import json
    data = json.loads(next(b.text for b in resp.content if b.type == "text"))
    claims = data["claims"]
    supported = sum(1 for c in claims if c["verdict"] == "supported")
    return {"faithfulness": supported / max(len(claims), 1), "detail": claims}

위 예시는 채점자로 claude-opus-4-8을 썼다(가장 정확). 채점 비용이 부담되면 claude-sonnet-4-6도 충분히 좋다. 구조화 출력(output_config.format)을 쓰면 채점 결과가 항상 유효한 JSON으로 와서 파싱이 안정적이다.

평가를 CI에 넣어라. 청킹·임베딩·프롬프트를 바꿀 때마다 평가셋을 돌려 회귀를 잡는다. "임베딩 모델 바꿨더니 hit_rate가 0.82 → 0.71로 떨어졌다"를 코드 머지 전에 발견해야 한다.

검색·리랭킹으로 좋은 청크를 골랐어도, 프롬프트를 어떻게 짜느냐에 따라 답변 품질이 갈린다.

컨텍스트 배치 순서 (lost in the middle): 긴 컨텍스트에서 LLM은 맨 앞과 맨 뒤의 정보를 가장 잘 활용하고, 가운데 정보는 놓치는 경향이 있다. 그래서 가장 관련 높은 청크(리랭크 1위)를 맨 앞이나 맨 뒤에 배치하는 게 좋다. 청크 수가 적으면(3~5개) 큰 문제는 아니지만, 10개 이상이면 신경 써야 한다.

각 청크에 출처 표시를 달아 인용을 가능하게 한다.

def build_prompt(query: str, chunks: list[dict]) -> tuple[str, str]:
    # 리랭크 점수 내림차순으로 들어온다고 가정.
    # 1위를 맨 뒤(질문 직전)에 두려면 reversed; 여기선 단순히 순서 유지.
    context_blocks = []
    for i, c in enumerate(chunks, start=1):
        src = f"{c.get('source','?')} p.{c.get('page','?')}"
        context_blocks.append(f"[출처 {i}: {src}]\n{c['text']}")
    context = "\n\n---\n\n".join(context_blocks)

    system = (
        "당신은 사내 문서 Q&A 어시스턴트입니다. 다음 규칙을 엄격히 지키세요.\n"
        "1. 아래 <context>의 정보만 근거로 답하세요. 외부 지식·추측 금지.\n"
        "2. 각 주장 끝에 사용한 출처 번호를 [출처 N] 형식으로 표기하세요.\n"
        "3. context에 답이 없으면 '제공된 문서에서 답을 찾을 수 없습니다'라고만 답하세요. "
        "억지로 지어내지 마세요.\n"
        "4. 답은 간결하게, 사용자가 바로 쓸 수 있는 형태로."
    )
    user = f"<context>\n{context}\n</context>\n\n질문: {query}"
    return system, user

환각 억제의 핵심은 "모르면 모른다고 하게 하는 것"이다. 위 규칙 3번이 그것이다. 이게 없으면 LLM은 컨텍스트에 답이 없어도 그럴듯한 거짓을 만든다. 리랭크 임계값으로 관련 청크가 0개일 때는 LLM을 부르지 않고 곧장 "관련 문서 없음"을 반환하는 것도 방법이다.

Claude의 인용(citations) 기능: 직접 [출처 N] 규칙을 짜는 대신, Claude API의 내장 citations를 쓰면 답변의 각 문장이 어떤 소스 청크의 어느 부분에서 왔는지 구조화된 인용으로 돌아온다. 문서를 document 블록으로 넣고 citations: {enabled: true}를 켜면 된다. 인용 정확도가 중요한 제품(법률·의료·고객지원)에서 특히 유용하다.

생성 호출 — 모델 선택은 다음 섹션에서 자세히 다루지만, 기본형은 이렇다.

def generate(query: str, chunks: list[dict], model="claude-opus-4-8") -> str:
    system, user = build_prompt(query, chunks)
    resp = client.messages.create(
        model=model,
        max_tokens=2000,
        system=system,
        messages=[{"role": "user", "content": user}],
    )
    return next(b.text for b in resp.content if b.type == "text")

함정 — 컨텍스트 과다 주입: "많이 넣을수록 좋겠지"라며 top-20을 다 넣으면, (1) 비용이 늘고 (2) 무관한 청크가 LLM을 산만하게 하고 (3) lost-in-the-middle으로 정작 중요한 게 묻힌다. 리랭킹으로 추린 3~8개가 보통 최적이다. 컨텍스트는 "많이"가 아니라 "정확히"다.

RAG의 마지막 단계인 생성(generation)에서 어떤 LLM을 쓸지, 컨텍스트 윈도우를 어떻게 다룰지 정리한다. 아래는 Claude 모델 기준이며, 모델 ID·컨텍스트·가격은 현재 기준 실제 값이다.

Claude 모델 선택 (RAG 생성용)

모델	모델 ID	컨텍스트	입력/출력 ($/1M)	RAG에서의 용도
Claude Opus 4.8	claude-opus-4-8	1M	$5 / $25	복잡한 추론·다중 문서 종합, 채점(judge)
Claude Sonnet 4.6	claude-sonnet-4-6	1M	$3 / $15	대부분의 프로덕션 Q&A — 비용·품질 균형
Claude Haiku 4.5	claude-haiku-4-5	200K	$1 / $5	단순 추출·분류, 대량·저지연

RAG에서는 컨텍스트가 충분히 좁혀져 있으면 claude-sonnet-4-6이 기본으로 합리적이다. 검색·리랭킹이 일을 잘 했다면 생성은 "주어진 5개 청크로 답하기"라 초고성능 모델이 항상 필요하진 않다. 다만 여러 청크의 정보를 종합·비교·추론해야 하는 질문, 또는 평가 채점자로는 claude-opus-4-8이 낫다.

def pick_model(query: str, chunks: list[dict]) -> str:
    # 단순 휴리스틱 예시: 청크가 많고 질문이 복잡하면 상위 모델
    if len(chunks) >= 6 or len(query) > 200:
        return "claude-opus-4-8"
    return "claude-sonnet-4-6"

컨텍스트 윈도우와 "긴 컨텍스트 vs RAG" 오해

Claude 모델은 1M 토큰 컨텍스트를 제공한다. 그래서 "문서 전체를 그냥 다 넣으면 RAG가 필요 없지 않나?"라는 질문이 나온다. 부분적으로 맞지만, RAG를 여전히 쓰는 이유가 있다.

1. 비용: 1M 토큰을 매 질문마다 넣으면 입력 비용이 폭발한다(Opus 기준 100만 토큰 = $5/요청). RAG로 5개 청크(~3K 토큰)만 넣으면 비용이 수백 배 적다.
2. 레이턴시: 컨텍스트가 길수록 첫 토큰까지 시간이 늘어난다.
3. 정확도(lost in the middle): 컨텍스트가 길수록 가운데 정보를 놓칠 위험. 좁혀서 넣은 5개 청크가 통째로 넣은 1000페이지보다 정확할 수 있다.
4. 확장성: 지식 베이스가 1M 토큰을 넘으면 애초에 다 못 넣는다.

결론: 긴 컨텍스트는 RAG를 대체하는 게 아니라 보완한다. RAG로 후보를 좁히되, 큰 컨텍스트 덕분에 청크를 좀 더 넉넉히(또는 더 크게) 넣을 여유가 생긴다.

적응형 사고(adaptive thinking): 다중 문서를 종합·추론해야 하는 어려운 RAG 질문에는 Claude의 적응형 사고를 켜면 답변 품질이 오른다. 모델이 답하기 전에 필요한 만큼 추론한다.

resp = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=4000,
    thinking={"type": "adaptive"},        # 복잡한 종합 질문에 유효
    output_config={"effort": "high"},     # low | medium | high | max
    system=system,
    messages=[{"role": "user", "content": user}],
)

스트리밍: 사용자에게 답변을 실시간으로 보여주려면 스트리밍을 쓴다. max_tokens가 크면(긴 답변) 스트리밍이 사실상 필수다 — 논스트리밍은 긴 응답에서 HTTP 타임아웃 위험이 있다.

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=4000,
    system=system,
    messages=[{"role": "user", "content": user}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()

함정 — 모델 ID에 날짜 접미사 붙이기: claude-sonnet-4-6이 정확한 ID다. claude-sonnet-4-6-20251114 같은 날짜 접미사를 임의로 붙이면 404가 난다. 별칭(alias)을 그대로 쓰라.

RAG에서 프롬프트 캐싱은 비용을 크게 줄일 수 있지만, RAG 특성상 잘못 쓰기 쉽다. 캐싱이 어떻게 작동하는지 이해하고 캐시 친화적으로 프롬프트를 배치해야 한다.

원리 — 캐싱은 접두사(prefix) 매칭이다. 프롬프트 앞부분(접두사)이 이전 요청과 바이트 단위로 동일하면 그 부분을 캐시에서 재사용한다. 접두사 중간에 한 바이트라도 달라지면 그 지점부터 뒤는 전부 캐시 무효화된다. 렌더 순서는 tools → system → messages다.

RAG의 핵심 함정: 검색된 컨텍스트는 질문마다 다르다. 그래서 컨텍스트를 시스템 프롬프트 앞쪽에 넣으면 캐시가 매번 깨진다. 캐싱 이득을 보려면 고정된 부분(시스템 지시문, 도구 정의)을 앞에, 변하는 부분(검색 컨텍스트, 질문)을 뒤에 둬야 한다.

# 좋은 배치: 고정 시스템 지시문에만 캐시 마커. 컨텍스트·질문은 messages 뒤쪽(캐시 대상 아님)
resp = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=2000,
    system=[{
        "type": "text",
        "text": STABLE_SYSTEM_INSTRUCTIONS,   # 항상 동일한 규칙·페르소나
        "cache_control": {"type": "ephemeral"},
    }],
    messages=[{
        "role": "user",
        # 검색 컨텍스트 + 질문 — 매번 다르므로 캐시하지 않는다
        "content": f"<context>\n{retrieved_context}\n</context>\n\n질문: {query}",
    }],
)

언제 캐싱이 이득인가 (RAG 관점)

상황	캐싱 효과
시스템 프롬프트가 크다(긴 규칙·few-shot 예시·포맷 지침)	큼 — 매 질문 재사용
같은 문서에 여러 질문(문서 Q&A 세션)	큼 — 문서를 앞에 캐시
컨텍스트가 질문마다 완전히 다르고 시스템 지시문이 짧다	거의 없음

특히 한 문서에 여러 질문을 던지는 패턴(예: 계약서 분석 세션)에서는 문서를 캐시 마커가 붙은 앞쪽에 두고 질문만 뒤에 바꿔 끼우면 두 번째 질문부터 입력 비용이 ~90% 절감된다.

# 한 문서에 여러 질문: 문서를 캐시, 질문만 교체
def ask_about_document(document: str, questions: list[str]):
    base_messages = [{
        "role": "user",
        "content": [
            {"type": "text", "text": f"다음 문서를 참고해 답하세요:\n{document}",
             "cache_control": {"type": "ephemeral"}},   # 문서를 캐시
        ],
    }]
    for q in questions:
        resp = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1500,
            messages=base_messages + [{"role": "user", "content": q}],
        )
        yield resp

캐시 적중 검증: 응답의 usage로 확인한다. 반복 요청에서 cache_read_input_tokens가 0이면 무언가 접두사를 깨고 있는 것이다.

print(resp.usage.cache_creation_input_tokens)  # 캐시에 쓴 토큰 (~1.25배 비용)
print(resp.usage.cache_read_input_tokens)       # 캐시에서 읽은 토큰 (~0.1배 비용)
print(resp.usage.input_tokens)                  # 캐시 안 된 토큰 (정가)

조용한 캐시 무효화 범인들 (시스템 프롬프트 접두사에 이게 있으면 매번 캐시가 깨진다):

• 시스템 프롬프트에 datetime.now()나 UUID를 끼워 넣기 → 매 요청 접두사가 달라짐
• json.dumps()를 sort_keys=True 없이 → 직렬화 순서가 비결정적
• 검색 컨텍스트를 시스템 프롬프트에 넣기(RAG의 전형적 실수) → 질문마다 달라짐
• 도구 목록을 질문/사용자마다 다르게 구성 → 도구는 위치 0이라 전체 캐시 파괴

최소 캐시 길이: 접두사가 모델별 최소 토큰(예: Sonnet 4.6은 2048토큰, Opus 4.8은 4096토큰)보다 짧으면 마커를 붙여도 조용히 캐시되지 않는다 — 에러 없이 그냥 cache_creation_input_tokens: 0이 나온다. 시스템 프롬프트가 짧으면 캐싱 이득이 없으니 무리해서 적용하지 마라.

경제성: 캐시 읽기는 정가 입력의 ~0.1배, 캐시 쓰기는 ~1.25배(5분 TTL). 즉 같은 접두사로 2번 이상 요청하면 본전을 넘긴다. RAG 문서 Q&A 세션처럼 같은 큰 컨텍스트를 반복 사용하는 패턴이 캐싱의 단골 수혜 케이스다.

RAG가 "답이 이상하다"고 할 때, 어느 단계가 범인인지 가르는 디버깅 순서다. 검색부터 의심하고, 생성은 마지막에 의심하라.

1단계 — 검색이 정답 청크를 가져왔는가? 질문을 던지고 top-k 결과를 사람이 눈으로 확인한다. 정답이 들어있는 청크가 결과에 없다면 문제는 검색이다.

def debug_retrieval(query: str, retriever, embed_fn, k=10):
    qv = embed_fn(query)
    results = retriever.search(query, qv, top_k=k)
    print(f"질문: {query}\n")
    for i, r in enumerate(results, 1):
        print(f"[{i}] (score={r.get('rerank_score', '?')}) "
              f"{r.get('source','?')} p.{r.get('page','?')}")
        print(f"    {r['text'][:150]}...\n")

증상	가능한 원인	해결
정답 청크가 top-50에도 없음	청킹이 정답을 두 청크로 쪼갬 / 임베딩 모델이 도메인 약함	청크 크기·겹침 조정, 임베딩 모델 교체, 섹션 제목 contextualize
키워드(코드·고유명사) 질문만 실패	벡터 검색만 씀	BM25 하이브리드 추가
한국어 키워드 매칭 실패	BM25 토큰화가 공백 기준	형태소 분석기로 토큰화
정답이 top-50엔 있는데 top-5엔 없음	리랭킹 부재 또는 약함	리랭커 추가/교체
정답이 항상 6~10위	top-k가 작음	top-k 키우고 리랭킹

2단계 — 컨텍스트는 맞는데 답이 틀린가? 검색은 정답 청크를 잘 가져왔는데 답변이 틀리거나 환각이라면, 문제는 프롬프트·생성이다.

증상	원인	해결
컨텍스트에 없는 내용을 답함	"모르면 모른다" 지시 부재	시스템 프롬프트에 환각 억제 규칙 추가
컨텍스트 가운데 정보를 놓침	lost in the middle / 청크 과다	청크 수 줄이고, 핵심을 앞/뒤 배치
여러 청크 종합 질문에서 부분만 답함	모델 추론력 부족	상위 모델 + 적응형 사고
인용이 부정확	인용 규칙 약함	Claude 내장 citations 사용

3단계 — 비용·레이턴시가 문제인가?

증상	원인	해결
입력 비용이 큼	컨텍스트 과다 / 캐시 미적중	리랭크로 청크 축소, 프롬프트 캐싱
첫 토큰 지연	컨텍스트 길이 / 리랭킹 느림	청크 축소, 캐시 프리워밍, 작은 리랭커
색인이 너무 느림	임베딩을 1개씩 호출	배치 임베딩

황금률 정리

1. 검색을 먼저 의심하라 — 실패의 대부분은 여기다.
2. 넓게 검색하고(top-k 크게) 좁게 리랭킹하라(top-n 작게).
3. 하이브리드(벡터+BM25)가 거의 항상 단일 검색보다 낫다.
4. 평가셋 없이 튜닝하지 마라 — 숫자로 회귀를 잡아라.
5. 모르면 모른다고 답하게 하라 — 환각보다 "근거 없음"이 낫다.
6. 임베딩 모델은 색인·질의에서 동일해야 하고, 바꾸면 인덱스를 재생성하라.
7. 고정된 건 앞에, 변하는 건 뒤에 — 캐시 친화 프롬프트 배치.