Build Skill-Augmented AI Agents with SkillNet for Search, Evaluation, Graph Analysis, and Task Planning

import sys, subprocess
def _pip(*pkgs):
   subprocess.run([sys.executable, "-m", "pip", "install", "-q", *pkgs], test=False)
print("Installing dependencies (skillnet-ai, networkx, matplotlib, requests)...")
_pip("skillnet-ai", "networkx", "matplotlib", "requests")
import os, re, json, textwrap, pathlib, traceback
import requests
API_KEY  = os.environ.get("API_KEY", "")
BASE_URL = os.environ.get("BASE_URL", "https://api.openai.com/v1")
MODEL    = os.environ.get("SKILLNET_MODEL", "gpt-4o")
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "")
GITHUB_MIRROR = os.environ.get("GITHUB_MIRROR", "")
if not API_KEY:
   strive:
       from google.colab import userdata
       API_KEY = userdata.get("API_KEY") or ""
   besides Exception:
       cross
REST_BASE = "http://api-skillnet.openkg.cn/v1"
WORKDIR   = pathlib.Path("./skillnet_demo"); WORKDIR.mkdir(exist_ok=True)
SKILLS_DIR = WORKDIR / "my_skills"; SKILLS_DIR.mkdir(exist_ok=True)
def banner(title):
   line = "=" * 78
   print(f"n{line}n  {title}n{line}")

banner("1) Initialize SkillInternet consumer (SDK with REST fallback)")
USE_SDK = False
consumer = None
strive:
   from skillnet_ai import SkillNetClient
   consumer = SkillNetClient(
       api_key=API_KEY or None,
       base_url=BASE_URL,
       github_token=GITHUB_TOKEN or None,
   )
   USE_SDK = True
   print("SDK loaded: skillnet_ai.SkillNetClient")
besides Exception as e:
   print(f"SDK unavailable ({e!r}); utilizing REST fallback for search/obtain.")
def _norm(merchandise):
   if isinstance(merchandise, dict):
       g = merchandise.get
   else:
       g = lambda okay, d=None: getattr(merchandise, okay, d)
   return {
       "skill_name":        g("skill_name") or g("identify") or "?",
       "skill_description": g("skill_description") or g("description") or "",
       "writer":            g("writer") or "",
       "stars":             g("stars") or 0,
       "skill_url":         g("skill_url") or g("url") or "",
       "class":          g("class") or "",
   }
def search(q, mode="key phrase", restrict=5, min_stars=0, sort_by="stars", threshold=0.8):
   if USE_SDK:
       strive:
           kw = dict(q=q, restrict=restrict, mode=mode)
           if mode == "key phrase":
               kw.replace(min_stars=min_stars, sort_by=sort_by)
           else:
               kw.replace(threshold=threshold)
           res = consumer.search(**kw)
           return [_norm(x) for x in (res or [])]
       besides Exception as e:
           print(f"  [SDK search failed -> REST] {e!r}")
   params = {"q": q, "mode": mode, "restrict": restrict}
   if mode == "key phrase":
       params.replace(min_stars=min_stars, sort_by=sort_by)
   else:
       params.replace(threshold=threshold)
   strive:
       r = requests.get(f"{REST_BASE}/search", params=params, timeout=30)
       r.raise_for_status()
       return [_norm(x) for x in r.json().get("data", [])]
   besides Exception as e:
       print(f"  [REST search failed] {e!r}")
       return []
def show_results(outcomes, title=""):
   if title:
       print(f"n-- {title} --")
   if not outcomes:
       print("   (no outcomes / endpoint unreachable)")
       return
   for i, s in enumerate(outcomes, 1):
       desc = textwrap.shorten(s["skill_description"], 70, placeholder="...")
       print(f"  {i}. {s['skill_name']:<34} {s['stars']:<5} [{s['category']}]")
       if desc:
           print(f"     {desc}")
banner("2) Search: key phrase vs. semantic (vector)")
kw_hits  = search("pdf", mode="key phrase", restrict=5, sort_by="stars")
show_results(kw_hits, "key phrase: 'pdf' (sorted by stars)")
vec_hits = search("analyze monetary experiences from paperwork",
                 mode="vector", restrict=5, threshold=0.80)
show_results(vec_hits, "vector: 'analyze monetary experiences from paperwork'")

banner("3) Install expertise (obtain from GitHub into ./skillnet_demo/my_skills)")
CURATED = [
   "https://github.com/anthropics/skills/tree/main/skills/skill-creator",
   "https://github.com/anthropics/skills/tree/main/skills/algorithmic-art",
]
for s in (kw_hits + vec_hits):
   if s["skill_url"] and s["skill_url"] not in CURATED:
       CURATED.append(s["skill_url"])
