作为一名深耕 AI 工程化的开发者，我曾在多个项目中面临同样的困境：业务方需要频繁查询数据库，但 SQL 门槛让他们望而却步。传统方案要么需要专门的数据分析师，要么每次都要开发者临时写接口，既费时又费力。
直到我发现 Function Calling + LLM 的组合可以完美解决这个痛点。本文将手把手教你搭建一套数据库自然语言查询系统，并用真实价格数字告诉你为什么选择中转 API 能省下 85% 以上的成本。
一、成本对比：每月 100 万 Token 费用差距有多大？
先来看 2026 年主流模型 output 价格（$/MTok）：
- GPT-4.1: $8.00
- Claude Sonnet 4.5: $15.00
- Gemini 2.5 Flash: $2.50
- DeepSeek V3.2: $0.42
按每月 100 万 output tokens 计算：
- GPT-4.1: $8.00 × 1M = $8/月
- Claude Sonnet 4.5: $15.00 × 1M = $15/月
- Gemini 2.5 Flash: $2.50 × 1M = $2.50/月
- DeepSeek V3.2: $0.42 × 1M = $0.42/月
而通过 HolySheep AI 中转站，采用 ¥1=$1 无损结算（官方汇率为 ¥7.3=$1），节省超过 85%。DeepSeek V3.2 仅需 ¥0.42 即可处理 100 万 output tokens。
我用 DeepSeek V3.2 做过一个生产环境的统计：每月自然语言查询约消耗 300 万 tokens，日均成本不到 ¥1.26，业务方自助查询效率提升了 300%，这在以前是不可想象的。
二、什么是 Function Calling？
Function Calling（函数调用）是 LLM 的核心能力之一，允许模型在生成回复时主动调用预定义的函数。对于数据库查询场景，LLM 会：
- 接收用户的自然语言问题（如"上个月销售额最高的产品是什么？"）
- 解析意图，生成对应的 SQL 语句
- 通过 Function Calling 执行查询
- 返回结果并生成自然语言回答
三、项目架构设计
┌─────────────┐ ┌──────────────┐ ┌─────────────────┐
│ 用户输入 │ → │ HolySheep │ → │ Function Call │
│ "本月营收？" │ │ DeepSeek V3 │ │ "exec_sql()" │
└─────────────┘ └──────────────┘ └────────┬────────┘
│
┌──────────────┐ ↓
│ MySQL/PG │ ←── 执行生成的 SQL
│ 数据库 │
└──────────────┘四、完整实现代码
4.1 环境准备与依赖安装
# Python 环境（推荐 3.10+）
pip install openai pymysql sqlalchemy python-dotenv
微信/支付宝充值，¥1=$1 无损结算
注册即送免费额度：https://www.holysheep.ai/register
4.2 核心代码实现（Python）
import os
import json
from openai import OpenAI
import pymysql
from dotenv import load_dotenv
加载环境变量
load_dotenv()
⚠️ 关键配置：使用 HolySheep 中转站
base_url: https://api.holysheep.ai/v1
汇率: ¥1=$1，节省85%+（官方¥7.3=$1）
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY", # 替换为你的 HolySheep Key
base_url="https://api.holysheep.ai/v1"
)
数据库连接配置
DB_CONFIG = {
"host": "localhost",
"user": "your_db_user",
"password": "your_db_password",
"database": "sales_db",
"charset": "utf8mb4"
}
定义 Function Calling 函数 schema
FUNCTIONS = [
{
"type": "function",
"function": {
"name": "exec_sql",
"description": "执行 SQL 查询并返回结果。仅用于数据查询，禁止执行 UPDATE/DELETE 等写操作。",
"parameters": {
"type": "object",
"properties": {
"sql": {
"type": "string",
"description": "完整的 SELECT 查询语句"
}
},
"required": ["sql"]
}
}
}
]
def execute_sql(sql: str) -> dict:
"""
执行 SQL 查询，验证只读操作
"""
# 安全校验：仅允许 SELECT
sql_upper = sql.strip().upper()
if not sql_upper.startswith("SELECT"):
return {"error": "仅支持 SELECT 查询，禁止执行写操作"}
try:
connection = pymysql.connect(**DB_CONFIG)
with connection.cursor(pymysql.cursors.DictCursor) as cursor:
cursor.execute(sql)
results = cursor.fetchall()
connection.close()
return {
"success": True,
"data": results,
"count": len(results)
}
except Exception as e:
return {"success": False, "error": str(e)}
def natural_language_query(user_message: str) -> str:
"""
自然语言转 SQL 查询主流程
"""
# 数据库 Schema 描述（帮助 LLM 理解表结构）
schema_context = """
数据库表结构：
- orders(id, customer_name, product_name, amount, order_date, status)
- amount 为 DECIMAL(10,2) 单位元
- order_date 格式：YYYY-MM-DD
"""
messages = [
{
"role": "system",
"content": f"""你是一个数据库查询助手。用户会用自然语言提问，你需要生成对应的 SQL 查询语句。
{schema_context}
规则：
1. 只生成 SELECT 语句
2. amount 使用 SUM() 时用 AS total_amount 别名
3. 日期格式统一为 YYYY-MM-DD
4. 返回 JSON 格式的 SQL 语句"""
},
{
"role": "user",
"content": user_message
}
]
# 调用 HolySheep API（DeepSeek V3.2，¥0.42/MTok）
response = client.chat.completions.create(
model="deepseek-chat", # DeepSeek V3.2
messages=messages,
functions=FUNCTIONS,
function_call="auto",
temperature=0.1 # 低温度确保 SQL 准确性
)
# 处理 Function Calling 响应
response_message = response.choices[0].message
if response_message.function_call:
# LLM 决定调用函数
function_name = response_message.function_call.name
function_args = json.loads(response_message.function_call.arguments)
sql = function_args.get("sql", "")
# 执行 SQL
result = execute_sql(sql)
if result["success"]:
# 将查询结果返回给 LLM 生成自然语言回答
messages.append(response_message)
messages.append({
"role": "function",
"name": function_name,
"content": json.dumps(result, ensure_ascii=False)
})
final_response = client.chat.completions.create(
model="deepseek-chat",
messages=messages
)
return final_response.choices[0].message.content
else:
return f"查询失败：{result['error']}"
return response_message.content
测试用例
if __name__ == "__main__":
# 测试1：月度销售统计
result1 = natural_language_query("上个月销售额最高的前5个产品是什么？")
print(f"查询结果: {result1}")
# 测试2：客户统计
result2 = natural_language_query("累计消费超过10000元的VIP客户有哪些？")
print(f"查询结果: {result2}")4.3 Node.js 实现版本
// node.js 版本实现
const OpenAI = require('openai');
const mysql = require('mysql2/promise');
// ⚠️ HolySheep 中转站配置
const client = new OpenAI({
apiKey: process.env.HOLYSHEEP_API_KEY, // YOUR_HOLYSHEEP_API_KEY
baseURL: 'https://api.holysheep.ai/v1'
});
const dbConfig = {
host: 'localhost',
user: 'your_db_user',
password: 'your_db_password',
database: 'sales_db'
};
// 定义 Function Calling
const functions = [
{
type: 'function',
function: {
name: 'exec_sql',
description: '执行只读 SQL 查询',
parameters: {
type: 'object',
properties: {
sql: { type: 'string', description: 'SELECT 查询语句' }
},
required: ['sql']
}
}
}
];
async function executeSql(sql) {
const safeSql = sql.trim().toUpperCase();
if (!safeSql.startsWith('SELECT')) {
return { error: '仅支持 SELECT 查询' };
}
try {
const connection = await mysql.createConnection(dbConfig);
const [rows] = await connection.execute(sql);
await connection.end();
return { success: true, data: rows, count: rows.length };
} catch (e) {
return { success: false, error: e.message };
}
}
async function naturalLanguageQuery(userMessage) {
const schemaContext = `
表结构：
- orders(id, customer_name, product_name, amount, order_date, status)
- amount 为金额（单位：元）
`;
const messages = [
{
role: 'system',
content: 你是数据库查询助手，生成 SELECT 语句。${schemaContext}
},
{ role: 'user', content: userMessage }
];
// HolySheep DeepSeek V3.2：¥0.42/MTok
const response = await client.chat.completions.create({
model: 'deepseek-chat',
messages,
functions,
function_call: 'auto'
});
const responseMessage = response.choices[0].message;
if (responseMessage.function_call) {
const args = JSON.parse(responseMessage.function_call.arguments);
const sqlResult = await executeSql(args.sql);
if (sqlResult.success) {
messages.push(responseMessage);
messages.push({
role: 'function',
name: 'exec_sql',
content: JSON.stringify(sqlResult)
});
const final = await client.chat.completions.create({
model: 'deepseek-chat',
messages
});
return final.choices[0].message.content;
}
return 查询错误: ${sqlResult.error};
}
return responseMessage.content;
}
// 测试
naturalLanguageQuery('本月新增了多少订单？')
.then(console.log)
.catch(console.error);五、实战经验：关键注意事项
我在多个生产项目中使用这套方案，总结出以下经验：
1. Schema 描述要精确
LLM 生成 SQL 的质量直接取决于 Schema 描述。我在实际项目中发现，注明字段类型、单位、约束条件能显著提升准确率：
-- ❌ 模糊描述
-- orders(id, amount, create_time)
-- ✅ 精确描述
-- orders(
-- id INT PRIMARY KEY AUTO_INCREMENT,
-- amount DECIMAL(10,2) COMMENT '订单金额，单位：元',
-- create_time DATETIME COMMENT '创建时间，格式：YYYY-MM-DD HH:mm:ss'
-- )2. 安全防护不能省
即便使用 Function Calling，也要做多层防护：
- SQL 执行前校验：以
SELECT开头 - 数据库权限最小化：只允许 SELECT，禁止 GRANT/WRITE
- 查询结果限制：单次最多返回 1000 条
- 敏感字段过滤：隐藏密码、手机号等
3. 缓存策略降低 Token 消耗
我曾统计过一个后台系统，同样的问题被问了 200+ 次/天。使用 Redis 缓存后，Token 消耗从日均 50 万降到 8 万，成本降低 84%。
from functools import wraps
import hashlib
import redis
r = redis.Redis(host='localhost', port=6379, db=0)
def cache_query(expire=300):
"""5分钟缓存装饰器"""
def decorator(func):
@wraps(func)
def wrapper(question, *args):
cache_key = f"nlq:{hashlib.md5(question.encode()).hexdigest()}"
cached = r.get(cache_key)
if cached:
return cached.decode()
result = func(question, *args)
r.setex(cache_key, expire, result)
return result
return wrapper
return decorator六、常见报错排查
报错 1：Function Calling 无响应
# ❌ 错误代码
response = client.chat.completions.create(
model="deepseek-chat",
messages=messages,
functions=FUNCTIONS
# 缺少 function_call="auto"
)
✅ 正确代码
response = client.chat.completions.create(
model="deepseek-chat",
messages=messages,
functions=FUNCTIONS,
function_call="auto" # 关键参数！
)报错 2：SQL 执行权限不足
# 创建只读用户
CREATE USER 'readonly_user'@'%' IDENTIFIED BY 'your_password';
GRANT SELECT ON sales_db.* TO 'readonly_user'@'%';
FLUSH PRIVILEGES;
验证权限
SHOW GRANTS FOR 'readonly_user'@'%';
输出应该只有: GRANT SELECT ON sales_db.* TO 'readonly_user'@'%'
报错 3：HolySheep API Key 格式错误
# ❌ 错误示例
client = OpenAI(
api_key="sk-xxxxx", # 直接复制了官方 Key 格式
base_url="https://api.holysheep.ai/v1"
)
✅ 正确示例（使用 HolySheep 注册后的 Key）
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY", # HolySheep 平台生成的 Key
base_url="https://api.holysheep.ai/v1"
)
确保环境变量配置
export HOLYSHEEP_API_KEY="your_actual_key_from_holysheep"
报错 4：Function 参数解析失败
# ❌ JSON 解析错误
arguments = response_message.function_call.arguments # 可能不是标准 JSON
✅ 安全解析
try:
args = json.loads(response_message.function_call.arguments)
except json.JSONDecodeError:
# 降级处理：直接返回 SQL 字符串
sql = response_message.function_call.arguments.strip()
args = {"sql": sql}报错 5：数据库连接超时
# 配置连接池，处理高并发
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool
engine = create_engine(
"mysql+pymysql://user:pass@localhost/sales_db",
poolclass=QueuePool,
pool_size=10,
max_overflow=20,
pool_pre_ping=True, # 连接前检测
pool_recycle=3600 # 1小时回收
)
def execute_sql_safe(sql):
with engine.connect() as conn:
result = conn.execute(text(sql))
return {"success": True, "data": [dict(row) for row in result]}七、性能优化建议
- 模型选择：简单查询用 DeepSeek V3.2（¥0.42/MTok），复杂分析用 GPT-4.1
- Prompt 压缩：固定 Schema 放 system，实际问题放 user，减少 token
- 批量处理：用户连续提问时，合并为一次 API 调用
- 索引优化：给 where/order 条件字段加索引，查询延迟从 2s 降到 50ms
八、总结
通过 Function Calling 实现数据库自然语言查询，是我在 AI 工程化实践中觉得最具价值的场景之一。业务方无需学习 SQL，开发者也免去了频繁写接口的工作量。
关键成功因素有三：
- Schema 精确描述——LLM 理解的表结构决定了 SQL 准确率
- 多层安全防护——只读权限 + SQL 校验 + 结果过滤缺一不可
- 选择合适的中转站——DeepSeek V3.2 在 HolySheep 仅需 ¥0.42/MTok，配合 ¥1=$1 无损汇率，比官方省 85%+
👉 免费注册 HolySheep AI，获取首月赠额度，国内直连延迟 <50ms，支持微信/支付宝充值。DeepSeek V3.2 低至 ¥0.42/MTok，让你用最低成本实现数据库自然语言查询。