Coverage for merco/context/processors/cache_optimize.py: 91%

1"""CacheOptimizeProcessor — 提高 LLM 缓存命中率""" 

2from __future__ import annotations 

3 

4from merco.context.pipeline import ContextProcessor 

5 

6 

7class CacheOptimizeProcessor(ContextProcessor): 

8 """缓存优化：重排序让稳定内容在前""" 

9 name = "cache_optimize" 

10 

11 async def process(self, messages: list[dict], **kwargs) -> list[dict]: 

12 stable = [] 

13 volatile = [] 

14 

15 for msg in messages: 

16 if self._is_stable(msg): 

17 stable.append(msg) 

18 else: 

19 volatile.append(msg) 

20 

21 return stable + volatile 

22 

23 def _is_stable(self, msg: dict) -> bool: 

24 """判断消息是否稳定（可缓存）""" 

25 role = msg.get("role", "") 

26 if role == "system": 

27 return True 

28 content = str(msg.get("content", "")) 

29 if "[Earlier conversation summary]" in content: 

30 return True 

31 if "[memory]" in content: 

32 return True 

33 return False