imes/test_fix.py

#!/usr/bin/env python3
"""测试修复后的chat接口"""

import asyncio
import json
import httpx
import os
from datetime import datetime

async def test_chat_api():
    """测试聊天API"""
    
    # 检查环境变量
    if not os.getenv("MES_AI_OPENAI_API_KEY"):
        print("❌ 请设置 MES_AI_OPENAI_API_KEY 环境变量")
        return
    
    base_url = "http://localhost:8000"
    
    # 测试健康检查
    print("🔍 测试健康检查...")
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(f"{base_url}/api/v1/health")
            if response.status_code == 200:
                print("✅ 健康检查通过")
                print(f"   响应: {response.json()}")
            else:
                print(f"❌ 健康检查失败: {response.status_code}")
                return
    except Exception as e:
        print(f"❌ 无法连接到服务器: {e}")
        print("   请确保服务已启动: python main.py")
        return
    
    # 测试聊天接口
    test_messages = [
        "你好",
        "我想进行物料入库",
        "如何使用MES系统",
        "今天的生产情况如何",
        "谢谢"
    ]
    
    session_id = f"test_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
    
    print(f"\n🤖 开始测试聊天功能 (会话ID: {session_id})...")
    
    for i, message in enumerate(test_messages, 1):
        print(f"\n--- 测试 {i}/{len(test_messages)} ---")
        print(f"👤 用户: {message}")
        
        try:
            async with httpx.AsyncClient(timeout=30.0) as client:
                response = await client.post(
                    f"{base_url}/api/v1/chat",
                    json={
                        "message": message,
                        "session_id": session_id,
                        "context": {}
                    }
                )
                
                if response.status_code == 200:
                    data = response.json()
                    print(f"🤖 AI: {data['message']}")
                    print(f"📊 问题类型: {data['question_type']}")
                    print(f"🔍 需要RAG: {data['need_rag']}")
                    print(f"🛠️  需要工具调用: {data['need_tool_call']}")
                    
                    if data.get('metadata'):
                        metadata = data['metadata']
                        if 'classification_confidence' in metadata:
                            print(f"📈 分类置信度: {metadata['classification_confidence']:.2f}")
                    
                else:
                    print(f"❌ 请求失败: {response.status_code}")
                    print(f"   错误信息: {response.text}")
                    
        except Exception as e:
            print(f"❌ 请求异常: {e}")
        
        # 稍作延迟
        await asyncio.sleep(1)
    
    # 测试会话历史
    print(f"\n📚 测试会话历史...")
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(f"{base_url}/api/v1/chat/history/{session_id}")
            if response.status_code == 200:
                history = response.json()
                print(f"✅ 会话历史获取成功")
                print(f"   消息数量: {history['message_count']}")
                print(f"   会话ID: {history['session_id']}")
            else:
                print(f"❌ 获取会话历史失败: {response.status_code}")
    except Exception as e:
        print(f"❌ 获取会话历史异常: {e}")
    
    print("\n🎉 测试完成!")

def main():
    """主函数"""
    print("🧪 MES AI 聊天接口测试工具")
    print("=" * 50)
    
    # 运行异步测试
    asyncio.run(test_chat_api())

if __name__ == "__main__":
    main()