[NL-to-SQL] Quest 2: Zero-Shot Baseline — 전체 스키마 때려박기

지난 편

[NL-to-SQL] Quest 1: 환경 세팅 + DB + 시맨틱 레이어 + pgvector RAGPostgreSQL + pgvector 환경을 만들고, Snowflake 스타일 시맨틱 레이어 YAML을 엔티티 단위로 분해해서 Solar Embedding으로 적재하기까지

Quest 1에서 시맨틱 레이어 + pgvector RAG 파이프라인까지 만들어놨는데 일단은 RAG 적용없이 진행하도록 하겠습니다.

RAG로 개선한다고 했을 때 “얼마나 개선됐냐” 를 증명하려면 Before -> After 가 필요하겠죠? 비교 대상 없는 정확도 숫자는 그냥 숫자일 뿐이니까요 !

그래서 Quest 2는 의도적으로 RAG 없이, 무식하게 전체 스키마를 통째로 프롬프트에 넣고 SQL 생성하는 Naive Baseline부터 진행해 보겠습니다. 그리고 반복 테스트를 편하게 하려고 Textual TUI도 함께 만들었습니다 ✌️

Naive Baseline ?

전략은 아주 단순합니다.

테이블 18개 전체 DDL을 항상 프롬프트에 주입 — 질문이 orders 한 테이블만 봐도 되든, 조인이 필요하든 무조건 전부 다 넣고 LLM이 테이블을 선택하게
질문과 관련 없는 테이블도 전부 포함 → 토큰 낭비가 구조적으로 발생. 노이즈.
RAG 없이, 예시(few-shot) 없이, 순수 제로샷 — 오직 스키마 + 질문만

Quest 1에서 만든 이커머스 DB는 18개 테이블인데, 그 DDL 전체를 프롬프트에 넣으면 질문당 약 2,000 토큰 정도 나오는 것을 확인할 수 있습니다.

항목	값
샘플 수	20 (전체 평가)
최소	1,971 토큰
최대	2,139 토큰
평균	2,037 토큰

참고로 구성요소별로 분해해보면:

schema.sql (DDL 18테이블) 단독: ~1,521 토큰
schema.sql + schema_meta.sql: ~1,706 토큰
위에 SYSTEM_PROMPT + 마크다운 래퍼 + 질문(평균 한 줄) 추가 → 실측 ~2,037 토큰

파이프라인 구현

프롬프트 설계

시스템 프롬프트는 간결하게 잡았습니다.

# prompts/zero_shot.py
SYSTEM_PROMPT = """\
You are an expert PostgreSQL SQL generator.

Rules:
- Output exactly ONE SQL query inside a fenced code block: ```sql ... ```
- Use only SELECT statements. Never use DROP, DELETE, UPDATE, INSERT, ALTER, TRUNCATE, or any DDL/DML.
- All table references must include the schema prefix (e.g. ecommerce.orders).
- Use proper PostgreSQL syntax (DATE_TRUNC, INTERVAL, EXTRACT, etc.).
- Do not explain; return only the fenced SQL block.
"""

User 프롬프트 쪽은 조금 더 단순합니다.

# prompts/zero_shot.py
def build_user_prompt(question: str, schema_ddl: str) -> str:
    return (
        "## Database Schema\n"
        f"```sql\n{schema_ddl}\n```\n"
        "## Question\n"
        f"{question}"
    )

schema_ddl은 파이프라인 초기화 시점에 한 번 읽어둔 값을 그대로 재사용 했습니다.

LLM 호출 — `generator.py`

generator.py는 LLM 호출 + 응답에서 SQL 추출을 담당합니다. 반환 타입은 데이터클래스 하나로 묶었습니다.

# src/generator.py
@dataclass
class GeneratedSQL:
    sql: str              # 최종 추출된 SQL
    raw_response: str     # LLM 원본 응답 전체
    usage: dict | None    # 토큰 사용량

디버깅을 목적으로 raw_response까지 보관했습니다. SQL 파싱이 잘못됐을 때 원본을 봐야 어디서 꼬였는지 알 수 있습니다. usage는 Quest 3에서 baseline 대비 토큰 절감을 비교할 때 쓸 거라 미리 추가했습니다!

generate_sql(question, schema_ddl) 함수 자체도 단순합니다 👍

# src/generator.py
def generate_sql(
    question: str,
    schema_ddl: str,
    *,
    client: OpenAI | None = None,
    model: str = "solar-pro",
) -> GeneratedSQL:
    client = client or get_default_client()
    response = client.chat.completions.create(
        model=model,
        temperature=0,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": build_user_prompt(question, schema_ddl)},
        ],
    )
    raw = response.choices[0].message.content or ""
    return GeneratedSQL(sql=extract_sql(raw), raw_response=raw, usage=response.usage.model_dump())

핵심 포인트 몇 개만 짚으면:

temperature=0 — baseline은 결정론적이어야 합니다. 같은 질문에 같은 SQL이 나와야 평가가 의미가 있습니다.
client·model 주입 가능 — 테스트할 때 fake client로 바꿔치기할 수 있습니다.
usage 보존 — Solar API의 prompt_tokens / completion_tokens / total_tokens를 그대로 가져옵니다.

SQL 추출 — 엣지 케이스가 있었습니다

LLM 응답에서 SQL을 꺼내는 부분은 생각보다 엣지 케이스가 있었습니다. 그냥 Structured Output 으로 할걸 후회중입니다..

# src/generator.py
def extract_sql(text: str) -> str:
    m = re.search(r"```sql\s*(.*?)```", text, re.DOTALL | re.IGNORECASE)
    if m:
        sql = m.group(1).strip()
    else:
        m = re.search(r"```\s*(.*?)```", text, re.DOTALL)
        if m:
            sql = m.group(1).strip()
        else:
            sql = text.strip()

    # 다중 세미콜론 문장이면 첫 문장만
    statements = [s.strip() for s in sql.split(";") if s.strip()]
    return statements[0] + ";" if statements else sql

우선순위는 ```sql → ``` → 텍스트 전체 순서입니다. LLM이 가끔 SET search_path = ...; 같은 구문을 SQL 앞에 붙이는 경우가 있어서, 세미콜론으로 분리해서 첫 번째 문장만 가져오도록 처리했습니다.

안전장치 — 위험 키워드 차단

SELECT 외에 다른 구문이 나오면 실행 자체를 막아버립니다!

# src/executor.py
UNSAFE_KEYWORDS = {
    "DROP", "DELETE", "UPDATE", "INSERT", "ALTER",
    "TRUNCATE", "GRANT", "REVOKE", "CREATE",
}

def is_safe(sql: str) -> bool:
    cleaned = _COMMENT_RE.sub(" ", sql)
    tokens = {m.group(1).upper() for m in _TOKEN_RE.finditer(cleaned)}
    return tokens.isdisjoint(UNSAFE_KEYWORDS)

실행 — `execute()`

is_safe()를 통과한 SQL만 execute()로 흘러갑니다. 반환 타입은 generator.py의 GeneratedSQL과 짝을 이루는 데이터클래스입니다.

# src/executor.py
@dataclass
class ExecutionResult:
    columns: list[str]       # 컬럼명
    rows: list[tuple]        # 결과 행
    row_count: int           # 행 수
    error: str | None        # 에러 메시지 (성공 시 None)
    elapsed_ms: float        # 소요 시간
    truncated: bool          # 200행 초과로 잘렸는지

실행 흐름은 단순한데 몇 가지 방어 코드를 추가했습니다.

# src/executor.py
def execute(sql: str, *, limit: int = 200) -> ExecutionResult:
    started = time.perf_counter()
    try:
        with get_conn() as conn, conn.cursor() as cur:
            cur.execute("SET LOCAL statement_timeout = 10000")  # 10초
            cur.execute(sql)
            fetched = cur.fetchmany(limit + 1)                  # ← 트릭
            truncated = len(fetched) > limit
            rows = fetched[:limit]
            columns = [d.name for d in cur.description] if cur.description else []
            return ExecutionResult(
                columns=columns,
                rows=rows,
                row_count=len(rows),
                error=None,
                elapsed_ms=(time.perf_counter() - started) * 1000,
                truncated=truncated,
            )
    except Exception as e:
        return ExecutionResult(
            columns=[], rows=[], row_count=0,
            error=str(e),
            elapsed_ms=(time.perf_counter() - started) * 1000,
            truncated=False,
        )

핵심 포인트 몇 개만 짚으면:

SET LOCAL statement_timeout = 10000 — 트랜잭션 단위로 10초 컷. SET LOCAL이라 다음 요청에 영향 없습니다.
fetchmany(limit + 1) — limit=200일 때 201행을 받아봐서, 201행이면 “더 있다는 뜻” 이니까 truncated=True로 마크하고 200행만 반환합니다.
예외를 raise하지 않음 — 에러를 그대로 던지면 TUI가 크래시하니까, error 필드에 메시지 담아서 정상 반환합니다.

