赞
踩
百川大模型开放提供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命令安装即可,否则无法正常使用。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。