239 lines
6.9 KiB
Python
239 lines
6.9 KiB
Python
"""
|
||
DeepSeek集成测试脚本
|
||
"""
|
||
import asyncio
|
||
import sys
|
||
import os
|
||
|
||
# 添加项目路径
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from app.services.llm_service import llm_service
|
||
from app.core.config import settings
|
||
|
||
|
||
async def test_deepseek_direct():
|
||
"""直接测试DeepSeek API调用"""
|
||
print("=" * 60)
|
||
print("测试1: 直接调用DeepSeek API")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
result = await llm_service.call_deepseek(
|
||
prompt="请用一句话介绍人工智能",
|
||
model="deepseek-chat",
|
||
temperature=0.7
|
||
)
|
||
print("✅ DeepSeek调用成功！")
|
||
print(f"响应: {result}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ DeepSeek调用失败: {str(e)}")
|
||
return False
|
||
|
||
|
||
async def test_deepseek_via_llm_service():
|
||
"""通过LLM服务通用接口测试"""
|
||
print("\n" + "=" * 60)
|
||
print("测试2: 通过LLM服务通用接口调用DeepSeek")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
result = await llm_service.call_llm(
|
||
prompt="请将以下文本翻译成英文：你好，世界",
|
||
provider="deepseek",
|
||
model="deepseek-chat",
|
||
temperature=0.7
|
||
)
|
||
print("✅ 通过LLM服务调用成功！")
|
||
print(f"响应: {result}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ 调用失败: {str(e)}")
|
||
return False
|
||
|
||
|
||
async def test_deepseek_coder():
|
||
"""测试DeepSeek Coder模型"""
|
||
print("\n" + "=" * 60)
|
||
print("测试3: 测试DeepSeek Coder模型")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
result = await llm_service.call_llm(
|
||
prompt="请用Python编写一个函数，计算斐波那契数列的第n项",
|
||
provider="deepseek",
|
||
model="deepseek-coder",
|
||
temperature=0.3
|
||
)
|
||
print("✅ DeepSeek Coder调用成功！")
|
||
print(f"响应:\n{result}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ 调用失败: {str(e)}")
|
||
return False
|
||
|
||
|
||
async def test_workflow_engine_llm_node():
|
||
"""测试工作流引擎中的LLM节点"""
|
||
print("\n" + "=" * 60)
|
||
print("测试4: 测试工作流引擎中的LLM节点")
|
||
print("=" * 60)
|
||
|
||
from app.services.workflow_engine import WorkflowEngine
|
||
|
||
# 创建一个简单的工作流数据
|
||
workflow_data = {
|
||
"nodes": [
|
||
{
|
||
"id": "start-1",
|
||
"type": "start",
|
||
"data": {"label": "开始"}
|
||
},
|
||
{
|
||
"id": "llm-1",
|
||
"type": "llm",
|
||
"data": {
|
||
"label": "DeepSeek节点",
|
||
"prompt": "请总结以下内容：{input}",
|
||
"provider": "deepseek",
|
||
"model": "deepseek-chat",
|
||
"temperature": 0.7
|
||
}
|
||
},
|
||
{
|
||
"id": "end-1",
|
||
"type": "end",
|
||
"data": {"label": "结束"}
|
||
}
|
||
],
|
||
"edges": [
|
||
{"id": "e1", "source": "start-1", "target": "llm-1"},
|
||
{"id": "e2", "source": "llm-1", "target": "end-1"}
|
||
]
|
||
}
|
||
|
||
# 创建引擎并执行
|
||
engine = WorkflowEngine("test-workflow", workflow_data)
|
||
|
||
try:
|
||
input_data = {
|
||
"input": "人工智能是计算机科学的一个分支，它试图理解智能的实质，并生产出一种新的能以人类智能相似的方式做出反应的智能机器。"
|
||
}
|
||
|
||
result = await engine.execute(input_data)
|
||
|
||
print("✅ 工作流执行成功！")
|
||
print(f"执行结果: {result.get('result', 'N/A')}")
|
||
print(f"节点结果: {result.get('node_results', {})}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ 工作流执行失败: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
async def test_prompt_template():
|
||
"""测试Prompt模板变量替换"""
|
||
print("\n" + "=" * 60)
|
||
print("测试5: 测试Prompt模板变量替换")
|
||
print("=" * 60)
|
||
|
||
from app.services.workflow_engine import WorkflowEngine
|
||
|
||
workflow_data = {
|
||
"nodes": [
|
||
{
|
||
"id": "start-1",
|
||
"type": "start",
|
||
"data": {"label": "开始"}
|
||
},
|
||
{
|
||
"id": "llm-1",
|
||
"type": "llm",
|
||
"data": {
|
||
"label": "DeepSeek节点",
|
||
"prompt": "用户说：{user_input}，请回复：",
|
||
"provider": "deepseek",
|
||
"model": "deepseek-chat"
|
||
}
|
||
},
|
||
{
|
||
"id": "end-1",
|
||
"type": "end",
|
||
"data": {"label": "结束"}
|
||
}
|
||
],
|
||
"edges": [
|
||
{"id": "e1", "source": "start-1", "target": "llm-1"},
|
||
{"id": "e2", "source": "llm-1", "target": "end-1"}
|
||
]
|
||
}
|
||
|
||
engine = WorkflowEngine("test-template", workflow_data)
|
||
|
||
try:
|
||
input_data = {
|
||
"user_input": "你好，请介绍一下自己"
|
||
}
|
||
|
||
result = await engine.execute(input_data)
|
||
|
||
print("✅ Prompt模板测试成功！")
|
||
print(f"输入: {input_data}")
|
||
print(f"输出: {result.get('result', 'N/A')}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ 测试失败: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
async def main():
|
||
"""主测试函数"""
|
||
print("\n" + "🚀 开始DeepSeek集成测试" + "\n")
|
||
|
||
# 检查配置
|
||
print("检查配置...")
|
||
print(f"DeepSeek API Key: {'已配置' if settings.DEEPSEEK_API_KEY else '❌ 未配置'}")
|
||
print(f"DeepSeek Base URL: {settings.DEEPSEEK_BASE_URL}")
|
||
print(f"DeepSeek客户端: {'已初始化' if llm_service.deepseek_client else '❌ 未初始化'}")
|
||
print()
|
||
|
||
if not settings.DEEPSEEK_API_KEY:
|
||
print("❌ DeepSeek API Key未配置，请先配置API Key")
|
||
return
|
||
|
||
if not llm_service.deepseek_client:
|
||
print("❌ DeepSeek客户端未初始化")
|
||
return
|
||
|
||
# 运行测试
|
||
results = []
|
||
|
||
results.append(await test_deepseek_direct())
|
||
results.append(await test_deepseek_via_llm_service())
|
||
results.append(await test_deepseek_coder())
|
||
results.append(await test_workflow_engine_llm_node())
|
||
results.append(await test_prompt_template())
|
||
|
||
# 汇总结果
|
||
print("\n" + "=" * 60)
|
||
print("测试结果汇总")
|
||
print("=" * 60)
|
||
passed = sum(results)
|
||
total = len(results)
|
||
print(f"通过: {passed}/{total}")
|
||
print(f"失败: {total - passed}/{total}")
|
||
|
||
if passed == total:
|
||
print("\n✅ 所有测试通过！DeepSeek集成正常！")
|
||
else:
|
||
print(f"\n⚠️ 有 {total - passed} 个测试失败，请检查配置和网络连接")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|