Add metrics CLI for project-scoped agent performance records.

Implement record, show, list, and export commands; document session-close
protocol template; extend cheat sheet and agency-framework docs; add CLI tests.
This commit is contained in:
2026-06-16 01:38:42 +02:00
parent 5cd3da3166
commit 97b7eb8cba
6 changed files with 303 additions and 5 deletions

View File

@@ -1,5 +1,6 @@
"""Command-line interface for Kaizen Agentic agent management."""
import json
import sys
import subprocess
import contextlib
@@ -938,6 +939,118 @@ def memory_clear(agent_name: str, target: str):
memory_path.parent.rmdir()
@cli.group()
def metrics():
"""Manage project-scoped agent metrics (.kaizen/metrics/<agent>/)."""
pass
@metrics.command("record")
@click.argument("agent_name")
@click.option("--target", "-t", default=".", help="Project root (default: current)")
@click.option("--success", "outcome_success", is_flag=True, help="Record successful execution")
@click.option("--failure", "outcome_failure", is_flag=True, help="Record failed execution")
@click.option("--time", "execution_time", type=float, help="Execution time in seconds")
@click.option("--quality", type=float, help="Quality score 0.01.0")
@click.option("--session-id", help="Optional session identifier")
@click.option("--idempotency-key", help="Skip append if this key was already recorded")
@click.option("--json", "json_input", is_flag=True, help="Read full record JSON from stdin")
def metrics_record(
agent_name: str,
target: str,
outcome_success: bool,
outcome_failure: bool,
execution_time: Optional[float],
quality: Optional[float],
session_id: Optional[str],
idempotency_key: Optional[str],
json_input: bool,
):
"""Append one execution record for an agent."""
store = MetricsStore(_project_root(target), agent_name)
if json_input:
payload = json.load(sys.stdin)
if not isinstance(payload, dict):
click.echo("Error: JSON input must be an object", err=True)
sys.exit(1)
else:
if outcome_success and outcome_failure:
click.echo("Error: use only one of --success or --failure", err=True)
sys.exit(1)
if not outcome_success and not outcome_failure:
click.echo("Error: specify --success or --failure (or use --json)", err=True)
sys.exit(1)
payload = {"success": outcome_success}
if execution_time is not None:
payload["execution_time_s"] = execution_time
if quality is not None:
payload["quality_score"] = quality
if session_id:
payload["session_id"] = session_id
if store.append(payload, idempotency_key=idempotency_key):
click.echo(f"Recorded metrics for '{agent_name}'")
else:
click.echo(f"Skipped duplicate record for '{agent_name}' (idempotency key exists)")
@metrics.command("show")
@click.argument("agent_name")
@click.option("--target", "-t", default=".", help="Project root (default: current)")
@click.option("--limit", "-n", default=5, show_default=True, help="Recent executions to show")
def metrics_show(agent_name: str, target: str, limit: int):
"""Print metrics summary and recent executions for an agent."""
store = MetricsStore(_project_root(target), agent_name)
if not store.executions_path.exists():
click.echo(f"No metrics found for agent '{agent_name}'.")
click.echo(f" Expected: {store.agent_dir}")
click.echo(f" Run: kaizen-agentic memory init {agent_name}")
return
summary = store.read_summary() or store.write_summary()
click.echo(f"Metrics for '{agent_name}':")
click.echo("=" * 40)
click.echo(json.dumps(summary, indent=2))
records = store.read_executions()
if records:
click.echo("\nRecent executions:")
for record in records[-limit:]:
click.echo(json.dumps(record, sort_keys=True))
@metrics.command("list")
@click.option("--target", "-t", default=".", help="Project root (default: current)")
def metrics_list(target: str):
"""List agents with metrics in the current project."""
agents = MetricsStore.list_agents(_project_root(target))
if not agents:
click.echo("No agent metrics found in this project.")
click.echo(" Run: kaizen-agentic memory init <agent>")
return
click.echo("Agents with metrics:")
for name in agents:
store = MetricsStore(_project_root(target), name)
summary = store.read_summary()
count = summary["execution_count"] if summary else len(store.read_executions())
click.echo(f"{name} ({count} executions)")
@metrics.command("export")
@click.argument("agent_name")
@click.option("--target", "-t", default=".", help="Project root (default: current)")
def metrics_export(agent_name: str, target: str):
"""Dump executions.jsonl for an agent to stdout."""
store = MetricsStore(_project_root(target), agent_name)
if not store.executions_path.exists():
click.echo(f"No metrics found for agent '{agent_name}'.", err=True)
sys.exit(1)
click.echo(store.executions_path.read_text(encoding="utf-8"), nl=False)
@cli.group()
def protocols():
"""Browse agent protocol runbooks (agents/protocols/<agent>/<slug>.md)."""
@@ -1011,8 +1124,12 @@ def protocols_show(agent_name: str, slug: str):
click.echo(protocol_path.read_text())
def _project_root(target: str) -> Path:
return Path(target).resolve()
def _memory_path(target: str, agent_name: str) -> Path:
return Path(target).resolve() / ".kaizen" / "agents" / agent_name / "memory.md"
return _project_root(target) / ".kaizen" / "agents" / agent_name / "memory.md"
def _today() -> str: