当前位置：首页 > article >正文

Python 基于 Chat Completions API 实现外部函数调用

article 2024/10/24 16:30:55

1. 使用 Deepbricks API中转平台

1.1 使用 curl

curl -x socks5h://127.0.0.1:1080 -X POST https://api.deepbricks.ai/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $API_KEY" \
  -d '{
  "stream": false,
  "model": "GPT-4o-mini",
  "messages": [
    {
      "role": "user",
      "content": "What'\''s the weather like in Boston today?"
    }
  ],
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "get_current_weather",
        "description": "Get the current weather in a given location",
        "parameters": {
          "type": "object",
          "properties": {
            "location": {
              "type": "string",
              "description": "The city and state, e.g. San Francisco, CA"
            },
            "unit": {
              "type": "string",
              "enum": ["celsius", "fahrenheit"]
            }
          },
          "required": ["location"]
        }
      }
    }
  ],
  "tool_choice": "auto"
}'

返回结果：

{"id":"chatcmpl-LZvV5LEnOFzVCScuJnOooKdcBKvOatl0",
"object":"chat.completion",
"created":1729751519,
"model":"gpt-4o-mini",
"system_fingerprint":"fp_8bfc6a7dc2",
"choices":[{"index":0,"finish_reason":"tool_calls","message":{"content":"","role":"assistant","tool_calls":[{"function":{"arguments":"{\"location\":\"Boston, MA\"}","name":"get_current_weather"},
"id":"call_F2m1RvYyJOvLJBivSQyGhrfU",
"type":"function"}]},"logprobs":null}],
"usage":{"prompt_tokens":80,
"completion_tokens":17,
"total_tokens":97}}

1.2 使用 python

from openai import OpenAI
import httpx
import os

API_KEY = os.getenv("API_KEY")
BASE_URL = "https://api.deepbricks.ai/v1/"


proxy_url = 'socks5://127.0.0.1:1080'
proxy_client = httpx.Client(proxies={
    "http://": proxy_url,
    "https://": proxy_url,
})

client = OpenAI(api_key=API_KEY, base_url=BASE_URL,http_client=proxy_client)
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Get the current weather in a given location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "The city and state, e.g. San Francisco, CA",
                    },
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
                },
                "required": ["location"],
            },
        }
    }
]
messages = [{"role": "user", "content": "What's the weather like in Boston today?"}]
completion = client.chat.completions.create(
    model="GPT-4o-mini",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)
print(completion.to_json())
mes = completion.choices[0].message
print(mes)

输出结果：

{
  "id": "chatcmpl-iyceALy2LwmqCScULb0OoKDCbkvoBbR0",
  "choices": [
    {
      "finish_reason": "tool_calls",
      "index": 0,
      "logprobs": null,
      "message": {
        "content": "",
        "role": "assistant",
        "tool_calls": [
          {
            "id": "call_PA7ZgXHGOhyDVnhAz8P1yOEs",
            "function": {
              "arguments": "{\"location\":\"Boston, MA\"}",
              "name": "get_current_weather"
            },
            "type": "function"
          }
        ]
      }
    }
  ],
  "created": 1729751725,
  "model": "gpt-4o-mini",
  "object": "chat.completion",
  "system_fingerprint": "fp_f59a81427f",
  "usage": {
    "completion_tokens": 17,
    "prompt_tokens": 80,
    "total_tokens": 97
  }
}
ChatCompletionMessage(content='', role='assistant', function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='call_PA7ZgXHGOhyDVnhAz8P1yOEs', function=Function(arguments='{"location":"Boston, MA"}', name='get_current_weather'), type='function')])

