百川大模型开放提供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 choiceselse:print(f"API Key: {api_key}, 请求失败,状态码: {response.status}")return None# 异步主函数
async def main():while True:# 接受用户输入user_input = input("用户: ")# 如果用户输入为空,则退出循环if not user_input:break# 随机选择一个 API Keyselected_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命令安装即可,否则无法正常使用。