bot.py

#!/usr/bin/env python3
"""
QQ机器人集成 - 使用NoneBot2框架
"""
import nonebot
from nonebot import on_message
from nonebot.adapters.onebot.v11 import Bot, Event, Message
import requests
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# DeepSeek API配置
DEEPSEEK_API_KEY = os.getenv('DEEPSEEK_API_KEY', 'sk-b37a7b019b094a27ae424507af28ca61')
DEEPSEEK_API_URL = os.getenv('DEEPSEEK_API_URL', 'https://api.deepseek.com/v1/chat/completions')
DEEPSEEK_MODEL = os.getenv('DEEPSEEK_MODEL', 'deepseek-reasoner')

class QQChatBot:
    def __init__(self):
        self.api_key = DEEPSEEK_API_KEY
        self.api_url = DEEPSEEK_API_URL
        self.model = DEEPSEEK_MODEL
        self.sessions = {}  # 存储每个用户的对话历史

    def get_response(self, user_id: str, message: str) -> str:
        """获取DeepSeek回复"""
        try:
            # 构建对话上下文
            if user_id not in self.sessions:
                self.sessions[user_id] = []

            # 添加用户消息
            self.sessions[user_id].append({"role": "user", "content": message})

            # 限制历史长度
            if len(self.sessions[user_id]) > 20:
                self.sessions[user_id] = self.sessions[user_id][-20:]

            # 调用DeepSeek API
            response = requests.post(
                self.api_url,
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": self.model,
                    "messages": self.sessions[user_id],
                    "temperature": 0.3,
                    "max_tokens": 2048
                },
                timeout=30
            )

            if response.status_code == 200:
                result = response.json()
                reply = result['choices'][0]['message']['content']

                # 添加助手回复到历史
                self.sessions[user_id].append({"role": "assistant", "content": reply})

                return reply
            else:
                return f"API错误: {response.status_code}"

        except Exception as e:
            return f"发生错误: {str(e)}"

# 创建机器人实例
chat_bot = QQChatBot()

# 创建消息处理器
chat_handler = on_message()

@chat_handler.handle()
async def handle_chat(bot: Bot, event: Event):
    """处理聊天消息"""
    user_id = str(event.get_user_id())
    message = str(event.get_message()).strip()

    # 跳过空消息
    if not message:
        return

    # 获取回复
    reply = chat_bot.get_response(user_id, message)

    # 发送回复
    await chat_handler.finish(Message(reply))

# NoneBot初始化
nonebot.init()

# 注册适配器
driver = nonebot.get_driver()
driver.register_adapter(nonebot.adapters.onebot.v11.Adapter)

# 加载插件
nonebot.load_builtin_plugins("echo")

if __name__ == "__main__":
    nonebot.run()