面向您的 DB 的 SQL 助手
向模型提供模式,让它编写 SQL,在沙盒化的 DB 上运行,并解释结果。
面向您的 DB 的 SQL 助手 是 CoddyKit 上的免费 AI Agents 课时。 这是第 4 节课,共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分,你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。
本课时的部分内容尚未翻译,以英文显示。
项目目标
构建一个代理,将自然语言问题转换为 SQL,在沙盒数据库中执行,并解释结果。
这是最实用的“与我的数据库对话”代理。
架构
- 用户:“本月有多少活跃用户?”
- 代理调用 list_tables 和 describe_table 来了解架构
- 代理使用生成的查询调用 run_sql
- 代理用自然语言解释结果
Step 1: Schema Tools
import psycopg
conn = psycopg.connect(DATABASE_URL)
def list_tables():
with conn.cursor() as cur:
cur.execute("SELECT table_name FROM information_schema.tables WHERE table_schema='public'")
return [r[0] for r in cur.fetchall()]
def describe_table(name):
with conn.cursor() as cur:
cur.execute('''
SELECT column_name, data_type FROM information_schema.columns
WHERE table_name = %s
''', (name,))
return cur.fetchall()步骤 2:安全的 run_sql 工具
您必须限制代理能够执行的操作——只读、查询超时和行数限制:
import re
def run_sql(query: str, limit: int = 100):
q = query.strip().rstrip(';').lower()
if not q.startswith('select'):
return {'error': 'Only SELECT statements are allowed.'}
if any(bad in q for bad in [' drop ', ' delete ', ' update ', ' insert ', ' alter ', ' truncate ']):
return {'error': 'Statement contains a disallowed keyword.'}
with conn.cursor() as cur:
cur.execute(f'SET statement_timeout = 5000') # 5 seconds
cur.execute(f'SELECT * FROM ({query}) sub LIMIT {limit}')
cols = [c.name for c in cur.description]
rows = cur.fetchall()
return {'columns': cols, 'rows': rows}Tool Definitions
tools = [
{'type': 'function', 'function': {'name': 'list_tables', 'description': 'List tables in the database', 'parameters': {'type': 'object', 'properties': {}}}},
{'type': 'function', 'function': {'name': 'describe_table', 'description': 'Get columns of a table', 'parameters': {'type': 'object', 'properties': {'name': {'type': 'string'}}, 'required': ['name']}}},
{'type': 'function', 'function': {'name': 'run_sql', 'description': 'Execute a SELECT query (read-only, max 100 rows, 5s timeout)', 'parameters': {'type': 'object', 'properties': {'query': {'type': 'string'}}, 'required': ['query']}}}
]
import json
print(json.dumps(tools, indent=2))
System Prompt
system = '''
You are a SQL analyst assistant for a Postgres database.
First use list_tables and describe_table to learn the schema.
Then write a single SELECT query to answer the user.
Never modify data.
After receiving results, explain them in plain language.
'''
print(system.strip())
在系统提示词中传入架构
为了获得更低的延迟,请在启动时获取一次架构,并将其放入系统提示词中——这样可以节省工具往返:
schema = ''
for t in list_tables():
cols = describe_table(t)
schema += f'{t}: {cols}\n'
system = system + f'\nSchema:\n{schema}'只读数据库用户
即使进行了代码检查,也请创建一个只读数据库用户,并且只授予其分析架构的权限。这是纵深防御。
PII 脱敏
某些列(电子邮件、电话号码)绝不能泄露。返回结果前请先对其进行脱敏:
PII_COLS = {'email', 'phone'}
for row in rows:
for i, col in enumerate(cols):
if col in PII_COLS:
row[i] = '[REDACTED]'查询成本估算
使用 EXPLAIN 估算查询成本;拒绝成本超过阈值的查询(防止在大型表上意外执行全表扫描)。
示例对话
用户:“上个月消费额最高的 5 位客户”
代理:
- list_tables → [users, orders, ...]
- describe_table(orders) → [id, user_id, total, created_at]
- run_sql("SELECT user_id, SUM(total) ...")
- 返回:“Alice($1240)、Bob($910)、...”
输出图表
为了提供更丰富的用户体验,请添加一个 render_chart 工具,接收列和行,并返回图表图片 URL。代理可以在查询后调用它。
优雅地处理失败
SQL 错误很常见。请原样返回 Postgres 错误消息——模型看到错误后,非常擅长修复自身生成的错误 SQL。
审计每个查询
记录用户、自然语言问题、生成的 SQL 以及结果数量。对于数据库访问代理,审计是不可或缺的。
为什么限制为 SELECT?
为什么要硬编码限制代理,使其只能运行 SELECT 语句?
回顾
SQL 代理可以立即产生价值。请谨慎构建——只读用户、语句超时、行数限制、查询类型白名单和审计日志。
常见问题解答
「面向您的 DB 的 SQL 助手」课时是免费的吗?
是的 — 「面向您的 DB 的 SQL 助手」的完整文本可在网页上免费阅读。要进行交互式练习(内置代码编辑器和全天候 AI 导师)并解锁 AI Agents 课程的其余内容,请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。
「面向您的 DB 的 SQL 助手」这节课中我会学到什么?
向模型提供模式,让它编写 SQL,在沙盒化的 DB 上运行,并解释结果。 你通过在浏览器中直接运行的动手代码来练习 AI Agents,全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Agents 需要有经验吗?
无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者,你可以从这里开始或从头开始,按照自己的节奏学习。 这是第 4 节课,共 4 节。
「面向您的 DB 的 SQL 助手」课时需要多长时间?
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动,所以你能稳步进步,并在网页和应用中从离开的地方继续。
我能在这节 AI Agents 课中编写并运行代码吗?
能。每节 AI Agents 课都包含内置代码编辑器,你可以在浏览器中直接编写并运行真实代码,并获得即时 AI 反馈 — 无需本地设置。
此课程中的所有课时
- 基于您的文档构建问答机器人
- 代码讲解智能体
- 网页浏览研究智能体
- 面向您的 DB 的 SQL 助手