Natural Language to SQL: 자연어로 데이터베이스 조회하기

데이터베이스에서 원하는 정보를 찾기 위해 복잡한 SQL 쿼리를 작성하는 것은 개발자에게도 때로는 번거로운 일입니다. 하물며 비개발자들에게는 더욱 어려운 일이죠. 이런 문제를 해결하기 위해 등장한 것이 바로 Natural Language to SQL(NL2SQL) 기술입니다.

NL2SQL이란?

NL2SQL은 자연어로 작성된 질문을 SQL 쿼리로 자동 변환하는 기술입니다. 예를 들어:

• 자연어: "지난달 매출이 가장 높은 제품은 무엇인가?"
• SQL: SELECT product_name FROM sales WHERE sale_date >= '2024-05-01' ORDER BY amount DESC LIMIT 1;

이처럼 일상 언어로 질문하면 시스템이 자동으로 해당하는 SQL 쿼리를 생성해줍니다.

핵심 동작 원리

1. 자연어 이해 (Natural Language Understanding)

사용자의 질문에서 핵심 요소들을 추출합니다:

• 의도: 조회, 집계, 비교 등
• 엔티티: 테이블명, 컬럼명에 해당하는 개념들
• 조건: 필터링, 정렬, 그룹화 조건들

2. 스키마 매핑 (Schema Mapping)

자연어 개념을 실제 데이터베이스 스키마와 연결합니다:

• "제품" → products 테이블
• "매출" → sales.amount 컬럼
• "지난달" → 날짜 범위 조건

3. SQL 생성 (SQL Generation)

파악된 정보를 바탕으로 실행 가능한 SQL 쿼리를 생성합니다.

실제 구현 예제

간단한 NL2SQL 시스템을 Python으로 구현해보겠습니다. 이 예제는 교육 목적으로 만든 것이며, 실제 프로덕션에서는 더 정교한 처리가 필요합니다.

기본 설정

import re
import sqlite3
from datetime import datetime, timedelta
from typing import Dict, List, Tuple

class SimpleNL2SQL:
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path)
        self.schema_mapping = {
            # 테이블 매핑
            '제품': 'products',
            '상품': 'products',
            '판매': 'sales',
            '매출': 'sales',
            '고객': 'customers',
            '사용자': 'customers',
            
            # 컬럼 매핑
            '이름': 'name',
            '가격': 'price',
            '금액': 'amount',
            '날짜': 'date',
            '수량': 'quantity'
        }
        
        # 의도 패턴
        self.intent_patterns = {
            'select': [r'무엇', r'어떤', r'보여줘', r'알려줘', r'찾아줘'],
            'count': [r'몇 개', r'개수', r'얼마나'],
            'max': [r'가장 높은', r'최대', r'최고'],
            'min': [r'가장 낮은', r'최소', r'최저'],
            'sum': [r'총합', r'합계', r'전체']
        }
        
        # 조건 패턴
        self.condition_patterns = {
            'last_month': [r'지난달', r'저번달'],
            'this_month': [r'이번달', r'현재달'],
            'greater_than': [r'이상', r'보다 큰'],
            'less_than': [r'이하', r'보다 작은']
        }

자연어 파싱 함수

    def parse_natural_language(self, question: str) -> Dict:
        """자연어 질문을 파싱하여 의도와 조건을 추출"""
        result = {
            'intent': 'select',
            'table': None,
            'columns': [],
            'conditions': [],
            'order_by': None,
            'limit': None
        }
        
        # 의도 파악
        for intent, patterns in self.intent_patterns.items():
            if any(re.search(pattern, question) for pattern in patterns):
                result['intent'] = intent
                break
        
        # 테이블 매핑
        for korean_term, english_term in self.schema_mapping.items():
            if korean_term in question:
                if english_term in ['products', 'sales', 'customers']:
                    result['table'] = english_term
                else:
                    result['columns'].append(english_term)
        
        # 조건 처리
        if any(re.search(pattern, question) for pattern in self.condition_patterns['last_month']):
            last_month = datetime.now() - timedelta(days=30)
            result['conditions'].append(f"date >= '{last_month.strftime('%Y-%m-%d')}'")
        
        # 정렬 및 제한
        if result['intent'] in ['max', 'min']:
            order_direction = 'DESC' if result['intent'] == 'max' else 'ASC'
            if 'amount' in result['columns'] or '금액' in question:
                result['order_by'] = f"ORDER BY amount {order_direction}"
            elif 'price' in result['columns'] or '가격' in question:
                result['order_by'] = f"ORDER BY price {order_direction}"
            result['limit'] = "LIMIT 1"
        
        return result

