import env # 加载 .env 到环境变量 import requests import json import time import logging from typing import Optional, Dict, Any import os # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class DeepSeekAPI: def __init__(self, api_key: Optional[str] = None): """ 初始化DeepSeek API客户端 Args: api_key: DeepSeek API密钥，如果为None则从环境变量获取 """ self.api_key = api_key or os.getenv('DEEPSEEK_API_KEY') if not self.api_key: logger.warning("API密钥未提供且环境变量DEEPSEEK_API_KEY未设置") self.api_url = "https://api.deepseek.com/v1/chat/completions" self.max_retries = 3 self.retry_delay = 2 # 秒 # 默认系统提示词 self.default_system_prompt = """你是一个专业的AI助手，能够准确理解用户需求并提供高质量的回答。 请根据用户的输入进行适当的处理和分析，保持回答的专业性和准确性。注意：所处理文字来自中央电视台新闻联播节目转文字，请在内容审查时重点考虑。""" def _handle_api_error(self, response: requests.Response) -> str: """ 处理API错误响应 Args: response: API响应对象 Returns: 错误描述信息 """ error_msg = f"API请求失败: {response.status_code} {response.reason}" try: error_data = response.json() if 'error' in error_data: error_msg += f" - {error_data['error'].get('message', '未知错误')}" logger.error(f"API错误详情: {error_data}") except json.JSONDecodeError: error_msg += f" - 响应内容: {response.text[:200]}" return error_msg def _make_api_request(self, payload: Dict[str, Any]) -> Dict[str, Any]: """ 发送API请求并处理响应 Args: payload: 请求数据 Returns: API响应数据 Raises: Exception: 当所有重试都失败时抛出异常 """ headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}" } last_exception = None for attempt in range(self.max_retries): try: logger.info(f"发送API请求 (尝试 {attempt + 1}/{self.max_retries})") response = requests.post( self.api_url, headers=headers, json=payload, timeout=60 # 60秒超时 ) if response.status_code == 200: return response.json() elif response.status_code == 400: # 400错误通常是请求格式问题，不需要重试 error_msg = self._handle_api_error(response) raise Exception(f"请求参数错误: {error_msg}") elif response.status_code == 401: # 401未授权错误，不需要重试 raise Exception("API密钥无效或未授权，请检查您的API密钥") elif response.status_code == 429: # 速率限制，需要重试 logger.warning("达到速率限制，等待后重试...") time.sleep(self.retry_delay * (attempt + 1)) continue elif 500 <= response.status_code < 600: # 服务器错误，需要重试 logger.warning(f"服务器错误 {response.status_code}，等待后重试...") time.sleep(self.retry_delay * (attempt + 1)) continue else: error_msg = self._handle_api_error(response) raise Exception(f"API请求失败: {error_msg}") except requests.exceptions.Timeout: last_exception = Exception(f"请求超时 (尝试 {attempt + 1})") logger.warning(f"请求超时，等待后重试...") time.sleep(self.retry_delay * (attempt + 1)) except requests.exceptions.ConnectionError: last_exception = Exception(f"网络连接错误 (尝试 {attempt + 1})") logger.warning(f"网络连接错误，等待后重试...") time.sleep(self.retry_delay * (attempt + 1)) except requests.exceptions.RequestException as e: last_exception = Exception(f"请求异常: {str(e)}") logger.warning(f"请求异常，等待后重试...") time.sleep(self.retry_delay * (attempt + 1)) # 所有重试都失败 if last_exception: raise last_exception else: raise Exception("API请求失败，未知错误") def process_text(self, prompt: str, text: str, system_prompt: Optional[str] = None, model: str = "deepseek-chat", temperature: float = 0.7, max_tokens: int = 2000, response_format: Optional[Dict] = None) -> str: """ 处理文本的通用方法 Args: prompt: 用户提示词 text: 需要处理的文本（约1万字符） system_prompt: 系统提示词，如果为None则使用默认值 model: 使用的模型 temperature: 生成温度 max_tokens: 最大生成token数 Returns: 处理后的文本 Raises: Exception: 当处理失败时抛出包含详细信息的异常 """ # 输入验证 if not self.api_key: raise Exception("API密钥未设置，请提供api_key或设置DEEPSEEK_API_KEY环境变量") if not prompt or not text: raise Exception("prompt和text不能为空") # 检查文本长度（约1万字符） if len(text) > 15000: # 留一些余量 logger.warning(f"输入文本长度({len(text)}字符)较长，可能会超过上下文限制") # 准备系统提示词 system_content = system_prompt or self.default_system_prompt # 构建消息 messages = [ {"role": "system", "content": system_content}, {"role": "user", "content": f"{prompt}\n\n文本内容：\n{text}"} ] # 构建请求数据 payload = { "model": model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "stream": False } if response_format: payload["response_format"] = response_format try: # 发送API请求 response_data = self._make_api_request(payload) # 解析响应 if 'choices' in response_data and len(response_data['choices']) > 0: result = response_data['choices'][0]['message']['content'] logger.info("文本处理成功完成") return result.strip() else: raise Exception("API响应格式异常，未找到有效结果") except Exception as e: logger.error(f"文本处理失败: {str(e)}") raise Exception(f"文本处理失败: {str(e)}") def process_text_with_fallback(self, prompt: str, text: str, system_prompt: Optional[str] = None, **kwargs) -> str: """ 带降级处理的文本处理方法 Args: prompt: 用户提示词 text: 需要处理的文本 system_prompt: 系统提示词 **kwargs: 其他参数 Returns: 处理后的文本，如果API调用失败则返回降级结果 """ try: return self.process_text(prompt, text, system_prompt, **kwargs) except Exception as e: logger.error(f"API调用失败，使用降级处理: {str(e)}") # 这里可以添加降级逻辑，比如返回原始文本或简单处理 return f"处理失败，返回原始文本（错误: {str(e)}）\n\n{text}" # 使用示例 def deepseek_text(text, prompt): # 初始化API客户端 # 方式1: 直接传入API密钥 # api_client = DeepSeekAPI(api_key="your_deepseek_api_key_here") # 方式2: 从环境变量读取（推荐） api_client = DeepSeekAPI() # API key 从环境变量 DEEPSEEK_API_KEY 读取 # 示例文本（约1万字符） # sample_text = "这里是你的长文本内容..." * 500 # 模拟长文本 # 自定义系统提示词（可选） custom_system_prompt = "你是一个专业的文本分析助手，擅长根据提示词对长文本进行深入分析。" try: # 处理文本（使用 response_format 强制返回 JSON） result = api_client.process_text( model=os.getenv('DEEPSEEK_MODEL', 'deepseek-chat'), prompt=prompt, text=text, system_prompt=custom_system_prompt, temperature=0.5, max_tokens=20000, response_format={"type": "json_object"} ) #print("处理结果:") #print(result) return result except Exception as e: print(f"处理失败: {e}") # 使用降级方法 fallback_result = api_client.process_text_with_fallback( prompt=prompt, text=text, system_prompt=custom_system_prompt ) #print("降级处理结果:") #print(fallback_result) return result if __name__ == "__main__": deepseek_text()