OpenAI Functions Calling 高级用法：多轮对话及精准控制函数调用

概述

上文 OpenAI 函数调用（1） – ChatGPT Plugin 是如何实现的？ 我们提到了 OpenAI Functions Calling（函数调用）功能的基本概念以及使用流程，本文会基于上文内容进行展开，主要包括 2 个方向：

1. 补充 Functions Calling 功能使用中的高级的用法，如多轮对话及精准控制函数调用与否；
2. 补充更多的 Functions Calling 的使用场景，如本地数据库的 SQL 查询；

好的，下面就继续使用查询天气的场景来详细讲解下 Functions Calling 的一些高级功能。

Functions Calling 高级用法

在开始高级用法之前，先简单回顾下 Functions Calling 的使用流程：

1. 声明函数并请求 API：定义当前函数的名称，描述，以及对应的参数信息，并执行 OpenAI 的 Chat 接口；
2. 执行本地函数：接受接口返回，并解析对应的函数参数信息，根据对应的参数信息调用本地函数；
3. 上报执行结果：将本地函数执行的结果上报给 OpenAI Chat 接口进行汇总、总结；

整体的时序图大致如下：

下面会使用查询天气的这个案例来讲解下一些高级用法，首先我们先封装一些基础的函数。

基础函数封装

首先封装一个 chat 函数，可以传入对话内容，functions，function_call 信息，定义大致如下：

@retry(wait=wait_random_exponential(min=1, max=40), stop=stop_after_attempt(3))
def chat_completion_request(messages, functions=None, function_call=None, model=GPT_MODEL):
    headers = {
        "Content-Type": "application/json",
        "Authorization": "Bearer " + openai.api_key,
    }
    json_data = {"model": model, "messages": messages}
    if functions is not None:
        json_data.update({"functions": functions})
    if function_call is not None:
        json_data.update({"function_call": function_call})
    try:
        response = requests.post(
            "https://api.openai.com/v1/chat/completions",
            headers=headers,
            json=json_data,
        )
        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",
    }
    formatted_messages = []
    for message in messages:
        if message["role"] == "system":
            formatted_messages.append(f"system: {message['content']}n")
        elif message["role"] == "user":
            formatted_messages.append(f"user: {message['content']}n")
        elif message["role"] == "assistant" and message.get("function_call"):
            formatted_messages.append(f"assistant: {message['function_call']}n")
        elif message["role"] == "assistant" and not message.get("function_call"):
            formatted_messages.append(f"assistant: {message['content']}n")
        elif message["role"] == "function":
            formatted_messages.append(f"function ({message['name']}): {message['content']}n")
    for formatted_message in formatted_messages:
        print(
            colored(
                formatted_message,
                role_to_color[messages[formatted_messages.index(formatted_message)]["role"]],
            )
        )

其打印效果大致如下，每一种角色都使用不同的颜色打印出来（不用关注具体的打印内容，下文会详细讲解）：

多轮对话

当我们定义的函数需要多个必传字段，并且用户在一轮对话中没有提供足够多的信息的时候，GPT 会询问用户，直到用户提供了必要的信息后才会触发函数调用。

下面还是用查询天气为例来详细说明下，首先需要先定义下对应查询天气的函数：

1. get_current_weather：查询当前的天气情况，需求传入对应的位置信息以及天气的格式；
2. get_n_day_weather_forecast：获取最近 N 天的天气情况，需求传入对应的位置信息、天气的格式以及查询的天数；

函数声明如下：

functions = [
    {
        "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"],
        },
    },
    {
        "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"]
        },
    },
]

但用户询问当前天气，并且没有告知询问哪里的天气时，GPT 回询问需要查询哪里的天气，代码如下：

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, functions=functions
)
assistant_message = chat_response.json()["choices"][0]["message"]
messages.append(assistant_message)
pretty_print_conversation(messages)

对话内容大致如下：

这就是所谓的多轮对话，当需要传入的参数没有提供时，会询问用户对应的信息，直到用户回答了所有的信息。这是 Chat 接口就会返回对应的 function_call 字段，返回内容大致如下：

{
    "role": "assistant",
    "content": null,
    "function_call": {
        "name": "get_current_weather",
        "arguments": "{n  "location": "Glasgow, Scotland",n  "format": "celsius"n}"
    }
}

完整的对话过程大致如下：

我们定义的 get_n_day_weather_forecast 函数也是通用的逻辑，大家可以自行尝试。

精准控制函数调用

上述的对话过程中，我们询问了和天气相关内容并且补全了对应的信息，GPT 最终会调用对应的函数。正常情况下，我们询问无关的问题时并不会出发对应的函数定义，如下对话内容：

