405 lines
16 KiB
Python
405 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""生成并机械校验方法技能分级目录与 Muse 编排技能目录。
|
||
|
||
发现合同（见 AGENTS.md）：任何智能体先读方法技能根目录，再按分类目录中的
|
||
“相对地址”读取所需 `SKILL.md`；编排技能直接从 `muse/技能目录.md` 发现。
|
||
物理位置取自 manifest 的 `skill_path`，技能说明取自 SKILL.md frontmatter。
|
||
|
||
用法：
|
||
.venv/bin/python muse/lifecycle/quality/harness/skills_index.py --check
|
||
.venv/bin/python muse/lifecycle/quality/harness/skills_index.py --write
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import difflib
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any, Optional, Sequence
|
||
|
||
import yaml
|
||
|
||
try:
|
||
from .skill_identity import (
|
||
SKILL_NAME_POLICY_TRANSITIONAL,
|
||
VALID_SKILL_NAME_POLICIES,
|
||
skill_name_validation_error,
|
||
)
|
||
except ImportError: # 允许直接执行本文件。
|
||
from skill_identity import (
|
||
SKILL_NAME_POLICY_TRANSITIONAL,
|
||
VALID_SKILL_NAME_POLICIES,
|
||
skill_name_validation_error,
|
||
)
|
||
|
||
|
||
LIFECYCLE_SECTIONS: tuple[tuple[str, str], ...] = (
|
||
("platform", "0 平台底座"),
|
||
("ingest", "1 素材导入与拆解"),
|
||
("knowledge", "2 知识与上下文供给"),
|
||
("concept", "3 概念与前期设计"),
|
||
("planning", "4 结构与规划"),
|
||
("writing", "5 正文写作与呈现"),
|
||
("review", "6 检测、评分与诊断"),
|
||
("humanization", "7 去 AI 味与人感"),
|
||
("sovereignty", "8 候选主权与落库"),
|
||
)
|
||
|
||
METHOD_ROOT_INDEX_RELPATH = Path(".agent") / "skills" / "目录.md"
|
||
BUSINESS_INDEX_RELPATH = Path("muse") / "技能目录.md"
|
||
LEGACY_INDEX_RELPATHS = (
|
||
Path(".agent") / "skills" / "_index.md",
|
||
Path("muse") / "_skills_index.md",
|
||
)
|
||
MANIFEST_RELPATH = (
|
||
Path("muse")
|
||
/ "lifecycle"
|
||
/ "quality"
|
||
/ "harness"
|
||
/ "manifests"
|
||
/ "skills.json"
|
||
)
|
||
|
||
_METHOD_GROUPS: dict[str, dict[str, Any]] = {
|
||
"规划": {
|
||
"order": 0,
|
||
"title": "规划类技能",
|
||
"description": "概念破题、故事骨架、伏笔铺垫与叙事动力等故事规划方法",
|
||
"usage": "故事构思、搭建节拍表、设计高潮与收尾时",
|
||
"requirement": "产出结构骨架与因果链，不代写正文",
|
||
},
|
||
"写作": {
|
||
"order": 1,
|
||
"title": "写作类技能",
|
||
"description": "人物设计、人物登场、对白暗劲、视角选择与场景执行等写作方法",
|
||
"usage": "正文生成、场景加厚、对白打磨与修辞校准时",
|
||
"requirement": "坚持具体压倒抽象、无心理描写法与白描",
|
||
},
|
||
"诊断": {
|
||
"order": 2,
|
||
"title": "诊断类技能",
|
||
"description": "小说全篇体检、结构断裂诊断与质量短板扫描",
|
||
"usage": "稿件完成、初审体检、分析弃读与拒稿根因时",
|
||
"requirement": "给出客观证据引文，定位具体改进技能",
|
||
},
|
||
}
|
||
|
||
_METHOD_HEADER = """# {title}目录
|
||
|
||
> **发现合同**：从 [`../目录.md`](../目录.md) 进入本分类后，按“相对地址”读取对应 `SKILL.md`。技能名和物理路径来自 manifest，技能说明来自 SKILL.md frontmatter；本文件由 `muse/lifecycle/quality/harness/skills_index.py --write` 生成，手改会被覆盖。
|
||
|
||
本分类共 {count} 个方法技能。
|
||
"""
|
||
|
||
_BUSINESS_HEADER = """# Muse 编排技能目录
|
||
|
||
> **业务发现合同**：主代理与业务编排按“技能文件”读取对应 `SKILL.md`。编排技能不进入智能体框架的运行期目录；物理位置完全取自 manifest 的 `skill_path`。
|
||
|
||
本目录只登记技能名、技能文件和技能说明。分类字段与责任方仍以 [`lifecycle/quality/harness/manifests/skills.json`](lifecycle/quality/harness/manifests/skills.json) 为准，由 `muse/lifecycle/quality/harness` 机械校验。
|
||
|
||
本文件由 `muse/lifecycle/quality/harness/skills_index.py --write` 生成，手改会被覆盖；按生命周期分域，共 {count} 个编排技能。
|
||
"""
|
||
|
||
|
||
def _load_manifest(root: Path) -> dict[str, dict[str, Any]]:
|
||
"""读取 skills.json，返回技能名到条目的唯一映射。"""
|
||
|
||
manifest_path = root / MANIFEST_RELPATH
|
||
if not manifest_path.exists():
|
||
legacy = root / "harness" / "manifests" / "skills.json"
|
||
if legacy.exists():
|
||
manifest_path = legacy
|
||
data = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
entries = data["skills"] if isinstance(data, dict) and "skills" in data else data
|
||
if not isinstance(entries, list):
|
||
raise ValueError("skills.json 的 skills 必须是数组")
|
||
name_policy = (
|
||
data.get("skill_name_policy", SKILL_NAME_POLICY_TRANSITIONAL)
|
||
if isinstance(data, dict)
|
||
else SKILL_NAME_POLICY_TRANSITIONAL
|
||
)
|
||
if name_policy not in VALID_SKILL_NAME_POLICIES:
|
||
raise ValueError("skills.json 的 skill_name_policy 非法")
|
||
|
||
result: dict[str, dict[str, Any]] = {}
|
||
for entry in entries:
|
||
if not isinstance(entry, dict):
|
||
raise ValueError("skills.json 的技能条目必须是对象")
|
||
name = entry.get("name")
|
||
name_error = skill_name_validation_error(name, policy=name_policy)
|
||
if name_error is not None:
|
||
raise ValueError(f"skills.json 的技能名非法：{name!r}：{name_error}")
|
||
assert isinstance(name, str)
|
||
if name in result:
|
||
raise ValueError(f"skills.json 重复登记技能名：{name}")
|
||
result[name] = entry
|
||
return result
|
||
|
||
|
||
def _read_description(skill_md: Path) -> str:
|
||
"""取 SKILL.md frontmatter 的 description 并折叠成单行。"""
|
||
|
||
text = skill_md.read_text(encoding="utf-8")
|
||
if not text.startswith("---\n"):
|
||
raise ValueError(f"{skill_md}: 缺少 frontmatter")
|
||
end = text.index("\n---\n", 4)
|
||
fields = yaml.safe_load(text[4:end])
|
||
description = (fields or {}).get("description")
|
||
if not isinstance(description, str) or not description.strip():
|
||
raise ValueError(f"{skill_md}: frontmatter 缺少非空 description")
|
||
return "".join(line.strip() for line in description.strip().splitlines())
|
||
|
||
|
||
def _skill_link(declared_path: str, index_parent: Path) -> str:
|
||
try:
|
||
relative_path = Path(declared_path).relative_to(index_parent)
|
||
except ValueError as error:
|
||
raise ValueError(
|
||
f"技能路径不在目录索引范围内：{declared_path} -> {index_parent}"
|
||
) from error
|
||
return f"[SKILL.md]({relative_path.as_posix()})"
|
||
|
||
|
||
def _business_table_lines(
|
||
names: Sequence[str],
|
||
disk: dict[str, tuple[str, str]],
|
||
) -> list[str]:
|
||
lines = ["| 技能名 | 技能文件 | 技能说明 |", "|---|---|---|"]
|
||
for name in names:
|
||
declared_path, description = disk[name]
|
||
lines.append(
|
||
f"| {name} | {_skill_link(declared_path, Path('muse'))} | "
|
||
f"{description} |"
|
||
)
|
||
return lines
|
||
|
||
|
||
def _method_table_lines(
|
||
names: Sequence[str],
|
||
disk: dict[str, tuple[str, str]],
|
||
group: str,
|
||
) -> list[str]:
|
||
lines = [
|
||
"| 名称 | 相对地址 | 内容描述 | 使用场景 | 使用要求 |",
|
||
"|------|----------|----------|----------|----------|",
|
||
]
|
||
index_parent = Path(".agent") / "skills" / group
|
||
for name in names:
|
||
declared_path, description = disk[name]
|
||
lines.append(
|
||
f"| {name} | {_skill_link(declared_path, index_parent)} | "
|
||
f"{description} | 命中内容描述中的适用场景时 | "
|
||
"先读取完整技能合同，遵守不适用场景与交接边界 |"
|
||
)
|
||
return lines
|
||
|
||
|
||
def _method_group_from_path(skill_path: str) -> str:
|
||
parts = Path(skill_path).parts
|
||
if len(parts) < 5 or parts[:2] != (".agent", "skills") or parts[-1] != "SKILL.md":
|
||
raise ValueError(f"方法技能路径不符合 .agent/skills/<分类>/<技能>/SKILL.md：{skill_path}")
|
||
group = parts[2]
|
||
if group not in _METHOD_GROUPS:
|
||
raise ValueError(f"方法技能使用了未登记分类目录：{group}")
|
||
return group
|
||
|
||
|
||
def build_method_index_texts(root: Path) -> tuple[dict[Path, str], int]:
|
||
"""生成方法技能根目录和各分类目录全文。"""
|
||
|
||
all_manifest = _load_manifest(root)
|
||
manifest = {
|
||
name: entry
|
||
for name, entry in all_manifest.items()
|
||
if entry.get("invocation") == "model_routed"
|
||
}
|
||
skills_root = root / ".agent" / "skills"
|
||
disk_by_path: dict[str, tuple[str, str]] = {}
|
||
for skill_md in sorted(skills_root.rglob("SKILL.md")):
|
||
relative = skill_md.relative_to(root).as_posix()
|
||
disk_by_path[relative] = (skill_md.parent.name, _read_description(skill_md))
|
||
|
||
disk: dict[str, tuple[str, str]] = {}
|
||
group_by_name: dict[str, str] = {}
|
||
for name, entry in manifest.items():
|
||
declared_path = str(entry.get("skill_path", ""))
|
||
disk_entry = disk_by_path.get(declared_path)
|
||
if disk_entry is None:
|
||
raise ValueError(f"skills.json 登记的方法技能不在磁盘上：{name}")
|
||
if disk_entry[0] != name:
|
||
raise ValueError(f"技能目录名与 name 不一致：{name} -> {declared_path}")
|
||
lifecycle = entry.get("lifecycle")
|
||
if lifecycle not in dict(LIFECYCLE_SECTIONS):
|
||
raise ValueError(f"{name}: 未知 lifecycle={lifecycle!r}")
|
||
disk[name] = (declared_path, disk_entry[1])
|
||
group_by_name[name] = _method_group_from_path(declared_path)
|
||
|
||
manifest_paths = {str(entry.get("skill_path", "")) for entry in manifest.values()}
|
||
unregistered = sorted(set(disk_by_path) - manifest_paths)
|
||
if unregistered:
|
||
raise ValueError("挂载面存在未登记技能：" + ", ".join(unregistered))
|
||
|
||
groups = sorted(
|
||
set(group_by_name.values()),
|
||
key=lambda group: (_METHOD_GROUPS[group]["order"], group),
|
||
)
|
||
root_lines = [
|
||
"| 名称 | 相对地址 | 内容描述 | 使用场景 | 使用要求 |",
|
||
"|------|----------|----------|----------|----------|",
|
||
]
|
||
outputs: dict[Path, str] = {}
|
||
for group in groups:
|
||
metadata = _METHOD_GROUPS[group]
|
||
group_names = sorted(name for name, value in group_by_name.items() if value == group)
|
||
group_relpath = Path(".agent") / "skills" / group / "目录.md"
|
||
root_lines.append(
|
||
f"| {metadata['title']} | [目录.md]({group}/目录.md) | "
|
||
f"{metadata['description']} | "
|
||
f"{metadata['usage']} | {metadata['requirement']} |"
|
||
)
|
||
|
||
lines = [
|
||
_METHOD_HEADER.format(title=metadata["title"], count=len(group_names)).rstrip()
|
||
]
|
||
for lifecycle, section_title in LIFECYCLE_SECTIONS:
|
||
names = [
|
||
name
|
||
for name in group_names
|
||
if manifest[name].get("lifecycle") == lifecycle
|
||
]
|
||
if not names:
|
||
continue
|
||
lines.extend(["", f"## {section_title}", ""])
|
||
lines.extend(_method_table_lines(names, disk, group))
|
||
lines.append("")
|
||
outputs[group_relpath] = "\n".join(lines)
|
||
|
||
root_lines.append("<!-- my-skills-cli:index:v2 -->")
|
||
root_lines.append("")
|
||
outputs[METHOD_ROOT_INDEX_RELPATH] = "\n".join(root_lines)
|
||
return outputs, len(disk)
|
||
|
||
|
||
def build_business_index_text(root: Path) -> tuple[str, int]:
|
||
"""按 manifest 生成 Muse 编排技能目录。"""
|
||
|
||
manifest = {
|
||
name: entry
|
||
for name, entry in _load_manifest(root).items()
|
||
if entry.get("invocation") == "orchestrated"
|
||
}
|
||
disk: dict[str, tuple[str, str]] = {}
|
||
for name, entry in manifest.items():
|
||
declared_path = str(entry.get("skill_path", ""))
|
||
if not declared_path.startswith("muse/"):
|
||
raise ValueError(f"编排技能不在 muse/ 业务树：{name}")
|
||
skill_md = root / declared_path
|
||
if not skill_md.is_file():
|
||
raise ValueError(f"编排技能路径不存在：{declared_path}")
|
||
if skill_md.parent.name != name:
|
||
raise ValueError(f"技能目录名与 name 不一致：{name} -> {declared_path}")
|
||
disk[name] = (declared_path, _read_description(skill_md))
|
||
|
||
lines = [_BUSINESS_HEADER.format(count=len(disk)).rstrip()]
|
||
for lifecycle, title in LIFECYCLE_SECTIONS:
|
||
names = sorted(
|
||
name for name, entry in manifest.items() if entry.get("lifecycle") == lifecycle
|
||
)
|
||
if not names:
|
||
continue
|
||
lines.extend(["", f"## {title}", ""])
|
||
lines.extend(_business_table_lines(names, disk))
|
||
lines.append("")
|
||
return "\n".join(lines), len(disk)
|
||
|
||
|
||
def _diff(relative_path: Path, actual: str, expected: str) -> str:
|
||
return "".join(
|
||
difflib.unified_diff(
|
||
actual.splitlines(keepends=True),
|
||
expected.splitlines(keepends=True),
|
||
fromfile=str(relative_path) + "（现状）",
|
||
tofile=str(relative_path) + "（按磁盘与 skills.json 重建）",
|
||
n=2,
|
||
)
|
||
)
|
||
|
||
|
||
def check(root: Path) -> Optional[str]:
|
||
"""校验所有目录与 manifest、磁盘一致；一致返回 None。"""
|
||
|
||
method_outputs, _ = build_method_index_texts(root)
|
||
business_text, _ = build_business_index_text(root)
|
||
expected_outputs = dict(method_outputs)
|
||
expected_outputs[BUSINESS_INDEX_RELPATH] = business_text
|
||
results: list[str] = []
|
||
for relative_path, expected in sorted(
|
||
expected_outputs.items(), key=lambda item: item[0].as_posix()
|
||
):
|
||
path = root / relative_path
|
||
actual = path.read_text(encoding="utf-8") if path.exists() else ""
|
||
if actual != expected:
|
||
results.append(_diff(relative_path, actual, expected))
|
||
|
||
actual_method_indexes = {
|
||
path.relative_to(root)
|
||
for path in (root / ".agent" / "skills").glob("*/目录.md")
|
||
if path.is_file()
|
||
}
|
||
expected_method_indexes = set(method_outputs) - {METHOD_ROOT_INDEX_RELPATH}
|
||
for stale_path in sorted(actual_method_indexes - expected_method_indexes):
|
||
results.append(f"存在未登记的方法技能分类目录：{stale_path}")
|
||
for legacy_path in LEGACY_INDEX_RELPATHS:
|
||
if (root / legacy_path).exists():
|
||
results.append(f"旧技能索引仍存在，必须删除：{legacy_path}")
|
||
return "\n".join(results) or None
|
||
|
||
|
||
def write(root: Path) -> tuple[int, int]:
|
||
"""重建全部技能目录，不删除未登记文件。"""
|
||
|
||
method_outputs, method_count = build_method_index_texts(root)
|
||
business_text, business_count = build_business_index_text(root)
|
||
for relative_path, text in method_outputs.items():
|
||
path = root / relative_path
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(text, encoding="utf-8")
|
||
business_path = root / BUSINESS_INDEX_RELPATH
|
||
business_path.parent.mkdir(parents=True, exist_ok=True)
|
||
business_path.write_text(business_text, encoding="utf-8")
|
||
return method_count, business_count
|
||
|
||
|
||
def main(argv: Optional[Sequence[str]] = None) -> int:
|
||
parser = argparse.ArgumentParser(description="生成与机械校验中文技能目录")
|
||
parser.add_argument("--root", default=".", help="项目根目录，默认为当前目录")
|
||
group = parser.add_mutually_exclusive_group()
|
||
group.add_argument("--check", action="store_true", help="校验目录与磁盘、skills.json 一致")
|
||
group.add_argument("--write", action="store_true", help="按磁盘与 skills.json 重新生成目录")
|
||
args = parser.parse_args(argv)
|
||
root = Path(args.root).resolve()
|
||
|
||
try:
|
||
if args.write:
|
||
method_count, business_count = write(root)
|
||
print(
|
||
f"已重新生成方法技能分级目录（{method_count} 个）与 "
|
||
f"{BUSINESS_INDEX_RELPATH}（{business_count} 个编排技能）"
|
||
)
|
||
return 0
|
||
difference = check(root)
|
||
except (ValueError, KeyError, FileNotFoundError) as error:
|
||
print(f"技能目录生成失败：{error}")
|
||
return 2
|
||
|
||
if difference is None:
|
||
print("技能目录一致：目录与磁盘、skills.json 相符")
|
||
return 0
|
||
print("技能目录漂移：请用 muse/lifecycle/quality/harness/skills_index.py --write 重新生成")
|
||
print(difference)
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|