NLP(七十九)函数调用(function_calling)

本文将介绍大模型中的函数调用(function calling),并介绍其在openai, langchain模块中的使用,以及Assistant API对function calling的支持。

函数调用(Function Calling)是OpenAI在今年6月13日对外发布的新能力。根据OpenAI官方博客描述,函数调用能力可以让大模型输出一个请求调用函数的消息,其中包含所需调用的函数信息、以及调用函数时所携带的参数信息。这是一种将大模型(LLM)能力与外部工具/API连接起来的新方式。

比如用户输入:

What’s the weather like in Shanghai?

使用function calling,可实现函数执行get_current_weather(location: string),从而获取函数输出,即得到对应地理位置的天气情况。这其中,location这个参数及其取值是借助大模型能力从用户输入中抽取出来的,同时,大模型判断得到调用的函数为get_current_weather

开发人员可以使用大模型的function calling能力实现:

  • 在进行自然语言交流时,通过调用外部工具回答问题(类似于ChatGPT插件);
  • 将自然语言转换为调用API调用,或数据库查询语句;
  • 从文本中抽取结构化数据
  • 其它

那么,在OpenAI发布的模型中,是如何实现function calling的呢?

本文中,使用的第三方模块信息如下:

1
2
openai==1.3.2
langchain==0.0.339

入门例子

我们以函数get_weather_info为例,其实现逻辑(模拟实现世界中的API调用,获取对应城市的天气状况)如下:

1
2
3
def get_weather_info(city: str):
weather_info = {"Shanghai": "Rainy", "Beijing": "Snow"}
return weather_info.get(city, "Sunny")

该函数只有一个参数:字符串变量city,即城市名称。为了实现function calling功能,需配置函数描述(类似JSON化的API描述),代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
functions = [
{
"name": "get_weather_info",
"description": "Get the weather information of a city",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city, e.g. Shanghai",
},
},
"required": ["city"],
}
}
]

对于一般的用户输入(query),大模型回复结果如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import json
from openai import OpenAI

client = OpenAI(api_key="sk-xxx")

query = "What is the capital of france?"
response = client.chat.completions.create(
model="gpt-3.5-turbo-0613",
messages=[{"role": "user", "content": query}],
functions=functions
)
message = response.dict()["choices"][0]["message"]
print(message)

>>> {'content': 'The capital of France is Paris.', 'role': 'assistant', 'function_call': None, 'tool_calls': None}

此时function_call为None,即大模型判断不需要function calling.

对于查询天气的query,大模型输出结果如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import json
from openai import OpenAI

client = OpenAI(api_key="sk-xxx")

query = "What is the weather like in Beijing?"
response = client.chat.completions.create(
model="gpt-3.5-turbo-0613",
messages=[{"role": "user", "content": query}],
functions=functions
)
message = response.dict()["choices"][0]["message"]
print(message)

>>> {'content': None, 'role': 'assistant', 'function_call': {'arguments': '{\n "city": "Beijing"\n}', 'name': 'get_weather_info'}, 'tool_calls': None}

