6728815174
- example_agent_role（5 角色）+ example_skill（24 技能），可变表带公共列 - 登记表是 Git 侧配置权威的库内影子（类比 meta_function_chain），Git 仍是权威 - sync_agent_registry.py：从 .claude/agents/*.md、.claude/skills/*/SKILL.md 抽 frontmatter + 尽力抽取涉及表名，upsert 入库，幂等可重跑；已登记 5 角色 + 24 技能
98 lines
4.1 KiB
Python
98 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
||
"""智能体/技能登记脚本——把 Git 侧的 agent/skill 元数据影子进库，供看板只读。
|
||
|
||
落库设计 §2.10 的 A 方案：
|
||
- Git 侧（.claude/agents/*.md、.claude/skills/*/SKILL.md）仍是配置权威；
|
||
- 本脚本抽取 frontmatter（name/description/model/tools）+ 尽力抽取涉及的表名，
|
||
upsert 进 example_agent_role / example_skill；
|
||
- 看板只读登记表、不读 Git。幂等可重跑（upsert），配置变更后重跑即同步。
|
||
|
||
跑法（仓库根目录）：.venv/bin/python .claude/skills/db/scripts/sync_agent_registry.py
|
||
"""
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
|
||
from db import connect # 复用锁死的 DSN（与 db.py 同目录，脚本目录自动在 sys.path）
|
||
|
||
ROOT = Path(__file__).resolve().parents[4] # .claude/skills/db/scripts → 仓库根
|
||
AGENTS_DIR = ROOT / ".claude" / "agents"
|
||
SKILLS_DIR = ROOT / ".claude" / "skills"
|
||
TABLE_RE = re.compile(r"\b(muse_[a-z_]+|example_[a-z_]+)\b")
|
||
|
||
|
||
def parse_frontmatter(text: str) -> dict:
|
||
"""解析开头 --- 围栏内的 frontmatter 为 key->字符串（每行按第一个冒号切，值可含冒号/逗号）。"""
|
||
lines = text.splitlines()
|
||
if not lines or lines[0].strip() != "---":
|
||
return {}
|
||
fm = {}
|
||
for line in lines[1:]:
|
||
if line.strip() == "---":
|
||
break
|
||
if ":" in line:
|
||
k, _, v = line.partition(":")
|
||
fm[k.strip()] = v.strip()
|
||
return fm
|
||
|
||
|
||
def sync_roles(conn) -> int:
|
||
n = 0
|
||
for md in sorted(AGENTS_DIR.glob("*.md")):
|
||
fm = parse_frontmatter(md.read_text(encoding="utf-8"))
|
||
role = fm.get("name") or md.stem
|
||
desc = fm.get("description", "")
|
||
display = desc.split("——", 1)[0].strip() or None # "网文写手——…" 取破折号前作显示名
|
||
tools_raw = fm.get("tools")
|
||
tools = json.dumps([t.strip() for t in tools_raw.split(",") if t.strip()],
|
||
ensure_ascii=False) if tools_raw else None
|
||
conn.execute(
|
||
"""INSERT INTO example_agent_role
|
||
(role, display_name, model, tools, responsibility, source_ref, synced_at, creator, updater)
|
||
VALUES (%s,%s,%s,%s::jsonb,%s,%s,CURRENT_TIMESTAMP,'sync_agent_registry','sync_agent_registry')
|
||
ON CONFLICT (tenant_id, role) DO UPDATE SET
|
||
display_name=EXCLUDED.display_name, model=EXCLUDED.model, tools=EXCLUDED.tools,
|
||
responsibility=EXCLUDED.responsibility, source_ref=EXCLUDED.source_ref,
|
||
synced_at=CURRENT_TIMESTAMP, updater='sync_agent_registry'""",
|
||
(role, display, fm.get("model"), tools, desc or None, str(md.relative_to(ROOT))))
|
||
n += 1
|
||
return n
|
||
|
||
|
||
def sync_skills(conn) -> int:
|
||
n = 0
|
||
for sk in sorted(SKILLS_DIR.glob("*/SKILL.md")):
|
||
text = sk.read_text(encoding="utf-8")
|
||
fm = parse_frontmatter(text)
|
||
name = fm.get("name") or sk.parent.name
|
||
tables = sorted(set(TABLE_RE.findall(text))) # 尽力抽取涉及的表名（不区分读写，待人工核）
|
||
model_used = None
|
||
if "MiniMax" in text:
|
||
model_used = "MiniMax-M3"
|
||
elif re.search(r"\bopus\b", text):
|
||
model_used = "opus"
|
||
conn.execute(
|
||
"""INSERT INTO example_skill
|
||
(skill_name, purpose, reads, model_used, source_ref, synced_at, creator, updater)
|
||
VALUES (%s,%s,%s::jsonb,%s,%s,CURRENT_TIMESTAMP,'sync_agent_registry','sync_agent_registry')
|
||
ON CONFLICT (tenant_id, skill_name) DO UPDATE SET
|
||
purpose=EXCLUDED.purpose, reads=EXCLUDED.reads, model_used=EXCLUDED.model_used,
|
||
source_ref=EXCLUDED.source_ref, synced_at=CURRENT_TIMESTAMP, updater='sync_agent_registry'""",
|
||
(name, fm.get("description"),
|
||
json.dumps(tables, ensure_ascii=False) if tables else None,
|
||
model_used, str(sk.relative_to(ROOT))))
|
||
n += 1
|
||
return n
|
||
|
||
|
||
def main():
|
||
with connect() as conn:
|
||
nr = sync_roles(conn)
|
||
ns = sync_skills(conn)
|
||
conn.commit()
|
||
print(f"已登记 {nr} 个角色、{ns} 个技能 → example_agent_role / example_skill")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|