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

本笔记涵盖如何在使用聊天完成API时结合外部函数来扩展GPT模型的功能。

tools 是聊天完成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="I need to know your location to provide you with the current weather. Could you please specify the city and state (or country) you're in?", 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_Dn2RJJSxzDm49vlVTehseJ0k', 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='Please specify the number of days (x) for which you want the weather forecast for Glasgow, Scotland.', 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_Yg5ydH9lHhLjjYQyXbNvh004', 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。这样做会强制模型对如何使用该函数做出假设。

# 在这个单元格中，我们强制模型使用 `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_aP8ZEtGcyseL0btTMYxTCKbk', function=Function(arguments='{"location":"Toronto, Canada","format":"celsius","num_days":1}', name='get_n_day_weather_forecast'), type='function')])

# 如果我们不强制模型使用 `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
)
chat_response.choices[0].message

ChatCompletionMessage(content=None, role='assistant', function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_5HqCVRaAoBuU0uTlO3MUwaWX', function=Function(arguments='{"location": "Toronto, Canada", "format": "celsius"}', name='get_current_weather'), type='function'), ChatCompletionMessageToolCall(id='call_C9kCha28xHEsxYl4PxZ1l5LI', 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="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_pFdKcCu5taDTtOOfX14vEDRp', function=Function(arguments='{"location": "San Francisco, CA", "format": "fahrenheit", "num_days": 4}', name='get_n_day_weather_forecast'), type='function'),
 ChatCompletionMessageToolCall(id='call_Veeyp2hYJOKp0wT7ODxmTjaS', function=Function(arguments='{"location": "Glasgow, UK", "format": "celsius", "num_days": 4}', name='get_n_day_weather_forecast'), type='function')]

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

在下一个示例中，我们将演示如何执行那些输入是模型生成的函数，并利用这一点来实现一个代理，可以帮助我们回答关于数据库的问题。为简单起见，我们将使用Chinook示例数据库。

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

指定执行SQL查询的函数

首先让我们定义一些有用的实用函数，用于从SQLite数据库中提取数据。

import sqlite3

conn = sqlite3.connect("data/Chinook.db")
print("Opened database successfully")

Opened database successfully

def get_table_names(conn):
    """返回一个表名列表。"""
    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):
    """返回列名列表。"""
    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):
    """返回一个包含字典的列表，每个字典包含数据库中每个表的表名和列信息。"""
    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):
    """提供SQL查询以查询SQLite数据库的函数。"""
    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以获取响应。

# 步骤 #1: 提供可能触发函数调用的内容。在这种情况下，模型可以识别用户请求的信息可能在传递给模型的工具描述中的数据库模式中可用。
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"
)

# 将消息追加到消息列表中
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_bXMf903yO78sdsMZble4yu90', function=Function(arguments='{"query":"SELECT A.Title, COUNT(T.TrackId) AS TrackCount FROM Album A JOIN Track T ON A.AlbumId = T.AlbumId GROUP BY A.Title ORDER BY TrackCount DESC LIMIT 1;"}', name='ask_database'), type='function')])

# 步骤2：判断模型的响应中是否包含工具调用。   
tool_calls = response_message.tool_calls
if tool_calls:
    # 如果模型判断正确，它将返回要调用的工具/函数的名称以及参数。  
    tool_call_id = tool_calls[0].id
    tool_function_name = tool_calls[0].function.name
    tool_query_string = eval(tool_calls[0].function.arguments)['query']
    
    # 步骤3：调用函数并获取结果。将结果追加到消息列表中。      
    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
        })
        
        # 步骤4：调用聊天补全API，并将函数响应附加到消息列表中
        # 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,
        )  # 从模型中获取一个新的响应，其中模型可以看到函数响应
        print(model_response_with_function_call.choices[0].message.content)
    else: 
        print(f"Error: function {tool_function_name} does not exist")
else: 
    # 模型未识别到要调用的函数，结果可返回给用户。 
    print(response_message.content) 

The album with the most tracks is titled "Greatest Hits," and it contains 57 tracks.

下一步

查看我们的另一个笔记本，演示如何使用Chat Completions API和用于知识检索的函数与知识库进行对话交互。