wehub-resource-sync
351 lines
11 KiB
Python
351 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Scaffold GOD executable skills and migrate agent configs.
|
|
|
|
This script is intentionally conservative: it creates missing `SKILL.md`,
|
|
`skill.json`, and `scripts/run_skill.py` files, but it does not overwrite an
|
|
existing skill's local implementation. Existing skill behavior belongs inside
|
|
that skill directory.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
SKILLS_ROOT = REPO_ROOT / "custom" / "skills"
|
|
CONFIG_PATH = (
|
|
REPO_ROOT
|
|
/ "quick_experiments"
|
|
/ "hypothesis_god_town"
|
|
/ "experiment_1"
|
|
/ "init"
|
|
/ "init_config.json"
|
|
)
|
|
|
|
COMMON_SKILL_IDS = [
|
|
"routine.daily",
|
|
"social.reply",
|
|
"memory.record",
|
|
"map.navigate",
|
|
"safety.respond",
|
|
]
|
|
|
|
PERSONA_SKILL_IDS = [
|
|
"community.coordinate",
|
|
"conflict.mediate",
|
|
"first_aid.basic",
|
|
"notice.write",
|
|
"messaging.group",
|
|
"tools.repair",
|
|
"inventory.count",
|
|
"route.localmap",
|
|
"ledger.basic",
|
|
"neighbor.support",
|
|
"class.organize",
|
|
"youth.communicate",
|
|
"writing.feedback",
|
|
"history.localtelling",
|
|
"library.curate",
|
|
"care.basic",
|
|
"chronic.followup",
|
|
"emotion.calm",
|
|
"health.educate",
|
|
"record.shortnote",
|
|
"cooking.lightmeal",
|
|
"listen.relay",
|
|
"shop.run",
|
|
"social.matchmake",
|
|
"community.observe",
|
|
"class.learn",
|
|
"sketch.draw",
|
|
"phone.photolog",
|
|
"computer.basic",
|
|
"peer.communicate",
|
|
"route.recall",
|
|
"garden.basic",
|
|
"story.localpast",
|
|
"writing.hand",
|
|
"neighbor.greet",
|
|
"computer.repair",
|
|
"script.automate",
|
|
"info.research",
|
|
"remote.communicate",
|
|
"privacy.protect",
|
|
"patrol.plan",
|
|
"roster.verify",
|
|
"repair.basic",
|
|
"crowd.guide",
|
|
"radio.comms",
|
|
"vegetable.source",
|
|
"stall.run",
|
|
"price.negotiate",
|
|
"ingredient.advise",
|
|
"gossip.filter",
|
|
]
|
|
|
|
|
|
def safe_dir(skill_id: str) -> str:
|
|
return skill_id.strip().replace("/", "_").replace("\\", "_").replace("..", "_")
|
|
|
|
|
|
def title(skill_id: str) -> str:
|
|
return " ".join(part.capitalize() for part in skill_id.replace(".", " ").split())
|
|
|
|
|
|
def output_name(skill_id: str) -> str:
|
|
return "".join(ch if ch.isalnum() else "_" for ch in skill_id).strip("_")
|
|
|
|
|
|
def default_effects(shared: bool) -> list[str]:
|
|
return (
|
|
["move", "interact", "set_state", "direct_message", "group_message", "remember"]
|
|
if not shared
|
|
else ["move", "interact", "set_state", "remember"]
|
|
)
|
|
|
|
|
|
def default_spec(skill_id: str, *, shared: bool) -> dict[str, Any]:
|
|
return {
|
|
"skill_id": skill_id,
|
|
"name": title(skill_id),
|
|
"description": f"Use {skill_id} as a real executable GOD skill.",
|
|
"shared": shared,
|
|
"effects": default_effects(shared),
|
|
"target_locations": ["park", "cafe", "home"],
|
|
"target_interactions": ["take_walk", "chat_over_coffee", "relax_at_home"],
|
|
"status": "active",
|
|
"emotion": "calm",
|
|
"speech": "我会根据当前场景执行这个能力。",
|
|
"memory_template": f"Used {skill_id}: {{summary}} at {{location_id}}.",
|
|
"failure_strategy": "set_state",
|
|
"strategy": "configured_action",
|
|
"trigger_examples": [f"the agent needs to use {skill_id}"],
|
|
"args_schema": {"type": "object", "additionalProperties": True},
|
|
}
|
|
|
|
|
|
def skill_md(spec: dict[str, Any]) -> str:
|
|
effects = "\n".join(f" - {effect}" for effect in spec["effects"])
|
|
triggers = "\n".join(f" - {item}" for item in spec["trigger_examples"])
|
|
return f"""---
|
|
name: {spec["skill_id"]}
|
|
description: {spec["description"]}
|
|
script: scripts/run_skill.py
|
|
shared: {str(bool(spec["shared"])).lower()}
|
|
effects:
|
|
{effects}
|
|
args_schema: {json.dumps(spec["args_schema"], ensure_ascii=False, separators=(",", ":"))}
|
|
trigger_examples:
|
|
{triggers}
|
|
outputs:
|
|
- state/skill_results/{output_name(spec["skill_id"])}.json
|
|
---
|
|
|
|
# {spec["name"]}
|
|
|
|
This skill is an executable GOD skill. Edit `skill.json` for local targets and
|
|
templates, and edit `scripts/run_skill.py` when the skill needs custom behavior.
|
|
The script returns a standard `SkillResult`; the agent runtime validates and
|
|
applies the effects.
|
|
"""
|
|
|
|
|
|
def scaffold_script(skill_id: str) -> str:
|
|
return f'''#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
SKILL_DIR = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(SKILL_DIR.parent / "_shared"))
|
|
|
|
from agent_skill_runtime import ( # noqa: E402
|
|
build_skill_result,
|
|
choose_available_interaction,
|
|
choose_available_location,
|
|
emit_result,
|
|
load_local_skill_spec,
|
|
memory_effect,
|
|
observation_from,
|
|
parse_args,
|
|
)
|
|
|
|
SKILL_ID = {skill_id!r}
|
|
|
|
|
|
def choose_world_effect(observation: dict, spec: dict) -> dict:
|
|
current = str(observation.get("location_id") or "")
|
|
location_id = choose_available_location(observation, list(spec.get("target_locations") or []))
|
|
if location_id and location_id != current:
|
|
return {{"type": "move", "location_id": location_id, "reason": f"{{SKILL_ID}}: move toward local skill target"}}
|
|
interaction_id = choose_available_interaction(
|
|
observation,
|
|
list(spec.get("target_interactions") or []),
|
|
location_id=current or location_id,
|
|
)
|
|
if interaction_id:
|
|
return {{"type": "interact", "interaction_id": interaction_id, "params": {{"message": spec.get("description", SKILL_ID)}}, "reason": f"{{SKILL_ID}}: perform local skill interaction"}}
|
|
return {{"type": "set_state", "action": str(spec.get("description") or SKILL_ID), "status": str(spec.get("status") or "active"), "emotion": str(spec.get("emotion") or "calm"), "reason": f"{{SKILL_ID}}: no configured target is reachable"}}
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
observation = observation_from(args)
|
|
spec = load_local_skill_spec(SKILL_DIR)
|
|
summary = str(spec.get("description") or SKILL_ID)
|
|
location_id = str(observation.get("location_id") or "unknown")
|
|
template = str(spec.get("memory_template") or "Used {{skill_id}}: {{summary}} at {{location_id}}.")
|
|
world_effect = choose_world_effect(observation, spec)
|
|
result = build_skill_result(
|
|
skill_id=SKILL_ID,
|
|
summary=summary,
|
|
reason=f"{{SKILL_ID}} uses only its local skill.json plus this local script.",
|
|
world_effect=world_effect,
|
|
memory_effects=[
|
|
memory_effect(
|
|
SKILL_ID,
|
|
template.format(skill_id=SKILL_ID, summary=summary, location_id=location_id),
|
|
location_id=location_id,
|
|
)
|
|
],
|
|
confidence=0.72,
|
|
)
|
|
return emit_result(args, SKILL_ID, result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|
|
'''
|
|
|
|
|
|
def write_if_missing(path: Path, text: str) -> bool:
|
|
if path.exists():
|
|
return False
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(text, encoding="utf-8")
|
|
if path.name == "run_skill.py":
|
|
path.chmod(0o755)
|
|
return True
|
|
|
|
|
|
def scaffold_skill(skill_id: str, *, shared: bool) -> list[str]:
|
|
dest = SKILLS_ROOT / safe_dir(skill_id)
|
|
spec = default_spec(skill_id, shared=shared)
|
|
created: list[str] = []
|
|
if write_if_missing(dest / "skill.json", json.dumps(spec, ensure_ascii=False, indent=2) + "\n"):
|
|
created.append(str(dest / "skill.json"))
|
|
if write_if_missing(dest / "SKILL.md", skill_md(spec)):
|
|
created.append(str(dest / "SKILL.md"))
|
|
if write_if_missing(dest / "scripts" / "run_skill.py", scaffold_script(skill_id)):
|
|
created.append(str(dest / "scripts" / "run_skill.py"))
|
|
return created
|
|
|
|
|
|
def load_config(path: Path) -> dict[str, Any]:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def default_skill_ids(agent_id: int, count: int = 5) -> list[str]:
|
|
start = ((agent_id - 1) * count) % len(PERSONA_SKILL_IDS)
|
|
rotated = PERSONA_SKILL_IDS[start:] + PERSONA_SKILL_IDS[:start]
|
|
return rotated[:count]
|
|
|
|
|
|
def normalize_personal_skill_ids(raw: Any, agent_id: int) -> list[str]:
|
|
source = raw if isinstance(raw, list) else []
|
|
known = set(PERSONA_SKILL_IDS)
|
|
result: list[str] = []
|
|
seen: set[str] = set()
|
|
for item in source:
|
|
skill_id = str(item).strip()
|
|
if skill_id in known and skill_id not in seen:
|
|
seen.add(skill_id)
|
|
result.append(skill_id)
|
|
return result or default_skill_ids(agent_id)
|
|
|
|
|
|
def collect_profile_skill_ids(data: dict[str, Any]) -> dict[int, list[str]]:
|
|
result: dict[int, list[str]] = {}
|
|
for agent in data.get("agents", []):
|
|
if not isinstance(agent, dict):
|
|
continue
|
|
agent_id = int(agent.get("agent_id") or agent.get("kwargs", {}).get("id") or 0)
|
|
kwargs = agent.get("kwargs") or {}
|
|
profile = kwargs.get("profile") or {}
|
|
existing = kwargs.get("skill_ids")
|
|
source = existing if isinstance(existing, list) else profile.get("skills")
|
|
result[agent_id] = normalize_personal_skill_ids(source, agent_id)
|
|
return result
|
|
|
|
|
|
def migrate_config(path: Path, profile_skill_ids: dict[int, list[str]]) -> None:
|
|
data = load_config(path)
|
|
for agent in data.get("agents", []):
|
|
if not isinstance(agent, dict):
|
|
continue
|
|
agent_id = int(agent.get("agent_id") or agent.get("kwargs", {}).get("id") or 0)
|
|
kwargs = agent.setdefault("kwargs", {})
|
|
profile = kwargs.setdefault("profile", {})
|
|
profile.pop("skills", None)
|
|
kwargs["enable_skill_runtime"] = True
|
|
kwargs["common_skill_ids"] = list(COMMON_SKILL_IDS)
|
|
kwargs["skill_ids"] = profile_skill_ids.get(agent_id, default_skill_ids(agent_id))
|
|
kwargs.pop("enable_daily_life", None)
|
|
kwargs.pop("daily_life_skill_path", None)
|
|
kwargs.pop("skill_runtime_skill_names", None)
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--config", default=str(CONFIG_PATH))
|
|
parser.add_argument("--all-configs", action="store_true")
|
|
parser.add_argument("--skip-config", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
config_paths = (
|
|
sorted((REPO_ROOT / "quick_experiments").glob("hypothesis_*/experiment_*/init/init_config.json"))
|
|
if args.all_configs
|
|
else [Path(args.config)]
|
|
)
|
|
per_config: dict[str, dict[int, list[str]]] = {}
|
|
for config_path in config_paths:
|
|
config_data = load_config(config_path) if config_path.exists() else {"agents": []}
|
|
per_config[str(config_path)] = collect_profile_skill_ids(config_data)
|
|
|
|
created: list[str] = []
|
|
for skill_id in COMMON_SKILL_IDS:
|
|
created.extend(scaffold_skill(skill_id, shared=True))
|
|
for skill_id in PERSONA_SKILL_IDS:
|
|
created.extend(scaffold_skill(skill_id, shared=False))
|
|
|
|
migrated_configs: list[str] = []
|
|
if not args.skip_config:
|
|
for raw_path, per_agent in per_config.items():
|
|
config_path = Path(raw_path)
|
|
if config_path.exists():
|
|
migrate_config(config_path, per_agent)
|
|
migrated_configs.append(str(config_path))
|
|
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"skills_checked": len(COMMON_SKILL_IDS) + len(PERSONA_SKILL_IDS),
|
|
"created_files": created,
|
|
"configs": migrated_configs,
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|