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 osAPI_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 modelwill not call a function and instead generates a message. `auto` means the modelcan pick between generating a message or calling a function. Specifying aparticular function via `{"name": "my_function"}` forces the model to call thatfunction.`none` is the default when no functions are present. `auto` is the default iffunctions 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 willnot call any tool and instead generates a message. `auto` means the model canpick between generating a message or calling one or more tools. `required` meansthe model must call one or more tools. Specifying a particular tool via`{"type": "function", "function": {"name": "my_function"}}` forces the model tocall that tool.`none` is the default when no tools are present. `auto` is the default if toolsare present.tools: A list of tools the model may call. Currently, only functions are supported as atool. Use this to provide a list of functions the model may generate JSON inputsfor. 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 coloredAPI_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 edef 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 coloredAPI_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 edef 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)输出结果：
3. 查询数据库
数据库 chinook.db 来自https://www.sqlitetutorial.net/wp-content/uploads/2018/03/chinook.zip
代码：
import httpx
import os
import sqlite3
import json
from tenacity import retry, wait_random_exponential, stop_after_attempt
from termcolor import coloredAPI_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,
}def get_table_names(conn):"""返回一个包含所有表名的列表"""table_names = [] # 创建一个空的表名列表# 执行SQL查询，获取数据库中所有表的名字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):"""返回一个给定表的所有列名的列表"""column_names = [] # 创建一个空的列名列表# 执行SQL查询，获取表的所有列的信息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):"""返回一个字典列表，每个字典包含一个表的名字和列信息"""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 # 返回字典列表conn = sqlite3.connect("../data/chinook.db")
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]
)tools = [{"type": "function","function": {"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"],},}},]def ask_database(conn, query):"""使用 query 来查询 SQLite 数据库的函数。"""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):"""执行函数调用"""# 判断功能调用的名称是否为 "ask_database"if message["tool_calls"][0]["function"]["name"] == "ask_database":# 如果是，则获取功能调用的参数，这里是 SQL 查询#.loads函数的作用是将一个JSON格式的字符串转换成Python能够理解的数据结构，比如字典、列表query = json.loads(message["tool_calls"][0]["function"]["arguments"])["query"]# 使用 ask_database 函数执行查询，并获取结果results = ask_database(conn, query)else:# 如果功能调用的名称不是 "ask_database"，则返回错误信息results = f"Error: function {message['function_call']['name']} does not exist"return results # 返回结果@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 edef pretty_print_conversation(messages):# 为不同角色设置不同的颜色role_to_color = {"system": "red","user": "green","assistant": "blue","tool": "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"] == "tool":print(colored(f"tool ({message['name']}): {message['content']}\n", role_to_color[message["role"]]))# 创建一个空的消息列表
messages = []# 向消息列表中添加一个系统角色的消息，内容是 "Answer user questions by generating SQL queries against the Chinook Music Database."
messages.append({"role": "system", "content": "Answer user questions by generating SQL queries against the Chinook Music Database."})# 向消息列表中添加一个用户角色的消息，内容是 "Hi, who are the top 5 artists by number of tracks?"
messages.append({"role": "user", "content": "Hi, who are the top 5 artists by number of tracks?"})# 使用 chat_completion_request 函数获取聊天响应
chat_response = chat_completion_request(messages, tools)# 从聊天响应中获取助手的消息
assistant_message = chat_response.json()["choices"][0]["message"]# 将助手的消息添加到消息列表中
messages.append(assistant_message)# # 如果助手的消息中有功能调用
if assistant_message.get("tool_calls"):# 使用 execute_function_call 函数执行功能调用，并获取结果results = execute_function_call(assistant_message)# 将功能的结果作为一个功能角色的消息添加到消息列表中messages.append({"role": "tool", "name": assistant_message["tool_calls"][0]["function"]["name"], "content": results,"tool_call_id":assistant_message["tool_calls"][0]["id"]})#向消息列表中添加一个用户的问题，内容是 "What is the name of the album with the most tracks?"
messages.append({"role": "user", "content": "What is the name of the album with the most tracks?"})# 使用 chat_completion_request 函数获取聊天响应
chat_response = chat_completion_request(messages, tools)# 从聊天响应中获取助手的消息
assistant_message = chat_response.json()["choices"][0]["message"]
# 将助手的消息添加到消息列表中
messages.append(assistant_message)# 如果助手的消息中有功能调用
if assistant_message.get("tool_calls"):# 使用 execute_function_call 函数执行功能调用，并获取结果results = execute_function_call(assistant_message)# 将功能的结果作为一个功能角色的消息添加到消息列表中messages.append({"role": "tool", "content": results, "name": assistant_message["tool_calls"][0]["function"]["name"]})pretty_print_conversation(messages)
输出结果：
system: Answer user questions by generating SQL queries against the Chinook Music Database.user: Hi, who are the top 5 artists by number of tracks?assistant[tool_calls]: {'arguments': '{"query":"SELECT artists.Name, COUNT(tracks.TrackId) AS TrackCount \\nFROM artists \\nJOIN albums ON artists.ArtistId = albums.ArtistId \\nJOIN tracks ON albums.AlbumId = tracks.AlbumId \\nGROUP BY artists.Name \\nORDER BY TrackCount DESC \\nLIMIT 5;"}', 'name': 'ask_database'}tool (ask_database): [('Iron Maiden', 213), ('U2', 135), ('Led Zeppelin', 114), ('Metallica', 112), ('Lost', 92)]user: What is the name of the album with the most tracks?assistant[tool_calls]: {'arguments': '{"query":"SELECT albums.Title, COUNT(tracks.TrackId) AS TrackCount \\nFROM albums \\nJOIN tracks ON albums.AlbumId = tracks.AlbumId \\nGROUP BY albums.AlbumId \\nORDER BY TrackCount DESC \\nLIMIT 1;"}', 'name': 'ask_database'}tool (ask_database): [('Greatest Hits', 57)]
