免费POC, 零成本试错
AI知识库

53AI知识库

学习大模型的前沿技术与行业应用场景


从3%到80%:揭秘Vanna如何用RAG技术革命性地解决AI生成SQL的准确率难题

发布日期:2025-08-25 07:56:33 浏览次数: 1552
作者:许泽宇的技术分享

微信搜一搜,关注“许泽宇的技术分享”

推荐语

Vanna用RAG技术将AI生成SQL的准确率从3%提升到80%,彻底改变了数据库查询的游戏规则。

核心内容:
1. AI生成SQL面临的数据库结构理解难题
2. Vanna基于RAG架构的智能SQL生成框架解析
3. 模块化设计带来的技术突破与实战效果

杨芳贤
53AI创始人/腾讯云(TVP)最具价值专家

"在数据驱动的时代,让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生成过程分解为三个可插拔的组件:

  1. 嵌入生成器:负责将文本转换为向量表示
  2. 向量存储器:负责存储和检索相关上下文
  3. 语言模型:负责基于上下文生成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-3.5
GPT-4
Bison
平均准确率
仅Schema
0%
10%
0%
3.3%
静态示例
25%
65%
30%
40%
上下文相关
70%
85%
85%
80%

这个结果揭示了一个重要的洞察:上下文比模型更重要。即使是最强大的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的核心创新在于其智能检索机制。当用户提出问题时,系统会:

  1. 向量化用户问题
  2. 在三个向量空间中并行搜索
  3. 基于相似度排序返回最相关的上下文
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团队的详细实验,我们可以清晰地看到不同策略对准确率的影响:

实验数据深度分析

  1. Schema-only方法的失败原因

  • 缺乏表间关系信息
  • 无法理解字段的业务含义
  • 不知道如何正确JOIN表
  • 无法处理复杂的业务逻辑
  • 静态示例的局限性

    • 示例数量有限(通常3-5个)
    • 无法覆盖所有查询模式
    • 缺乏针对性
  • 上下文相关方法的优势

    • 动态选择最相关的示例
    • 能够组合多个示例的模式
    • 随着训练数据增加而改进

    4.2 不同LLM的表现差异

    有趣的是,实验结果显示了不同LLM在不同上下文策略下的表现差异:

    GPT-4.1

    • 在所有策略下都表现最佳
    • 特别擅长理解复杂的业务逻辑
    • 能够很好地泛化示例

    Google Bison

    • 在上下文丰富时表现优异
    • 对示例的依赖性更强
    • 在复杂推理上稍逊于GPT-4

    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[0if result elseNone,
                row_count=len(result[1]) if result and result[1isnotNoneelse0
            )
            
            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 技术挑战与机遇

    挑战

    1. 隐私和安全:如何在保护数据隐私的同时提供智能服务
    2. 可解释性:如何让AI的决策过程更透明
    3. 准确性保证:如何确保生成的SQL在所有情况下都是正确的

    机遇

    1. 边缘计算:在本地部署小型化的AI模型
    2. 实时分析:结合流处理技术实现实时智能查询
    3. 自动化运维: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%的准确率提升不仅仅是一个数字的变化,更代表着一种全新的数据分析范式的诞生。

    关键洞察总结

    1. 上下文比模型更重要:即使是最强大的LLM,没有合适的上下文也无法生成准确的SQL
    2. RAG是解决方案:通过智能检索相关上下文,可以显著提升AI的表现
    3. 模块化设计的价值:Vanna的成功很大程度上归功于其优秀的架构设计
    4. 持续学习的重要性:AI系统需要能够从每次交互中学习和改进

    对未来的思考

    Vanna只是AI+SQL领域的一个开始。随着技术的不断发展,我们可以期待:

    • 更智能的自然语言理解
    • 更准确的SQL生成
    • 更丰富的数据可视化
    • 更强大的跨数据源查询能力

    行动建议

    对于技术从业者:

    1. 学习RAG技术:这是当前AI应用的核心技术之一
    2. 关注开源项目:像Vanna这样的项目提供了宝贵的学习资源
    3. 实践应用:在自己的项目中尝试集成AI能力

    对于企业决策者:

    1. 评估数据民主化的价值:考虑AI工具对业务效率的提升
    2. 投资数据基础设施:良好的数据治理是AI应用的基础
    3. 培养复合型人才:既懂业务又懂技术的人才将更加重要

    互动时间:让我们一起探索AI的无限可能

    读到这里,相信你对Vanna和RAG技术有了深入的了解。但学习的旅程永远不会结束,我特别想听听你的想法:

    🤔 讨论话题

    1. 你的使用场景:在你的工作中,有哪些场景可以应用类似Vanna的技术?
    2. 技术挑战:你认为当前AI生成SQL还面临哪些主要挑战?
    3. 创新想法:基于Vanna的架构,你能想到哪些有趣的扩展应用?
    4. 实践经验:如果你已经尝试过类似的技术,遇到了哪些问题和收获?

    💡 实践挑战

    我为大家准备了一个小挑战:

    挑战题目:基于本文介绍的Vanna架构,设计一个针对你所在行业的AI SQL助手。请考虑:

    • 你的行业有哪些特殊的数据特点?
    • 需要哪些定制化的功能?
    • 如何确保数据安全和隐私?

53AI,企业落地大模型首选服务商

产品:场景落地咨询+大模型应用平台+行业解决方案

承诺:免费POC验证,效果达标后再合作。零风险落地应用大模型,已交付160+中大型企业

联系我们

售前咨询
186 6662 7370
预约演示
185 8882 0121

微信扫码

添加专属顾问

回到顶部

加载中...

扫码咨询