사내 위키, 제품 매뉴얼, 계약서, 고객 문의 로그 — 회사가 가진 데이터는 대부분 LLM의 학습 데이터에 들어 있지 않습니다. 그래서 Claude에게 "우리 환불 정책이 뭐냐"고 물으면, 모델은 일반론을 답하거나 그럴듯한 거짓말(환각)을 만들어 냅니다. RAG(Retrieval-Augmented Generation, 검색 증강 생성)는 이 문제를 정공법으로 해결합니다. 질문이 들어오면 먼저 내 데이터에서 관련 조각을 검색하고, 그 조각을 프롬프트에 끼워 넣은 뒤, Claude가 그 근거만 보고 답하게 만드는 구조입니다. 이 글에서는 청킹·임베딩·벡터 검색·컨텍스트 주입·출처 인용까지, 실제로 동작하는 코드와 함께 RAG 파이프라인을 처음부터 끝까지 만들어 봅니다.

한 가지 중요한 사실을 먼저 못 박아 둡니다. Anthropic은 임베딩 API를 제공하지 않습니다. RAG에서 임베딩(텍스트를 벡터로 바꾸는 단계)은 OpenAI의 text-embedding-3-small/-large 또는 Anthropic이 공식 권장하는 Voyage AI를 사용하고, 생성(답변 작성)만 Claude가 담당합니다. "Claude 임베딩 엔드포인트" 같은 건 존재하지 않으니, 그런 코드를 보면 잘못된 것입니다.

RAG는 세 단계다: 검색 → 증강 → 생성

RAG의 이름 그대로 흐름은 세 단계로 나뉩니다. 각 단계가 무엇을 하는지 먼저 머릿속에 그려 두면 나머지 코드가 전부 이 골격에 붙습니다.

• 검색(Retrieval) — 사용자 질문을 임베딩 벡터로 바꾸고, 미리 벡터화해 저장해 둔 문서 청크 중 의미적으로 가장 가까운 상위 N개를 코사인 유사도로 찾아옵니다.
• 증강(Augmentation) — 찾아온 청크들을 프롬프트의 컨텍스트 영역에 삽입하고, "이 자료만 근거로 답하라"는 지시와 출처 표기 규칙을 함께 넣습니다.
• 생성(Generation) — Claude가 주입된 컨텍스트를 읽고 답변을 작성합니다. 근거가 없으면 "자료에 없습니다"라고 말하게 유도해 환각을 억제합니다.

이때 데이터 준비(인덱싱) 과정은 질의와 분리되어 있습니다. 인덱싱은 오프라인으로 한 번(또는 문서가 바뀔 때마다) 수행하고, 질의는 온라인으로 매 요청마다 일어납니다. 두 흐름을 헷갈리지 않는 게 RAG 설계의 출발점입니다.

문서 청킹: 검색 품질의 70%가 여기서 결정된다

임베딩 모델은 긴 문서를 통째로 벡터 하나로 뭉개면 의미가 흐려집니다. 그래서 문서를 적당한 크기의 청크(chunk)로 잘라 각각 임베딩합니다. 청크가 너무 크면 한 벡터에 여러 주제가 섞여 검색 정밀도가 떨어지고, 너무 작으면 문맥이 잘려 답변 근거가 부실해집니다. 실무에서는 다음 전략을 씁니다.

• 크기 — 보통 토큰 기준 300~800 토큰. 문장이 잘리지 않도록 문단·문장 경계에서 자릅니다.
• 오버랩(overlap) — 인접 청크가 50~100 토큰쯤 겹치게 합니다. 경계에 걸친 정보가 양쪽 청크 모두에 남아 검색 누락을 줄입니다.
• 구조 보존 — 마크다운 헤더, 표, 코드 블록은 가능한 한 한 청크 안에 유지합니다. 제목을 각 청크 앞에 붙여 주면(예: "환불 정책 > 7일 이내") 문맥이 살아납니다.
• 메타데이터 — 청크마다 출처 문서, URL, 페이지/섹션 번호를 함께 저장합니다. 나중에 출처 인용에 필수입니다.

