Agent开发初体验:自然语言转 SQL 智能体，揭开 Agent 的底层真相
·
当下 AI Agent 概念火热，大量框架、多轮规划、长期记忆、工具链方案层出不穷，很容易让人误以为 Agent 是智能核心。但我的观点很直接：AI Agent 开发本质是文字游戏，Agent 本身并不 “聪明”，真正强大的是背后的基座大模型。
Agent 更像一个精密的 “话术调度器”：负责把用户需求包装成合适 Prompt、调用大模型产出结构化内容、对接外部工具、校验输出、整理结果。本文搭建一个轻量 SQL 翻译智能体：用户用日常语言提查询需求，基座大模型负责理解业务语义、输出可执行 SQL，Agent 承担串联调度工作，带你剥离光环，看懂 Agent 最朴素的实现逻辑。
1. 项目依赖
pip install 'langchain[openai]' dotenv
pip install sqlalchemy pymysql loguru
pip install --upgrade "langgraph-cli[inmem]"
2. 测试的建表脚本
-- =============================================
-- RuoYi RBAC五张权限表 修复无冲突完整版
-- MySQL5.7 / MySQL8.0
-- =============================================
DROP TABLE IF EXISTS sys_role_menu;
DROP TABLE IF EXISTS sys_user_role;
DROP TABLE IF EXISTS sys_menu;
DROP TABLE IF EXISTS sys_role;
DROP TABLE IF EXISTS sys_user;
-- ----------------------------
-- 用户表 sys_user
-- ----------------------------
CREATE TABLE sys_user (
user_id BIGINT NOT NULL AUTO_INCREMENT COMMENT '用户ID',
user_name VARCHAR(50) NOT NULL COMMENT '登录账号',
nick_name VARCHAR(50) NOT NULL COMMENT '用户昵称',
password VARCHAR(100) NOT NULL COMMENT 'BCrypt加密密码',
status CHAR(1) DEFAULT '0' COMMENT '账号状态 0正常 1停用',
create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
PRIMARY KEY (user_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户信息表';
-- ----------------------------
-- 角色表 sys_role
-- ----------------------------
CREATE TABLE sys_role (
role_id BIGINT NOT NULL AUTO_INCREMENT COMMENT '角色ID',
role_name VARCHAR(50) NOT NULL COMMENT '角色名称',
role_key VARCHAR(50) NOT NULL COMMENT '角色编码',
sort INT DEFAULT 0 COMMENT '排序',
status CHAR(1) DEFAULT '0' COMMENT '状态 0正常 1停用',
create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
PRIMARY KEY (role_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='角色信息表';
-- ----------------------------
-- 用户角色关联 sys_user_role 多对多
-- ----------------------------
CREATE TABLE sys_user_role (
user_id BIGINT NOT NULL COMMENT '用户ID',
role_id BIGINT NOT NULL COMMENT '角色ID',
PRIMARY KEY (user_id, role_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户角色关联表';
-- ----------------------------
-- 菜单权限 sys_menu
-- menu_type：C目录 M菜单 F按钮
-- ----------------------------
CREATE TABLE sys_menu (
menu_id BIGINT NOT NULL AUTO_INCREMENT COMMENT '菜单ID',
parent_id BIGINT DEFAULT 0 COMMENT '父菜单ID，0为顶级',
menu_name VARCHAR(50) NOT NULL COMMENT '菜单名称',
menu_type CHAR(1) NOT NULL COMMENT 'C目录 M菜单 F按钮',
path VARCHAR(200) DEFAULT '' COMMENT '前端路由地址',
component VARCHAR(255) DEFAULT '' COMMENT 'Vue组件路径',
perms VARCHAR(100) DEFAULT '' COMMENT '后端权限标识',
icon VARCHAR(100) DEFAULT '#' COMMENT '菜单图标',
sort INT DEFAULT 0 COMMENT '排序',
status CHAR(1) DEFAULT '0' COMMENT '状态 0正常 1停用',
create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
PRIMARY KEY (menu_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='菜单权限表';
-- ----------------------------
-- 角色菜单关联 sys_role_menu 多对多
-- ----------------------------
CREATE TABLE sys_role_menu (
role_id BIGINT NOT NULL COMMENT '角色ID',
menu_id BIGINT NOT NULL COMMENT '菜单ID',
PRIMARY KEY (role_id, menu_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='角色菜单关联表';
-- ===================== 测试数据开始 =====================
INSERT INTO sys_user(user_name, nick_name, password, status) VALUES
('admin','超级管理员','$2a$10$7JB720yubVSZvUI0rEqK/.VqGOZTH.ulu33dHOiBE8ByOhJIrdAu2','0'),
('test','普通业务用户','$2a$10$N.zmdr9k7uOCQb376NoUnuTJ8iAt6Z5EHsM8lE9lBOsl7iKTVKIUi','0'),
('zhangsan','运维管理员','$2a$10$X7H1GpQnM0O2wF5dU7yR8eK9aS3bD4cF5gH6jI7kL8mN9oP0','0'),
('lisi','访客只读账号','$2a$10$A8B9C0D1E2F3G4H5I6J7K8L9M0N1O2P3Q4R5S6T7U8V9W0X1Y2','0'),
('wangwu','多角色测试账号','$2a$10$Z9Y8X7W6V5U4T3S2R1Q0P9O8N7M6L5K4J3I2H1G0F','0');
INSERT INTO sys_role(role_name, role_key, sort, status) VALUES
('超级管理员','admin',1,'0'),
('普通业务角色','common',2,'0'),
('运维角色','operator',3,'0'),
('访客角色','guest',4,'0');
INSERT INTO sys_user_role(user_id, role_id) VALUES
(1, 1),
(2, 2),
(3, 3),
(4, 4),
(5, 2),
(5, 4);
-- menu_id自增1~21
INSERT INTO sys_menu(parent_id, menu_name, menu_type, path, component, perms, icon, sort) VALUES
(0, '系统管理', 'C', 'system', '', '', 'system', 1),
(1, '用户管理', 'M', 'user', 'system/user/index', '', 'user', 1),
(2, '用户查询', 'F', '', '', 'system:user:list', '#',1),
(2, '用户新增', 'F', '', '', 'system:user:add', '#',2),
(2, '用户修改', 'F', '', '', 'system:user:edit', '#',3),
(2, '用户删除', 'F', '', '', 'system:user:remove', '#',4),
(1, '角色管理', 'M', 'role', 'system/role/index', '', 'role', 2),
(7, '角色查询', 'F', '', '', 'system:role:list', '#',1),
(7, '角色新增', 'F', '', '', 'system:role:add', '#',2),
(7, '角色修改', 'F', '', '', 'system:role:edit', '#',3),
(7, '角色删除', 'F', '', '', 'system:role:remove', '#',4),
(0, '工具模块', 'C', 'tool', '', '', 'tool', 2),
(12, '代码生成', 'M', 'gen', 'tool/gen/index', '', 'code',1),
(13, '生成查询', 'F', '', '', 'tool:gen:list', '#',1),
(13, '预览代码', 'F', '', '', 'tool:gen:preview', '#',2),
(13, '下载代码', 'F', '', '', 'tool:gen:download', '#',3),
(12, '定时任务', 'M', 'job', 'tool/job/index', '', 'time',2),
(17, '任务查询', 'F', '', '', 'tool:job:list', '#',1),
(17, '任务新增', 'F', '', '', 'tool:job:add', '#',2),
(17, '任务修改', 'F', '', '', 'tool:job:edit', '#',3),
(17, '任务删除', 'F', '', '', 'tool:job:remove', '#',4);
-- =================修复后的sys_role_menu，无重复主键=================
INSERT INTO sys_role_menu(role_id, menu_id) VALUES
-- role1 admin超级管理员，全部权限
(1,1),(1,2),(1,3),(1,4),(1,5),(1,6),
(1,7),(1,8),(1,9),(1,10),(1,11),
(1,12),(1,13),(1,14),(1,15),(1,16),
(1,17),(1,18),(1,19),(1,20),(1,21),
-- role2 common普通业务角色：系统管理、用户管理、用户查询、角色管理、角色查询
(2,1),(2,2),(2,3),(2,7),(2,8),
-- role3 operator运维，全部权限
(3,1),(3,2),(3,3),(3,4),(3,5),(3,6),
(3,7),(3,8),(3,9),(3,10),(3,11),
(3,12),(3,13),(3,14),(3,15),(3,16),
(3,17),(3,18),(3,19),(3,20),(3,21),
-- role4 guest访客角色：工具模块、代码生成、定时任务页面
(4,12),(4,13),(4,17);
-- ====================验证SQL====================
-- 查询用户id=5(wangwu)的所有权限标识
SELECT DISTINCT m.perms
FROM sys_user u
INNER JOIN sys_user_role ur ON u.user_id = ur.user_id
INNER JOIN sys_role_menu rm ON ur.role_id = rm.role_id
INNER JOIN sys_menu m ON rm.menu_id = m.menu_id
WHERE u.user_id = 5 AND m.perms <> '';
-- 查询侧边栏菜单（过滤按钮F）
SELECT menu_id,parent_id,menu_name,menu_type,path,component,icon,sort
FROM sys_menu
WHERE menu_type IN ('C','M')
ORDER BY parent_id, sort;
3. 工具的开发
from typing import Any, Optional, List
from langchain_core.tools import BaseTool
from langchain_core.utils.pydantic import create_model
from pydantic import Field
from sqlalchemy import text
from sqlalchemy.ext.asyncio import result
from agent.utils import LogUtil
from agent.utils.db_utils import MySQLDataBaseManager
class ListTablesTool(BaseTool):
name:str = "ListTablesTool"
description:str = "列出MySQL数据库中所有的表名及描述信息"
db_manager: MySQLDataBaseManager
def _run(self) -> str:
try:
tables_info = self.db_manager.get_tables_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']
if not table_comment:
table_comment = "暂无描述"
result += f"{i+1}.表名:{table_name}\n"
result += f" 描述:{table_comment}\n"
return result
except Exception as e:
LogUtil.exception(e)
return f"列出数据库中所有表时出错{str(e)}"
async def _arun(self) -> str:
return self._run();
class TableSchemaTool(BaseTool):
name:str = "TableSchemaTool"
description:str = "获取MySQL数据库中指定表的详细模式信息,包括定义,主键,外键等.输入应为表名列表,以获取所有表信息."
db_manager: MySQLDataBaseManager
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.args_schema = create_model(
"TableSchemaToolArgs",
tables_names=(Optional[List[str]], Field(default=None, description="需要查询的表名列表，不传则获取全部表"))
)
def _run(self, tables_names:Optional[List[str]]=None) -> str:
"""
返回表结构信息
:param tables:
:return:
"""
try:
schema_info = self.db_manager.get_tables_schema(tables_names)
return schema_info if schema_info else "未找到匹配的表"
except Exception as e:
LogUtil.exception(e)
return f"获取表模式信息出错{str(e)}"
async def _arun(self,tables_names: Optional[List[str]] = None) -> str:
return self._run(tables_names);
class SqlQueryTool(BaseTool):
name:str = "SqlQueryTool"
description:str = "在MySql数据库中执行安全的SELECT查询语句并返回结果.输入应为有效的SELECT类型的SQL语句"
db_manager: MySQLDataBaseManager
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.args_schema = create_model(
"SqlQueryToolArgs",
query=(str, Field(default=None, description="有效的SELECT类型的SQL语句"))
)
def _run(self,query:str) -> str:
if not query:
return "输入SQL语句不能为空"
try:
res = self.db_manager.check_execute_query(query)
return res
except Exception as e:
LogUtil.exception(e)
return f"执行查询语句报错{str(e)}"
async def _arun(self,query:str) -> str:
return self._run(query);
class SqlCheckTool(BaseTool):
name:str = "SqlCheckTool"
description:str = "检查SQL语句的语法是否正确,提供验证反馈.输入应为要验证的SQL查询."
db_manager: MySQLDataBaseManager
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.args_schema = create_model(
"SqlQueryToolArgs",
query=(str, Field(default=None, description="有效的SELECT类型的SQL语句"))
)
def _run(self,query:str) -> str:
if not query:
return "SQL语句不能为空"
try:
res = self.db_manager.validate_query(query)
return res
except Exception as e:
LogUtil.exception(e)
return f"SQL校验报错{str(e)}"
async def _arun(self,query:str) -> str:
return self._run(query);
if __name__ == '__main__':
tool = SqlCheckTool(db_manager=MySQLDataBaseManager("mysql+pymysql://root:123456@127.0.0.1:3306/my_test01?charset=utf8mb4"))
result = tool.invoke({"query":"select * from aa"})
print(f"结果:{result}")
4. 大模型的调用
from langchain_openai import ChatOpenAI
from agent.env_utils import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL
# 大模型
llm = ChatOpenAI(
model="deepseek-v4-pro",
temperature=0,
api_key=DEEPSEEK_API_KEY,
base_url=DEEPSEEK_BASE_URL
)
5. Agent模块开发
from langchain.agents import create_agent
from langchain_core.tools import BaseTool
from agent.llm_deepseek import llm
from agent.tools.text_to_sql_tool import ListTablesTool, TableSchemaTool, SqlQueryTool, SqlCheckTool
from agent.utils.db_utils import MySQLDataBaseManager
def get_tools(db_url:str) -> list[BaseTool]:
db_manager = MySQLDataBaseManager(db_url)
return [
ListTablesTool(db_manager=db_manager),
TableSchemaTool(db_manager=db_manager),
SqlQueryTool(db_manager=db_manager),
SqlCheckTool(db_manager=db_manager)
]
sys_prompt = """你是MySQL文本转SQL助手。
1. 仅生成 SELECT / WITH 只读查询，禁止增删改、建表等任何修改类SQL。
2. 根据用户的自然语言提问生成SQL；不清楚表结构就调用TableSchemaTool获取表和字段，严禁编造表名字段。
3. 生成SQL后调用SqlCheckTool校验语法，报错就修正。
4. 最终输出SQL并返回结果，不要多余说明。"""
tools = get_tools(db_url="mysql+pymysql://root:123456@127.0.0.1:3306/my_test01?charset=utf8mb4")
agent = create_agent(
llm,
tools=tools,
system_prompt=sys_prompt
)
6. 配置文件
{
"dependencies": ["."],
"graphs": {
"agent": "./src/agent/text_to_sql_agent.py:agent"
},
"env": ".env"
}
7. LangSmith验证
启动脚本:langgraph dev
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
5
0- 0
已为社区贡献1条内容
所有评论(0)