RAG와 하네스 엔지니어링: 검색 증강 생성 파이프라인을 감싸는 실행 스캐폴드

RAG(Retrieval-Augmented Generation, 검색 증강 생성)는 LLM이 답변을 생성하기 전에 외부 데이터 저장소에서 관련 문서를 먼저 검색해서, 그 내용을 프롬프트에 끼워 넣어 답변의 근거로 삼게 하는 구조다. 모델을 다시 학습시키지 않고도 최신 정보나 사내 문서 같은 걸 답변에 반영할 수 있어서, 챗봇이나 문서 기반 QA 서비스 대부분이 이 구조를 쓴다. 동작 원리는 비교적 단순하다 — 문서를 임베딩 모델로 벡터화해서 벡터 DB에 저장해두고, 사용자 질문이 들어오면 그 질문도 같은 방식으로 벡터화한 뒤 유사도가 가장 높은 문서 몇 개를 꺼내와서, "다음 문서를 참고해서 답해줘"라는 형태로 프롬프트에 이어 붙인다.

하네스란 무엇인가

하네스(harness)는 원래 테스트하려는 대상을 실행 가능한 상태로 세팅해주고, 입력을 넣고, 출력을 받는 주변 장치를 가리키는 소프트웨어 엔지니어링 용어다. 유닛 테스트 프레임워크가 함수 하나를 격리된 환경에서 반복 실행해주는 것과 같은 역할을, AI 시스템 전체에도 그대로 적용한 게 요즘 말하는 "AI 하네스"다. 모델 자체는 그냥 "입력을 받아 텍스트를 뱉는 함수"에 가깝고, 실제로 쓸만한 서비스가 되려면 그 앞뒤로 입력 준비, 재시도/타임아웃, 로깅, 도구 호출 오케스트레이션 같은 걸 감싸는 코드가 필요하다 — 이 감싸는 코드 전체를 하네스라고 부른다. 코딩 에이전트 CLI도 결국 "모델 호출 + 도구 실행 루프 + 권한 체크 + 로깅"을 감싸는 하네스이고, RAG도 여기에 "검색"이라는 단계가 하나 더 끼어드는 구조일 뿐이다.

RAG 하네스가 필요한 이유

RAG 파이프라인은 임베딩 API, 벡터 DB, LLM API라는 세 개의 서로 다른 외부 의존성을 순서대로 거친다. 각각이 독립적으로 느려지거나 실패할 수 있는 네트워크 호출이라, "검색 결과를 프롬프트에 붙여서 모델을 호출한다"는 단 세 줄짜리 로직도 실제로 안정적으로 돌리려면 신경 쓸 게 꽤 많아진다.

• 재현성: 같은 질문을 다시 넣었을 때 어떤 문서가 검색됐고 어떤 프롬프트가 조립됐는지 다시 확인할 수 있어야, 답변이 이상할 때 원인이 검색 단계인지 생성 단계인지 구분할 수 있다
• 관측성: 검색된 문서, 유사도 점수, 최종 프롬프트, 모델 응답, 각 단계별 소요 시간을 구조화된 형태로 남겨야 나중에 디버깅이나 품질 분석이 가능하다
• 신뢰성: 임베딩 API나 벡터 DB, LLM 호출 중 하나가 타임아웃 나도 전체 요청이 그냥 죽어버리지 않도록 재시도와 폴백 경로가 필요하다
• 교체 가능성: 임베딩 모델이나 벡터 DB, LLM 제공사를 나중에 바꾸더라도 파이프라인의 나머지 로직은 그대로 두고 어댑터 부분만 갈아 끼울 수 있어야 유지보수가 된다

이 네 가지를 각 컴포넌트마다 따로 구현하는 대신, 공통 스캐폴드 하나로 감싸서 어떤 RAG 애플리케이션이든 그 위에 얹어 쓸 수 있게 만드는 게 RAG 하네스의 목적이다.

실제로 짜보기

아래는 최소한의 형태로 짠 파이썬 예시다. 임베딩 모델이나 벡터 DB, LLM API 클라이언트는 실제 서비스(예: text-embedding-3-small + Chroma/pgvector + 실제 LLM API)로 갈아 끼울 수 있도록 인터페이스만 얇게 감싸서, 하네스 자체는 그 아래 구현에 의존하지 않게 짰다.

import time
import logging
from dataclasses import dataclass, field
from typing import Protocol

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rag_harness")

@dataclass
class Document:
    id: str
    text: str
    score: float = 0.0

@dataclass
class Trace:
    """한 번의 요청에서 각 단계가 어떻게 진행됐는지 담는 관측 기록."""
    query: str
    retrieved: list[Document] = field(default_factory=list)
    prompt: str = ""
    response: str = ""
    timings_ms: dict = field(default_factory=dict)

