百川大模型AI对话实战——Python开发一个对话机器人-编程知识

百川大模型开放提供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())