가이드: 임베딩으로 RAG 구현하기

검색 증강 생성(RAG)은 사용자가 가진 문서로 질문에 답합니다. 문서를 청크로 나누고 청크마다 한 번씩 임베딩해 두었다가, 질문이 들어오면 질문을 임베딩해 가장 가까운 청크를 찾고, 필요하면 재순위를 매긴 뒤, 가장 좋은 청크를 출처를 인용하라는 지시와 함께 채팅 모델에 보냅니다.

아래의 모든 코드는 base URL https://api14.com/api/v1, sk-api14-... 키, 공식 openai 패키지를 사용합니다. 비용은 선불 USD 잔액에서 차감되며 모든 응답의 usage.cost로 반환됩니다.

import os
from openai import OpenAI

client = OpenAI(base_url="https://api14.com/api/v1", api_key=os.environ["API14_API_KEY"])
EMBED_MODEL = "openai/text-embedding-3-small"
CHAT_MODEL = "openai/gpt-5-mini"

1. 문서를 청크로 나누기

청크는 구체적일 만큼 작고, 그 자체로 의미가 통할 만큼 커야 합니다. 시작점으로는 문단 경계에서 나누고 약간 겹치게 하는 500~1,000자 정도가 좋습니다.

def chunk(text, size=800, overlap=150):
    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
    chunks, current = [], ""
    for p in paragraphs:
        if current and len(current) + len(p) > size:
            chunks.append(current)
            current = current[-overlap:]  # carry context into the next chunk
        current = f"{current}\n\n{p}".strip()
    if current:
        chunks.append(current)
    return chunks

docs = {"handbook.md": open("handbook.md").read(), "faq.md": open("faq.md").read()}
chunks = [{"source": name, "text": c} for name, text in docs.items() for c in chunk(text)]

모든 청크에 출처(파일 이름, URL, 페이지 번호)를 함께 보관하세요. 인용에 필요합니다.

2. POST /embeddings로 임베딩하기

요청 하나에 여러 청크를 보내세요. 입력 100개를 담은 요청 하나는 속도 제한에서 요청 한 번으로 계산됩니다.

def embed(texts):
    vectors = []
    for i in range(0, len(texts), 100):
        r = client.embeddings.create(model=EMBED_MODEL, input=texts[i:i + 100])
        vectors += [d.embedding for d in sorted(r.data, key=lambda d: d.index)]
    return vectors

for c, v in zip(chunks, embed([c["text"] for c in chunks])):
    c["vector"] = v

벡터와 모델 id를 함께 저장하세요. 서로 다른 모델의 벡터는 비교할 수 없으므로 EMBED_MODEL을 바꾸면 코퍼스를 다시 임베딩해야 합니다. Embeddings를 참고하세요.

3. 코사인 유사도로 검색하기

청크가 수천 개 정도라면 메모리에서 선형 탐색을 해도 충분히 빠릅니다. 그보다 많다면 벡터를 벡터 데이터베이스나 pgvector가 설치된 Postgres 테이블에 넣으세요.

import math

def cosine(a, b):
    dot = sum(x * y for x, y in zip(a, b))
    return dot / (math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b)))

def search(question, k=20):
    q = embed([question])[0]
    return sorted(chunks, key=lambda c: cosine(q, c["vector"]), reverse=True)[:k]

4. POST /rerank로 재순위 매기기

임베딩 검색은 재현율은 좋지만 순서는 거칠게 매깁니다. 재순위 모델은 질문과 각 후보를 함께 읽고 관련도를 더 정밀하게 채점합니다. 후보를 20100개 가져와 재순위를 매기고 상위 38개만 남기세요. OpenAI SDK에는 rerank 메서드가 없으므로 HTTP로 호출합니다(httpx는 openai와 함께 설치됩니다).

import httpx

def rerank(question, candidates, top_n=5):
    r = httpx.post(
        "https://api14.com/api/v1/rerank",
        headers={"Authorization": f"Bearer {os.environ['API14_API_KEY']}"},
        json={
            "model": "cohere/rerank-v3.5",
            "query": question,
            "documents": [c["text"] for c in candidates],
            "top_n": top_n,
        },
        timeout=30,
    )
    r.raise_for_status()
    return [candidates[x["index"]] for x in r.json()["results"]]

