AI 赋能传统报表系统：自然语言查询到 SQL 生成的工程化管道-尧图网络科技

AI 赋能传统报表系统：自然语言查询到 SQL 生成的工程化管道

一、报表系统的用户之痛：当业务方想知道上季度东北区的退货率，但 SQL 对他们来说是天书

报表系统最核心的用户是业务方——运营、销售、财务。但他们不懂 SQL。于是工作流变成了：业务方提需求 → 数据分析师写 SQL → 跑数据 → 生成报表。这个流程的问题在于：分析师的带宽是有限的，业务方的需求是即时的。一个"东北区 Q2 退货率"的问题可能要等半天才能得到答案。

Text-to-SQL 技术（自然语言转 SQL）让业务方可以用自然语言直接查询数据库。LLM 在这个场景下天然适合——它理解自然语言的语义，能将"查一下上季度退货率最高的 5 个商品"翻译为SELECT product_name, return_rate FROM products WHERE quarter='Q2' ORDER BY return_rate DESC LIMIT 5。

但 Text-to-SQL 不是一个简单的"让 LLM 直接写 SQL"就能解决的问题。数据库 Schema 需要上下文、SQL 必须语法正确且不会拖垮数据库、查询结果需要安全的权限控制。

graph TB A[用户自然语言查询] --> B[预处理<br/>意图识别 + 关键词提取] B --> C{查询类型} C -->|预定义模板| D1[模板匹配<br/>直接生成 SQL] C -->|需要生成| D2[LLM Text-to-SQL] D2 --> E[Schema 上下文注入<br/>表结构 + 字段说明 + 示例] E --> F[LLM 生成 SQL] F --> G{SQL 校验} G -->|语法错误| H[自动修复 + 重试] G -->|通过| I{安全检查} I -->|包含 DROP/DELETE| J[拒绝执行] I -->|只读查询| K[执行 SQL + 限时] K --> L[结果格式化<br/>Chart/Table] L --> M[返回给用户] style J fill:#ff6b6b,color:#fff style M fill:#51cf66,color:#fff

二、Text-to-SQL 的工程挑战：不止于 Prompt Engineering

让 LLM 写出正确的 SQL，需要解决三个关键问题：

Schema 上下文：LLM 需要知道数据库中有哪些表、每个表有哪些字段、字段的类型和含义。但大多数数据库有几十张表——全部塞进 Prompt 会导致 Token 爆炸。解决方案是按用户查询的意图筛选相关表。例如用户问"退货率"，只把orders、returns、products三张表的结构发给 LLM。

SQL 安全检查：LLM 生成的 SQL 可能包含危险操作（DROP、DELETE、UPDATE）。管道中必须对生成的 SQL 做白名单检查——只允许 SELECT 语句，禁止任何修改操作。同时使用只读数据库账号做隔离。

约束执行：LLM 可能生成一个 JOIN 5 张表、扫描百万行数据的查询。需要：设置查询超时（5 秒），超时自动终止；限制返回行数（默认 LIMIT 100，除非用户明确要求更多）。

三、完整的 Text-to-SQL 管道实现