# --- 각 단계를 인터페이스로 분리 (구현체는 나중에 자유롭게 교체) ---
class Retriever(Protocol):
    def search(self, query: str, k: int) -> list[Document]: ...

class Generator(Protocol):
    def generate(self, prompt: str) -> str: ...

def build_prompt(query: str, docs: list[Document]) -> str:
    context = "\n\n".join(f"[{d.id}] {d.text}" for d in docs)
    return f"""다음 문서를 참고해서 사용자 질문에 답해줘.

[문서]
{context}

[질문]
{query}"""

# --- 하네스: 재시도, 타임 측정, 로깅, 트레이스 수집을 한곳에서 담당 ---
class RagHarness:
    def __init__(self, retriever: Retriever, generator: Generator, k: int = 3, max_retries: int = 2):
        self.retriever = retriever
        self.generator = generator
        self.k = k
        self.max_retries = max_retries

    def _with_retry(self, fn, *args):
        last_err = None
        for attempt in range(self.max_retries + 1):
            try:
                return fn(*args)
            except Exception as e:
                last_err = e
                logger.warning(f"attempt {attempt} failed: {e}")
                time.sleep(0.2 * (attempt + 1))
        raise last_err

    def run(self, query: str) -> Trace:
        trace = Trace(query=query)

        t0 = time.perf_counter()
        trace.retrieved = self._with_retry(self.retriever.search, query, self.k)
        trace.timings_ms["retrieve"] = round((time.perf_counter() - t0) * 1000, 1)

        trace.prompt = build_prompt(query, trace.retrieved)

        t1 = time.perf_counter()
        trace.response = self._with_retry(self.generator.generate, trace.prompt)
        trace.timings_ms["generate"] = round((time.perf_counter() - t1) * 1000, 1)

        logger.info(
            f"query={query!r} docs={[d.id for d in trace.retrieved]} "
            f"timings={trace.timings_ms}"
        )
        return trace

# --- 데모용 더미 구현체 (실무에서는 실제 임베딩/벡터DB/LLM 클라이언트로 교체) ---
class DummyRetriever:
    def __init__(self, corpus: list[Document]):
        self.corpus = corpus

    def search(self, query: str, k: int) -> list[Document]:
        scored = sorted(
            self.corpus,
            key=lambda d: len(set(query) & set(d.text)),
            reverse=True,
        )
        return scored[:k]

class DummyGenerator:
    def generate(self, prompt: str) -> str:
        return "문서를 참고해서 요약한 답변입니다."

if __name__ == "__main__":
    corpus = [
        Document("d1", "회사 휴가 정책: 연차는 입사 1년 후부터 발생합니다."),
        Document("d2", "회사 원격근무 정책: 주 3일까지 재택근무가 가능합니다."),
    ]
    harness = RagHarness(DummyRetriever(corpus), DummyGenerator())
    trace = harness.run("원격근무 정책이 뭐야?")
    print(trace.response)

핵심은 RagHarness 클래스가 검색과 생성 로직 자체는 전혀 모른다는 점이다. Retriever와 Generator는 그냥 search/generate 메서드만 있으면 되는 인터페이스라서, DummyRetriever/DummyGenerator 자리에 실제 벡터 DB 클라이언트와 LLM API 클라이언트를 꽂아 넣기만 하면 재시도·타이밍·로깅·트레이스 수집은 그대로 재사용된다. 새로운 RAG 프로젝트를 시작할 때마다 이 배관 코드를 다시 짜는 대신, 하네스 하나를 여러 프로젝트에서 공유하는 식으로 쓸 수 있다.

하네스 위에 얹을 수 있는 것들

이렇게 파이프라인을 하네스로 감싸두면, 나중에 필요해지는 것들을 각 단계 사이에 끼워 넣기가 훨씬 쉬워진다. 예를 들어 run() 안에 검색 결과를 로깅하는 지점이 이미 있으니, 거기에 응답 품질을 자동 채점하는 평가 훅을 붙이거나, 프롬프트에 들어가기 전 문서 내용을 검증하는 필터를 끼워 넣거나, 도구 호출이 있는 에이전트형 RAG라면 그 사이에 권한 체크 로직을 두는 것도 같은 자리에 자연스럽게 얹힌다. 파이프라인 로직과 이런 부가 기능이 하네스 레벨에서 한 번에 분리되어 있으면, 나중에 어떤 요구사항이 추가되든 핵심 코드를 건드리지 않고 하네스 쪽만 확장하면 된다는 게 이 구조의 실질적인 이점이다.