334 lines
14 KiB
Python
334 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""kpi_ledger.py — agent-operating KPI 수집기 (finding #19).
|
|
|
|
리뷰 지적: agent-operating-kpi.yaml 에 rework/hallucination/context-bloat/release-failure 등이
|
|
정의돼 있으나 **수집기가 없어 측정되지 않는다**. 이 도구가 그 구멍을 닫는다 —
|
|
① 기존 append-only 아티팩트(completion-records 시도수·acceptance-events 결정·token-ledger)에서
|
|
**파생 가능한 KPI를 실제로 계산**하고, ② 파생 불가한 것은 수동 이벤트로 적재하며,
|
|
③ 대시보드에서 각 KPI를 measured/derived · manual · **unmeasured(정직 표시)** 로 구분한다.
|
|
→ "측정 안 됨"을 "측정됨"처럼 위장하지 않는다.
|
|
|
|
Usage:
|
|
kpi_ledger.py derive [--workflow WF] # 아티팩트에서 파생 KPI 계산 → kpi-ledger.jsonl 적재
|
|
kpi_ledger.py log --metric M --value V [--workflow WF --role R --note "..."] # 수동 이벤트
|
|
kpi_ledger.py dashboard # reports/KPI.md 렌더(measured/manual/unmeasured)
|
|
"""
|
|
import glob
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
import yaml
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
ROOT = os.environ.get("CLAUDE_PROJECT_DIR") or os.path.dirname(os.path.dirname(HERE))
|
|
sys.path.insert(0, HERE)
|
|
import _workspace as W # noqa: E402
|
|
import acceptance_log as AL # noqa: E402
|
|
import state_engine as SE # noqa: E402
|
|
|
|
KPI_SPEC = os.path.join(ROOT, "org-os", "06-agent-work", "agent-operating-kpi.yaml")
|
|
|
|
|
|
def _ledger():
|
|
return os.path.join(W.state_dir(), "kpi-ledger.jsonl")
|
|
|
|
|
|
def _now():
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
def _append(rec):
|
|
lp = _ledger()
|
|
os.makedirs(os.path.dirname(lp), exist_ok=True)
|
|
with open(lp, "a", encoding="utf-8") as f:
|
|
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
|
|
|
|
|
|
def _rows():
|
|
lp = _ledger()
|
|
if not os.path.exists(lp):
|
|
return []
|
|
out = []
|
|
for line in open(lp, encoding="utf-8"):
|
|
line = line.strip()
|
|
if line:
|
|
try:
|
|
out.append(json.loads(line))
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------- derive
|
|
def _report_files():
|
|
return glob.glob(os.path.join(W.records_dir(), "**", "*.report.yaml"), recursive=True)
|
|
|
|
|
|
def _role_of(path):
|
|
"""report yaml에서 role-id(없으면 파일명 stem 앞부분) 추출."""
|
|
try:
|
|
doc = yaml.safe_load(open(path, encoding="utf-8")) or {}
|
|
rid = doc.get("role-id") or doc.get("role-name")
|
|
wf = doc.get("workflow-id")
|
|
except Exception:
|
|
rid, wf = None, None
|
|
stem = os.path.basename(path).replace(".report.yaml", "")
|
|
role = rid or stem.rsplit("-", 1)[0]
|
|
wf = wf or os.path.basename(os.path.dirname(path))
|
|
return str(wf), str(role)
|
|
|
|
|
|
def derive(workflow=None):
|
|
"""기존 아티팩트에서 파생 KPI를 계산해 적재. (measured=derived source)"""
|
|
artifact_events = [event for event in SE.read_artifact_events()
|
|
if event.get("event-type") == "artifact-submitted"
|
|
and (not workflow or event.get("workflow-id") == workflow)]
|
|
wf_set = {event.get("workflow-id") for event in artifact_events if event.get("workflow-id")}
|
|
total_reports = len(artifact_events)
|
|
event_ids = [event.get("artifact-event-id") for event in artifact_events]
|
|
artifact_keys = [(event.get("workflow-id"), event.get("artifact-id"), event.get("artifact-sha256"))
|
|
for event in artifact_events]
|
|
extra_attempts = ((len(event_ids) - len(set(event_ids)))
|
|
+ (len(artifact_keys) - len(set(artifact_keys))))
|
|
|
|
# acceptance 이벤트에서 결정 분포.
|
|
events = [e for e in AL.read_events()
|
|
if not workflow or e.get("workflow-id") == workflow]
|
|
dec = {"accepted": 0, "changes-requested": 0, "blocked": 0}
|
|
human_wf = set()
|
|
for e in events:
|
|
d = (e.get("decision") or "").strip().lower()
|
|
if d in dec:
|
|
dec[d] += 1
|
|
reviewer = e.get("reviewer") if isinstance(e.get("reviewer"), dict) else {}
|
|
appr = str(reviewer.get("actor-id") or reviewer.get("role-id") or e.get("role-id") or "")
|
|
if "HUMAN" in appr.upper():
|
|
human_wf.add(e.get("workflow-id"))
|
|
total_dec = sum(dec.values())
|
|
release_events = []
|
|
state_dir = W.state_dir()
|
|
for path in glob.glob(os.path.join(state_dir, "*", "workflow-events.jsonl")):
|
|
for line in open(path, encoding="utf-8"):
|
|
try:
|
|
event = json.loads(line)
|
|
except Exception:
|
|
continue
|
|
if event.get("event-type") != "release-decision-recorded":
|
|
continue
|
|
if workflow and event.get("workflow-id") != workflow:
|
|
continue
|
|
release_events.append(event)
|
|
failed_release = sum(1 for event in release_events
|
|
if event.get("status") != "Approved" or event.get("unresolved-critical-risks"))
|
|
|
|
# Context bloat is token-weighted, not item-count weighted. Planned context comes from
|
|
# exact packages bound in the subagent registry; actual reads come from usage-events.
|
|
planned = {}
|
|
registry_path = os.path.join(state_dir, "subagent-registry.jsonl")
|
|
if os.path.exists(registry_path):
|
|
for line in open(registry_path, encoding="utf-8"):
|
|
try:
|
|
row = json.loads(line)
|
|
except Exception:
|
|
continue
|
|
if workflow and row.get("workflow_id") != workflow:
|
|
continue
|
|
package_ref = row.get("context_package")
|
|
if not package_ref:
|
|
continue
|
|
package_path = package_ref if os.path.isabs(package_ref) else os.path.join(ROOT, package_ref)
|
|
try:
|
|
package = yaml.safe_load(open(package_path, encoding="utf-8")) or {}
|
|
except Exception:
|
|
continue
|
|
for item in package.get("must-read", []) or []:
|
|
if not isinstance(item, dict) or not item.get("context-id"):
|
|
continue
|
|
estimate = item.get("estimated-tokens")
|
|
if estimate is None:
|
|
uri = item.get("uri")
|
|
path = uri if os.path.isabs(str(uri or "")) else os.path.join(ROOT, str(uri or ""))
|
|
try:
|
|
estimate = max(1, os.path.getsize(path) // 4)
|
|
except OSError:
|
|
estimate = 0
|
|
planned[(package_ref, str(item["context-id"]))] = max(0, int(estimate or 0))
|
|
reads = {}
|
|
usage_path = os.path.join(state_dir, "usage-events.jsonl")
|
|
if os.path.exists(usage_path):
|
|
for line in open(usage_path, encoding="utf-8"):
|
|
try:
|
|
event = json.loads(line)
|
|
except Exception:
|
|
continue
|
|
if event.get("event-type") != "ContextItemRead":
|
|
continue
|
|
if workflow and event.get("workflow-id") != workflow:
|
|
continue
|
|
key = (event.get("context-package"), str(event.get("context-id")))
|
|
reads[key] = reads.get(key, 0) + 1
|
|
planned_tokens = sum(planned.values())
|
|
unused_tokens = sum(value for key, value in planned.items() if not reads.get(key))
|
|
duplicate_tokens = sum(planned.get(key, 0) * max(0, count - 1) for key, count in reads.items())
|
|
context_bloat = (min(planned_tokens, unused_tokens + duplicate_tokens) / planned_tokens
|
|
if planned_tokens else None)
|
|
|
|
metrics = {
|
|
# rework-rate = changes-requested / canonical submitted outputs
|
|
"rework-rate": (dec["changes-requested"] / total_reports) if total_reports else None,
|
|
# duplicate-report-rate = duplicate canonical event/revision keys / submissions
|
|
"duplicate-report-rate": (extra_attempts / total_reports) if total_reports else None,
|
|
# human-intervention-rate = 인간 개입 워크플로 / 총 워크플로
|
|
"human-intervention-rate": (len(human_wf) / len(wf_set)) if wf_set else None,
|
|
"release-gate-failure-rate": (failed_release / len(release_events)) if release_events else None,
|
|
"context-bloat-rate": context_bloat,
|
|
}
|
|
stamp = _now()
|
|
logged = 0
|
|
for m, v in metrics.items():
|
|
if v is None:
|
|
continue
|
|
_append({"at": stamp, "metric": m, "value": round(v, 4), "source": "derived",
|
|
"workflow": workflow or "*",
|
|
"basis": {"reports": total_reports, "decisions": total_dec,
|
|
"extra_attempts": extra_attempts, "workflows": len(wf_set),
|
|
"release_checks": len(release_events),
|
|
"planned_context_tokens": planned_tokens,
|
|
"unused_context_tokens": unused_tokens,
|
|
"duplicate_context_tokens": duplicate_tokens}})
|
|
logged += 1
|
|
# 파생 불가하지만 유용한 원자료도 함께 기록(counts).
|
|
_append({"at": stamp, "metric": "_counts", "value": total_reports, "source": "derived",
|
|
"workflow": workflow or "*",
|
|
"basis": {"reports": total_reports, "decisions": dec, "workflows": len(wf_set),
|
|
"extra_attempts": extra_attempts}})
|
|
print(f"[kpi_ledger] derived {logged} metric(s) from {total_reports} reports · "
|
|
f"{total_dec} acceptance decisions · {len(wf_set)} workflow(s)")
|
|
|
|
|
|
# ---------------------------------------------------------------- dashboard
|
|
# 파생 가능(코드가 아티팩트에서 계산) vs 수동만(사람/외부 계측 필요) vs 미측정(수집기 없음).
|
|
DERIVED_METRICS = {
|
|
"rework-rate", "duplicate-report-rate", "human-intervention-rate",
|
|
"release-gate-failure-rate", "context-bloat-rate",
|
|
}
|
|
MANUAL_METRICS = { # log 이벤트로만 채워질 수 있는 것(사람 판정/외부 계측)
|
|
"hallucination-rate", "blocker-reopen-rate", "slo-risk-escape-rate",
|
|
"shift-left-detection-rate", "skipped-role-incident-rate", "learning-capture-rate",
|
|
}
|
|
|
|
|
|
def _spec_metrics():
|
|
try:
|
|
return (yaml.safe_load(open(KPI_SPEC, encoding="utf-8")) or {})["agent-operating-kpi"]["metrics"]
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def dashboard():
|
|
rows = _rows()
|
|
latest = {}
|
|
manual_counts = {}
|
|
for r in rows:
|
|
m = r.get("metric")
|
|
if m == "_counts":
|
|
continue
|
|
if r.get("source") == "manual":
|
|
manual_counts[m] = manual_counts.get(m, 0) + 1
|
|
latest[m] = r # 마지막 기록이 최신
|
|
spec = _spec_metrics()
|
|
ts = datetime.now().strftime("%Y-%m-%d %H:%M")
|
|
L = ["# 📈 Agent KPI 대시보드 (finding #19)", "",
|
|
f"생성: {ts} · 원장: `state/kpi-ledger.jsonl`",
|
|
"> 각 KPI를 **측정(derived)** · **수동(manual)** · **미측정(no collector)** 로 정직히 구분한다. "
|
|
"미측정을 측정된 것처럼 위장하지 않는다.",
|
|
"", "| KPI | 목표 | 상태 | 최근 값 |", "|---|---|---|---|"]
|
|
measured = 0
|
|
for name, spec_v in spec.items():
|
|
target = spec_v.get("target", "-") if isinstance(spec_v, dict) else "-"
|
|
if name in latest:
|
|
measured += 1
|
|
val = latest[name].get("value")
|
|
src = latest[name].get("source")
|
|
status = "✅ derived" if src == "derived" else "✍️ manual"
|
|
vals = f"{val:.1%}" if isinstance(val, float) and val <= 1 else str(val)
|
|
elif name in DERIVED_METRICS:
|
|
status, vals = "⏳ derivable (run `derive`)", "-"
|
|
elif name in MANUAL_METRICS:
|
|
status, vals = "✍️ manual-only (log 이벤트 필요)", "-"
|
|
else:
|
|
status, vals = "⚪ 미측정 (수집기 없음 — 정직)", "-"
|
|
L.append(f"| {name} | {target} | {status} | {vals} |")
|
|
total = len(spec)
|
|
L += ["", f"측정중(derived/manual): **{measured}** · 파생가능 미실행: "
|
|
f"{len(DERIVED_METRICS - set(latest))} · 미측정: "
|
|
f"{total - measured - len(DERIVED_METRICS - set(latest))} / 총 {total} KPI"]
|
|
out = os.path.join(W.reports_dir(), "KPI.md")
|
|
os.makedirs(os.path.dirname(out), exist_ok=True)
|
|
with open(out, "w", encoding="utf-8") as f:
|
|
f.write("\n".join(L) + "\n")
|
|
print(f"[kpi_ledger] dashboard -> {os.path.relpath(out, ROOT)} "
|
|
f"({measured}/{total} KPI 측정중)")
|
|
|
|
|
|
# ---------------------------------------------------------------- CLI
|
|
def _parse(args):
|
|
opt = {}
|
|
i = 0
|
|
while i < len(args):
|
|
if args[i].startswith("--"):
|
|
k = args[i][2:]
|
|
v = args[i + 1] if i + 1 < len(args) and not args[i + 1].startswith("--") else True
|
|
opt[k] = v
|
|
i += 2
|
|
else:
|
|
i += 1
|
|
return opt
|
|
|
|
|
|
def main():
|
|
a = sys.argv[1:]
|
|
if not a:
|
|
sys.stderr.write(__doc__)
|
|
sys.exit(1)
|
|
cmd = a[0]
|
|
opt = _parse(a[1:])
|
|
try:
|
|
if cmd == "derive":
|
|
derive(opt.get("workflow") if isinstance(opt.get("workflow"), str) else None)
|
|
elif cmd == "log":
|
|
metric = opt.get("metric")
|
|
if metric not in _spec_metrics():
|
|
raise ValueError(f"미등록 metric: {metric!r}")
|
|
if not opt.get("window-start") or not opt.get("window-end"):
|
|
raise ValueError("manual KPI는 --window-start/--window-end 필수")
|
|
value = float(opt.get("value"))
|
|
if not (value == value and abs(value) != float("inf")):
|
|
raise ValueError("KPI value는 finite number여야 한다")
|
|
if metric.endswith("-rate") and not 0 <= value <= 1:
|
|
raise ValueError("rate KPI는 0..1 범위여야 한다")
|
|
_append({"at": _now(), "metric": metric,
|
|
"value": value, "unit": opt.get("unit") or ("ratio" if metric.endswith("-rate") else "count"),
|
|
"window-start": opt.get("window-start"), "window-end": opt.get("window-end"),
|
|
"formula-version": 2, "source": "manual",
|
|
"workflow": opt.get("workflow", "*"), "role": opt.get("role"),
|
|
"note": opt.get("note")})
|
|
print(f"[kpi_ledger] logged manual {metric}={value}")
|
|
elif cmd == "dashboard":
|
|
dashboard()
|
|
else:
|
|
sys.stderr.write(f"unknown command: {cmd}\n")
|
|
sys.exit(1)
|
|
except W.WorkspaceNotSetError as e:
|
|
sys.stderr.write(f"[kpi_ledger] 워크스페이스 미설정: {e}\n")
|
|
sys.exit(1)
|
|
except (TypeError, ValueError) as e:
|
|
sys.stderr.write(f"[kpi_ledger] 입력 거부: {e}\n")
|
|
sys.exit(2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|