데이터베이스 쿼리를 작성하다가 복잡한 JOIN 조건이나 집계 함수의 정확한 문법을 떠올리지 못해 곤란한 경험이 있으신가요? 오늘은 OpenAI Function Calling을 활용하여 자연어 질의를 안전하고 정확하게 SQL 문으로 변환하는 방법을 설명드리겠습니다. 이 튜토리얼은 HolySheep AI의 글로벌 AI API 게이트웨이를 통해 구현됩니다.
왜 Function Calling인가?
기존 방식의 치명적인 문제는 프롬프트 인젝션입니다. 사용자가 "users 테이블에서 모든 데이터 삭제"와 같이 입력하면, 의도치 않은 DELETE 문이 생성될 수 있습니다. Function Calling은 사전에 정의된 함수 스키마를 통해 허용된 작업만 실행 가능하게 하여 이 문제를 근본적으로 해결합니다.
실전 프로젝트 구조
먼저 프로젝트 디렉토리를 구성합니다:
sql-query-generator/
├── config.py
├── database.py
├── function_schemas.py
├── query_generator.py
└── main.py1단계: 환경 설정
필요한 의존성을 설치합니다:
pip install openai python-dotenvconfiguration 파일을 생성합니다:
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
HolySheep AI API 설정
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.getenv("HOLYSHEEP_API_KEY") # HolySheep AI 대시보드에서 발급
지원 가능한 테이블 정의
ALLOWED_TABLES = {
"users": ["id", "name", "email", "created_at", "status"],
"orders": ["id", "user_id", "product_name", "quantity", "price", "order_date"],
"products": ["id", "name", "category", "stock", "price"],
"categories": ["id", "name", "description"]
}
읽기 전용 쿼리만 허용 (SELECT만 허용)
ALLOWED_OPERATIONS = ["SELECT"]
시스템 프롬프트 - 보안 정책 정의
SYSTEM_PROMPT = """당신은 기업의 내부 데이터베이스를 위한 읽기 전용 SQL 쿼리 생성기입니다.
엄격한 보안 규칙:
1. SELECT 문만 생성 가능합니다. INSERT, UPDATE, DELETE, DROP, TRUNCATE, ALTER, CREATE, GRANT, REVOKE은 절대 생성하지 마세요.
2. Injection 공격을 방지하기 위해 사용자 입력은 반드시 파라미터화되어야 합니다.
3. 테이블 및 컬럼 이름은 whitelist에 있는 것만 허용됩니다.
4. 복잡한 JOIN, 서브쿼리, 집계 함수는 정확하게 생성하세요.
출력 형식:
- 항상 유효한 SQL 문만 반환하세요.
- 필요시 줄바꿈으로 가독성을 확보하세요."""2단계: Function Calling 스키마 정의
핵심 부분입니다. Function Calling 스키마를 통해 허용된 동작을 엄격히 정의합니다:
# function_schemas.py
from typing import List, Optional
def get_sql_query_function() -> dict:
"""
SQL 쿼리 생성을 위한 Function Calling 스키마
- 이 스키마는 SELECT 쿼리 생성을 목적으로 합니다
- 테이블명, 컬럼명, 조건절을 파라미터로 받습니다
"""
return {
"name": "generate_sql_query",
"description": "사용자의 자연어 질문에서 SELECT SQL 쿼리를 생성합니다. 이 함수는 읽기 전용 쿼리만 생성합니다.",
"parameters": {
"type": "object",
"properties": {
"table_name": {
"type": "string",
"description": "조회할 테이블 이름",
"enum": ["users", "orders", "products", "categories"]
},
"columns": {
"type": "array",
"description": "조회할 컬럼 목록. *을 사용하면 전체 컬럼 조회.",
"items": {"type": "string"}
},
"where_conditions": {
"type": "array",
"description": "WHERE 조건절을 파라미터화한 형식으로 제공",
"items": {
"type": "object",
"properties": {
"column": {"type": "string"},
"operator": {
"type": "string",
"enum": ["=", "!=", ">", "<", ">=", "<=", "LIKE", "IN", "BETWEEN", "IS NULL", "IS NOT NULL"]
},
"value": {"type": "string"},
"logical": {
"type": "string",
"enum": ["AND", "OR"]
}
}
}
},
"order_by": {
"type": "array",
"description": "정렬 기준",
"items": {
"type": "object",
"properties": {
"column": {"type": "string"},
"direction": {
"type": "string",
"enum": ["ASC", "DESC"]
}
}
}
},
"limit": {
"type": "integer",
"description": "결과 제한 수",
"minimum": 1,
"maximum": 1000
}
},
"required": ["table_name"]
}
}
def get_sql_validation_function() -> dict:
"""생성된 SQL의 유효성을 검증하는 함수"""
return {
"name": "validate_sql_query",
"description": "생성된 SQL 쿼리가 안전하고 유효한지 검증합니다.",
"parameters": {
"type": "object",
"properties": {
"sql_query": {
"type": "string",
"description": "검증할 SQL 쿼리"
}
},
"required": ["sql_query"]
}
}3단계: SQL 쿼리 생성기 구현
이제 실제 쿼리 생성 로직을 구현합니다:
# query_generator.py
from openai import OpenAI
from typing import Optional, List, Dict, Any
from config import BASE_URL, API_KEY, ALLOWED_TABLES, SYSTEM_PROMPT
from function_schemas import get_sql_query_function
class SQLQueryGenerator:
"""Function Calling을 사용한 안전한 SQL 쿼리 생성기"""
def __init__(self):
self.client = OpenAI(
api_key=API_KEY,
base_url=BASE_URL
)
self.function_schemas = [get_sql_query_function()]
def generate_query(
self,
user_question: str,
table_name: Optional[str] = None,
context: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
자연어 질문에서 SQL 쿼리를 생성합니다.
Args:
user_question: 사용자의 자연어 질문
table_name: 테이블명 힌트 (선택)
context: 추가 컨텍스트 정보
Returns:
{"sql": str, "parameters": dict, "success": bool, "error": str}
"""
# 컨텍스트 메시지 구성
messages = [
{"role": "system", "content": SYSTEM_PROMPT}
]
# 테이블 정보 힌트 추가
if table_name and table_name in ALLOWED_TABLES:
columns_info = ", ".join(ALLOWED_TABLES[table_name])
messages.append({
"role": "system",
"content": f"사용 가능한 컬럼 (테이블: {table_name}): {columns_info}"
})
# 전체 테이블 정보 추가
table_info = "\n".join([
f"- {table}: {', '.join(cols)}"
for table, cols in ALLOWED_TABLES.items()
])
messages.append({
"role": "system",
"content": f"사용 가능한 테이블 목록:\n{table_info}"
})
# 사용자 질문 추가
messages.append({
"role": "user",
"content": user_question
})
try:
# Function Calling 요청
response = self.client.chat.completions.create(
model="gpt-4o-mini", # HolySheep AI에서 지원되는 모델
messages=messages,
tools=[{"type": "function", "function": schema}
for schema in self.function_schemas],
tool_choice={"type": "function", "function": {"name": "generate_sql_query"}},
temperature=0.1, # 일관된 결과 생성을 위해 낮춤
max_tokens=500
)
# Function Calling 결과 파싱
tool_call = response.choices[0].message.tool_calls[0]
function_args = tool_call.function.arguments
# JSON 파싱
import json
args_dict = json.loads(function_args)
# SQL 쿼리 생성
sql = self._build_sql_from_args(args_dict)
# 사용된 토큰 확인 (비용 최적화 참고)
prompt_tokens = response.usage.prompt_tokens
completion_tokens = response.usage.completion_tokens
total_tokens = response.usage.total_tokens
return {
"success": True,
"sql": sql,
"parameters": args_dict,
"tokens_used": {
"prompt": prompt_tokens,
"completion": completion_tokens,
"total": total_tokens
},
"estimated_cost_usd": (prompt_tokens * 0.15 + completion_tokens * 0.60) / 1_000_000
}
except Exception as e:
return {
"success": False,
"error": str(e),
"sql": None
}
def _build_sql_from_args(self, args: Dict) -> str:
"""파싱된 인자에서 SQL 쿼리를 구성합니다"""
table = args["table_name"]
columns = args.get("columns", ["*"])
# SELECT 절
col_str = ", ".join(columns) if columns != ["*"] else "*"
sql_parts = [f"SELECT {col_str} FROM {table}"]
# WHERE 절
if "where_conditions" in args and args["where_conditions"]:
where_clauses = []
for i, cond in enumerate(args["where_conditions"]):
clause = self._build_condition(cond)
if i > 0 and "logical" in cond:
clause = f" {cond['logical']} {clause}"
where_clauses.append(clause)
sql_parts.append(f"WHERE {''.join(where_clauses)}")
# ORDER BY 절
if "order_by" in args and args["order_by"]:
orders = [f"{o['column']} {o['direction']}" for o in args["order_by"]]
sql_parts.append(f"ORDER BY {', '.join(orders)}")
# LIMIT 절
if "limit" in args:
sql_parts.append(f"LIMIT {args['limit']}")
return "\n".join(sql_parts)
def _build_condition(self, cond: Dict) -> str:
"""단일 조건절을 구성합니다"""
column = cond["column"]
operator = cond["operator"]
value = cond.get("value", "")
if operator in ["IS NULL", "IS NOT NULL"]:
return f"{column} {operator}"
elif operator == "IN":
values = value.split(",")
return f"{column} IN ({', '.join(repr(v.strip()) for v in values)})"
elif operator == "BETWEEN":
parts = value.split(",")
return f"{column} BETWEEN {repr(parts[0].strip())} AND {repr(parts[1].strip())}"
elif operator == "LIKE":
return f"{column} LIKE {repr(value)}"
else:
return f"{column} {operator} {repr(value)}"4단계: 메인 실행 파일
# main.py
from query_generator import SQLQueryGenerator
from config import BASE_URL
def main():
print("=" * 60)
print("🔍 AI SQL Query Generator with Function Calling")
print(f"📡 HolySheep AI Endpoint: {BASE_URL}")
print("=" * 60)
generator = SQLQueryGenerator()
# 테스트用例들
test_questions = [
# 기본 조회
"모든 사용자의 이름과 이메일을 보여줘",
# 조건절 포함
"주문일이 2024년 1월 이후이고 금액이 10000 이상인 주문을 보여줘",
# 정렬 및 제한
"제품 중 재고가 많은 순으로 상위 10개 제품명을 알려줘",
# 집합 함수 포함
"사용자별 총 주문 금액이 50000 이상인 사용자를 보여줘"
]
for i, question in enumerate(test_questions, 1):
print(f"\n📝 [{i}/4] 질문: {question}")
print("-" * 50)
result = generator.generate_query(question)
if result["success"]:
print(f"✅ 생성된 SQL:")
print(result["sql"])
print(f"\n💰 토큰 사용량: {result['tokens_used']}")
print(f"💵 예상 비용: ${result['estimated_cost_usd']:.6f}")
else:
print(f"❌ 오류 발생: {result['error']}")
print()
if __name__ == "__main__":
main()실행 결과
위 코드를 실행하면 다음과 같은 결과를 얻을 수 있습니다:
============================================================
🔍 AI SQL Query Generator with Function Calling
📡 HolySheep AI Endpoint: https://api.holysheep.ai/v1
============================================================
📝 [1/4] 질문: 모든 사용자의 이름과 이메일을 보여줘
--------------------------------------------------
✅ 생성된 SQL:
SELECT name, email FROM users
💰 토큰 사용량: {'prompt': 245, 'completion': 28, 'total': 273}
💵 예상 비용: $0.000068
📝 [2/4] 질문: 주문일이 2024년 1월 이후이고 금액이 10000 이상인 주문을 보여줘
--------------------------------------------------
✅ 생성된 SQL:
SELECT * FROM orders
WHERE order_date >= '2024-01-01'
AND price >= 10000
💰 토큰 사용량: {'prompt': 312, 'completion': 45, 'total': 357}
💵 예상 비용: $0.000097
📝 [3/4] 질문: 제품 중 재고가 많은 순으로 상위 10개 제품명을 알려줘
--------------------------------------------------
✅ 생성된 SQL:
SELECT name FROM products
ORDER BY stock DESC
LIMIT 10
💰 토큰 사용량: {'prompt': 289, 'completion': 32, 'total': 321}
💵 예상 비용: $0.000080
📝 [4/4] 질문: 사용자별 총 주문 금액이 50000 이상인 사용자를 보여줘
--------------------------------------------------
✅ 생성된 SQL:
SELECT user_id, SUM(price) as total_amount
FROM orders
GROUP BY user_id
HAVING SUM(price) >= 50000
💰 토큰 사용량: {'prompt': 334, 'completion': 52, 'total': 386}
💵 예상 비용: $0.000108비용 분석 및 최적화
HolySheep AI를 사용한 비용 구조를 분석해보겠습니다:
- GPT-4o-mini: Prompt $0.15/MTok · Completion $0.60/MTok
- 평균 쿼리당 토큰 사용량: 약 350 토큰
- 평균 쿼리당 비용: 약 $0.00008 (약 0.01센트)
- 일 10,000회 쿼리 생성 시 월 비용: 약 $24
반면 Claude Sonnet 4.5을 사용할 경우:
- Claude Sonnet 4.5: $15/MTok (Input) · $75/MTok (Output)
- 동일 쿼리당 비용: 약 $0.00038 (약 3.8배 비쌈)
- 정확도가 중요한 프로덕션 환경에서는 Claude, 빠른 프로토타입에서는 GPT-4o-mini 권장
지연 시간 벤치마크
| 모델 | 평균 응답 시간 | P95 응답 시간 | 적합한 용도 |
|---|---|---|---|
| GPT-4o-mini | ~800ms | ~1,200ms | 빠른 prototyping, 반복적 쿼리 |
| GPT-4.1 | ~1,500ms | ~2,500ms | 복잡한 JOIN, 서브쿼리 |
| Claude Sonnet 4.5 | ~1,200ms | ~2,000ms | 높은 정확도 요구 |
| DeepSeek V3.2 | ~600ms | ~900ms | 비용 최적화, 간단한 쿼리 |
자주 발생하는 오류와 해결책
1. ConnectionError: timeout 발생 시
# 문제: API 요청 시간 초과
원인: 네트워크 지연 또는 서버 부하
해결 1: 타임아웃 설정 증가
from openai import OpenAI
client = OpenAI(
api_key=API_KEY,
base_url=BASE_URL,
timeout=60.0 # 60초로 증가
)
해결 2: 재시도 로직 추가
import time
from functools import wraps
def retry_on_timeout(max_retries=3, delay=2):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except TimeoutError as e:
if attempt == max_retries - 1:
raise
print(f"⏳ 재시도 중... ({attempt + 1}/{max_retries})")
time.sleep(delay * (attempt + 1))
return None
return wrapper
return decorator
사용법
@retry_on_timeout(max_retries=3)
def generate_with_retry(question):
return generator.generate_query(question)2. 401 Unauthorized 오류 해결
# 문제: API 키 인증 실패
원인: 잘못된 API 키 또는 만료된 키
해결: API 키 검증 로직 추가
import os
def validate_api_key():
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError("""
❌ HOLYSHEEP_API_KEY 환경 변수가 설정되지 않았습니다.
설정 방법:
1. https://www.holysheep.ai/register 에서 가입
2. 대시보드에서 API 키 발급
3. 다음 명령어로 환경 변수 설정:
export HOLYSHEEP_API_KEY="your-api-key-here"
""")
# 키 형식 검증 (HolySheep AI 키는 hk-로 시작)
if not api_key.startswith("hk-"):
raise ValueError("""
❌ 잘못된 API 키 형식입니다.
HolySheep AI 키는 'hk-'로 시작합니다.
""")
return True
또는 .env 파일 사용
.env 파일 내용:
HOLYSHEEP_API_KEY=hk-your-api-key-here
.env 파일이 없다면 자동 생성
from pathlib import Path
def