TUI에서 실제로 잘리는 모습은 이렇게 나옵니다.

배송 완료된 주문 고객별 주문 금액 결과 — 결과가 200행으로 잘렸습니다 메시지 — 결과가 200행을 넘으면 Results 패널 하단에 '결과가 200행으로 잘렸습니다.' 알림이 뜹니다

pipeline.py 입장에서는 is_safe() → execute() 두 단계만 거치면 항상 PipelineResult가 돌아온다는 게 보장돼서 분기가 깔끔해집니다 👍

파이프라인 연결 — `pipeline.py`

pipeline.py는 지금까지 본 모듈들을 직렬로 연결하는 오케스트레이터입니다. 결과 타입부터 보면 어떤 정보를 모으는지 한눈에 보입니다.

# src/pipeline.py
@dataclass
class PipelineResult:
    question: str                       # 입력 질문
    sql: str                            # 최종 SQL
    generation: GeneratedSQL | None     # LLM 생성 결과
    execution: ExecutionResult | None   # DB 실행 결과
    blocked_reason: str | None          # 안전 검사 차단 사유
    error: str | None                   # generation 단계 에러
    total_ms: float                     # 전체 소요 시간

# src/pipeline.py
class NaivePipeline:
    def __init__(self, schema_path: str = "data/schema.sql") -> None:
        self._schema_ddl = load_schema_ddl(schema_path)  # 초기화 시 1회만 로드

    def run(self, question: str) -> PipelineResult:
        started = time.perf_counter()
        try:
            gen = generate_sql(question, self._schema_ddl)
        except Exception as e:
            return PipelineResult(question=question, sql="", generation=None,
                                  execution=None, blocked_reason=None,
                                  error=str(e), total_ms=_ms(started))

        if not is_safe(gen.sql):
            return PipelineResult(question=question, sql=gen.sql, generation=gen,
                                  execution=None,
                                  blocked_reason="Unsafe SQL detected",
                                  error=None, total_ms=_ms(started))

        execution = execute(gen.sql)
        return PipelineResult(question=question, sql=gen.sql, generation=gen,
                              execution=execution, blocked_reason=None,
                              error=None, total_ms=_ms(started))

3단계 중 어디서 실패했는지가 필드별로 갈라집니다.

단계	실패 시 채워지는 필드
LLM 생성	`error`
안전 검사	`blocked_reason`
DB 실행	`execution.error`

이렇게 분리해두면 TUI나 eval_baseline에서 “왜 실패했는지” 를 그대로 표시할 수 있습니다. 그리고 스키마 DDL은 초기화 시 한 번만 로드합니다. TUI에서 매 요청마다 파일을 읽으면 괜히 느려지거든요 👊

사실 진짜 의미는 인터페이스 계약

지금은 NaivePipeline 하나뿐이지만, Quest 3에서 만들 RAGPipeline도 똑같이 PipelineResult를 반환하게 설계할 겁니다. 그러면 TUI나 평가 스크립트는 파이프라인 클래스만 갈아끼우면 끝, 다른 코드 손댈 게 없습니다 !

즉 이 pipeline.py가 Quest 2 → Quest 3(RAG) → Quest 6(Agentic) 으로 가는 파이프라인 교체의 기준점 역할을 합니다. baseline은 단순히 “성능 비교용” 이 아니라 인터페이스 계약을 박아두는 작업 이기도 한 거죠:)

Textual TUI — CLI 대신 4패널 테스트 환경

Jupyter나 스크립트로 테스트하는 것도 방법이지만, TUI(Terminal UI)로 만들면 반복 테스트가 훨씬 쾌적하더라구요. Python TUI 라이브러리 중에서 Textual이 요즘 제일 활발하고 예쁘기도 한거 같아서 골랐습니다.

TUI는 4패널 구성입니다.

패널	내용
Input	자연어 질문 입력 + Run 버튼
SQL	생성된 SQL (syntax highlight)
Results	실행 결과 DataTable
History	세션 내 질문 히스토리

Textual TUI 초기 화면 — Question / Generated SQL / Results / Session History 4패널 — 앱을 띄우면 보이는 첫 화면 — 4패널 비어있고 Question 패널에 입력 대기