我们可以在请求接口的时候使用 function_call 字段强制 GPT 调用对应的函数，代码如下：

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": "hello"})
chat_response = chat_completion_request(
    messages, functions=functions, function_call={"name": "get_n_day_weather_forecast"}
)
assistant_message = chat_response.json()["choices"][0]["message"]
messages.append(assistant_message)
pretty_print_conversation(messages)

至于缺失的参数 GPT 会自动生成，返回的结果大致如下：

{
    //...
    "choices": [
      {
        "index": 0,
        "message": {
          "role": "assistant",
          "content": null,
          "function_call": {
            "name": "get_n_day_weather_forecast",
            "arguments": "{\n  "location": "San Francisco, CA",\n  "format": "celsius",\n  "num_days": 5\n}"
          }
        },
        "finish_reason": "stop"
      }
    ]
  }

完整的对话流程如下：

可知，GPT 自动生成需要的地址（San Francisco, CA）、天气格式（celsius）以及需要查询的天数（3）。

关于始终被调用的这个特性在一些 LLM 应用中将会变得非常实用，我们可以始终要求返回对应的 JSON 结构，而不再是处理一些纯文本对话了。比如，返回 command 和 payload 形式内容，我们就可以很好的去处理对应的逻辑了。当然，这属于函数调用的 hack 用法了，这种场景返回的内容不是参数，而是对应的 Response 了。

精准控制函数不被调用

可以通知 GPT 始终调用函数，同样的也可以设置始终不调用函数，将对应的 function_call 设置为 none 即可。如下：

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, functions=functions, function_call="none"
)
assistant_message = chat_response.json()["choices"][0]["message"]
messages.append(assistant_message)
pretty_print_conversation(messages)

assistant_message 对应的 JOSN 内容如下：

{
"role": "assistant",
"content": "{n "location": "Toronto, Canada",n "format": "celsius"n}"
}

注意，上述并没有调用函数，只是 content 的内容返回的 JSON 格式。对话流程大致如下：

当然，不传入 functions 参数也可以达到同样的效果。

上述的内容都在围绕着 OpenAI 的 Chat 接口如何调用，关于 get_current_weather 与 get_n_day_weather_forecast 函数的具体定义以及执行的并没有展开阐述，这部分其实就是本地的函数定义部分，与具体的业务逻辑有关，与 OpenAI 接口关系不大。

由于查询天气本质上就是网络接口请求，大家应该都比较熟悉，这部不再描述详细的实现细节，下面会以查询本地数据库的场景，来详细阐述下 OpenAI Factions Calling 于本地函数执行的完整逻辑。

案例讲解： SQL 查询

数据库定义

假设现在有一个名为 Chinook.db 的数据库文件，其中包含多张表，每张表中又包含多个字段信息。大致定义如下：

Table: Album
       Columns: AlbumId, Title, ArtistId
Table: Artist
       Columns: ArtistId, Name
Table: Customer
       Columns: CustomerId, FirstName, LastName, Company, Address, City, State, Country, PostalCode, Phone, Fax, Email, SupportRepId
Table: Employee
       Columns: EmployeeId, LastName, FirstName, Title, ReportsTo, BirthDate, HireDate, Address, City, State, Country, PostalCode, Phone, Fax, Email
Table: Genre
       Columns: GenreId, Name
Table: Invoice
       Columns: InvoiceId, CustomerId, InvoiceDate, BillingAddress, BillingCity, BillingState, BillingCountry, BillingPostalCode, Total
Table: InvoiceLine
       Columns: InvoiceLineId, InvoiceId, TrackId, UnitPrice, Quantity
Table: MediaType
       Columns: MediaTypeId, Name
Table: Playlist
       Columns: PlaylistId, Name
Table: PlaylistTrack
       Columns: PlaylistId, TrackId
Table: Track
       Columns: TrackId, Name, AlbumId, MediaTypeId, GenreId, Composer, Milliseconds, Bytes, UnitPrice

想要 GPT 帮我们生成一些查询数据库的参数，必须要告诉 GPT 我们数据库的一些结构信息，否则它并不会生成准确的信息。所以下面就是定义一些基础函数来生成一些数据库的描述信息，并将其传递给 GPT 。

数据信息生成

首先我们需要定义一些函数来生成类似上述的数据库描述信息（表结构），首先定义获取数据库信息的函数，大致代码如下：

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

然后我们将对应的数据描述信息整理成一段可理解的描述信息，代码大致如下：

import sqlite3

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

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

