百川大模型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 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命令安装即可，否则无法正常使用。