大模型 Function Call（函数调用）详解：原理、实践与数据库智能查询 Agent
·
目录
一、什么是 Function Call
1.1 概念定义
Function Call 是 OpenAI 在 2023 年 6 月 13 日随 API 更新一同发布的一项能力，允许开发者在调用 Chat Completions API 时，以 JSON Schema 的形式向模型描述一组可用的函数（工具）。模型会根据用户的输入内容，自主判断是否需要调用某个函数，并在需要时返回一个结构化的 JSON 对象，其中包含要调用的函数名称和参数。
一句话总结：Function Call 让大模型从"只会说话"变成"能干活"——它知道什么时候该调用什么工具，并且能正确地填写参数。
1.2 为什么需要 Function Call
在 Function Call 出现之前，开发者如果想让大模型使用外部工具，通常需要依赖提示词工程（Prompt Engineering），让模型在回复中以特定格式（如特定的 JSON 格式或标记语言）输出调用意图，然后再由程序解析。这种方式存在诸多问题：
| 问题 | 说明 |
|---|---|
| 格式不稳定 | 模型可能输出不符合预期格式的内容，导致解析失败 |
| 参数提取不可靠 | 模型可能遗漏必要参数或生成错误的参数值 |
| 维护成本高 | 每次增减工具都需要重新设计提示词和解析逻辑 |
| 多轮调用困难 | 需要多个工具协作时，流程编排非常复杂 |
Function Call 将这些"非正式"的约定提升为API 层面的标准化能力，由模型经过专门训练来识别函数调用意图并生成规范的 JSON 输出，从根本上提高了可靠性和可维护性。
1.3 支持 Function Call 的模型
目前，Function Call（或等价的工具调用能力）已经成为主流大模型的标配功能：
| 模型/平台 | 支持时间 | 备注 |
|---|---|---|
| OpenAI GPT-4 / GPT-3.5-turbo | 2023 年 6 月 | 首次发布 Function Call |
| OpenAI GPT-4o / GPT-4o-mini | 2024 年 | 支持并行函数调用、更低成本 |
| 百度文心一言（ERNIE） | 2023 年下半年 | 国内首批支持 |
| 智谱 ChatGLM3-6B | 2023 年 10 月 | 开源模型中较早支持 |
| 讯飞星火 3.0 | 2023 年 | 支持 Function Call |
| 阿里通义千问（Qwen） | 2024 年 | 支持 Tool Calling |
| DeepSeek | 2024 年 | 支持 Function Call，兼容 OpenAI 接口 |
| Anthropic Claude | 2024 年 | 称为 Tool Use |
| Google Gemini | 2024 年 | 称为 Function Calling |
二、Function Call 解决了大模型的哪些问题
大语言模型虽然强大，但受限于其工作原理，存在几个根本性的短板。Function Call 的出现正是为了弥补这些不足：
2.1 信息实时性
大模型的训练数据有时间截止点（knowledge cutoff），无法获取训练完成之后的信息。例如，模型不可能知道今天的天气、当前的股价或最新的新闻。
Function Call 的解决方案：通过调用外部 API（如天气 API、新闻 API、金融数据接口），模型可以实时获取最新数据，打破"信息茧房"。
2.2 数据局限性
模型的训练数据虽然庞大，但无法覆盖所有专业领域的详细知识。例如，特定企业的内部数据库、私有知识库、用户个人数据等，都不在训练集之中。
Function Call 的解决方案：模型可以调用企业内部的数据库查询工具、知识库检索系统、RAG 检索接口等，获取特定领域的精确信息。
2.3 功能扩展性
大模型本质上只是"文本生成器"，无法直接执行操作——它不能发邮件、不能操作数据库、不能控制智能家居。
Function Call 的解决方案：通过为模型配备各种"工具函数"，模型可以根据用户意图自动选择并调用合适的工具，实现从"理解意图"到"执行动作"的完整闭环。
2.4 结构化输出
直接让模型输出 JSON 等结构化数据时，格式往往不稳定，可能出现多余的说明文字、格式错误等问题。
Function Call 的解决方案：模型经过专门训练，能够输出严格符合函数参数定义的 JSON 对象，格式稳定可靠。
三、Function Call 工作原理
3.1 无 Function Call 时的工作流
在没有 Function Call 时，大模型应用的交互流程非常简单：
┌────────┐ ┌────────────┐ ┌──────────┐
│ 用户 │──请求──▶│ 应用服务 │──提示词──▶│ GPT 模型 │
│ Client │ │ Chat Server│ │ │
│ │◀─响应──│ │◀──文本──│ │
└────────┘ └────────────┘ └──────────┘
流程：
- 1.用户向应用服务发送请求
- 2.应用服务将用户的请求组装成提示词，发送给 GPT 模型
- 3.模型生成文本响应，返回给应用服务
- 4.应用服务将响应返回给用户
这个流程中，模型只能基于自身知识回答，无法访问外部信息。
3.2 有 Function Call 时的工作流
引入 Function Call 后，交互流程变得更加丰富：
┌────────┐ ┌────────────┐ ┌──────────┐ ┌────────┐
│ 用户 │──请求──▶│ 应用服务 │──提示词+工具定义──▶│ GPT 模型 │ │ 外部工具 │
│ Client │ │ Chat Server│ │ │ │ (函数) │
│ │ │ │◀──函数调用指令──│ │ │ │
│ │ │ │──执行函数，获取结果──▶│ │──────▶│ │
│ │ │ │◀──最终自然语言回答──│ │◀──────│ │
│ │◀─响应──│ │ │ │ │ │
└────────┘ └────────────┘ └──────────┘ └────────┘
详细步骤：
- 1.用户发送请求，应用服务将用户的提示词以及可用的函数定义（JSON Schema 格式）一起发送给模型
- 2.模型分析用户意图，判断是否需要调用函数：
- 如果不需要调用函数，直接返回自然语言回答
- 如果需要调用函数，返回一个函数调用指令（包含函数名和参数的 JSON）
- 3.应用服务接收函数调用指令，在本地执行对应的函数，获取执行结果
- 4.应用服务将函数执行结果返回给模型
- 5.模型整合函数返回的数据，生成最终的自然语言回答
关键点：Function Call 不是模型直接执行函数，而是模型"告诉"应用层它想调用哪个函数、传什么参数，由应用层负责实际执行并将结果回传。
四、Function Call 核心概念详解
4.1 工具定义（Tool Definition）
向模型描述可用函数时，需要使用特定的 JSON Schema 格式。一个完整的工具定义包含以下关键字段：
{
"type": "function",
"function": {
"name": "get_weather",
"description": "根据城市名称获取当前天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，例如：北京、上海"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位，默认为摄氏度"
}
},
"required": ["city"]
}
}
}
字段说明：
| 字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
type |
string | 是 | 固定为 "function" |
function.name |
string | 是 | 函数名称，需与代码中的函数名一致 |
function.description |
string | 是 | 函数功能的自然语言描述，模型据此判断何时调用该函数 |
function.parameters |
object | 是 | 函数参数的 JSON Schema 定义 |
function.parameters.type |
string | 是 | 固定为 "object" |
function.parameters.properties |
object | 是 | 各参数的类型和描述 |
function.parameters.required |
array | 否 | 必填参数列表 |
提示：
description字段的质量直接影响模型的调用准确率。描述应该清晰、准确、包含使用场景和约束条件。
4.2 工具调用响应
当模型决定调用函数时，返回的消息结构如下：
{
"role": "assistant",
"content": null,
"tool_calls": [
{
"id": "call_abc123",
"type": "function",
"function": {
"name": "get_weather",
"arguments": "{\"city\": \"北京\"}"
}
}
]
}
关键字段：
content为null，表示模型这次没有生成文本，而是发出了函数调用请求tool_calls是一个数组，支持一次返回多个函数调用请求（并行调用）id是调用的唯一标识，用于后续回传结果时关联arguments是一个 JSON 字符串（注意不是对象），需要json.loads()解析
4.3 工具执行结果回传
应用层执行完函数后，需要将结果以特定格式回传给模型：
{
"role": "tool",
"tool_call_id": "call_abc123",
"name": "get_weather",
"content": "晴，28°C"
}
关键字段：
role固定为"tool"tool_call_id必须与模型返回的调用id一一对应content是函数执行的结果（字符串形式）
4.4 tool_choice 参数
调用 API 时，可以通过 tool_choice 参数控制模型的工具调用行为：
| 值 | 行为 |
|---|---|
"auto" |
由模型自主决定是否调用工具（推荐） |
"none" |
强制模型不调用任何工具 |
"required" |
强制模型必须调用某个工具 |
{"type": "function", "function": {"name": "xxx"}} |
强制模型调用指定的函数 |
五、实践一：基于 OpenAI SDK 原生实现
5.1 需求说明
当用户向大模型提问时，大模型可以自动调用本地的两个方法：
get_weather：获取指定城市的天气信息dress_advice：根据天气情况给出穿衣建议
并将工具的执行结果整合成最终的自然语言回答返回给用户。
5.2 环境准备
pip install openai python-dotenv 5.3 项目配置
创建 .env 文件存放 API 配置（放在项目根目录）：
MODEL_API_KEY=your_api_key_here
MODEL_BASE_URL=https://api.deepseek.com
MODEL_NAME=deepseek-chat
创建配置加载模块 config.py：
import os
from dotenv import load_dotenv
def load_project_env() -> None:
"""
加载项目根目录下的 .env 文件。
"""
repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
env_path = os.path.join(repo_root, ".env")
load_dotenv(env_path)
5.4 完整实现代码
"""
Function Call 基础示例 —— 基于 OpenAI SDK 原生实现
功能：根据用户出行计划，自动查询各城市天气并给出穿衣建议
"""
import json
import os
from openai import OpenAI
from config import load_project_env
load_project_env()
# ===================== 第一步：定义本地工具函数 =====================
def get_weather(city: str) -> str:
"""
根据城市名称返回模拟天气情况。
实际项目中应替换为真实的天气 API 调用。
"""
dummy_weather = {
"北京": "晴，28°C",
"上海": "多云，22°C",
"广州": "小雨，19°C",
"长沙": "阴，25°C",
"深圳": "多云转晴，26°C",
"成都": "小雨，18°C",
}
return dummy_weather.get(city, f"未找到 {city} 的天气信息")
def dress_advice(weather: str) -> str:
"""
根据天气情况给出穿衣建议。
"""
if "雨" in weather:
return "建议穿防水外套，记得带伞，鞋子选择防水款式。"
elif "晴" in weather:
# 判断温度
try:
temp = int(weather.split("°C")[0].split("，")[-1].strip())
if temp >= 30:
return "天气炎热，建议穿短袖短裤，注意防晒和补水。"
elif temp >= 20:
return "天气晴朗温度适宜，穿轻薄长袖或T恤即可。"
else:
return "虽然晴天但温度偏低，建议穿薄外套搭配长袖。"
except (ValueError, IndexError):
return "晴天建议适当防晒，根据体感增减衣物。"
elif "阴" in weather:
return "阴天温差较小，适合春秋装，穿薄外套即可。"
elif "多云" in weather:
return "多云天气，适合穿轻便舒适的衣物，可备一件薄外套。"
else:
return "根据体感温度适当增减衣物，建议随身携带一件外套。"
# ===================== 第二步：注册 Tools（JSON Schema 定义） =====================
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "根据城市名称返回该城市的当前天气情况，包括天气状况和温度。"
"适用于用户询问某个城市天气的场景。",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，例如：北京、上海、广州、长沙"
}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "dress_advice",
"description": "根据天气情况给出穿衣搭配建议。"
"适用于用户询问在某种天气下应该怎么穿衣服的场景。",
"parameters": {
"type": "object",
"properties": {
"weather": {
"type": "string",
"description": "天气描述信息，例如：晴，28°C / 小雨，19°C"
}
},
"required": ["weather"]
}
}
}
]
# ===================== 第三步：定义消息历史和工具映射 =====================
messages = [
{
"role": "system",
"content": (
"你是一个贴心的出行助手。当用户描述出行计划时，"
"你需要先查询各地天气，再根据天气给出穿衣建议。"
"回答要自然、亲切，像朋友一样给出建议。"
)
},
{
"role": "user",
"content": "我今天要去上海，明天去长沙，后天去北京，大后天去广州，我该怎么准备衣服？"
}
]
# 工具映射：函数名 -> 实际函数对象
tool_map = {
"get_weather": get_weather,
"dress_advice": dress_advice,
}
# ===================== 第四步：初始化客户端并执行调用循环 =====================
client = OpenAI(
api_key=os.getenv("MODEL_API_KEY"),
base_url=os.getenv("MODEL_BASE_URL"),
)
# Function Call 的核心循环
while True:
# 向模型发起请求（携带工具定义）
resp = client.chat.completions.create(
model=os.getenv("MODEL_NAME", "deepseek-chat"),
messages=messages,
tools=tools,
tool_choice="auto", # 让模型自主决定是否调用工具
)
msg = resp.choices[0].message
# 如果模型没有调用任何工具，说明已经生成了最终回答，退出循环
if not msg.tool_calls:
print("=" * 60)
print("最终回答：")
print(msg.content)
print("=" * 60)
break
# 将模型的 assistant 消息（含 tool_calls）加入历史
messages.append(msg.model_dump())
# 依次处理每一个工具调用
for call in msg.tool_calls:
func_name = call.function.name
func_args = json.loads(call.function.arguments or "{}")
print(f"[调用工具] {func_name}({func_args})")
# 执行对应的本地函数
if func_name in tool_map:
result = tool_map[func_name](**func_args)
else:
result = f"未知工具：{func_name}"
print(f"[工具结果] {result}")
# 将工具执行结果回传给模型
messages.append({
"role": "tool",
"tool_call_id": call.id,
"name": func_name,
"content": str(result),
})
5.5 运行结果示例
[调用工具] get_weather({"city": "上海"})
[工具结果] 多云，22°C
[调用工具] get_weather({"city": "长沙"})
[工具结果] 阴，25°C
[调用工具] get_weather({"city": "北京"})
[工具结果] 晴，28°C
[调用工具] get_weather({"city": "广州"})
[工具结果] 小雨，19°C
[调用工具] dress_advice({"weather": "多云，22°C"})
[工具结果] 多云天气，适合穿轻便舒适的衣物，可备一件薄外套。
[调用工具] dress_advice({"weather": "阴，25°C"})
[工具结果] 阴天温差较小，适合春秋装，穿薄外套即可。
[调用工具] dress_advice({"weather": "晴，28°C"})
[工具结果] 天气晴朗温度适宜，穿轻薄长袖或T恤即可。
[调用工具] dress_advice({"weather": "小雨，19°C"})
[工具结果] 建议穿防水外套，记得带伞，鞋子选择防水款式。
============================================================
最终回答：
你好！根据你接下来几天的出行计划，我帮你查看了各地天气并整理了穿衣建议：
**今天 - 上海（多云，22°C）**：适合穿轻便舒适的衣物，建议备一件薄外套。
**明天 - 长沙（阴，25°C）**：阴天温差较小，穿春秋装搭配薄外套即可。
**后天 - 北京（晴，28°C）**：天气晴朗温度适宜，穿轻薄长袖或T恤即可。
**大后天 - 广州（小雨，19°C）**：记得带伞，穿防水外套，鞋子选择防水款式。
总体建议：带一件薄外套、一件防水外套，一把折叠伞是必备的。祝旅途愉快！
============================================================
5.6 代码流程解析
整个调用过程可以概括为以下循环：
┌──────────────────────────────────────────┐
│ 开始：发送请求+工具定义 │
│ │
▼ │
┌─────────┐ │
│ 模型响应 │ │
└────┬────┘ │
│ │
▼ │
┌──────────────┐ 是 ┌─────────────┐ │
│ 有 tool_calls?│─────────▶│ 执行本地函数 │ │
└──────┬───────┘ └──────┬──────┘ │
│ 否 │ │
▼ ▼ │
┌──────────┐ ┌──────────────┐ │
│ 输出最终 │ │ 结果加入 messages│───────┘
│ 回答 │ └──────────────┘
└──────────┘
六、实践二：基于 LangChain 框架实现
6.1 LangChain 简介
LangChain 是目前最流行的 LLM 应用开发框架之一，它将工具定义、Agent 编排、消息管理等进行了高度封装，让开发者可以用更少的代码实现复杂的 Agent 逻辑。
LangChain 对 Function Call 的封装主要体现在：
- 使用
@tool装饰器定义工具，函数的 docstring 自动作为工具描述 - 使用
create_react_agent（基于 LangGraph）自动编排 Agent 循环 - 自动管理消息历史和工具结果的回传
6.2 环境准备
pip install langchain-core langchain-openai langgraph python-dotenv 6.3 完整实现代码
"""
Function Call 示例 —— 基于 LangChain + LangGraph 实现
功能：根据用户出行计划，自动查询各城市天气并给出穿衣建议
"""
import os
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from config import load_project_env
load_project_env()
# ===================== 第一步：定义工具（使用 @tool 装饰器） =====================
# 注意：函数的 docstring 非常重要！
# LangChain 会自动将 docstring 作为工具的 description 传给模型
@tool
def get_weather(city: str) -> str:
"""根据城市名称返回该城市的当前天气情况，包括天气状况和温度。
Args:
city: 城市名称，例如：北京、上海、广州
"""
dummy_weather = {
"北京": "晴，28°C",
"上海": "多云，22°C",
"广州": "小雨，19°C",
"长沙": "阴，25°C",
"深圳": "多云转晴，26°C",
"成都": "小雨，18°C",
}
return dummy_weather.get(city, f"未找到 {city} 的天气信息")
@tool
def dress_advice(weather: str) -> str:
"""根据天气情况给出穿衣搭配建议。
Args:
weather: 天气描述信息，例如：晴，28°C / 小雨，19°C
"""
if "雨" in weather:
return "建议穿防水外套，记得带伞，鞋子选择防水款式。"
elif "晴" in weather:
try:
temp = int(weather.split("°C")[0].split("，")[-1].strip())
if temp >= 30:
return "天气炎热，建议穿短袖短裤，注意防晒和补水。"
elif temp >= 20:
return "天气晴朗温度适宜，穿轻薄长袖或T恤即可。"
else:
return "虽然晴天但温度偏低，建议穿薄外套搭配长袖。"
except (ValueError, IndexError):
return "晴天建议适当防晒，根据体感增减衣物。"
elif "阴" in weather:
return "阴天温差较小，适合春秋装，穿薄外套即可。"
elif "多云" in weather:
return "多云天气，适合穿轻便舒适的衣物，可备一件薄外套。"
else:
return "根据体感温度适当增减衣物，建议随身携带一件外套。"
# ===================== 第二步：初始化模型 =====================
llm = ChatOpenAI(
api_key=os.getenv("MODEL_API_KEY"),
base_url=os.getenv("MODEL_BASE_URL"),
model=os.getenv("MODEL_NAME", "deepseek-chat"),
temperature=0, # 降低随机性，提高工具调用准确率
)
# ===================== 第三步：创建 Agent =====================
tools = [get_weather, dress_advice]
agent = create_react_agent(
model=llm,
tools=tools,
prompt="你是一个贴心的出行助手。当用户描述出行计划时，"
"你需要先查询各地天气，再根据天气给出穿衣建议。"
"回答要自然、亲切，像朋友一样给出建议。",
)
# ===================== 第四步：运行交互循环 =====================
if __name__ == "__main__":
print("出行穿衣助手已启动！输入 'exit' 退出。")
print("-" * 50)
while True:
user_input = input("\n请输入你的问题：").strip()
if user_input.lower() in ("exit", "quit", "q"):
print("再见！")
break
if not user_input:
continue
# invoke 会自动处理整个 Agent 循环（调用工具 -> 获取结果 -> 再次推理）
state = agent.invoke({"messages": [HumanMessage(content=user_input)]})
# 输出最终回答
messages = state.get("messages", [])
if messages:
final_message = messages[-1]
print(f"\n助手回答：\n{final_message.content}")
6.4 与原生 SDK 实现的对比
| 维度 | OpenAI SDK 原生 | LangChain + LangGraph |
|---|---|---|
| 工具定义 | 手写 JSON Schema | @tool 装饰器 + docstring 自动提取 |
| Agent 循环 | 手动 while 循环 | create_react_agent 自动编排 |
| 消息管理 | 手动维护 messages 列表 | 框架自动管理 |
| 工具执行 | 手动解析参数并调用 | 框架自动解析和调用 |
| 灵活性 | 完全可控 | 封装度高，适合快速开发 |
| 调试难度 | 需自行添加日志 | 可开启 debug 模式查看详细流程 |
| 适用场景 | 需要精细控制流程的场景 | 快速原型开发、标准 Agent 场景 |
七、综合实战：MySQL 数据库智能查询 Agent
接下来，我们实现一个更复杂的案例：用自然语言查询 MySQL 数据库。用户只需用自然语言描述需求（如"用户最多的城市是哪个？"），Agent 就能自动获取数据库表结构、生成 SQL、执行查询、并以自然语言返回分析结果。
7.1 架构设计
┌──────────────┐
│ 用户输入 │ "用户最多的城市是哪个？"
└──────┬───────┘
▼
┌──────────────┐
│ LLM Agent │ 分析意图，决定调用哪些工具
└──┬───┬───┬───┘
│ │ │
▼ ▼ ▼
┌──────┐ ┌──────────┐ ┌──────────┐
│获取表 │ │执行SQL查询│ │查看表数据 │
│结构 │ │ │ │ │
└──┬───┘ └────┬─────┘ └────┬─────┘
│ │ │
▼ ▼ ▼
┌──────────────────────────────┐
│ MySQL 数据库 │
└──────────────────────────────┘
│
▼
┌──────────────┐
│ LLM 整合结果 │ 生成自然语言分析报告
└──────┬───────┘
▼
┌──────────────┐
│ 最终回答 │ "用户最多的城市是北京，共 15,230 人..."
└──────────────┘
7.2 数据库准备
首先创建示例数据库和表：
-- 创建数据库
CREATE DATABASE IF NOT EXISTS sales_analysis_db
DEFAULT CHARSET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE sales_analysis_db;
-- 用户表
CREATE TABLE IF NOT EXISTS `users` (
`id` INT PRIMARY KEY AUTO_INCREMENT COMMENT '用户ID',
`name` VARCHAR(50) NOT NULL COMMENT '用户姓名',
`city` VARCHAR(50) NOT NULL COMMENT '所在城市',
`age` INT COMMENT '年龄',
`gender` ENUM('男', '女') COMMENT '性别',
`register_date` DATE COMMENT '注册日期',
`vip_level` INT DEFAULT 0 COMMENT 'VIP等级(0-5)'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户信息表';
-- 订单表
CREATE TABLE IF NOT EXISTS `orders` (
`id` INT PRIMARY KEY AUTO_INCREMENT COMMENT '订单ID',
`user_id` INT NOT NULL COMMENT '用户ID',
`product_name` VARCHAR(100) NOT NULL COMMENT '商品名称',
`category` VARCHAR(50) COMMENT '商品类别',
`amount` DECIMAL(10, 2) NOT NULL COMMENT '订单金额',
`quantity` INT NOT NULL DEFAULT 1 COMMENT '购买数量',
`order_date` DATE NOT NULL COMMENT '订单日期',
`status` ENUM('待付款', '已付款', '已发货', '已完成', '已取消') DEFAULT '待付款' COMMENT '订单状态',
FOREIGN KEY (`user_id`) REFERENCES `users`(`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='订单表';
-- 插入示例数据
INSERT INTO `users` (`name`, `city`, `age`, `gender`, `register_date`, `vip_level`) VALUES
('张三', '北京', 28, '男', '2023-01-15', 3),
('李四', '上海', 35, '女', '2023-02-20', 2),
('王五', '北京', 22, '男', '2023-03-10', 1),
('赵六', '广州', 30, '女', '2023-04-05', 4),
('钱七', '深圳', 45, '男', '2023-05-18', 5),
('孙八', '上海', 27, '女', '2023-06-22', 1),
('周九', '北京', 33, '男', '2023-07-08', 3),
('吴十', '成都', 29, '女', '2023-08-15', 2);
INSERT INTO `orders` (`user_id`, `product_name`, `category`, `amount`, `quantity`, `order_date`, `status`) VALUES
(1, 'MacBook Pro 14寸', '电脑', 14999.00, 1, '2024-01-10', '已完成'),
(1, 'AirPods Pro', '配件', 1899.00, 1, '2024-02-15', '已完成'),
(2, 'iPhone 15 Pro', '手机', 8999.00, 1, '2024-01-20', '已完成'),
(3, 'iPad Air', '平板', 4799.00, 1, '2024-03-05', '已发货'),
(3, 'Apple Watch', '穿戴', 2999.00, 1, '2024-03-05', '已发货'),
(4, 'MacBook Air', '电脑', 8999.00, 1, '2024-02-28', '已完成'),
(5, 'iPhone 15 Pro Max', '手机', 9999.00, 1, '2024-01-05', '已完成'),
(5, 'Mac Studio', '电脑', 29999.00, 1, '2024-04-01', '已完成'),
(6, 'AirPods 3', '配件', 1399.00, 2, '2024-03-15', '已完成'),
(7, 'iPad Pro', '平板', 8999.00, 1, '2024-04-10', '待付款'),
(8, 'iPhone 15', '手机', 6999.00, 1, '2024-02-20', '已完成');
7.3 完整实现代码
"""
MySQL 数据库智能查询 Agent
功能：用户通过自然语言提问，Agent 自动查询 MySQL 数据库并返回分析结果
"""
from collections import defaultdict
import os
import pymysql
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from config import load_project_env
load_project_env()
# ===================== 第一步：数据库基础操作 =====================
DB_NAME = "sales_analysis_db"
def get_connection():
"""创建数据库连接"""
return pymysql.connect(
host="localhost",
user="root",
password="123456",
database=DB_NAME,
charset="utf8mb4",
cursorclass=pymysql.cursors.Cursor,
)
def run_sql(sql: str) -> list:
"""
执行只读 SQL 查询，返回结果集。
注意：生产环境中应该增加 SQL 审核机制，仅允许 SELECT 语句。
"""
# 安全检查：禁止危险操作
forbidden_keywords = ["DROP", "DELETE", "TRUNCATE", "ALTER", "INSERT", "UPDATE", "CREATE"]
sql_upper = sql.upper().strip()
for keyword in forbidden_keywords:
if sql_upper.startswith(keyword):
return f"安全拒绝：不允许执行 {keyword} 操作"
conn = get_connection()
try:
with conn.cursor() as cursor:
cursor.execute(sql)
result = cursor.fetchall()
return result
except Exception as e:
return f"SQL 执行错误：{str(e)}"
finally:
conn.close()
# ===================== 第二步：数据库结构获取 =====================
def load_schema() -> dict:
"""获取数据库中所有表的结构信息"""
sql = """
SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_COMMENT
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = %s
ORDER BY TABLE_NAME, ORDINAL_POSITION
"""
conn = get_connection()
try:
with conn.cursor() as cursor:
cursor.execute(sql, (DB_NAME,))
result = cursor.fetchall()
finally:
conn.close()
schema_dict = defaultdict(list)
for row in result:
table_name = row[0]
schema_dict[table_name].append(row)
return schema_dict
def schema_to_markdown(schema_dict: dict) -> str:
"""将数据库结构转换为 Markdown 格式"""
lines = [f"# 数据库 `{DB_NAME}` 结构信息\n"]
for table, cols in schema_dict.items():
lines.append(f"## 表：{table}\n")
lines.append("| 列名 | 数据类型 | 可空 | 说明 |")
lines.append("|------|---------|------|------|")
for _, column_name, column_type, is_nullable, column_comment in cols:
nullable = "是" if is_nullable == "YES" else "否"
comment = column_comment or "-"
lines.append(f"| `{column_name}` | {column_type} | {nullable} | {comment} |")
lines.append("") # 空行分隔
return "\n".join(lines)
# 缓存 schema，避免重复查询
_schema_cache = None
def get_cached_schema() -> str:
"""获取缓存的数据库结构（Markdown 格式）"""
global _schema_cache
if _schema_cache is None:
schema = load_schema()
_schema_cache = schema_to_markdown(schema)
return _schema_cache
# ===================== 第三步：定义工具 =====================
@tool
def get_mysql_schema() -> str:
"""获取数据库的完整结构信息，包括所有表名、字段名、字段类型和字段说明。
在不确定表结构时应先调用此工具了解数据库架构。"""
return get_cached_schema()
@tool
def run_mysql_sql_tool(sql: str) -> str:
"""执行 SQL 查询语句并返回查询结果。仅支持 SELECT 查询。
Args:
sql: 要执行的 SELECT SQL 查询语句。表名和字段名建议使用反引号包裹。
"""
result = run_sql(sql)
return str(result)
@tool
def peek_table(table: str, n: int = 5) -> str:
"""快速查看指定表的前 n 行数据，用于了解表中的实际数据内容。
Args:
table: 要查看的表名
n: 要查看的行数，默认为 5
"""
sql = f"SELECT * FROM `{table}` LIMIT {n}"
return str(run_sql(sql))
# ===================== 第四步：定义系统提示词 =====================
SYSTEM_PROMPT = """
# 角色定位
你是一位资深的 MySQL 数据库分析师，擅长将用户的自然语言问题转化为精准的 SQL 查询，并用通俗易懂的语言解释查询结果。
# 工作流程
1. **理解需求**：仔细分析用户的问题，确定需要查询哪些表和字段
2. **了解结构**：如果不确定表结构，先调用 get_mysql_schema 工具获取数据库结构
3. **验证数据**：对不熟悉的表，可以用 peek_table 先查看前几行数据
4. **编写SQL**：编写准确的 SQL 查询语句
5. **执行查询**：调用 run_mysql_sql_tool 执行 SQL
6. **分析结果**：用自然语言解释查询结果，给出有价值的洞察
# SQL 编写规范
- 所有表名和字段名使用反引号(`)包裹
- 优先使用索引字段作为查询条件
- 预估数据量大的查询添加 LIMIT 子句
- 避免 SELECT *，明确指定需要的字段
- 多表连接必须明确指定连接条件
- 使用聚合函数时注意 GROUP BY 的搭配
# 回答要求
- 用中文回答，语言专业但易懂
- 先展示关键数据，再给出分析和建议
- 如果查询结果异常，主动分析可能的原因
"""
# ===================== 第五步：创建 Agent 并运行 =====================
def create_db_agent():
"""创建数据库查询 Agent"""
llm = ChatOpenAI(
api_key=os.getenv("MODEL_API_KEY"),
base_url=os.getenv("MODEL_BASE_URL"),
model=os.getenv("MODEL_NAME", "deepseek-chat"),
temperature=0,
)
tools = [get_mysql_schema, run_mysql_sql_tool, peek_table]
agent = create_react_agent(
model=llm,
tools=tools,
prompt=SYSTEM_PROMPT,
)
return agent
def ask(question: str) -> str:
"""向 Agent 提问并获取回答"""
agent = create_db_agent()
state = agent.invoke({"messages": [HumanMessage(content=question)]})
messages = state.get("messages", [])
if messages:
return messages[-1].content
return "查询完成，但未获得预期结果。"
# ===================== 第六步：主程序入口 =====================
if __name__ == "__main__":
print("MySQL 智能查询助手已启动！")
print("示例问题：")
print(" - 用户最多的城市是哪个？")
print(" - 各商品类别的总销售额是多少？")
print(" - VIP等级最高的用户有哪些？")
print(" - 每个月的订单数量变化趋势如何？")
print("输入 'exit' 退出。")
print("-" * 60)
while True:
user_input = input("\n请输入你的问题：").strip()
if user_input.lower() in ("exit", "quit", "q"):
print("再见！")
break
if not user_input:
continue
try:
result = ask(user_input)
print(f"\n{result}")
except Exception as e:
print(f"\n查询出错：{str(e)}")
7.4 运行效果示例
请输入你的问题：用户最多的城市是哪个？
[Agent 内部调用流程]
→ 调用 get_mysql_schema() → 获取到 users 和 orders 表结构
→ 调用 run_mysql_sql_tool("SELECT `city`, COUNT(*) AS user_count FROM `users` GROUP BY `city` ORDER BY user_count DESC LIMIT 5")
→ 返回结果：(('北京', 3), ('上海', 2), ('广州', 1), ('深圳', 1), ('成都', 1))
最终回答：
根据查询结果，**用户最多的城市是北京**，共有 3 位用户。
各城市用户数量排名如下：
| 排名 | 城市 | 用户数 |
|------|------|--------|
| 1 | 北京 | 3 |
| 2 | 上海 | 2 |
| 3 | 广州 | 1 |
| 3 | 深圳 | 1 |
| 3 | 成都 | 1 |
可以看出，北京和上海的用户占比超过一半（5/8），建议将营销资源重点投放在这两个城市。
八、Function Call 的演进与最佳实践
8.1 Function Call 的版本演进
| 时间 | 版本/更新 | 主要变化 |
|---|---|---|
| 2023.06 | 初始版本 | 首次支持 Function Call，单次调用一个函数 |
| 2023.11 | 并行函数调用 | 支持在一次响应中返回多个 tool_calls，可并行执行 |
| 2024.01 | Structured Outputs | 引入 strict: true 模式，保证输出严格符合 JSON Schema |
| 2024.05 | GPT-4o 发布 | 原生支持函数调用，速度更快、成本更低 |
| 2024.08 | Structured Outputs 全面开放 | 所有支持函数调用的模型均支持 Structured Outputs |
8.2 最佳实践
8.2.1 工具描述要清晰准确
工具的 description 是模型判断何时调用的唯一依据。描述不清晰会导致模型误判或遗漏调用。
# ❌ 不好的描述
"description": "查询数据"
# ✅ 好的描述
"description": "根据 SQL 查询语句查询 MySQL 数据库并返回结果。"
"仅支持 SELECT 查询，不支持 INSERT/UPDATE/DELETE 等写操作。"
"适用于用户询问数据分析、统计、查询等场景。"
8.2.2 参数描述要有示例
# ❌ 不好的参数描述
"city": {"type": "string", "description": "城市"}
# ✅ 好的参数描述
"city": {
"type": "string",
"description": "城市名称，例如：北京、上海、广州、深圳"
}
8.2.3 合理控制工具数量
工具数量过多会增加模型的选择负担，降低准确率。一般建议：
- 单次请求中的工具数量不超过 20 个
- 如果工具确实很多，考虑使用工具分组或动态加载策略
8.2.4 做好错误处理
@tool
def run_mysql_sql_tool(sql: str) -> str:
"""执行 SQL 查询"""
try:
result = run_sql(sql)
if isinstance(result, str) and "安全拒绝" in result:
return result
if not result:
return "查询结果为空，请检查查询条件是否正确。"
return str(result)
except Exception as e:
return f"查询执行失败：{str(e)}。请检查 SQL 语法是否正确。"
8.2.5 控制 Token 消耗
当查询结果很大时，返回给模型的 token 会很多，既浪费成本又可能超出上下文窗口。建议：
@tool
def run_mysql_sql_tool(sql: str) -> str:
"""执行 SQL 查询"""
result = run_sql(sql)
# 限制返回行数
MAX_ROWS = 50
if isinstance(result, (list, tuple)) and len(result) > MAX_ROWS:
truncated = result[:MAX_ROWS]
return f"{str(truncated)}\n\n... 共 {len(result)} 条结果，已截取前 {MAX_ROWS} 条"
return str(result)
8.2.6 安全性考虑
在生产环境中，使用 Function Call 操作数据库时必须注意安全：
def validate_sql(sql: str) -> tuple[bool, str]:
"""SQL 安全审核"""
sql_upper = sql.upper().strip()
# 1. 只允许 SELECT
if not sql_upper.startswith("SELECT"):
return False, "仅允许 SELECT 查询"
# 2. 禁止子查询中包含写操作
dangerous = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "TRUNCATE", "GRANT"]
for keyword in dangerous:
if keyword in sql_upper:
return False, f"SQL 中不允许包含 {keyword}"
# 3. 强制添加 LIMIT（防止全表扫描）
if "LIMIT" not in sql_upper:
sql = sql.rstrip(";") + " LIMIT 100"
return True, sql
8.3 并行函数调用示例
当模型需要同时获取多个独立数据时，可以一次返回多个 tool_calls：
{
"tool_calls": [
{
"id": "call_001",
"function": {"name": "get_weather", "arguments": "{\"city\": \"北京\"}"}
},
{
"id": "call_002",
"function": {"name": "get_weather", "arguments": "{\"city\": \"上海\"}"}
},
{
"id": "call_003",
"function": {"name": "get_weather", "arguments": "{\"city\": \"广州\"}"}
}
]
}
这样应用层可以并行执行这三个查询（例如使用 asyncio 或多线程），显著提高响应速度。
import asyncio
from concurrent.futures import ThreadPoolExecutor
# 并行执行多个工具调用
async def execute_tools_parallel(tool_calls, tool_map):
"""并行执行多个工具调用"""
loop = asyncio.get_event_loop()
results = []
with ThreadPoolExecutor() as executor:
tasks = []
for call in tool_calls:
name = call.function.name
args = json.loads(call.function.arguments or "{}")
func = tool_map.get(name)
if func:
tasks.append(loop.run_in_executor(executor, lambda f=func, a=args: f(**a)))
results = await asyncio.gather(*tasks)
return results
九、总结与展望
9.1 总结
本文从概念到实践，系统介绍了 Function Call 的方方面面：
| 层次 | 内容 |
|---|---|
| 概念层 | Function Call 的定义、意义、解决的问题 |
| 原理层 | 工作流程、请求-响应结构、核心字段含义 |
| 实践层 | OpenAI SDK 原生实现、LangChain 框架实现 |
| 进阶层 | MySQL 智能查询 Agent、并行调用、安全机制、最佳实践 |
Function Call 的本质是为大模型提供了结构化的外部交互接口，使得大模型从一个"聊天机器人"进化为一个真正有能力执行操作的 AI Agent。
9.2 展望
Function Call 正在成为 AI Agent 生态的基础设施。随着技术的演进，以下趋势值得关注：
- 1.MCP（Model Context Protocol）：Anthropic 提出的标准化协议，旨在统一模型与外部工具的交互方式
- 2.Multi-Agent 系统：多个 Agent 协作完成复杂任务，每个 Agent 拥有不同的工具集
- 3.代码生成与执行：模型不再只是调用预定义函数，而是动态生成代码并执行
- 4.工具自动发现：模型能够自动发现和学习使用新的工具，无需预先定义
掌握 Function Call，就是掌握了构建 AI Agent 应用的核心能力。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
15
0- 0
已为社区贡献16条内容
所有评论(0)