Spaces:
Paused
Paused
File size: 5,803 Bytes
fd21f34 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 |
#!/usr/bin/env python3
"""
简化的性能测试，避免过多日志输出
"""
import sys
import os
import time
import json
import logging
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# 临时禁用日志以避免性能测试中的噪音
logging.getLogger().setLevel(logging.CRITICAL)
from app.utils.sse_tool_handler import SSEToolHandler
def test_optimized_performance():
"""测试优化后的性能"""
print("🧪 测试优化后的 JSON 修复性能\n")
# 测试用例
test_cases = [
{
"name": "简单JSON",
"input": '{"command":"echo hello","description":"test"}',
"iterations": 100
},
{
"name": "复杂命令行参数",
"input": '{"command":"echo \\"添加更多内容\\uff1a$(date)\\\\\\" >> \\\\\\"C:\\\\\\\\Users\\\\\\\\test\\\\\\\\1.txt\\\\\\"\\"","description":"test"}',
"iterations": 50
},
{
"name": "缺少开始括号",
"input": '"command":"echo hello","description":"test"}',
"iterations": 50
},
{
"name": "Windows路径问题",
"input": '{"path":"C:\\\\\\\\Users\\\\\\\\Documents","command":"dir"}',
"iterations": 50
}
]
handler = SSEToolHandler("test-model", stream=False)
total_time = 0
total_iterations = 0
for test_case in test_cases:
print(f"测试: {test_case['name']}")
print(f" 输入长度: {len(test_case['input'])} 字符")
print(f" 迭代次数: {test_case['iterations']}")
# 预热
for _ in range(5):
handler._fix_tool_arguments(test_case['input'])
# 性能测试
start_time = time.time()
for _ in range(test_case['iterations']):
result = handler._fix_tool_arguments(test_case['input'])
end_time = time.time()
duration = end_time - start_time
if duration > 0:
avg_time = duration / test_case['iterations'] * 1000 # 毫秒
throughput = test_case['iterations'] / duration
else:
avg_time = 0
throughput = float('inf')
print(f" 总时间: {duration:.4f}s")
print(f" 平均时间: {avg_time:.4f}ms")
print(f" 吞吐量: {throughput:.1f} ops/s")
total_time += duration
total_iterations += test_case['iterations']
# 验证结果正确性
try:
parsed = json.loads(result)
print(f" ✅ 结果有效")
except:
print(f" ❌ 结果无效")
print()
print(f"📊 总体性能:")
print(f" 总时间: {total_time:.4f}s")
print(f" 总迭代: {total_iterations}")
if total_time > 0:
print(f" 平均性能: {total_iterations/total_time:.1f} ops/s")
print(f" 平均延迟: {total_time/total_iterations*1000:.4f}ms")
else:
print(f" 平均性能: ∞ ops/s")
print(f" 平均延迟: 0.0000ms")
def test_code_simplification_benefits():
"""测试代码简化的好处"""
print("\n🧪 测试代码简化的好处\n")
# 测试不同复杂度的JSON
test_cases = [
'{"command":"echo hello"}', # 简单
'{"command":"echo \\"hello\\"","description":"test"}', # 转义引号
'"command":"echo hello","description":"test"}', # 缺少开始括号
'{"command":"echo hello > file.txt\\"","description":"test"}', # 多余引号
]
handler = SSEToolHandler("test-model", stream=False)
print("测试各种JSON修复场景:")
for i, test_input in enumerate(test_cases, 1):
print(f"\n场景 {i}: {test_input[:50]}{'...' if len(test_input) > 50 else ''}")
start_time = time.time()
result = handler._fix_tool_arguments(test_input)
end_time = time.time()
duration = (end_time - start_time) * 1000 # 毫秒
try:
parsed = json.loads(result)
status = "✅ 成功"
except:
status = "❌ 失败"
print(f" 处理时间: {duration:.4f}ms")
print(f" 修复状态: {status}")
print(f" 结果长度: {len(result)} 字符")
def test_memory_efficiency():
"""测试内存效率"""
print("\n🧪 测试内存效率\n")
try:
import psutil
process = psutil.Process()
# 基线内存
baseline_memory = process.memory_info().rss / 1024 / 1024 # MB
print(f"基线内存: {baseline_memory:.2f} MB")
handler = SSEToolHandler("test-model", stream=False)
# 测试大量小JSON
test_data = '{"command":"echo test","description":"test"}'
start_memory = process.memory_info().rss / 1024 / 1024
for i in range(100):
result = handler._fix_tool_arguments(test_data)
end_memory = process.memory_info().rss / 1024 / 1024
print(f"处理100次后内存: {end_memory:.2f} MB")
print(f"内存增长: {end_memory - baseline_memory:.2f} MB")
print(f"平均每次处理: {(end_memory - start_memory) / 100 * 1024:.2f} KB")
except ImportError:
print("psutil 未安装，跳过内存测试")
if __name__ == "__main__":
test_optimized_performance()
test_code_simplification_benefits()
test_memory_efficiency()
print("\n🎯 优化总结:")
print("✅ 简化了预处理逻辑")
print("✅ 统一了修复流程")
print("✅ 减少了代码复杂度")
print("✅ 保持了修复质量")
print("✅ 提高了可维护性")
|