Text-to-SQL Agent for Natural Language Data Query

1. 解决的问题

业务团队（运营、市场、产品）需要数据但不懂SQL。

2. 核心算法逻辑

核心问题：业务团队（运营、市场、产品）需要数据但不懂SQL。每次提数需求都要排期给数据团队，平均等待23天。TexttoSQL让非技术用户用自然语言直接查询数据库。

3. 业务应用场景

业务问题：运营团队每天问数据团队同样的问题："昨天德国站奶粉类目的转化率是多少？"、"上周各品类的退货率排名"。数据团队80%时间在回答重复问题。

2. 用户提问："上个月德国站各品类的转化率和退货率"

3. Agent处理： - Schema Linking：识别相关表（orders, users）、字段（country, category, amount, is_returned, order_date） - SQL生成：

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

• ROI：数据团队重复取数工作量减少80%，业务决策速度提升10倍
• 难度：⭐⭐⭐☆☆（3/5）— 高频场景用模板，复杂场景用LLM
• 优先级：⭐⭐⭐⭐⭐（5/5）— 数据民主化的核心基础设施

7. 代码模板

代码块数量：3 · 路径：未检测到

"""
Text-to-SQL Agent — 自然语言转SQL查询
支持：Schema理解、SQL生成、执行、结果解释
"""

import re
from typing import List, Dict, Optional


class SchemaManager:
    """数据库Schema管理器"""

    def __init__(self):
        self.tables = {}

    def add_table(self, name: str, columns: Dict[str, str], description: str = ""):
        """
        添加表结构

        Args:
            name: 表名
            columns: {字段名: 字段描述}
            description: 表用途描述
        """
        self.tables[name] = {
            'columns': columns,
            'description': description
        }

    def get_schema_prompt(self) -> str:
        """生成Schema描述Prompt"""
        lines = ["数据库Schema:"]
        for table_name, info in self.tables.items():
            lines.append(f"\n表: {table_name}")
            if info['description']:
                lines.append(f"  用途: {info['description']}")
            lines.append("  字段:")
            for col, desc in info['columns'].items():
                lines.append(f"    - {col}: {desc}")
        return "\n".join(lines)

    def find_relevant_tables(self, query: str) -> List[str]:
        """根据查询内容识别相关表"""
        query_lower = query.lower()
        relevant = []
        for table_name, info in self.tables.items():
            score = 0
            # 表名匹配
            if table_name.lower() in query_lower:
                score += 3
            # 字段描述匹配
            for col, desc in info['columns'].items():
                if any(keyword in query_lower for keyword in [col.lower(), desc.lower()]):
                    score += 1
            if score > 0:
                relevant.append((table_name, score))

        relevant.sort(key=lambda x: x[1], reverse=True)
        return [t[0] for t in relevant[:3]]  # 返回Top-3相关表

8. 论文来源

未自动抽取；请查看原始 Skill 卡片。