#!/usr/bin/env python3 """ Aria Knowledge Pipeline (AKP) โ€” Stage 2: Triage KAR-74 ยท 2026-05-12 Liest alle ingested Videos ohne triaged-Eintrag, schickt Transcript-Excerpt an Haiku 4.5 mit JSON-Rubric (depth/novelty/aria_relevance/production_ready 0-3), schreibt Verdict (promote/summary_only/skip) nach triage/.json, trackt Cost gegen Daily-Cap. Gated auf ANTHROPIC_API_KEY โ€” fehlt der, wird sauber gequeued (kein Crash). """ from __future__ import annotations import sys as _sys _sys.path.insert(0, "/root/aria/lib") from aria_logging import get_logger as _get_logger _log = _get_logger("aria-akp-triage") from aria_audit import audit as _audit # KAR-220 P2 from aria_llm_trace import trace_llm_call # KAR-744 LLM-Tracing import json, os, sqlite3, sys, time, yaml from datetime import datetime, timezone from pathlib import Path CONFIG_PATH = Path("/root/aria/brain/youtube/.config.yaml") TRIAGE_PROMPT = """Du bist Arias Triage-Filter fuer Wissens-Videos. Bewerte das Video auf 4 Qualitaets-Achsen PLUS 1 Junk-Penalty-Achse, Skala 0-3. Qualitaets-Achsen (0=schwach, 1=mittelmaessig, 2=gut, 3=stark): 1. depth: Tiefe der Insights vs. Oberflaeche (Skript-Tiefe, Argumentations-Schritte) 2. novelty: Neuigkeit vs. wiedergekaute Best-Practices (Datum/Stand-Bezug, neue Konzepte) 3. aria_relevance: Relevanz fuer Aria-Ziele (Multi-Agent-Patterns, Eval-Setup, Memory-Design, LLM-Distribution, Engineering-Principles, Brain-Architektur, Observability, Cost-Optimierung) 4. production_ready: Direkt umsetzbar vs. theoretisch (Code/Workflow/Architektur in der Praxis nutzbar) Junk-Penalty-Achse (LLM-Brain-Rot-Guard, Xing et al. 2025 / arXiv:2510.13928 โ€” Junk-Content verursacht nachweislich dauerhaften Capability-Decay): 5. junk_penalty: Wie stark zeigt der Content schaedliche Junk-Marker? 0 = sauber, faktenbasiert, ausargumentiert 1 = leichte Hype-Tendenz, aber Substanz ueberwiegt 2 = deutlicher Clickbait/Engagement-Bait ODER Thought-Skipping (Ergebnis-Behauptung ohne Reasoning-Schritte) 3 = reiner Sensationalismus/Hype ohne Substanz, uebertriebene/unbelegte Claims, Meinung als Fakt Junk-Marker: "WOW"/"SHOCKING"/"you won't believe", Hype ohne Beleg, fehlende Reasoning-Kette. WICHTIG: Viralitaet/Kuerze ist KEIN Wert-Signal. Substanz schlaegt Reichweite. Verdict wird deterministisch aus den Scores berechnet (du musst es nicht selbst bestimmen), aber setze "verdict": "skip" wenn der Content komplett wertlos ist. Die Junk-Penalty zieht den effektiven Score runter; junk_penalty=3 erzwingt skip, junk_penalty>=2 verhindert promote. Output AUSSCHLIESSLICH ein JSON-Objekt mit dieser Struktur (keine Erklaerung drumherum): {{"depth": int, "novelty": int, "aria_relevance": int, "production_ready": int, "junk_penalty": int, "verdict": "promote"|"summary_only"|"skip", "one_line_reason": str}} Video: - Titel: {title} - Channel: {channel} - Dauer: {duration_min} min - Upload: {upload_date} Transcript-Excerpt: {transcript_excerpt} """ def load_config() -> dict: with CONFIG_PATH.open() as f: return yaml.safe_load(f) def db_conn(db_path: str) -> sqlite3.Connection: con = sqlite3.connect(db_path) con.row_factory = sqlite3.Row return con def ensure_schema(con: sqlite3.Connection) -> None: """Idempotente Migration: junk_penalty-Spalte (KAR-685, Brain-Rot-Guard).""" cols = {r[1] for r in con.execute("PRAGMA table_info(triaged)")} if "junk_penalty" not in cols: con.execute("ALTER TABLE triaged ADD COLUMN junk_penalty INTEGER DEFAULT 0") con.commit() def compute_verdict(scores: dict, junk_penalty: int, promote_avg: float, promote_aria: int, summary_avg: float) -> tuple[str, float, float]: """Pure verdict-Logik (testbar, ohne API). Junk-Penalty zieht effektiven Score runter. Returns: (verdict, avg_positive, effective_avg) Regeln (KAR-685): - junk_penalty=3 -> immer skip (reiner Junk, Brain-Rot-Risiko) - junk_penalty>=2 -> nie promote (max summary_only) - effective = max(0, avg_positive - 0.5*junk_penalty) """ jp = max(0, min(3, int(junk_penalty))) avg_positive = (scores["depth"] + scores["novelty"] + scores["aria_relevance"] + scores["production_ready"]) / 4.0 eff = max(0.0, avg_positive - 0.5 * jp) if jp >= 3: verdict = "skip" elif eff >= promote_avg and scores["aria_relevance"] >= promote_aria and jp < 2: verdict = "promote" elif eff >= summary_avg or (scores["aria_relevance"] >= promote_aria and jp < 2): verdict = "summary_only" else: verdict = "skip" return verdict, avg_positive, eff def get_pending(con: sqlite3.Connection, limit: int = 250) -> list[sqlite3.Row]: return list(con.execute( """SELECT i.video_id, i.channel, i.title, i.duration_seconds, i.url, i.published_at AS upload_date, i.raw_path FROM ingested i LEFT JOIN triaged t ON t.video_id = i.video_id WHERE i.skipped_reason IS NULL AND i.raw_path IS NOT NULL AND t.video_id IS NULL ORDER BY i.ingested_at ASC LIMIT ?""", (limit,), )) def todays_cost(con: sqlite3.Connection, key: str) -> float: today = datetime.now(timezone.utc).date().isoformat() row = con.execute(f"SELECT {key} FROM daily_costs WHERE date=?", (today,)).fetchone() return float(row[0]) if row else 0.0 def add_cost(con: sqlite3.Connection, triage_delta: float = 0.0, deep_delta: float = 0.0) -> None: today = datetime.now(timezone.utc).date().isoformat() con.execute( "INSERT INTO daily_costs (date, triage_usd, deep_usd, total_usd) VALUES (?,?,?,?) " "ON CONFLICT(date) DO UPDATE SET triage_usd=triage_usd+?, deep_usd=deep_usd+?, total_usd=total_usd+?", (today, triage_delta, deep_delta, triage_delta + deep_delta, triage_delta, deep_delta, triage_delta + deep_delta), ) con.commit() def haiku_pricing_usd(input_tokens: int, output_tokens: int) -> float: # Haiku 4.5 pricing (per Anthropic Pricing Page, Stand 2026): $1/MTok input, $5/MTok output return (input_tokens / 1_000_000) * 1.0 + (output_tokens / 1_000_000) * 5.0 def call_haiku(client, model: str, prompt: str, video_id: str | None = None) -> tuple[dict | None, dict]: """Call Haiku, expect JSON. Returns (parsed_json or None, usage_dict).""" t0 = time.monotonic() try: resp = client.messages.create( model=model, max_tokens=400, messages=[{"role": "user", "content": prompt}], ) except Exception as e: trace_llm_call(script="aria-akp-triage", model=model, prompt=prompt, latency_ms=int((time.monotonic() - t0) * 1000), video_id=video_id, error=str(e)) return None, {"error": str(e), "input_tokens": 0, "output_tokens": 0} raw_text = "".join(b.text for b in resp.content if hasattr(b, "text")) text = raw_text parsed = None try: if "```" in text: text = text.split("```")[1] if text.startswith("json"): text = text[4:] text = text.strip() first = text.find("{"); last = text.rfind("}") if first != -1 and last != -1: parsed = json.loads(text[first:last+1]) except (json.JSONDecodeError, ValueError): parsed = None usage = { "input_tokens": resp.usage.input_tokens, "output_tokens": resp.usage.output_tokens, } trace_llm_call(script="aria-akp-triage", model=model, prompt=prompt, response_text=raw_text, usage=usage, latency_ms=int((time.monotonic() - t0) * 1000), video_id=video_id) return parsed, usage def store_triage(con: sqlite3.Connection, video_id: str, scores: dict, verdict: str, reason: str, path: str, cost: float, junk_penalty: int = 0) -> None: avg = (scores["depth"] + scores["novelty"] + scores["aria_relevance"] + scores["production_ready"]) / 4.0 con.execute( "INSERT OR REPLACE INTO triaged (video_id, depth, novelty, aria_relevance, production_ready, junk_penalty, avg_score, verdict, one_line_reason, triage_path, cost_usd) VALUES (?,?,?,?,?,?,?,?,?,?,?)", (video_id, scores["depth"], scores["novelty"], scores["aria_relevance"], scores["production_ready"], int(junk_penalty), avg, verdict, reason, path, cost), ) con.commit() def write_triage_file(cfg: dict, video_id: str, payload: dict) -> str: out_dir = Path(cfg["paths"]["triage_dir"]) out_dir.mkdir(parents=True, exist_ok=True) p = out_dir / f"{video_id}.json" p.write_text(json.dumps(payload, indent=2, ensure_ascii=False)) return str(p) def log_run_start(con: sqlite3.Connection) -> int: cur = con.execute("INSERT INTO run_log (stage, status) VALUES ('triage', 'running')") con.commit() return cur.lastrowid def log_run_finish(con: sqlite3.Connection, run_id: int, processed: int, skipped: int, failed: int, notes: str = "") -> None: con.execute( "UPDATE run_log SET finished_at=datetime('now'), status='completed', items_processed=?, items_skipped=?, items_failed=?, notes=? WHERE id=?", (processed, skipped, failed, notes, run_id), ) con.commit() def main(argv: list[str]) -> int: cfg = load_config() con = db_conn(cfg["paths"]["state_db"]) ensure_schema(con) api_key = os.environ.get("ANTHROPIC_API_KEY") if not api_key: pending = get_pending(con, limit=10000) print(f"[akp-triage] ANTHROPIC_API_KEY not set. {len(pending)} videos queued.") log_run_start(con) log_run_finish(con, con.execute("SELECT MAX(id) FROM run_log").fetchone()[0], 0, len(pending), 0, "no_api_key") return 0 try: from anthropic import Anthropic except ImportError: print("[akp-triage] anthropic SDK missing. pip install anthropic") return 2 client = Anthropic(api_key=api_key) cost_cap = cfg["cost_caps"]["triage_max_usd"] todays = todays_cost(con, "triage_usd") if todays >= cost_cap: print(f"[akp-triage] daily cost-cap reached ({todays:.4f}/{cost_cap} USD). Skipping run.") return 0 pending = get_pending(con) if not pending: print("[akp-triage] no pending videos.") return 0 print(f"[akp-triage] {len(pending)} pending. cost-budget left: ${cost_cap - todays:.4f}") run_id = log_run_start(con) excerpt_max = cfg["triage"]["transcript_max_chars"] promote_avg = cfg["triage"]["promote_threshold_avg"] promote_aria = cfg["triage"]["promote_threshold_aria"] summary_avg = cfg["triage"]["summary_threshold_avg"] model = cfg["models"]["triage_model"] processed = 0 failed = 0 skipped = 0 for row in pending: if (todays_cost(con, "triage_usd") >= cost_cap): print("[akp-triage] cost-cap reached mid-run, stopping.") break try: raw = json.loads(Path(row["raw_path"]).read_text()) except (OSError, json.JSONDecodeError): failed += 1 continue transcript = (raw.get("transcript") or "")[:excerpt_max] if not transcript.strip(): store_triage(con, row["video_id"], {"depth":0,"novelty":0,"aria_relevance":0,"production_ready":0}, "skip", "empty_transcript", "", 0.0) skipped += 1 continue prompt = TRIAGE_PROMPT.format( title=row["title"] or "", channel=row["channel"] or "", duration_min=int((row["duration_seconds"] or 0) / 60), upload_date=row["upload_date"] or "unknown", transcript_excerpt=transcript, ) parsed, usage = call_haiku(client, model, prompt, video_id=row["video_id"]) cost = haiku_pricing_usd(usage["input_tokens"], usage["output_tokens"]) add_cost(con, triage_delta=cost) if not parsed or "verdict" not in parsed: store_triage(con, row["video_id"], {"depth":0,"novelty":0,"aria_relevance":0,"production_ready":0}, "skip", f"parse_fail: {usage.get('error','')[:100]}", "", cost) failed += 1 continue scores = { "depth": int(parsed.get("depth", 0)), "novelty": int(parsed.get("novelty", 0)), "aria_relevance": int(parsed.get("aria_relevance", 0)), "production_ready": int(parsed.get("production_ready", 0)), } junk_penalty = max(0, min(3, int(parsed.get("junk_penalty", 0)))) verdict, avg, eff = compute_verdict(scores, junk_penalty, promote_avg, promote_aria, summary_avg) # Modell-Veto honorieren: explizites skip schlaegt die Score-Logik if parsed.get("verdict") == "skip": verdict = "skip" reason = parsed.get("one_line_reason", "")[:300] payload = { "video_id": row["video_id"], "channel": row["channel"], "title": row["title"], "scores": scores, "junk_penalty": junk_penalty, "avg_score": avg, "effective_score": eff, "verdict": verdict, "one_line_reason": reason, "model": model, "cost_usd": cost, "triaged_at": datetime.now(timezone.utc).isoformat(), } path = write_triage_file(cfg, row["video_id"], payload) store_triage(con, row["video_id"], scores, verdict, reason, path, cost, junk_penalty) processed += 1 time.sleep(0.2) log_run_finish(con, run_id, processed, skipped, failed, f"todays_cost=${todays_cost(con,'triage_usd'):.4f}") print(f"[akp-triage] processed={processed} skipped={skipped} failed={failed} cost_today=${todays_cost(con,'triage_usd'):.4f}") con.close() return 0 if __name__ == "__main__": _log.event("script_start") sys.exit(main(sys.argv[1:]))