Claude의 count_tokens로 청크 토큰 수를 정확히 재면서 자르는 예시입니다. 임베딩 모델과 토크나이저가 다르긴 하지만, 컨텍스트 예산을 잡는 기준으로는 충분히 유용합니다(tiktoken은 Claude 토큰을 부정확하게 세므로 컨텍스트 예산 산정에는 쓰지 마세요).

import re
from anthropic import Anthropic

client = Anthropic()  # ANTHROPIC_API_KEY 환경변수 사용

def count_tokens(text: str) -> int:
    # 정확한 토큰 수는 모델별. 컨텍스트 예산 산정용 기준으로 사용.
    return client.messages.count_tokens(
        model="claude-opus-4-8",
        messages=[{"role": "user", "content": text}],
    ).input_tokens

def chunk_document(text: str, max_tokens: int = 500, overlap_sentences: int = 2):
    """문장 경계로 자르고, 인접 청크에 문장 단위 오버랩을 둔다."""
    sentences = re.split(r"(?<=[.!?。\n])\s+", text.strip())
    chunks, current, current_tokens = [], [], 0

    for sent in sentences:
        t = count_tokens(sent)
        if current_tokens + t > max_tokens and current:
            chunks.append(" ".join(current))
            # 끝 문장 몇 개를 다음 청크 앞에 겹쳐 둔다
            current = current[-overlap_sentences:]
            current_tokens = sum(count_tokens(s) for s in current)
        current.append(sent)
        current_tokens += t

    if current:
        chunks.append(" ".join(current))
    return chunks

임베딩 생성: OpenAI 또는 Voyage AI (Claude 아님)

다시 강조합니다. 임베딩은 Claude가 하지 않습니다. 가장 보편적인 선택은 OpenAI text-embedding-3-small(1536차원, 저렴·빠름)이고, 더 높은 정확도가 필요하면 text-embedding-3-large(3072차원)나 Anthropic이 RAG용으로 권장하는 Voyage AI를 씁니다. 핵심 규칙: 인덱싱 때 쓴 임베딩 모델과 질의 때 쓰는 모델은 반드시 동일해야 합니다. 모델이 다르면 벡터 공간이 달라 유사도가 무의미해집니다.

from openai import OpenAI

oai = OpenAI()  # OPENAI_API_KEY 환경변수 사용
EMBED_MODEL = "text-embedding-3-small"  # 1536차원. 인덱싱/질의에 동일 모델 사용

def embed(texts: list[str]) -> list[list[float]]:
    """여러 텍스트를 한 번에 임베딩(배치가 비용/지연에 유리)."""
    resp = oai.embeddings.create(model=EMBED_MODEL, input=texts)
    # 입력 순서와 동일한 순서로 data가 반환된다
    return [item.embedding for item in resp.data]

# 인덱싱 단계: 청크 → 벡터
chunks = chunk_document(open("refund_policy.md").read())
vectors = embed(chunks)
print(len(vectors), "개 청크 임베딩 완료, 차원:", len(vectors[0]))

OpenAI 모델 ID(text-embedding-3-small/-large)는 안정적이지만, 새 모델이 추가될 수 있으니 운영 전 공급사 현행 문서에서 제공 모델을 확인하는 것을 권장합니다. Voyage를 쓸 경우에도 차원 수와 모델명을 같은 방식으로 고정하면 됩니다.

벡터 저장과 유사도 검색: pgvector

벡터를 어디에 저장할지가 다음 결정입니다. 전용 벡터 DB(Pinecone, Qdrant 등)도 있지만, 이미 PostgreSQL을 쓰고 있다면 pgvector 확장이 가장 실무적입니다. 기존 관계형 데이터와 메타데이터를 한 트랜잭션·한 쿼리에서 다룰 수 있기 때문입니다. 먼저 확장과 테이블을 만듭니다.

-- pgvector 확장 활성화
CREATE EXTENSION IF NOT EXISTS vector;

