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

,
Jun 13, 2023
在 Github 上打开

本笔记本介绍了如何使用 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')]))

我们可以强制模型使用特定函数,例如 get_n_day_weather_forecast,通过使用 function_call 参数。通过这样做,我们强制模型对如何使用它做出假设。

# 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:使用消息列表调用 chat completions 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 和函数进行知识检索,以便与知识库进行对话式交互。