LangChain与数据库的深度集成Agent驱动的自然语言查询与可视化实践一、业务方想用自然语言查数据SQL写不出来数据部门的日常困境运营团队想知道上周通过微信渠道进来的用户购买转化率最高的是哪三个品类这条查询涉及多表JOIN、时间窗口筛选和聚合排序运营自己写不出SQL数据开发排期又要三天。更糟糕的是即使给了SQL非技术人员也无法理解复杂的查询逻辑当数据不符合预期时无法判断是SQL写错了还是数据本身有问题。让LLM将自然语言翻译为SQLNL2SQL不是一个新想法但要在生产环境落地需要解决几个实际问题数据库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:#a5d6a7Schema检索是链路中最关键但最容易被忽视的环节。一个典型的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, fSQL包含禁止的操作: {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(fSQL安全检查异常: {e}) return False, fSQL安全校验失败: {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(fLIMIT {self.max_result_rows}) # 添加查询超时 if max_execution_time not in sql.lower(): limits.insert(0, fSET 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: fSQL生成失败: {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(fSQL 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示例库。