skill_loader.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400
  1. """
  2. Skill Loader - 从 Markdown 文件加载 Skills
  3. 支持两种格式:
  4. 格式1 - YAML Frontmatter(推荐):
  5. ---
  6. name: skill-name
  7. description: Skill description
  8. category: category-name
  9. scope: agent:*
  10. parent: parent-id
  11. ---
  12. ## When to use
  13. - Use case 1
  14. - Use case 2
  15. ## Guidelines
  16. - Guideline 1
  17. - Guideline 2
  18. 格式2 - 行内元数据(向后兼容):
  19. # Skill Name
  20. > category: web-automation
  21. > scope: agent:*
  22. ## Description
  23. ...
  24. ## Guidelines
  25. ...
  26. """
  27. from pathlib import Path
  28. from typing import List, Dict, Optional
  29. import logging
  30. from agent.skill.models import Skill
  31. logger = logging.getLogger(__name__)
  32. class SkillLoader:
  33. """从 Markdown 文件加载 Skills"""
  34. def __init__(self, skills_dir: str):
  35. """
  36. 初始化 SkillLoader
  37. Args:
  38. skills_dir: skills 目录路径
  39. """
  40. self.skills_dir = Path(skills_dir)
  41. if not self.skills_dir.exists():
  42. logger.warning(f"Skills 目录不存在: {skills_dir}")
  43. def load_all(self) -> List[Skill]:
  44. """
  45. 加载目录下所有 .md 文件
  46. Returns:
  47. Skill 列表
  48. """
  49. if not self.skills_dir.exists():
  50. return []
  51. skills = []
  52. for md_file in self.skills_dir.glob("*.md"):
  53. try:
  54. skill = self.load_file(md_file)
  55. if skill:
  56. skills.append(skill)
  57. logger.info(f"成功加载 skill: {skill.name} from {md_file.name}")
  58. except Exception as e:
  59. logger.error(f"加载 skill 失败 {md_file}: {e}")
  60. return skills
  61. def load_file(self, file_path: Path) -> Optional[Skill]:
  62. """
  63. 从单个 Markdown 文件加载 Skill
  64. Args:
  65. file_path: Markdown 文件路径
  66. Returns:
  67. Skill 对象,解析失败返回 None
  68. """
  69. if not file_path.exists():
  70. logger.warning(f"文件不存在: {file_path}")
  71. return None
  72. with open(file_path, "r", encoding="utf-8") as f:
  73. content = f.read()
  74. return self.parse_markdown(content, file_path.stem)
  75. def parse_markdown(self, content: str, filename: str) -> Optional[Skill]:
  76. """
  77. 解析 Markdown 内容为 Skill
  78. 支持两种格式:
  79. 格式1 - YAML Frontmatter(推荐):
  80. ---
  81. name: skill-name
  82. description: Skill description
  83. category: category-name
  84. scope: agent:*
  85. ---
  86. ## When to use
  87. - Use case 1
  88. ## Guidelines
  89. - Guideline 1
  90. 格式2 - 行内元数据(向后兼容):
  91. # Skill Name
  92. > category: web-automation
  93. > scope: agent:*
  94. ## Description
  95. 描述内容...
  96. ## Guidelines
  97. - 指导原则1
  98. Args:
  99. content: Markdown 内容
  100. filename: 文件名(不含扩展名)
  101. Returns:
  102. Skill 对象
  103. """
  104. # 检测格式:是否有 YAML frontmatter
  105. if content.strip().startswith("---"):
  106. return self._parse_frontmatter_format(content, filename)
  107. else:
  108. return self._parse_inline_format(content, filename)
  109. def _parse_frontmatter_format(self, content: str, filename: str) -> Optional[Skill]:
  110. """
  111. 解析 YAML frontmatter 格式
  112. ---
  113. name: skill-name
  114. description: Skill description
  115. category: category-name
  116. scope: agent:*
  117. parent: parent-id
  118. ---
  119. ## When to use
  120. ...
  121. ## Guidelines
  122. ...
  123. """
  124. lines = content.split("\n")
  125. # 提取 YAML frontmatter
  126. if not lines[0].strip() == "---":
  127. logger.warning("格式错误:缺少开始的 ---")
  128. return None
  129. frontmatter = {}
  130. i = 1
  131. while i < len(lines):
  132. line = lines[i].strip()
  133. if line == "---":
  134. break
  135. if ":" in line:
  136. key, value = line.split(":", 1)
  137. frontmatter[key.strip()] = value.strip()
  138. i += 1
  139. # 提取元数据
  140. name = frontmatter.get("name") or self._filename_to_title(filename)
  141. description = frontmatter.get("description", "")
  142. category = frontmatter.get("category", "general")
  143. scope = frontmatter.get("scope", "agent:*")
  144. parent_id = frontmatter.get("parent")
  145. # 提取章节内容(从 frontmatter 之后开始)
  146. remaining_content = "\n".join(lines[i+1:])
  147. remaining_lines = remaining_content.split("\n")
  148. # 提取 "When to use" 章节(可选)
  149. when_to_use = self._extract_list_items(remaining_lines, "When to use")
  150. if when_to_use:
  151. # 将 "When to use" 添加到描述中
  152. description += "\n\n适用场景:\n" + "\n".join(f"- {item}" for item in when_to_use)
  153. # 提取 Guidelines
  154. guidelines = self._extract_list_items(remaining_lines, "Guidelines")
  155. # 保存完整的内容(去掉 frontmatter)
  156. content = remaining_content.strip()
  157. # 创建 Skill
  158. return Skill.create(
  159. scope=scope,
  160. name=name,
  161. description=description.strip(),
  162. category=category,
  163. content=content, # 完整的 Markdown 内容
  164. guidelines=guidelines,
  165. parent_id=parent_id,
  166. )
  167. def _parse_inline_format(self, content: str, filename: str) -> Optional[Skill]:
  168. """
  169. 解析行内元数据格式(向后兼容)
  170. # Skill Name
  171. > category: web-automation
  172. > scope: agent:*
  173. ## Description
  174. ...
  175. ## Guidelines
  176. ...
  177. """
  178. lines = content.split("\n")
  179. # 提取标题作为 name
  180. name = self._extract_title(lines) or self._filename_to_title(filename)
  181. # 提取元数据
  182. metadata = self._extract_metadata(lines)
  183. category = metadata.get("category", "general")
  184. scope = metadata.get("scope", "agent:*")
  185. parent_id = metadata.get("parent")
  186. # 提取描述
  187. description = self._extract_section(lines, "Description") or ""
  188. # 提取指导原则
  189. guidelines = self._extract_list_items(lines, "Guidelines")
  190. # 提取完整内容(去掉元数据行和标题行)
  191. content_lines = []
  192. skip_metadata = False
  193. for line in lines:
  194. stripped = line.strip()
  195. # 跳过标题
  196. if stripped.startswith("# "):
  197. continue
  198. # 跳过元数据
  199. if stripped.startswith(">"):
  200. skip_metadata = True
  201. continue
  202. # 如果之前是元数据,跳过后续的空行
  203. if skip_metadata and not stripped:
  204. skip_metadata = False
  205. continue
  206. content_lines.append(line)
  207. content = "\n".join(content_lines).strip()
  208. # 创建 Skill
  209. return Skill.create(
  210. scope=scope,
  211. name=name,
  212. description=description.strip(),
  213. category=category,
  214. content=content, # 完整的 Markdown 内容
  215. guidelines=guidelines,
  216. parent_id=parent_id,
  217. )
  218. def _extract_title(self, lines: List[str]) -> Optional[str]:
  219. """提取 # 标题"""
  220. for line in lines:
  221. line = line.strip()
  222. if line.startswith("# "):
  223. return line[2:].strip()
  224. return None
  225. def _filename_to_title(self, filename: str) -> str:
  226. """将文件名转换为标题(kebab-case -> Title Case)"""
  227. return " ".join(word.capitalize() for word in filename.split("-"))
  228. def _extract_metadata(self, lines: List[str]) -> Dict[str, str]:
  229. """
  230. 提取元数据块(> key: value)
  231. Example:
  232. > category: web-automation
  233. > scope: agent:*
  234. """
  235. metadata = {}
  236. for line in lines:
  237. line = line.strip()
  238. if line.startswith(">"):
  239. # 去掉 > 符号
  240. content = line[1:].strip()
  241. # 分割 key: value
  242. if ":" in content:
  243. key, value = content.split(":", 1)
  244. metadata[key.strip()] = value.strip()
  245. return metadata
  246. def _extract_section(self, lines: List[str], section_name: str) -> Optional[str]:
  247. """
  248. 提取指定章节的内容
  249. Args:
  250. lines: 文件行列表
  251. section_name: 章节名称(如 "Description")
  252. Returns:
  253. 章节内容(纯文本)
  254. """
  255. in_section = False
  256. section_lines = []
  257. for line in lines:
  258. stripped = line.strip()
  259. # 遇到目标章节
  260. if stripped.startswith("## ") and section_name.lower() in stripped.lower():
  261. in_section = True
  262. continue
  263. # 遇到下一个章节,结束
  264. if in_section and stripped.startswith("##"):
  265. break
  266. # 收集章节内容
  267. if in_section:
  268. section_lines.append(line)
  269. return "\n".join(section_lines).strip() if section_lines else None
  270. def _extract_list_items(self, lines: List[str], section_name: str) -> List[str]:
  271. """
  272. 提取指定章节的列表项
  273. Args:
  274. lines: 文件行列表
  275. section_name: 章节名称(如 "Guidelines")
  276. Returns:
  277. 列表项数组
  278. """
  279. section_content = self._extract_section(lines, section_name)
  280. if not section_content:
  281. return []
  282. items = []
  283. for line in section_content.split("\n"):
  284. line = line.strip()
  285. # 匹配列表项(- item 或 * item)
  286. if line.startswith("- ") or line.startswith("* "):
  287. items.append(line[2:].strip())
  288. return items
  289. # 便捷函数
  290. def load_skills_from_dir(skills_dir: Optional[str] = None) -> List[Skill]:
  291. """
  292. 从目录加载所有 Skills
  293. 加载优先级:
  294. 1. 始终加载内置 skills(agent/skills/)
  295. 2. 如果指定了 skills_dir,额外加载该目录的 skills
  296. Args:
  297. skills_dir: 用户自定义 skills 目录路径(可选)
  298. Returns:
  299. Skill 列表(内置 + 自定义)
  300. """
  301. all_skills = []
  302. # 1. 加载内置 skills(agent/skill/skills/)
  303. builtin_skills_dir = Path(__file__).parent / "skills"
  304. if builtin_skills_dir.exists():
  305. loader = SkillLoader(str(builtin_skills_dir))
  306. builtin_skills = loader.load_all()
  307. all_skills.extend(builtin_skills)
  308. logger.info(f"加载了 {len(builtin_skills)} 个内置 skills")
  309. # 2. 加载用户自定义 skills(如果提供)
  310. if skills_dir:
  311. loader = SkillLoader(skills_dir)
  312. custom_skills = loader.load_all()
  313. all_skills.extend(custom_skills)
  314. logger.info(f"加载了 {len(custom_skills)} 个自定义 skills")
  315. return all_skills