Agent Skills 入门：把研究流程封装成可复用、可测试的技能

REGISTRY = {
    "clean_panel": "balance a panel drop duplicates handle missing values",
    "event_study": "panel staggered treatment event time relative dummies",
    "make_table":  "render regression results into a publication table",
}

def select(task, registry):
    tok = set(task.lower().split())
    scored = {name: len(tok & set(desc.split())) for name, desc in registry.items()}
    return max(scored, key=scored.get), scored

task = "estimate an event study for a staggered treatment in panel data"
best, scores = select(task, REGISTRY)
print("scores:", scores)
print("selected skill:", best)

预期输出

scores: {'clean_panel': 2, 'event_study': 4, 'make_table': 1}
selected skill: event_study

def clean_panel(state):
    state["clean"] = True; return state
def event_study(state):
    assert state.get("clean"), "needs a clean panel first"
    state["coefs"] = [0.01, 0.04, 0.12]; return state
def make_table(state):
    assert "coefs" in state, "needs estimates first"
    state["table"] = "outputs/event_study.tex"; return state

pipeline = [clean_panel, event_study, make_table]
state = {"panel": "firm-year.csv"}
for skill in pipeline:
    state = skill(state)
print("artifacts:", {k: state[k] for k in ("clean", "coefs", "table")})

预期输出

artifacts: {'clean': True, 'coefs': [0.01, 0.04, 0.12], 'table': 'outputs/event_study.tex'}

import math
pass_rates = {"clean_panel": 0.99, "event_study": 0.95, "make_table": 0.98}
R = math.prod(pass_rates.values())
print("per-skill pass rates:", pass_rates)
print(f"pipeline reliability = {R:.3f}")
print(f"failure rate = {1 - R:.3f}  -> test each skill, do not trust the chain")

预期输出

per-skill pass rates: {'clean_panel': 0.99, 'event_study': 0.95, 'make_table': 0.98}
pipeline reliability = 0.922
failure rate = 0.078  -> test each skill, do not trust the chain

skills = {
    "clean_panel": {"index": 40, "body": 1200},
    "event_study": {"index": 45, "body": 1500},
    "make_table":  {"index": 38, "body": 900},
}
triggered = ["event_study"]                       # only this matches the task
index_only = sum(s["index"] for s in skills.values())
progressive = index_only + sum(skills[n]["body"] for n in triggered)
load_all = index_only + sum(s["body"] for s in skills.values())
print("index-only tokens:        ", index_only)
print("progressive-disclosure:   ", progressive)
print("load-everything tokens:   ", load_all)

预期输出

index-only tokens:         123
progressive-disclosure:    1623
load-everything tokens:    3723

def event_study_skill(n_periods):
    if n_periods < 2:
        raise ValueError("need >= 2 periods")
    return {"ok": True, "periods": n_periods}

test_cases = [5, 3, 1, 8, 0]
results = []
for tc in test_cases:
    try:
        event_study_skill(tc); results.append(True)
    except Exception:
        results.append(False)
print("results:", results)
print(f"pass rate = {sum(results)/len(results):.2f}  (skills need tests, like code)")

预期输出

results: [True, True, False, True, False]
pass rate = 0.60  (skills need tests, like code)