AI在线开发

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

实验准备：
获取API Key: https://console.volcengine.com/ark/region:ark+cn-beijing/apiKey
2. 获取模型接入点ID: https://console.volcengine.com/ark/region:ark+cn-beijing/endpoint
实验步骤：
cd AI_online #进入功能包
nano config.py #替换个人API Key和模型接入点，模型接入点以ep-开头
python doubao_chat.py #运行接入豆包ai脚本
实验效果如下：
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
火山引擎豆包AI对话脚本
精简版本，直接使用API接口
"""

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 不存在或配置错误")
    print("📝 请确保 config.py 文件存在并正确配置")
    sys.exit(1)

class DoubaoChat:
    def __init__(self):
        # 火山引擎豆包API配置
        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):
        """检查API配置"""
        if not self.api_key or self.api_key == "你的API_KEY":
            print("❌ 请先配置API Key")
            print("📝 请在 config.py 文件中设置 API_KEY = '你的API_KEY'")
            print("🔗 获取方式：https://console.volcengine.com/ark/region:ark+cn-beijing/apiKey")
            sys.exit(1)
        
        if not self.model or self.model == "你的接入点ID":
            print("❌ 请先配置模型接入点ID")
            print("📝 请在 config.py 文件中设置 MODEL_ENDPOINT = '你的接入点ID'")
            print("🔗 获取方式：https://console.volcengine.com/ark/region:ark+cn-beijing/endpoint")
            sys.exit(1)
  
    def send_message(self, user_input):
        """发送消息到豆包API"""
        # 添加用户消息
        self.messages.append({"role": "user", "content": user_input})
    
        # 准备请求数据 - 根据火山引擎API文档格式
        data = {
            "model": self.model,
            "messages": self.messages,
            "stream": False,
            "temperature": 0.7,
            "max_tokens": 2000
        }
    
        # 设置请求头 - 使用Bearer认证
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}",
            "Accept": "application/json"
        }
    
        try:
            print("🔄 正在请求豆包API...", end="", flush=True)
        
            # 发送请求
            response = requests.post(self.api_url, json=data, headers=headers, timeout=self.timeout)
        
            print("\r" + " " * 30 + "\r", end="", flush=True)  # 清除加载提示
        
            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返回格式错误，请检查模型接入点ID是否正确"
            elif response.status_code == 401:
                return "❌ 认证失败 (401)，请检查API Key是否正确"
            elif response.status_code == 404:
                return "❌ 接入点不存在 (404)，请检查模型接入点ID是否正确"
            elif response.status_code == 429:
                return "❌ 请求过于频繁 (429)，请稍后再试"
            else:
                error_info = f"❌ API请求失败，状态码: {response.status_code}"
                try:
                    error_data = response.json()
                    if 'error' in error_data:
                        if isinstance(error_data['error'], dict):
                            error_msg = error_data['error'].get('message', '未知错误')
                            error_info += f"\n错误信息: {error_msg}"
                        else:
                            error_info += f"\n错误信息: {error_data['error']}"
                except:
                    error_info += f"\n响应内容: {response.text[:200]}"
                return error_info
            
        except requests.exceptions.Timeout:
            return "❌ 请求超时，请检查网络连接或稍后重试"
        except requests.exceptions.ConnectionError:
            return "❌ 连接失败，请检查网络连接"
        except requests.exceptions.RequestException as e:
            return f"❌ 网络请求异常: {str(e)}"
        except Exception as e:
            return f"❌ 未知异常: {str(e)}"
  
    def clear_history(self):
        """清空对话历史"""
        self.messages = [
            {"role": "system", "content": SYSTEM_PROMPT}
        ]
        print("✅ 对话历史已清空")
  
    def show_help(self):
        """显示帮助信息"""
        print("\n📖 使用说明:")
        print("• 直接输入消息开始对话")
        print("• 输入 'quit' 或 'exit' 退出程序")
        print("• 输入 'clear' 清空对话历史")
        print("• 输入 'help' 显示此帮助信息")
  
    def run(self):
        """运行对话程序"""
        print("🚀 火山引擎豆包AI对话系统")
        print("=" * 50)
        print(f"🔗 API地址: {self.api_url}")
        print(f"🤖 模型: {self.model}")
        print("=" * 50)
    
        self.show_help()
    
        while True:
            try:
                user_input = input("\n👤 你: ").strip()
            
                if not user_input:
                    continue
            
                if user_input.lower() in ['quit', 'exit', '退出']:
                    print("👋 再见！")
                    break
                elif user_input.lower() in ['clear', '清空']:
                    self.clear_history()
                    continue
                elif user_input.lower() in ['help', '帮助']:
                    self.show_help()
                    continue
            
                print("🤖 豆包: ", end="", flush=True)
                response = self.send_message(user_input)
                print(response)
            
            except KeyboardInterrupt:
                print("\n👋 程序已退出")
                break
            except Exception as e:
                print(f"\n❌ 程序异常: {e}")

def main():
    """主函数"""
    print("🔧 配置检查...")
  
    # 创建对话实例
    chat = DoubaoChat()
  
    # 运行对话
    chat.run()

if __name__ == "__main__":
    main()