bf003cd944
Implements issue #14: 7 metrics that prove knowledge compounding. Metrics: - Knowledge velocity: new facts/day (from index.json) - Knowledge coverage: % domains with 10+ facts (from YAML files) - Hit rate: % sessions referencing bootstrap knowledge - Error recurrence: same errors across sessions (should decrease) - Task completion: % sessions with successful end_reason - First-try success: actions without backtracking (tool/msg ratio) - Knowledge age: staleness of facts (freshness score) Data sources: - knowledge/index.json + YAML files for fact metrics - ~/.hermes/state.db sessions + messages tables Features: - JSON and markdown output formats - --since, --repo, --format flags - 7-day trend tracking via snapshot persistence - Runs in 33ms on 11.9K sessions / 192K messages - Dashboard auto-generation with --save-snapshot Closes #14
404 lines
16 KiB
Python
404 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Compounding Intelligence Metrics Engine.
|
|
|
|
Computes 7 metrics that prove whether the knowledge compounding loop is working:
|
|
1. Knowledge velocity -- new facts per day
|
|
2. Knowledge coverage -- % of domains with >10 facts
|
|
3. Hit rate -- % of sessions referencing bootstrap knowledge
|
|
4. Error recurrence -- same errors across sessions (should decrease)
|
|
5. Task completion -- % of sessions ending successfully
|
|
6. First-try success -- actions without backtracking
|
|
7. Knowledge age -- staleness of facts
|
|
|
|
Usage:
|
|
python3 measurer.py # All metrics, all time
|
|
python3 measurer.py --since 2026-04-01 # Time range
|
|
python3 measurer.py --repo the-nexus # Per-repo metrics
|
|
python3 measurer.py --format json # JSON output (default)
|
|
python3 measurer.py --format markdown # Human-readable
|
|
python3 measurer.py --knowledge-dir ./knowledge # Custom knowledge path
|
|
python3 measurer.py --db ~/.hermes/state.db # Custom DB path
|
|
|
|
Data sources:
|
|
- knowledge/index.json -- fact index
|
|
- knowledge/ -- YAML fact files for coverage
|
|
- ~/.hermes/state.db -- session/message metadata
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
from collections import Counter, defaultdict
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
# --- Defaults ---
|
|
|
|
DEFAULT_KNOWLEDGE_DIR = Path(__file__).parent.parent / "knowledge"
|
|
DEFAULT_DB_PATH = Path.home() / ".hermes" / "state.db"
|
|
SEVEN_DAYS = timedelta(days=7)
|
|
|
|
|
|
# --- Knowledge Store ---
|
|
|
|
def load_facts(knowledge_dir):
|
|
index_path = knowledge_dir / "index.json"
|
|
if not index_path.exists():
|
|
return []
|
|
with open(index_path) as f:
|
|
data = json.load(f)
|
|
return data.get("facts", [])
|
|
|
|
|
|
def count_yaml_facts(knowledge_dir):
|
|
domain_counts = {}
|
|
for subdir in ["repos", "global", "agents"]:
|
|
dirpath = knowledge_dir / subdir
|
|
if not dirpath.exists():
|
|
continue
|
|
for yaml_file in dirpath.glob("*.yaml"):
|
|
count = 0
|
|
try:
|
|
content = yaml_file.read_text()
|
|
count = len(re.findall(r"^\s*-\s*id:", content, re.MULTILINE))
|
|
except Exception:
|
|
pass
|
|
domain = yaml_file.stem
|
|
domain_counts[domain] = domain_counts.get(domain, 0) + count
|
|
return domain_counts
|
|
|
|
|
|
# --- Session Database ---
|
|
|
|
def open_db(db_path):
|
|
if not db_path.exists():
|
|
print(f"WARNING: Database not found at {db_path}", file=sys.stderr)
|
|
return None
|
|
conn = sqlite3.connect(str(db_path))
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def query_sessions(conn, since=None, repo=None):
|
|
if conn is None:
|
|
return []
|
|
query = "SELECT id, started_at, ended_at, end_reason, message_count, tool_call_count, model FROM sessions WHERE 1=1"
|
|
params = []
|
|
if since:
|
|
since_ts = datetime.fromisoformat(since).replace(tzinfo=timezone.utc).timestamp()
|
|
query += " AND started_at >= ?"
|
|
params.append(since_ts)
|
|
query += " ORDER BY started_at ASC"
|
|
cur = conn.execute(query, params)
|
|
return [dict(row) for row in cur.fetchall()]
|
|
|
|
|
|
def query_messages(conn, session_ids=None, since_ts=None):
|
|
if conn is None:
|
|
return []
|
|
query = "SELECT m.session_id, m.role, m.content, m.tool_name, m.timestamp FROM messages m WHERE 1=1"
|
|
params = []
|
|
if since_ts:
|
|
query += " AND m.timestamp >= ?"
|
|
params.append(since_ts)
|
|
if session_ids:
|
|
placeholders = ",".join("?" for _ in session_ids)
|
|
query += f" AND m.session_id IN ({placeholders})"
|
|
params.extend(session_ids)
|
|
cur = conn.execute(query, params)
|
|
return [dict(row) for row in cur.fetchall()]
|
|
|
|
|
|
# --- Metric Computations ---
|
|
|
|
def compute_knowledge_velocity(facts, since=None):
|
|
if not facts:
|
|
return {"value": 0.0, "total_facts": 0, "period_days": 0, "new_facts": 0}
|
|
dates = []
|
|
for f in facts:
|
|
d = f.get("first_seen") or f.get("created")
|
|
if d:
|
|
try:
|
|
dt = datetime.fromisoformat(d.replace("Z", "+00:00"))
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
dates.append(dt)
|
|
except (ValueError, AttributeError):
|
|
pass
|
|
if not dates:
|
|
return {"value": 0.0, "total_facts": len(facts), "period_days": 0, "new_facts": 0}
|
|
if since:
|
|
cutoff = datetime.fromisoformat(since).replace(tzinfo=timezone.utc)
|
|
dates = [d for d in dates if d >= cutoff]
|
|
if not dates:
|
|
return {"value": 0.0, "total_facts": len(facts), "period_days": 0, "new_facts": 0}
|
|
earliest = min(dates)
|
|
latest = max(dates)
|
|
period_days = max((latest - earliest).days, 1)
|
|
return {"value": round(len(dates) / period_days, 2), "total_facts": len(facts), "period_days": period_days, "new_facts": len(dates)}
|
|
|
|
|
|
def compute_knowledge_coverage(facts, yaml_counts):
|
|
domain_fact_counts = defaultdict(int)
|
|
for f in facts:
|
|
domain = f.get("domain", "unknown")
|
|
domain_fact_counts[domain] += 1
|
|
for domain, count in yaml_counts.items():
|
|
domain_fact_counts[domain] = max(domain_fact_counts[domain], count)
|
|
total_domains = len(domain_fact_counts)
|
|
if total_domains == 0:
|
|
return {"value": 0.0, "covered_domains": 0, "total_domains": 0, "domain_details": {}}
|
|
covered = sum(1 for c in domain_fact_counts.values() if c >= 10)
|
|
return {"value": round(covered / total_domains, 3), "covered_domains": covered, "total_domains": total_domains, "domain_details": dict(sorted(domain_fact_counts.items(), key=lambda x: -x[1])[:20])}
|
|
|
|
|
|
def compute_hit_rate(sessions, messages, facts):
|
|
if not sessions or not facts:
|
|
return {"value": 0.0, "hit_sessions": 0, "total_sessions": len(sessions)}
|
|
fact_fragments = set()
|
|
for f in facts:
|
|
text = f.get("fact", "").lower().strip()
|
|
if len(text) > 10:
|
|
fact_fragments.add(text)
|
|
words = re.findall(r'\w{4,}', text)
|
|
for w in words:
|
|
fact_fragments.add(w)
|
|
if not fact_fragments:
|
|
return {"value": 0.0, "hit_sessions": 0, "total_sessions": len(sessions)}
|
|
session_messages = defaultdict(list)
|
|
for m in messages:
|
|
content = (m.get("content") or "").lower()
|
|
if content:
|
|
session_messages[m["session_id"]].append(content)
|
|
hit_sessions = 0
|
|
for session in sessions:
|
|
sid = session["id"]
|
|
all_content = " ".join(session_messages.get(sid, []))
|
|
if any(frag in all_content for frag in fact_fragments):
|
|
hit_sessions += 1
|
|
return {"value": round(hit_sessions / len(sessions), 3) if sessions else 0.0, "hit_sessions": hit_sessions, "total_sessions": len(sessions)}
|
|
|
|
|
|
def compute_error_recurrence(messages):
|
|
if not messages:
|
|
return {"value": 0.0, "unique_errors": 0, "recurring_errors": 0, "top_errors": []}
|
|
error_pattern = re.compile(r'(?:error|Error|ERROR|failed|FAIL|exception|Exception)[:\s]*(.{10,80})', re.IGNORECASE)
|
|
error_to_sessions = defaultdict(set)
|
|
for m in messages:
|
|
content = m.get("content") or ""
|
|
if not content:
|
|
continue
|
|
for match in error_pattern.finditer(content):
|
|
sig = match.group(1).strip().lower()
|
|
sig = re.sub(r'\s+', ' ', sig)
|
|
if len(sig) > 5:
|
|
error_to_sessions[sig].add(m["session_id"])
|
|
if not error_to_sessions:
|
|
return {"value": 0.0, "unique_errors": 0, "recurring_errors": 0, "top_errors": []}
|
|
recurring = {e: s for e, s in error_to_sessions.items() if len(s) > 1}
|
|
total_errors = len(error_to_sessions)
|
|
recurring_count = len(recurring)
|
|
top = sorted(recurring.items(), key=lambda x: -len(x[1]))[:10]
|
|
return {"value": round(recurring_count / total_errors, 3) if total_errors else 0.0, "unique_errors": total_errors, "recurring_errors": recurring_count, "top_errors": [{"error": e, "sessions": len(s)} for e, s in top]}
|
|
|
|
|
|
def compute_task_completion(sessions):
|
|
if not sessions:
|
|
return {"value": 0.0, "completed": 0, "total": 0, "breakdown": {}}
|
|
breakdown = Counter()
|
|
for s in sessions:
|
|
reason = s.get("end_reason") or "unknown"
|
|
breakdown[reason] += 1
|
|
completed = breakdown.get("cron_complete", 0) + breakdown.get("session_reset", 0)
|
|
normal_endings = completed + breakdown.get("cli_close", 0) + breakdown.get("compression", 0)
|
|
return {"value": round(completed / len(sessions), 3) if sessions else 0.0, "normal_end_rate": round(normal_endings / len(sessions), 3) if sessions else 0.0, "completed": completed, "total": len(sessions), "breakdown": dict(breakdown.most_common())}
|
|
|
|
|
|
def compute_first_try_success(sessions):
|
|
if not sessions:
|
|
return {"value": 0.0, "avg_tool_msg_ratio": 0.0, "sampled": 0}
|
|
ratios = []
|
|
for s in sessions:
|
|
msgs = s.get("message_count", 0) or 0
|
|
tools = s.get("tool_call_count", 0) or 0
|
|
if msgs > 2:
|
|
ratios.append(tools / msgs if msgs > 0 else 0)
|
|
if not ratios:
|
|
return {"value": 0.0, "avg_tool_msg_ratio": 0.0, "sampled": 0}
|
|
avg_ratio = sum(ratios) / len(ratios)
|
|
first_try = sum(1 for r in ratios if r < 0.5)
|
|
return {"value": round(first_try / len(ratios), 3), "avg_tool_msg_ratio": round(avg_ratio, 3), "sampled": len(ratios), "interpretation": "Higher value = fewer backtracks = better first-try success"}
|
|
|
|
|
|
def compute_knowledge_age(facts):
|
|
if not facts:
|
|
return {"value": 0.0, "avg_age_days": 0, "stale_facts": 0, "total_facts": 0}
|
|
now = datetime.now(timezone.utc)
|
|
ages = []
|
|
stale_count = 0
|
|
for f in facts:
|
|
confirmed = f.get("last_confirmed") or f.get("first_seen")
|
|
if confirmed:
|
|
try:
|
|
dt = datetime.fromisoformat(confirmed.replace("Z", "+00:00"))
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
age = (now - dt).days
|
|
ages.append(age)
|
|
if age > 30:
|
|
stale_count += 1
|
|
except (ValueError, AttributeError):
|
|
pass
|
|
if not ages:
|
|
return {"value": 0.0, "avg_age_days": 0, "stale_facts": 0, "total_facts": len(facts)}
|
|
avg_age = sum(ages) / len(ages)
|
|
freshness = max(0.0, 1.0 - (avg_age / 90))
|
|
return {"value": round(freshness, 3), "avg_age_days": round(avg_age, 1), "stale_facts": stale_count, "total_facts": len(facts), "interpretation": "1.0 = all facts fresh. 0.0 = all facts 90+ days old"}
|
|
|
|
|
|
# --- Trend Computation ---
|
|
|
|
def compute_trend(current, previous, metric_key="value"):
|
|
if not previous:
|
|
return {"delta": "N/A", "direction": "unknown"}
|
|
curr_val = current.get(metric_key, 0)
|
|
prev_val = previous.get(metric_key, 0)
|
|
if prev_val == 0:
|
|
return {"delta": "N/A (no baseline)", "direction": "unknown"}
|
|
pct = ((curr_val - prev_val) / abs(prev_val)) * 100
|
|
direction = "up" if pct > 0 else "down" if pct < 0 else "flat"
|
|
if metric_key == "error_recurrence" or metric_key == "knowledge_age":
|
|
direction_label = "good" if pct < 0 else "bad" if pct > 0 else "neutral"
|
|
else:
|
|
direction_label = "good" if pct > 0 else "bad" if pct < 0 else "neutral"
|
|
return {"delta": f"{'+' if pct > 0 else ''}{pct:.1f}%", "direction": direction, "assessment": direction_label}
|
|
|
|
|
|
# --- Output Formatters ---
|
|
|
|
def format_json(metrics):
|
|
return json.dumps(metrics, indent=2)
|
|
|
|
|
|
def format_markdown(metrics):
|
|
lines = ["# Compounding Intelligence Metrics", f"**Generated:** {metrics.get('generated_at', 'unknown')}", ""]
|
|
trend = metrics.get("trend_7d", {})
|
|
|
|
def metric_block(name, data, desc, good_direction="up"):
|
|
val = data.get("value", 0)
|
|
t = trend.get(name, {})
|
|
delta = t.get("delta", "N/A")
|
|
assessment = t.get("assessment", "unknown")
|
|
arrow = "up" if assessment == "good" else "down" if assessment == "bad" else "---"
|
|
lines.extend([f"## {name}", desc, "", f"**Value:** {val} | **7d trend:** {delta} {arrow} ({assessment})", ""])
|
|
for k, v in data.items():
|
|
if k != "value" and k != "interpretation":
|
|
if isinstance(v, (int, float, str)):
|
|
lines.append(f"- {k}: {v}")
|
|
lines.append("")
|
|
|
|
metric_block("knowledge_velocity", metrics.get("knowledge_velocity", {}), "New facts extracted per day. Higher = compounding loop working.")
|
|
metric_block("knowledge_coverage", metrics.get("knowledge_coverage", {}), "Percentage of domains/repos with 10+ facts. Measures breadth.")
|
|
metric_block("hit_rate", metrics.get("hit_rate", {}), "Percentage of sessions referencing bootstrapped knowledge.")
|
|
metric_block("error_recurrence", metrics.get("error_recurrence", {}), "Ratio of recurring errors. Lower = fleet learning from mistakes.", good_direction="down")
|
|
metric_block("task_completion", metrics.get("task_completion", {}), "Percentage of sessions ending with successful completion.")
|
|
metric_block("first_try_success", metrics.get("first_try_success", {}), "Percentage of sessions completed without backtracking.")
|
|
metric_block("knowledge_age", metrics.get("knowledge_age", {}), "Freshness of knowledge store. 1.0 = all fresh, 0.0 = all stale.", good_direction="up")
|
|
return "\n".join(lines)
|
|
|
|
|
|
# --- Snapshot Persistence ---
|
|
|
|
def load_snapshot(metrics_dir):
|
|
snapshot_path = metrics_dir / "latest_snapshot.json"
|
|
if snapshot_path.exists():
|
|
with open(snapshot_path) as f:
|
|
return json.load(f)
|
|
return {}
|
|
|
|
|
|
def save_snapshot(metrics_dir, metrics):
|
|
metrics_dir.mkdir(parents=True, exist_ok=True)
|
|
snapshot_path = metrics_dir / "latest_snapshot.json"
|
|
with open(snapshot_path, "w") as f:
|
|
json.dump(metrics, f, indent=2)
|
|
|
|
|
|
# --- Main ---
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Compounding Intelligence Metrics")
|
|
parser.add_argument("--since", help="Start date (YYYY-MM-DD)")
|
|
parser.add_argument("--repo", help="Filter by repo/domain")
|
|
parser.add_argument("--format", choices=["json", "markdown"], default="json")
|
|
parser.add_argument("--knowledge-dir", type=Path, default=DEFAULT_KNOWLEDGE_DIR)
|
|
parser.add_argument("--db", type=Path, default=DEFAULT_DB_PATH)
|
|
parser.add_argument("--save-snapshot", action="store_true", help="Save current metrics as snapshot for trend tracking")
|
|
parser.add_argument("--metrics-dir", type=Path, default=Path(__file__).parent.parent / "metrics", help="Directory for snapshots and dashboard")
|
|
args = parser.parse_args()
|
|
|
|
facts = load_facts(args.knowledge_dir)
|
|
yaml_counts = count_yaml_facts(args.knowledge_dir)
|
|
if args.repo:
|
|
facts = [f for f in facts if f.get("domain") == args.repo]
|
|
conn = open_db(args.db)
|
|
sessions = query_sessions(conn, since=args.since)
|
|
messages = query_messages(conn) if conn else []
|
|
if conn:
|
|
conn.close()
|
|
|
|
velocity = compute_knowledge_velocity(facts, since=args.since)
|
|
coverage = compute_knowledge_coverage(facts, yaml_counts)
|
|
hit_rate = compute_hit_rate(sessions, messages, facts)
|
|
error_recurrence = compute_error_recurrence(messages)
|
|
task_completion = compute_task_completion(sessions)
|
|
first_try = compute_first_try_success(sessions)
|
|
age = compute_knowledge_age(facts)
|
|
|
|
previous = load_snapshot(args.metrics_dir)
|
|
trend = {
|
|
"knowledge_velocity": compute_trend(velocity, previous.get("knowledge_velocity", {})),
|
|
"knowledge_coverage": compute_trend(coverage, previous.get("knowledge_coverage", {})),
|
|
"hit_rate": compute_trend(hit_rate, previous.get("hit_rate", {})),
|
|
"error_recurrence": compute_trend(error_recurrence, previous.get("error_recurrence", {}), "value"),
|
|
"task_completion": compute_trend(task_completion, previous.get("task_completion", {})),
|
|
"first_try_success": compute_trend(first_try, previous.get("first_try_success", {})),
|
|
"knowledge_age": compute_trend(age, previous.get("knowledge_age", {})),
|
|
}
|
|
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
metrics = {
|
|
"generated_at": now,
|
|
"knowledge_velocity": velocity,
|
|
"knowledge_coverage": coverage,
|
|
"hit_rate": hit_rate,
|
|
"error_recurrence": error_recurrence,
|
|
"task_completion": task_completion,
|
|
"first_try_success": first_try,
|
|
"knowledge_age": age,
|
|
"trend_7d": trend,
|
|
}
|
|
if args.since:
|
|
metrics["since"] = args.since
|
|
|
|
if args.save_snapshot:
|
|
save_snapshot(args.metrics_dir, metrics)
|
|
dashboard_path = args.metrics_dir / "dashboard.md"
|
|
with open(dashboard_path, "w") as f:
|
|
f.write(format_markdown(metrics))
|
|
|
|
if args.format == "json":
|
|
print(format_json(metrics))
|
|
else:
|
|
print(format_markdown(metrics))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|