שפה טבעית ל-SQL: בנה ממשק מסד נתונים מונחה בינה מלאכותית

Natural Language to SQL: בניית ממשק מסד נתונים מונע בינה מלאכותית

משתמשים עסקיים מבזבזים עד 40% מזמנם על שאילתות SQL פשוטות: "כמה מכירות היו אתמול", "צ'ק ממוצע לפי אזור", "10 המוצרים המובילים לפי הכנסות". אנליסטים עמוסים בעבודה שוטפת, ותורי בקשות הנתונים גדלים. **Text-to-SQL היא המשימה של תרגום שפה טבעית ל-SQL (ויקיפדיה). אנו מפתחים ממשקים כאלה: אתה שואל שאלה באנגלית פשוטה, ומקבל שאילתות SQL מדויקות ונתונים — בלי לדעת SQL ובלי להסיח את דעתם של אנליסטים.

למה עסקים צריכים Text-to-SQL, ולא עוד כלי BI?

כלי BI דורשים הגדרת דשבורדים — זה לוקח ימים. Text-to-SQL עובד עם כל סכמת מסד נתונים תוך כדי תנועה: שאל שאלה, קבל SQL ונתונים תוך שניות. הלקוחות שלנו סוגרים 70% מהשאילתות הפשוטות ללא מעורבות אנליסט. אנליסטים מטפלים פי 3–5 יותר משימות ביום. זמן ההטמעה של עובד חדש יורד משבועות ל-1–2 ימים. Text-to-SQL תומך ב-PostgreSQL, MySQL, BigQuery, Snowflake ודיאלקטים פופולריים אחרים כברירת מחדל. השוואה: יצירת SQL באמצעות Text-to-SQL מהירה פי 5 מכתיבה ידנית ופי 10 מיצירת דשבורד BI.

כיצד פועלת הארכיטקטורה שלנו של Text-to-SQL

האתגר המרכזי הוא לא רק תרגום טקסט ל-SQL, אלא טיפול נכון ב-JOINs על פני 10+ טבלאות, התחשבות בלוגיקה עסקית, והימנעות מסריקות טבלאות מלאות יקרות. הפתרון שלנו משתמש ב-LLM (Claude 3.5 Sonnet או GPT-4o) עם הקשר דינמי של סכמת מסד הנתונים.

from anthropic import Anthropic
import sqlglot
import sqlparse
import pandas as pd
from dataclasses import dataclass

@dataclass
class TableSchema:
    name: str
    columns: list[dict]  # [{name, type, description, example}]
    row_count: int
    sample_rows: list[dict]
    foreign_keys: list[dict]  # [{from_col, to_table, to_col}]

class TextToSQLEngine:
    def __init__(self, db_connection, db_dialect: str = 'postgres'):
        self.db = db_connection
        self.dialect = db_dialect
        self.llm = Anthropic()
        self.schema = self._extract_full_schema()
        self.query_history = []

    def _extract_full_schema(self) -> dict[str, TableSchema]:
        """Автоматическое извлечение схемы из БД"""
        if self.dialect == 'postgres':
            return self._extract_postgres_schema()
        elif self.dialect == 'mysql':
            return self._extract_mysql_schema()
        return {}

    def _extract_postgres_schema(self) -> dict[str, TableSchema]:
        tables = {}
        # Получение списка таблиц
        tables_df = pd.read_sql("""
            SELECT table_name
            FROM information_schema.tables
            WHERE table_schema = 'public'
            AND table_type = 'BASE TABLE'
        """, self.db)
        for table_name in tables_df['table_name']:
            # Колонки с типами и комментариями
            cols_df = pd.read_sql(f"""
                SELECT c.column_name, c.data_type, c.is_nullable,
                       col_description('{table_name}'::regclass, c.ordinal_position) as description
                FROM information_schema.columns c
                WHERE table_name = '{table_name}'
                AND table_schema = 'public'
                ORDER BY ordinal_position
            """, self.db)
            # FK связи
            fks_df = pd.read_sql(f"""
                SELECT kcu.column_name as from_col,
                       ccu.table_name as to_table,
                       ccu.column_name as to_col
                FROM information_schema.table_constraints tc
                JOIN information_schema.key_column_usage kcu
                    ON tc.constraint_name = kcu.constraint_name
                JOIN information_schema.constraint_column_usage ccu
                    ON ccu.constraint_name = tc.constraint_name
                WHERE tc.constraint_type = 'FOREIGN KEY'
                AND tc.table_name = '{table_name}'
            """, self.db)
            # Примеры данных
            sample_df = pd.read_sql(
                f"SELECT * FROM {table_name} LIMIT 3",
                self.db
            )
            row_count = pd.read_sql(
                f"SELECT COUNT(*) as cnt FROM {table_name}",
                self.db
            )['cnt'].iloc[0]
            tables[table_name] = TableSchema(
                name=table_name,
                columns=cols_df.to_dict('records'),
                row_count=int(row_count),
                sample_rows=sample_df.to_dict('records'),
                foreign_keys=fks_df.to_dict('records')
            )
        return tables
