百川大模型开放提供API体验中心，体验不错，有小伙伴也对搭建自己的对话机器人比较兴趣，今天通过Python来简单介绍下，如何调用百川大模型的API来构建自己的小产品。
在开发环境中安装Python，如何安装？参照网络资料，这里假设已经有正常的Python环境。
import requests
# API 请求 URL
url = "https://api.baichuan-ai.com/v1/chat/completions"
# 请求头部参数
headers = {
"Content-Type": "application/json",
"Authorization": "Bearer sk-333333333" # 替换为实际的 API Key
}
# 请求体参数
data = {
"model": "Baichuan2-Turbo",
"messages": [
{
"role": "user",
"content": "你好，"
}
# {
# "role": "assistant",
# "content": "你好！有什么我可以帮助你的吗？"
# }
# 添加更多对话消息...
],
"stream": True, # 是否使用流式接口
"temperature": 0.3,
"top_p": 0.85,
"top_k": 5,
"with_search_enhance": False
# 添加其他参数...
}
# 发送 POST 请求
response = requests.post(url, headers=headers, json=data)
# 输出响应内容
print(response.text)
执行上述代码，结果如下：
hogworts@hogworts baichuan % python3 chat.py
data: {"id":"chatcmpl-M233b00CLE3goFd","object":"chat.completion.chunk","created":1703138623,"model":"Baichuan2-Turbo","choices":[{"index":0,"delta":{"role":"assistant","content":"你好"}}]}
data: {"id":"chatcmpl-M233b00CLE3goFd","object":"chat.completion.chunk","created":1703138623,"model":"Baichuan2-Turbo","choices":[{"index":0,"delta":{"role":"assistant","content":"！有什么我可以帮助"}}]}
data: {"id":"chatcmpl-M233b00CLE3goFd","object":"chat.completion.chunk","created":1703138623,"model":"Baichuan2-Turbo","choices":[{"index":0,"delta":{"role":"assistant","content":"你的吗？"},"finish_reason":"stop"}],"usage":{"prompt_tokens":4,"completion_tokens":9,"total_tokens":13}}
data: [DONE]
1. 样式不够美观，能不能将输出的结果进行JSON格式，方便阅读？
2. 上述只能进行一次对话，使用体验很不好，能不能做成连续对话的模式？
下面重构为连续对话的模式。
import subprocess
import requests
import json
# API 请求 URL
url = "https://api.baichuan-ai.com/v1/chat/completions"
# 替换为实际的 API Key
api_key = "sk-33333333"
# 请求头部参数
headers = {
"Content-Type": "application/json",
"Authorization":f"Bearer {api_key}"
}
# 持续对话循环
while True:
# 用户输入
user_input = input("用户: ")
# 如果用户输入为空，则退出循环
if not user_input:
break
# 构建对话消息
messages = [
{"role": "user", "content": user_input}
# 如果有历史消息，可以继续添加
]
# 请求体参数
data = {
"model": "Baichuan2-Turbo",
"messages": messages,
"stream": False,
"temperature": 0.3,
"top_p": 0.85,
"top_k": 5,
"with_search_enhance": False
# 添加其他参数...
}
# 发送 POST 请求
response = requests.post(url, headers=headers, json=data)
# 处理响应
if response.status_code == 200:
# 使用 jq 美化 JSON 输出
result = json.loads(response.text)
formatted_result = subprocess.run(['jq', '.'], input=json.dumps(result), text=True, capture_output=True)
# 输出助手的回复
print(f"助手: {formatted_result.stdout}")
else:
print(f"请求失败，状态码: {response.status_code}")
# 用户输入为空，退出循环
print("对话结束。")
执行程序，返回结果如下
hogworts@hogworts baichuan % python3 chat-json.py
用户: hi guys
助手: {
"id": "chatcmpl-M8c0000CLE7to5U",
"object": "chat.completion",
"created": 1703138875,
"model": "Baichuan2-Turbo",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "Hi!"
},
"finish_reason": "stop"
}
],
"usage": {
"prompt_tokens": 4,
"completion_tokens": 3,
"total_tokens": 7
}
}
用户: where are you going
助手: {
"id": "chatcmpl-M400600CLE88o2H",
"object": "chat.completion",
"created": 1703138888,
"model": "Baichuan2-Turbo",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "我暂时不能离开这个平台，但我可以帮助您回答各种问题、提供信息或执行一些任务。如果您有任何问题，请随时提问。"
},
"finish_reason": "stop"
}
],
"usage": {
"prompt_tokens": 6,
"completion_tokens": 29,
"total_tokens": 35
}
}
问题：单个API_KEY，往往会有并发量的限制，这在面对大量用户使用时，很容易受到接口请求的限制。
重构一下，支持多个Key进行轮询，就可以更多的响应用户请求。
注意：API_KEY在生成后，直接脱敏显示，在很多平台都是这样的规则，所以，谨慎保管你的KEY，否则就算是你自己生的，你也没法找回来。
import aiohttp
import asyncio
import random
import subprocess
import json
# 替换为你的多个 API Key
api_keys = ["sk-222", "sk-333", "sk-44", "sk-555", "sk-666"]
# API 请求 URL
url = "https://api.baichuan-ai.com/v1/chat/completions"
# 请求体参数
data = {
"model": "Baichuan2-Turbo",
"messages": [],
"stream": False,
"temperature": 0.3,
"top_p": 0.85,
"top_k": 5,
"with_search_enhance": False
# 添加其他参数...
}
# 异步发送请求的函数
async def send_request(api_key, user_input):
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}"
}
# 添加用户输入到 messages 中
data["messages"].append({"role": "user", "content": user_input})
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=data) as response:
if response.status == 200:
result = await response.json()
choices = result.get("choices", [])
return choices
else:
print(f"API Key: {api_key}, 请求失败，状态码: {response.status}")
return None
# 异步主函数
async def main():
while True:
# 接受用户输入
user_input = input("用户: ")
# 如果用户输入为空，则退出循环
if not user_input:
break
# 随机选择一个 API Key
selected_api_key = random.choice(api_keys)
# 使用异步发送请求
choices = await send_request(selected_api_key, user_input)
# 处理请求的结果
if choices:
# json_result = json.loads(choices)
# formatted_result = subprocess.run(['jq', '.'], input=json.dumps(json_result), text=True, capture_output=True)
print(f"API Key: {selected_api_key}, Choices: {choices}")
# 使用 jq 美化 JSON 输出
#result = json.loads(choices)
#formatted_result = subprocess.run(['jq', '.'], input=json.dumps(result), text=True, capture_output=True)
# 输出助手的回复
#print(f"API Key: {selected_api_key}, Choices: {formatted_result.stdout}")
else:
print(f"API Key: {selected_api_key}, 请求失败")
# 清空 messages，准备下一轮聊天
data["messages"] = []
# Run the event loop
if __name__ == "__main__":
asyncio.run(main())
上面会涉及到很多Python常见的组件库，比如requests,json,subprocess,aiohttp等等，都是应对特殊功能，必须的安装包，只需要通过pip命令安装即可，否则无法正常使用。