-- 청크 + 1536차원 임베딩 + 출처 메타데이터
CREATE TABLE doc_chunks (
    id          bigserial PRIMARY KEY,
    source      text        NOT NULL,   -- 출처 문서명/URL
    section     text,                   -- 섹션/페이지 (출처 인용용)
    content     text        NOT NULL,   -- 청크 원문
    embedding   vector(1536) NOT NULL   -- text-embedding-3-small 차원
);

-- 코사인 거리용 근사 최근접 인덱스(HNSW). 데이터가 많을 때 검색 속도 핵심.
CREATE INDEX ON doc_chunks
    USING hnsw (embedding vector_cosine_ops);

pgvector는 거리 연산자를 제공합니다. <=>는 코사인 거리, <->는 L2(유클리드) 거리, <#>는 음의 내적입니다. 임베딩이 정규화돼 있으면 코사인이 표준 선택입니다. 코사인 유사도 = 1 − 코사인 거리이므로, 거리가 작을수록 유사합니다. 질의 벡터로 상위 K개를 뽑는 검색 쿼리는 다음과 같습니다.

-- :query_embedding 자리에 질문 임베딩(벡터 리터럴)을 바인딩
SELECT
    source,
    section,
    content,
    1 - (embedding <=> :query_embedding) AS cosine_similarity
FROM doc_chunks
ORDER BY embedding <=> :query_embedding   -- 거리 오름차순 = 가까운 순
LIMIT 5;

이제 데이터를 적재하고 검색하는 Python 코드입니다. pgvector 벡터는 [0.1,0.2,...] 형태의 문자열 리터럴로 바인딩하거나 pgvector.psycopg 어댑터를 쓰면 됩니다. 여기서는 의존성을 줄이려고 리터럴로 직접 만들었습니다.

import psycopg

def to_pgvector(vec: list[float]) -> str:
    return "[" + ",".join(f"{x:.8f}" for x in vec) + "]"

def index_chunks(conn, source: str, chunks: list[str], sections: list[str]):
    vecs = embed(chunks)
    with conn.cursor() as cur:
        for content, section, vec in zip(chunks, sections, vecs):
            cur.execute(
                "INSERT INTO doc_chunks (source, section, content, embedding) "
                "VALUES (%s, %s, %s, %s)",
                (source, section, content, to_pgvector(vec)),
            )
    conn.commit()

def search(conn, question: str, top_k: int = 5):
    q_vec = embed([question])[0]              # 질의도 같은 임베딩 모델!
    with conn.cursor() as cur:
        cur.execute(
            "SELECT source, section, content, "
            "       1 - (embedding <=> %s) AS similarity "
            "FROM doc_chunks "
            "ORDER BY embedding <=> %s "
            "LIMIT %s",
            (to_pgvector(q_vec), to_pgvector(q_vec), top_k),
        )
        rows = cur.fetchall()
    return [
        {"source": r[0], "section": r[1], "content": r[2], "similarity": r[3]}
        for r in rows
    ]

상위 청크를 컨텍스트로 주입하고 Claude로 생성

검색이 끝났으니 이제 핵심 단계, 증강과 생성입니다. 찾아온 청크를 프롬프트에 번호를 붙여 넣고, Claude에게 "이 자료만 근거로 답하고, 문장마다 [출처 번호]를 달고, 자료에 없으면 모른다고 말하라"고 지시합니다. 이 지시문이 환각 억제와 출처 인용의 핵심입니다.

여기서 Anthropic SDK의 정확한 사용법을 지킵니다. 모델 ID는 claude-opus-4-8, 시스템 프롬프트는 messages와 별개인 system 파라미터로 전달하며, 응답 msg.content는 블록 리스트이므로 block.type == "text"를 확인하고 읽습니다.

from anthropic import Anthropic

client = Anthropic()

SYSTEM_PROMPT = """당신은 사내 문서 기반 Q&A 어시스턴트입니다. 다음 규칙을 반드시 지키세요.
- 아래 <컨텍스트>에 주어진 자료만 근거로 답하세요. 사전 지식이나 추측으로 채우지 마세요.
- 근거로 사용한 문장 끝마다 [번호] 형식으로 출처를 표기하세요. 예: ...환불됩니다 [2].
- 컨텍스트에 답이 없으면 "제공된 자료에서 해당 정보를 찾을 수 없습니다."라고만 답하세요.
- 추측, 일반론, 자료 밖 정보는 절대 만들어 내지 마세요."""