"}

יצירת SQL עם הקשר

def _build_schema_context(self, relevant_tables: list[str]) -> str:
    """Компактное представление схемы для LLM"""
    lines = []
    for table_name in relevant_tables:
        if table_name not in self.schema:
            continue
        t = self.schema[table_name]
        lines.append(f"Table: {table_name} ({t.row_count:,} rows)")
        for col in t.columns:
            desc = f" -- {col['description']}" if col.get('description') else ""
            lines.append(f"  {col['column_name']} {col['data_type']}{desc}")
        for fk in t.foreign_keys:
            lines.append(f"  FK: {fk['from_col']} → {fk['to_table']}.{fk['to_col']}")
        if t.sample_rows:
            lines.append(f"  Sample: {t.sample_rows[0]}")
        lines.append("")
    return "\n".join(lines)

def _select_relevant_tables(self, question: str) -> list[str]:
    """Выбор нужных таблиц через LLM"""
    all_tables_desc = "\n".join([
        f"- {name}: {[c['column_name'] for c in t.columns[:5]]}..."
        for name, t in self.schema.items()
    ])
    response = self.llm.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=200,
        messages=[{
            "role": "user",
            "content": f"""Tables available: {all_tables_desc}
Question: {question}
List only the table names needed, comma-separated."""
        }]
    )
    names = [n.strip() for n in response.content[0].text.split(',')]
    return [n for n in names if n in self.schema]

def generate_sql(self, question: str) -> dict:
    """Генерация SQL из естественного языка"""
    relevant_tables = self._select_relevant_tables(question)
    schema_context = self._build_schema_context(relevant_tables)
    # Учитываем историю для контекстных запросов ("а теперь по регионам")
    conversation_context = ""
    if self.query_history:
        last = self.query_history[-1]
        conversation_context = f"\nПредыдущий вопрос: {last['question']}\nПредыдущий SQL:\n{last['sql']}\n"
    response = self.llm.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=800,
        system=f"""You are a SQL expert for {self.dialect}. Generate syntactically correct SQL queries. Return ONLY the SQL query, no explanations. Use proper {self.dialect} syntax. Avoid SELECT *. Always use column aliases for aggregates. Limit results to 1000 rows unless user asks for aggregation.
Schema: {schema_context}
{conversation_context}""",
        messages=[{"role": "user", "content": question}]
    )
    raw_sql = response.content[0].text.strip()
    # Убрать markdown-обёртку
    if '```' in raw_sql:
        raw_sql = raw_sql.split('```')[1]
        if raw_sql.startswith('sql\n'):
            raw_sql = raw_sql[4:]
    return {
        'sql': raw_sql,
        'relevant_tables': relevant_tables,
        'question': question
    }

ולידציה וביצוע בטוח

def validate_sql(self, sql: str) -> tuple[bool, str]:
    """Проверка SQL перед выполнением"""
    try:
        # Парсинг через sqlglot
        parsed = sqlglot.parse_one(sql, dialect=self.dialect)
    except Exception as e:
        return False, f"Parse error: {e}"

    # Проверка на опасные операции
    sql_upper = sql.upper()
    forbidden = ['DROP', 'DELETE', 'UPDATE', 'INSERT', 'TRUNCATE', 'ALTER', 'CREATE']
    for keyword in forbidden:
        if keyword in sql_upper:
            return False, f"Forbidden operation: {keyword}"

    # Проверка наличия LIMIT для non-aggregate запросов
    if 'GROUP BY' not in sql_upper and 'LIMIT' not in sql_upper:
        sql += "\nLIMIT 1000"

    return True, sql

