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示例库。