搭建NL2SQL基本架构
·
目录
目标
使用LangChain搭建自然语言转SQL案例。
版本
Python 3.12
LangChain 1.2.10
MySQL 8.0.37
Ollama 0.16.3
官网
LangChain官网
https://docs.langchain.com/oss/python/langchain/install
LangSmith官网https://smith.langchain.com/
实战
创建项目并下载依赖
第一步：搭建本地大模型平台。我选用的是Ollama，因为它属于轻量级平台，具体搭建方法参考我之前的博客。
第二步：安装LangChain、MySQL客户端库、LangGraph等相关依赖。
pip install -U langchain
pip install -U langchain-openai
pip install dotenv
pip install -U langgraph
pip install -U langchain-ollama
pip install --upgrade "langgraph-cli[inmem]"
pip install sqlalchemy pymysql loguru第三步：新建Python项目。根据官方文档，我们需要把Python版本定义为3.10+，我这里定义为3.12版本。
第四步：创建src目录并将它设置为源代码根目录。在src下创建agent目录，后续我们的智能体、工具类等就在这个目录下创建。创建README.md文件，它是LangChain的要求规范，所以即便我们什么都不写，也把README.md文件创建好。
项目名称
|
src
|
agent
|
README.md
第五步：去LangSmith官网获取API密钥，并将密钥设置到环境变量（.env文件，它相当于SpringBoot项目中的application.yml文件）中。我给这个密钥接收变量是LANGSMITH_API_KEY。
项目名称
|
.env获取API密钥https://smith.langchain.com/o/21cd98bd-cf51-413c-acef-d5abda7c618f/settings/apikeys
第六步：创建langgraph.json文件，这是LangGraph的配置文件，里面配置了环境变量的配置文件和对应的智能体。
{
"dependencies": ["."],
"graphs": {
"agent" :"这里先空着，后面配置agent路径。"
},
"env": ".env"
}项目名称
|
langgraph.json第七步：创建pyproject.toml配置文件，它是Python项目的统一配置文件，相当于Maven的pom.xml文件。内容如下：
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "agent"
version = "0.0.1"
description = "An excellent agent build for LangSmith."
authors = [
{ name = "Polly the parrot", email = "1223+polly@users.noreply.github.com" }
]
license = { text = "MIT" }
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"langgraph>=0.6.0",
"langchain-fireworks>=0.1.3",
"protobuf>=6.31,<7"
]
#[tool.hatch.build.targets.wheel]
#packages = ["agent"]
[tool.setuptools.packages.find]
where = ["src"]第八步：执行以下命令（相当于Maven项目中的mvn install）。此时会在项目下生成.langgraph_api文件夹，里面有很多pckl文件。
pip install -e .
创建MySQL工具类
第一步：创建MySQL数据库，并新建一些表和测试数据。我创建了六个表和部分测试数据。
/*
SQLyog Trial v13.2.1 (64 bit)
MySQL - 8.0.37 : Database - nl_to_sql_db
*********************************************************************
*/
/*!40101 SET NAMES utf8 */;
/*!40101 SET SQL_MODE=''*/;
/*!40014 SET @OLD_UNIQUE_CHECKS=@@UNIQUE_CHECKS, UNIQUE_CHECKS=0 */;
/*!40014 SET @OLD_FOREIGN_KEY_CHECKS=@@FOREIGN_KEY_CHECKS, FOREIGN_KEY_CHECKS=0 */;
/*!40101 SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='NO_AUTO_VALUE_ON_ZERO' */;
/*!40111 SET @OLD_SQL_NOTES=@@SQL_NOTES, SQL_NOTES=0 */;
CREATE DATABASE /*!32312 IF NOT EXISTS*/`nl_to_sql_db` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */;
USE `nl_to_sql_db`;
/*Table structure for table `class_info` */
DROP TABLE IF EXISTS `class_info`;
CREATE TABLE `class_info` (
`id` bigint NOT NULL AUTO_INCREMENT COMMENT '班级ID',
`class_name` varchar(32) NOT NULL COMMENT '班级名称',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='班级信息表';
/*Data for the table `class_info` */
insert into `class_info`(`id`,`class_name`) values
(1,'一班'),
(2,'二班'),
(3,'三班');
/*Table structure for table `class_teacher_map` */
DROP TABLE IF EXISTS `class_teacher_map`;
CREATE TABLE `class_teacher_map` (
`id` bigint NOT NULL AUTO_INCREMENT,
`class_id` bigint NOT NULL COMMENT '班级ID',
`teacher_id` bigint NOT NULL COMMENT '教师ID',
`subject_id` bigint DEFAULT NULL COMMENT '所教科目ID',
PRIMARY KEY (`id`),
KEY `fk_class_teacher_class` (`class_id`),
KEY `fk_class_teacher_teacher` (`teacher_id`),
KEY `fk_class_teacher_subject` (`subject_id`),
CONSTRAINT `fk_class_teacher_class` FOREIGN KEY (`class_id`) REFERENCES `class_info` (`id`) ON DELETE CASCADE ON UPDATE CASCADE,
CONSTRAINT `fk_class_teacher_subject` FOREIGN KEY (`subject_id`) REFERENCES `subject_info` (`id`) ON DELETE SET NULL ON UPDATE CASCADE,
CONSTRAINT `fk_class_teacher_teacher` FOREIGN KEY (`teacher_id`) REFERENCES `teacher_info` (`id`) ON DELETE CASCADE ON UPDATE CASCADE
) ENGINE=InnoDB AUTO_INCREMENT=12 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='班级_教师_所教科目映射表';
/*Data for the table `class_teacher_map` */
insert into `class_teacher_map`(`id`,`class_id`,`teacher_id`,`subject_id`) values
(2,1,1,1),
(3,1,2,2),
(4,1,3,3),
(6,2,4,1),
(7,2,4,2),
(8,2,5,3),
(9,3,6,1),
(10,3,2,2),
(11,3,5,3);
/*Table structure for table `student_info` */
DROP TABLE IF EXISTS `student_info`;
CREATE TABLE `student_info` (
`id` bigint NOT NULL AUTO_INCREMENT COMMENT '学号',
`stu_name` varchar(32) DEFAULT NULL COMMENT '姓名',
`age` int DEFAULT NULL COMMENT '年龄',
`sex` int DEFAULT NULL COMMENT '性别：1=男；2=女',
`stu_class` bigint DEFAULT NULL COMMENT '班级ID',
PRIMARY KEY (`id`),
KEY `fk_student_class` (`stu_class`),
CONSTRAINT `fk_student_class` FOREIGN KEY (`stu_class`) REFERENCES `class_info` (`id`) ON DELETE SET NULL ON UPDATE CASCADE
) ENGINE=InnoDB AUTO_INCREMENT=6 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='学生信息表';
/*Data for the table `student_info` */
insert into `student_info`(`id`,`stu_name`,`age`,`sex`,`stu_class`) values
(1,'张三',11,1,NULL),
(2,'李四',12,2,NULL),
(3,'王五',12,1,NULL),
(4,'洪六',13,2,NULL),
(5,'钱七',14,1,NULL);
/*Table structure for table `student_score` */
DROP TABLE IF EXISTS `student_score`;
CREATE TABLE `student_score` (
`id` bigint NOT NULL AUTO_INCREMENT,
`stu_id` bigint NOT NULL COMMENT '学号',
`subject_id` bigint DEFAULT NULL COMMENT '科目ID',
`score` decimal(4,1) DEFAULT NULL COMMENT '分数',
`exam_type` varchar(64) DEFAULT NULL COMMENT '考试类型',
PRIMARY KEY (`id`),
KEY `fk_score_student` (`stu_id`),
KEY `fk_score_subject` (`subject_id`),
CONSTRAINT `fk_score_student` FOREIGN KEY (`stu_id`) REFERENCES `student_info` (`id`) ON DELETE CASCADE ON UPDATE CASCADE,
CONSTRAINT `fk_score_subject` FOREIGN KEY (`subject_id`) REFERENCES `subject_info` (`id`) ON DELETE SET NULL ON UPDATE CASCADE
) ENGINE=InnoDB AUTO_INCREMENT=16 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='学生成绩表';
/*Data for the table `student_score` */
insert into `student_score`(`id`,`stu_id`,`subject_id`,`score`,`exam_type`) values
(1,1,1,80.0,'期末考试'),
(2,1,2,95.0,'期末考试'),
(3,1,3,95.5,'期末考试'),
(4,2,1,69.0,'期末考试'),
(5,2,2,85.0,'期末考试'),
(6,2,3,75.0,'期末考试'),
(7,3,1,52.0,'期末考试'),
(8,3,2,74.0,'期末考试'),
(9,3,3,99.0,'期末考试'),
(10,4,1,100.0,'期末考试'),
(11,4,2,96.0,'期末考试'),
(12,4,3,48.0,'期末考试'),
(13,5,1,0.0,'期末考试'),
(14,5,2,66.0,'期末考试'),
(15,5,3,87.0,'期末考试');
/*Table structure for table `subject_info` */
DROP TABLE IF EXISTS `subject_info`;
CREATE TABLE `subject_info` (
`id` bigint NOT NULL AUTO_INCREMENT,
`subject_name` varchar(32) NOT NULL COMMENT '科目名称，比如：语文、数学、英语、物理等学科',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='科目信息表';
/*Data for the table `subject_info` */
insert into `subject_info`(`id`,`subject_name`) values
(1,'英语'),
(2,'语文'),
(3,'数学');
/*Table structure for table `teacher_info` */
DROP TABLE IF EXISTS `teacher_info`;
CREATE TABLE `teacher_info` (
`id` bigint NOT NULL AUTO_INCREMENT COMMENT '教师ID',
`teacher_name` varchar(32) NOT NULL COMMENT '教师姓名',
`tel` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL COMMENT '电话号码',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=8 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='教师信息表';
/*Data for the table `teacher_info` */
insert into `teacher_info`(`id`,`teacher_name`,`tel`) values
(1,'陆小凤','13854125652'),
(2,'花满楼','13854125651'),
(3,'西门吹雪','13854125653'),
(4,'谢晓峰','13854125654'),
(5,'韩立','13854125655'),
(6,'张小凡','13854125656'),
(7,'向之礼','13854125657');
/*!40101 SET SQL_MODE=@OLD_SQL_MODE */;
/*!40014 SET FOREIGN_KEY_CHECKS=@OLD_FOREIGN_KEY_CHECKS */;
/*!40014 SET UNIQUE_CHECKS=@OLD_UNIQUE_CHECKS */;
/*!40111 SET SQL_NOTES=@OLD_SQL_NOTES */;第二步：创建MySQL数据库连接工具并进行测试。这些工具（方法）包含：
- get_tables_with_comments：获取数据库中所有表的名称和描述信息。
- get_table_schema：获取指定表的模式信息（包含字段注释/主键/外键/索引）。
- preprocess_sql：SQL预处理。
- execute_sql：执行SQL。
项目名称
|
src
|
utils
|
db_utils.py# sqlalchemy相当于mybatis
from typing import Optional, List, Dict, Any
from sqlalchemy import create_engine, inspect, text
from sqlalchemy.exc import SQLAlchemyError
from utils.log_utils import log
class MySQLDatabaseManager:
def __init__(self, connection_string: str):
self.engine = create_engine(
connection_string,
pool_size=10,
pool_recycle=3600,
pool_pre_ping=True,
)
# =========================
# 元数据
# =========================
def get_table_names(self) -> List[str]:
try:
inspector = inspect(self.engine)
return inspector.get_table_names()
except Exception as e:
log.exception(e)
return []
def get_tables_with_comments(self) -> List[Dict[str, str]]:
"""
获取数据库中所有表的名称和描述信息
"""
try:
query = text("""
SELECT TABLE_NAME, TABLE_COMMENT
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_TYPE = 'BASE TABLE'
AND TABLE_SCHEMA = DATABASE()
ORDER BY TABLE_NAME
""")
with self.engine.connect() as connection:
result = connection.execute(query)
return [
{"table_name": row[0], "table_comment": row[1]}
for row in result
]
except SQLAlchemyError as e:
log.exception(e)
raise ValueError(f"获取表名及描述信息失败：{str(e)}")
def get_table_schema(self, table_names: Optional[List[str]] = None) -> str:
"""
获取指定表的模式信息（包含字段注释/主键/外键/索引）
Args:
table_names: 表名列表，为空则获取所有表
"""
try:
inspector = inspect(self.engine)
schema_info: List[str] = []
tables_to_process = table_names if table_names else self.get_table_names()
for table_name in tables_to_process:
columns = inspector.get_columns(table_name)
pk_constraint = inspector.get_pk_constraint(table_name)
primary_keys = pk_constraint.get("constrained_columns", []) if pk_constraint else []
foreign_keys = inspector.get_foreign_keys(table_name)
indexes = inspector.get_indexes(table_name)
table_schema = f"表名: {table_name}\n列信息:\n"
for col in columns:
pk_indicator = "(主键)" if col["name"] in primary_keys else ""
comment = col.get("comment", "无注释")
table_schema += f" - {col['name']} : {col['type']} {pk_indicator} [注释: {comment}]\n"
if foreign_keys:
table_schema += "外键信息:\n"
for fk in foreign_keys:
table_schema += f" - 列 {fk['constrained_columns']} -> {fk['referred_table']}({fk['referred_columns']})\n"
if indexes:
table_schema += "索引信息:\n"
for idx in indexes:
table_schema += f" - {idx['name']} : {idx['column_names']} 唯一: {idx.get('unique', False)}\n"
table_schema += "\n"
schema_info.append(table_schema)
return "\n".join(schema_info)
except Exception as e:
log.exception(e)
raise ValueError(f"获取表结构失败: {str(e)}")
# =========================
# SQL 预处理
# =========================
def preprocess_sql(self, sql: str) -> str:
try:
if not sql:
raise ValueError("SQL 不能为空")
sql = sql.strip().rstrip(";")
lower_sql = sql.lower()
if not lower_sql.startswith("select"):
raise ValueError("仅允许执行 SELECT 查询")
forbidden = [
" drop ",
" truncate ",
" delete ",
" update ",
" insert ",
" alter ",
" create ",
" grant ",
" revoke ",
]
for keyword in forbidden:
if keyword in lower_sql:
raise ValueError(f"检测到危险SQL操作: {keyword.strip()}")
# 语法校验
explain_sql = f"EXPLAIN {sql}"
with self.engine.connect() as connection:
connection.execute(text(explain_sql))
return sql #返回原SQL
except SQLAlchemyError as e:
raise ValueError(f"SQL 语法或表结构错误: {str(e)}")
except Exception as e:
log.exception(e)
raise
# =========================
# SQL 执行
# =========================
def execute_sql(
self,
sql: str,
params: Optional[Dict[str, Any]] = None,
limit: Optional[int] = 100,
) -> List[Dict[str, Any]]:
try:
safe_sql = self.preprocess_sql(sql)
# 自动加 limit
if limit and " limit " not in safe_sql.lower():
safe_sql = f"{safe_sql} LIMIT {limit}"
with self.engine.connect() as connection:
result = connection.execute(text(safe_sql), params or {})
columns = result.keys()
return [dict(zip(columns, row)) for row in result]
except SQLAlchemyError as e:
log.exception(e)
raise ValueError(f"SQL 执行失败: {str(e)}")
if __name__ == '__main__':
host = "localhost"
port = 3306
username = "root"
password = "123456"
database = "nl_to_sql_db"
connection_string = f"mysql+pymysql://{username}:{password}@{host}/{database}?charset=utf8"
manager = MySQLDatabaseManager(connection_string)
# 测试表名及描述
tables_info = manager.get_tables_with_comments()
print(tables_info)
# 测试表结构
schema = manager.get_table_schema(['student_info'])
print(schema)
# 测试SQL预处理
sql_safe = manager.preprocess_sql("SELECT id FROM student_info")
print(sql_safe)
# 测试SQL执行
result = manager.execute_sql("SELECT id FROM student_info")
print(result)
创建日志工具类
项目名称
|
src
|
agent
|
log_utils.pyimport os
import sys
from loguru import logger
#获得项目的绝对路径
root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
#存放项目日志目录的绝对路径
log_dir = os.path.join(root_dir, 'logs')
if not os.path.exists(log_dir):
#如果日志目录不存在则创建
os.makedirs(log_dir)
class MyLogger:
def __init__(self):
#写日志的对象
self.logger=logger
#清空所有设置
self.logger.remove()
#添加控制台输出格式
self.logger.add(sys.stdout, level='DEBUG',format=
"<green>{time:YYYYMMDD HH:mm:ss}</green> | " # 额色>时间
"{process.name} | " # 进程名
"{thread.name} | " #线程名
"<cyan>{mddule}</cyan>.<cyan>{function}</cyan>" # 模块名.方法名
":<cyan>{line}</cyan> |" #行号
"<level>{level}</level>:" #级
"<level>{message}</level>",#日志内容
)
def get_logger(self):
return self.logger
log = MyLogger().get_logger()
if __name__ == '__main__':
print('str.pdf'['str.pdf'.rindex('.'):])
def test():
try:
print(3/0)
except ZeroDivisionError as e:
log.exception(e)创建本地大模型工具类
第一步：引用配置文件.env中配置的本地大模型，我用的是千问的两个版本（qwen3:8b和qwen3-vl:2b）。
项目名称
|
src
|
agent
|
env_utils.py
import os
from dotenv import load_dotenv
load_dotenv(override=True)
OLLAMA_MODEL=os.getenv('OLLAMA_MODEL')
QWEN38B_MODEL=os.getenv('QWEN38B_MODEL')第二步：创建大模型工具类并进行测试。
from langchain_ollama import ChatOllama
from agent import env_utils
llm = ChatOllama(
model=env_utils.QWEN38B_MODEL,
temperature=0.7
)
llm_mini = ChatOllama(
model=env_utils.OLLAMA_MODEL,
temperature=0.7
)
if __name__ == '__main__':
resp=llm.invoke("请告诉我荒野求生怎么生火。")
print(resp)
print(type(resp))
创建并测试LangChain的工具模块
分析：Agent调用各种工具，不同工具分别调用我们刚才创建好的MySQL的四个方法。
第一步：创建并测试LangChain的工具模块。
项目名称
|
src
|
agent
|
tools
|
text_to_sql_tools.pyimport asyncio
from typing import Optional, List
from langchain_core.tools import BaseTool
from pydantic import Field, create_model
from utils.db_utils import MySQLDatabaseManager
from utils.log_utils import log
class ListTablesTool(BaseTool):
"""列出数据库中的所有表及其描述信息"""
name: str = "sql_db_list_tables"
description: str = "列出MySQL数据库中的所有表名及其描述信息。"
db_manager: MySQLDatabaseManager
def _run(self) -> str:
try:
tables_info = self.db_manager.get_tables_with_comments()
result = f"数据库总共有{len(tables_info)}个表：\n\n"
for i, table_info in enumerate(tables_info):
table_name = table_info["table_name"]
table_comment = table_info["table_comment"]
description_display = table_comment if table_comment and not table_comment.isspace() else "(暂无描述)"
result += f"{i + 1}. 表名：{table_name}\n 描述：{description_display}\n\n"
return result
except Exception as e:
log.exception(e)
return f"列出表时出错：{str(e)}"
async def _arun(self) -> str:
return await asyncio.to_thread(self._run)
class TableSchemaTool(BaseTool):
"""获取表模式信息"""
name: str = "sql_db_schema"
description: str = "获取表结构。"
db_manager: MySQLDatabaseManager
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.args_schema = create_model(
"TableSchemaToolArgs",
table_names=(Optional[List[str]], Field(..., description="表名列表"))
)
def _run(self, table_names: Optional[List[str]] = None) -> str:
try:
schema_info = self.db_manager.get_table_schema(table_names)
return schema_info if schema_info else "未找到匹配的表。"
except Exception as e:
log.exception(e)
return f"返回表结构时出错：{str(e)}"
async def _arun(self, table_names: Optional[List[str]] = None) -> str:
return await asyncio.to_thread(self._run, table_names)
class SQLQueryTool(BaseTool):
"""执行SQL查询"""
name: str = "sql_db_query"
description: str = "在MySQL数据库上执行安全的SELECT查询并返回结果。"
db_manager: MySQLDatabaseManager
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.args_schema = create_model(
"SQLQueryToolArgs",
query=(str, Field(..., description="有效的SQL查询语句"))
)
#同步兼容
def _run(self, query: str) -> str:
try:
result = self.db_manager.execute_sql(query)
return str(result)
except Exception as e:
log.exception(e)
return f"执行查询时出错：{str(e)}"
async def _arun(self, query: str) -> str:
try:
result = await asyncio.to_thread(self.db_manager.execute_sql, query)
return str(result)
except Exception as e:
log.exception(e)
return f"执行查询时出错：{str(e)}"
class SQLQueryCheckerTool(BaseTool):
"""检查SQL查询语法"""
name: str = "sql_db_query_checker"
description: str = "检查SQL语法"
db_manager: MySQLDatabaseManager
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.args_schema = create_model(
"SQLQueryCheckerToolArgs",
query=(str, Field(..., description="需要检查的SQL"))
)
def _run(self, query: str) -> str:
try:
return self.db_manager.preprocess_sql(query)
except Exception as e:
log.exception(e)
return f"执行检查时出错：{str(e)}"
async def _arun(self, query: str) -> str:
try:
return await asyncio.to_thread(self.db_manager.preprocess_sql, query)
except Exception as e:
log.exception(e)
return f"执行检查时出错：{str(e)}"
if __name__ == '__main__':
# 配置数据库连接信息
host = "localhost"
port = 3306
username = "root"
password = "123456"
database = "nl_to_sql_db"
connection_string = f"mysql+pymysql://{username}:{password}@{host}/{database}?charset=utf8"
manager = MySQLDatabaseManager(connection_string)
#测试工具
tool = ListTablesTool(db_manager=manager)
print(tool.invoke({}))
tool = TableSchemaTool(db_manager=manager)
print(tool.invoke({"table_names": ['student_info']}))
#
tool = SQLQueryCheckerTool(db_manager=manager)
print(tool.invoke({"query": "SELECT count(1) FROM student_info"}))
#
tool = SQLQueryTool(db_manager=manager)
print(tool.invoke({"query": "SELECT id FROM student_info"}))
创建Agent
第一步：创建Agent。
项目名称
|
src
|
agent
|
text_to_sql_agent.pyfrom typing import List
from langchain.agents import create_agent
from langchain_core.tools import BaseTool
from agent.my_llm import llm
from agent.tools.text_to_sql_tools import ListTablesTool, TableSchemaTool, SQLQueryCheckerTool, SQLQueryTool
from utils.db_utils import MySQLDatabaseManager
def get_tools(host: str, port: int, username: str, password: str, database: str) -> List[BaseTool]:
# 构建连接字符串
connection_string = f"mysql+pymysql://{username}:{password}@{host}/{database}?charset=utf8"
manager = MySQLDatabaseManager(connection_string)
return [
ListTablesTool(db_manager=manager),
TableSchemaTool(db_manager=manager),
SQLQueryCheckerTool(db_manager=manager),
SQLQueryTool(db_manager=manager),
]
# 配置数据库连接信息
host = "localhost"
port = 3306
username = "root"
password = "123456"
database = "nl_to_sql_db"
tools = get_tools(host, port, username, password, database)
system_prompt = """
你是一个专门用于与 SQL 数据库交互的 AI 智能体，负责将自然语言问题转换为安全、准确的 SQL 查询并返回结果。
你必须严格遵循以下工作流程：
先查看数据库中可用的表及其结构信息。
根据用户问题生成符合 {dialect} 语法的 SQL 查询语句。
查询语句仅可使用只读 SELECT 操作。
除非用户明确指定数量，否则结果最多返回 {top_k} 条。
仅选择必要字段，禁止使用 SELECT *。
可通过排序提升结果相关性。
在执行前必须检查 SQL 语法与安全性。
若执行失败，需自动修正 SQL 并重试。
基于最终查询结果生成答案，而非凭空推测。
""".format(
dialect="MySQL",
top_k=5
)
agent = create_agent(
llm,
tools=tools,
system_prompt=system_prompt
)第二步：上述代码中create_agent方法返回Agent，我用agent变量接收。所以我们要在langgraph.json文件中配置相关的Agent地址。这与我们之前的pyproject.toml配置文件也是对应的。
{
"dependencies": ["."],
"graphs": {
"agent" :"./src/agent/text_to_sql_agent.py:agent"
},
"env": ".env"
}调试
第一步：启动本地的Agent服务器。启动成功后会自动打开LangSmith Studio图形界面，在这里可以进行提问。注意：如果多次启动后无法打开LangSmith Studio图形界面或有界面但Studio没有流程图，可以尝试重启开发工具再运行项目。如果还不行，则重启电脑。
langgraph dev
总结
刚才我们搭建的NL2SQL还有很多不足之处，比如SQL准确率低、无法给出分段SQL、没有自定义前端页面等问题。因此该系统只能算一个案例或基本架构，感兴趣的同学可以去下载源代码进行优化升级。
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
24
0- 0
已为社区贡献8条内容
所有评论(0)