#!/usr/bin/env python3 """Hermes Session Cleaner — delete short sessions, generate missing titles. Usage: hermes_cleaner.py [--dry-run] [--min-messages N] [--no-delete] [--no-title] Scans the Hermes session DB and: 1. Deletes sessions with fewer than N messages (default: 5) 2. Generates titles for sessions that don't have one (using local Ollama) Safety: --dry-run shows what would happen without making changes. """ import argparse import json import os import re import sys from datetime import datetime from pathlib import Path from urllib.request import Request, urlopen from urllib.error import URLError HERMES_HOME = Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes")) sys.path.insert(0, str(HERMES_HOME / "hermes-agent")) from hermes_state import SessionDB # noqa: E402 # ── Markdown cleanup ──────────────────────────────────────────────────────── def strip_markdown(text: str) -> str: """Strip markdown formatting.""" text = re.sub(r"```.*?```", "", text, flags=re.DOTALL) text = re.sub(r"`([^`]+)`", r"\1", text) text = re.sub(r"\*\*\*([^*]+)\*\*\*", r"\1", text) text = re.sub(r"\*\*([^*]+)\*\*", r"\1", text) text = re.sub(r"\*([^*]+)\*", r"\1", text) text = re.sub(r"__([^_]+)__", r"\1", text) text = re.sub(r"_([^_]+)_", r"\1", text) text = re.sub(r"^>\s*", "", text, flags=re.MULTILINE) return text.strip() # ── Title generation ──────────────────────────────────────────────────────── def build_title_prompt(msgs: list[dict]) -> str: """Build prompt from conversation messages.""" conversation = [] for m in msgs: content = (m.get("content") or "").strip() if len(content) > 400: content = content[:400] + "..." if content: conversation.append(f"[{m['role']}] {content}") transcript = "\n\n".join(conversation) return ( "You are a title generator. Your ONLY job is to output a plain text title.\n\n" "Generate a short, descriptive title (3-7 words, max 50 characters) for the " "conversation below. Capture the MAIN topic.\n\n" "FORMAT RULES:\n" "- Output ONLY the title text. Nothing else.\n" "- Plain text only — NO markdown, NO quotes, NO formatting.\n" "- NO punctuation at the end. NO 'Title:' prefix.\n\n" f"CONVERSATION:\n{transcript}\n\n" "TITLE:" ) def call_llm(prompt: str, endpoint: str, model: str, api_key: str = "", timeout: int = 60) -> str | None: """Call OpenAI-compatible API for title generation.""" body = json.dumps({ "model": model, "messages": [{"role": "user", "content": prompt}], "max_tokens": 500, # reasoning models need space for thinking "temperature": 0.1, }).encode("utf-8") headers = {"Content-Type": "application/json"} if api_key: headers["Authorization"] = f"Bearer {api_key}" try: req = Request(endpoint, data=body, headers=headers) with urlopen(req, timeout=timeout) as resp: data = json.loads(resp.read()) msg = data["choices"][0]["message"] raw = (msg.get("content") or msg.get("reasoning_content") or msg.get("reasoning") or "").strip() title = strip_markdown(raw) title = title.strip("\"'«»„”") title = re.sub(r"^(Title|Başlık|TITLE)[:\s-]*", "", title, flags=re.IGNORECASE) if len(title) > 50: title = title[:47] + "..." return title if title else None except Exception as e: print(f" LLM error: {e}", file=sys.stderr) return None def generate_title_for_session(db: SessionDB, session_id: str, endpoint: str, model: str, api_key: str = "") -> str | None: """Generate and set a title for a session.""" msgs = db.get_messages_as_conversation(session_id) user_msgs = [m for m in msgs if m.get("role") in ("user", "assistant")] if len(user_msgs) < 2: return None prompt = build_title_prompt(user_msgs) title = call_llm(prompt, endpoint, model, api_key) if title: db.set_session_title(session_id, title) return title # ── Main ──────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Hermes session cleaner — delete short sessions + missing titles" ) parser.add_argument("--dry-run", action="store_true", help="Show what would happen, don't change anything") parser.add_argument("--min-messages", type=int, default=5, help="Delete sessions with fewer than N messages (default: 5)") parser.add_argument("--no-delete", action="store_true", help="Skip deletion, only generate titles") parser.add_argument("--no-title", action="store_true", help="Skip title generation, only delete") parser.add_argument("--max-titles", type=int, default=20, help="Max titles to generate in one run (default: 20)") args = parser.parse_args() db_path = HERMES_HOME / "state.db" if not db_path.exists(): print(f"Session DB not found: {db_path}", file=sys.stderr) sys.exit(1) db = SessionDB(db_path) # Default: DeepSeek API (works everywhere, needs DEEPSEEK_API_KEY in .env) endpoint = os.environ.get( "HERMES_CLEANER_ENDPOINT", "https://api.deepseek.com/v1/chat/completions" ) model = os.environ.get("HERMES_CLEANER_MODEL", "deepseek-v4-flash") # Load API key from .env (DeepSeek) or env override api_key = os.environ.get("HERMES_CLEANER_API_KEY", "") if not api_key: env_path = HERMES_HOME / ".env" if env_path.exists(): for line in env_path.read_text().splitlines(): if "DEEPSEEK_API_KEY" in line and "=" in line: api_key = line.split("=", 1)[-1].strip().strip('"').strip("'") break # ── Scan ──────────────────────────────────────────────────────────── sessions = db.list_sessions_rich(limit=500) to_delete = [] to_title = [] for s in sessions: sid = s["id"] msg_count = int(s.get("message_count", 0)) title = s.get("title") or "" source = s.get("source", "?") if msg_count < args.min_messages: to_delete.append((sid, msg_count, title, source)) elif not title: to_title.append((sid, msg_count, source)) print(f"Sessions scanned : {len(sessions)}") print(f"To delete (<{args.min_messages} msgs) : {len(to_delete)}") print(f"To title (>= {args.min_messages} msgs, no title): {len(to_title)}") # ── Delete short sessions ─────────────────────────────────────────── if not args.no_delete and to_delete: print(f"\n--- Deleting {len(to_delete)} short sessions ---") for sid, cnt, title, src in to_delete: title_info = f' "{title}"' if title else "" if args.dry_run: print(f" [DRY RUN] Would delete: {sid[:30]} msgs={cnt} src={src}{title_info}") else: try: db.delete_session(sid) print(f" Deleted: {sid[:30]} msgs={cnt} src={src}{title_info}") except Exception as e: print(f" FAILED: {sid[:30]} — {e}", file=sys.stderr) elif args.no_delete: print("\n(delete skipped via --no-delete)") # ── Generate missing titles ───────────────────────────────────────── if not args.no_title and to_title: limit = min(len(to_title), args.max_titles) print(f"\n--- Generating titles ({limit} of {len(to_title)}) ---") generated = 0 for sid, cnt, src in to_title[:limit]: if args.dry_run: print(f" [DRY RUN] Would title: {sid[:30]} msgs={cnt} src={src}") continue print(f" {sid[:30]} msgs={cnt} src={src} ...", end=" ", flush=True) title = generate_title_for_session(db, sid, endpoint, model, api_key) if title: print(f'→ "{title}"') generated += 1 else: print("FAILED") print(f"\nTitles generated: {generated}/{limit}") elif args.no_title: print("\n(title generation skipped via --no-title)") if args.dry_run: print("\n[DRY RUN] No changes made.") if __name__ == "__main__": main()