LangChain与数据库的深度集成：Agent驱动的自然语言查询与可视化实践

一、业务方想用自然语言查数据，SQL写不出来

数据部门的日常困境：运营团队想知道"上周通过微信渠道进来的用户，购买转化率最高的是哪三个品类"，这条查询涉及多表JOIN、时间窗口筛选和聚合排序，运营自己写不出SQL，数据开发排期又要三天。更糟糕的是，即使给了SQL，非技术人员也无法理解复杂的查询逻辑，当数据不符合预期时无法判断是SQL写错了还是数据本身有问题。

让LLM将自然语言翻译为SQL（NL2SQL）不是一个新想法，但要在生产环境落地需要解决几个实际问题：数据库Schema的理解（几百张表、几千个列，LLM的上下文窗口装不下全量Schema），查询的安全性（DELETE/DROP等恶意或误操作的禁止），以及结果的准确性验证（LLM可能写出语法正确但语义错误的SQL）。Agent架构是将NL2SQL从demo提升到生产级的关键。

二、Text-to-SQL Agent的架构：Schema理解、查询生成与安全检查

一个生产级的Text-to-SQL Agent需要四个核心组件协同工作：

• Schema检索器：从完整的数据库Schema中筛选出与用户问题相关的表和列，压缩后送入LLM上下文
• SQL生成器：基于筛选后的Schema和用户问题，用Few-Shot Prompting生成SQL
• 安全检查器：对生成的SQL做语法解析和安全规则验证
• 执行与反馈：执行SQL、捕获错误、根据错误信息自动修正

flowchart TD
    A[用户自然语言问题] --> B[Schema检索器<br/>向量相似度筛选相关表]
    B --> C[Few-Shot Prompt构造<br/>Schema + 示例SQL + 问题]
    C --> D[LLM SQL生成]
    D --> E{安全检查器}
    E -->|含危险操作| F[拒绝执行<br/>返回错误提示]
    E -->|安全| G[执行SQL]
    G --> H{执行成功?}
    H -->|失败| I[错误分析与SQL修正]
    I --> D
    H -->|成功| J[结果格式化与可视化]
    J --> K[NL摘要 + 图表 + 数据表格]
    
    style B fill:#c8e6c9
    style D fill:#fff3e0
    style E fill:#ffcdd2
    style K fill:#a5d6a7

Schema检索是链路中最关键但最容易被忽视的环节。一个典型的OLAP数据库可能有500+张表，每张表10-100列，总计数万列。把它们全部塞进LLM的prompt不仅浪费Token，还会让模型在无关信息中"迷路"。Schema检索的一般做法是：为每张表的描述和列注释建立Embedding向量，计算用户问题与各表/列的语义相似度，只选择Top-K最相关的表和列送入LLM。这种方法在实践中能将Schema Token从数万压缩到数百。

三、一个带权限隔离的NL2SQL Agent实现

import re
import sqlparse
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
import logging

logger = logging.getLogger(__name__)


@dataclass
class TableInfo:
    name: str
    description: str
    columns: List[Dict[str, str]]  # [{"name": "id", "type": "int", "comment": "..."}]


class SQLSecurityChecker:
    """SQL安全检查器——防止危险操作和权限越界"""

    FORBIDDEN_KEYWORDS = [
        'DROP', 'DELETE', 'TRUNCATE', 'ALTER', 'CREATE',
        'INSERT', 'UPDATE', 'GRANT', 'REVOKE', 'EXEC',
    ]

    # 允许的白名单操作
    ALLOWED_KEYWORDS = ['SELECT', 'WITH', 'EXPLAIN']

    def __init__(self, max_result_rows: int = 10000,
                 query_timeout_ms: int = 30000):
        self.max_result_rows = max_result_rows
        self.query_timeout_ms = query_timeout_ms

    def validate(self, sql: str, user_tables: List[str]) -> Tuple[bool, str]:
        """验证SQL的安全性和权限"""
        try:
            parsed = sqlparse.parse(sql)
            if not parsed:
                return False, "无法解析SQL"

            statement = parsed[0]
            stmt_type = statement.get_type()

            # 检查操作类型
            if stmt_type.upper() not in self.ALLOWED_KEYWORDS:
                return False, f"不允许的操作类型: {stmt_type}"

            # 检查是否包含禁止关键词
            sql_upper = sql.upper()
            for keyword in self.FORBIDDEN_KEYWORDS:
                if re.search(r'\b' + keyword + r'\b', sql_upper):
                    return False, f"SQL包含禁止的操作: {keyword}"

            # 检查表权限：用户只能查询授权表
            tables_in_sql = self._extract_tables(sql)
            for table in tables_in_sql:
                if table not in user_tables:
                    return False, f"无权访问表: {table}"

            # 添加资源限制
            limited_sql = self._add_resource_limits(sql)
            
            return True, limited_sql

        except Exception as e:
            logger.error(f"SQL安全检查异常: {e}")
            return False, f"SQL安全校验失败: {str(e)}"

    def _extract_tables(self, sql: str) -> List[str]:
        """从SQL中提取表名"""
        tables = set()
        parsed = sqlparse.parse(sql)
        for statement in parsed:
            from_seen = False
            for token in statement.flatten():
                if token.ttype is None and token.value.upper() == 'FROM':
                    from_seen = True
                    continue
                if from_seen and token.ttype is sqlparse.tokens.Name:
                    # 去schema前缀
                    table_name = token.value.split('.')[-1].strip('`"\'')
                    tables.add(table_name)
                    from_seen = False
                if token.ttype is None and token.value.upper() == 'JOIN':
                    from_seen = True
        return list(tables)

    def _add_resource_limits(self, sql: str) -> str:
        """为SQL添加资源限制"""
        sql = sql.rstrip(';').strip()
        limits = []
        
        if 'LIMIT' not in sql.upper():
            limits.append(f"LIMIT {self.max_result_rows}")
        
        # 添加查询超时
        if 'max_execution_time' not in sql.lower():
            limits.insert(0, f"SET max_execution_time={self.query_timeout_ms};")

        if limits:
            return '; '.join(limits) + ' ' + sql
        return sql