SQL 생성 함수

    def generate_sql(self, parsed_data: Dict) -> str:
        """파싱된 데이터를 기반으로 SQL 쿼리 생성"""
        # 기본 SELECT 구조
        if parsed_data['intent'] == 'count':
            select_clause = "SELECT COUNT(*)"
        elif parsed_data['intent'] in ['sum']:
            column = parsed_data['columns'][0] if parsed_data['columns'] else 'amount'
            select_clause = f"SELECT SUM({column})"
        else:
            columns = ', '.join(parsed_data['columns']) if parsed_data['columns'] else '*'
            select_clause = f"SELECT {columns}"
        
        # FROM 절
        table = parsed_data['table'] or 'sales'  # 기본 테이블
        from_clause = f"FROM {table}"
        
        # WHERE 절
        where_clause = ""
        if parsed_data['conditions']:
            where_clause = f"WHERE {' AND '.join(parsed_data['conditions'])}"
        
        # ORDER BY 절
        order_clause = parsed_data['order_by'] or ""
        
        # LIMIT 절
        limit_clause = parsed_data['limit'] or ""
        
        # 최종 쿼리 조합
        query_parts = [select_clause, from_clause, where_clause, order_clause, limit_clause]
        sql_query = ' '.join(part for part in query_parts if part)
        
        return sql_query
    
    def query(self, question: str) -> Tuple[str, List]:
        """자연어 질문을 SQL로 변환하고 실행"""
        try:
            parsed = self.parse_natural_language(question)
            sql_query = self.generate_sql(parsed)
            
            cursor = self.conn.cursor()
            cursor.execute(sql_query)
            results = cursor.fetchall()
            
            return sql_query, results
        
        except Exception as e:
            return f"Error: {str(e)}", []

테스트 데이터베이스 설정

def setup_test_database():
    """테스트용 데이터베이스 생성"""
    conn = sqlite3.connect('test.db')
    cursor = conn.cursor()
    
    # 테이블 생성
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY,
            name TEXT,
            price REAL,
            category TEXT
        )
    ''')
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS sales (
            id INTEGER PRIMARY KEY,
            product_id INTEGER,
            amount REAL,
            quantity INTEGER,
            date TEXT,
            FOREIGN KEY (product_id) REFERENCES products (id)
        )
    ''')
    
    # 샘플 데이터 삽입
    products_data = [
        (1, '노트북', 1200000, '전자제품'),
        (2, '마우스', 25000, '전자제품'),
        (3, '키보드', 80000, '전자제품'),
        (4, '모니터', 300000, '전자제품')
    ]
    
    sales_data = [
        (1, 1, 1200000, 1, '2024-05-15'),
        (2, 2, 50000, 2, '2024-05-16'),
        (3, 3, 160000, 2, '2024-05-17'),
        (4, 1, 2400000, 2, '2024-05-18'),
        (5, 4, 300000, 1, '2024-06-01')
    ]
    
    cursor.executemany('INSERT OR REPLACE INTO products VALUES (?, ?, ?, ?)', products_data)
    cursor.executemany('INSERT OR REPLACE INTO sales VALUES (?, ?, ?, ?, ?)', sales_data)
    
    conn.commit()
    conn.close()

사용 예제

# 데이터베이스 설정
setup_test_database()

# NL2SQL 시스템 초기화
nl2sql = SimpleNL2SQL('test.db')

# 테스트 질문들
test_questions = [
    "지난달 매출이 가장 높은 제품은 무엇인가?",
    "전체 판매 개수는 몇 개인가?",
    "모든 제품을 보여줘",
    "총 매출 합계는 얼마인가?"
]

print("=== NL2SQL 테스트 결과 ===\n")

for question in test_questions:
    print(f"질문: {question}")
    sql_query, results = nl2sql.query(question)
    print(f"생성된 SQL: {sql_query}")
    print(f"결과: {results}")
    print("-" * 50)

실행 결과:

=== NL2SQL 테스트 결과 ===