此时我们看到了令人吃惊的输出,大模型的输出内容为空,而判断需要function calling, 函数名称为get_weather_info,参数为{'arguments': '{\n "city": "Beijing"\n}

下一步,我们可以调用该函数,传入参数,得到函数输出,并再次调用大模型得到答案回复。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
func_name = message["function_call"]["name"]
func_args = json.loads(message["function_call"]["arguments"])
print("func name and args: ", func_name, func_args)
func_response = get_weather_info(**func_args)

final_response = client.chat.completions.create(
model="gpt-3.5-turbo-0613",
messages=[
{"role": "user", "content": query},
{"role": "assistant", "content": None, "function_call": message["function_call"]},
{
"role": "function",
"name": func_name,
"content": func_response
},
],
)
print("answer: ", final_response.dict()["choices"][0]["message"]["content"])

输出结果如下:

1
2
func name and args:  get_weather_info {'city': 'Beijing'}
answer: The weather in Beijing is currently snowy.

以上仅是function calling的简单示例,采用一步一步的详细过程来演示大模型中function calling如何使用。

在实际场景中,我们还需要实现中间过程的函数执行过程。

以下将介绍在OpenAI, LangChain中如何实现function calling。后面我们将使用的3个函数(这些函数仅用于测试,实际场景中可替换为具体的工具或API)如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def get_pizza_info(pizza_name: str):
# get pizza info by pizza name
pizza_info = {
"name": pizza_name,
"price": "10.99"
}
return json.dumps(pizza_info)


def get_weather_info(city: str):
# get city weather info with mock result
weather_info = {"Shanghai": "Rainy", "Beijing": "Snow"}
return weather_info.get(city, "Sunny")


def get_rectangle_area(width: float, length: float):
# calculate the rectangle with given width and length
return f"The area of this rectangle is {width * length}."

openai调用function calling

在OpenAI的官方模块openai中实现function calling的代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
# -*- coding: utf-8 -*-
from openai import OpenAI
import json

client = OpenAI(api_key="sk-xxx")


def get_pizza_info(pizza_name: str):
pizza_info = {
"name": pizza_name,
"price": "10.99"
}
return json.dumps(pizza_info)


def get_weather_info(city: str):
weather_info = {"Shanghai": "Rainy", "Beijing": "Snow"}
return weather_info.get(city, "Sunny")


def get_rectangle_area(width: float, length: float):
return f"The area of this rectangle is {width * length}."


function_mapping = {"get_pizza_info": get_pizza_info,
"get_weather_info": get_weather_info,
"get_rectangle_area": get_rectangle_area}


functions = [
{
"name": "get_pizza_info",
"description": "Get name and price of a pizza of the restaurant",
"parameters": {
"type": "object",
"properties": {
"pizza_name": {
"type": "string",
"description": "The name of the pizza, e.g. Salami",
},
},
"required": ["pizza_name"],
}
},
{
"name": "get_weather_info",
"description": "Get the weather information of a city",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city, e.g. Shanghai",
},
},
"required": ["city"],
}
},
{
"name": "get_rectangle_area",
"description": "Get the area of a rectangle with given width and length",
"parameters": {
"type": "object",
"properties": {
"width": {
"type": "number",
"description": "The width of a rectangle",
},
"length": {
"type": "number",
"description": "The length of a rectangle",
}
},
"required": ["width", "length"],
}
}
]


def chat(query):
response = client.chat.completions.create(
model="gpt-3.5-turbo-0613",
messages=[{"role": "user", "content": query}],
functions=functions
)
message = response.dict()["choices"][0]["message"]
print('message: ', message)

function_call_info = message.get("function_call")
if not function_call_info:
return message
else:
function_name = function_call_info["name"]
arg_name = json.loads(function_call_info["arguments"])
print(f"function name and arg name: ", function_name, arg_name)

if function_name in function_mapping:
function_response = function_mapping[function_name](**arg_name)

final_response = client.chat.completions.create(
model="gpt-3.5-turbo-0613",
messages=[
{"role": "user", "content": query},
{"role": "assistant", "content": None, "function_call": function_call_info},
{
"role": "function",
"name": function_name,
"content": function_response
},
],
)
return final_response.dict()["choices"][0]["message"]
else:
return "wrong function name with function call"


query1 = "What is the capital of france?"
print("answer: ", chat(query1), end="\n\n")

query2 = "How much does pizza Domino cost?"
print("answer: ", chat(query2), end="\n\n")

query3 = "What is the weather like in Beijing?"
print("answer: ", chat(query3), end="\n\n")

query4 = "calculate the rectangle area with width 3 and length 5."
print("answer: ", chat(query4), end="\n\n")

输出结果如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
query:  What is the capital of france?
message: {'content': 'The capital of France is Paris.', 'role': 'assistant', 'function_call': None, 'tool_calls': None}
answer: {'content': 'The capital of France is Paris.', 'role': 'assistant', 'function_call': None, 'tool_calls': None}

query: How much does pizza Domino cost?
message: {'content': None, 'role': 'assistant', 'function_call': {'arguments': '{\n "pizza_name": "Domino"\n}', 'name': 'get_pizza_info'}, 'tool_calls': None}
function name and arg name: get_pizza_info {'pizza_name': 'Domino'}
answer: The cost of a Domino pizza is $10.99.

query: What is the weather like in Beijing?
message: {'content': None, 'role': 'assistant', 'function_call': {'arguments': '{\n "city": "Beijing"\n}', 'name': 'get_weather_info'}, 'tool_calls': None}
function name and arg name: get_weather_info {'city': 'Beijing'}
answer: The weather in Beijing is currently experiencing snow.

query: calculate the rectangle area with width 3 and length 5.
message: {'content': None, 'role': 'assistant', 'function_call': {'arguments': '{\n "width": 3,\n "length": 5\n}', 'name': 'get_rectangle_area'}, 'tool_calls': None}
function name and arg name: get_rectangle_area {'width': 3, 'length': 5}
answer: The area of a rectangle can be calculated by multiplying its width by its length. In this case, the width is 3 and the length is 5. Therefore, the area of the rectangle is 3 * 5 = 15 square units.

LangChain调用function calling

langchain中实现function calling的代码相对简洁写,function calling的结果在Message中的additional_kwargs变量中,实现代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
# -*- coding: utf-8 -*-
import os
import json
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, AIMessage, ChatMessage

os.environ["OPENAI_API_KEY"] = "sk-xxx"


def get_pizza_info(pizza_name: str):
pizza_info = {
"name": pizza_name,
"price": "10.99"
}
return json.dumps(pizza_info)


def get_weather_info(city: str):
weather_info = {"Shanghai": "Rainy", "Beijing": "Snow"}
return weather_info.get(city, "Sunny")


def get_rectangle_area(width: float, length: float):
return f"The area of this rectangle is {width * length}."


function_mapping = {"get_pizza_info": get_pizza_info,
"get_weather_info": get_weather_info,
"get_rectangle_area": get_rectangle_area}


functions = [
{
"name": "get_pizza_info",
"description": "Get name and price of a pizza of the restaurant",
"parameters": {
"type": "object",
"properties": {
"pizza_name": {
"type": "string",
"description": "The name of the pizza, e.g. Salami",
},
},
"required": ["pizza_name"],
}
},
{
"name": "get_weather_info",
"description": "Get the weather information of a city",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city, e.g. Shanghai",
},
},
"required": ["city"],
}
},
{
"name": "get_rectangle_area",
"description": "Get the area of a rectangle with given width and length",
"parameters": {
"type": "object",
"properties": {
"width": {
"type": "number",
"description": "The width of a rectangle",
},
"length": {
"type": "number",
"description": "The length of a rectangle",
}
},
"required": ["width", "length"],
}
}
]


def chat(query):
llm = ChatOpenAI(model="gpt-3.5-turbo-0613")
message = llm.predict_messages(
[HumanMessage(content=query)], functions=functions
)
print('message: ', message, type(message))

function_call_info = message.additional_kwargs.get("function_call", None)
if not function_call_info:
return message
else:
function_name = function_call_info["name"]
arg_name = json.loads(function_call_info["arguments"])
print(f"function name and arg name: ", function_name, arg_name)

if function_name in function_mapping:
function_response = function_mapping[function_name](**arg_name)

final_response = llm.predict_messages(
[
HumanMessage(content=query),
AIMessage(content=str(message.additional_kwargs)),
ChatMessage(
role="function",
additional_kwargs={
"name": function_name
},
content=function_response
),
]
)
return final_response.content
else:
return "wrong function name with function call"


query1 = "What is the capital of Japan?"
print("answer:", chat(query1), end='\n\n')

query2 = "How much does pizza Domino cost?"
print("answer:", chat(query2), end='\n\n')

query3 = "What is the weather like in Paris?"
print("answer: ", chat(query3), end="\n\n")

query4 = "calculate the rectangle area with width 3 and length 10"
print("answer: ", chat(query4), end="\n\n")

OpenAI Assistant API支持function calling

Assistant API是OpenAI在今年OpenAI开发者大会中提出的创新功能。Assistants API允许用户在自己的应用程序中构建AI助手。助手有指令,可以利用模型、工具和知识来响应用户查询。Assistants API目前支持三种类型的工具:代码解释器Code Interpreter)、检索Retrieval)和函数调用Function Calling)。

我们来看看,在openai中的Assistant API如何支持function calling。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
import time
import json
from openai import OpenAI


def get_pizza_info(pizza_name: str):
pizza_info = {
"name": pizza_name,
"price": "10.99"
}
return json.dumps(pizza_info)


