Coverage for merco/core/loop_policy.py: 98%

1"""Agent LoopPolicy — 可拔插循环策略""" 

2from __future__ import annotations 

3 

4from abc import ABC, abstractmethod 

5from dataclasses import dataclass 

6 

7 

8@dataclass 

9class LoopState: 

10 """Loop 当前状态""" 

11 iteration: int 

12 tool_calls_count: int 

13 max_tool_calls: int 

14 has_tool_calls: bool 

15 finish_reason: str | None = None 

16 

17 

18@dataclass 

19class LoopDecision: 

20 """Loop 策略决策""" 

21 action: str # "continue" | "exit" 

22 reason: str = "" 

23 

24 

25class LoopPolicy(ABC): 

26 """Agent Loop 策略基类""" 

27 name: str = "" 

28 

29 @abstractmethod 

30 async def decide(self, response: dict, state: LoopState) -> LoopDecision: 

31 """根据 LLM response 和当前 state 决定继续或退出""" 

32 ... 

33 

34 

35class DefaultLoopPolicy(LoopPolicy): 

36 """默认策略：完全复刻当前行为""" 

37 name = "default" 

38 

39 async def decide(self, response: dict, state: LoopState) -> LoopDecision: 

40 if state.has_tool_calls: 

41 return LoopDecision(action="continue", reason="tool_calls present") 

42 return LoopDecision(action="exit", reason="no tool_calls") 

43 

44 

45class LoopPolicyRegistry: 

46 """LoopPolicy 注册表""" 

47 

48 def __init__(self): 

49 self._policies: dict[str, LoopPolicy] = {} 

50 self._active: str = "default" 

51 

52 def register(self, policy: LoopPolicy) -> None: 

53 self._policies[policy.name] = policy 

54 

55 def get(self, name: str) -> LoopPolicy | None: 

56 return self._policies.get(name) 

57 

58 def list(self) -> list[LoopPolicy]: 

59 return list(self._policies.values()) 

60 

61 def set_active(self, name: str) -> None: 

62 if name not in self._policies: 

63 raise KeyError(name) 

64 self._active = name 

65 

66 @property 

67 def active(self) -> LoopPolicy: 

68 return self._policies[self._active]