질문: 지난달 매출이 가장 높은 제품은 무엇인가?
생성된 SQL: SELECT * FROM sales WHERE date >= '2024-05-03' ORDER BY amount DESC LIMIT 1
결과: [(4, 1, 2400000.0, 2, '2024-05-18')]

질문: 전체 판매 개수는 몇 개인가?
생성된 SQL: SELECT COUNT(*) FROM sales
결과: [(5,)]

질문: 모든 제품을 보여줘
생성된 SQL: SELECT * FROM products
결과: [(1, '노트북', 1200000.0, '전자제품'), (2, '마우스', 25000.0, '전자제품'), ...]

고급 기능 구현

LLM 기반 NL2SQL

실제 프로덕션 환경에서는 OpenAI GPT나 다른 LLM을 활용할 수 있습니다:

import openai
from typing import Dict, Any

class LLMBasedNL2SQL:
    def __init__(self, api_key: str, db_schema: Dict[str, Any]):
        self.client = openai.OpenAI(api_key=api_key)
        self.db_schema = db_schema
    
    def generate_sql_with_llm(self, question: str) -> str:
        """LLM을 사용해 자연어를 SQL로 변환"""
        
        schema_description = self._format_schema_for_prompt()
        
        prompt = f"""
        다음 데이터베이스 스키마를 참고하여 자연어 질문을 SQL 쿼리로 변환해주세요.
        
        데이터베이스 스키마:
        {schema_description}
        
        질문: {question}
        
        응답은 실행 가능한 SQL 쿼리만 반환해주세요.
        """
        
        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "당신은 자연어를 SQL로 변환하는 전문가입니다."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.1
        )
        
        return response.choices[0].message.content.strip()
    
    def _format_schema_for_prompt(self) -> str:
        """스키마 정보를 프롬프트용으로 포맷팅"""
        schema_text = ""
        for table_name, table_info in self.db_schema.items():
            schema_text += f"테이블: {table_name}\n"
            schema_text += f"컬럼: {', '.join(table_info['columns'])}\n\n"
        return schema_text

주요 도전과제와 해결 방안

1. 스키마 복잡성

실제 데이터베이스는 수백 개의 테이블을 가질 수 있습니다. 이를 해결하기 위해:

• 스키마 임베딩: 테이블과 컬럼 정보를 벡터로 변환하여 의미적 유사성 기반 매칭
• 계층적 스키마 탐색: 관련성 높은 테이블부터 우선 고려

2. 모호성 해결

"큰 주문"이 금액인지 수량인지 명확하지 않은 경우:

• 컨텍스트 학습: 이전 대화 내용 참고
• 사용자 확인: 모호한 경우 명확화 질문

3. 복합 쿼리 처리

여러 조건과 조인이 포함된 복잡한 질문:

• 단계별 분해: 복잡한 질문을 단순한 하위 질문으로 분해
• 템플릿 기반 접근: 자주 사용되는 쿼리 패턴을 템플릿화

실제 활용 사례

1. 비즈니스 인텔리전스

# 예시: 대시보드에서 자연어 질의
questions = [
    "이번 분기 매출 TOP 10 제품은?",
    "작년 대비 매출 증가율은?",
    "지역별 고객 분포는 어떻게 되나?"
]

2. 고객 서비스 챗봇

# 예시: 고객이 주문 상태를 자연어로 문의
def handle_customer_query(customer_id: str, question: str):
    # "내 지난달 주문 내역 보여줘" -> SQL 변환
    personalized_question = f"고객 ID {customer_id}의 {question}"
    return nl2sql.query(personalized_question)

성능 최적화 팁

1. 캐싱 전략

from functools import lru_cache

class OptimizedNL2SQL(SimpleNL2SQL):
    @lru_cache(maxsize=1000)
    def cached_query(self, question: str):
        """자주 사용되는 질문을 캐싱"""
        return self.query(question)

2. 인덱스 최적화

-- 자주 조회되는 컬럼에 인덱스 생성
CREATE INDEX idx_sales_date ON sales(date);
CREATE INDEX idx_sales_amount ON sales(amount);

마무리

NL2SQL 기술은 데이터 접근성을 크게 향상시키는 혁신적인 기술입니다. 간단한 규칙 기반 시스템부터 최신 LLM을 활용한 고도화된 시스템까지, 프로젝트의 요구사항에 맞는 적절한 수준에서 구현할 수 있습니다.