Update test_tool_call_fix.py

tests/test_tool_call_fix.py

@@ -2,214 +2,132 @@
 # -*- coding: utf-8 -*-
 
 """
-测试修复后的工具调用功能
+测试工具调用
 """
 
 import json
-import asyncio
-import httpx
+import urllib.request
+import urllib.parse
 from typing import Dict, Any
 
-# 测试配置
-TEST_URL = "http://localhost:8080/v1/chat/completions"
-TEST_AUTH_TOKEN = "sk-test-key"
-
-# 测试工具定义
-TEST_TOOLS = [
-    {
-        "type": "function",
-        "function": {
-            "name": "get_weather",
-            "description": "获取指定城市的天气信息",
-            "parameters": {
-                "type": "object",
-                "properties": {
-                    "city": {
-                        "type": "string",
-                        "description": "城市名称"
-                    },
-                    "unit": {
-                        "type": "string",
-                        "enum": ["celsius", "fahrenheit"],
-                        "description": "温度单位"
-                    }
-                },
-                "required": ["city"]
-            }
-        }
-    }
-]
+def test_tool_call():
+    """测试工具调用功能"""
 
-async def test_tool_call_streaming():
-    """测试流式工具调用"""
-    print("🧪 开始测试流式工具调用...")
-    
-    payload = {
+    # 测试请求
+    test_request = {
         "model": "glm-4.5",
         "messages": [
             {
-                "role": "user", 
-                "content": "请帮我查询北京的天气，使用摄氏度"
+                "role": "user",
+                "content": "请打开Google网站"
             }
         ],
-        "tools": TEST_TOOLS,
-        "stream": True,
-        "temperature": 0.7
-    }
-    
-    headers = {
-        "Content-Type": "application/json",
-        "Authorization": f"Bearer {TEST_AUTH_TOKEN}"
+        "tools": [
+            {
+                "type": "function",
+                "function": {
+                    "name": "playwri-browser_navigate",
+                    "description": "Navigate to a URL in the browser",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "url": {
+                                "type": "string",
+                                "description": "The URL to navigate to"
+                            }
+                        },
+                        "required": ["url"]
+                    }
+                }
+            }
+        ],
+        "stream": True
     }
-    
+
+    print("🚀 发送工具调用测试请求...")
+    print(f"📦 请求内容: {json.dumps(test_request, ensure_ascii=False, indent=2)}")
+
+    # 准备HTTP请求
+    url = "http://localhost:8080/v1/chat/completions"
+    data = json.dumps(test_request).encode('utf-8')
+
+    req = urllib.request.Request(url, data=data)
+    req.add_header('Content-Type', 'application/json')
+    req.add_header('Authorization', 'Bearer sk-test-key')
+
     try:
-        async with httpx.AsyncClient(timeout=30.0) as client:
-            async with client.stream(
-                "POST", 
-                TEST_URL, 
-                json=payload, 
-                headers=headers
-            ) as response:
-                print(f"📡 响应状态: {response.status_code}")
-                print(f"📡 响应头: {dict(response.headers)}")
-                
-                if response.status_code != 200:
-                    error_text = await response.aread()
-                    print(f"❌ 请求失败: {error_text.decode()}")
-                    return
-                
-                print("\n📦 开始接收流式数据:")
-                print("-" * 80)
-                
+        with urllib.request.urlopen(req) as response:
+            print(f"📈 响应状态: {response.status}")
+
+            if response.status == 200:
+                print("✅ 开始接收流式响应...")
+
+                tool_calls_found = []
                 chunk_count = 0
-                tool_calls_found = False
-                
-                async for line in response.aiter_lines():
-                    if not line:
-                        continue
-                        
-                    if line.startswith("data: "):
+
+                for line in response:
+                    line = line.decode('utf-8').strip()
+                    if line.startswith('data: '):
                         chunk_count += 1
-                        data_str = line[6:].strip()
-                        
-                        if data_str == "[DONE]":
-                            print(f"🏁 [{chunk_count:03d}] 流结束: [DONE]")
+                        data_str = line[6:]  # 去掉 'data: ' 前缀
+
+                        if data_str == '[DONE]':
+                            print("🏁 接收到结束信号")
                             break
-                            
+
                         try:
                             chunk = json.loads(data_str)
-                            
+
                             # 检查是否包含工具调用
