【大模型实战笔记 3】大模型Function Call的原理及应用
《大模型Function Call的原理及应用》
【注:代码附于文末,代码简单容易入手】
- 《大模型Function Call的原理及应用》
- 一、什么是 Function Call?
- Function Call 能解决哪些问题?
- 二、Function Call 的工作原理
- 三、项目实战
- 实战一:单函数调用 —— 实时天气查询
- (1)Function Call 的单一函数应用
- (2)场景目标
- (3)实现步骤
- 1. 准备工作
- 2. 定义工具函数与描述
- 3. 主逻辑调用流程
- 实战二:多函数协同 —— 航班与票价查询
- (1)场景目标
- (2)关键设计
- (3)核心代码片段(简化)
- 实战三:连接数据库 —— SQL 查询助手
- (1)场景目标
- (2)实现要点
- 四、总结与思考
- 五、代码附录
- (1)实时天气查询
- (2)航班与票价查询
- (3)SQL 查询助手
摘要:Function Call 是大语言模型(LLM)与外部世界交互的关键能力之一。本文系统讲解 Function Call 的核心原理,并通过三个典型应用场景(单函数调用、多函数协同、数据库查询)进行代码级实战演示,帮助开发者快速掌握其在实际项目中的落地方法。
一、什么是 Function Call?
Function Call(函数调用)是 OpenAI 于 2023 年 6 月正式推出的一项重要功能,其核心思想是:允许大语言模型在生成文本的过程中,主动调用外部函数或 API,从而获取实时数据、访问专业数据库或执行特定工具操作。
Function Call 能解决哪些问题?
- 信息实时性不足:LLM 的训练数据截止于某一时间点,无法获取最新新闻、股价、天气等动态信息。Function Call 可桥接实时 API。
- 数据领域局限性:模型虽广博,但难以覆盖医学、法律、金融等垂直领域的深度知识。Function Call允许模型调用外部数据库或API,获取特定领域的详细信息。
- 功能扩展性受限:模型本身不具备执行复杂计算、文件处理、数据库查询等能力。Function Call 提供了“插件式”能力扩展机制。
⚠️ 关键认知:大模型 不会直接执行函数,仅会返回结构化的函数调用请求(含函数名与参数)。真正的函数执行由开发者在后端完成,并将结果回传给模型以生成最终回答。
二、Function Call 的工作原理
在无 Function Call 的传统模式中,用户请求 → 模型响应,流程简单直接。
而在启用 Function Call 后,交互流程变为:
- 用户发送请求(Prompt)及可用工具列表(Tools)至服务端;
- LLM 判断是否需要调用外部函数:
- 若不需要,直接返回自然语言回答;
- 若需要,返回一个结构化的 函数调用请求(含函数名与参数);
- 服务端解析该请求,调用对应本地函数,获取执行结果;
- 将函数结果作为新消息(role: tool)加入对话上下文;
- 再次调用模型,结合原始问题与函数结果,生成最终连贯回答。
整个过程通常涉及 两次模型调用,确保回答既准确又自然。
三、项目实战
实战一:单函数调用 —— 实时天气查询
(1)Function Call 的单一函数应用
(2)场景目标
构建一个能根据用户输入的城市名,返回当日天气的聊天机器人。
(3)实现步骤
1. 准备工作
- 使用智谱 AI 的 GLM-4 模型(支持 Function Call);
- 申请 API Key 并配置环境变量;
- 安装依赖:
pip install zhipuai requests python-dotenv
。
2. 定义工具函数与描述
# tools.py
import json
import requestsdef get_current_weather(location):"""获取给定城市的当前天气"""with open('./cityCode_use.json', 'r') as f:city_data = json.load(f)city_code = next((item["编码"] for item in city_data if item["市名"] == location), None)if not city_code:return json.dumps({"error": "城市未找到"}, ensure_ascii=False)url = f"http://t.weather.itboy.net/api/weather/city/{city_code}"resp = requests.get(url)data = resp.json()forecast = data["data"]["forecast"][0]return json.dumps({"location": location,"high_temperature": forecast["high"],"low_temperature": forecast["low"],"week": forecast["week"],"type": forecast["type"]}, ensure_ascii=False)tools = [{"type": "function","function": {"name": "get_current_weather","description": "获取给定位置的当前天气","parameters": {"type": "object","properties": {"location": {"type": "string", "description": "城市名,如北京、上海"}},"required": ["location"]}}
}]
3. 主逻辑调用流程
# weather_zhipu.py
from zhipuai import ZhipuAI
from dotenv import load_dotenv
import os
from tools import tools, get_current_weather, parse_responseload_dotenv()
client = ZhipuAI(api_key=os.environ['zhupu_api'])def chat_completion(messages, tools=None):return client.chat.completions.create(model="glm-4", messages=messages, tools=tools, tool_choice="auto")def main():messages = [{"role": "system", "content": "你是天气助手,请根据用户提供的城市回答天气,不要编造信息。"},{"role": "user", "content": "今天北京的天气如何?"}]# 第一次调用:获取函数调用请求response = chat_completion(messages, tools)messages.append(response.choices[0].message.model_dump())# 执行函数func_result = parse_response(response)tool_call = response.choices[0].message.tool_calls[0]messages.append({"role": "tool","tool_call_id": tool_call.id,"name": tool_call.function.name,"content": func_result})# 第二次调用:生成最终回答final_resp = chat_completion(messages)print(final_resp.choices[0].message.content)if __name__ == "__main__":main()
输出示例:
“根据您的查询,我获取到了北京市当前的天气情况。今天是星期一,北京的天气情况是晴天,最高气温为33℃,最低气温为17℃。”
实战二:多函数协同 —— 航班与票价查询
(1)场景目标
用户输入“查郑州到北京4月2日的航班票价”,系统需先查航班号,再查票价,最终整合回答。
(2)关键设计
- 定义两个函数:
get_plane_number(start, end, date)
和get_ticket_price(number, date)
; - 模型可能需 分两步调用函数,因此主逻辑需支持 多次工具调用循环。
(3)核心代码片段(简化)
# airplane_function_tools.py
tools = [{"function": {"name": "get_plane_number","description": "根据始发地、目的地和日期查询航班号","parameters": { ... }}},{"function": {"name": "get_ticket_price","description": "查询某航班在某日的价格","parameters": { ... }}}
]# muti_utils.py
def parse_function_call(model_response):tool_call = model_response.choices[0].message.tool_calls[0]func_name = tool_call.function.nameargs = json.loads(tool_call.function.arguments)if func_name == "get_plane_number":return get_plane_number(**args)elif func_name == "get_ticket_price":return get_ticket_price(**args)
执行流程:
用户提问 → 模型调用get_plane_number
→ 获取航班号 → 模型再调用get_ticket_price
→ 最终输出:“2024年4月2日,郑州到北京的航班号为1123,票价为668元。”
实战三:连接数据库 —— SQL 查询助手
(1)场景目标
用户问“工资最高的员工是谁?”,模型自动生成 SQL 并查询数据库返回结果。
(2)实现要点
- 提供数据库表结构(Schema)作为函数描述的一部分;
- 函数
ask_database(query)
负责执行 SQL; - 模型生成的
query
必须是合法 SQL 语句。
# sql_function_tools.py
database_schema_string1 = """
CREATE TABLE emp (empno int, ename varchar(50), sal int, ...
);
CREATE TABLE DEPT (...);
"""tools = [{"function": {"name": "ask_database","description": "使用此函数回答业务问题,要求输出是一个SQL查询语句","parameters": {"properties": {"query": {"description": f"SQL应基于以下Schema编写:{database_schema_string1},仅使用MySQL语法。"}}}}
}]def ask_database(query):conn = pymysql.connect(host='localhost', user='root', password='123456', database='it_heima')cursor = conn.cursor()cursor.execute(query)result = cursor.fetchall()cursor.close()conn.close()return str(result)
效果:
用户输入:“查询最高工资的员工姓名及工资”
模型输出:“根据您的查询,工资最高的员工是 KING,工资为 5000 元。”
四、总结与思考
Function Call 极大地拓展了大模型的应用边界,使其从“纯文本生成器”进化为“智能调度中枢”。在实际开发中需注意:
- 函数描述(tools)必须清晰准确,否则模型无法正确调用;
- 参数校验与错误处理必不可少,避免因无效输入导致崩溃;
- 多轮工具调用需设计良好的消息管理机制,确保上下文连贯;
- 安全性:切勿让模型直接生成并执行任意 SQL 或系统命令,应做严格白名单限制。
Function Call 不是魔法,而是工程。只有将模型能力与可靠后端逻辑紧密结合,才能构建真正可用的智能应用。
五、代码附录
将一个项目下的代码发在同一个文件下
(1)实时天气查询
weather_zhipu.py
import os
from dotenv import load_dotenv, find_dotenv
from tools import *
from zhipuai import ZhipuAI_ = load_dotenv(find_dotenv())
zhupu_ak = os.environ['zhupu_api']
client = ZhipuAI(api_key=zhupu_ak) # 填写您自己的APIKey
ChatGLM = "glm-4"def chat_completion_request(messages, tools=None, tool_choice=None, model=ChatGLM):try:response = client.chat.completions.create(model=model,messages=messages,tools=tools,tool_choice=tool_choice,)return responseexcept Exception as e:print("Unable to generate ChatCompletion response")print(f"Exception: {e}")return edef main():messages = []messages.append({"role": "system","content": "你是一个天气播报小助手,你需要根据用户提供的地址来回答当地的天气情况,如果用户提供的问题具有不确定性,不要自己编造内容,提示用户明确输入"})messages.append({"role": "user", "content": "今天北京的天气如何"})print(messages)response = chat_completion_request(messages, tools=tools, tool_choice="auto")# 获取函数的结果function_response = parse_response(response)# 添加第一次模型得到的结果assistant_message = response.choices[0].messageprint(f'assistant_message-->{assistant_message}')messages.append(assistant_message.model_dump()) # extend conversation with assistant's replyfunction_name = response.choices[0].message.tool_calls[0].function.nameprint(f'function_name--》{function_name}')function_id = response.choices[0].message.tool_calls[0].idprint(f'function_id--》{function_id}')# 添加函数返回的结果messages.append({"role": "tool","tool_call_id": function_id,"name": function_name,"content": function_response,}) # extend conversation with function responselast_response = chat_completion_request(messages, tools=tools, tool_choice="auto")print(f'last_response--》{last_response.choices[0].message}')if __name__ == '__main__':main()
tools.py
import json
import requeststools = [{"type": "function","function": {"name": "get_current_weather","description": "获取给定位置的当前天气","parameters": {"type": "object","properties": {"location": {"type": "string","description": "城市或区,例如北京、海淀",},},"required": ["location"],},}}
]# todo:1.调用API接口,实现天气查询
def get_current_weather(location):"""得到给定地址的当前天气信息"""with open('./cityCode_use.json', 'r') as file:# 使用 json.load() 函数加载 JSON 数据data = json.load(file)city_code = ""weather_info = {}for loc in data:if location == loc["市名"]:city_code = loc["编码"]if city_code:weather_url = "http://t.weather.itboy.net/api/weather/city/" + city_coderesponse = requests.get(weather_url)result1 = eval(response.text)forecast = result1["data"]["forecast"][0]weather_info = {"location": location,"high_temperature": forecast["high"],"low_temperature": forecast["low"],"week": forecast["week"],"type": forecast["type"],}return json.dumps(weather_info, ensure_ascii=False)# todo: 2.根据模型回复来确定使用哪一个工具函数:def parse_response(response):response_message = response.choices[0].message# 检测是否需要调用函数if response_message.tool_calls:# 调用函数available_functions = {"get_current_weather": get_current_weather,} # only one function test in this example, but you can have multiplefunction_name = response_message.tool_calls[0].function.namefuction_to_call = available_functions[function_name]function_args = json.loads(response_message.tool_calls[0].function.arguments)function_response = fuction_to_call(location=function_args.get("location"),)return function_response
cityCode_use.json
[{"市名": "北京","编码": "101010100"
},{"市名": "昌平","编码": "101010700"
}
]
(2)航班与票价查询
muti_function_zhipu.py
from zhipuai import ZhipuAI
from dotenv import load_dotenv, find_dotenv
from muti_utils import *
from airplane_function_tools import *
import os
_ = load_dotenv(find_dotenv())
# 获取环境变量 ZhiPu_API_KEY
zhupu_ak = os.environ['zhupu_api']
client = ZhipuAI(api_key=zhupu_ak) # 填写您自己的APIKey
ChatGLM = "glm-4"def chat_completion_request(messages, tools=None, tool_choice=None, model=ChatGLM):try:response = client.chat.completions.create(model=model,messages=messages,tools=tools,tool_choice=tool_choice,)return responseexcept Exception as e:print("Unable to generate ChatCompletion response")print(f"Exception: {e}")return edef main():messages = []messages.append({"role": "system","content": "现在你是一个航班查询助手,将根据用户问题提供答案,但是不要假设或猜测传入函数的参数值。如果用户的描述不明确,请要求用户提供必要信息 "})messages.append({"role": "user", "content": "帮我查询2024年4月2日,郑州到北京的航班的票价"})# 1.得到第一次回复:调用:get_plane_number函数first_response = chat_completion_request(messages, tools=tools, tool_choice="auto")assistant_message1 = first_response.choices[0].messageprint(f'assistant_message1-->{assistant_message1}')# 2. 将第一次得到的模型回复结果加入messagesmessages.append(first_response.choices[0].message.model_dump())# 3. 第一次得到函数的结果first_function = parse_function_call(model_response=first_response)print(f'first_function--》{first_function}')tool_call = first_response.choices[0].message.tool_calls[0]# 4. 将函数的结果添加到messages中,继续送入模型问答messages.append({"role": "tool","tool_call_id": tool_call.id,"content": str(json.dumps(first_function))})# 5. 第二次调用模型print(messages)second_response = chat_completion_request(messages, tools=tools, tool_choice="auto")print(f'second_response--》{second_response.choices[0].message}')# 6. 将第二次得到函数结果加入信息中messages.append(second_response.choices[0].message.model_dump())second_function = parse_function_call(model_response=second_response)print(f'second_function--》{second_function}')tool2_call = second_response.choices[0].message.tool_calls[0]# 4. 将函数的结果添加到messages中,继续送入模型问答messages.append({"role": "tool","tool_call_id": tool2_call.id,"content": str(json.dumps(second_function))})last_response = chat_completion_request(messages, tools=tools, tool_choice="auto")print(f'last_response--》{last_response.choices[0].message}')
if __name__ == '__main__':main()
airplane_function_tools.py
tools = [{"type": "function","function": {"name": "get_plane_number","description": "根据始发地、目的地和日期,查询对应日期的航班号","parameters": {"type": "object","properties": {"start": {"description": "出发地","type": "string"},"end": {"description": "目的地","type": "string"},"date": {"description": "日期","type": "string",}},"required": ["start", "end", "date"]},}},{"type": "function","function": {"name": "get_ticket_price","description": "查询某航班在某日的价格","parameters": {"type": "object","properties": {"number": {"description": "航班号","type": "string"},"date": {"description": "日期","type": "string",}},"required": [ "number", "date"]},}},
]
muti_utils.py
import jsondef get_plane_number(date, start , end):plane_number = {"北京": {"深圳": "126","广州": "356",},"郑州": {"北京": "1123","天津": "3661",}}return {"date": date, "number": plane_number[start][end]}def get_ticket_price(date:str , number:str):print(date)print(number)return {"ticket_price": "668"}def parse_function_call(model_response):''':param model_response: 模型返回的结果:return: 返回函数的结果'''function_result = ''if model_response.choices[0].message.tool_calls:tool_call = model_response.choices[0].message.tool_calls[0]args = tool_call.function.argumentsfunction_result = {}if tool_call.function.name == "get_plane_number":function_result = get_plane_number(**json.loads(args))if tool_call.function.name == "get_ticket_price":function_result = get_ticket_price(**json.loads(args))return function_result
(3)SQL 查询助手
sql_zhipu.py
from zhipuai import ZhipuAI
from dotenv import load_dotenv, find_dotenv
from sql_function_tools import *
import os
_ = load_dotenv(find_dotenv())
# 获取环境变量 ZhiPu_API_KEY
zhupu_ak = os.environ['zhupu_api']
client = ZhipuAI(api_key=zhupu_ak) # 填写您自己的APIKey
ChatGLM = "glm-4"def chat_completion_request(messages, tools=None, tool_choice=None, model=ChatGLM):try:response = client.chat.completions.create(model=model,messages=messages,tools=tools,tool_choice=tool_choice,)return responseexcept Exception as e:print("Unable to generate ChatCompletion response")print(f"Exception: {e}")return edef main():messages = []messages.append({"role": "system","content": "通过针对业务数据库生成 SQL 查询来回答用户的问题"})messages.append({"role": "user", "content": "查询一下最高工资的员工姓名及对应的工资"})response = chat_completion_request(messages, tools=tools, tool_choice="auto")assistant_message = response.choices[0].messageprint(f'assistant_message1-->{assistant_message}')function_name = response.choices[0].message.tool_calls[0].function.namefunction_id = response.choices[0].message.tool_calls[0].idfunction_response = parse_response(response)print(f'assistant_message.model_dump()-->{assistant_message.model_dump()}')messages.append(assistant_message.model_dump()) # extend conversation with assistant's replymessages.append({"role": "tool","tool_call_id": function_id,"name": function_name,"content": str(function_response),}) # extend conversation with function responseprint(f'messages-->{messages}')last_response = chat_completion_request(messages, tools=tools, tool_choice="auto")print(f'last_response--》{last_response}')print(f'last_response--》{last_response.choices[0].message}')
if __name__ == '__main__':main()
sql_function_tools.py
import json
import requests
import os
import pymysql
from dotenv import load_dotenv, find_dotenv# todo: 1.描述数据库表结构(单一个表格)
database_schema_string = """CREATE TABLE `emp` (`empno` int DEFAULT NULL, --员工编号, 默认为空`ename` varchar(50) DEFAULT NULL, --员工姓名, 默认为空`job` varchar(50) DEFAULT NULL,--员工工作, 默认为空`mgr` int DEFAULT NULL,--员工领导, 默认为空`hiredate` date DEFAULT NULL,--员工入职日期, 默认为空`sal` int DEFAULT NULL,--员工的月薪, 默认为空`comm` int DEFAULT NULL,--员工年终奖, 默认为空`deptno` int DEFAULT NULL,--员工部分编号, 默认为空
)"""# todo: 2.描述数据库表结构(多个表格)
database_schema_string1 = """
CREATE TABLE `emp` (`empno` int DEFAULT NULL, --员工编号, 默认为空`ename` varchar(50) DEFAULT NULL, --员工姓名, 默认为空`job` varchar(50) DEFAULT NULL,--员工工作, 默认为空`mgr` int DEFAULT NULL,--员工领导, 默认为空`hiredate` date DEFAULT NULL,--员工入职日期, 默认为空`sal` int DEFAULT NULL,--员工的月薪, 默认为空`comm` int DEFAULT NULL,--员工年终奖, 默认为空`deptno` int DEFAULT NULL,--员工部分编号, 默认为空
);CREATE TABLE `DEPT` (`DEPTNO` int NOT NULL, -- 部门编码, 默认为空`DNAME` varchar(14) DEFAULT NULL,--部门名称, 默认为空`LOC` varchar(13) DEFAULT NULL,--地点, 默认为空PRIMARY KEY (`DEPTNO`)
);"""
tools = [{"type": "function","function": {"name": "ask_database","description": "使用此函数回答业务问题,要求输出是一个SQL查询语句","parameters": {"type": "object","properties": {"query": {"type": "string","description": f"SQL查询提取信息以回答用户的问题。"f"SQL应该使用以下数据库模式编写:{database_schema_string1}"f"查询应该以纯文本返回,而不是JSON。"f"查询应该只包含MySQL支持的语法。",}},"required": ["query"],},}}
]# todo:1.连接数据库,进行sql语句的查询def ask_database(query):"""连接数据库,进行查询"""# 1.连接到 MySQL 数据库print("进入函数内部")conn = pymysql.connect(host='localhost',port=3306,user='root',password='123456',database='it_heima',charset='utf8mb4', # 指定游标类,返回结果为字典)# 2. 创建游标cursor = conn.cursor()print(f'开始测试')# 3. 执行sql语句测试# 示例:执行 SQL 查询# sql = "SELECT * FROM emp"print(f'query--》{query}')cursor.execute(query)# 4. 获取查询结果result = cursor.fetchall()# 5.关闭游标cursor.close()# 6.关闭连接conn.close()return result# # todo: 2.根据模型回复来确定使用工具函数:def parse_response(response):response_message = response.choices[0].message# 检测是否需要调用函数if response_message.tool_calls:# 调用函数available_functions = {"ask_database": ask_database} # only one function test in this example, but you can have multiplefunction_name = response_message.tool_calls[0].function.namefuction_to_call = available_functions[function_name]function_args = json.loads(response_message.tool_calls[0].function.arguments)function_response = fuction_to_call(query=function_args.get("query"),)return function_responseif __name__ == '__main__':query = "select count(*) from emp"a = ask_database(query)print(a)