feat(search): app/search/skills.py — 技能注册表与 frontmatter 解析
从 TRM4 core/search/skills.py 保真迁移。提供 parse_frontmatter、 strip_frontmatter、SkillRegistry、discover_skills 四个公共 API。 逻辑完全一致，仅调整导入路径并添加中文 docstring。 17 个单元测试全部通过，覆盖正常/异常/边界场景。 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,195 @@
|
||||
"""技能注册表与 Markdown frontmatter 解析工具。
|
||||
|
||||
提供 Skill 文件的 frontmatter 解析、正文提取、注册表管理和目录扫描功能，
|
||||
供搜索 Agent 装配层使用。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from loguru import logger
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
_FRONTMATTER_FIELDS = {"name", "description", "always", "task_type"}
|
||||
|
||||
|
||||
def _extract_frontmatter_lines(text: str) -> tuple[list[str], int] | None:
|
||||
"""提取 frontmatter 行与正文起始偏移。
|
||||
|
||||
参数:
|
||||
text: 原始 Markdown 文本。
|
||||
|
||||
返回:
|
||||
(frontmatter 行列表, 正文起始字节偏移) 二元组；
|
||||
若不存在完整 frontmatter 则返回 None。
|
||||
"""
|
||||
lines = text.splitlines(keepends=True)
|
||||
if not lines or lines[0].strip() != "---":
|
||||
return None
|
||||
|
||||
offset = len(lines[0])
|
||||
frontmatter_lines: list[str] = []
|
||||
for line in lines[1:]:
|
||||
if line.strip() == "---":
|
||||
return frontmatter_lines, offset + len(line)
|
||||
frontmatter_lines.append(line)
|
||||
offset += len(line)
|
||||
|
||||
logger.debug("frontmatter 缺少结束分隔符，按普通正文处理")
|
||||
return None
|
||||
|
||||
|
||||
def strip_frontmatter(text: str) -> str:
|
||||
"""移除 Markdown 文本开头的 frontmatter，并返回正文。
|
||||
|
||||
参数:
|
||||
text: 原始 Markdown 文本。
|
||||
|
||||
返回:
|
||||
去除 frontmatter 后的正文；若 frontmatter 不完整或不存在，则返回原文。
|
||||
"""
|
||||
extracted = _extract_frontmatter_lines(text)
|
||||
if extracted is None:
|
||||
return text
|
||||
|
||||
_, body_start = extracted
|
||||
return text[body_start:]
|
||||
|
||||
|
||||
def parse_frontmatter(text: str) -> dict[str, str]:
|
||||
"""解析 Markdown frontmatter 中的目标字段。
|
||||
|
||||
仅识别 ``name``、``description``、``always``、``task_type`` 四个字段，
|
||||
其余字段会被忽略。引号包裹的值会自动去除引号。
|
||||
|
||||
参数:
|
||||
text: 原始 Markdown 文本。
|
||||
|
||||
返回:
|
||||
仅包含目标字段的字符串字典。
|
||||
若不存在完整 frontmatter，则返回空字典。
|
||||
"""
|
||||
extracted = _extract_frontmatter_lines(text)
|
||||
if extracted is None:
|
||||
return {}
|
||||
|
||||
frontmatter_lines, _ = extracted
|
||||
parsed: dict[str, str] = {}
|
||||
for raw_line in frontmatter_lines:
|
||||
line = raw_line.strip()
|
||||
if not line or ":" not in line:
|
||||
continue
|
||||
|
||||
key, _, raw_value = line.partition(":")
|
||||
normalized_key = key.strip()
|
||||
if normalized_key not in _FRONTMATTER_FIELDS:
|
||||
continue
|
||||
|
||||
value = raw_value.strip()
|
||||
if len(value) >= 2 and (
|
||||
(value.startswith('"') and value.endswith('"'))
|
||||
or (value.startswith("'") and value.endswith("'"))
|
||||
):
|
||||
value = value[1:-1]
|
||||
parsed[normalized_key] = value
|
||||
|
||||
return parsed
|
||||
|
||||
|
||||
class SkillRegistry:
|
||||
"""管理技能名称到文件路径映射并读取技能正文。
|
||||
|
||||
通过 ``set_paths`` 注入名称→路径映射后，
|
||||
可用 ``read`` 按名读取技能 Markdown 正文（自动去除 frontmatter）。
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._paths: dict[str, Path] = {}
|
||||
|
||||
def set_paths(self, mapping: dict[str, Path]) -> None:
|
||||
"""注入技能名称到文件路径的映射。
|
||||
|
||||
参数:
|
||||
mapping: 技能名到 Markdown 文件路径的映射。
|
||||
"""
|
||||
self._paths = dict(mapping)
|
||||
logger.debug("SkillRegistry 已载入 {} 个技能路径", len(self._paths))
|
||||
|
||||
def read(self, name: str) -> str:
|
||||
"""读取指定技能文件，并返回去除 frontmatter 后的正文。
|
||||
|
||||
参数:
|
||||
name: 技能名称。
|
||||
|
||||
返回:
|
||||
技能 Markdown 正文。
|
||||
|
||||
异常:
|
||||
KeyError: 技能名称未注册时抛出。
|
||||
"""
|
||||
try:
|
||||
path = self._paths[name]
|
||||
except KeyError:
|
||||
logger.error("技能未注册: {}", name)
|
||||
raise
|
||||
|
||||
logger.debug("读取技能文件: name={}, path={}", name, path)
|
||||
return strip_frontmatter(path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def discover_skills(
|
||||
skills_dir: Path,
|
||||
) -> tuple[str, dict[str, str], str, SkillRegistry]:
|
||||
"""扫描 skills 目录，按 frontmatter 分类返回。
|
||||
|
||||
遍历 ``*.md`` 文件，根据 frontmatter 的 ``always`` / ``task_type`` 字段分类：
|
||||
|
||||
- ``always=true`` 的 skill 拼入 ``always_skills_text``
|
||||
- 有 ``task_type`` 的 skill 加入 ``task_skill_map``
|
||||
- 非 always 的 skill 生成 ``catalog_text`` 并注册到 registry
|
||||
|
||||
参数:
|
||||
skills_dir: Skill 文件目录。
|
||||
|
||||
返回:
|
||||
``(always_skills_text, task_skill_map, catalog_text, registry)`` 四元组。
|
||||
"""
|
||||
if not skills_dir.exists():
|
||||
return "", {}, "", SkillRegistry()
|
||||
|
||||
always_parts: list[str] = []
|
||||
task_skill_map: dict[str, str] = {}
|
||||
catalog_lines: list[str] = []
|
||||
registry_paths: dict[str, Path] = {}
|
||||
|
||||
for path in sorted(skills_dir.glob("*.md")):
|
||||
raw = path.read_text(encoding="utf-8")
|
||||
meta = parse_frontmatter(raw)
|
||||
if "name" not in meta:
|
||||
logger.warning("跳过无 name 的 skill 文件: {}", path)
|
||||
continue
|
||||
|
||||
body = strip_frontmatter(raw)
|
||||
name = meta["name"]
|
||||
desc = meta.get("description", "")
|
||||
task_type = meta.get("task_type", "")
|
||||
is_always = str(meta.get("always", "false")).lower() == "true"
|
||||
|
||||
if is_always:
|
||||
always_parts.append(body)
|
||||
else:
|
||||
if task_type:
|
||||
task_skill_map[task_type] = body
|
||||
catalog_lines.append(f"- **{name}**: {desc}")
|
||||
registry_paths[name] = path
|
||||
|
||||
always_text = "\n\n---\n\n".join(always_parts)
|
||||
catalog_text = "\n".join(catalog_lines)
|
||||
|
||||
registry = SkillRegistry()
|
||||
registry.set_paths(registry_paths)
|
||||
|
||||
return always_text, task_skill_map, catalog_text, registry
|
||||
@@ -0,0 +1,214 @@
|
||||
"""app/search/skills 模块的单元测试。
|
||||
|
||||
覆盖 parse_frontmatter、strip_frontmatter、SkillRegistry、discover_skills。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
|
||||
from app.search.skills import (
|
||||
SkillRegistry,
|
||||
discover_skills,
|
||||
parse_frontmatter,
|
||||
strip_frontmatter,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
# ── parse_frontmatter ──────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestParseFrontmatter:
|
||||
"""parse_frontmatter 的测试集。"""
|
||||
|
||||
def test_normal(self) -> None:
|
||||
"""正常 frontmatter 应解析出目标字段。"""
|
||||
text = (
|
||||
"---\n"
|
||||
"name: my_skill\n"
|
||||
'description: "A cool skill"\n'
|
||||
"always: true\n"
|
||||
"task_type: qa\n"
|
||||
"---\n"
|
||||
"Body text here.\n"
|
||||
)
|
||||
result = parse_frontmatter(text)
|
||||
assert result == {
|
||||
"name": "my_skill",
|
||||
"description": "A cool skill",
|
||||
"always": "true",
|
||||
"task_type": "qa",
|
||||
}
|
||||
|
||||
def test_missing_closing_delimiter(self) -> None:
|
||||
"""缺少结束 --- 应返回空字典。"""
|
||||
text = "---\nname: orphan\ndescription: no end\n"
|
||||
assert parse_frontmatter(text) == {}
|
||||
|
||||
def test_no_frontmatter(self) -> None:
|
||||
"""不以 --- 开头的文本应返回空字典。"""
|
||||
text = "Just plain markdown.\n"
|
||||
assert parse_frontmatter(text) == {}
|
||||
|
||||
def test_ignores_unknown_fields(self) -> None:
|
||||
"""非目标字段应被忽略。"""
|
||||
text = "---\nname: s1\nauthor: someone\n---\nBody\n"
|
||||
result = parse_frontmatter(text)
|
||||
assert result == {"name": "s1"}
|
||||
assert "author" not in result
|
||||
|
||||
def test_single_quoted_value(self) -> None:
|
||||
"""单引号包裹的值应去除引号。"""
|
||||
text = "---\nname: 'quoted_name'\n---\nBody\n"
|
||||
result = parse_frontmatter(text)
|
||||
assert result["name"] == "quoted_name"
|
||||
|
||||
|
||||
# ── strip_frontmatter ──────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestStripFrontmatter:
|
||||
"""strip_frontmatter 的测试集。"""
|
||||
|
||||
def test_strips_frontmatter(self) -> None:
|
||||
"""正常情况下应去除 frontmatter，返回正文。"""
|
||||
text = "---\nname: x\n---\nBody content.\n"
|
||||
assert strip_frontmatter(text) == "Body content.\n"
|
||||
|
||||
def test_no_frontmatter_returns_original(self) -> None:
|
||||
"""无 frontmatter 时应返回原文。"""
|
||||
text = "No frontmatter here.\n"
|
||||
assert strip_frontmatter(text) == text
|
||||
|
||||
def test_incomplete_frontmatter_returns_original(self) -> None:
|
||||
"""不完整 frontmatter（缺结束符）应返回原文。"""
|
||||
text = "---\nname: x\nstill going\n"
|
||||
assert strip_frontmatter(text) == text
|
||||
|
||||
|
||||
# ── SkillRegistry ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestSkillRegistry:
|
||||
"""SkillRegistry 的测试集。"""
|
||||
|
||||
def test_read_normal(self, tmp_path: Path) -> None:
|
||||
"""read 应返回去除 frontmatter 后的正文。"""
|
||||
skill_file = tmp_path / "skill_a.md"
|
||||
skill_file.write_text("---\nname: skill_a\n---\nSkill A body.\n", encoding="utf-8")
|
||||
|
||||
registry = SkillRegistry()
|
||||
registry.set_paths({"skill_a": skill_file})
|
||||
assert registry.read("skill_a") == "Skill A body.\n"
|
||||
|
||||
def test_read_unregistered_raises_key_error(self) -> None:
|
||||
"""读取未注册的技能应抛出 KeyError。"""
|
||||
registry = SkillRegistry()
|
||||
with pytest.raises(KeyError):
|
||||
registry.read("nonexistent")
|
||||
|
||||
|
||||
# ── discover_skills ────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestDiscoverSkills:
|
||||
"""discover_skills 的测试集。"""
|
||||
|
||||
def _write_skill(self, path: Path, content: str) -> None:
|
||||
"""辅助方法：写入技能文件。"""
|
||||
path.write_text(content, encoding="utf-8")
|
||||
|
||||
def test_always_skill(self, tmp_path: Path) -> None:
|
||||
"""always=true 的技能应出现在 always_text 中。"""
|
||||
self._write_skill(
|
||||
tmp_path / "always_skill.md",
|
||||
"---\nname: a1\ndescription: always on\nalways: true\n---\nAlways body.\n",
|
||||
)
|
||||
always_text, task_map, catalog_text, registry = discover_skills(tmp_path)
|
||||
|
||||
assert "Always body." in always_text
|
||||
assert task_map == {}
|
||||
assert "a1" not in catalog_text
|
||||
|
||||
def test_task_type_skill(self, tmp_path: Path) -> None:
|
||||
"""有 task_type 的非 always 技能应出现在 task_skill_map 中。"""
|
||||
self._write_skill(
|
||||
tmp_path / "task_skill.md",
|
||||
"---\nname: t1\ndescription: task skill\ntask_type: qa\n---\nTask body.\n",
|
||||
)
|
||||
always_text, task_map, catalog_text, registry = discover_skills(tmp_path)
|
||||
|
||||
assert always_text == ""
|
||||
assert task_map == {"qa": "Task body.\n"}
|
||||
assert "t1" in catalog_text
|
||||
|
||||
def test_catalog_skill(self, tmp_path: Path) -> None:
|
||||
"""普通技能应出现在 catalog_text 和 registry 中。"""
|
||||
self._write_skill(
|
||||
tmp_path / "cat_skill.md",
|
||||
"---\nname: c1\ndescription: catalog skill\n---\nCatalog body.\n",
|
||||
)
|
||||
always_text, task_map, catalog_text, registry = discover_skills(tmp_path)
|
||||
|
||||
assert always_text == ""
|
||||
assert task_map == {}
|
||||
assert "**c1**" in catalog_text
|
||||
assert "catalog skill" in catalog_text
|
||||
assert registry.read("c1") == "Catalog body.\n"
|
||||
|
||||
def test_empty_directory(self, tmp_path: Path) -> None:
|
||||
"""空目录应返回所有空值。"""
|
||||
always_text, task_map, catalog_text, registry = discover_skills(tmp_path)
|
||||
|
||||
assert always_text == ""
|
||||
assert task_map == {}
|
||||
assert catalog_text == ""
|
||||
|
||||
def test_nonexistent_directory(self, tmp_path: Path) -> None:
|
||||
"""不存在的目录应返回所有空值。"""
|
||||
missing = tmp_path / "no_such_dir"
|
||||
always_text, task_map, catalog_text, registry = discover_skills(missing)
|
||||
|
||||
assert always_text == ""
|
||||
assert task_map == {}
|
||||
assert catalog_text == ""
|
||||
|
||||
def test_mixed_skills(self, tmp_path: Path) -> None:
|
||||
"""混合 always / task_type / catalog 技能应正确分类。"""
|
||||
self._write_skill(
|
||||
tmp_path / "01_always.md",
|
||||
"---\nname: a1\ndescription: always\nalways: true\n---\nA body.\n",
|
||||
)
|
||||
self._write_skill(
|
||||
tmp_path / "02_task.md",
|
||||
"---\nname: t1\ndescription: task\ntask_type: summary\n---\nT body.\n",
|
||||
)
|
||||
self._write_skill(
|
||||
tmp_path / "03_catalog.md",
|
||||
"---\nname: c1\ndescription: catalog\n---\nC body.\n",
|
||||
)
|
||||
always_text, task_map, catalog_text, registry = discover_skills(tmp_path)
|
||||
|
||||
assert "A body." in always_text
|
||||
assert task_map == {"summary": "T body.\n"}
|
||||
assert "**t1**" in catalog_text
|
||||
assert "**c1**" in catalog_text
|
||||
# always 技能不应出现在 catalog 或 registry 中
|
||||
assert "a1" not in catalog_text
|
||||
|
||||
def test_skip_no_name(self, tmp_path: Path) -> None:
|
||||
"""没有 name 字段的技能文件应被跳过。"""
|
||||
self._write_skill(
|
||||
tmp_path / "bad.md",
|
||||
"---\ndescription: no name\n---\nBody.\n",
|
||||
)
|
||||
always_text, task_map, catalog_text, registry = discover_skills(tmp_path)
|
||||
|
||||
assert always_text == ""
|
||||
assert task_map == {}
|
||||
assert catalog_text == ""
|
||||
Reference in New Issue
Block a user