函数调用（function calling）_assistant api和function call

目录

入门例子

openai调用function calling

LangChain调用function calling

OpenAI Assistant API支持function calling

LangChain Assistant API支持function calling

网页Assistant支持function calling

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

比如用户输入:

What’s the weather like in Tokyo?

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

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

• • 在进行自然语言交流时，通过调用外部工具回答问题（类似于ChatGPT插件）；

• • 将自然语言转换为调用API调用，或数据库查询语句；

• • 从文本中抽取结构化数据

• • 其它

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

本文中，使用的第三方模块信息如下：

openai==1.3.2
langchain==0.0.339

入门例子

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

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

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

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），大模型回复结果如下：

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，大模型输出结果如下：

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}。

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

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"])

输出结果如下：

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）如下：

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的代码如下：

# -*- 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")

输出结果如下：

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变量中，实现代码如下：

# -*- 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。

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))

输出结果如下：

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。

实现代码如下：

# -*- 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.