[LangChain RAG 구축 시리즈 Ep.26] 🧩 멀티 테이블 구조에 맞춘 자동 컬렉션 선택 및 태깅 전략

RAG 시스템이 다양한 Iceberg 테이블을 다루게 되면,
단일 벡터 컬렉션에 모든 문서를 저장하는 방식은 검색 정확도와 성능의 한계에 부딪히게 됩니다.

예를 들어, products, orders, customers와 같은 테이블이 각각 존재한다면,
각 테이블의 문서를 분리 저장하고 질문에 따라 적절한 컬렉션을 자동으로 선택해야 합니다.

그래서 이 글에서는:

• ✅ 사용자 질문에서 테이블명을 자동으로 추출하고
• ✅ 해당 테이블에 맞는 Chroma 컬렉션을 자동으로 선택하여
• ✅ 정확하고 빠른 검색이 가능한 RAG 시스템을 구현합니다.

이 전략은 문서 수가 많거나 테이블이 10개 이상인 데이터 플랫폼에서 매우 유용합니다.

---

🎯 목표

• 사용자 질문에서 Iceberg 테이블명을 추출하는 로직 구현
• 테이블별 벡터 컬렉션을 자동으로 선택하는 API 설계
• 향후 GPT 기반 테이블 매핑 전략으로 확장 가능한 구조 구성

---

🧠 Step 1. 사용자 질문에서 테이블명 추출 함수

# src/utils.py

import re

def extract_table_name(question: str) -> str:
    """
    사용자의 질문에서 테이블명을 추출하는 유틸 함수입니다.
    현재는 단순 문자열 포함 방식이며, 추후 LLM 기반 분류기로 고도화할 수 있습니다.
    """

    # ✅ 사전에 정의된 테이블명 목록
    known_tables = ["products", "customers", "orders", "items"]

    # ✅ 대소문자 구분 없이 검색하기 위해 소문자로 변환
    question = question.lower()

    # ✅ 테이블명 키워드가 질문에 포함되어 있다면 해당 테이블명을 반환
    for table in known_tables:
        if table in question:
            return table

    # ✅ 일치하는 테이블명이 없을 경우 기본값 반환
    return "unknown"

---

⚙️ Step 2. FastAPI + 자동 컬렉션 선택 API 구현

# src/rag_server_dynamic.py

from fastapi import FastAPI, Request, Depends               # FastAPI 서버 및 요청 핸들링
from pydantic import BaseModel                              # 요청 데이터 구조화 (question)
from src.auth import verify_api_key                         # API 인증 미들웨어
from src.utils import extract_table_name                    # 테이블 추출 함수

# LangChain 관련 모듈
from langchain.vectorstores import Chroma                   # 벡터 검색기 (Chroma)
from langchain.embeddings import OpenAIEmbeddings           # 텍스트 임베딩
from langchain.chat_models import ChatOpenAI                # GPT-3.5 모델
from langchain.chains import RetrievalQA                    # Retrieval QA 체인

import uvicorn

# ✅ FastAPI 애플리케이션 인스턴스 생성
app = FastAPI()

# ✅ 요청 본문으로 받을 데이터 구조 (질문)
class QueryRequest(BaseModel):
    question: str

# ✅ POST API: /rag/query
@app.post("/rag/query")
async def rag_query(request: QueryRequest, _: str = Depends(verify_api_key)):
    """
    사용자 질문에서 테이블명을 자동으로 추출하고,
    해당 테이블에 맞는 컬렉션을 자동 선택하여 GPT 응답을 생성합니다.
    """

    # ✅ 클라이언트 질문 추출
    question = request.question

    # ✅ 테이블명 추출 (예: "products")
    table = extract_table_name(question)

    # ✅ Chroma 컬렉션 이름 생성
    collection_name = f"iceberg_{table}"

    # ✅ 벡터 저장소 로딩
    vectordb = Chroma(
        persist_directory="chroma_db",               # 벡터 DB가 저장된 로컬 디렉토리
        collection_name=collection_name,             # 자동 선택된 컬렉션
        embedding_function=OpenAIEmbeddings()        # 동일한 임베딩 모델 사용
    )

    # ✅ 검색기 구성 (top-2 검색)
    retriever = vectordb.as_retriever(
        search_kwargs={"k": 2}
    )

    # ✅ GPT 모델 설정
    llm = ChatOpenAI(
        model_name="gpt-3.5-turbo",
        temperature=0.2
    )

    # ✅ QA 체인 구성
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        retriever=retriever
    )

    # ✅ 질문 실행 및 응답 생성
    result = qa_chain.run(question)

    # ✅ 응답 반환
    return {
        "question": question,
        "table": table,
        "collection": collection_name,
        "answer": result
    }

# ✅ 서버 실행 (로컬 테스트용)
if __name__ == "__main__":
    uvicorn.run("src.rag_server_dynamic:app", host="0.0.0.0", port=8000, reload=True)

---

▶️ 테스트 예시

http POST http://localhost:8000/rag/query \
  Authorization:"my-secret-key" \
  question="products 테이블의 컬럼은 뭐야?"

 

📤 결과:

{
  "question": "products 테이블의 컬럼은 뭐야?",
  "table": "products",
  "collection": "iceberg_products",
  "answer": "products 테이블은 product_id, product_name, category 컬럼으로 구성되어 있습니다."
}

---

📎 요약 및 핵심 정리

• 사용자의 질문을 분석하여 Iceberg 테이블명을 자동 추출하는 구조를 구현했습니다.
• 추출된 테이블명을 기반으로 Chroma 컬렉션을 자동으로 선택하여
  더 정확하고 빠른 문서 검색이 가능해졌습니다.
• 이 구조는 향후 GPT를 활용한 테이블 매핑 전략 또는 메타데이터 기반 필터링 전략으로 확장 가능합니다.