1.Function Calling是什么?
1.1 概念
Function calling是一种将LLM(大语言模型Large language model)连接外部工具的能力,LLM经过微调后,可以检测出何时需要调用函数以及需要调用函数的方法名及参数,并返回给调用方以便调用外部函数,然后把调用外部函数的结果再传给LLM,最后LLM给出用户问题的最终响应。
1.2 流程图
下面是一张Function Calling工作的流程图,重点有3个:
- LLM会告诉我们需要调用什么工具函数以及调用工具函数的具体参数
- 调用函数发生在我们的系统里(一般使用反射实现),并把函数的执行情况加入到与LLM交互的历史会话中
- 会有多次跟LLM的交互过程
2. Function Calling的作用?
序号 | 作用 | 内容 | 应用场景 |
---|---|---|---|
1 | 扩展能力 | 能允许LLM与外部函数或者服务交互,极大地扩展了LLM的能力 | 数据库查询、数学计算、第三方API |
2 | 实时性与动态性 | 可以使LLM获取实时的数据,突破LLM的训练数据静态知识的限制 | 股票信息、天气预报 |
3 | 任务自动化 | 能实现任务的自动编排 | 智能家具系统中灯光、温度、湿度等设备的自动调节 |
4 | 自然语言到标准化的转化 | 能识别出需要用到的外部函数和参数,完成自然语言到函数和参数的转换 | 我想知道上海的天气? -> {‘name’: ‘get_city_weather’, ‘arguments’: ‘{“city”: “上海市”}’} |
5 | 业务逻辑集成 | 能让LLM的智能处理能力与企业的业务逻辑紧密结合 | 一家客服公司基于现有的知识沉淀来做智能客服 |
基于Function Calling的上述作用,能极大地推进AI在产品形态上的落地,以往我们总希望AI是无所不能的,期望能给它很少的输入,能完成很复杂的事情,但结果往往事与愿违,这样很难得到我们想要的结果。其实应该把复杂的任务拆得足够简单,每次给到AI清晰的prompt,完成单次的任务,最后所有任务完成了再给出最终的响应,这其实就是Function Call的工作机制,而LLM在其中扮演了任务拆分的重要工作。
特别是在已有系统的基础上,利用已有系统的API作为外部函数,让LLM来实现任务拆分和编排,就能很快地让AI为我们业务赋能。
3. Function Calling怎么用?
很多大模型都在微调之后支持了Function Calling,下面是基于百度千帆的ERNIE-Functions-8K的具体实现
注意:ERNIE-Functions-8K的function calling的能力只有python版本的SDK,没有其他语言的。如果python基础好的,可以直接看官方文档,不过官方文档里没有给出调用函数的过程。
3.1 单函数调用
接下来,我们用Python代码看如何实现用户的问题:“请帮我查询一下数据库中用python撰写的代码文件数量”
3.1.1 首先定义函数get_file_num
如下所示,主要就是通过指定语言获取文件数量,这个函数主要是在LLM给予我们返回参数后调用的
def get_file_num(params: dict) -> str:"""获取数据库中指定语言的代码文件数量"""language_low = params.get('language', '').lower()language_map = {"c/c++": 35,"java": 10,"javascript": 25,"python": 35,"go": 32,}return str(language_map.get(language_low, 0))
3.1.2 定义函数Schema
为了让LLM知道什么时候需要调用函数,以及调用函数名及参数,我们需要定义函数的schema传给LLM
# 定义函数schema
func_list = [{"name": "get_file_num", # 函数名称"description": "获取内部数据库中以某一编程语言编写的文件数量", # 函数描述"parameters": {"type": "object","properties": { # 参数schema,如果参数为空,设为空字典即可"language": { # 参数名称"type": "string", # 参数类型"description": "代码所运用的编程语言,例如:python、c/c++、go、java" # 参数描述}},"required": ["language"] # 必填参数(无默认值)}
}]
3.1.3 第一次调用LLM接口
可以看到,我们把用户的问题(msgs)和函数schema(funclist)传给了LLM
chat_comp = qianfan.Function()msgs = qianfan.QfMessages()msgs.append(query, role='user')resp = chat_comp.do(messages=msgs,functions=func_list)
上述是千帆的SDK,实际调用LLM的Prompt如下:
{"role": "user","content": "接下来的所有对话中,你可以使用外部的工具来回答问题。\n你必须按照规定的格式来使用工具,当你使用工具时,我会在下一轮对话给你工具调用结果,然后你应该根据实际结果判断是否需要进一步使用工具,或给出你的回答。\n工具可能有多个,每个工具由名称、描述、参数组成,参数符合标准的json schema。\n\n下面是工具列表:\n名称:get_file_num\n描述:获取内部数据库中以某一编程语言编写的文件数量\n参数:{\"type\": \"object\", \"properties\": {\"language\": {\"type\": \"string\", \"description\": \"代码所运用的编程语言,例如:python、c/c++、go、java\"}}, \"required\": [\"language\"]}\n\n\n如果你需要使用外部工具,那么你的输出必须按照如下格式,只包含2行,不需要输出任何解释或其他无关内容:\nAction: 使用的工具名称\nAction Input: 使用工具的参数,json格式\n\n如果你不需要使用外部工具,不需要输出Action和Action Input,请输出你的回答。\n你的问题:请帮我查询一下数据库中用python撰写的代码文件数量"}
可以看出,整体是由:千帆system_prompt + 函数schema + 用户问题
3.1.4 解析第一次调用结果
第一次调用LLM,LLM发现要解决用户的问题,需要调用函数,则会返回如下的格式:
{"id": "as-09dj7y7hzz","object": "chat.completion","created": 1736684991,"result": "","is_truncated": false,"need_clear_history": false,"finish_reason": "normal","usage": {"prompt_tokens": 226,"completion_tokens": 17,"total_tokens": 243},"function_call": {"name": "get_file_num","arguments": "{\"language\": \"python\"}"}
}
千帆会把需要调用的函数和参数放到function_call中返回,所以我们用如下代码获得函数及调用的参数:
func_call_result = resp['body'].get('function_call')
3.1.5 调用函数获得结果
如下所示,我们获取需要调用的函数和参数,然后使用globals()反射的方式使用"{“language”: “python”}"作为参数调用函数get_file_num函数得到结果func_resp
if func_call_result:# 获取函数名称、入参func_name = func_call_result["name"]func_param = json.loads(func_call_result["arguments"])func_resp = None# 使用 globals() 动态调用函数if func_name in globals() and callable(globals()[func_name]):func_resp = globals()[func_name](func_param)print("函数" + func_name + "执行结果:"+ func_resp)else:print(f"Function {func_name} does not exist or is not callable.")
3.1.6 把执行结果和调用返回添加到历史会话
我们执行函数得到结果后,需要把执行结果和调用函数名以及参数添加到历史返回中,方便下次调用LLM接口
# 将函数返回值转换成json字符串func_content = json.dumps({"return": func_resp})# 添加到历史对话记录msgs.append(resp, role="assistant")msgs.append(func_content, role="function")
上述代码实际上是把下面的内容加入到了历史会话:
{"role": "assistant","content": "Action: get_file_num\\nAction Input: {\"language\": \"python\"}"},{"role": "user","content": "{\"return\": \"35\"}"}
3.1.7 第二次调用LLM接口
我们得到函数的执行结果后,再次调用LLM接口
# 再次调用chat_completionsecond_resp = chat_comp.do(messages=msgs,functions=func_list)
这里要注意,还需要把函数schema和历史会话、用户问题一起发送给LLM,实际的prompt如下:
{"role": "user","content": "接下来的所有对话中,你可以使用外部的工具来回答问题。\n你必须按照规定的格式来使用工具,当你使用工具时,我会在下一轮对话给你工具调用结果,然后你应该根据实际结果判断是否需要进一步使用工具,或给出你的回答。\n工具可能有多个,每个工具由名称、描述、参数组成,参数符合标准的json schema。\n\n下面是工具列表:\n名称:get_file_num\n描述:获取内部数据库中以某一编程语言编写的文件数量\n参数:{\"type\": \"object\", \"properties\": {\"language\": {\"type\": \"string\", \"description\": \"代码所运用的编程语言,例如:python、c/c++、go、java\"}}, \"required\": [\"language\"]}\n\n\n如果你需要使用外部工具,那么你的输出必须按照如下格式,只包含2行,不需要输出任何解释或其他无关内容:\nAction: 使用的工具名称\nAction Input: 使用工具的参数,json格式\n\n如果你不需要使用外部工具,不需要输出Action和Action Input,请输出你的回答。\n你的问题:请帮我查询一下数据库中用python撰写的代码文件数量"},{"role": "assistant","content": "Action: get_file_num\\nAction Input: {\"language\": \"python\"}"},{"role": "user","content": "{\"return\": \"35\"}"}
3.1.8 得到最后的结果
这时LLM发现不需要再调用函数而能直接得出结果,并且把结果放到了result中,具体格式如下:
{"id": "as-10dw9zdfic","object": "chat.completion","created": 1736684992,"result": "数据库中用Python撰写的代码文件数量为35。","is_truncated": false,"need_clear_history": false,"finish_reason": "normal","usage": {"prompt_tokens": 252,"completion_tokens": 10,"total_tokens": 262}
}
可以看到,这时候没有function_call的字段了,我们只要直接解析result返回就可以:
# 得到答案return second_resp['body']['result']
3.1.9 完整的Python代码
import os
import qianfan
import jsonos.environ["QIANFAN_AK"] = os.getenv("QIANFAN_API_KEY")
os.environ["QIANFAN_SK"] = os.getenv("QIANFAN_SECRET_KEY")# 定义可被调用函数
def get_file_num(params: dict) -> str:"""获取数据库中指定语言的代码文件数量"""language_low = params.get('language', '').lower()language_map = {"c/c++": 35,"java": 10,"javascript": 25,"python": 35,"go": 32,}return str(language_map.get(language_low, 0))# 定义函数schema
func_list = [{"name": "get_file_num", # 函数名称"description": "获取内部数据库中以某一编程语言编写的文件数量", # 函数描述"parameters": {"type": "object","properties": { # 参数schema,如果参数为空,设为空字典即可"language": { # 参数名称"type": "string", # 参数类型"description": "代码所运用的编程语言,例如:python、c/c++、go、java" # 参数描述}},"required": ["language"] # 必填参数(无默认值)}
}]def ai_querying_data(query : str) -> str:"""AI查询数据"""#QianFan Function is an agent for calling QianFan ChatCompletion with function call API.chat_comp = qianfan.Function()msgs = qianfan.QfMessages()msgs.append(query, role='user')# 第一次调用,目的是获取需要调用的方法和参数print("第一次调用msgs:", msgs._msg_list)resp = chat_comp.do(messages=msgs,functions=func_list)print("第一次调用真实的请求:\n", json.dumps(resp.request.json_body, indent=4, ensure_ascii=False))print("第一次调用返回的 resp.body:\n", json.dumps(resp['body'], indent=4))func_call_result = resp['body'].get('function_call')print("function_call:", func_call_result)# 如果存在function_call,则直接实际方法的调用print("******"*100)if func_call_result:# 获取函数名称、入参func_name = func_call_result["name"]func_param = json.loads(func_call_result["arguments"])func_resp = None# 使用 globals() 动态调用函数if func_name in globals() and callable(globals()[func_name]):func_resp = globals()[func_name](func_param)print("函数" + func_name + "执行结果:"+ func_resp)else:print(f"Function {func_name} does not exist or is not callable.")# 将函数返回值转换成json字符串func_content = json.dumps({"return": func_resp})# 添加到历史对话记录msgs.append(resp, role="assistant")msgs.append(func_content, role="function")# 第二次调用,目的是获取最后的结果print("第二次调用msgs:", msgs._msg_list)# 再次调用chat_completionsecond_resp = chat_comp.do(messages=msgs,functions=func_list)print("第二次调用真实的请求:\n", json.dumps(second_resp.request.json_body, indent=4, ensure_ascii=False))print("第二次调用返回的 resp.body:\n", json.dumps(second_resp['body'], indent=4, ensure_ascii=False))# 得到答案return second_resp['body']['result']if __name__ == '__main__':#query = "请帮我查询一下数据库中用java撰写的代码文件数量"query = "请帮我查询一下数据库中用python撰写的代码文件数量"print("最终结果:", ai_querying_data(query))
3.1.10 流程图
3.2 多函数调用
当LLM发现要解决用户的问题,还需要继续用到工具,则会再次返回用户function_call,也就是下面绿色部分可能会执行多次。
我们举例说明下,比如我们的问题变成了:请帮我查询一下数据库中用java撰写的代码文件数量,并且查询哈尔滨的天气
3.2.1 查询天气函数
这时候发现只依赖之前的get_file_num方法不能完全解决这个问题,所以我们先增加一个查询天气的函数:
def get_city_weather(params: dict) -> str:"""获取指定城市的天气"""city_weather_map = {"上海": 15,"北京": 10,"广州": 25,"深圳": 28,"哈尔滨": 5}return str(city_weather_map.get(params.get('city', ''), 0))
3.2.2 函数schema
这时候把上述get_city_weather也在函数schema中声明:
# 定义函数schema
func_list = [{"name": "get_file_num", # 函数名称"description": "获取内部数据库中以某一编程语言编写的文件数量", # 函数描述"parameters": {"type": "object","properties": { # 参数schema,如果参数为空,设为空字典即可"language": { # 参数名称"type": "string", # 参数类型"description": "代码所运用的编程语言,例如:python、c/c++、go、java" # 参数描述}},"required": ["language"] # 必填参数(无默认值)}},{"name": "get_city_weather", # 函数名称"description": "获取指定城市的天气", # 函数描述"parameters": {"type": "object","properties": { # 参数schema,如果参数为空,设为空字典即可"city": { # 参数名称"type": "string", # 参数类型"description": "需要查询的城市,例如:上海、北京、深圳、广州,最后不要带市" # 参数描述}},"required": ["city"] # 必填参数(无默认值)}}
]
3.2.3 循环调用
因为这时候LLM会不止一次返回给我们需要调用外部函数,所以我们要利用如果返回的内容包括function_call则就一直循环调用返回的函数,并把执行结果加入到历史的会话以便下次调用LLM接口,直到返回的结果内容不包含funcion_call,直接获取result作为最终的相应。
def ai_querying_data(query: str) -> str:"""AI查询数据"""chat_comp = qianfan.Function()msgs = qianfan.QfMessages()msgs.append(query, role='user')# 开始循环调用 chat_comp.dowhile True:resp = chat_comp.do(messages=msgs, functions=func_list)func_call_result = resp['body'].get('function_call')# 检查是否有 function_call,如果没有则跳出循环if not func_call_result:break# 调用指定的函数func_name = func_call_result["name"]func_param = json.loads(func_call_result["arguments"])func_resp = Noneif func_name in globals() and callable(globals()[func_name]):func_resp = globals()[func_name](func_param)else:# 如果函数不存在,可以选择退出循环或处理错误break# 将函数返回值转换为 JSON 字符串,并更新对话记录func_content = json.dumps({"return": func_resp})msgs.append(resp, role="assistant")msgs.append(func_content, role="function")i = i + 1# 循环结束后,获取并打印最终结果result = resp['body'].get('result')if result:return resultelse:print("未获取到最终结果。")return None
3.2.4 完整代码
import os
import qianfan
import jsonos.environ["QIANFAN_AK"] = os.getenv("QIANFAN_API_KEY")
os.environ["QIANFAN_SK"] = os.getenv("QIANFAN_SECRET_KEY")# 定义可被调用函数
def get_file_num(params: dict) -> str:"""获取数据库中指定语言的代码文件数量"""language_low = params.get('language', '').lower()language_map = {"c/c++": 35,"java": 10,"javascript": 25,"python": 35,"go": 32,}return str(language_map.get(language_low, 0))# 定义可被调用函数
def get_city_weather(params: dict) -> str:"""获取指定城市的天气"""city_weather_map = {"上海": 15,"北京": 10,"广州": 25,"深圳": 28,"哈尔滨": 5}return str(city_weather_map.get(params.get('city', ''), 0))# 定义函数schema
func_list = [{"name": "get_file_num", # 函数名称"description": "获取内部数据库中以某一编程语言编写的文件数量", # 函数描述"parameters": {"type": "object","properties": { # 参数schema,如果参数为空,设为空字典即可"language": { # 参数名称"type": "string", # 参数类型"description": "代码所运用的编程语言,例如:python、c/c++、go、java" # 参数描述}},"required": ["language"] # 必填参数(无默认值)}},{"name": "get_city_weather", # 函数名称"description": "获取指定城市的天气", # 函数描述"parameters": {"type": "object","properties": { # 参数schema,如果参数为空,设为空字典即可"city": { # 参数名称"type": "string", # 参数类型"description": "需要查询的城市,例如:上海、北京、深圳、广州,最后不要带市" # 参数描述}},"required": ["city"] # 必填参数(无默认值)}}
]def ai_querying_data(query: str) -> str:"""AI查询数据"""chat_comp = qianfan.Function()msgs = qianfan.QfMessages()msgs.append(query, role='user')# 开始循环调用 chat_comp.doi = 1;while True:print("第" + str(i) + "次调用msgs:", msgs._msg_list)resp = chat_comp.do(messages=msgs, functions=func_list)func_call_result = resp['body'].get('function_call')print("第" + str(i) + "次调用真实的请求:\n", json.dumps(resp.request.json_body, indent=4, ensure_ascii=False))print("第" + str(i) + "次调用返回的 resp.body:\n", json.dumps(resp['body'], indent=4, ensure_ascii=False))print("第" + str(i) + "次function_call:", func_call_result)# 检查是否有 function_call,如果没有则跳出循环if not func_call_result:break# 调用指定的函数func_name = func_call_result["name"]func_param = json.loads(func_call_result["arguments"])func_resp = Noneif func_name in globals() and callable(globals()[func_name]):func_resp = globals()[func_name](func_param)else:print(f"Function {func_name} does not exist or is not callable.")# 如果函数不存在,可以选择退出循环或处理错误break# 将函数返回值转换为 JSON 字符串,并更新对话记录func_content = json.dumps({"return": func_resp})msgs.append(resp, role="assistant")msgs.append(func_content, role="function")i = i + 1# 循环结束后,获取并打印最终结果result = resp['body'].get('result')if result:return resultelse:print("未获取到最终结果。")return Noneif __name__ == '__main__':query = "请帮我查询一下数据库中用java撰写的代码文件数量,并且查询哈尔滨的天气"print("最终结果:", ai_querying_data(query))
4. 需要注意的点
序号 | 问题 | 内容 |
---|---|---|
1 | 性能 | 复杂任务会存在多次外部工具调用,只能串行调用 |
2 | 程序健壮性 | 调度交给了LLM,可能会死循环,记得加中止条件 |
5. 总结
Function Calling极大地扩展了LLM的能力,特别是在LLM在实际业务场景落地时能发挥巨大作用,比如把系统不同模块的API(权限、客服、主业务、数据库的知识沉淀、第三方API等)定义到函数schema中,以智能机器人的方式向外提供服务,用户只需要输入自然语言,系统就能自动匹配出用户需要使用的模块API,并完成任务编排,最后直接给到用户最终的答案。
欢迎留言说出AI在你们业务场景中的应用,一起讨论!