4단계
RAG 파이프라인
35 분
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 · 정확도 +1020%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 · "~로만 답하세요" 세 가지로 대체로 잘 동작합니다. 튜닝은 실제 사용 로그를 보며 조금씩.
Next
- 05-gemini-openai-api