#!/usr/bin/env python3 """ Test the Ollama model's ability to classify release notes. Runs leave-one-out: for each example, uses the rest as few-shot context and checks whether the model predicts the correct action. Usage: python test_classifier.py [service_name] service_name defaults to "immich" """ import json import os import sys from pathlib import Path from urllib.request import Request, urlopen import re as _re OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://192.168.1.20:11434") OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "gemma4:26b-64k") SERVICE_NAME = sys.argv[1] if len(sys.argv) > 1 else "immich" EXAMPLES_DIR = Path(__file__).parent / "services" / SERVICE_NAME / "review-examples" def guess_version_from_notes(notes: str) -> str: """Extract a version number from release notes (e.g. '# v2.7.0' or '### v8.1.5').""" match = _re.search(r"v?(\d+\.\d+\.\d+)", notes) return match.group(1) if match else "1.0.0" def make_previous_version(version: str) -> str: """Decrement version to simulate upgrading from a previous release.""" parts = version.split(".") if len(parts) == 3 and parts[2].isdigit(): patch = int(parts[2]) if patch > 0: return f"{parts[0]}.{parts[1]}.{patch - 1}" # x.y.0 → decrement minor minor = int(parts[1]) if minor > 0: return f"{parts[0]}.{minor - 1}.0" # x.0.0 → decrement major major = int(parts[0]) if major > 0: return f"{major - 1}.0.0" return version def load_examples(directory: Path) -> list[dict]: examples = [] for path in sorted(directory.glob("*.txt")): text = path.read_text(encoding="utf-8") if "---" not in text: print(f"WARNING: {path.name} has no frontmatter separator, skipping") continue header, _, body = text.partition("---\n") action = None reason = None for line in header.strip().splitlines(): if line.startswith("action:"): action = line.split(":", 1)[1].strip() elif line.startswith("reason:"): reason = line.split(":", 1)[1].strip() if action not in ("auto", "manual"): print(f"WARNING: {path.name} has invalid action '{action}', skipping") continue examples.append({ "file": path.name, "action": action, "reason": reason or "", "notes": body.strip(), }) return examples def build_messages(examples: list[dict], target_notes: str, current: str = "1.0.0", latest: str = "2.0.0") -> list[dict]: context_file = EXAMPLES_DIR.parent / "context.txt" context = context_file.read_text(encoding="utf-8").strip() if context_file.is_file() else None system = ( f"You are an update safety classifier for self-hosted {SERVICE_NAME} instances.\n" f"Given release notes for a {SERVICE_NAME} update, decide whether the update can be\n" "applied automatically (action: auto) or requires the admin to intervene\n" "manually before or during the update (action: manual).\n\n" ) if context: system += f"Instance context: {context}\n\n" system += ( "Rules:\n" "- Only flag as manual if the breaking changes actually affect this specific setup.\n" "- Classify only THIS release, not future or past releases mentioned in the notes.\n" "- Warnings or announcements about breaking changes in upcoming versions do NOT\n" " make the current release manual.\n" "- Conditional notes like 'follow X instructions if upgrading from <= Y' only apply\n" " if the current running version is <= Y. If we are already past Y, they do not apply.\n" "- Respond ONLY with a JSON object: {\"action\": \"auto\", \"reason\": \"...\"}\n" " or {\"action\": \"manual\", \"reason\": \"...\"}\n" "- Keep reason under 60 words." ) messages = [{"role": "system", "content": system}] for ex in examples: messages.append({"role": "user", "content": f"Release notes:\n{ex['notes']}"}) messages.append({ "role": "assistant", "content": json.dumps({"action": ex["action"], "reason": ex["reason"]}), }) messages.append({"role": "user", "content": f"Upgrading from {current} to {latest}.\n\nRelease notes:\n{target_notes}"}) return messages def call_ollama(messages: list[dict]) -> dict: payload = json.dumps({ "model": OLLAMA_MODEL, "messages": messages, "stream": False, "options": {"temperature": 0}, }).encode("utf-8") req = Request( f"{OLLAMA_BASE_URL}/v1/chat/completions", data=payload, headers={"Content-Type": "application/json"}, method="POST", ) with urlopen(req, timeout=300) as resp: data = json.loads(resp.read()) return data def parse_response(text: str) -> dict: text = text.strip() # extract JSON from response (model may wrap it in markdown) start = text.find("{") end = text.rfind("}") + 1 if start == -1 or end == 0: return {"action": "parse_error", "reason": text[:200]} try: return json.loads(text[start:end]) except json.JSONDecodeError: return {"action": "parse_error", "reason": text[:200]} def main(): examples = load_examples(EXAMPLES_DIR) if not examples: print("No examples found.") sys.exit(1) print(f"Loaded {len(examples)} examples from {EXAMPLES_DIR}") print(f"Model: {OLLAMA_MODEL}") print(f"Running leave-one-out test ({len(examples)} calls)...\n") passed = 0 failed = 0 for i, target in enumerate(examples): few_shot = [ex for j, ex in enumerate(examples) if j != i] latest = guess_version_from_notes(target["notes"]) current = make_previous_version(latest) messages = build_messages(few_shot, target["notes"], current, latest) try: response = call_ollama(messages) content = response["choices"][0]["message"]["content"] result = parse_response(content) except Exception as exc: result = {"action": "error", "reason": str(exc)} predicted = result.get("action", "error") expected = target["action"] ok = predicted == expected status = "PASS" if ok else "FAIL" if ok: passed += 1 else: failed += 1 print(f"[{status}] {target['file']}") print(f" expected : {expected}") print(f" predicted: {predicted}") print(f" reason : {result.get('reason', '')}") print() print(f"Results: {passed}/{len(examples)} passed", end="") if failed: print(f", {failed} failed") else: print(" — all correct!") sys.exit(0 if failed == 0 else 1) if __name__ == "__main__": main()