def build_context(chunks: list[dict]) -> str:
    blocks = []
    for i, c in enumerate(chunks, start=1):
        cite = f"{c['source']}" + (f" ({c['section']})" if c["section"] else "")
        blocks.append(f"[{i}] 출처: {cite}\n{c['content']}")
    return "\n\n".join(blocks)

def answer(conn, question: str) -> str:
    chunks = search(conn, question, top_k=5)
    context = build_context(chunks)

    user_msg = (
        f"<컨텍스트>\n{context}\n</컨텍스트>\n\n"
        f"질문: {question}\n\n"
        "위 컨텍스트만 근거로, 출처 번호를 달아 답하세요."
    )

    msg = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=1024,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": user_msg}],
    )

    # content는 블록 리스트 — 텍스트 블록만 모은다
    parts = [b.text for b in msg.content if b.type == "text"]
    return "".join(parts)

# 사용 예
with psycopg.connect("postgresql://localhost/ragdb") as conn:
    print(answer(conn, "구매 후 며칠 안에 환불받을 수 있나요?"))

출처 인용과 환각 줄이기

위 코드는 이미 환각 억제의 3대 장치를 담고 있습니다. 정리하면 다음과 같습니다.

• 근거 한정 — "주어진 자료만 근거로"라는 명시적 제약. 모델이 사전 지식으로 빈칸을 채우는 걸 막습니다.
• 모름의 허용 — "찾을 수 없습니다"라는 탈출구를 명시. 답을 억지로 지어내는 압박을 없앱니다. 이 한 줄이 환각을 가장 크게 줄입니다.
• 출처 강제 — 문장마다 [번호] 표기. 사용자가 근거 청크를 역추적해 검증할 수 있고, 모델도 근거 없는 주장을 하기 어려워집니다.

응답에서 인용 번호를 파싱해 실제 출처 객체와 매핑하면, UI에 "각주"를 그대로 렌더링할 수 있습니다. [2]는 chunks[1]의 source/section에 대응됩니다. 더 엄격한 인용이 필요하면 구조화 출력으로 답변과 인용을 분리해 받을 수 있습니다. Anthropic의 현행 구조화 출력은 output_config={"format": {...}}를 사용합니다(구버전 output_format 파라미터는 폐기됐습니다). 스키마에는 additionalProperties: false가 필요하고, minLength 같은 제약은 미지원이라 클라이언트 측에서 검증합니다.

SCHEMA = {
    "type": "object",
    "additionalProperties": False,
    "properties": {
        "answer": {"type": "string"},
        "citations": {
            "type": "array",
            "items": {"type": "integer"},  # 사용한 컨텍스트 번호들
        },
        "found": {"type": "boolean"},      # 자료에서 답을 찾았는지
    },
    "required": ["answer", "citations", "found"],
}

def answer_structured(conn, question: str) -> dict:
    chunks = search(conn, question, top_k=5)
    context = build_context(chunks)
    user_msg = (
        f"<컨텍스트>\n{context}\n</컨텍스트>\n\n질문: {question}"
    )

    msg = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=1024,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": user_msg}],
        output_config={"format": {"type": "json_schema", "schema": SCHEMA}},
    )

    import json
    text = "".join(b.text for b in msg.content if b.type == "text")
    data = json.loads(text)   # 스키마에 맞춘 JSON
    # 인용 번호 → 실제 출처로 매핑
    data["sources"] = [
        {"source": chunks[i - 1]["source"], "section": chunks[i - 1]["section"]}
        for i in data["citations"] if 1 <= i <= len(chunks)
    ]
    return data

프롬프트 캐싱으로 비용·지연 줄이기

