2023年6月13日

如何使用聊天模型调用函数

,

本笔记本介绍如何将Chat Completions API与外部函数结合使用,以扩展GPT模型的能力。

tools 是Chat Completion API中的一个可选参数,可用于提供函数规范。其目的是使模型能够生成符合所提供规范的函数参数。请注意,API实际上不会执行任何函数调用。开发者需要根据模型输出来执行函数调用。

tools参数中,如果提供了functions参数,默认情况下模型会自行决定何时适合使用某个函数。可以通过将tool_choice参数设置为{"type": "function", "function": {"name": "my_function"}}来强制API使用特定函数。也可以通过将tool_choice参数设置为"none"来强制API不使用任何函数。如果使用了函数,响应输出中将包含"finish_reason": "tool_calls"以及一个tool_calls对象,该对象包含函数名称和生成的函数参数。

概述

本笔记本包含以下2个部分:

  • 如何生成函数参数: 指定一组函数并使用API来生成函数参数。
  • 如何使用模型生成的参数调用函数: 通过实际执行带有模型生成参数的函数来闭环流程。
!pip install scipy --quiet
!pip install tenacity --quiet
!pip install tiktoken --quiet
!pip install termcolor --quiet
!pip install openai --quiet
import json
from openai import OpenAI
from tenacity import retry, wait_random_exponential, stop_after_attempt
from termcolor import colored  

GPT_MODEL = "gpt-4o"
client = OpenAI()

实用工具

首先,我们定义一些实用工具,用于调用Chat Completions API以及维护和跟踪对话状态。

@retry(wait=wait_random_exponential(multiplier=1, max=40), stop=stop_after_attempt(3))
def chat_completion_request(messages, tools=None, tool_choice=None, model=GPT_MODEL):
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=tools,
            tool_choice=tool_choice,
        )
        return response
    except Exception as e:
        print("Unable to generate ChatCompletion response")
        print(f"Exception: {e}")
        return e
def pretty_print_conversation(messages):
    role_to_color = {
        "system": "red",
        "user": "green",
        "assistant": "blue",
        "function": "magenta",
    }
    
    for message in messages:
        if message["role"] == "system":
            print(colored(f"system: {message['content']}\n", role_to_color[message["role"]]))
        elif message["role"] == "user":
            print(colored(f"user: {message['content']}\n", role_to_color[message["role"]]))
        elif message["role"] == "assistant" and message.get("function_call"):
            print(colored(f"assistant: {message['function_call']}\n", role_to_color[message["role"]]))
        elif message["role"] == "assistant" and not message.get("function_call"):
            print(colored(f"assistant: {message['content']}\n", role_to_color[message["role"]]))
        elif message["role"] == "function":
            print(colored(f"function ({message['name']}): {message['content']}\n", role_to_color[message["role"]]))

基本概念

让我们创建一些函数规范来与一个假设的天气API进行交互。我们将把这些函数规范传递给Chat Completions API,以生成符合规范的函数参数。

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Get the current weather",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "The city and state, e.g. San Francisco, CA",
                    },
                    "format": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "The temperature unit to use. Infer this from the users location.",
                    },
                },
                "required": ["location", "format"],
            },
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_n_day_weather_forecast",
            "description": "Get an N-day weather forecast",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "The city and state, e.g. San Francisco, CA",
                    },
                    "format": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "The temperature unit to use. Infer this from the users location.",
                    },
                    "num_days": {
                        "type": "integer",
                        "description": "The number of days to forecast",
                    }
                },
                "required": ["location", "format", "num_days"]
            },
        }
    },
]

如果我们向模型询问当前天气情况,它会回应一些澄清性问题。

messages = []
messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."})
messages.append({"role": "user", "content": "What's the weather like today"})
chat_response = chat_completion_request(
    messages, tools=tools
)
assistant_message = chat_response.choices[0].message
messages.append(assistant_message)
assistant_message
ChatCompletionMessage(content='Sure, can you please provide me with the name of your city and state?', role='assistant', function_call=None, tool_calls=None)

一旦我们提供了缺失的信息,它将为我们生成适当的函数参数。

