Coverage for tools / skills.py: 33%

36 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-03-29 02:55 +0800

1""" 

2内置工具:Skills 相关工具 

3""" 

4 

5import json 

6from pydantic import BaseModel, Field 

7from qrclaw.tools.registry import register 

8from qrclaw.skills.registry import SkillRegistry 

9from qrclaw.logger import get_logger 

10 

11logger = get_logger("qrclaw.tools.skills") 

12 

13 

14# ── 参数模型 ────────────────────────────────────────────── 

15 

16class UseSkillArgs(BaseModel): 

17 skill_name: str = Field(description="要使用的技能名称,例如 analyze-project") 

18 args: dict = Field(default_factory=dict, description="技能参数,根据技能定义传入") 

19 

20 

21def get_skill_registry() -> SkillRegistry: 

22 """获取当前 agent 的 Skill Registry(从 agent 的 workspace 加载)""" 

23 from qrclaw.agent import get_session 

24 from qrclaw.workspace import Workspace 

25 session = get_session() 

26 # 通过 session 路径反推 workspace(sessions_dir 的上两级是 agent root) 

27 if session is not None: 

28 agent_root = session._path.parent.parent 

29 skills_dir = agent_root / "skills" 

30 else: 

31 skills_dir = Workspace("default").skills_dir 

32 registry = SkillRegistry() 

33 registry.load_from_dir(skills_dir) 

34 return registry 

35 

36 

37# ── 工具函数 ────────────────────────────────────────────── 

38 

39@register(description="使用指定的技能(Skill)来完成复杂任务,技能是预定义的工作流", args_model=UseSkillArgs) 

40def use_skill(skill_name: str, args: dict = None) -> str: 

41 """ 

42 使用技能 

43 

44 Args: 

45 skill_name: 技能名称 

46 args: 技能参数 

47 

48 Returns: 

49 str: 技能的完整信息和执行指导 

50 """ 

51 if args is None: 

52 args = {} 

53 

54 logger.debug(f"使用技能: {skill_name}, 参数: {args}") 

55 

56 # 获取技能注册表 

57 registry = get_skill_registry() 

58 

59 # 检查技能是否存在 

60 if not registry.has_skill(skill_name): 

61 available_skills = ", ".join(registry.get_skills_list()) 

62 error_msg = f"错误:找不到技能 '{skill_name}'\n\n可用技能:\n{available_skills}" 

63 logger.warning(error_msg) 

64 return error_msg 

65 

66 # 获取技能完整信息 

67 skill = registry.get_skill(skill_name) 

68 

69 # 构建技能执行指导 

70 lines = [ 

71 f"## 开始执行技能:{skill.name}", 

72 "", 

73 skill.get_full_info(), 

74 "", 

75 "## 执行指导", 

76 "请按照上述步骤,逐步完成这个技能的任务。你可以:", 

77 "- 使用 read_file、write_file、run_shell 等工具", 

78 "- 按照步骤顺序执行", 

79 "- 根据实际情况灵活调整", 

80 "", 

81 f"**用户提供的参数**:{json.dumps(args, ensure_ascii=False, indent=2) if args else '无'}", 

82 ] 

83 

84 result = "\n".join(lines) 

85 logger.info(f"注入技能完整信息: {skill_name}") 

86 

87 return result