RAG에서 시스템 프롬프트와 공통 지시문은 매 요청 동일합니다. 이런 안정적 prefix는 프롬프트 캐싱으로 재사용하면 비용과 지연을 크게 낮춥니다. 콘텐츠 블록에 cache_control={"type": "ephemeral"}를 붙이면 됩니다. 단, 캐시는 prefix 일치 방식이라 캐싱 대상(고정 시스템 프롬프트)을 앞에, 변하는 부분(검색된 컨텍스트·질문)을 뒤에 둬야 합니다. 검색 결과는 질의마다 달라지므로 캐싱하지 말고, 고정 시스템 프롬프트만 캐싱하는 게 핵심입니다.

msg = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": SYSTEM_PROMPT,                 # 매 요청 동일 → 캐시 대상
            "cache_control": {"type": "ephemeral"},
        }
    ],
    messages=[{"role": "user", "content": user_msg}],  # 컨텍스트+질문 → 매번 다름
)

# 캐시 적중 확인: 0이 계속 나오면 prefix가 매번 달라지는 것
print("cache_read:", msg.usage.cache_read_input_tokens)
print("cache_write:", msg.usage.cache_creation_input_tokens)

캐시 쓰기는 약 1.25배, 읽기는 약 0.1배 비용입니다. 최소 캐시 대상 prefix는 모델별로 다르며(Opus 4.8은 약 4096토큰), 그보다 짧으면 조용히 캐싱되지 않습니다. 시스템 프롬프트에 datetime.now() 같은 매번 변하는 값을 넣으면 prefix가 깨져 캐시가 무효화되니 주의하세요.

Laravel(PHP)에서 RAG 답변 생성

독자 중 Laravel 개발자가 많으니 PHP 예제도 봅니다. Anthropic 공식 PHP SDK(composer require anthropic-ai/sdk)로 동일한 생성 단계를 구현합니다. 검색(pgvector)은 Laravel의 DB 레이어로 처리하고, 가져온 청크를 컨텍스트로 주입합니다. PHP에서도 응답 content는 블록 배열이라 텍스트 블록만 골라 읽습니다.

<?php

use Anthropic\Client;
use Illuminate\Support\Facades\DB;

// 1) 질의 임베딩은 OpenAI/Voyage로 생성(여기선 $queryEmbedding이 이미 있다고 가정)
//    pgvector 리터럴 문자열로 변환
$vectorLiteral = '[' . implode(',', $queryEmbedding) . ']';

// 2) pgvector 유사도 검색 — 거리 오름차순 상위 5개
$chunks = DB::select(
    'SELECT source, section, content,
            1 - (embedding <=> ?) AS similarity
     FROM doc_chunks
     ORDER BY embedding <=> ?
     LIMIT 5',
    [$vectorLiteral, $vectorLiteral]
);

// 3) 컨텍스트 구성(출처 번호 부여)
$contextBlocks = [];
foreach ($chunks as $i => $c) {
    $n = $i + 1;
    $cite = $c->source . ($c->section ? " ({$c->section})" : '');
    $contextBlocks[] = "[{$n}] 출처: {$cite}\n{$c->content}";
}
$context = implode("\n\n", $contextBlocks);

$systemPrompt = "당신은 사내 문서 기반 Q&A 어시스턴트입니다. "
    . "아래 컨텍스트만 근거로 답하고, 문장마다 [번호]로 출처를 표기하세요. "
    . "자료에 없으면 '제공된 자료에서 찾을 수 없습니다'라고만 답하세요.";

$question = "구매 후 며칠 안에 환불받을 수 있나요?";
$userMsg  = "<컨텍스트>\n{$context}\n</컨텍스트>\n\n질문: {$question}";

// 4) Claude로 생성
$client  = new Client(apiKey: getenv('ANTHROPIC_API_KEY'));
$message = $client->messages->create(
    model: 'claude-opus-4-8',
    maxTokens: 1024,
    system: $systemPrompt,
    messages: [
        ['role' => 'user', 'content' => $userMsg],
    ],
);

// content는 블록 배열 — 텍스트 블록만 출력
foreach ($message->content as $block) {
    if ($block->type === 'text') {
        echo $block->text;
    }
}

운영에서 챙길 것들

파이프라인이 돌아간다고 끝이 아닙니다. 검색 품질과 비용을 좌우하는 실무 포인트를 정리합니다.

