279 lines
8.3 KiB
Python
Executable File
279 lines
8.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
测试API密钥是否有效
|
||
"""
|
||
import os
|
||
import sys
|
||
from dotenv import load_dotenv
|
||
|
||
# 添加项目根目录到Python路径
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
# 加载环境变量
|
||
load_dotenv()
|
||
|
||
def test_deepseek_api():
|
||
"""测试DeepSeek API"""
|
||
print("\n测试 DeepSeek API...")
|
||
|
||
api_key = os.environ.get('DEEPSEEK_API_KEY')
|
||
if not api_key:
|
||
print("❌ 错误: 未设置 DEEPSEEK_API_KEY")
|
||
return False
|
||
|
||
try:
|
||
from openai import OpenAI
|
||
|
||
base_url = os.environ.get('DEEPSEEK_BASE_URL', 'https://api.deepseek.com/v1')
|
||
|
||
# 检测是否是火山引擎
|
||
if 'volces.com' in base_url:
|
||
print(" 检测到火山引擎 ARK 平台")
|
||
r1_model = "deepseek-r1-250120"
|
||
v3_model = "deepseek-v3-241226"
|
||
else:
|
||
r1_model = "deepseek-reasoner"
|
||
v3_model = "deepseek-chat"
|
||
|
||
client = OpenAI(
|
||
api_key=api_key,
|
||
base_url=base_url
|
||
)
|
||
|
||
# 测试V3模型（先测试V3，因为它更稳定）
|
||
print(f" 测试 V3 模型 ({v3_model})...")
|
||
try:
|
||
response = client.chat.completions.create(
|
||
model=v3_model,
|
||
messages=[{"role": "user", "content": "Hello, this is a test. Reply with OK."}],
|
||
max_tokens=10
|
||
)
|
||
|
||
if response.choices[0].message.content:
|
||
print(" ✓ V3 模型测试成功")
|
||
else:
|
||
print(" ❌ V3 模型响应异常")
|
||
return False
|
||
except Exception as e:
|
||
print(f" ❌ V3 模型测试失败: {e}")
|
||
# 如果是火山引擎，可能需要使用endpoint ID
|
||
if 'volces.com' in base_url:
|
||
print(" 提示: 火山引擎可能需要使用自定义的 endpoint ID (如 ep-xxxxx)")
|
||
return False
|
||
|
||
# 测试R1模型
|
||
print(f" 测试 R1 模型 ({r1_model})...")
|
||
try:
|
||
response = client.chat.completions.create(
|
||
model=r1_model,
|
||
messages=[{"role": "user", "content": "Hello, this is a test. Reply with OK."}],
|
||
max_tokens=10
|
||
)
|
||
|
||
if response.choices[0].message.content:
|
||
print(" ✓ R1 模型测试成功")
|
||
else:
|
||
print(" ❌ R1 模型响应异常")
|
||
# R1失败不影响整体，因为V3可以工作
|
||
except Exception as e:
|
||
print(f" ⚠️ R1 模型测试失败: {e}")
|
||
print(" 注意: R1模型可能需要特殊配置或不可用")
|
||
|
||
print("✅ DeepSeek API 测试通过")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ DeepSeek API 测试失败: {e}")
|
||
return False
|
||
|
||
def test_tavily_api():
|
||
"""测试Tavily API"""
|
||
print("\n测试 Tavily API...")
|
||
|
||
api_key = os.environ.get('TAVILY_API_KEY')
|
||
if not api_key:
|
||
print("❌ 错误: 未设置 TAVILY_API_KEY")
|
||
return False
|
||
|
||
try:
|
||
from tavily import TavilyClient
|
||
|
||
client = TavilyClient(api_key=api_key)
|
||
|
||
# 执行测试搜索
|
||
print(" 执行测试搜索...")
|
||
response = client.search("test query", max_results=1)
|
||
|
||
if response and 'results' in response:
|
||
print(f" ✓ 搜索返回 {len(response['results'])} 条结果")
|
||
print("✅ Tavily API 测试通过")
|
||
return True
|
||
else:
|
||
print(" ❌ 搜索响应异常")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ Tavily API 测试失败: {e}")
|
||
return False
|
||
|
||
def test_task_manager():
|
||
"""测试任务管理器"""
|
||
print("\n测试任务管理器...")
|
||
|
||
try:
|
||
from app.services.task_manager import task_manager
|
||
|
||
# 测试提交任务
|
||
def test_func(x):
|
||
return x * 2
|
||
|
||
task_id = task_manager.submit_task(test_func, 5)
|
||
print(f" ✓ 任务提交成功: {task_id}")
|
||
|
||
# 等待任务完成
|
||
import time
|
||
time.sleep(1)
|
||
|
||
# 检查任务状态
|
||
status = task_manager.get_task_status(task_id)
|
||
if status and status['status'] == 'completed':
|
||
print(" ✓ 任务执行成功")
|
||
print("✅ 任务管理器测试通过")
|
||
return True
|
||
else:
|
||
print(f" ❌ 任务状态异常: {status}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 任务管理器测试失败: {e}")
|
||
return False
|
||
|
||
def test_mongodb_connection():
|
||
"""测试MongoDB连接（可选）"""
|
||
print("\n测试 MongoDB 连接（可选）...")
|
||
|
||
mongodb_uri = os.environ.get('MONGODB_URI', 'mongodb://localhost:27017/deepresearch')
|
||
|
||
try:
|
||
from pymongo import MongoClient
|
||
|
||
client = MongoClient(mongodb_uri, serverSelectionTimeoutMS=5000)
|
||
|
||
# 测试连接
|
||
client.server_info()
|
||
print("✅ MongoDB 连接测试通过")
|
||
|
||
# 测试数据库操作
|
||
db = client.get_database()
|
||
test_collection = db['test_collection']
|
||
|
||
# 插入测试文档
|
||
result = test_collection.insert_one({'test': 'document'})
|
||
|
||
# 查询测试文档
|
||
doc = test_collection.find_one({'_id': result.inserted_id})
|
||
|
||
# 删除测试文档
|
||
test_collection.delete_one({'_id': result.inserted_id})
|
||
|
||
if doc and doc['test'] == 'document':
|
||
print(" ✓ MongoDB 读写测试通过")
|
||
return True
|
||
else:
|
||
print(" ❌ MongoDB 读写测试失败")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ MongoDB 连接测试失败（可选）: {e}")
|
||
print(" 提示: MongoDB是可选的，不影响基本功能")
|
||
return False
|
||
|
||
def check_python_version():
|
||
"""检查Python版本"""
|
||
print("\n检查 Python 版本...")
|
||
|
||
version = sys.version_info
|
||
if version.major == 3 and version.minor >= 8:
|
||
print(f"✅ Python 版本: {version.major}.{version.minor}.{version.micro}")
|
||
return True
|
||
else:
|
||
print(f"❌ Python 版本过低: {version.major}.{version.minor}.{version.micro}")
|
||
print(" 需要 Python 3.8 或更高版本")
|
||
return False
|
||
|
||
def check_dependencies():
|
||
"""检查依赖包"""
|
||
print("\n检查依赖包...")
|
||
|
||
required_packages = [
|
||
'flask',
|
||
'flask_cors',
|
||
'flask_socketio',
|
||
'openai',
|
||
'tavily',
|
||
'pydantic',
|
||
'python-dotenv'
|
||
]
|
||
|
||
missing_packages = []
|
||
|
||
for package in required_packages:
|
||
try:
|
||
__import__(package)
|
||
except ImportError:
|
||
missing_packages.append(package)
|
||
|
||
if missing_packages:
|
||
print(f"❌ 缺少以下依赖包: {', '.join(missing_packages)}")
|
||
print(" 请运行: pip install -r requirements.txt")
|
||
return False
|
||
else:
|
||
print("✅ 所有必需的依赖包已安装")
|
||
return True
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("=" * 60)
|
||
print("DeepResearch API 密钥测试工具")
|
||
print("=" * 60)
|
||
|
||
all_passed = True
|
||
|
||
# 检查Python版本
|
||
if not check_python_version():
|
||
all_passed = False
|
||
|
||
# 检查依赖包
|
||
if not check_dependencies():
|
||
all_passed = False
|
||
|
||
# 测试DeepSeek API
|
||
if not test_deepseek_api():
|
||
all_passed = False
|
||
|
||
# 测试Tavily API
|
||
if not test_tavily_api():
|
||
all_passed = False
|
||
|
||
# 测试任务管理器
|
||
if not test_task_manager():
|
||
all_passed = False
|
||
|
||
# 测试MongoDB连接（可选）
|
||
test_mongodb_connection() # 不影响整体结果
|
||
|
||
print("\n" + "=" * 60)
|
||
|
||
if all_passed:
|
||
print("✅ 所有必需的测试都已通过！")
|
||
print("\n您可以运行以下命令启动应用:")
|
||
print("1. 启动应用: python app.py")
|
||
print("\n注意: 不再需要启动 Celery Worker 和 Redis")
|
||
else:
|
||
print("❌ 有些测试未通过，请检查上述错误信息")
|
||
print("\n常见问题:")
|
||
print("1. 确保在.env文件中正确设置了API密钥")
|
||
print("2. 检查网络连接是否正常")
|
||
|
||
if __name__ == '__main__':
|
||
main() |