Add label master skill
This commit is contained in:
wuyang6
+331
@@ -0,0 +1,331 @@
|
||||
#!/usr/bin/env python3
|
||||
"""从 Markdown 标签知识库生成机器可读索引。
|
||||
|
||||
这个脚本只做确定性整理，不做语义判断：
|
||||
- 人继续维护 knowledge/ 下的中文 Markdown。
|
||||
- 脚本把标签、intent、function、object、边界文件整理成 JSON manifest。
|
||||
- Agent 和校验脚本优先读取 manifest，避免每次全量扫描知识库。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
SCRIPT_PATH = Path(__file__).resolve()
|
||||
SKILL_ROOT = SCRIPT_PATH.parents[1]
|
||||
KNOWLEDGE_ROOT = SKILL_ROOT / "knowledge"
|
||||
DEFAULT_OUTPUT = KNOWLEDGE_ROOT / "索引" / "label_manifest.json"
|
||||
|
||||
|
||||
def read_text(path: Path) -> str:
|
||||
return path.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def rel(path: Path) -> str:
|
||||
return path.relative_to(SKILL_ROOT).as_posix()
|
||||
|
||||
|
||||
def first_heading(text: str, fallback: str) -> str:
|
||||
match = re.search(r"^#\s+(.+?)\s*$", text, re.MULTILINE)
|
||||
return match.group(1).strip() if match else fallback
|
||||
|
||||
|
||||
def extract_section(text: str, heading: str) -> str:
|
||||
pattern = rf"^##\s+{re.escape(heading)}\s*$"
|
||||
match = re.search(pattern, text, re.MULTILINE)
|
||||
if not match:
|
||||
return ""
|
||||
start = match.end()
|
||||
next_match = re.search(r"^##\s+", text[start:], re.MULTILINE)
|
||||
end = start + next_match.start() if next_match else len(text)
|
||||
return text[start:end].strip()
|
||||
|
||||
|
||||
def extract_bullet_value(text: str, label: str) -> str:
|
||||
match = re.search(rf"^-\s*{re.escape(label)}[：:]\s*(.+?)\s*$", text, re.MULTILINE)
|
||||
return match.group(1).strip() if match else ""
|
||||
|
||||
|
||||
def parse_markdown_table(text: str) -> list[dict[str, str]]:
|
||||
rows: list[dict[str, str]] = []
|
||||
table_lines = [line.strip() for line in text.splitlines() if line.strip().startswith("|")]
|
||||
if len(table_lines) < 2:
|
||||
return rows
|
||||
header = [cell.strip() for cell in table_lines[0].strip("|").split("|")]
|
||||
for line in table_lines[2:]:
|
||||
cells = [cell.strip() for cell in line.strip("|").split("|")]
|
||||
if len(cells) != len(header):
|
||||
continue
|
||||
rows.append(dict(zip(header, cells)))
|
||||
return rows
|
||||
|
||||
|
||||
def parse_enum_values(description: str) -> list[str]:
|
||||
if "枚举" not in description:
|
||||
return []
|
||||
desc = re.sub(r"[，。；;].*$", "", description.split("枚举", 1)[-1])
|
||||
desc = desc.replace("：", ":")
|
||||
if ":" in desc:
|
||||
desc = desc.split(":", 1)[1]
|
||||
parts = re.split(r"[、,，/| 或 和]+", desc)
|
||||
values: list[str] = []
|
||||
for part in parts:
|
||||
cleaned = part.strip().strip("`").strip()
|
||||
cleaned = re.sub(r"[（(].*?[）)]", "", cleaned).strip()
|
||||
if cleaned and re.fullmatch(r"[A-Za-z0-9_]+", cleaned):
|
||||
values.append(cleaned)
|
||||
return values
|
||||
|
||||
|
||||
def parse_label_cards() -> list[dict[str, Any]]:
|
||||
labels_root = KNOWLEDGE_ROOT / "标签"
|
||||
labels: list[dict[str, Any]] = []
|
||||
if not labels_root.exists():
|
||||
return labels
|
||||
|
||||
for path in sorted(labels_root.rglob("*.md")):
|
||||
text = read_text(path)
|
||||
name = first_heading(text, path.stem)
|
||||
domain = path.parent.name
|
||||
label = {
|
||||
"name": name,
|
||||
"domain": domain,
|
||||
"path": rel(path),
|
||||
"old_tag": extract_bullet_value(text, "旧 tag 标签"),
|
||||
"agent_candidate": extract_bullet_value(text, "Agent 包装候选（不代表最终）"),
|
||||
"function_candidate": extract_bullet_value(text, "Function 输出候选"),
|
||||
"recommended_output_shape": extract_bullet_value(text, "推荐输出形态"),
|
||||
"scope": extract_section(text, "适用范围"),
|
||||
"examples": extract_section(text, "典型 Query"),
|
||||
"confusing_labels": extract_section(text, "易混淆标签"),
|
||||
"principles": extract_section(text, "划分原则"),
|
||||
}
|
||||
labels.append(label)
|
||||
return labels
|
||||
|
||||
|
||||
def parse_function_catalog() -> list[dict[str, Any]]:
|
||||
path = KNOWLEDGE_ROOT / "输出能力" / "函数目录.md"
|
||||
if not path.exists():
|
||||
return []
|
||||
text = read_text(path)
|
||||
functions: list[dict[str, Any]] = []
|
||||
sections = list(re.finditer(r"^##\s+(.+?)\s*$", text, re.MULTILINE))
|
||||
for index, match in enumerate(sections):
|
||||
name = match.group(1).strip()
|
||||
start = match.end()
|
||||
end = sections[index + 1].start() if index + 1 < len(sections) else len(text)
|
||||
body = text[start:end].strip()
|
||||
params: dict[str, dict[str, Any]] = {}
|
||||
param_block = ""
|
||||
if "参数：" in body:
|
||||
param_block = body.split("参数：", 1)[1]
|
||||
if "示例：" in param_block:
|
||||
param_block = param_block.split("示例：", 1)[0]
|
||||
for row in parse_markdown_table(param_block):
|
||||
param_name = row.get("参数", "").strip()
|
||||
description = row.get("说明", "").strip()
|
||||
if not param_name:
|
||||
continue
|
||||
params[param_name] = {
|
||||
"description": description,
|
||||
"enum_values": parse_enum_values(description),
|
||||
}
|
||||
functions.append(
|
||||
{
|
||||
"name": name,
|
||||
"path": rel(path),
|
||||
"params": params,
|
||||
"allow_unknown_params": False,
|
||||
"summary": re.sub(r"\s+", " ", extract_summary(body)).strip(),
|
||||
}
|
||||
)
|
||||
return functions
|
||||
|
||||
|
||||
def parse_object_catalog() -> list[dict[str, Any]]:
|
||||
path = KNOWLEDGE_ROOT / "输出能力" / "对象目录.md"
|
||||
if not path.exists():
|
||||
return []
|
||||
text = read_text(path)
|
||||
objects: list[dict[str, Any]] = []
|
||||
|
||||
sections = list(re.finditer(r"^##\s+(.+?)\s*$", text, re.MULTILINE))
|
||||
for index, match in enumerate(sections):
|
||||
name = match.group(1).strip()
|
||||
start = match.end()
|
||||
end = sections[index + 1].start() if index + 1 < len(sections) else len(text)
|
||||
body = text[start:end].strip()
|
||||
|
||||
if name in {"VisionQA 辅助对象", "生成类辅助对象"}:
|
||||
for row in parse_markdown_table(body):
|
||||
object_name = row.get("对象", "").strip()
|
||||
if object_name:
|
||||
objects.append(
|
||||
{
|
||||
"name": object_name,
|
||||
"path": rel(path),
|
||||
"params": {},
|
||||
"allow_unknown_params": True,
|
||||
"summary": row.get("说明", "").strip(),
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
params: dict[str, dict[str, Any]] = {}
|
||||
param_block = body.split("参数：", 1)[1] if "参数：" in body else ""
|
||||
if "示例：" in param_block:
|
||||
param_block = param_block.split("示例：", 1)[0]
|
||||
for row in parse_markdown_table(param_block):
|
||||
param_name = row.get("参数", "").strip()
|
||||
description = row.get("说明", "").strip()
|
||||
if not param_name:
|
||||
continue
|
||||
params[param_name] = {
|
||||
"description": description,
|
||||
"enum_values": parse_enum_values(description),
|
||||
}
|
||||
objects.append(
|
||||
{
|
||||
"name": name,
|
||||
"path": rel(path),
|
||||
"params": params,
|
||||
"allow_unknown_params": False,
|
||||
"summary": re.sub(r"\s+", " ", extract_summary(body)).strip(),
|
||||
}
|
||||
)
|
||||
return objects
|
||||
|
||||
|
||||
def parse_intent_catalog() -> list[dict[str, Any]]:
|
||||
path = KNOWLEDGE_ROOT / "输出能力" / "意图目录.md"
|
||||
if not path.exists():
|
||||
return []
|
||||
text = read_text(path)
|
||||
intents: list[dict[str, Any]] = []
|
||||
for row in parse_markdown_table(text):
|
||||
name = row.get("intent", "").replace("\\|", "|").strip()
|
||||
if not name:
|
||||
continue
|
||||
intents.append(
|
||||
{
|
||||
"name": name,
|
||||
"agent": row.get("Agent", "").strip(),
|
||||
"instruction_type": row.get("指令类型", "").strip(),
|
||||
"scope": row.get("简要范围", "").strip(),
|
||||
"path": rel(path),
|
||||
}
|
||||
)
|
||||
return intents
|
||||
|
||||
|
||||
def extract_summary(body: str) -> str:
|
||||
match = re.search(r"功能范围[：:]\s*(.+?)(?:\n\n|$)", body, re.DOTALL)
|
||||
return match.group(1).strip() if match else ""
|
||||
|
||||
|
||||
def parse_boundary_files() -> list[dict[str, str]]:
|
||||
boundary_root = KNOWLEDGE_ROOT / "边界"
|
||||
if not boundary_root.exists():
|
||||
return []
|
||||
files: list[dict[str, str]] = []
|
||||
for path in sorted(boundary_root.rglob("*.md")):
|
||||
if path.name == "README.md":
|
||||
continue
|
||||
text = read_text(path)
|
||||
parts = path.relative_to(boundary_root).parts
|
||||
category = parts[0] if len(parts) > 1 else "root"
|
||||
files.append({"name": first_heading(text, path.stem), "path": rel(path), "category": category})
|
||||
return files
|
||||
|
||||
|
||||
def parse_dimension_files() -> list[dict[str, str]]:
|
||||
root = KNOWLEDGE_ROOT / "判断维度"
|
||||
if not root.exists():
|
||||
return []
|
||||
files: list[dict[str, str]] = []
|
||||
for path in sorted(root.glob("*.md")):
|
||||
if path.name == "README.md":
|
||||
continue
|
||||
text = read_text(path)
|
||||
files.append({"name": first_heading(text, path.stem), "path": rel(path)})
|
||||
return files
|
||||
|
||||
|
||||
def build_manifest() -> dict[str, Any]:
|
||||
labels = parse_label_cards()
|
||||
functions = parse_function_catalog()
|
||||
objects = parse_object_catalog()
|
||||
intents = parse_intent_catalog()
|
||||
|
||||
old_tags = sorted({item["old_tag"] for item in labels if item.get("old_tag") and item["old_tag"] != "待确认。"})
|
||||
label_names = sorted({item["name"] for item in labels})
|
||||
agent_tags = sorted(set(old_tags) | set(label_names))
|
||||
|
||||
return {
|
||||
"schema_version": 1,
|
||||
"skill": "label-master",
|
||||
"display_name": "标签大师",
|
||||
"description": "标签大师的中控标签知识库机器索引；由 scripts/build_label_manifest.py 从 Markdown 生成。",
|
||||
"counts": {
|
||||
"labels": len(labels),
|
||||
"agent_tags": len(agent_tags),
|
||||
"functions": len(functions),
|
||||
"objects": len(objects),
|
||||
"intents": len(intents),
|
||||
},
|
||||
"agent_tags": agent_tags,
|
||||
"labels": labels,
|
||||
"functions": functions,
|
||||
"objects": objects,
|
||||
"intents": intents,
|
||||
"boundaries": parse_boundary_files(),
|
||||
"dimensions": parse_dimension_files(),
|
||||
}
|
||||
|
||||
|
||||
def write_manifest(path: Path, manifest: dict[str, Any]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
text = json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
||||
path.write_text(text, encoding="utf-8")
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description="生成 label-master 标签大师机器索引")
|
||||
parser.add_argument("--output", default=str(DEFAULT_OUTPUT), help="输出 JSON 路径")
|
||||
parser.add_argument("--check", action="store_true", help="只检查输出文件是否为最新，不写入")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
output = Path(args.output)
|
||||
if not output.is_absolute():
|
||||
output = SKILL_ROOT / output
|
||||
|
||||
manifest = build_manifest()
|
||||
new_text = json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
||||
|
||||
if args.check:
|
||||
if not output.exists():
|
||||
print(f"ERROR: manifest 不存在：{output}", file=sys.stderr)
|
||||
return 1
|
||||
old_text = output.read_text(encoding="utf-8")
|
||||
if old_text != new_text:
|
||||
print(f"ERROR: manifest 不是最新：{output}", file=sys.stderr)
|
||||
return 1
|
||||
print(f"OK: manifest 已是最新：{output}")
|
||||
return 0
|
||||
|
||||
output.parent.mkdir(parents=True, exist_ok=True)
|
||||
output.write_text(new_text, encoding="utf-8")
|
||||
print(f"Wrote {output}")
|
||||
print(json.dumps(manifest["counts"], ensure_ascii=False, sort_keys=True))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user