248 lines
10 KiB
Python
248 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""智能体/技能登记脚本——把 Git 侧的 agent/skill 元数据影子进库，供看板只读。
|
||
|
||
落库设计 §2.10 的 A 方案：
|
||
- Git 侧（角色合同文档、.agent/agents/*.md、.agent/skills/*/SKILL.md）仍是配置权威；
|
||
- 本脚本从中心角色合同读取责任、模型策略和工具政策，角色 frontmatter 只提供 name/description；
|
||
upsert 进 example_agent_role / example_skill；
|
||
- 看板只读登记表、不读 Git。幂等可重跑（upsert），配置变更后重跑即同步。
|
||
|
||
跑法（仓库根目录）：
|
||
.venv/bin/python .agent/skills/access-database/scripts/sync_agent_registry.py --check
|
||
.venv/bin/python .agent/skills/access-database/scripts/sync_agent_registry.py --check-db
|
||
.venv/bin/python .agent/skills/access-database/scripts/sync_agent_registry.py
|
||
"""
|
||
import argparse
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
|
||
from muse_db import connect
|
||
from muse_role_contract import ROLE_CONTRACT_RELATIVE_PATH, load_role_contract_catalog
|
||
|
||
ROOT = Path(__file__).resolve().parents[4] # .agent/skills/access-database/scripts → 仓库根
|
||
AGENTS_DIR = ROOT / ".agent" / "agents"
|
||
SKILLS_DIR = ROOT / ".agent" / "skills"
|
||
TABLE_RE = re.compile(r"\b(muse_[a-z_]+|example_[a-z_]+)\b")
|
||
SKILL_NAME_RE = re.compile(r"^[a-z][a-z0-9]*(?:-[a-z0-9]+)+$")
|
||
SKILL_FRONTMATTER_KEYS = frozenset({"name", "description", "disable-model-invocation"})
|
||
ROLE_FRONTMATTER_KEYS = frozenset({"name", "description", "skills", "tools"})
|
||
EXPECTED_ROLES = frozenset({"writer", "planner", "extractor", "detector", "judge"})
|
||
|
||
|
||
FRONTMATTER_KEY_RE = re.compile(r"^([A-Za-z0-9_-]+)\s*:")
|
||
|
||
|
||
def parse_frontmatter(text: str) -> dict:
|
||
"""解析开头 --- 围栏内的 frontmatter 为 key->字符串（值可含冒号/逗号）。
|
||
|
||
只有顶格且键名合法的行才算键：block scalar（`description: |`）的续行一律缩进，
|
||
其正文里的冒号（如"要求严格 1:1 复刻"）不得被误读成字段名。
|
||
"""
|
||
lines = text.splitlines()
|
||
if not lines or lines[0].strip() != "---":
|
||
return {}
|
||
fm = {}
|
||
for line in lines[1:]:
|
||
if line.strip() == "---":
|
||
break
|
||
match = FRONTMATTER_KEY_RE.match(line)
|
||
if match:
|
||
_, _, value = line.partition(":")
|
||
fm[match.group(1)] = value.strip()
|
||
return fm
|
||
|
||
|
||
def validate_role_catalog(agents_dir: Path = AGENTS_DIR) -> list[tuple[Path, dict]]:
|
||
"""校验五个角色文件、frontmatter 与宿主无关的身份字段。"""
|
||
|
||
entries = []
|
||
seen = set()
|
||
for md in sorted(agents_dir.glob("*.md")):
|
||
fm = parse_frontmatter(md.read_text(encoding="utf-8"))
|
||
unexpected = sorted(set(fm) - ROLE_FRONTMATTER_KEYS)
|
||
if unexpected:
|
||
raise ValueError(f"{md}: 角色 frontmatter 含未登记字段: {unexpected}")
|
||
role = fm.get("name", "")
|
||
if role != md.stem:
|
||
raise ValueError(f"{md}: 文件名 {md.stem!r} 与 name {role!r} 不一致")
|
||
if role in seen:
|
||
raise ValueError(f"{md}: 角色 name 重复: {role}")
|
||
if not fm.get("description"):
|
||
raise ValueError(f"{md}: frontmatter 缺少 description")
|
||
for key in ("skills", "tools"):
|
||
if key not in fm or not fm[key].strip():
|
||
raise ValueError(f"{md}: frontmatter 缺少 {key}")
|
||
seen.add(role)
|
||
entries.append((md, fm))
|
||
if seen != EXPECTED_ROLES:
|
||
raise ValueError(
|
||
f"角色目录必须精确包含 {sorted(EXPECTED_ROLES)}，实际 {sorted(seen)}"
|
||
)
|
||
if agents_dir.resolve() == AGENTS_DIR.resolve():
|
||
catalog = load_role_contract_catalog(ROOT)
|
||
if set(catalog.roles) != EXPECTED_ROLES:
|
||
raise ValueError("角色合同文档与角色目录不一致")
|
||
return entries
|
||
|
||
|
||
def validate_skill_catalog(skills_dir: Path = SKILLS_DIR) -> list[tuple[Path, dict]]:
|
||
"""校验 Skill 目录、frontmatter 与动作式名称，返回稳定排序的目录项。"""
|
||
|
||
entries = []
|
||
seen = set()
|
||
for sk in sorted(skills_dir.glob("*/SKILL.md")):
|
||
fm = parse_frontmatter(sk.read_text(encoding="utf-8"))
|
||
name = fm.get("name", "")
|
||
directory = sk.parent.name
|
||
unexpected = sorted(set(fm) - SKILL_FRONTMATTER_KEYS)
|
||
if unexpected:
|
||
raise ValueError(f"{sk}: frontmatter 含未登记字段: {unexpected}")
|
||
if not name:
|
||
raise ValueError(f"{sk}: frontmatter 缺少 name")
|
||
if directory != name:
|
||
raise ValueError(f"{sk}: 目录名 {directory!r} 与 name {name!r} 不一致")
|
||
if not SKILL_NAME_RE.fullmatch(name):
|
||
raise ValueError(f"{sk}: name 必须使用小写 动作-对象")
|
||
if name in seen:
|
||
raise ValueError(f"{sk}: Skill name 重复: {name}")
|
||
if not fm.get("description"):
|
||
raise ValueError(f"{sk}: frontmatter 缺少 description")
|
||
invocation_flag = fm.get("disable-model-invocation")
|
||
if invocation_flag is not None and invocation_flag not in {"true", "false"}:
|
||
raise ValueError(
|
||
f"{sk}: disable-model-invocation 必须是 true 或 false"
|
||
)
|
||
seen.add(name)
|
||
entries.append((sk, fm))
|
||
if not entries:
|
||
raise ValueError(f"{skills_dir}: 未发现任何 SKILL.md")
|
||
return entries
|
||
|
||
|
||
def sync_roles(conn, catalog: list[tuple[Path, dict]] | None = None) -> int:
|
||
entries = catalog or validate_role_catalog()
|
||
contracts = load_role_contract_catalog(ROOT)
|
||
names = []
|
||
for md, fm in entries:
|
||
role = fm["name"]
|
||
contract = contracts.for_role(role)
|
||
names.append(role)
|
||
display = contract.display_name
|
||
responsibility = contract.contract_prompt
|
||
conn.execute(
|
||
"""INSERT INTO example_agent_role
|
||
(role, display_name, model, tools, responsibility, source_ref, synced_at, creator, updater)
|
||
VALUES (%s,%s,NULL,NULL,%s,%s,CURRENT_TIMESTAMP,'sync_agent_registry','sync_agent_registry')
|
||
ON CONFLICT (tenant_id, role) DO UPDATE SET
|
||
display_name=EXCLUDED.display_name, model=NULL, tools=NULL,
|
||
responsibility=EXCLUDED.responsibility, source_ref=EXCLUDED.source_ref,
|
||
synced_at=CURRENT_TIMESTAMP, updater='sync_agent_registry', deleted=FALSE""",
|
||
(role, display, responsibility, ROLE_CONTRACT_RELATIVE_PATH.as_posix()))
|
||
conn.execute(
|
||
"""UPDATE example_agent_role
|
||
SET deleted=TRUE, synced_at=CURRENT_TIMESTAMP, updater='sync_agent_registry'
|
||
WHERE tenant_id=0 AND deleted=FALSE AND NOT (role=ANY(%s))""",
|
||
(names,),
|
||
)
|
||
return len(entries)
|
||
|
||
|
||
def sync_skills(conn, catalog: list[tuple[Path, dict]] | None = None) -> int:
|
||
entries = catalog or validate_skill_catalog()
|
||
names = []
|
||
for sk, fm in entries:
|
||
text = sk.read_text(encoding="utf-8")
|
||
name = fm["name"]
|
||
names.append(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', deleted=FALSE""",
|
||
(name, fm.get("description"),
|
||
json.dumps(tables, ensure_ascii=False) if tables else None,
|
||
model_used, str(sk.relative_to(ROOT))))
|
||
conn.execute(
|
||
"""UPDATE example_skill
|
||
SET deleted=TRUE, synced_at=CURRENT_TIMESTAMP, updater='sync_agent_registry'
|
||
WHERE tenant_id=0 AND deleted=FALSE AND NOT (skill_name=ANY(%s))""",
|
||
(names,),
|
||
)
|
||
return len(entries)
|
||
|
||
|
||
def check_database(conn, roles, skills) -> None:
|
||
"""把数据库当前活跃影子与 Git 目录逐项对账。"""
|
||
|
||
contracts = load_role_contract_catalog(ROOT)
|
||
expected_roles = {
|
||
fm["name"]: {
|
||
"source_ref": ROLE_CONTRACT_RELATIVE_PATH.as_posix(),
|
||
"responsibility": contracts.for_role(fm["name"]).contract_prompt,
|
||
"model": None,
|
||
}
|
||
for _path, fm in roles
|
||
}
|
||
actual_roles = {
|
||
row[0]: {"source_ref": row[1], "responsibility": row[2], "model": row[3]}
|
||
for row in conn.execute(
|
||
"SELECT role, source_ref, responsibility, model FROM example_agent_role "
|
||
"WHERE tenant_id=0 AND deleted=FALSE"
|
||
).fetchall()
|
||
}
|
||
expected_skills = {
|
||
fm["name"]: {
|
||
"source_ref": str(path.relative_to(ROOT)),
|
||
"purpose": fm["description"],
|
||
}
|
||
for path, fm in skills
|
||
}
|
||
actual_skills = {
|
||
row[0]: {"source_ref": row[1], "purpose": row[2]}
|
||
for row in conn.execute(
|
||
"SELECT skill_name, source_ref, purpose FROM example_skill "
|
||
"WHERE tenant_id=0 AND deleted=FALSE"
|
||
).fetchall()
|
||
}
|
||
if actual_roles != expected_roles:
|
||
raise ValueError("example_agent_role 与 Git 角色目录不一致，请执行同步")
|
||
if actual_skills != expected_skills:
|
||
raise ValueError("example_skill 与 Git Skill 目录不一致，请执行同步")
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="校验或同步 Agent/Skill 登记")
|
||
mode = parser.add_mutually_exclusive_group()
|
||
mode.add_argument("--check", action="store_true", help="只校验 Git 侧角色与 Skill 目录")
|
||
mode.add_argument("--check-db", action="store_true", help="只读对账 Git 与数据库活跃影子")
|
||
args = parser.parse_args()
|
||
roles = validate_role_catalog()
|
||
skills = validate_skill_catalog()
|
||
if args.check:
|
||
print(f"目录校验通过：{len(roles)} 个角色、{len(skills)} 个 Skill")
|
||
return
|
||
if args.check_db:
|
||
with connect(readonly=True) as conn:
|
||
check_database(conn, roles, skills)
|
||
print(f"数据库影子一致：{len(roles)} 个角色、{len(skills)} 个 Skill")
|
||
return
|
||
with connect() as conn:
|
||
nr = sync_roles(conn, roles)
|
||
ns = sync_skills(conn, skills)
|
||
conn.commit()
|
||
print(f"已登记 {nr} 个角色、{ns} 个技能 → example_agent_role / example_skill")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|