RAG 파이프라인

청킹 → 임베딩 → 저장 → 검색 → 프롬프트 주입 → 생성. 여섯 단계가 전부.

1. 청킹 — 문서 자르기

LLM 컨텍스트는 제한되고, 너무 긴 청크는 임베딩 품질도 떨어집니다.

def chunk_by_sentence(text: str, max_chars=1000, overlap=100):
    sentences = re.split(r'(?<=[.!?。])\s+', text)
    chunks, cur = [], ""
    for s in sentences:
        if len(cur) + len(s) > max_chars:
            chunks.append(cur)
            cur = cur[-overlap:] + " " + s
        else:
            cur += " " + s
    if cur: chunks.append(cur)
    return chunks

1000 자 (약 500 토큰) 기준 · 100 자 오버랩
문장 경계 존중 · 코드 블록은 별도 보존

2. 적재 — 배치 임베딩

async def index_document(doc_id: int, text: str):
    chunks = chunk_by_sentence(text)
    for i in range(0, len(chunks), 10):
        batch = chunks[i:i+10]
        resp = genai.embed_content(
            model="models/text-embedding-004",
            content=batch, task_type="retrieval_document",
        )
        async with pool.acquire() as con:
            await con.executemany(
                "INSERT INTO document_chunks (document_id, chunk_index, content, embedding) VALUES ($1, $2, $3, $4)",
                [(doc_id, i+j, c, v) for j, (c, v) in enumerate(zip(batch, resp["embedding"]))]
            )

10 ~ 50 개 배치. API rate limit 주의
실패 시 부분 적재 복구 위해 document_id 단위 트랜잭션

3. 검색 — top-k

async def retrieve(query: str, k=5):
    q_emb = genai.embed_content(
        model="models/text-embedding-004",
        content=query, task_type="retrieval_query",
    )["embedding"]
    rows = await pool.fetch(
        """SELECT content, 1 - (embedding <=> $1::vector) AS score
           FROM document_chunks ORDER BY embedding <=> $1::vector LIMIT $2""",
        q_emb, k,
    )
    return [(r["content"], r["score"]) for r in rows]

top-k 는 3 ~ 10 이 실용 범위. 더 많이 가져오면 컨텍스트 희석.

4. (선택) rerank

top-k=20 → rerank 모델로 top-5 재정렬. 품질이 중요한 프로덕션에서 유용.

Cohere Rerank · BGE-Reranker · cross-encoder
응답 latency +200~~500ms · 정확도 +10~~20%p

MVP 에서는 생략.

5. 프롬프트 주입

def build_prompt(query: str, chunks: list[str]):
    context = "\n\n---\n\n".join(chunks)
    return f"""당신은 아래 문서를 근거로만 답하는 어시스턴트입니다.
문서에 없는 내용은 "문서에서 찾을 수 없습니다" 라고 답하세요.

# 문서
{context}

# 질문
{query}

# 답변 (인용 포함)
"""

"~로만" · "찾을 수 없습니다" · 인용 요구 — hallucination 방지 3 요소.

6. 생성

def generate(prompt: str) -> str:
    resp = openai_client.chat.completions.create(
        model="gemma-2-9b-it",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,
        max_tokens=500,
    )
    return resp.choices[0].message.content

temperature 0.3 정도가 RAG 의 안정 범위.

7. 전체 플로우

async def ask(query: str):
    chunks = await retrieve(query, k=5)
    prompt = build_prompt(query, [c for c, _ in chunks])
    answer = generate(prompt)
    return {"answer": answer, "sources": chunks}

응답에 sources 를 포함하면 UI 에서 "이 답변의 근거" 를 클릭 펼치기 가능.

8. 자주 걸리는 자리

청크가 너무 작음 — 의미 단위 깨짐
청크가 너무 큼 — 임베딩 희석 · 컨텍스트 초과
검색 점수 낮은데 그대로 주입 — threshold (score < 0.5 skip) 필요
citations 요청하지 않음 — 환각이 늘어남

하고픈 말

첫 RAG 은 top-k=5 · temperature 0.3 · "~로만 답하세요" 세 가지로 대체로 잘 동작합니다. 튜닝은 실제 사용 로그를 보며 조금씩.

05-gemini-openai-api