-                            choices = chunk.get("choices", [])
-                            if choices:
-                                choice = choices[0]
-                                delta = choice.get("delta", {})
-                                tool_calls = delta.get("tool_calls", [])
-                                
-                                if tool_calls:
-                                    tool_calls_found = True
-                                    print(f"🔧 [{chunk_count:03d}] 工具调用块:")
-                                    for tool_call in tool_calls:
-                                        print(f"    ID: {tool_call.get('id', 'N/A')}")
-                                        print(f"    类型: {tool_call.get('type', 'N/A')}")
-                                        function = tool_call.get('function', {})
-                                        print(f"    函数名: {function.get('name', 'N/A')}")
-                                        print(f"    参数: {function.get('arguments', 'N/A')}")
-                                        print(f"    参数类型: {type(function.get('arguments', 'N/A'))}")
-                                
-                                finish_reason = choice.get("finish_reason")
-                                if finish_reason:
-                                    print(f"🏁 [{chunk_count:03d}] 完成原因: {finish_reason}")
-                                
-                                # 显示其他内容
-                                content = delta.get("content")
-                                if content:
-                                    print(f"💬 [{chunk_count:03d}] 内容: {content}")
-                            
-                            # 显示usage信息
-                            usage = chunk.get("usage")
-                            if usage:
-                                print(f"📊 [{chunk_count:03d}] 使用统计: {usage}")
-                                
+                            if 'choices' in chunk and chunk['choices']:
+                                choice = chunk['choices'][0]
+                                if 'delta' in choice and 'tool_calls' in choice['delta']:
+                                    tool_calls = choice['delta']['tool_calls']
+                                    if tool_calls:
+                                        for tool_call in tool_calls:
+                                            print(f"🔧 发现工具调用: {json.dumps(tool_call, ensure_ascii=False, indent=2)}")
+                                            tool_calls_found.append(tool_call)
+
+                                # 检查完成原因
+                                if choice.get('finish_reason') == 'tool_calls':
+                                    print("✅ 工具调用完成")
+
                         except json.JSONDecodeError as e:
-                            print(f"❌ [{chunk_count:03d}] JSON解析错误: {e}")
-                            print(f"    原始数据: {data_str[:200]}...")
-                
-                print("-" * 80)
-                print(f"✅ 测试完成，共处理 {chunk_count} 个数据块")
-                print(f"🔧 工具调用检测: {'成功' if tool_calls_found else '失败'}")
-                
-    except Exception as e:
-        print(f"❌ 测试异常: {e}")
-        import traceback
-        traceback.print_exc()
-
-async def test_tool_call_non_streaming():
-    """测试非流式工具调用"""
-    print("\n🧪 开始测试非流式工具调用...")
-    
-    payload = {
-        "model": "glm-4.5",
-        "messages": [
-            {
-                "role": "user", 
-                "content": "请帮我查询上海的天气"
-            }
-        ],
-        "tools": TEST_TOOLS,
-        "stream": False,
-        "temperature": 0.7
-    }
-    
-    headers = {
-        "Content-Type": "application/json",
-        "Authorization": f"Bearer {TEST_AUTH_TOKEN}"
-    }
-    
-    try:
-        async with httpx.AsyncClient(timeout=30.0) as client:
-            response = await client.post(TEST_URL, json=payload, headers=headers)
-            
-            print(f"📡 响应状态: {response.status_code}")
-            
-            if response.status_code == 200:
-                result = response.json()
-                print("📦 响应结果:")
-                print(json.dumps(result, indent=2, ensure_ascii=False))
-                
-                # 检查工具调用
-                choices = result.get("choices", [])
-                if choices:
-                    message = choices[0].get("message", {})
-                    tool_calls = message.get("tool_calls", [])
-                    if tool_calls:
-                        print(f"🔧 发现 {len(tool_calls)} 个工具调用")
-                        for i, tool_call in enumerate(tool_calls):
-                            print(f"  工具 {i+1}: {tool_call}")
-                    else:
-                        print("❌ ��发现工具调用")
+                            print(f"❌ JSON解析错误: {e}, 数据: {data_str[:200]}")
+
+                print(f"📊 总共接收到 {chunk_count} 个数据块")
+                print(f"🔧 发现 {len(tool_calls_found)} 个工具调用")
+
+                # 分析工具调用格式
+                for i, tool_call in enumerate(tool_calls_found):
+                    print(f"\n🔍 工具调用 {i+1} 分析:")
+                    print(f"  ID: {tool_call.get('id', 'N/A')}")
+                    print(f"  类型: {tool_call.get('type', 'N/A')}")
+
+                    if 'function' in tool_call:
+                        func = tool_call['function']
+                        print(f"  函数名: {func.get('name', 'N/A')}")
+
+                        arguments = func.get('arguments', '')
+                        print(f"  参数类型: {type(arguments)}")
+                        print(f"  参数内容: {arguments}")
+
+                        # 尝试解析参数
+                        if isinstance(arguments, str) and arguments:
+                            try:
+                                parsed_args = json.loads(arguments)
+                                print(f"  ✅ 参数解析成功: {parsed_args}")
+                            except json.JSONDecodeError as e:
+                                print(f"  ❌ 参数解析失败: {e}")
+                        elif isinstance(arguments, dict):
+                            print(f"  ⚠️  参数是对象格式（应该是字符串）: {arguments}")
+
             else:
-                print(f"❌ 请求失败: {response.text}")
-                
+                error_text = response.read().decode('utf-8')
+                print(f"❌ 请求失败: {error_text}")
+
     except Exception as e:
-        print(f"❌ 测试异常: {e}")
-
-async def main():
-    """主测试函数"""
-    print("🚀 开始工具调用修复验证测试")
-    print("=" * 80)
-    
-    # 测试流式工具调用
-    await test_tool_call_streaming()
-    
-    # 等待一下
-    await asyncio.sleep(2)
-    
-    # 测试非流式工具调用
-    await test_tool_call_non_streaming()
-    
-    print("\n" + "=" * 80)
-    print("🎯 测试完成")
+        print(f"❌ 请求异常: {e}")
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    test_tool_call()