messages.append({"role": "user", "content": "I'm in Glasgow, Scotland."})
chat_response = chat_completion_request(
    messages, tools=tools
)
assistant_message = chat_response.choices[0].message
messages.append(assistant_message)
assistant_message
ChatCompletionMessage(content=None, role='assistant', function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_xb7QwwNnx90LkmhtlW0YrgP2', function=Function(arguments='{"location":"Glasgow, Scotland","format":"celsius"}', name='get_current_weather'), type='function')])

通过不同的提示方式,我们可以让它针对我们告诉它的另一个函数进行操作。

messages = []
messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."})
messages.append({"role": "user", "content": "what is the weather going to be like in Glasgow, Scotland over the next x days"})
chat_response = chat_completion_request(
    messages, tools=tools
)
assistant_message = chat_response.choices[0].message
messages.append(assistant_message)
assistant_message
ChatCompletionMessage(content='To provide you with the weather forecast for Glasgow, Scotland, could you please specify the number of days you would like the forecast for?', role='assistant', function_call=None, tool_calls=None)

模型再次要求我们澄清,因为它目前掌握的信息还不够充分。在这个例子中,它已经知道天气预报的地点,但还需要了解预报需要多少天的数据。

messages.append({"role": "user", "content": "5 days"})
chat_response = chat_completion_request(
    messages, tools=tools
)
chat_response.choices[0]
Choice(finish_reason='tool_calls', index=0, logprobs=None, message=ChatCompletionMessage(content=None, role='assistant', function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_34PBraFdNN6KR95uD5rHF8Aw', function=Function(arguments='{"location":"Glasgow, Scotland","format":"celsius","num_days":5}', name='get_n_day_weather_forecast'), type='function')]))

我们可以通过使用function_call参数强制模型调用特定函数,例如get_n_day_weather_forecast。通过这种方式,我们迫使模型对如何使用该函数做出假设。

# in this cell we force the model to use get_n_day_weather_forecast
messages = []
messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."})
messages.append({"role": "user", "content": "Give me a weather report for Toronto, Canada."})
chat_response = chat_completion_request(
    messages, tools=tools, tool_choice={"type": "function", "function": {"name": "get_n_day_weather_forecast"}}
)
chat_response.choices[0].message
ChatCompletionMessage(content=None, role='assistant', function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_FImGxrLowOAOszCaaQqQWmEN', function=Function(arguments='{"location":"Toronto, Canada","format":"celsius","num_days":7}', name='get_n_day_weather_forecast'), type='function')])
# if we don't force the model to use get_n_day_weather_forecast it may not
messages = []
messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."})
messages.append({"role": "user", "content": "Give me a weather report for Toronto, Canada."})
chat_response = chat_completion_request(
    messages, tools=tools
)
chat_response.choices[0].message
ChatCompletionMessage(content=None, role='assistant', function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_n84kYFqjNFDPNGDEnjnrd2KC', function=Function(arguments='{"location": "Toronto, Canada", "format": "celsius"}', name='get_current_weather'), type='function'), ChatCompletionMessageToolCall(id='call_AEs3AFhJc9pn42hWSbHTaIDh', function=Function(arguments='{"location": "Toronto, Canada", "format": "celsius", "num_days": 3}', name='get_n_day_weather_forecast'), type='function')])

我们也可以强制模型完全不使用某个函数。通过这种方式,我们可以阻止它生成正确的函数调用。

messages = []
messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."})
messages.append({"role": "user", "content": "Give me the current weather (use Celcius) for Toronto, Canada."})
chat_response = chat_completion_request(
    messages, tools=tools, tool_choice="none"
)
chat_response.choices[0].message
ChatCompletionMessage(content="Sure, I'll get the current weather for Toronto, Canada in Celsius.", role='assistant', function_call=None, tool_calls=None)

并行函数调用

较新的模型如gpt-4o或gpt-3.5-turbo可以在单次交互中调用多个函数。

messages = []
messages.append({"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."})
messages.append({"role": "user", "content": "what is the weather going to be like in San Francisco and Glasgow over the next 4 days"})
chat_response = chat_completion_request(
    messages, tools=tools, model=GPT_MODEL
)

assistant_message = chat_response.choices[0].message.tool_calls
assistant_message
[ChatCompletionMessageToolCall(id='call_ObhLiJwaHwc3U1KyB4Pdpx8y', function=Function(arguments='{"location": "San Francisco, CA", "format": "fahrenheit", "num_days": 4}', name='get_n_day_weather_forecast'), type='function'),
 ChatCompletionMessageToolCall(id='call_5YRgeZ0MGBMFKE3hZiLouwg7', function=Function(arguments='{"location": "Glasgow, SCT", "format": "celsius", "num_days": 4}', name='get_n_day_weather_forecast'), type='function')]

如何调用带有模型生成参数的函数

在下一个示例中,我们将演示如何执行输入由模型生成的函数,并利用这一功能实现一个能为我们解答数据库相关问题的智能体。为简化起见,我们将使用Chinook示例数据库

注意: 在生产环境中,SQL生成可能存在高风险,因为模型在生成正确的SQL方面并非完全可靠。

import sqlite3

conn = sqlite3.connect("data/Chinook.db")
print("Opened database successfully")
Opened database successfully
def get_table_names(conn):
    """Return a list of table names."""
    table_names = []
    tables = conn.execute("SELECT name FROM sqlite_master WHERE type='table';")
    for table in tables.fetchall():
        table_names.append(table[0])
    return table_names


def get_column_names(conn, table_name):
    """Return a list of column names."""
    column_names = []
    columns = conn.execute(f"PRAGMA table_info('{table_name}');").fetchall()
    for col in columns:
        column_names.append(col[1])
    return column_names


def get_database_info(conn):
    """Return a list of dicts containing the table name and columns for each table in the database."""
    table_dicts = []
    for table_name in get_table_names(conn):
        columns_names = get_column_names(conn, table_name)
        table_dicts.append({"table_name": table_name, "column_names": columns_names})
    return table_dicts

现在我们可以使用这些实用函数来提取数据库模式的表示。

database_schema_dict = get_database_info(conn)
database_schema_string = "\n".join(
    [
        f"Table: {table['table_name']}\nColumns: {', '.join(table['column_names'])}"
        for table in database_schema_dict
    ]
)

和之前一样,我们将为希望API生成参数的函数定义一个函数规范。请注意,我们正在将数据库架构插入到函数规范中。这对于模型了解相关信息非常重要。

tools = [
    {
        "type": "function",
        "function": {
            "name": "ask_database",
            "description": "Use this function to answer user questions about music. Input should be a fully formed SQL query.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": f"""
                                SQL query extracting info to answer the user's question.
                                SQL should be written using this database schema:
                                {database_schema_string}
                                The query should be returned in plain text, not in JSON.
                                """,
                    }
                },
                "required": ["query"],
            },
        }
    }
]

执行SQL查询

现在让我们实现实际对数据库执行查询的函数。

def ask_database(conn, query):
    """Function to query SQLite database with a provided SQL query."""
    try:
        results = str(conn.execute(query).fetchall())
    except Exception as e:
        results = f"query failed with error: {e}"
    return results
使用Chat Completions API调用函数步骤:

步骤1: 向模型提供可能导致其选择使用工具的内容提示。工具的描述(如函数名称和签名)在'Tools'列表中定义,并通过API调用传递给模型。如果被选中,响应中将包含函数名称和参数。

步骤2: 通过编程方式检查模型是否想要调用函数。如果为真,则继续步骤3。

步骤3: 从响应中提取函数名称和参数,使用参数调用该函数。将结果附加到消息列表中。

步骤4: 使用消息列表调用聊天补全API以获取响应。

# Step #1: Prompt with content that may result in function call. In this case the model can identify the information requested by the user is potentially available in the database schema passed to the model in Tools description. 
messages = [{
    "role":"user", 
    "content": "What is the name of the album with the most tracks?"
}]

response = client.chat.completions.create(
    model='gpt-4o', 
    messages=messages, 
    tools= tools, 
    tool_choice="auto"
)

# Append the message to messages list
response_message = response.choices[0].message 
messages.append(response_message)

print(response_message)
ChatCompletionMessage(content=None, role='assistant', function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_wDN8uLjq2ofuU6rVx1k8Gw0e', function=Function(arguments='{"query":"SELECT Album.Title, COUNT(Track.TrackId) AS TrackCount FROM Album INNER JOIN Track ON Album.AlbumId = Track.AlbumId GROUP BY Album.Title ORDER BY TrackCount DESC LIMIT 1;"}', name='ask_database'), type='function')])
# Step 2: determine if the response from the model includes a tool call.   
tool_calls = response_message.tool_calls
if tool_calls:
    # If true the model will return the name of the tool / function to call and the argument(s)  
    tool_call_id = tool_calls[0].id
    tool_function_name = tool_calls[0].function.name
    tool_query_string = json.loads(tool_calls[0].function.arguments)['query']

    # Step 3: Call the function and retrieve results. Append the results to the messages list.      
    if tool_function_name == 'ask_database':
        results = ask_database(conn, tool_query_string)
        
        messages.append({
            "role":"tool", 
            "tool_call_id":tool_call_id, 
            "name": tool_function_name, 
            "content":results
        })
        
        # Step 4: Invoke the chat completions API with the function response appended to the messages list
        # Note that messages with role 'tool' must be a response to a preceding message with 'tool_calls'
        model_response_with_function_call = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
        )  # get a new response from the model where it can see the function response
        print(model_response_with_function_call.choices[0].message.content)
    else: 
        print(f"Error: function {tool_function_name} does not exist")
else: 
    # Model did not identify a function to call, result can be returned to the user 
    print(response_message.content) 
The album with the most tracks is titled "Greatest Hits," which contains 57 tracks.

下一步

请参阅我们的另一个notebook,其中演示了如何使用Chat Completions API和知识检索功能来与知识库进行对话式交互。