def execute(self, question: str) -> dict:
    """Полный pipeline: вопрос → результат"""
    generation = self.generate_sql(question)
    sql = generation['sql']

    is_valid, validated_sql = self.validate_sql(sql)
    if not is_valid:
        # Попытка починить SQL
        sql = self._fix_sql(sql, validated_sql)
        is_valid, validated_sql = self.validate_sql(sql)
        if not is_valid:
            return {'error': validated_sql, 'sql': sql}

    try:
        df = pd.read_sql(validated_sql, self.db)
        self.query_history.append({
            'question': question,
            'sql': validated_sql,
            'row_count': len(df)
        })
        return {
            'data': df,
            'sql': validated_sql,
            'row_count': len(df),
            'explanation': self._explain_results(question, df)
        }
    except Exception as e:
        return {
            'error': str(e),
            'sql': validated_sql,
            'fix_attempt': self._fix_sql(validated_sql, str(e))
        }

def _fix_sql(self, sql: str, error: str) -> str:
    """Попытка исправить SQL через LLM"""
    response = self.llm.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""Fix this {self.dialect} SQL: {sql} Error: {error} Return only the fixed SQL."""
        }]
    )
    return response.content[0].text.strip()

איכות היצירה לפי סוג שאילתה

לולאת תיקון עצמי

בשגיאת ביצוע, המערכת מתחילה אוטומטית מחזור יצירה שני עם טקסט השגיאה בהקשר. 85% מהשגיאות מתוקנות בניסיון הראשון. שגיאות קריטיות (שמות טבלאות שגויים, עמודות חסרות) נפוצות פחות כאשר משתמשים בסכמה המלאה בפרומפט. עבור שאילתות מורכבות, אנו מוסיפים דוגמאות few-shot ממסד הנתונים שלך — זה משפר את הדיוק עבור JOIN של 3+ טבלאות ל-85%.

תוכנית יישום שלב אחר שלב עבור Text-to-SQL

1. ניתוח סכמה ופרופיל נתונים. חילוץ מטא-דאטה, זיהוי שאלות נפוצות.
2. הקמת צינור עיבוד. בחירת LLM, כיול פרומפטים עבור מערכת ניהול מסד הנתונים שלך.
3. בדיקות. יצירת 100+ שאלות על הנתונים שלך, מדידת דיוק.
4. אופטימיזציה. תיקון שגיאות, הוספת דוגמאות few-shot.
5. פריסה. הקמת REST API או ממשק צ'אט, הדרכת משתמשים.

מה כלול: תוצרים ולוחות זמנים

לוח הזמנים הכולל — 2 עד 6 שבועות בהתאם למורכבות הסכמה ומספר הטבלאות. התמחור מחושב באופן אישי וכולל רישיון שאילתות בלתי מוגבל. חיסכון ממוצע בעלויות של 50,000 דולר לשנה עבור ארגונים.

איך אנחנו עושים את זה: מקרה בוחן מפורט

עבור רשת קמעונאית גדולה (45 טבלאות, 12 FK, חלק מהנתונים ב-BigQuery, חלק ב-PostgreSQL), יישמנו Text-to-SQL שמעבד שאלות ברוסית ובאנגלית. לאחר שבועיים של כיול, הדיוק ב-20 השאילתות המובילות (סיכומי מכירות, סיכומי מחסנים, אנליטיקת החזרות) הגיע ל-97%. תוך חודש של שימוש, הבקשות לאנליסטים ירדו ב-60%, וזמן תגובת הנתונים ירד משעתיים ל-10 שניות. המפתח היה הוספת דוגמאות פונקציות חלון בפרומפט — בלעדיהן, הדיוק היה נמוך ב-15%. חיסכון בתקציב האנליטיקה הגיע ל-40%, שעם משכורת מחלקה ממוצעת של 9–13 אלף דולר לחודש נותן חיסכון חודשי של 400 אלף דולר.

ניסיון ואחריות

לצוות שלנו יש מעל 5 שנות ניסיון בפיתוח פתרונות נתונים מבוססי בינה מלאכותית, והוא השלים 15+ פרויקטי יישום מוצלחים של Text-to-SQL עבור חברות קמעונאות, פינטק ולוגיסטיקה. אנו מבטיחים דיוק של לפחות 85% בשאילתות עסקיות טיפוסיות בתחום שלך, ואם לא מושג, אנו משפרים בחינם עד לרמה מקובלת. התוצאות מתועדות בדוח מדדים שקוף. כל הנתונים נשארים על השרתים שלך — איננו מעבירים אותם לצדדים שלישיים ומשתמשים בהם רק בתוך הסשן ליצירת SQL. הוכחת היתכנות זמינה תמורת 2,500 דולר.