打造專屬ChatGPT：透過OpenAI Tool Calling深度整合資源

from pydantic import BaseModel
from enum import Enum


class DailyRevenue(BaseModel):
    class Config:
        extra = "forbid"


def get_daily_revenue(params=None):
    # 用於tool call呼叫，在資料庫中查詢當日店鋪收益

    return {
        "revenue": 15_000_000,
        "currency": "NT$",
        "memo": "當日匯率1美元兌換31.5元台幣",
    }


class DayOfTheWeek(str, Enum):
    SUN = "SUN"
    MON = "MON"
    TUE = "TUE"
    WED = "WED"
    THU = "THU"
    FRI = "FRI"
    SAT = "SAT"


class HistoricalPedestrianFlow(BaseModel):
    day_of_the_week: DayOfTheWeek
    # = Field(
    #    title="一週中的某一天", description="一週中的某一天"
    # )

    class Config:
        extra = "forbid"


def get_historical_pedestrian_flow(params: HistoricalPedestrianFlow):
    # 用於tool call呼叫，查詢每週的某一天的歷史人流
    return {
        "day_of_the_week": params.day_of_the_week,
        "historical_pedestrian_flow": 300,
    }


def tool_name_mapping(tool_name, json_text: str = ""):
    # 將 tool_name與tool function對應
    if tool_name == "get_daily_revenue":
        fn = get_daily_revenue
        params = DailyRevenue.model_validate_json(json_text)
    elif tool_name == "get_historical_pedestrian_flow":
        fn = get_historical_pedestrian_flow
        params = HistoricalPedestrianFlow.model_validate_json(json_text)

    return fn(params)

{
    "type": "function",
    "function" : {
        "name": string，Required，function(tool)的名稱,
        "description": string，Optional，描述function(tool)的用途，讓模型知道使用時機
        "parameters": JSON Schema,
        "strict": "boolean or null，Optional，Defaults to false，控制是否嚴格遵守papameters的設定(只支援部分的JSON Schema)"
    }
}

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_daily_revenue",
            "description": "在資料庫中查詢當日店鋪收益",
            "parameters": DailyRevenue.model_json_schema(),
            "strict": True,
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_historical_pedestrian_flow",
            "description": "查詢每週的某一天的歷史人流",
            "parameters": HistoricalPedestrianFlow.model_json_schema(),
            "strict": True,
        },
    },
]

from openai import OpenAI

client = OpenAI()

messages = [
    {"role": "system", "content": "你是店家小幫手，使用繁體中文"},
    {
        "role": "user",
        "content": f"不知道今天賺了多少，明天(星期四)歷史人流大概有多少",
    },
]
completion = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="auto",
)

choice = completion.choices[0]

finish_reason = choice.finish_reason
print(f"finish_reason {finish_reason}")

import json

tool_call_assistant = []
tool_messages = []
for tool in choice.message.tool_calls:
    tool_res = tool_name_mapping(tool.function.name, tool.function.arguments)

    tool_call_assistant.append(
        {
            "id": tool.id,
            "type": "function",
            "function": {
                "name": tool.function.name,
                "arguments": tool.function.arguments,
            },
        }
    )
    tool_messages.append(
        {"role": "tool", "content": json.dumps(tool_res), "tool_call_id": tool.id}
    )

messages.append(
    {
        "role": "assistant",
        "tool_calls": tool_call_assistant,
    }
)
messages.extend(tool_messages)

messages.append({"role": "system", "content": "將資訊轉換為美元"})

completion = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="auto",
)
print(completion.choices[0].message.content)
# 今天的店鋪收益是15,000,000台幣，大約相當於476,190美元（根據當日匯率1美金兌換31.5台幣）。

# 至於明天（星期四）的歷史人流，大約有300人。

# Check if the conversation was too long for the context window
if response['choices'][0]['message']['finish_reason'] == "length":
    print("Error: The conversation was too long for the context window.")
    # Handle the error as needed, e.g., by truncating the conversation or asking for clarification
    handle_length_error(response)
    
# Check if the model's output included copyright material (or similar)
if response['choices'][0]['message']['finish_reason'] == "content_filter":
    print("Error: The content was filtered due to policy violations.")
    # Handle the error as needed, e.g., by modifying the request or notifying the user
    handle_content_filter_error(response)
    
# Check if the model has made a tool_call. This is the case either if the "finish_reason" is "tool_calls" or if the "finish_reason" is "stop" and our API request had forced a function call
if (response['choices'][0]['message']['finish_reason'] == "tool_calls" or 
    # This handles the edge case where if we forced the model to call one of our functions, the finish_reason will actually be "stop" instead of "tool_calls"
    (our_api_request_forced_a_tool_call and response['choices'][0]['message']['finish_reason'] == "stop")):
    # Handle tool call
    print("Model made a tool call.")
    # Your code to handle tool calls
    handle_tool_call(response)
    
# Else finish_reason is "stop", in which case the model was just responding directly to the user
elif response['choices'][0]['message']['finish_reason'] == "stop":
    # Handle the normal stop case
    print("Model responded directly to the user.")
    # Your code to handle normal responses
    handle_normal_response(response)
    
# Catch any other case, this is unexpected
else:
    print("Unexpected finish_reason:", response['choices'][0]['message']['finish_reason'])
    # Handle unexpected cases as needed
    handle_unexpected_case(response)