"""A8 · Agent Skills v1.1 demo. 离线 Skill Registry / Runner：发现、版本解析、输入契约、工具权限、 审批门禁与结果验证。标准库 only；不是真实 LLM / Skills 服务。 运行：python3 A8-Agent-Skills-v1.1-demo.py """ from __future__ import annotations import hashlib import json from dataclasses import dataclass, field from typing import Any, Dict, FrozenSet, List, Optional @dataclass(frozen=True) class SkillSpec: name: str version: str description: str triggers: FrozenSet[str] required_tools: FrozenSet[str] required_inputs: FrozenSet[str] approval_for: FrozenSet[str] = frozenset() class Registry: def __init__(self): self._skills: Dict[tuple[str, str], SkillSpec] = {} def register(self, spec: SkillSpec) -> None: self._skills[(spec.name, spec.version)] = spec def discover(self, intent: str) -> List[SkillSpec]: intent_l = intent.lower() hits = [] for spec in self._skills.values(): if any(t.lower() in intent_l or intent_l in t.lower() for t in spec.triggers): hits.append(spec) elif any(word in intent_l for word in spec.name.replace("_", " ").split()): hits.append(spec) # de-dupe by (name, version) preserving order seen = set() out = [] for s in hits: key = (s.name, s.version) if key not in seen: seen.add(key) out.append(s) return out def resolve(self, name: str, version: str) -> SkillSpec: try: return self._skills[(name, version)] except KeyError as exc: raise KeyError(f"skill not found: {name}@{version}") from exc @dataclass class ToolGateway: granted_tools: FrozenSet[str] call_log: List[str] = field(default_factory=list) sources: Dict[str, str] = field(default_factory=dict) def call(self, tool: str, **kwargs: Any) -> Dict[str, Any]: if tool not in self.granted_tools: raise PermissionError(tool) self.call_log.append(tool) if tool == "read_source": source_id = kwargs["source_id"] text = self.sources.get(source_id, "") digest = hashlib.sha256(text.encode("utf-8")).hexdigest() return {"ok": True, "text": text, "sha256": digest} if tool == "summarize": text = kwargs.get("text", "") # 确定性摘要：非真实 LLM brief = "Brief: Agents use tools and feedback." if text else "" return {"ok": True, "report": brief} if tool == "publish_report": return {"ok": True, "published": kwargs.get("report", "")} raise ValueError(f"unknown tool: {tool}") class SkillRunner: def __init__(self, registry: Registry, sources: Optional[Dict[str, str]] = None): self.registry = registry self.sources = sources or { "src-1": "Agents use tools and feedback." } def run( self, name: str, version: str, inputs: Dict[str, Any], granted_tools: FrozenSet[str], approvals: FrozenSet[str], ) -> Dict[str, Any]: spec = self.registry.resolve(name, version) missing_inputs = sorted(set(spec.required_inputs) - set(inputs)) if missing_inputs: return { "status": "FAILED", "error": "MISSING_INPUT", "missing": missing_inputs, "events": [], "output": None, } missing_tools = sorted(set(spec.required_tools) - set(granted_tools)) if missing_tools: return { "status": "FAILED", "error": "MISSING_TOOL_PERMISSION", "missing": missing_tools, "events": [], "output": None, } pending = sorted(set(spec.approval_for) - set(approvals)) if pending: return { "status": "WAITING_APPROVAL", "pending": pending, "events": [], "output": None, } gw = ToolGateway(granted_tools=granted_tools, sources=self.sources) events: List[Dict[str, Any]] = [] # research_brief: read_source → summarize if name == "research_brief": read = gw.call("read_source", source_id=inputs["source_id"]) events.append({"step": "read_source", "sha256": read["sha256"]}) if not read["text"].strip(): return { "status": "FAILED", "error": "VALIDATION_FAILED", "reason": "empty_source", "events": events, "output": None, } summary = gw.call("summarize", text=read["text"]) events.append({"step": "summarize"}) report = summary["report"] if not report: return { "status": "FAILED", "error": "VALIDATION_FAILED", "reason": "empty_report", "events": events, "output": None, } return { "status": "COMPLETED", "events": events, "output": {"report": report}, } # publish_brief: requires approval_for publish_report if name == "publish_brief": report = inputs.get("report", "") pub = gw.call("publish_report", report=report) events.append({"step": "publish_report"}) return { "status": "COMPLETED", "events": events, "output": {"published": pub["published"]}, } return { "status": "FAILED", "error": "UNKNOWN_SKILL_HANDLER", "events": events, "output": None, } def make_registry() -> Registry: reg = Registry() reg.register( SkillSpec( name="research_brief", version="1.0.0", description="Read a source and produce a short research brief", triggers=frozenset({"research_brief", "generate brief", "source brief"}), required_tools=frozenset({"read_source", "summarize"}), required_inputs=frozenset({"source_id"}), ) ) reg.register( SkillSpec( name="publish_brief", version="1.0.0", description="Publish an existing brief after approval", triggers=frozenset({"publish_brief", "publish report"}), required_tools=frozenset({"publish_report"}), required_inputs=frozenset({"report"}), approval_for=frozenset({"publish_report"}), ) ) return reg if __name__ == "__main__": reg = make_registry() runner = SkillRunner(reg) # ① 按意图发现技能 found = reg.discover("please generate brief from source") assert any(s.name == "research_brief" for s in found) # ② 正常执行两步工具操作并生成产物 ok = runner.run( "research_brief", "1.0.0", {"source_id": "src-1"}, frozenset({"read_source", "summarize"}), frozenset(), ) assert ok["status"] == "COMPLETED" assert ok["events"][0]["step"] == "read_source" assert "sha256" in ok["events"][0] assert ok["events"][1]["step"] == "summarize" assert ok["output"]["report"] == "Brief: Agents use tools and feedback." # ③ 拒绝缺失输入 missing_in = runner.run( "research_brief", "1.0.0", {}, frozenset({"read_source", "summarize"}), frozenset(), ) assert missing_in["status"] == "FAILED" assert missing_in["error"] == "MISSING_INPUT" assert missing_in["events"] == [] # ④ 拒绝缺失工具权限且不调用工具 denied = runner.run( "research_brief", "1.0.0", {"source_id": "src-1"}, frozenset({"summarize"}), # missing read_source frozenset(), ) assert denied["status"] == "FAILED" assert denied["error"] == "MISSING_TOOL_PERMISSION" assert denied["events"] == [] # ⑤ 拒绝空来源并报告验证失败 empty_runner = SkillRunner(reg, sources={"src-empty": ""}) empty = empty_runner.run( "research_brief", "1.0.0", {"source_id": "src-empty"}, frozenset({"read_source", "summarize"}), frozenset(), ) assert empty["status"] == "FAILED" assert empty["error"] == "VALIDATION_FAILED" assert empty["events"] and empty["events"][0]["step"] == "read_source" # ⑥ 未审批时等待、获批后执行 waiting = runner.run( "publish_brief", "1.0.0", {"report": "Brief: Agents use tools and feedback."}, frozenset({"publish_report"}), frozenset(), ) assert waiting["status"] == "WAITING_APPROVAL" approved = runner.run( "publish_brief", "1.0.0", {"report": "Brief: Agents use tools and feedback."}, frozenset({"publish_report"}), frozenset({"publish_report"}), ) assert approved["status"] == "COMPLETED" print("A8 skill contract: 6 checks passed") print(json.dumps(ok, ensure_ascii=False, indent=2))