#!/usr/bin/env python3 # text_to_sql.py import json import re import sqlite3 from typing import List, Dict, Optional from openai import OpenAI client = OpenAI(api_key=os.environ['OPENAI_API_KEY']) # 数据库 Schema（简化版） DB_SCHEMA = """ Tables: - orders(order_id INT PRIMARY KEY, product_id INT, customer_id INT, amount DECIMAL, status VARCHAR, region VARCHAR, created_at DATE) - products(product_id INT PRIMARY KEY, name VARCHAR, category VARCHAR, cost DECIMAL, price DECIMAL) - returns(return_id INT PRIMARY KEY, order_id INT, product_id INT, reason VARCHAR, amount DECIMAL, created_at DATE) - customers(customer_id INT PRIMARY KEY, name VARCHAR, region VARCHAR, level VARCHAR) Relations: - orders.product_id -> products.product_id - returns.order_id -> orders.order_id - orders.customer_id -> customers.customer_id Example Queries: 1. "退货率最高的 5 个商品" SELECT p.name, COUNT(r.return_id) as return_count FROM returns r JOIN products p ON r.product_id = p.product_id GROUP BY p.product_id ORDER BY return_count DESC LIMIT 5 2. "东北区 Q2 退货金额" SELECT SUM(r.amount) FROM returns r JOIN orders o ON r.order_id = o.order_id WHERE o.region = '东北' AND o.created_at BETWEEN '2024-04-01' AND '2024-06-30' """ SYSTEM_PROMPT = f"""你是一个 SQL 查询生成助手。基于以下数据库 Schema 生成 SQL 查询。 规则： 1. 只生成 SELECT 语句，禁止 INSERT/UPDATE/DELETE/DROP 2. 默认 LIMIT 100，除非用户明确指定 3. 涉及日期的查询，注意处理时间范围 4. 使用标准的 SQL 语法（兼容 SQLite） 5. 返回 JSON 格式：{{"sql": "...", "explanation": "中文解释"}} {DB_SCHEMA}""" def generate_sql(question: str) -> Optional[str]: """LLM 生成 SQL""" try: response = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": question}, ], temperature=0, response_format={"type": "json_object"}, ) result = json.loads(response.choices[0].message.content) sql = result.get("sql", "") if not sql or not sql.strip().upper().startswith("SELECT"): return None return sql except Exception as e: print(f"SQL generation error: {e}") return None def validate_sql(sql: str) -> bool: """SQL 安全检查""" sql_upper = sql.upper() # 禁止的危险操作 dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE", "TRUNCATE"] for keyword in dangerous: if keyword in sql_upper: return False # 必须包含 SELECT if not sql_upper.strip().startswith("SELECT"): return False return True def add_limit(sql: str, max_limit: int = 100) -> str: """确保查询有 LIMIT""" sql_upper = sql.upper().strip() # 已有 LIMIT 就不加 if "LIMIT" in sql_upper: # 检查 LIMIT 值 match = re.search(r'LIMIT\s+(\d+)', sql_upper) if match: limit_val = int(match.group(1)) if limit_val > max_limit: sql = re.sub(r'LIMIT\s+\d+', f'LIMIT {max_limit}', sql, flags=re.IGNORECASE) return sql # 去除末尾分号 sql = sql.rstrip(';') return f"{sql} LIMIT {max_limit}" def execute_query(sql: str, db_path: str, timeout: int = 5) -> List[Dict]: """执行 SQL 查询""" conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row try: conn.execute(f"PRAGMA query_only = ON") cursor = conn.execute(sql) conn.commit() # 设置超时 import signal def handler(signum, frame): raise TimeoutError("Query timeout") signal.signal(signal.SIGALRM, handler) signal.alarm(timeout) rows = [dict(row) for row in cursor.fetchall()] signal.alarm(0) return rows except TimeoutError: return [{"error": "查询超时，请缩小查询范围"}] except Exception as e: return [{"error": str(e)}] finally: conn.close() def ask(question: str, db_path: str) -> Dict: """完整的查询管道""" # Step 1: 生成 SQL sql = generate_sql(question) if not sql: return {"error": "无法理解该查询，请换一种方式描述"} # Step 2: 安全检查 if not validate_sql(sql): return {"error": "生成的查询不安全，已拒绝执行"} # Step 3: 添加 LIMIT sql = add_limit(sql, max_limit=100) # Step 4: 执行 results = execute_query(sql, db_path, timeout=5) return { "question": question, "sql": sql, "results": results, "count": len(results), } # CLI 使用 if __name__ == "__main__": result = ask("东北区退货率最高的 5 个商品", "sales.db") print(json.dumps(result, ensure_ascii=False, indent=2))