FazBrowse GitHub Viewer | Trending | | Home

Tools: [Download Repo ZIP] [View Raw Code] [Original HTTPS Page]

LLM-API-benchmark/test_executor.py at main · inoutcode/LLM-API-benchmark · GitHub

59 lines (49 loc) · 1.75 KB

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

"""

测试任务执行器

"""

import sys

import time

from backend.executor import TaskExecutor

from backend import create_app, db

from backend.models import Task

def test_executor():

"""测试任务执行"""

app = create_app()

with app.app_context():

# 获取第一个任务

task = Task.query.first()

if not task:

print("❌ 没有找到任务，请先创建一个任务")

return

print(f"📋 测试任务: {task.name} (ID: {task.id})")

print(f" 类型: {task.task_type}")

print(f" 配置: {task.config}")

print()

# 创建执行器

executor = TaskExecutor()

print("🚀 开始执行任务...")

executor.execute_async(task)

# 等待执行完成

print("⏳ 等待执行完成...")

time.sleep(5)

# 检查结果

db.session.refresh(task)

print(f"\n📊 执行结果:")

print(f" 状态: {task.status}")

print(f" 最后执行时间: {task.last_run_time}")

if task.task_type == 'perf_test':

results = task.perf_results.all()

print(f" 结果数量: {len(results)}")

if results:

result = results[-1]

print(f" 最新结果 ID: {result.id}")

print(f" 输出文件: {result.output_file}")

else:

results = task.quality_results.all()

print(f" 结果数量: {len(results)}")

if results:

result = results[-1]

print(f" 最新结果 ID: {result.id}")

print(f" 输出文件: {result.output_file}")

if __name__ == '__main__':

test_executor()

Back | FazBrowse Home | New Git URL