index는 보낸 documents 배열의 위치를 가리킵니다. Rerank를 참고하세요.

5. 출처를 인용하며 답변하기

프롬프트에서 청크에 번호를 매기고, 모델에게 그 번호를 인용하도록 요청하세요. 컨텍스트에 답이 없으면 그렇게 말하라고 지시합니다.

def answer(question):
    context = rerank(question, search(question))
    sources = "\n\n".join(f"[{i + 1}] ({c['source']})\n{c['text']}" for i, c in enumerate(context))
    r = client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[
            {"role": "system", "content": (
                "Answer only from the numbered sources. Cite them like [1] or [2][3] after each claim. "
                "If the sources do not contain the answer, say you don't know."
            )},
            {"role": "user", "content": f"Sources:\n\n{sources}\n\nQuestion: {question}"},
        ],
    )
    return r.choices[0].message.content, context, r.usage.cost

앱에서 인용을 데이터로 사용해야 한다면, 대괄호를 파싱하는 대신 { "answer": string, "citations": integer[] } 같은 스키마와 함께 Structured outputs를 사용하세요.

Node에서의 동일한 파이프라인

import OpenAI from "openai";

const client = new OpenAI({ baseURL: "https://api14.com/api/v1", apiKey: process.env.API14_API_KEY });

const embed = async (input) =>
  (await client.embeddings.create({ model: "openai/text-embedding-3-small", input })).data.map((d) => d.embedding);

const cosine = (a, b) => {
  let dot = 0, na = 0, nb = 0;
  for (let i = 0; i < a.length; i++) { dot += a[i] * b[i]; na += a[i] ** 2; nb += b[i] ** 2; }
  return dot / Math.sqrt(na * nb);
};

async function answer(question, chunks /* [{ source, text, vector }] */) {
  const [q] = await embed([question]);
  const candidates = chunks.map((c) => ({ ...c, score: cosine(q, c.vector) }))
    .sort((a, b) => b.score - a.score).slice(0, 20);

  const ranked = await client.post("/rerank", {
    body: { model: "cohere/rerank-v3.5", query: question, documents: candidates.map((c) => c.text), top_n: 5 },
  });
  const context = ranked.results.map((r) => candidates[r.index]);

  const sources = context.map((c, i) => `[${i + 1}] (${c.source})\n${c.text}`).join("\n\n");
  const res = await client.chat.completions.create({
    model: "openai/gpt-5-mini",
    messages: [
      { role: "system", content: "Answer only from the numbered sources and cite them like [1]. If they do not contain the answer, say so." },
      { role: "user", content: `Sources:\n\n${sources}\n\nQuestion: ${question}` },
    ],
  });
  return res.choices[0].message.content;
}

비용 추정

RAG 시스템에는 세 가지 비용 항목이 있으며, 모두 USD이고 usage.cost로 보고됩니다.

실행하기 전에 추정하려면 GET /models/{id}에서 모델 가격을 가져오고(pricing.prompt와 pricing.completion은 토큰당 USD), 영어 텍스트는 토큰당 대략 4자라고 가정하세요.

import httpx

def price(model_id):
    return httpx.get(f"https://api14.com/api/v1/models/{model_id}").json()["pricing"]

corpus_tokens = sum(len(c["text"]) for c in chunks) / 4
print("One-off embedding:", corpus_tokens * float(price(EMBED_MODEL)["prompt"]), "USD")

p = price(CHAT_MODEL)
prompt_tokens, answer_tokens = 5 * 800 / 4 + 200, 300  # 5 chunks + instructions, short answer
print("Per answer:", prompt_tokens * float(p["prompt"]) + answer_tokens * float(p["completion"]), "USD")

가장 큰 비용은 보통 채팅 호출입니다. 이를 줄이려면 청크를 더 적고 짧게 보내고(재순위가 이를 안전하게 해 줍니다), 프롬프트 캐싱이 적용되도록 시스템 프롬프트를 맨 앞에 고정하고, max_tokens를 설정하세요. 실제 비용은 나중에 usage.cost 또는 Generation 조회로 확인하세요.