打印出 database_schema_string 信息如下：

Table: Album
Columns: AlbumId, Title, ArtistId
Table: Artist
Columns: ArtistId, Name
Table: Customer
Columns: CustomerId, FirstName, LastName, Company, Address, City, State, Country, PostalCode, Phone, Fax, Email, SupportRepId
Table: Employee
Columns: EmployeeId, LastName, FirstName, Title, ReportsTo, BirthDate, HireDate, Address, City, State, Country, PostalCode, Phone, Fax, Email
Table: Genre
Columns: GenreId, Name
Table: Invoice
Columns: InvoiceId, CustomerId, InvoiceDate, BillingAddress, BillingCity, BillingState, BillingCountry, BillingPostalCode, Total
Table: InvoiceLine
Columns: InvoiceLineId, InvoiceId, TrackId, UnitPrice, Quantity
Table: MediaType
Columns: MediaTypeId, Name
Table: Playlist
Columns: PlaylistId, Name
Table: PlaylistTrack
Columns: PlaylistId, TrackId
Table: Track
Columns: TrackId, Name, AlbumId, MediaTypeId, GenreId, Composer, Milliseconds, Bytes, UnitPrice

有了对应的信息之后，我们就可以声明对应的 functions 了。

Function Calling 声明与定义

想要 GPT 帮我们生成一些查询数据库的参数，必须要告诉 GPT 我们数据库的一些结构信息，对应的定义如下：

1. name：函数名为 ask_database，可以查询数据库的一些信息；
2. description：详细描述下函数的作用以及期望的格式；
3. parameters：函数接受一个名为 query 的参数，将对应的自然语言转换成对应的 SQL 语句，而对 query 的描述就需要添加上对应的数据库信息了；

对应的声明大致如下：

functions = [
    {
        "name": "ask_database",
        "description": "Use this function to answer user questions about music. Output 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"],
        },
    }
]

下面就是需要定义一个 ask_database 这个函数的真实逻辑了。

本地函数定义

下面就是一个查询本地数据库的具体实现，该函数会接受对应的 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

def execute_function_call(message):
    if message["function_call"]["name"] == "ask_database":
        query = json.loads(message["function_call"]["arguments"])["query"]
        results = ask_database(conn, query)
    else:
        results = f"Error: function {message['function_call']['name']} does not exist"
    return results

Chat 完整流程

所有的准备工作都已经完成了，我们看一下完成流程的效果，用户正常对话的过程中会有什么效果。比如我们想要查询歌曲量前 5 的歌手都有哪些，代码如下：

messages = []
messages.append({"role": "system", "content": "Answer user questions by generating SQL queries against the Chinook Music Database."})
messages.append({"role": "user", "content": "Hi, who are the top 5 artists by number of tracks?"})
chat_response = chat_completion_request(messages, functions)
assistant_message = chat_response.json()["choices"][0]["message"]
# 解析对应的结果，并添加到对话历史中
messages.append(assistant_message)
if assistant_message.get("function_call"):
		# 执行本地的数据库查询逻辑
    results = execute_function_call(assistant_message)
		# 将执行结果添加对话历史中
    messages.append({"role": "function", "name": assistant_message["function_call"]["name"], "content": results})

pretty_print_conversation(messages)

这个过程中产生的对话历史如下：

到这里整个环节还没有完成，我还需将上述的对话历史请求 OpenAI 的 Chat 接口，GPT 会根据函数的结果整理成容易理解的自然语言。代码如下：

# 使用上述的对话历史再次请求 Chat 接口
chat_response = chat_completion_request(messages, functions)
# 将 GPT 返回的那天添加到对话历史中
assistant_message = chat_response.json()["choices"][0]["message"]
messages.append(assistant_message)
pretty_print_conversation(messages)

整个过程完整的对话历史如下：

可见，GPT 的确是将数据库的查询结果转换成容易理解的自然语言了。

总结

本文介绍了 Function Calling 的一些高级用法，如：

1. GPT 的多轮对话机制，来询问用户提供必要的关键信息（通过 Prompt 控制）；
2. 强制要求 GPT 使用指定函数，并生成一些参数信息；
3. 也可以要求 GPT 不用使用函数；

为了演示完整真实的使用场景，我们以查询本地数据库的具体案例讲解了 Function Calling 声明的细节（一定要将对应的信息完整的给到GPT，否则它并不能完全理解“参数”传递的合理性），以及函数执行的完整流程。

后面会讲解打造自己的 Plugin 商店的一些思路，可以无缝的将现有的 ChatGPT Plugin 接入到自己的业务中。