❝
"在数据驱动的时代，让AI理解你的数据库就像教会外星人说人话一样困难。但Vanna做到了，而且做得相当优雅。"
引言：当ChatGPT遇上企业数据库的"水土不服"
想象一下这样的场景：你兴冲冲地打开ChatGPT，输入"帮我查询一下德国有多少客户"，期待着AI能够生成一条完美的SQL语句。结果呢？AI给你返回了一个看起来很专业的查询：
SELECT COUNT(*) FROM customers WHERE country = 'Germany';
看起来不错对吧？但当你兴奋地复制到数据库中执行时，系统无情地抛出了错误：Table 'customers' doesn't exist。
这就是当前AI生成SQL面临的核心困境：LLM虽然掌握了SQL语法的精髓，却对你的具体数据库结构一无所知。就像一个语言天才试图在不了解当地文化的情况下进行深度交流一样，注定会闹出笑话。
但是，如果我告诉你有一个开源项目能够将SQL生成的准确率从令人绝望的3%提升到令人惊艳的80%，你会相信吗？这就是我们今天要深入探讨的主角——Vanna。
第一章：RAG遇上SQL——一场技术革命的开始
1.1 什么是Vanna？不只是另一个SQL生成器
Vanna不是简单的"ChatGPT + SQL"的组合，而是一个基于RAG（Retrieval-Augmented Generation）架构的智能SQL生成框架。它的核心理念可以用一句话概括：让AI不仅懂SQL语法，更要懂你的数据。
从技术架构上看，Vanna采用了经典的RAG模式：
graph TD
A[用户问题] --> B[向量化检索]
B --> C[相关上下文]
C --> D[LLM生成SQL]
D --> E[执行验证]
E --> F[结果反馈]
F --> G[自动训练]
G --> B
这个架构的精妙之处在于，它不是简单地把问题扔给LLM，而是先从知识库中检索出最相关的上下文信息，然后再让LLM基于这些信息生成SQL。这就像给一个外国朋友不仅提供了字典，还提供了当地的文化背景和使用习惯。
1.2 核心技术栈：模块化设计的艺术
让我们深入Vanna的技术内核。通过分析其源码结构，我们可以发现Vanna采用了高度模块化的设计：
# Vanna的核心抽象基类
class VannaBase(ABC):
def __init__(self, config=None):
self.config = config
self.run_sql_is_set = False
self.static_documentation = ""
self.dialect = self.config.get("dialect", "SQL")
self.language = self.config.get("language", None)
self.max_tokens = self.config.get("max_tokens", 14000)
@abstractmethod
def generate_embedding(self, data: str, **kwargs) -> List[float]:
"""生成文本嵌入向量"""
pass
@abstractmethod
def get_similar_question_sql(self, question: str, **kwargs) -> list:
"""检索相似的问题-SQL对"""
pass
@abstractmethod
def submit_prompt(self, prompt, **kwargs) -> str:
"""提交提示词到LLM"""
pass
这种设计的巧妙之处在于，它将复杂的SQL生成过程分解为三个可插拔的组件：
-
嵌入生成器：负责将文本转换为向量表示 -
向量存储器：负责存储和检索相关上下文 -
语言模型：负责基于上下文生成SQL
1.3 支持的生态系统：一个真正的"瑞士军刀"
Vanna的另一个令人印象深刻的特点是其广泛的生态系统支持。从项目结构可以看出，它支持：
LLM提供商（9+）：
-
OpenAI (GPT-3.5/4) -
Anthropic (Claude) -
Google (Gemini/Bison) -
本地模型 (Ollama, HuggingFace) -
国产大模型 (智谱AI, 千问, 千帆, DeepSeek)
向量数据库（10+）：
-
ChromaDB, Pinecone, Qdrant -
Milvus, Weaviate, FAISS -
PostgreSQL (pgvector), Oracle
关系数据库（10+）：
-
PostgreSQL, MySQL, Oracle -
Snowflake, BigQuery, ClickHouse -
SQLite, DuckDB, SQL Server
这种"大一统"的设计哲学让Vanna能够适应几乎任何技术栈，这在企业级应用中尤为重要。
第二章：深度解析——RAG如何让SQL生成脱胎换骨
2.1 传统方法的困境：为什么ChatGPT在SQL生成上"水土不服"？
在深入Vanna的解决方案之前，我们先来理解传统方法的局限性。Vanna团队进行了一项令人印象深刻的实验，使用Cybersyn SEC数据集测试了不同方法的SQL生成准确率：
实验设置：
-
数据集：Cybersyn SEC filings（复杂的金融数据） -
测试问题：20个业务问题 -
测试模型：GPT-3.5, GPT-4, Google Bison -
评估方法：手动验证SQL执行结果
结果令人震惊：
|
|
|
|
|
|
|---|---|---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
这个结果揭示了一个重要的洞察：上下文比模型更重要。即使是最强大的GPT-4，在没有合适上下文的情况下，准确率也只有可怜的10%。
2.2 Vanna的三层上下文策略
Vanna的成功秘诀在于其精心设计的三层上下文策略：
第一层：Schema上下文（DDL信息）
def add_ddl(self, ddl: str, **kwargs) -> str:
"""添加数据定义语言到训练数据"""
id = deterministic_uuid(ddl) + "-ddl"
self.ddl_collection.add(
documents=ddl,
embeddings=self.generate_embedding(ddl),
ids=id,
)
return id
这一层提供了数据库的结构信息，包括表名、字段名、数据类型等。但仅有这些还不够，因为它无法告诉AI如何正确地使用这些表。
第二层：文档上下文（业务逻辑）
def add_documentation(self, documentation: str, **kwargs) -> str:
"""添加业务文档到训练数据"""
id = deterministic_uuid(documentation) + "-doc"
self.documentation_collection.add(
documents=documentation,
embeddings=self.generate_embedding(documentation),
ids=id,
)
return id
这一层包含了业务规则、字段含义、计算逻辑等信息。比如"revenue"字段的具体定义，或者某个表中数据的业务含义。
第三层：SQL示例上下文（最佳实践）
def add_question_sql(self, question: str, sql: str, **kwargs) -> str:
"""添加问题-SQL对到训练数据"""
question_sql_json = json.dumps({
"question": question,
"sql": sql,
}, ensure_ascii=False)
id = deterministic_uuid(question_sql_json) + "-sql"
self.sql_collection.add(
documents=question_sql_json,
embeddings=self.generate_embedding(question_sql_json),
ids=id,
)
return id
这是最关键的一层，它提供了具体的问题-SQL对应关系，让AI能够学习到如何将自然语言问题转换为正确的SQL查询。
2.3 智能检索：向量相似度的魔法
Vanna的核心创新在于其智能检索机制。当用户提出问题时，系统会：
-
向量化用户问题 -
在三个向量空间中并行搜索 -
基于相似度排序返回最相关的上下文
def generate_sql(self, question: str, allow_llm_to_see_data=False, **kwargs) -> str:
# 检索相似的问题-SQL对
question_sql_list = self.get_similar_question_sql(question, **kwargs)
# 检索相关的DDL信息
ddl_list = self.get_related_ddl(question, **kwargs)
# 检索相关的文档
doc_list = self.get_related_documentation(question, **kwargs)
# 构建提示词
prompt = self.get_sql_prompt(
initial_prompt=initial_prompt,
question=question,
question_sql_list=question_sql_list,
ddl_list=ddl_list,
doc_list=doc_list,
**kwargs,
)
# 提交给LLM生成SQL
llm_response = self.submit_prompt(prompt, **kwargs)
return self.extract_sql(llm_response)
这种方法的精妙之处在于，它不是简单地把所有信息都塞给LLM，而是智能地选择最相关的信息。这样既保证了上下文的质量，又避免了超出LLM的上下文窗口限制。
第三章：实战解析——从代码到生产的完整链路
3.1 快速上手：五分钟搭建你的AI SQL助手
让我们通过一个具体的例子来看看Vanna是如何工作的：
from vanna.openai.openai_chat import OpenAI_Chat
from vanna.chromadb.chromadb_vector import ChromaDB_VectorStore
# 创建自定义的Vanna类
class MyVanna(ChromaDB_VectorStore, OpenAI_Chat):
def __init__(self, config=None):
ChromaDB_VectorStore.__init__(self, config=config)
OpenAI_Chat.__init__(self, config=config)
# 初始化
vn = MyVanna(config={
'api_key': 'your-openai-key',
'model': 'gpt-4'
})
# 连接数据库
vn.connect_to_postgres(
host="localhost",
dbname="ecommerce",
user="admin",
password="password"
)
3.2 训练过程：让AI理解你的数据
训练Vanna就像教一个新员工熟悉公司的数据库：
# 1. 添加表结构信息
vn.train(ddl="""
CREATE TABLE customers (
id SERIAL PRIMARY KEY,
name VARCHAR(100),
email VARCHAR(100),
country VARCHAR(50),
created_at TIMESTAMP
);
""")
# 2. 添加业务文档
vn.train(documentation="""
customers表存储了所有注册用户的基本信息。
country字段使用ISO 3166-1标准的国家代码。
created_at表示用户注册时间。
""")
# 3. 添加示例查询
vn.train(
question="德国有多少客户？",
sql="SELECT COUNT(*) FROM customers WHERE country = 'DE';"
)
vn.train(
question="最近一个月新注册的用户数量？",
sql="""
SELECT COUNT(*)
FROM customers
WHERE created_at >= CURRENT_DATE - INTERVAL '1 month';
"""
)
3.3 智能问答：体验AI的魔法时刻
训练完成后，你就可以开始享受AI助手的服务了：
# 提问
sql, df, fig = vn.ask("显示每个国家的客户数量，按数量降序排列")
# Vanna会自动：
# 1. 理解问题意图
# 2. 检索相关上下文
# 3. 生成SQL查询
# 4. 执行查询
# 5. 返回结果和可视化图表
生成的SQL可能是这样的：
SELECT
country,
COUNT(*) as customer_count
FROM customers
GROUP BY country
ORDER BY customer_count DESC;
3.4 高级特性：让AI更聪明的秘密武器
自动学习机制
def ask(self, question: str, auto_train: bool = True, **kwargs):
# ... 生成和执行SQL ...
# 如果查询成功且auto_train=True，自动添加到训练数据
if len(df) > 0 and auto_train:
self.add_question_sql(question=question, sql=sql)
这个特性让Vanna能够从每次成功的查询中学习，不断改进自己的性能。
中间SQL支持
# 当需要探索数据时，Vanna可以生成中间查询
if 'intermediate_sql' in llm_response:
intermediate_sql = self.extract_sql(llm_response)
df = self.run_sql(intermediate_sql)
# 基于中间结果生成最终SQL
prompt = self.get_sql_prompt(
# ... 包含中间结果的上下文 ...
doc_list=doc_list + [f"中间查询结果: n{df.to_markdown()}"],
)
这个特性让AI能够像人类分析师一样，先探索数据再生成最终查询。
多语言支持
def _response_language(self) -> str:
if self.language is None:
return ""
return f"Respond in the {self.language} language."
Vanna支持多种语言的问答，这对国际化企业尤为重要。
第四章：性能深度分析——数据背后的真相
4.1 准确率提升的秘密：上下文就是一切
通过Vanna团队的详细实验，我们可以清晰地看到不同策略对准确率的影响：
实验数据深度分析：
-
Schema-only方法的失败原因：
-
缺乏表间关系信息 -
无法理解字段的业务含义 -
不知道如何正确JOIN表 -
无法处理复杂的业务逻辑
-
静态示例的局限性：
-
示例数量有限（通常3-5个） -
无法覆盖所有查询模式 -
缺乏针对性 -
上下文相关方法的优势：
-
动态选择最相关的示例 -
能够组合多个示例的模式 -
随着训练数据增加而改进 -
在所有策略下都表现最佳 -
特别擅长理解复杂的业务逻辑 -
能够很好地泛化示例 -
在上下文丰富时表现优异 -
对示例的依赖性更强 -
在复杂推理上稍逊于GPT-4 -
性价比最高 -
在简单查询上表现不错 -
复杂查询准确率较低
4.2 不同LLM的表现差异
有趣的是，实验结果显示了不同LLM在不同上下文策略下的表现差异：
GPT-4.1：
Google Bison：
GPT-4.1-mini：
4.3 性能优化策略
基于实验结果，我们可以总结出几个关键的性能优化策略：
策略1：智能上下文选择
def get_sql_prompt(self, question: str, **kwargs):
# 动态调整检索数量
n_results = min(10, max(3, len(question.split()) // 2))
question_sql_list = self.get_similar_question_sql(
question, n_results=n_results
)
# ...
策略2：分层训练数据管理
# 高质量的核心示例
core_examples = [
{"question": "...", "sql": "...", "priority": "high"},
# ...
]
# 自动生成的示例
auto_examples = [
{"question": "...", "sql": "...", "priority": "medium"},
# ...
]
策略3：持续学习机制
def continuous_learning(self):
# 定期分析查询日志
successful_queries = self.get_successful_queries()
# 自动提取新的训练样本
for query in successful_queries:
if self.is_novel_pattern(query):
self.add_question_sql(query.question, query.sql)
第五章：企业级应用——从概念验证到生产部署
5.1 架构设计：构建可扩展的AI SQL服务
在企业环境中部署Vanna需要考虑更多的因素：
# 企业级配置示例
class EnterpriseVanna:
def __init__(self):
self.config = {
# 多模型支持
'primary_llm': 'gpt-4',
'fallback_llm': 'gpt-3.5-turbo',
# 向量数据库集群
'vector_store': {
'type': 'qdrant',
'cluster_urls': ['http://qdrant-1:6333', 'http://qdrant-2:6333'],
'collection_name': 'enterprise_sql_kb'
},
# 安全配置
'security': {
'enable_query_validation': True,
'allowed_operations': ['SELECT'],
'max_result_rows': 10000,
'query_timeout': 30
},
# 监控配置
'monitoring': {
'enable_logging': True,
'log_level': 'INFO',
'metrics_endpoint': '/metrics'
}
}
5.2 安全考虑：保护企业数据的防线
企业级部署必须考虑安全性：
class SecureVanna(VannaBase):
def validate_sql(self, sql: str) -> bool:
"""SQL安全验证"""
# 检查危险操作
dangerous_keywords = ['DROP', 'DELETE', 'UPDATE', 'INSERT', 'TRUNCATE']
sql_upper = sql.upper()
for keyword in dangerous_keywords:
if keyword in sql_upper:
raise SecurityError(f"Dangerous operation detected: {keyword}")
# 检查表访问权限
tables = self.extract_table_names(sql)
for table in tables:
ifnot self.user_has_access(table):
raise PermissionError(f"Access denied to table: {table}")
returnTrue
def run_sql(self, sql: str, **kwargs) -> pd.DataFrame:
# 验证SQL安全性
self.validate_sql(sql)
# 添加行数限制
if'LIMIT'notin sql.upper():
sql += f" LIMIT {self.config['max_result_rows']}"
return super().run_sql(sql, **kwargs)
5.3 监控和可观测性：让AI透明化
import logging
from prometheus_client import Counter, Histogram, Gauge
class MonitoredVanna(VannaBase):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Prometheus指标
self.query_counter = Counter('vanna_queries_total', 'Total queries')
self.query_duration = Histogram('vanna_query_duration_seconds', 'Query duration')
self.accuracy_gauge = Gauge('vanna_accuracy_rate', 'Current accuracy rate')
# 日志配置
logging.basicConfig(level=logging.INFO)
self.logger = logging.getLogger(__name__)
def ask(self, question: str, **kwargs):
start_time = time.time()
self.query_counter.inc()
try:
result = super().ask(question, **kwargs)
# 记录成功查询
self.logger.info(f"Successful query: {question}")
duration = time.time() - start_time
self.query_duration.observe(duration)
return result
except Exception as e:
# 记录失败查询
self.logger.error(f"Failed query: {question}, Error: {str(e)}")
raise
5.4 多租户支持：一套系统服务多个业务
class MultiTenantVanna(VannaBase):
def __init__(self, tenant_id: str, **kwargs):
self.tenant_id = tenant_id
# 租户隔离的配置
config = kwargs.get('config', {})
config['collection_name'] = f"vanna_{tenant_id}"
super().__init__(config=config)
def add_question_sql(self, question: str, sql: str, **kwargs):
# 添加租户标识
metadata = kwargs.get('metadata', {})
metadata['tenant_id'] = self.tenant_id
return super().add_question_sql(
question, sql, metadata=metadata, **kwargs
)
第六章：实战案例——真实场景中的Vanna应用
6.1 案例一：电商数据分析平台
背景：某大型电商公司有复杂的数据仓库，包含用户、订单、商品、物流等多个业务域的数据。业务分析师经常需要进行复杂的数据查询。
挑战：
-
数据表超过200个 -
复杂的表关系 -
业务逻辑复杂 -
分析师SQL水平参差不齐
Vanna解决方案：
# 训练数据示例
training_examples = [
{
"question": "最近30天每日GMV趋势",
"sql": """
SELECT
DATE(order_time) as date,
SUM(total_amount) as gmv
FROM orders
WHERE order_time >= CURRENT_DATE - INTERVAL '30 days'
AND order_status = 'completed'
GROUP BY DATE(order_time)
ORDER BY date;
"""
},
{
"question": "各品类的复购率",
"sql": """
WITH user_category_orders AS (
SELECT
u.user_id,
p.category_id,
COUNT(DISTINCT o.order_id) as order_count
FROM users u
JOIN orders o ON u.user_id = o.user_id
JOIN order_items oi ON o.order_id = oi.order_id
JOIN products p ON oi.product_id = p.product_id
WHERE o.order_status = 'completed'
GROUP BY u.user_id, p.category_id
)
SELECT
c.category_name,
COUNT(CASE WHEN uco.order_count > 1 THEN 1 END) * 100.0 / COUNT(*) as repurchase_rate
FROM user_category_orders uco
JOIN categories c ON uco.category_id = c.category_id
GROUP BY c.category_name;
"""
}
]
效果：
-
分析师查询效率提升300% -
SQL错误率下降90% -
新员工上手时间从2周缩短到2天
6.2 案例二：金融风控数据查询
背景：金融公司需要实时监控各种风险指标，业务人员需要快速获取风控数据。
特殊要求：
-
高安全性要求 -
实时性要求高 -
查询结果需要审计
Vanna定制方案：
class FinanceVanna(VannaBase):
def __init__(self, user_role: str, **kwargs):
super().__init__(**kwargs)
self.user_role = user_role
self.audit_logger = AuditLogger()
def ask(self, question: str, **kwargs):
# 审计日志
self.audit_logger.log_query_request(
user_role=self.user_role,
question=question,
timestamp=datetime.now()
)
# 基于角色的查询限制
if self.user_role == 'analyst':
# 分析师只能查询汇总数据
kwargs['aggregation_only'] = True
elif self.user_role == 'manager':
# 经理可以查询详细数据但有行数限制
kwargs['max_rows'] = 1000
result = super().ask(question, **kwargs)
# 记录查询结果
self.audit_logger.log_query_result(
user_role=self.user_role,
question=question,
sql=result[0] if result elseNone,
row_count=len(result[1]) if result and result[1] isnotNoneelse0
)
return result
6.3 案例三：制造业IoT数据分析
背景：制造企业有大量IoT设备数据，需要进行设备状态监控和预测性维护分析。
技术挑战：
-
时序数据查询复杂 -
数据量巨大 -
需要支持多种聚合函数
解决方案：
# 专门的时序数据训练
time_series_examples = [
{
"question": "设备A最近24小时的温度异常点",
"sql": """
WITH temp_stats AS (
SELECT
AVG(temperature) as avg_temp,
STDDEV(temperature) as std_temp
FROM sensor_data
WHERE device_id = 'A'
AND timestamp >= NOW() - INTERVAL '24 hours'
)
SELECT
timestamp,
temperature,
ABS(temperature - ts.avg_temp) / ts.std_temp as z_score
FROM sensor_data sd
CROSS JOIN temp_stats ts
WHERE sd.device_id = 'A'
AND sd.timestamp >= NOW() - INTERVAL '24 hours'
AND ABS(sd.temperature - ts.avg_temp) / ts.std_temp > 2
ORDER BY timestamp;
"""
}
]
第七章：未来展望——AI+SQL的下一个十年
7.1 技术发展趋势
1. 多模态查询支持
未来的Vanna可能支持：
-
语音输入："帮我查一下销售数据" -
图表输入：上传一个图表，自动生成类似的查询 -
自然语言+手势：在移动设备上的新交互方式
2. 自动化数据治理
# 未来可能的功能
class SmartVanna(VannaBase):
def auto_discover_schema(self):
"""自动发现和理解数据库结构"""
pass
def suggest_data_quality_checks(self):
"""基于查询模式建议数据质量检查"""
pass
def auto_generate_documentation(self):
"""自动生成数据字典和业务文档"""
pass
3. 联邦查询支持
# 跨数据源查询
vn.ask("比较我们在MySQL中的销售数据和Snowflake中的财务数据")
# 自动生成跨数据源的查询计划
7.2 行业应用前景
数据民主化的实现
Vanna这样的工具正在推动"数据民主化"的实现：
-
降低技术门槛：非技术人员也能进行复杂数据分析 -
提高决策速度：从提出问题到获得答案的时间大幅缩短 -
减少人为错误：AI生成的SQL通常比人工编写的更准确
新的职业角色
-
AI训练师：专门负责训练和优化AI模型 -
数据产品经理：设计AI驱动的数据产品 -
智能分析师：结合AI工具进行高级数据分析
7.3 技术挑战与机遇
挑战
-
隐私和安全：如何在保护数据隐私的同时提供智能服务 -
可解释性：如何让AI的决策过程更透明 -
准确性保证：如何确保生成的SQL在所有情况下都是正确的
机遇
-
边缘计算：在本地部署小型化的AI模型 -
实时分析：结合流处理技术实现实时智能查询 -
自动化运维：AI驱动的数据库优化和维护
第八章：动手实践——构建你的第一个AI SQL助手
8.1 环境准备
# 安装Vanna
pip install vanna
# 安装可选依赖
pip install vanna[openai,chromadb,postgres]
8.2 完整示例：构建一个电商数据助手
import os
from vanna.openai.openai_chat import OpenAI_Chat
from vanna.chromadb.chromadb_vector import ChromaDB_VectorStore
class EcommerceVanna(ChromaDB_VectorStore, OpenAI_Chat):
def __init__(self, config=None):
ChromaDB_VectorStore.__init__(self, config=config)
OpenAI_Chat.__init__(self, config=config)
# 初始化
vn = EcommerceVanna(config={
'api_key': os.getenv('OPENAI_API_KEY'),
'model': 'gpt-4',
'path': './vanna_db'# ChromaDB存储路径
})
# 连接数据库
vn.connect_to_postgres(
host="localhost",
dbname="ecommerce",
user="postgres",
password="password"
)
# 训练数据
def train_ecommerce_model():
# 添加表结构
ddl_statements = [
"""
CREATE TABLE users (
user_id SERIAL PRIMARY KEY,
username VARCHAR(50) UNIQUE NOT NULL,
email VARCHAR(100) UNIQUE NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
country VARCHAR(2)
);
""",
"""
CREATE TABLE products (
product_id SERIAL PRIMARY KEY,
product_name VARCHAR(200) NOT NULL,
category_id INTEGER,
price DECIMAL(10,2),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""",
"""
CREATE TABLE orders (
order_id SERIAL PRIMARY KEY,
user_id INTEGER REFERENCES users(user_id),
order_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
total_amount DECIMAL(10,2),
status VARCHAR(20) DEFAULT 'pending'
);
"""
]
for ddl in ddl_statements:
vn.train(ddl=ddl)
# 添加业务文档
documentation = [
"用户表(users)存储所有注册用户信息，country字段使用ISO 3166-1 alpha-2标准",
"订单表(orders)记录所有订单，status字段可能值：pending, completed, cancelled",
"产品表(products)存储商品信息，price字段为美元价格"
]
for doc in documentation:
vn.train(documentation=doc)
# 添加示例查询
examples = [
{
"question": "今天有多少新用户注册？",
"sql": "SELECT COUNT(*) FROM users WHERE DATE(created_at) = CURRENT_DATE;"
},
{
"question": "最近7天的日均订单金额是多少？",
"sql": """
SELECT AVG(daily_total) as avg_daily_amount
FROM (
SELECT DATE(order_date) as date, SUM(total_amount) as daily_total
FROM orders
WHERE order_date >= CURRENT_DATE - INTERVAL '7 days'
AND status = 'completed'
GROUP BY DATE(order_date)
) daily_totals;
"""
},
{
"question": "哪个国家的用户最多？",
"sql": """
SELECT country, COUNT(*) as user_count
FROM users
WHERE country IS NOT NULL
GROUP BY country
ORDER BY user_count DESC
LIMIT 1;
"""
}
]
for example in examples:
vn.train(question=example["question"], sql=example["sql"])
# 执行训练
train_ecommerce_model()
# 开始使用
if __name__ == "__main__":
whileTrue:
question = input("请输入你的问题（输入'quit'退出）: ")
if question.lower() == 'quit':
break
try:
sql, df, fig = vn.ask(question)
print(f"n生成的SQL:n{sql}")
print(f"n查询结果:n{df}")
if fig:
fig.show() # 显示图表
except Exception as e:
print(f"查询失败: {str(e)}")
8.3 高级定制：添加业务特定功能
class AdvancedEcommerceVanna(EcommerceVanna):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.business_metrics = {
'GMV': 'Gross Merchandise Value - 总商品交易额',
'AOV': 'Average Order Value - 平均订单价值',
'LTV': 'Lifetime Value - 客户生命周期价值',
'CAC': 'Customer Acquisition Cost - 客户获取成本'
}
def preprocess_question(self, question: str) -> str:
"""预处理问题，替换业务术语"""
for abbr, full_name in self.business_metrics.items():
if abbr.lower() in question.lower():
question = question.replace(abbr, full_name)
return question
def ask(self, question: str, **kwargs):
# 预处理问题
processed_question = self.preprocess_question(question)
# 添加业务上下文
if any(metric in processed_question.lower() for metric in self.business_metrics.values()):
kwargs['include_business_context'] = True
return super().ask(processed_question, **kwargs)
def generate_business_report(self, period: str = "last_30_days"):
"""生成业务报告"""
questions = [
f"What was the GMV for the {period}?",
f"What was the AOV for the {period}?",
f"How many new customers did we acquire in the {period}?",
f"What was the top-selling product category in the {period}?"
]
report = {}
for question in questions:
try:
sql, df, _ = self.ask(question, print_results=False)
report[question] = {
'sql': sql,
'result': df.to_dict('records') if df isnotNoneelseNone
}
except Exception as e:
report[question] = {'error': str(e)}
return report
结语：拥抱AI时代的数据分析新范式
通过深入分析Vanna项目，我们看到了RAG技术在SQL生成领域的巨大潜力。从3%到80%的准确率提升不仅仅是一个数字的变化，更代表着一种全新的数据分析范式的诞生。
关键洞察总结
-
上下文比模型更重要：即使是最强大的LLM，没有合适的上下文也无法生成准确的SQL -
RAG是解决方案：通过智能检索相关上下文，可以显著提升AI的表现 -
模块化设计的价值：Vanna的成功很大程度上归功于其优秀的架构设计 -
持续学习的重要性：AI系统需要能够从每次交互中学习和改进
对未来的思考
Vanna只是AI+SQL领域的一个开始。随着技术的不断发展，我们可以期待：
-
更智能的自然语言理解 -
更准确的SQL生成 -
更丰富的数据可视化 -
更强大的跨数据源查询能力
行动建议
对于技术从业者：
-
学习RAG技术：这是当前AI应用的核心技术之一 -
关注开源项目：像Vanna这样的项目提供了宝贵的学习资源 -
实践应用：在自己的项目中尝试集成AI能力
对于企业决策者：
-
评估数据民主化的价值：考虑AI工具对业务效率的提升 -
投资数据基础设施：良好的数据治理是AI应用的基础 -
培养复合型人才：既懂业务又懂技术的人才将更加重要
互动时间：让我们一起探索AI的无限可能
读到这里，相信你对Vanna和RAG技术有了深入的了解。但学习的旅程永远不会结束，我特别想听听你的想法：
🤔 讨论话题
-
你的使用场景：在你的工作中，有哪些场景可以应用类似Vanna的技术？ -
技术挑战：你认为当前AI生成SQL还面临哪些主要挑战？ -
创新想法：基于Vanna的架构，你能想到哪些有趣的扩展应用？ -
实践经验：如果你已经尝试过类似的技术，遇到了哪些问题和收获？
💡 实践挑战
我为大家准备了一个小挑战：
挑战题目：基于本文介绍的Vanna架构，设计一个针对你所在行业的AI SQL助手。请考虑：
-
你的行业有哪些特殊的数据特点？ -
需要哪些定制化的功能？ -
如何确保数据安全和隐私？
