|
|
#!/usr/bin/env python3
|
|
|
# -*- coding: utf-8 -*-
|
|
|
"""
|
|
|
DeepSeek AI助手 - API客户端模块
|
|
|
|
|
|
该模块实现了与DeepSeek服务器通信的API客户端，负责处理API请求、响应解析
|
|
|
以及日程和待办事项的智能提取功能。
|
|
|
"""
|
|
|
|
|
|
# 导入标准库模块
|
|
|
import requests # HTTP请求库
|
|
|
import json # JSON数据处理
|
|
|
from typing import Dict, List, Any # 类型注解
|
|
|
import logging # 日志模块
|
|
|
|
|
|
# 导入项目内部模块
|
|
|
from config.config import config_manager # 配置管理器
|
|
|
|
|
|
# 设置日志
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DeepSeekAPI:
|
|
|
"""DeepSeek API客户端类
|
|
|
|
|
|
该类负责与DeepSeek AI服务器进行通信，处理所有API请求和响应。
|
|
|
支持模型切换、API密钥管理、消息发送以及日程和待办事项的智能提取。
|
|
|
"""
|
|
|
|
|
|
def __init__(self, logger=None):
|
|
|
"""初始化DeepSeek API客户端
|
|
|
|
|
|
从配置管理器加载初始配置，包括API密钥、API地址和默认模型。
|
|
|
|
|
|
参数:
|
|
|
logger: 日志记录器对象，用于记录API调用错误
|
|
|
"""
|
|
|
# 从配置管理器获取API密钥
|
|
|
self.api_key = config_manager.get_api_key()
|
|
|
# 从配置管理器获取API URL
|
|
|
self.api_url = config_manager.get_api_url()
|
|
|
# 从配置管理器获取默认模型
|
|
|
self.model = config_manager.get_default_model()
|
|
|
# 日志记录器
|
|
|
self.logger = logger
|
|
|
|
|
|
def set_model(self, new_model):
|
|
|
"""切换AI模型
|
|
|
|
|
|
参数:
|
|
|
new_model: 新的模型名称（例如："deepseek-chat"或"deepseek-coder"）
|
|
|
|
|
|
该方法用于切换不同的AI模型，并将新模型保存到配置中，以便下次启动时使用。
|
|
|
"""
|
|
|
self.model = new_model # 更新当前实例的模型设置
|
|
|
config_manager.set_default_model(new_model) # 保存到配置管理器
|
|
|
|
|
|
def set_api_key(self, new_key):
|
|
|
"""设置API密钥
|
|
|
|
|
|
参数:
|
|
|
new_key: 新的API密钥字符串
|
|
|
|
|
|
该方法用于更新API密钥，并将新密钥保存到配置中，以便下次启动时使用。
|
|
|
API密钥是与DeepSeek服务器通信的必要凭证。
|
|
|
"""
|
|
|
self.api_key = new_key # 更新当前实例的API密钥
|
|
|
config_manager.set_api_key(new_key) # 保存到配置管理器
|
|
|
|
|
|
def set_api_url(self, new_url):
|
|
|
"""设置API URL
|
|
|
|
|
|
参数:
|
|
|
new_url: 新的API端点URL字符串
|
|
|
|
|
|
该方法用于更新API服务器的URL地址，并将新URL保存到配置中。
|
|
|
通常不需要修改此设置，除非DeepSeek更改了API端点。
|
|
|
"""
|
|
|
self.api_url = new_url # 更新当前实例的API URL
|
|
|
config_manager.set_api_url(new_url) # 保存到配置管理器
|
|
|
|
|
|
def send_message(self, message):
|
|
|
"""发送单条消息获取AI回复
|
|
|
|
|
|
参数:
|
|
|
message: 用户消息内容字符串
|
|
|
|
|
|
返回:
|
|
|
成功时返回AI助手的回复内容字符串
|
|
|
失败时返回错误信息字符串
|
|
|
|
|
|
该方法是一个简化的API调用接口，用于发送单条用户消息并获取AI回复。
|
|
|
它会自动构造包含单个用户消息的对话历史，然后调用call_api方法。
|
|
|
"""
|
|
|
# 构造包含单条用户消息的对话历史
|
|
|
chat_history = [{"role": "user", "content": message}]
|
|
|
# 调用底层API方法获取回复
|
|
|
return self.call_api(chat_history)
|
|
|
|
|
|
def call_api(self, chat_history):
|
|
|
"""调用DeepSeek API获取AI回复
|
|
|
|
|
|
参数:
|
|
|
chat_history: 对话历史列表，格式为：
|
|
|
[{"role": "user/assistant", "content": "消息内容"}, ...]
|
|
|
|
|
|
返回:
|
|
|
成功时返回AI助手的回复内容字符串
|
|
|
失败时返回错误信息字符串
|
|
|
|
|
|
该方法是API客户端的核心方法，负责：
|
|
|
1. 验证API密钥的有效性
|
|
|
2. 构造符合API要求的请求头和参数
|
|
|
3. 发送HTTP POST请求到DeepSeek API服务器
|
|
|
4. 处理响应并提取AI回复内容
|
|
|
5. 捕获并处理各种可能的错误
|
|
|
"""
|
|
|
# 校验API Key是否已正确设置
|
|
|
if not self.api_key or self.api_key == "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx":
|
|
|
return "错误：请先在config.py中填写有效的API Key！"
|
|
|
|
|
|
# 构造API请求头
|
|
|
headers = {
|
|
|
"Content-Type": "application/json", # 指定请求内容类型为JSON
|
|
|
"Authorization": f"Bearer {self.api_key}" # 使用Bearer令牌认证
|
|
|
}
|
|
|
|
|
|
# 构造API请求参数
|
|
|
payload = {
|
|
|
"model": self.model, # 指定使用的AI模型
|
|
|
"messages": chat_history, # 对话历史
|
|
|
"temperature": config_manager.get_temperature(), # 温度参数（控制回复的创造性）
|
|
|
"max_tokens": config_manager.get_max_tokens(), # 最大令牌数（控制回复长度）
|
|
|
"stream": False # 是否使用流式响应（当前设置为非流式）
|
|
|
}
|
|
|
|
|
|
try:
|
|
|
# 发送HTTP POST请求到API服务器
|
|
|
response = requests.post(
|
|
|
self.api_url, # API端点URL
|
|
|
headers=headers, # 请求头
|
|
|
data=json.dumps(payload), # 将payload转换为JSON字符串
|
|
|
timeout=30 # 设置30秒超时
|
|
|
)
|
|
|
|
|
|
# 检查响应状态码，如果不是2xx则抛出异常
|
|
|
response.raise_for_status()
|
|
|
|
|
|
# 解析JSON响应
|
|
|
result = response.json()
|
|
|
|
|
|
# 提取并返回AI助手的回复内容
|
|
|
return result["choices"][0]["message"]["content"].strip()
|
|
|
|
|
|
except requests.exceptions.Timeout:
|
|
|
# 处理请求超时错误
|
|
|
error_msg = "错误：请求超时，请检查网络！"
|
|
|
if self.logger:
|
|
|
self.logger.log("error", error_msg)
|
|
|
return error_msg
|
|
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
|
# 处理其他HTTP请求错误
|
|
|
error_msg = f"API请求失败：{str(e)}"
|
|
|
if self.logger:
|
|
|
self.logger.log("error", error_msg)
|
|
|
return error_msg
|
|
|
|
|
|
except KeyError:
|
|
|
# 处理API响应格式异常错误
|
|
|
error_msg = "错误：API响应格式异常，请检查API Key是否正确！"
|
|
|
if self.logger:
|
|
|
self.logger.log("error", error_msg)
|
|
|
return error_msg
|
|
|
|
|
|
def parse_schedule(self, user_input):
|
|
|
"""解析用户输入中的日程信息
|
|
|
|
|
|
参数:
|
|
|
user_input: 用户输入的自然语言字符串，包含可能的日程信息
|
|
|
|
|
|
返回:
|
|
|
成功时返回日程信息字典，格式为：
|
|
|
{
|
|
|
"title": "日程标题",
|
|
|
"description": "日程描述",
|
|
|
"start_time": "ISO格式的开始时间",
|
|
|
"end_time": "ISO格式的结束时间",
|
|
|
"location": "地点",
|
|
|
"reminder": "提醒设置（可选）"
|
|
|
}
|
|
|
失败时返回None
|
|
|
|
|
|
该方法通过调用AI模型智能解析用户输入中的日程信息，支持自然语言表达的各种日程格式，
|
|
|
能够识别标题、时间、地点等关键信息，并将其转换为结构化的JSON格式。
|
|
|
"""
|
|
|
from datetime import datetime
|
|
|
system_prompt = config_manager.get_schedule_assistant_prompt()
|
|
|
|
|
|
# 获取当前时间并添加到用户输入中，帮助AI正确理解相对时间
|
|
|
current_time = datetime.now().isoformat()
|
|
|
user_input_with_time = f"当前时间：{current_time}\n\n{user_input}"
|
|
|
|
|
|
messages = [
|
|
|
{"role": "system", "content": system_prompt},
|
|
|
{"role": "user", "content": user_input_with_time}
|
|
|
]
|
|
|
|
|
|
response = self.call_api(messages)
|
|
|
|
|
|
# 检查API调用是否返回了错误信息
|
|
|
if response.startswith("错误："):
|
|
|
return None
|
|
|
|
|
|
try:
|
|
|
# 尝试解析JSON响应
|
|
|
schedule_data = json.loads(response)
|
|
|
# 确保返回的数据至少包含title字段
|
|
|
if not isinstance(schedule_data, dict) or "title" not in schedule_data:
|
|
|
return None
|
|
|
return schedule_data
|
|
|
except json.JSONDecodeError:
|
|
|
# 如果解析失败，返回None
|
|
|
return None
|
|
|
|
|
|
def parse_todo(self, user_message):
|
|
|
"""解析用户消息中的待办事项信息
|
|
|
|
|
|
参数:
|
|
|
user_message: 用户输入的自然语言字符串，包含可能的待办事项信息
|
|
|
|
|
|
返回:
|
|
|
成功时返回待办事项信息字典，格式为：
|
|
|
{
|
|
|
"title": "待办事项标题",
|
|
|
"description": "待办事项描述",
|
|
|
"due_date": "截止日期",
|
|
|
"priority": "优先级（high/medium/low）"
|
|
|
}
|
|
|
失败时返回None
|
|
|
|
|
|
该方法通过调用AI模型智能解析用户输入中的待办事项信息，支持自然语言表达的各种待办事项格式，
|
|
|
能够识别标题、截止日期、优先级等关键信息，并将其转换为结构化的JSON格式。
|
|
|
"""
|
|
|
try:
|
|
|
from datetime import datetime
|
|
|
system_prompt = config_manager.get_todo_assistant_prompt()
|
|
|
|
|
|
# 获取当前时间并添加到用户输入中，帮助AI正确理解相对时间
|
|
|
current_time = datetime.now().isoformat()
|
|
|
user_message_with_time = f"当前时间：{current_time}\n\n{user_message}"
|
|
|
|
|
|
messages = [
|
|
|
{"role": "system", "content": system_prompt},
|
|
|
{"role": "user", "content": user_message_with_time}
|
|
|
]
|
|
|
|
|
|
response = self.call_api(messages)
|
|
|
|
|
|
# 检查API调用是否返回了错误信息
|
|
|
if response.startswith("错误："):
|
|
|
return None
|
|
|
|
|
|
# 解析JSON响应
|
|
|
todo_data = json.loads(response)
|
|
|
# 确保返回的数据至少包含title字段
|
|
|
if not isinstance(todo_data, dict) or "title" not in todo_data:
|
|
|
return None
|
|
|
return todo_data
|
|
|
except json.JSONDecodeError as e:
|
|
|
print(f"解析JSON失败: {e}")
|
|
|
return None
|
|
|
except Exception as e:
|
|
|
print(f"解析待办事项失败: {e}")
|
|
|
return None |