• 임베딩 모델 일관성 — 인덱싱과 질의는 같은 모델·같은 차원. 모델을 바꾸면 전체 재인덱싱이 필요합니다.
• Top-K와 컨텍스트 예산 — K를 키우면 누락은 줄지만 노이즈와 토큰 비용이 늡니다. 보통 3~8에서 시작해 평가셋으로 조정합니다.
• 메타데이터 필터링 — pgvector는 WHERE source = ... 같은 조건과 벡터 검색을 한 쿼리에서 결합할 수 있습니다. 부서별·기간별로 검색 범위를 좁히면 정밀도가 오릅니다.
• 재순위(re-ranking) — 1차로 벡터 Top-20을 뽑은 뒤 cross-encoder나 Voyage rerank로 Top-5를 다시 고르면 정밀도가 크게 좋아집니다.
• 긴 입력은 스트리밍 — 컨텍스트가 커지고 max_tokens가 16000을 넘으면 타임아웃 방지를 위해 client.messages.stream(...)으로 받으세요.
• 모델 선택 — 답변 품질이 중요하면 claude-opus-4-8($5/$25 per 1M), 대량·저비용이면 claude-haiku-4-5($1/$5)로 비용을 조절합니다.

자주 묻는 질문 (FAQ)

Q. Claude로 임베딩을 만들 수 있나요?
아니요. Anthropic은 임베딩 API를 제공하지 않습니다. RAG의 임베딩 단계는 OpenAI text-embedding-3-small/-large 또는 Anthropic이 권장하는 Voyage AI를 사용하고, 검색된 컨텍스트로 답변을 생성하는 단계만 Claude가 담당합니다. "Claude 임베딩 엔드포인트"는 존재하지 않습니다.

Q. 파인튜닝과 RAG 중 무엇을 써야 하나요?
자주 바뀌는 사실·문서 기반 지식에는 RAG가 적합합니다. 데이터가 바뀌어도 재학습 없이 인덱스만 갱신하면 되고, 출처 인용으로 근거를 보일 수 있어 환각 검증이 쉽습니다. 파인튜닝은 말투·형식·도메인 행동을 바꿀 때 유리하며, 둘을 함께 쓰기도 합니다. 대부분의 "내 데이터로 답하기"는 RAG가 1차 선택입니다.

Q. 청크 크기는 어떻게 정하나요?
정답은 없지만 300~800 토큰에 50~100 토큰 오버랩이 흔한 출발점입니다. 너무 크면 한 벡터에 여러 주제가 섞여 검색 정밀도가 떨어지고, 너무 작으면 문맥이 잘립니다. 문장·문단 경계에서 자르고, 마크다운 헤더·표·코드 블록은 한 청크에 유지하며, 실제 질문 샘플로 검색 적중을 평가해 조정하세요.

Q. pgvector와 전용 벡터 DB 중 무엇을 골라야 하나요?
이미 PostgreSQL을 쓰고 메타데이터·관계형 데이터와 함께 다뤄야 한다면 pgvector가 가장 실무적입니다. 한 쿼리에서 필터+벡터 검색이 가능하고 운영 부담이 적습니다. 수억 벡터 규모의 초대형 검색이나 특화된 ANN 튜닝이 필요하면 Pinecone·Qdrant 같은 전용 DB를 고려하세요. HNSW 인덱스만 잘 잡으면 pgvector도 상당한 규모를 감당합니다.

Q. RAG가 그래도 환각을 하면 어떻게 줄이나요?
세 가지를 동시에 적용하세요. (1) "주어진 자료만 근거로"라는 명시적 제약, (2) "자료에 없으면 모른다고 답하라"는 탈출구, (3) 문장별 출처 [번호] 강제입니다. 그래도 부족하면 재순위로 검색 정밀도를 높이고, 구조화 출력의 found 플래그로 "근거 없음"을 코드 레벨에서 분기 처리하세요.

AI 개발이 필요하신가요?

DevTeam은 OpenAI·Claude 등 AI API 연동, 챗봇, RAG, 업무 자동화를 설계부터 배포까지 구현합니다. AI 연동 개발 또는 무료 견적 문의.