def get_weather_info(city: str):
weather_info = {"Shanghai": "Rainy", "Beijing": "Snow"}
return weather_info.get(city, "Sunny")


def get_rectangle_area(width: float, length: float):
return f"The area of this rectangle is {width * length}."


function_mapping = {"get_pizza_info": get_pizza_info,
"get_weather_info": get_weather_info,
"get_rectangle_area": get_rectangle_area}


functions = [
{
"name": "get_pizza_info",
"description": "Get name and price of a pizza of the restaurant",
"parameters": {
"type": "object",
"properties": {
"pizza_name": {
"type": "string",
"description": "The name of the pizza, e.g. Salami",
},
},
"required": ["pizza_name"],
}
},
{
"name": "get_weather_info",
"description": "Get the weather information of a city",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city, e.g. Shanghai",
},
},
"required": ["city"],
}
},
{
"name": "get_rectangle_area",
"description": "Get the area of a rectangle with given width and length",
"parameters": {
"type": "object",
"properties": {
"width": {
"type": "number",
"description": "The width of a rectangle",
},
"length": {
"type": "number",
"description": "The length of a rectangle",
}
},
"required": ["width", "length"],
}
}
]


client = OpenAI(api_key="sk-xxx")

assistant = client.beta.assistants.create(
name="assistant test",
instructions="You are a helpful assistant, ready to answer user's questions.",
model="gpt-3.5-turbo-0613",
)
MATH_ASSISTANT_ID = assistant.id
print("assistant id: ", MATH_ASSISTANT_ID)

assistant = client.beta.assistants.update(
MATH_ASSISTANT_ID,
tools=[
{"type": "function", "function": function} for function in functions
],
)
print("assistant id: ", assistant.id)


def submit_message(assistant_id, thread, user_message):
client.beta.threads.messages.create(
thread_id=thread.id, role="user", content=user_message
)
return client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant_id,
)


def create_thread_and_run(user_input):
thread = client.beta.threads.create()
run = submit_message(MATH_ASSISTANT_ID, thread, user_input)
return thread, run


def wait_on_run(run, thread):
while run.status == "queued" or run.status == "in_progress":
run = client.beta.threads.runs.retrieve(
thread_id=thread.id,
run_id=run.id,
)
time.sleep(0.5)
return run


def get_response(thread):
return client.beta.threads.messages.list(thread_id=thread.id, order="asc")


# Pretty printing helper
def pretty_print(messages):
print("# Messages")
for m in messages:
print(f"{m.role}: {m.content[0].text.value}")
print()


query = "How much does pizza Domino cost?"

thread, run = create_thread_and_run(query)
run = wait_on_run(run, thread)

# Extract single tool call
tool_call = run.required_action.submit_tool_outputs.tool_calls[0]
name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
my_response = function_mapping[name](**arguments)
print("function response: ", my_response)

# use function response for rerun
final_run = client.beta.threads.runs.submit_tool_outputs(
thread_id=thread.id,
run_id=run.id,
tool_outputs=[
{
"tool_call_id": tool_call.id,
"output": json.dumps(my_response),
}
],
)

final_run = wait_on_run(final_run, thread)
pretty_print(get_response(thread))

输出结果如下:

1
2
3
4
5
6
assistant id:  asst_ElovRUJRLqBeYk2Gu2CUIiU9
assistant id: asst_ElovRUJRLqBeYk2Gu2CUIiU9
function response: {"name": "Domino", "price": "10.99"}
# Messages
user: How much does pizza Domino cost?
assistant: The pizza Domino from the restaurant costs $10.99.

LangChain Assistant API支持function calling

可以看到在openai模块中,在Assistant API中实现function calling,较为麻烦。而新版的langchain(0.0.339)中已经添加对Assistant API的支持,我们来看看在langchain中如何支持function calling。

实现代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
# -*- coding: utf-8 -*-
# @place: Pudong, Shanghai
# @contact: lianmingjie@shanda.com
# @file: assistant_api_with_functions.py
# @time: 2023/11/23 11:00
import os
import json
from langchain.agents.openai_assistant import OpenAIAssistantRunnable
from langchain.schema.agent import AgentFinish
from langchain.agents import Tool
from langchain.tools import StructuredTool


