Coverage for agentos/models/routing_strategy.py: 58%

1""" 

2智能路由策略 — 按任务复杂度自动选择模型。 

3""" 

4 

5from __future__ import annotations 

6 

7from enum import Enum 

8 

9 

10class Complexity(str, Enum): 

11 

12 """复杂度评估结果。""" 

13 

14 SIMPLE = "simple" 

15 MODERATE = "moderate" 

16 COMPLEX_REASONING = "complex_reasoning" 

17 LARGE_CODEBASE = "large_codebase" 

18 CRITICAL = "critical" 

19 

20 

21class Budget(str, Enum): 

22 

23 """成本预算配置。""" 

24 

25 UNLIMITED = "unlimited" 

26 NORMAL = "normal" 

27 TIGHT = "tight" 

28 

29 

30ROUTING_TABLE: dict[tuple[Complexity, Budget], str] = { 

31 (Complexity.SIMPLE, Budget.TIGHT): "minimax-m2.7", 

32 (Complexity.SIMPLE, Budget.NORMAL): "deepseek-v3.1", 

33 (Complexity.SIMPLE, Budget.UNLIMITED): "deepseek-v3.1", 

34 (Complexity.MODERATE, Budget.TIGHT): "deepseek-v3.1", 

35 (Complexity.MODERATE, Budget.NORMAL): "kimi-k2.6", 

36 (Complexity.MODERATE, Budget.UNLIMITED): "kimi-k2.6", 

37 (Complexity.COMPLEX_REASONING, Budget.TIGHT): "deepseek-r1", 

38 (Complexity.COMPLEX_REASONING, Budget.NORMAL): "deepseek-r1", 

39 (Complexity.COMPLEX_REASONING, Budget.UNLIMITED): "claude-opus-4.8", 

40 (Complexity.LARGE_CODEBASE, Budget.TIGHT): "qwen-3.6-plus", 

41 (Complexity.LARGE_CODEBASE, Budget.NORMAL): "qwen-3.6-plus", 

42 (Complexity.LARGE_CODEBASE, Budget.UNLIMITED): "qwen-3.6-plus", 

43 (Complexity.CRITICAL, Budget.TIGHT): "kimi-k2.6", 

44 (Complexity.CRITICAL, Budget.NORMAL): "kimi-k2.6", 

45 (Complexity.CRITICAL, Budget.UNLIMITED): "claude-opus-4.8", 

46} 

47 

48# 简单启发式的复杂度与预算判定规则 

49SIMPLE_KEYWORDS = ["list", "show", "display", "read", "what is", "how to", "print"] 

50COMPLEX_KEYWORDS = ["analyze", "debug", "optimize", "refactor", "design", "architecture", "research"] 

51LARGE_KEYWORDS = ["entire codebase", "all files", "monorepo", "whole project"] 

52CRITICAL_KEYWORDS = ["production", "critical", "must not fail", "urgent"] 

53 

54TIGHT_BUDGET_KEYWORDS = ["cheapest", "save cost", "budget mode", "最便宜", "省钱"] 

55 

56 

57class RoutingStrategy: 

58 """根据任务描述自动决定使用哪个模型。""" 

59 

60 @staticmethod 

61 def assess_complexity(task: str) -> Complexity: 

62 task_lower = task.lower() 

63 if any(kw in task_lower for kw in CRITICAL_KEYWORDS): 

64 return Complexity.CRITICAL 

65 if any(kw in task_lower for kw in LARGE_KEYWORDS): 

66 return Complexity.LARGE_CODEBASE 

67 if any(kw in task_lower for kw in COMPLEX_KEYWORDS): 

68 return Complexity.COMPLEX_REASONING 

69 if any(kw in task_lower for kw in SIMPLE_KEYWORDS): 

70 return Complexity.SIMPLE 

71 return Complexity.MODERATE 

72 

73 @staticmethod 

74 def assess_budget(task: str) -> Budget: 

75 task_lower = task.lower() 

76 if any(kw in task_lower for kw in TIGHT_BUDGET_KEYWORDS): 

77 return Budget.TIGHT 

78 return Budget.NORMAL 

79 

80 @classmethod 

81 def route(cls, task: str, budget: Budget | None = None) -> str: 

82 complexity = cls.assess_complexity(task) 

83 if budget is None: 

84 budget = cls.assess_budget(task) 

85 return ROUTING_TABLE.get((complexity, budget), "kimi-k2.6")