CURATED = CURATED[:4]
def obtain(url, target_dir):
   if USE_SDK:
       strive:
           kw = {}
           if GITHUB_MIRROR:
               kw["mirror"] = GITHUB_MIRROR
           return consumer.obtain(url=url, target_dir=str(target_dir), **kw)
       besides TypeError:
           return consumer.obtain(url=url, target_dir=str(target_dir))
       besides Exception as e:
           print(f"  obtain failed for {url}: {e!r}")
           return None
   print("  (SDK not current — skipping stay obtain for this URL)")
   return None
put in = []
for url in CURATED:
   print(f"  downloading: {url}")
   path = obtain(url, SKILLS_DIR)
   if path:
       put in.append(path)
       print(f"    -> {path}")
print(f"nInstalled {len(put in)} ability(s).")

banner("4) Inspect put in expertise (SKILL.md frontmatter)")
def parse_skill_md(skill_path):
   p = pathlib.Path(skill_path)
   md = None
   if p.is_dir():
       for cand in p.rglob("SKILL.md"):
           md = cand; break
   elif p.identify.higher() == "SKILL.MD":
       md = p
   if not md or not md.exists():
       return {"path": str(skill_path), "identify": p.identify, "meta": {}, "discovered": False}
   textual content = md.read_text(encoding="utf-8", errors="ignore")
   meta = {}
   m = re.match(r"^---s*n(.*?)n---", textual content, re.DOTALL)
   if m:
       for line in m.group(1).splitlines():
           if ":" in line:
               okay, v = line.cut up(":", 1)
               meta[k.strip()] = v.strip().strip('"').strip("'")
   return {"path": str(md), "identify": meta.get("identify", p.identify),
           "meta": meta, "discovered": True}
inspected = [parse_skill_md(pp) for pp in installed] if put in else []
for data in inspected:
   print(f"  • {data['name']}  ({'SKILL.md discovered' if data['found'] else 'no SKILL.md'})")
   desc = data["meta"].get("description", "")
   if desc:
       print(f"      {textwrap.shorten(desc, 90, placeholder='...')}")
if not inspected:
   print("  (nothing put in regionally — possible no SDK/community; sections 2 & 7 nonetheless run)")

banner("5) Evaluate expertise on 5 high quality dimensions (high quality gate)")
DIMS = ["safety", "completeness", "executability", "maintainability", "cost_awareness"]
LEVEL_SCORE = {"Excellent": 4, "Good": 3, "Fair": 2, "Poor": 1, "Bad": 0}
def consider(goal):
   if USE_SDK and API_KEY:
       strive:
           return consumer.consider(goal=goal)
       besides Exception as e:
           print(f"  consider failed for {goal}: {e!r}")
   return None
def mock_eval(identify):
   import hashlib
   h = int(hashlib.md5(identify.encode()).hexdigest(), 16)
   ranges = ["Excellent", "Good", "Fair", "Poor"]
   return {d: {"degree": ranges[(h >> (i * 3)) % 4], "motive": "offline mock rating"}
           for i, d in enumerate(DIMS)}
def gate_score(report):
   tot = sum(LEVEL_SCORE.get(report.get(d, {}).get("degree", "Fair"), 2) for d in DIMS)
   return tot / (len(DIMS) * 4)
GATE_THRESHOLD = 0.55
targets = [s["skill_url"] for s in (kw_hits + vec_hits) if s["skill_url"]][:3] 
         or [i["name"] for i in inspected] or ["pdf-extractor", "chart-reader", "web-scraper"]
handed, scored = [], []
for t in targets:
   rep = consider(t)
   by way of = "LLM"
   if rep is None:
       rep, by way of = mock_eval(str(t)), "mock"
   rating = gate_score(rep)
   scored.append((t, rating, by way of))
   flags = " ".be a part of(f"{d[:4]}={rep.get(d,{}).get('degree','?')[:4]}" for d in DIMS)
   standing = "PASS " if rating >= GATE_THRESHOLD else "FAIL "
   print(f"  [{via:4}] {standing} rating={rating:.2f}  {textwrap.shorten(str(t),46,placeholder='...')}")
   print(f"          {flags}")
   if rating >= GATE_THRESHOLD:
       handed.append(t)
print(f"n{len(handed)}/{len(targets)} expertise handed the standard gate (threshold={GATE_THRESHOLD}).")
banner("6) Analyze relationships and draw the Skill Graph")
def analyze(skills_dir):
   if USE_SDK and API_KEY:
       strive:
           return consumer.analyze(skills_dir=str(skills_dir))
       besides Exception as e:
           print(f"  analyze failed: {e!r}")
   return None
rels = analyze(SKILLS_DIR)
if not rels:
   names = [i["name"] for i in inspected] or ["PDF_Parser", "Text_Summarizer",
                                              "Chart_Reader", "Web_Scraper"]
   whereas len(names) < 4:
       names.append(f"Skill_{len(names)}")
   rels = [
       {"source": names[0], "sort": "compose_with", "goal": names[1]},
       {"supply": names[2], "sort": "similar_to",   "goal": names[0]},
       {"supply": names[3], "sort": "depend_on",    "goal": names[1]},
       {"supply": names[1], "sort": "belong_to",    "goal": names[2]},
   ]
   print("  (utilizing offline mock relationships — set API_KEY for actual evaluation)")
