Python 基于 Chat Completions API 实现外部函数调用
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)
输出结果: