aiagent/backend/test_deepseek.py

"""
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())