OpenAI Function Calling学习笔记02：如何使用聊天模型调用函数

 1import json
 2from openai import OpenAI
 3from tenacity import retry, wait_random_exponential, stop_after_attempt
 4from termcolor import colored
 5import os
 6
 7OPENAI_API_BASE = os.environ.get("OPENAI_API_BASE")
 8OPENAI_API_KEY = os.environ.get("OPEN_AI_API_KEY")
 9MODEL_NAME = os.environ.get("MODEL_NAME")
10
11client = OpenAI(
12    base_url=OPENAI_API_BASE,
13    api_key=OPENAI_API_KEY,
14)

 1from tenacity import retry, wait_random_exponential, stop_after_attempt
 2from termcolor import colored
 3
 4# 设置重试策略：指数随机等待时间，最多重试3次
 5@retry(wait=wait_random_exponential(multiplier=1, max=40), stop=stop_after_attempt(3))
 6def chat_completion_request(messages, tools=None, tool_choice=None, model=MODEL_NAME):
 7    try:
 8        # 调用OpenAI的chat.completions.create方法，生成对话响应
 9        response = client.chat.completions.create(
10            model=model,
11            messages=messages,
12            tools=tools,
13            tool_choice=tool_choice,
14        )
15        return response
16    except Exception as e:
17        # 异常处理，捕获并打印错误信息
18        print("Unable to generate ChatCompletion response")
19        print(f"Exception: {e}")
20        return e
21
22
23def pretty_print_conversation(messages):
24    # 定义不同角色的输出颜色
25    role_to_color = {
26        "system": "red",
27        "user": "green",
28        "assistant": "blue",
29        "function": "magenta",
30    }
31
32    for message in messages:
33        if message["role"] == "system":
34            # 打印系统消息，红色显示
35            print(
36                colored(
37                    f"system: {message['content']}\n", role_to_color[message["role"]]
38                )
39            )
40        elif message["role"] == "user":
41            # 打印用户消息，绿色显示
42            print(
43                colored(f"user: {message['content']}\n", role_to_color[message["role"]])
44            )
45        elif message["role"] == "assistant" and message.get("function_call"):
46            # 打印助手的函数调用信息，蓝色显示
47            print(
48                colored(
49                    f"assistant: {message['function_call']}\n",
50                    role_to_color[message["role"]],
51                )
52            )
53        elif message["role"] == "assistant" and not message.get("function_call"):
54            # 打印助手的普通消息，蓝色显示
55            print(
56                colored(
57                    f"assistant: {message['content']}\n", role_to_color[message["role"]]
58                )
59            )
60        elif message["role"] == "function":
61            # 打印函数调用结果，品红色显示
62            print(
63                colored(
64                    f"function ({message['name']}): {message['content']}\n",
65                    role_to_color[message["role"]],
66                )
67            )

 1def main():
 2    tools = [
 3        {
 4            "type": "function",
 5            "function": {
 6                "name": "get_current_weather",
 7                "description": "Get the current weather",
 8                "parameters": {
 9                    "type": "object",
10                    "properties": {
11                        "location": {
12                            "type": "string",
13                            "description": "The city and state, e.g. San Francisco, CA",
14                        },
15                        "format": {
16                            "type": "string",
17                            "enum": ["celsius", "fahrenheit"],
18                            "description": "The temperature unit to use. Infer this from the users location.",
19                        },
20                    },
21                    "required": ["location", "format"],
22                },
23            },
24        },
25        {
26            "type": "function",
27            "function": {
28                "name": "get_n_day_weather_forecast",
29                "description": "Get an N-day weather forecast",
30                "parameters": {
31                    "type": "object",
32                    "properties": {
33                        "location": {
34                            "type": "string",
35                            "description": "The city and state, e.g. San Francisco, CA",
36                        },
37                        "format": {
38                            "type": "string",
39                            "enum": ["celsius", "fahrenheit"],
40                            "description": "The temperature unit to use. Infer this from the users location.",
41                        },
42                        "num_days": {
43                            "type": "integer",
44                            "description": "The number of days to forecast",
45                        },
46                    },
47                    "required": ["location", "format", "num_days"],
48                },
49            },
50        },
51    ]
52
53
54if __name__ == "__main__":
55    main()

 1    messages = []
 2    messages.append(
 3        {
 4            "role": "system",
 5            "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous.",
 6        }
 7    )
 8    messages.append({"role": "user", "content": "What's the weather like today"})
 9
10    chat_response = chat_completion_request(messages, tools=tools)
11
12    print("finish_reason=", chat_response.choices[0].finish_reason)
13    assistant_message = chat_response.choices[0].message
14    messages.append(assistant_message)
15    print(assistant_message)

1    messages.append({"role": "user", "content": "I'm in Glasgow, Scotland."})
2    chat_response = chat_completion_request(messages, tools=tools)
3    print("finish_reason=", chat_response.choices[0].finish_reason)
4    assistant_message = chat_response.choices[0].message
5    messages.append(assistant_message)
6    print(assistant_message)

