AI在线开发

实验01-接入火山引擎豆包 AI

实验准备：

1. 获取API Key: https://console.volcengine.com/ark/region:ark+cn-beijing/apiKey

2. 获取模型接入点ID: https://console.volcengine.com/ark/region:ark+cn-beijing/endpoint

实验步骤：

1. cd AI_online #进入功能包
2. nano config.py #替换个人API Key和模型接入点，模型接入点以ep-开头

python doubao_chat.py #运行接入豆包ai脚本

实验效果如下：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import requests
import json
import os
import sys
try:
    from config import API_KEY, MODEL_ENDPOINT, API_BASE_URL, SYSTEM_PROMPT, REQUEST_TIMEOUT
except ImportError:
    print("配置文件 config.py 不存在或配置错误")
    sys.exit(1)
class DoubaoChat:
    def __init__(self):
        self.api_url = API_BASE_URL
        self.api_key = API_KEY
        self.model = MODEL_ENDPOINT
        self.timeout = REQUEST_TIMEOUT
        self.messages = [{"role": "system", "content": SYSTEM_PROMPT}]
        self.check_config()
    def check_config(self):
        if not self.api_key or self.api_key == "你的API_KEY":
            sys.exit(1)
        if not self.model or self.model == "你的接入点ID":
            sys.exit(1)
    def send_message(self, user_input):
        self.messages.append({"role": "user", "content": user_input})
        data = {
            "model": self.model,
            "messages": self.messages,
            "stream": False,
            "temperature": 0.7,
            "max_tokens": 2000
        }
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}",
            "Accept": "application/json"
        }
        try:
            response = requests.post(self.api_url, json=data, headers=headers, timeout=self.timeout)
            if response.status_code == 200:
                result = response.json()
                if 'choices' in result and len(result['choices']) > 0:
                    assistant_message = result['choices'][0]['message']['content']
                    self.messages.append({"role": "assistant", "content": assistant_message})
                    return assistant_message
                else:
                    return "API返回格式错误"
            else:
                return f"API请求失败，状态码: {response.status_code}"
        except Exception as e:
            return f"异常: {str(e)}"
    def clear_history(self):
        self.messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    def run(self):
        while True:
            try:
                user_input = input("\n你: ").strip()
                if not user_input:
                    continue
                if user_input.lower() in ['quit', 'exit', '退出']:
                    break
                elif user_input.lower() in ['clear', '清空']:
                    self.clear_history()
                    continue
                print("豆包: ", end="", flush=True)
                response = self.send_message(user_input)
                print(response)
            except KeyboardInterrupt:
                break
            except Exception as e:
                print(f"\n异常: {e}")
def main():
    chat = DoubaoChat()
    chat.run()
if __name__ == "__main__":
    main()