os.environ["OPENAI_API_KEY"] = "sk-xxx"


def get_pizza_info(pizza_name: str):
pizza_info = {
"name": pizza_name,
"price": "10.99"
}
return json.dumps(pizza_info)


def get_weather_info(city: str):
weather_info = {"Shanghai": "Rainy", "Beijing": "Snow"}
return weather_info.get(city, "Sunny")


def get_rectangle_area(width: float = 1.0, length: float = 1.0):
return f"The area of this rectangle is {width * length}."


function_mapping = {"get_pizza_info": get_pizza_info,
"get_weather_info": get_weather_info,
"get_rectangle_area": get_rectangle_area}


functions = [
{
"name": "get_pizza_info",
"description": "Get name and price of a pizza of the restaurant",
"parameters": {
"type": "object",
"properties": {
"pizza_name": {
"type": "string",
"description": "The name of the pizza, e.g. Salami",
},
},
"required": ["pizza_name"],
}
},
{
"name": "get_weather_info",
"description": "Get the weather information of a city",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city, e.g. Shanghai",
},
},
"required": ["city"],
}
},
{
"name": "get_rectangle_area",
"description": "Get the area of a rectangle with given width and length",
"parameters": {
"type": "object",
"properties": {
"width": {
"type": "number",
"description": "The width of a rectangle",
},
"length": {
"type": "number",
"description": "The length of a rectangle",
}
},
"required": ["width", "length"],
}
}
]

tools = [Tool(name=func["name"],
func=function_mapping[func["name"]],
description=func["description"]) for func in functions[:-1]]

tools.append(StructuredTool.from_function(get_rectangle_area, name="get_rectangle_area", description="Get the area of a rectangle with given width and length"))

agent = OpenAIAssistantRunnable.create_assistant(
name="langchain assistant",
instructions="You are a helpful assistant, ready to answer user's questions.",
tools=tools,
model="gpt-3.5-turbo-0613",
as_agent=True,
)


def execute_agent(agent, tools, input):
tool_map = {tool.name: tool for tool in tools}
response = agent.invoke(input)
while not isinstance(response, AgentFinish):
tool_outputs = []
for action in response:
print(action.tool, action.tool_input)
tool_output = tool_map[action.tool].invoke(action.tool_input)
print(action.tool, action.tool_input, tool_output, end="\n\n")
tool_outputs.append(
{"output": tool_output, "tool_call_id": action.tool_call_id}
)
response = agent.invoke(
{
"tool_outputs": tool_outputs,
"run_id": action.run_id,
"thread_id": action.thread_id,
}
)

return response


query = "How much does pizza Domino cost?"
query = "What is the capital of france?"
query = "What is the weather like in Beijing?"
query = "calculate the rectangle area with width 3 and length 5."

response = execute_agent(agent, tools, {"content": query})
print(response.return_values["output"])

注意,函数get_rectangle_area为多参数输入,因此需使用StructuredTool.

网页Assistant支持function calling

在OpenAI中的官网中,Assistant已经支持function calling.

Assistant加入function calling

总结

本文是这几天来笔者对于function calling的一个总结。原本以为function calling功能简单好用,但在实际的代码实现中,还是有点难度的,尤其是Assistant API出来后,如何加入外部工具显得尤为重要。

本文作为function calling的一个系统性小结,并给出了详细的代码,希望能对读者有所帮助。

参考文献

  1. Function calling and other API updates: https://openai.com/blog/function-calling-and-other-api-updates
  2. OpenAI assistants in LangChain: https://python.langchain.com/docs/modules/agents/agent_types/openai_assistants
  3. Multi-Input Tools in LangChain: https://python.langchain.com/docs/modules/agents/tools/multi_input_tool
  4. examples/Assistants_API_overview_python.ipynb: https://github.com/openai/openai-cookbook/blob/main/examples/Assistants_API_overview_python.ipynb
欢迎关注我的公众号NLP奇幻之旅,原创技术文章第一时间推送。

欢迎关注我的知识星球“自然语言处理奇幻之旅”,笔者正在努力构建自己的技术社区。


NLP(七十九)函数调用(function_calling)
https://percent4.github.io/NLP(七十九)函数调用(function-calling)/
作者
Jclian91
发布于
2024年1月11日
许可协议