解析 client.chat.completions.create

          ...
          function_call: Deprecated in favor of `tool_choice`.

              Controls which (if any) function is called by the model. `none` means the model
              will not call a function and instead generates a message. `auto` means the model
              can pick between generating a message or calling a function. Specifying a
              particular function via `{"name": "my_function"}` forces the model to call that
              function.

              `none` is the default when no functions are present. `auto` is the default if
              functions are present.

          functions: Deprecated in favor of `tools`.

              A list of functions the model may generate JSON inputs for.
          ....
          tool_choice: Controls which (if any) tool is called by the model. `none` means the model will
              not call any tool and instead generates a message. `auto` means the model can
              pick between generating a message or calling one or more tools. `required` means
              the model must call one or more tools. Specifying a particular tool via
              `{"type": "function", "function": {"name": "my_function"}}` forces the model to
              call that tool.

              `none` is the default when no tools are present. `auto` is the default if tools
              are present.

          tools: A list of tools the model may call. Currently, only functions are supported as a
              tool. Use this to provide a list of functions the model may generate JSON inputs
              for. A max of 128 functions are supported.

弃用 function_call 与 functions ，以 tool_choice 和 tools 代替
其中 tool_choice 可选参数有：

None
"auto"
{"type": "function", "function": {"name": "my_function"}}

None 表示模型不调用任何工具，而是生成一条消息
“auto” 意味着模型可以在生成消息或调用一个或多个工具之间进行选择
{“type”: “function”, “function”: {“name”: “my_function”}} 模型必须调用一个或多个工具

如果没有可用的工具，tool_choice的默认值是None；如果存在工具，那么默认值是"auto"

2. 集合代码

2.1 强制

import httpx
import os
from tenacity import retry, wait_random_exponential, stop_after_attempt
from termcolor import colored


API_KEY = os.getenv("API_KEY")
BASE_URL = "https://api.deepbricks.ai/v1/chat/completions"
GPT_MODEL = "GPT-4o-mini"#"GPT-4o"#GPT-3.5-turbo"#
proxy_url = 'socks5://127.0.0.1:1080'
proxies={
    "http://": proxy_url,
    "https://": proxy_url,
}


@retry(wait=wait_random_exponential(multiplier=1, max=30), stop=stop_after_attempt(5))
def chat_completion_request(messages, tools=None, tool_choice=None, model=GPT_MODEL):

    headers = {
        "Content-Type": "application/json",
        "Authorization": "Bearer " + API_KEY,
    }

    # 设定请求的JSON数据，包括GPT模型名和要进行补全的消息
    json_data = {"model": model, "messages": messages}

    # 如果传入了tools，将其加入到json_data中
    if tools is not None:
        json_data.update({"tools": tools})

    # 如果传入了tool_choice，将其加入到json_data中
    if tool_choice is not None:
        json_data.update({"tool_choice": tool_choice})


    # 尝试发送POST请求到deepbricks服务器的chat/completions接口
    try:
        response = httpx.post(
            BASE_URL,
            headers=headers,
            json=json_data,
            proxies=proxies)
        # 返回服务器的响应
        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:

        # 如果消息的角色是"system"，则用红色打印“content”
        if message["role"] == "system":
            print(colored(f"system: {message['content']}\n", role_to_color[message["role"]]))

        # 如果消息的角色是"user"，则用绿色打印“content”
        elif message["role"] == "user":
            print(colored(f"user: {message['content']}\n", role_to_color[message["role"]]))

        # 如果消息的角色是"assistant"，并且消息中包含"tool_calls"，则用蓝色打印"tool_calls"
        elif message["role"] == "assistant" and message.get("tool_calls"):
            print(colored(f"assistant[tool_calls]: {message['tool_calls']}\n", role_to_color[message["role"]]))

        # 如果消息的角色是"assistant"，但是消息中不包含"tool_calls"，则用蓝色打印“content”
        elif message["role"] == "assistant" and not message.get("tool_calls"):
            print(colored(f"assistant[content]: {message['content']}\n", role_to_color[message["role"]]))

        # 如果消息的角色是"tools"，则用品红色打印“tools”
        elif message["role"] == "tools":
            print(colored(f"tools ({message['name']}): {message['content']}\n", role_to_color[message["role"]]))


tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Get the current weather in a given location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "The city and state, e.g. San Francisco, CA",
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "The temperature unit to use. Infer this from the users location.",
                    },
                },
                "required": ["location","unit"],
            },
        }

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

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",  # 消息的角色是"user"
    "content": "what is the weather going to be like in Beijing, China over the next x days"
})


chat_response = chat_completion_request(
    messages, tools=tools,tool_choice="auto"

)

