Setting the file. One moment. 11.10
OpenAI
TypePythonscripts/skill_index.py
Python·273 lines·11 KB
from
pathlib
import
Path
14
15from inventory_skills import iter_skill_files, parse_frontmatter, tokenize
16
17
18QUERY_EXPANSIONS = [
19 (
20 ("技能组合", "技能栈", "技能包", "配齐", "skill stack", "一套skills", "一套 skills"),
21 "agent-skill-stack build curate skill stack capability workflow project profile local index compare conflicts install 组合 技能栈 能力 工作流 项目",
22 ),
23 (
24 ("找一个", "找个", "找一款", "find a skill", "有没有skill", "有没有 skill", "有没有能"),
25 "find-skills find skills discover install common capability 查找 单个 技能",
26 ),
27 (
28 ("去ai", "ai味", "humanize", "natural writing", "文风", "自然一点"),
29 "humanizer humanize writing rewrite natural style tone voice 文案 改写 自然 文风",
30 ),
31 (
32 ("事实核查", "fact check", "引用", "citation", "可信"),
33 "fact check verify evidence citation grounded accuracy 核查 引用 证据 准确",
34 ),
35 (
36 ("合规", "compliance", "版权", "copyright", "规则"),
37 "compliance policy copyright legal safety audit 合规 版权 规则 审核",
38 ),
39 (
40 ("调研", "research", "对标", "竞品", "搜集"),
41 "research search collect compare benchmark competitor evidence 调研 搜索 收集 对标 竞品",
42 ),
43 (
44 ("发布", "publish", "定时", "schedule"),
45 "publish schedule post upload automation approval 发布 定时 上传 自动化 审批",
46 ),
47 (
48 ("整理", "入库", "知识库", "organize", "knowledge base"),
49 "organize knowledge base notes database deduplicate structure 整理 入库 知识库 去重 结构化",
50 ),
51]
52
53
54def utc_iso(timestamp: float | None = None) -> str:
55 moment = datetime.fromtimestamp(timestamp, tz=timezone.utc) if timestamp is not None else datetime.now(timezone.utc)
56 return moment.isoformat()
57
58
59def sha256_file(path: Path) -> str:
60 digest = hashlib.sha256()
61 with path.open("rb") as handle:
62 for chunk in iter(lambda: handle.read(1024 * 1024), b""):
63 digest.update(chunk)
64 return digest.hexdigest()
65
66
67def atomic_write_json(path: Path, payload: dict[str, object]) -> None:
68 path.parent.mkdir(parents=True, exist_ok=True)
69 with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle:
70 json.dump(payload, handle, ensure_ascii=False, indent=2)
71 handle.write("\n")
72 temporary = handle.name
73 os.replace(temporary, path)
74
75
76def infer_scope(skill_file: Path, project_root: Path | None) -> str:
77 if project_root is not None:
78 try:
79 skill_file.relative_to(project_root)
80 return "project"
81 except ValueError:
82 pass
83 return "global"
84
85
86def build_record(skill_file: Path, root: Path, project_root: Path | None) -> dict[str, object]:
87 text = skill_file.read_text(encoding="utf-8", errors="replace")
88 metadata, issues = parse_frontmatter(text[:300_000])
89 name = metadata.get("name", "")
90 description = metadata.get("description", "")
91 headings = [
92 re.sub(r"\s+#+$", "", heading).strip()
93 for heading in re.findall(r"^#{1,3}\s+(.+)$", text, flags=re.MULTILINE)
94 ][:40]
95 capability_terms = sorted(tokenize(" ".join([name, description, *headings])))
96 if name and skill_file.parent.name != name:
97 issues.append(f"directory name '{skill_file.parent.name}' differs from skill name '{name}'")
98 fingerprint = sha256_file(skill_file)
99 summary = re.sub(r"\s+", " ", description).strip()
100 if len(summary) > 360:
101 summary = summary[:357].rstrip() + "..."
102 return {
103 "id": f"{name or 'invalid'}:{fingerprint[:12]}",
104 "name": name,
105 "display_name": name.replace("-", " ").strip().title() if name else "Invalid Skill",
106 "summary": summary,
107 "scope": infer_scope(skill_file, project_root),
108 "aliases": capability_terms[:80],
109 "capability_terms": capability_terms,
110 "headings": headings,
111 "last_local_change": utc_iso(skill_file.stat().st_mtime),
112 "issues": issues,
113 "technical": {
114 "source_root": str(root),
115 "path": str(skill_file.parent),
116 "skill_file_fingerprint": fingerprint,
117 },
118 }
119
120
121def build_index(args: argparse.Namespace) -> int:
122 project_root = Path(args.project_root).expanduser().resolve() if args.project_root else None
123 roots: list[Path] = []
124 missing: list[str] = []
125 for raw in args.root:
126 root = Path(os.path.expandvars(os.path.expanduser(raw))).resolve()
127 if root.is_dir():
128 roots.append(root)
129 else:
130 missing.append(str(root))
131
132 records: list[dict[str, object]] = []
133 seen: set[Path] = set()
134 for root in roots:
135 for skill_file in iter_skill_files(root):
136 resolved = skill_file.resolve()
137 if resolved in seen:
138 continue
139 seen.add(resolved)
140 records.append(build_record(skill_file, root, project_root))
141 records.sort(key=lambda item: (str(item.get("name", "")), str(item["technical"]["path"])))
142
143 names: dict[str, list[str]] = {}
144 for record in records:
145 if record["name"]:
146 names.setdefault(str(record["name"]), []).append(str(record["id"]))
147 duplicates = {name: ids for name, ids in sorted(names.items()) if len(ids) > 1}
148
149 output = Path(os.path.expandvars(os.path.expanduser(args.output))).resolve()
150 payload: dict[str, object] = {
151 "schema": 1,
152 "generated_at": utc_iso(),
153 "roots": [str(root) for root in roots],
154 "missing_roots": missing,
155 "skills": records,
156 "duplicates": duplicates,
157 "privacy": "This index stores Skill metadata only. It contains no prompts, usage history, or routing feedback.",
158 }
159 atomic_write_json(output, payload)
160 print(json.dumps({
161 "status": "built",
162 "output": str(output),
163 "skills_indexed": len(records),
164 "duplicate_names": len(duplicates),
165 "missing_roots": missing,
166 }, ensure_ascii=False, indent=2))
167 return 0
168
169
170def expanded_query(query: str) -> str:
171 lower = query.lower()
172 additions = [terms for triggers, terms in QUERY_EXPANSIONS if any(trigger in lower for trigger in triggers)]
173 return " ".join([query, *additions])
174
175
176def score_record(
177 record: dict[str, object],
178 query: str,
179 direct_tokens: set[str],
180 expanded_tokens: set[str],
181) -> tuple[float, list[str]]:
182 name = str(record.get("name", "")).lower()
183 summary = str(record.get("summary", "")).lower()
184 aliases = set(str(item) for item in record.get("aliases", []))
185 capability_terms = set(str(item) for item in record.get("capability_terms", []))
186 lower_query = query.lower().strip()
187 record_tokens = aliases | capability_terms
188 direct_matched = sorted(direct_tokens & record_tokens)
189 helper_matched = sorted((expanded_tokens - direct_tokens) & record_tokens)
190 matched = [*direct_matched, *helper_matched]
191
192 # What the user actually said must outrank generic query-expansion terms.
193 score = float(len(direct_matched) * 4 + len(helper_matched))
194 if lower_query and lower_query == name:
195 score += 20
196 elif lower_query and lower_query in name:
197 score += 10
198 if lower_query and lower_query in summary:
199 score += 8
200 if name and name in direct_tokens:
201 score += 18
202 elif name and name in expanded_tokens:
203 score += 14
204 if direct_tokens:
205 score += 10 * len(direct_matched) / len(direct_tokens)
206 if score > 0 and record.get("scope") == "project":
207 score += 1
208 if record.get("issues"):
209 score -= 2
210 return score, matched
211
212
213def search_index(args: argparse.Namespace) -> int:
214 index_path = Path(os.path.expandvars(os.path.expanduser(args.index))).resolve()
215 payload = json.loads(index_path.read_text(encoding="utf-8"))
216 expanded = expanded_query(args.query)
217 direct_tokens = tokenize(args.query)
218 expanded_tokens = tokenize(expanded)
219 results: list[dict[str, object]] = []
220 for record in payload.get("skills", []):
221 score, matched = score_record(record, args.query, direct_tokens, expanded_tokens)
222 if score <= 0:
223 continue
224 results.append({
225 "name": record.get("name"),
226 "display_name": record.get("display_name"),
227 "summary": record.get("summary"),
228 "scope": record.get("scope"),
229 "score": round(score, 3),
230 "matched_terms": matched[:20],
231 "issues": record.get("issues", []),
232 "technical": record.get("technical", {}),
233 })
234 results.sort(key=lambda item: (-float(item["score"]), str(item["name"])))
235 results = results[:args.limit]
236
237 if args.format == "simple":
238 for position, result in enumerate(results, start=1):
239 scope = "项目内" if result["scope"] == "project" else "全局"
240 print(f"{position}. {result['display_name']}（{scope}）— {result['summary']}")
241 else:
242 print(json.dumps({
243 "query": args.query,
244 "expanded_query": expanded,
245 "results": results,
246 "notice": "Search scores are retrieval hints, not quality or installation scores.",
247 }, ensure_ascii=False, indent=2))
248 return 0
249
250
251def main() -> int:
252 parser = argparse.ArgumentParser(description=__doc__)
253 subparsers = parser.add_subparsers(dest="command", required=True)
254
255 build = subparsers.add_parser("build", help="Build or refresh a local Skill index")
256 build.add_argument("--root", action="append", required=True, help="Skill root; repeatable")
257 build.add_argument("--output", required=True, help="Index JSON output path")
258 build.add_argument("--project-root", help="Optional project root used to mark project-scoped Skills")
259 build.set_defaults(handler=build_index)
260
261 search = subparsers.add_parser("search", help="Search a previously built local Skill index")
262 search.add_argument("--index", required=True, help="Index JSON path")
263 search.add_argument("--query", required=True, help="Natural-language capability query")
264 search.add_argument("--limit", type=int, default=10)
265 search.add_argument("--format", choices=("json", "simple"), default="json")
266 search.set_defaults(handler=search_index)
267
268 args = parser.parse_args()
269 return int(args.handler(args))
270
271
272if __name__ == "__main__":
273 raise SystemExit(main())