1    messages.append({"role": "user", "content": "I'm in Shanghai, China."})
2    chat_response = chat_completion_request(messages, tools=tools)
3    print("finish_reason=", chat_response.choices[0].finish_reason)
4    assistant_message = chat_response.choices[0].message
5    messages.append(assistant_message)
6    print(assistant_message)

 1    messages = []
 2    messages.append(
 3        {
 4            "role": "system",
 5            "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous.",
 6        }
 7    )
 8    messages.append(
 9        {
10            "role": "user",
11            "content": "what is the weather going to be like in Shanghai, China over the next x days",
12        }
13    )
14    chat_response = chat_completion_request(messages, tools=tools)
15    print("finish_reason=", chat_response.choices[0].finish_reason)
16    assistant_message = chat_response.choices[0].message
17    messages.append(assistant_message)
18    print(assistant_message)

 1    # 强制模型使用函数get_n_day_weather_forecast
 2    messages = []
 3    messages.append(
 4        {
 5            "role": "system",
 6            "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous.",
 7        }
 8    )
 9    messages.append(
10        {"role": "user", "content": "Give me a weather report for Toronto, Canada."}
11    )
12    chat_response = chat_completion_request(
13        messages,
14        tools=tools,
15        tool_choice={
16            "type": "function",
17            "function": {"name": "get_n_day_weather_forecast"},
18        },
19    )
20    print("finish_reason=", chat_response.choices[0].finish_reason)
21    assistant_message = chat_response.choices[0].message
22    messages.append(assistant_message)
23    print(assistant_message)

 1messages = []
 2    messages.append(
 3        {
 4            "role": "system",
 5            "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous.",
 6        }
 7    )
 8    messages.append(
 9        {
10            "role": "user",
11            "content": "Give me the current weather (use Celcius) for Toronto, Canada.",
12        }
13    )
14    chat_response = chat_completion_request(messages, tools=tools, tool_choice="none")
15    print("finish_reason=", chat_response.choices[0].finish_reason)
16    assistant_message = chat_response.choices[0].message
17    messages.append(assistant_message)
18    print(assistant_message)

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

1import sqlite3
2
3
4def main():
5    conn = sqlite3.connect("data/chinook.db")
6    print("Opened database successfully")

 1def get_table_names(conn):
 2    """返回数据库中的所有表名列表。"""
 3    table_names = []
 4    tables = conn.execute("SELECT name FROM sqlite_master WHERE type='table';")
 5    for table in tables.fetchall():
 6        table_names.append(table[0])
 7    return table_names
 8
 9
10def get_column_names(conn, table_name):
11    """返回指定表的列名列表。"""
12    column_names = []
13    columns = conn.execute(f"PRAGMA table_info('{table_name}');").fetchall()
14    for col in columns:
15        column_names.append(col[1])
16    return column_names
17
18
19def get_database_info(conn):
20    """返回数据库中每个表的字典列表，包含表名和列名。"""
21    table_dicts = []
22    for table_name in get_table_names(conn):
23        columns_names = get_column_names(conn, table_name)
24        table_dicts.append({"table_name": table_name, "column_names": columns_names})
25    return table_dicts

1    database_schema_dict = get_database_info(conn)
2    database_schema_string = "\n".join(
3        [
4            f"Table: {table['table_name']}\nColumns: {', '.join(table['column_names'])}"
5            for table in database_schema_dict
6        ]
7    )
8    print(database_schema_string)

 1 tools = [
 2        {
 3            "type": "function",
 4            "function": {
 5                "name": "ask_database",
 6                "description": "Use this function to answer user questions about music. Input should be a fully formed SQL query.",
 7                "parameters": {
 8                    "type": "object",
 9                    "properties": {
10                        "query": {
11                            "type": "string",
12                            "description": f"""
13                                SQL query extracting info to answer the user's question.
14                                SQL should be written using this database schema:
15                                {database_schema_string}
16                                The query should be returned in plain text, not in JSON.
17                                """,
18                        }
19                    },
20                    "required": ["query"],
21                },
22            },
23        }
24    ]

1def ask_database(conn, query):
2    """使用提供的 SQL 查询查询 SQLite 数据库的函数"""
3    try:
4        results = str(conn.execute(query).fetchall())
5    except Exception as e:
6        results = f"query failed with error: {e}"
7    return results

 1    # 第一步：使用可能触发函数调用的内容进行提示。
 2    # 在这种情况下，模型可以识别出用户请求的信息，其可能存在于传递给模型的工具描述中的数据库模式中。
 3    messages = [
 4        {
 5            "role": "user",
 6            "content": "What is the name of the album with the most tracks?",
 7        }
 8    ]
 9
10    response = chat_completion_request(
11        messages=messages, tools=tools, tool_choice="auto"
12    )
13    print("finish_reason=", response.choices[0].finish_reason)
14    response_message = response.choices[0].message
15    messages.append(response_message)
16    print(response_message)

 1    # 第2步：确定模型的响应是否包含工具调用。
 2    tool_calls = response_message.tool_calls
 3    if tool_calls:
 4        # 如果为真，模型将返回要调用的工具/函数名称和参数
 5        tool_call_id = tool_calls[0].id
 6        tool_function_name = tool_calls[0].function.name
 7        tool_query_string = json.loads(tool_calls[0].function.arguments)["query"]
 8
 9        # 第3步：调用函数并检索结果。将结果附加到消息列表。
10        if tool_function_name == "ask_database":
11            results = ask_database(conn, tool_query_string)
12
13            messages.append(
14                {
15                    "role": "tool",
16                    "tool_call_id": tool_call_id,
17                    "name": tool_function_name,
18                    "content": results,
19                }
20            )
21
22            # 第4步：调用聊天补全API，将函数响应附加到消息列表中
23            # 注意，角色为“tool”的消息必须是响应前一条带有“tool_calls”的消息
24            model_response_with_function_call = chat_completion_request(
25                messages=messages,
26            )  # 获取模型的新响应，它可以看到函数的响应
27            print(model_response_with_function_call.choices[0].message.content)
28        else:
29            print(f"Error: function {tool_function_name} does not exist")
30    else:
31        # 模型未识别出要调用的函数，结果可以返回给用户
32        print(response_message.content)