RAG 쿼리 변환 (RAG Query Transformation, RAG 쿼리 변환)

사용자 질문
    │
    ▼
┌────────────────────────┐
│ 쿼리 변환 (LLM 호출 등)  │  재작성 / 독립화 / multi-query / HyDE / 분해 / step-back / 라우팅
└───────────┬────────────┘
            ▼
     검색용 쿼리 1~N개
            │
            ▼
        검색 → (리랭킹) → LLM 생성
                           ▲
사용자 원래 질문 ───────────┘   ← 답변 생성에는 보통 원래 질문을 그대로 쓴다

질문: "회사 노트북 잃어버렸는데 어떡해요 ㅠㅠ"
문서: "IT 자산 분실 시 처리 절차: 분실 인지 즉시 보안팀(내선 1234)에 신고하고..."

사용자: 연차는 1년에 며칠이에요?
봇:     입사 1년 이상이면 15일입니다.
사용자: 그럼 반차는요?

"작년 대비 올해 A팀과 B팀의 예산 증가율을 비교해 줘"

원래:   "회사 노트북 잃어버렸는데 어떡해요 ㅠㅠ"
재작성: "업무용 노트북(IT 자산) 분실 시 신고 및 처리 절차"

from openai import OpenAI

client = OpenAI()
MODEL = "gpt-4o-mini"  # 쿼리 변환에는 작고 빠른 모델이면 충분한 경우가 많다

def llm(prompt: str) -> str:
    resp = client.chat.completions.create(
        model=MODEL,
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    return resp.choices[0].message.content.strip()

REWRITE = """다음 질문을 사내 문서 검색에 쓸 검색어 문장으로 다시 쓰세요.
공식 용어를 쓰고, 불필요한 말은 빼고, 한 줄로만 출력하세요.

질문: {q}"""

search_query = llm(REWRITE.format(q="회사 노트북 잃어버렸는데 어떡해요 ㅠㅠ"))

대화 기록:
  사용자: 연차는 1년에 며칠이에요?
  봇:     입사 1년 이상이면 15일입니다.
현재 질문: "그럼 반차는요?"

독립 질문: "반차를 사용하면 연차 15일에서 얼마나 차감되나요?"

CONDENSE = """대화 기록과 마지막 질문이 주어집니다.
마지막 질문을 대화 기록 없이도 이해할 수 있는 독립된 질문 하나로 다시 쓰세요.
대명사와 생략된 대상을 구체적으로 채우세요. 질문만 출력하세요.

대화 기록:
{history}

마지막 질문: {q}"""

def standalone(history: list[tuple[str, str]], q: str) -> str:
    if not history:
        return q  # 첫 질문은 변환할 필요 없음
    h = "\n".join(f"{role}: {text}" for role, text in history[-6:])  # 최근 몇 턴만
    return llm(CONDENSE.format(history=h, q=q))

원래 질문: "재택근무 중 장비 지원 받을 수 있나요?"

생성된 쿼리
 q1: "재택근무자 장비 지원 정책"
 q2: "원격근무 모니터 키보드 구매 비용 지원"
 q3: "재택 근무 환경 지원금 신청 방법"

      q1 검색  ─┐
      q2 검색  ─┼─> 결과 합치기 (RRF) ─> top-k
      q3 검색  ─┘

RRF 점수(d) = Σ  1 / (k + rank_i(d))      (k는 보통 60)
             i: 각 결과 목록

질문: "회사 노트북 잃어버렸는데 어떡해요"
           │
           ▼  LLM: "이 질문에 답하는 사내 문서를 써 봐"
가상 문서: "업무용 노트북을 분실한 경우, 즉시 보안팀에 신고해야 합니다.
           신고 후 원격 잠금 및 데이터 삭제 조치가 진행되며, 경위서를 제출..."
           │
           ▼  임베딩
       벡터 검색 → 실제 "IT 자산 분실 처리 절차" 문서

import numpy as np
from sentence_transformers import SentenceTransformer

embedder = SentenceTransformer("BAAI/bge-m3")

HYDE = """다음 질문에 답하는 사내 규정 문서의 한 단락을 작성하세요.
정확하지 않아도 됩니다. 실제 문서처럼 공식적인 문체로 쓰세요.

질문: {q}"""

def hyde_vector(q: str) -> np.ndarray:
    fake_doc = llm(HYDE.format(q=q))
    return embedder.encode(fake_doc, normalize_embeddings=True)

질문: "작년 대비 올해 A팀과 B팀의 예산 증가율을 비교해 줘"

하위 질문
 1. A팀 2025년 예산
 2. A팀 2026년 예산
 3. B팀 2025년 예산
 4. B팀 2026년 예산

각각 검색 → 문서 모음 → LLM이 계산·비교해서 답변

DECOMPOSE = """다음 질문에 답하려면 어떤 정보를 각각 찾아야 하는지
독립적으로 검색 가능한 하위 질문으로 나누세요. 최대 {n}개, 한 줄에 하나씩.
나눌 필요가 없으면 원래 질문 하나만 출력하세요.

질문: {q}"""

def decompose_search(q: str, search, n: int = 4) -> list[str]:
    subs = [s for s in llm(DECOMPOSE.format(q=q, n=n)).splitlines() if s.strip()]
    docs: list[str] = []
    for s in subs:
        for d in search(s)[:3]:
            if d not in docs:
                docs.append(d)
    return docs

원래 질문:  "v2.3.1에서 비밀번호 5회 틀리면 왜 30분 잠기나요?"
step-back: "계정 잠금 정책은 어떻게 동작하나요?"

원래 질문으로 검색   → 릴리스 노트 v2.3.1 (구체적 변경 사항)
step-back으로 검색  → 보안 정책 문서 (잠금 규칙의 원리와 이유)
                       ↓
          두 결과를 함께 LLM에 제공

STEP_BACK = """다음 질문의 배경이 되는 더 일반적인 개념이나 원리를 묻는 질문 하나를 만드세요.
질문만 출력하세요.

질문: {q}"""

def step_back_search(q: str, search, k: int = 3) -> list[str]:
    general = llm(STEP_BACK.format(q=q))
    return list(dict.fromkeys(search(q)[:k] + search(general)[:k]))  # 순서 유지 중복 제거

                    ┌──> 인사 규정 인덱스
                    │
질문 ──> 라우터 ─────┼──> 기술 문서 인덱스
       (LLM 또는     │
        분류 모델)   ├──> SQL DB (매출, 수치 데이터)
                    │
                    └──> 검색 없이 바로 답변 (인사말, 잡담)

ROUTES = {
    "hr": "휴가, 급여, 복리후생, 인사 평가 등 인사 규정",
    "tech": "개발 가이드, API, 배포, 장애 대응 등 기술 문서",
    "sql": "매출, 주문 수, 사용자 수 등 숫자 데이터 조회",
    "none": "인사, 잡담 등 문서 검색이 필요 없는 대화",
}

ROUTE = """질문을 처리할 경로 하나를 고르세요. 경로 이름만 출력하세요.

경로:
{routes}

질문: {q}"""

def route(q: str) -> str:
    routes = "\n".join(f"- {k}: {v}" for k, v in ROUTES.items())
    choice = llm(ROUTE.format(routes=routes, q=q)).strip().lower()
    return choice if choice in ROUTES else "hr"  # 알 수 없는 출력은 기본 경로로