fix(api): 改进DeepSeek API密钥的动态加载和验证

统一各模块中DeepSeek API密钥的加载方式,确保使用最新的环境变量
添加密钥有效性检查,并在无效时提供降级方案
This commit is contained in:
你的姓名 2026-01-15 23:59:18 +08:00
parent 2c0c045b2f
commit 52c426b241
3 changed files with 49 additions and 8 deletions

View File

@ -59,13 +59,38 @@ class DeepSeekIntegrator:
def _call_api(self, prompt, max_tokens=1000, temperature=0.7):
"""调用DeepSeek API"""
# 重新加载环境变量确保使用最新的API密钥
load_dotenv()
current_api_key = os.getenv("DEEPSEEK_API_KEY")
# 如果实例化时没有提供API密钥使用当前环境变量中的值
api_key_to_use = self.api_key or current_api_key
# 确保API密钥已配置
if not self.api_key:
raise ValueError("DeepSeek API密钥未配置")
if not api_key_to_use or api_key_to_use == "your-key-here":
# 返回模拟结果作为降级方案
print("DeepSeek API密钥未配置或无效返回模拟响应")
if "历史案例" in prompt or "相似事件" in prompt:
return {
"choices": [{"message": {"content": json.dumps(self.similar_cases, ensure_ascii=False)}}]
}
elif "公关声明" in prompt or "道歉信" in prompt:
return {
"choices": [{"message": {"content": self._generate_pr_statement(prompt)}}]
}
elif "传播路径" in prompt or "影响预测" in prompt:
return {
"choices": [{"message": {"content": self._predict_propagation(prompt)}}]
}
else:
return {
"choices": [{"message": {"content": "API密钥未配置返回模拟响应。"}}]
}
# 实际API调用代码
headers = {
"Authorization": f"Bearer {self.api_key}",
"Authorization": f"Bearer {api_key_to_use}",
"Content-Type": "application/json"
}
payload = {

View File

@ -61,10 +61,15 @@ def call_deepseek_api(prompt: str, max_tokens: int = 500, temperature: float = 0
"""调用DeepSeek大模型API"""
import time
import requests
import json
# 重新加载环境变量确保使用最新的API密钥
load_dotenv()
current_api_key = os.getenv("DEEPSEEK_API_KEY")
# 检查API Key是否配置
if not DEEPSEEK_API_KEY:
print("DeepSeek API密钥未配置返回模拟响应")
if not current_api_key or current_api_key == "your-key-here":
print("DeepSeek API密钥未配置或无效,返回模拟响应")
# 返回模拟响应
if "问题分类" in prompt:
return json.dumps({"primary_category": "其他", "secondary_category": "其他"}, ensure_ascii=False)
@ -85,7 +90,7 @@ def call_deepseek_api(prompt: str, max_tokens: int = 500, temperature: float = 0
try:
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {DEEPSEEK_API_KEY}"
"Authorization": f"Bearer {current_api_key}"
}
payload = {

View File

@ -610,16 +610,27 @@ class SmartCustomerService:
def _call_deepseek_api(self, user_id: str, churn_prediction: Dict[str, Any], query: str = "") -> str:
"""调用DeepSeek API生成挽留措施"""
try:
# 重新加载环境变量确保使用最新的API密钥
load_dotenv()
current_api_key = os.getenv("DEEPSEEK_API_KEY", "")
# 检查API密钥是否有效
if not DEEPSEEK_API_KEY or DEEPSEEK_API_KEY == "your-key-here":
if not current_api_key or current_api_key == "your-key-here":
st.warning("DeepSeek API密钥未配置或无效将使用备用方案")
return self._generate_mock_retention_plan(user_id, churn_prediction, query)
# 在函数内部创建client对象确保使用最新的API密钥
from openai import OpenAI
current_client = OpenAI(
api_key=current_api_key,
base_url=DEEPSEEK_API_BASE
)
# 构建智能提示词(包含用户查询内容)
prompt = self._build_retention_prompt(user_id, churn_prediction, query)
# 调用DeepSeek API使用新的openai接口
response = client.chat.completions.create(
response = current_client.chat.completions.create(
model="deepseek-chat",
messages=[
{"role": "system", "content": "你是一个专业的客户关系管理专家,擅长制定客户挽留策略。请基于客户的具体问题和流失风险分析,提供切实可行的个性化挽留措施。"},