messages.append(chat_response.json()["choices"][0]["message"])
messages.append({
    "role": "user",  # 消息的角色是"user"
    "content": "5 days"
})

chat_response = chat_completion_request(
    messages, tools=tools,tool_choice={"type": "function", "function": {"name": "get_n_day_weather_forecast"}}
)

assistant_message = chat_response.json()["choices"][0]["message"]

messages.append(assistant_message)

pretty_print_conversation(messages)

输出结果：
在这里插入图片描述

2.2 auto

import httpx
import os
from tenacity import retry, wait_random_exponential, stop_after_attempt
from termcolor import colored


API_KEY = os.getenv("API_KEY")
BASE_URL = "https://api.deepbricks.ai/v1/chat/completions"
GPT_MODEL = "GPT-4o-mini"#"GPT-4o"#GPT-3.5-turbo"#
proxy_url = 'socks5://127.0.0.1:1080'
proxies={
    "http://": proxy_url,
    "https://": proxy_url,
}


@retry(wait=wait_random_exponential(multiplier=1, max=30), stop=stop_after_attempt(5))
def chat_completion_request(messages, tools=None, tool_choice=None, model=GPT_MODEL):

    headers = {
        "Content-Type": "application/json",
        "Authorization": "Bearer " + API_KEY,
    }

    # 设定请求的JSON数据，包括GPT模型名和要进行补全的消息
    json_data = {"model": model, "messages": messages}

    # 如果传入了tools，将其加入到json_data中
    if tools is not None:
        json_data.update({"tools": tools})

    # 如果传入了tool_choice，将其加入到json_data中
    if tool_choice is not None:
        json_data.update({"tool_choice": tool_choice})


    # 尝试发送POST请求到deepbricks服务器的chat/completions接口
    try:
        response = httpx.post(
            BASE_URL,
            headers=headers,
            json=json_data,
            proxies=proxies)
        # 返回服务器的响应
        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:

        # 如果消息的角色是"system"，则用红色打印“content”
        if message["role"] == "system":
            print(colored(f"system: {message['content']}\n", role_to_color[message["role"]]))

        # 如果消息的角色是"user"，则用绿色打印“content”
        elif message["role"] == "user":
            print(colored(f"user: {message['content']}\n", role_to_color[message["role"]]))

        # 如果消息的角色是"assistant"，并且消息中包含"tool_calls"，则用蓝色打印"tool_calls"
        elif message["role"] == "assistant" and message.get("tool_calls"):
            print(colored(f"assistant[tool_calls]: {message['tool_calls'][0]['function']}\n", role_to_color[message["role"]]))

        # 如果消息的角色是"assistant"，但是消息中不包含"tool_calls"，则用蓝色打印“content”
        elif message["role"] == "assistant" and not message.get("tool_calls"):
            print(colored(f"assistant[content]: {message['content']}\n", role_to_color[message["role"]]))

        # 如果消息的角色是"tools"，则用品红色打印“tools”
        elif message["role"] == "tools":
            print(colored(f"tools ({message['name']}): {message['content']}\n", role_to_color[message["role"]]))


tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "Get the current weather in a given location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "The city and state, e.g. San Francisco, CA",
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "The temperature unit to use. Infer this from the users location.",
                    },
                },
                "required": ["location","unit"],
            },
        }

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

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",  # 消息的角色是"user"
    "content": "What's the weather like today"  # 用户询问今天的天气情况
})

chat_response = chat_completion_request(
    messages, tools=tools,tool_choice="auto"

)

messages.append(chat_response.json()["choices"][0]["message"])
messages.append({
    "role": "user",  # 消息的角色是"user"
    "content": "I'm in Shanghai, China."  # 用户的消息内容
})

chat_response = chat_completion_request(
    messages, tools=tools,tool_choice="auto"
)

messages.append(chat_response.json()["choices"][0]["message"])
messages.append({
    "role": "user",  # 消息的角色是"user"
    "content": "Celsius"  # 用户的消息内容
})

chat_response = chat_completion_request(
    messages, tools=tools,tool_choice="auto"
)

assistant_message = chat_response.json()["choices"][0]["message"]
messages.append(assistant_message)

pretty_print_conversation(messages)