Coverage for merco/skills/processors.py: 41%

1"""Skill-related result processors.""" 

2from __future__ import annotations 

3 

4import logging 

5from merco.core.pipeline import Processor, ProcessContext 

6 

7logger = logging.getLogger("merco.pipeline") 

8 

9 

10class SkillViewProcessor(Processor): 

11 """Skill 注入：skill_view 结果以 user message 注入上下文。 

12 

13 对标 Hermes：skill 内容以 role=user 注入（高优先级 + prompt cache 友好）。 

14 不追加到 system prompt（跨 provider 兼容更好）。 

15 """ 

16 

17 name = "skill_view" 

18 

19 async def process(self, ctx: ProcessContext) -> bool: 

20 if ctx.tool_name != "skill_view": 

21 return False 

22 if "error" in ctx.result: 

23 return False 

24 if "content" not in ctx.result: 

25 return False 

26 

27 skill_name = ctx.result.get("name", "unknown") 

28 skill_content = ctx.result["content"] 

29 content_len = len(skill_content) 

30 

31 # 工具结果只留占位信息 

32 ctx.result["content"] = ( 

33 f"技能 {skill_name} 已加载（{content_len:,} 字符），详见上下文。" 

34 ) 

35 

36 # 完整内容以 user message 注入 

37 user_msg = ( 

38 f"技能 **{skill_name}** 已加载，请遵循以下指引：\n\n" 

39 f"{skill_content}" 

40 ) 

41 # 仍然截断保护：user message 上限 8000 

42 if len(user_msg) > 8000: 

43 user_msg = user_msg[:7800] + "\n\n...(技能内容过长，已截断)" 

44 

45 ctx.extra_messages.append({ 

46 "role": "user", 

47 "content": user_msg, 

48 }) 

49 

50 return False # 不停止管线