class NL2SQLAgent:
    """NL2SQL Agent主控"""

    def __init__(self, db_schema: List[TableInfo], 
                 user_permissions: Dict[str, List[str]]):
        """
        user_permissions: {"user_id": ["table1", "table2"]}
        """
        self.schema = db_schema
        self.permissions = user_permissions
        self.security = SQLSecurityChecker()

    def query(self, user_id: str, question: str) -> Dict:
        """处理用户的自然语言查询"""
        # 1. 权限检查
        user_tables = self.permissions.get(user_id, [])
        if not user_tables:
            return {"error": "无数据查询权限", "sql": None, "data": None}

        # 2. Schema检索
        relevant_tables = self._retrieve_schema(question, user_tables)
        if not relevant_tables:
            return {"error": "未找到与问题相关的数据表", "sql": None, "data": None}

        # 3. 生成SQL（这里简化为调用LLM的占位符）
        try:
            schema_text = self._format_schema(relevant_tables)
            sql = self._generate_sql(schema_text, question)
        except Exception as e:
            return {"error": f"SQL生成失败: {str(e)}", "sql": None, "data": None}

        # 4. 安全检查
        is_safe, result = self.security.validate(sql, user_tables)
        if not is_safe:
            return {"error": result, "sql": sql, "data": None}

        # 5. 返回安全的SQL（实际执行需异步处理）
        return {
            "sql": result,
            "original_sql": sql,
            "tables_used": [t.name for t in relevant_tables],
            "explanation": f"查询涉及 {len(relevant_tables)} 张表",
        }

    def _retrieve_schema(self, question: str, 
                         user_tables: List[str]) -> List[TableInfo]:
        """基于问题检索相关Schema（简化版：全返回授权表）"""
        return [t for t in self.schema if t.name in user_tables]

    def _format_schema(self, tables: List[TableInfo]) -> str:
        """格式化Schema为prompt文本"""
        lines = []
        for table in tables:
            lines.append(f"## {table.name}: {table.description}")
            for col in table.columns:
                lines.append(f"  - {col['name']} ({col['type']}): {col.get('comment', '')}")
        return '\n'.join(lines)

    def _generate_sql(self, schema_text: str, question: str) -> str:
        """调用LLM生成SQL（占位实现）"""
        prompt = f"""根据以下数据库Schema生成SQL查询语句，只返回纯SQL不要解释。

{schema_text}

用户问题: {question}

SQL:"""
        # 实际应调用LLM API
        logger.info(f"SQL prompt: {prompt[:200]}...")
        return "SELECT 1"  # 占位

许可证模型的设计是权限隔离的核心。不是简单的"能查/不能查"的二元控制，而是列级别甚至行级别的精细权限。上述实现支持表级别的权限控制，在安全检查器中提取SQL中引用的所有表并逐一验证。对于金融等高合规场景，还需要增加数据脱敏规则——即使查询返回了敏感列，在输出给用户前做脱敏处理。

四、当SQL生成错误时：自动修正 vs 人工干预的边界在哪里

LLM生成的SQL有三种常见错误模式。语法错误最容易自动修正——将错误信息反馈给LLM做一次重试，大多数情况下能纠正。语义错误（SQL语法正确但逻辑不对，如选了错误的聚合列、JOIN条件遗漏）最难检测，因为SQL能正确执行返回数据，但数据不符合用户预期。性能风险在高频调用场景中尤为危险——一条缺少过滤条件的全表扫描SQL，在百亿级别的大表上可能跑几分钟还跑不完。

自动修正与人工干预的边界应当由置信度和风险等级来决定。对于低风险操作（只读、表小于10万行、执行预估小于1秒），可以自动执行并修正；中风险操作需要用户确认SQL；高风险操作（涉及超过100万行的表、预估执行时间超过10秒）要求用户审核SQL并在确认后才能执行。这个分层策略通过SQL的EXPLAIN结果和实施前代价估算来自动判定风险等级。

五、总结

NL2SQL的生产落地不是简单的"自然语言→LLM→SQL"，而是Schema检索、SQL生成、安全检查、权限控制和错误修正五个环节的精密配合。Agent架构将每个环节模块化，支持独立的优化和监控。在当前的技术水平下，简单的过滤聚合查询准确率达到85%-90%，但复杂的多表JOIN和嵌套子查询准确率只有60%-70%。建议从简单查询场景起步，逐步扩展到复杂查询，同时建立用户反馈闭环来持续优化Few-Shot示例库。