|
import json, subprocess, os, re, glob, concurrent.futures, time, sys |
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) |
|
from cases import CASES |
|
|
|
# Point this at a checkout of larksuite/cli's skills/ dir (or any lark-* skill dir). |
|
SKILLS_DIR = os.environ.get("SKILLS_DIR", os.path.join(os.path.dirname(os.path.abspath(__file__)), "skills")) |
|
|
|
def load_descriptions(): |
|
descs = {} |
|
for d in sorted(glob.glob(os.path.join(SKILLS_DIR, "lark-*"))): |
|
p = os.path.join(d, "SKILL.md") |
|
if not os.path.isfile(p): |
|
continue |
|
txt = open(p, encoding="utf-8").read() |
|
fm = txt.split("---", 2)[1] if txt.startswith("---") else "" |
|
m = re.search(r'^description:\s*(.+?)(?=\n[a-z_-]+:|\n---|\Z)', fm, re.S | re.M) |
|
desc = (m.group(1).strip() if m else "").strip().strip('"') |
|
descs[os.path.basename(d)] = desc |
|
return descs |
|
|
|
DESCS = load_descriptions() |
|
ROUTER_DESC = '飞书/Lark 资源统筹路由：当用户给出任何飞书资源 URL/token（/docx/ /wiki/ /sheets/ /base/ /slides/ 路径）或描述飞书操作意图时使用本 skill 做统筹分发，先识别资源类型再分发到对应具体 skill。' |
|
|
|
def build_prompt(strat, user_prompt): |
|
valid_names = sorted(DESCS.keys()) |
|
name_list = ", ".join(valid_names) |
|
name_constraint = ( |
|
f"VALID SKILL NAMES (you MUST pick one of these EXACT strings, nothing else — " |
|
f"do not invent or use any name outside this list): {name_list}\n" |
|
) |
|
if strat == "decentralized": |
|
catalog = "\n".join(f"- {n}: {d}" for n, d in sorted(DESCS.items())) |
|
instr = ( |
|
f"Below are the available skills (name: description). Read the user request and output EXACTLY one line: PICK: <skill-name>\n" |
|
f"{name_constraint}" |
|
f"Pick the single skill whose description best matches the request. Output only the PICK line, no explanation.\n\n{catalog}\n\nUSER REQUEST: {user_prompt}" |
|
) |
|
else: |
|
# Routed strategy: router skill ON TOP of the sub-skills' ORIGINAL full |
|
# descriptions (same surface as decentralized, plus a router entry). |
|
# This tests whether adding a router layer helps/hurts when sub-skill |
|
# descriptions are fully present. |
|
catalog = "\n".join(f"- {n}: {d}" for n, d in sorted(DESCS.items())) |
|
instr = ( |
|
f"Below are the available skills (name: description), headed by a routing skill. Read the user request and output EXACTLY one line: PICK: <sub-skill-name>\n" |
|
f"{name_constraint}" |
|
f"Route via lark-router conceptually first, then pick the terminal sub-skill it would dispatch to (do not pick lark-router itself). Output only the PICK line, no explanation.\n\n" |
|
f"ROUTER: lark-router: {ROUTER_DESC}\nSUB-SKILLS:\n{catalog}\n\nUSER REQUEST: {user_prompt}" |
|
) |
|
return instr |
|
|
|
def run(strat, case): |
|
cid, prompt, expected = case |
|
instr = build_prompt(strat, prompt) |
|
start = time.time() |
|
env = dict(os.environ, HOME="/tmp/lark-e2e/fakehome") |
|
# cwd = the strategy's skill dir so ./.claude/skills (the clone) is discovered |
|
cwd = f"/tmp/lark-e2e/{strat}" |
|
proc = subprocess.run( |
|
["claude", "-p", "--output-format", "json", instr], |
|
cwd=cwd, capture_output=True, text=True, timeout=180, stdin=subprocess.DEVNULL, env=env) |
|
pick = "" |
|
try: |
|
ev = json.loads(proc.stdout) |
|
txt = ev.get("result", "") |
|
raw_pick = "" |
|
for ln in txt.splitlines(): |
|
if "PICK:" in ln: |
|
raw_pick = ln.split("PICK:", 1)[1].strip().strip("`,").split()[0] |
|
break |
|
# If the model picked a name outside the catalog (e.g. its own |
|
# installed office:lark router), that counts as no clone selected. |
|
if raw_pick in DESCS: |
|
pick = raw_pick |
|
else: |
|
# fall back: any catalog name mentioned in the full text |
|
mentions = [n for n in DESCS if n in txt] |
|
pick = mentions[0] if mentions else "" |
|
except Exception: |
|
pass |
|
return dict(strat=strat, case=cid, expected=expected, pick=pick, |
|
correct=(pick == expected), dur=round(time.time() - start, 1)) |
|
|
|
cases = [(s, c) for s in ["decentralized", "routed"] for c in CASES] |
|
results = [] |
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex: |
|
futs = {ex.submit(run, s, c): (s, c) for s, c in cases} |
|
for fut in concurrent.futures.as_completed(futs): |
|
try: |
|
results.append(fut.result()) |
|
except Exception as e: |
|
s, c = futs[fut] |
|
results.append(dict(strat=s, case=c[0], error=str(e))) |
|
results.sort(key=lambda r: (r["strat"], r["case"])) |
|
print(json.dumps(results, ensure_ascii=False, indent=2)) |