RAG 검색 결과 패널은 Quest 3에서 채울 예정이라 자리만 잡아뒀습니다. 질문을 하나 던지면 SQL → 결과 테이블 → Session History까지 한 번에 채워집니다

'브랜드별 평균 상품 가격' 질문 실행 결과 — JOIN + AVG SQL과 결과 테이블 — brands JOIN products + ROUND(AVG()) GROUP BY — Solar Pro가 알아서 쿼리를 짜줍니다

실행은 uv run python -m src.tui로 됩니다. Textual의 @work 데코레이터를 써서 SQL 생성·실행이 비동기로 돌아가기 때문에, LLM 응답 기다리는 동안 TUI가 블로킹되지 않습니다

평가 결과

5개 정도 샘플 질문으로 먼저 돌려봤습니다. 단일 테이블 / JOIN / 윈도우 함수가 골고루 섞이게 골랐습니다.

차트 로딩 중...

질문	행 수	SQL 특징
월별 매출 추이	25	`orders` + `order_items` JOIN, `DATE_TRUNC`
카테고리별 상품 분포	30	`categories` LEFT JOIN `products`, 활성 여부 `CASE`
평점별 리뷰 분포	5	`rating` GROUP BY + 백분율 윈도우 함수
택배사별 평균 배송일수	5	`EXTRACT(DAY FROM delivered_at - shipped_at)`
고객별 총 주문 금액 상위 10명	10	`customers` JOIN `order_items`, `SUM` + ORDER BY LIMIT

5개 전부 한 번에 통과. 평균 ~1,300ms / ~2,000 tokens 수준이고, 단일 테이블이든 다중 조인이든 큰 차이가 없었습니다.

메트릭	값
Parse Success Rate	100% (20/20)
Execution Accuracy	100% (20/20)
평균 레이턴시	1,129ms
평균 토큰 / 질문	2,036 tokens
총 토큰 사용량	40,734 tokens

솔직히 예상보다 훨씬 잘 나왔습니다! 18개 테이블 스키마를 전부 넣어도 Solar Pro가 꽤 잘 골라서 SQL을 만들어내더라구요

그리고 지금 질문당 평균 2,036 토큰을 쓰고 있는데, 이게 Quest 3에서 시맨틱 RAG를 붙이면 얼마나 줄어드는지가 이 시리즈의 핵심 포인트 중 하나입니다. 테이블 수가 많아질수록 이 차이는 더 벌어지겠죠?!

재미있었던 케이스 — “오늘의 날씨?”

본 평가 외에 장난 삼아 도메인 밖 질문도 던져봤습니다. 이커머스 DB에는 당연히 날씨 데이터가 없습니다!!

'오늘의 날씨 ?' 질문에 대한 LLM의 graceful 응답 — 메시지를 SELECT로 감싸서 반환 — 없는 데이터를 만들어내는 대신, 메시지 문자열을 SELECT로 감싸서 정중하게 거절합니다

SELECT '오늘의 날씨 정보를 제공할 수 없습니다. PostgreSQL 데이터베이스에는 날씨 데이터가 포함되어 있지 않습니다.' AS message;

이게 흥미로운 이유 몇 가지:

존재하지 않는 weather 테이블을 할루시네이션하지 않음 — 그냥 없는 테이블 만들어서 SELECT * FROM weather 같은 거 뱉을 수도 있었거든요
안전 검사 통과 — SELECT 외 키워드가 없으니 is_safe() 가 OK
실행 성공 — 1행짜리 message 컬럼이 실제로 화면에 나옵니다 🤔

물론 이게 항상 이렇게 동작한다는 보장은 없습니다. Quest 7 즈음에서 “답할 수 없는 질문” 을 명시적으로 라우팅하는 가드를 따로 두는 게 맞을 것 같아요

마무리

Quest 2는 말 그대로 “기준선을 기록하는 것” 이 목적이었습니다. 전체 스키마를 다 넣는 방식이 지금은 나쁘지 않은 이유는 테이블이 18개라서 그나마 버티는 거고, 50개·100개가 되면 이야기가 달라질 겁니다 ㅎㅎ

Quest 3에서는 Quest 1에서 만들어둔 시맨틱 RAG를 파이프라인에 연결해서, 관련 테이블만 골라 주입하는 방식으로 전환합니다. 토큰 절감 + 정확도 향상을 baseline 대비로 진행해 보겠습니다.

NL-to-SQL Agent (4부작)