for r in rels:
   print(f"  {r['source']} --[{r['type']}]--> {r['target']}")
strive:
   import networkx as nx
   import matplotlib.pyplot as plt
   G = nx.DiGraph()
   COLORS = {"similar_to": "#4C9BE8", "belong_to": "#E8A14C",
             "compose_with": "#6BBF59", "depend_on": "#D45D79"}
   for r in rels:
       G.add_edge(r["source"], r["target"], sort=r["type"])
   pos = nx.spring_layout(G, seed=42, okay=1.2)
   plt.determine(figsize=(9, 6))
   nx.draw_networkx_nodes(G, pos, node_size=2200, node_color="#EDEDED", edgecolors="#444")
   nx.draw_networkx_labels(G, pos, font_size=9)
   for et, col in COLORS.objects():
       edges = [(u, v) for u, v, d in G.edges(data=True) if d["type"] == et]
       if edges:
           nx.draw_networkx_edges(G, pos, edgelist=edges, edge_color=col,
                                  width=2, arrows=True, arrowsize=18,
                                  connectionstyle="arc3,rad=0.08")
   plt.legend(handles=[plt.Line2D([0], [0], shade=c, lw=2, label=t)
                       for t, c in COLORS.objects()], loc="finest", fontsize=8)
   plt.title("SkillInternet — Skill Relationship Graph")
   plt.axis("off"); plt.tight_layout()
   plt.savefig(WORKDIR / "skill_graph.png", dpi=130)
   plt.present()
   print(f"  graph saved -> {WORKDIR/'skill_graph.png'}")
besides Exception as e:
   print(f"  graph drawing skipped: {e!r}")

banner("7) Skill-augmented agent planner")
GOAL = "Analyze scRNA-seq information to search out and validate most cancers drug targets, then write a report"
def llm_decompose(objective):
   if API_KEY:
       strive:
           payload = {
               "mannequin": MODEL,
               "messages": [
                   {"role": "system", "content":
                    "Decompose the user's goal into 3-6 short, ordered subtasks. "
                    "Reply ONLY as a JSON array of strings, no prose, no markdown."},
                   {"role": "user", "content": goal},
               ],
               "temperature": 0.2,
           }
           r = requests.submit(f"{BASE_URL}/chat/completions",
                             headers={"Authorization": f"Bearer {API_KEY}"},
                             json=payload, timeout=60)
           r.raise_for_status()
           txt = r.json()["choices"][0]["message"]["content"]
           txt = re.sub(r"^```(?:json)?|```$", "", txt.strip()).strip()
           subs = json.hundreds(txt)
           if isinstance(subs, checklist) and subs:
               return [str(x) for x in subs]
       besides Exception as e:
           print(f"  LLM decompose failed -> heuristic ({e!r})")
   return ["acquire single-cell RNA-seq dataset",
           "preprocess and cluster cells",
           "identify candidate cancer target genes",
           "validate targets against pathway database",
           "generate a discovery report"]
def keywords_for(subtask):
   cease = {"the", "and", "a", "to", "of", "from", "into", "for", "with", "then", "an"}
   toks = [w for w in re.findall(r"[a-zA-Z-]+", subtask.decrease()) if w not in cease]
   return " ".be a part of(toks[:4])
subtasks = llm_decompose(GOAL)
print(f"GOAL: {GOAL}nnPLAN ({len(subtasks)} steps):")
plan = []
for i, st in enumerate(subtasks, 1):
   q = keywords_for(st)
   hits = search(q, mode="vector", restrict=2, threshold=0.6) or 
          search(q, mode="key phrase", restrict=2)
   finest = hits[0] if hits else None
   chosen = finest["skill_name"] if finest else "(no ability discovered — fallback to base mannequin)"
   plan.append({"step": i, "subtask": st, "question": q, "ability": chosen})
   print(f"n  Step {i}: {st}")
   print(f"     search('{q}') -> {chosen}" + (f"  {finest['stars']}" if finest else ""))
print("nExecution order (assembled pipeline):")
print("  " + "  ->  ".be a part of(p["skill"].cut up()[0] if p["skill"][0] != "(" else "base-model"
                           for p in plan))
banner("Tutorial full")
print(textwrap.dedent(f"""
 Recap:
   • Search (key phrase + vector) ............ ran by way of {'SDK' if USE_SDK else 'REST'}
   • Install (GitHub -> native) ............ {len(put in)} ability(s)
   • Inspect SKILL.md metadata ............ {len(inspected)} parsed
   • Evaluate + high quality gate .............. {len(handed)}/{len(targets)} handed {'(LLM)' if API_KEY else '(offline mock)'}
   • Relationship graph ................... {len(rels)} edges -> skill_graph.png
   • Agent planner ........................ {len(plan)} steps mapped to expertise
 Docs: https://github.com/zjunlp/SkillInternet
"""))