#!/usr/bin/env python3 """Varion Local Tester v2.4.0. Runs the original and the real Varion-optimised request directly against OpenAI, Anthropic Claude, or an OpenAI-compatible OpenClaw gateway. Privacy model: - Provider API keys stay on this computer and are sent only to the provider. - The request body is sent to Varion's optimise-only endpoint. - Varion does not call the provider and does not store prompt content. - Only metrics are uploaded to the dashboard when UPLOAD_METRICS is enabled. Works in Jupyter, VS Code, and a normal terminal. """ from __future__ import annotations import argparse import copy import getpass import json import os import sys import time import urllib.error import urllib.request from difflib import SequenceMatcher from pathlib import Path from typing import Any, Callable # SIMPLE SETTINGS FOR JUPYTER / VS CODE PROVIDER = "openai" # openai, claude, or openclaw REQUEST_FILE = "" # blank = safe built-in example UPLOAD_METRICS = False # optional metrics only; no keys, prompts, or answers OPENAI_SAMPLE: dict[str, Any] = { "model": "gpt-4o-mini", "messages": [ {"role": "system", "content": "You are a concise support assistant."}, {"role": "user", "content": "Confirm politely that order 12345 was received."}, {"role": "user", "content": "Confirm politely that order 12345 was received."}, ], "temperature": 0, } CLAUDE_SAMPLE: dict[str, Any] = { "model": "claude-3-5-haiku-latest", "max_tokens": 120, "messages": [ {"role": "user", "content": "Confirm politely that order 12345 was received."}, {"role": "user", "content": "Confirm politely that order 12345 was received."}, ], "temperature": 0, } def is_notebook() -> bool: return "ipykernel" in sys.modules or hasattr(sys, "ps1") def parse_options() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run original versus real Varion-optimised AI requests.") parser.add_argument("--provider", choices=["openai", "claude", "openclaw"], default=None) parser.add_argument("--request", default=None, help="Path to one JSON request object") parser.add_argument("--no-upload", action="store_true", help="Do not upload metrics to the dashboard") # Jupyter injects -f ; parse_known_args ignores it safely. args, _unknown = parser.parse_known_args() args.provider = (args.provider or os.getenv("VARION_TEST_PROVIDER") or PROVIDER).strip().lower() args.request = args.request if args.request is not None else os.getenv("VARION_REQUEST_FILE", REQUEST_FILE) args.upload = bool(UPLOAD_METRICS and not args.no_upload) return args def read_secret(env_name: str, label: str, required: bool = True) -> str: value = os.getenv(env_name, "").strip() if value: return value if not required: return "" try: value = getpass.getpass(f"{label} (hidden, not saved): ").strip() except (EOFError, KeyboardInterrupt) as exc: raise RuntimeError(f"Missing {env_name}") from exc if not value: raise RuntimeError(f"Missing {env_name}") return value def load_request(provider: str, request_file: str) -> dict[str, Any]: request_file = (request_file or "").strip() if not request_file: return copy.deepcopy(CLAUDE_SAMPLE if provider == "claude" else OPENAI_SAMPLE) path = Path(request_file).expanduser() if not path.is_file(): raise RuntimeError(f"Request file not found: {path}") try: value = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: raise RuntimeError(f"Invalid JSON in {path}: {exc}") from exc if not isinstance(value, dict): raise RuntimeError("Request JSON must contain one object") return value def post_json(url: str, payload: dict[str, Any], headers: dict[str, str], timeout: int = 180) -> tuple[dict[str, Any], int]: started = time.perf_counter() req = urllib.request.Request( url, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), headers={"Content-Type": "application/json", "User-Agent": "VarionLocalTester/2.4.0", **headers}, method="POST", ) try: with urllib.request.urlopen(req, timeout=timeout) as response: body = response.read().decode("utf-8", "replace") return json.loads(body), int((time.perf_counter() - started) * 1000) except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", "replace") raise RuntimeError(f"HTTP {exc.code} from {url}: {detail[:1200]}") from exc except urllib.error.URLError as exc: raise RuntimeError(f"Connection failed for {url}: {exc.reason}") from exc def provider_config(provider: str) -> tuple[str, dict[str, str], Callable[[dict[str, Any]], tuple[int, int]], str]: if provider == "claude": key = read_secret("ANTHROPIC_API_KEY", "Anthropic API key") base = os.getenv("ANTHROPIC_BASE_URL", "https://api.anthropic.com/v1").rstrip("/") return f"{base}/messages", {"x-api-key": key, "anthropic-version": "2023-06-01"}, usage_claude, "Anthropic Claude" if provider == "openclaw": key = read_secret("OPENCLAW_API_KEY", "OpenClaw API key", required=False) base = os.getenv("OPENCLAW_BASE_URL", "http://127.0.0.1:3000/v1").rstrip("/") headers = {"Authorization": f"Bearer {key}"} if key else {} return f"{base}/chat/completions", headers, usage_openai, "OpenClaw" key = read_secret("OPENAI_API_KEY", "OpenAI API key") base = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1").rstrip("/") return f"{base}/chat/completions", {"Authorization": f"Bearer {key}"}, usage_openai, "OpenAI" def usage_openai(data: dict[str, Any]) -> tuple[int, int]: usage = data.get("usage") or {} return int(usage.get("prompt_tokens") or usage.get("input_tokens") or 0), int(usage.get("completion_tokens") or usage.get("output_tokens") or 0) def usage_claude(data: dict[str, Any]) -> tuple[int, int]: usage = data.get("usage") or {} return int(usage.get("input_tokens") or 0), int(usage.get("output_tokens") or 0) def answer_text(data: dict[str, Any]) -> str: choices = data.get("choices") or [] if choices and isinstance(choices[0], dict): return json.dumps(choices[0].get("message") or choices[0], ensure_ascii=False, sort_keys=True) content = data.get("content") if isinstance(content, list): return "".join(str(item.get("text") or "") for item in content if isinstance(item, dict)) return json.dumps(data, ensure_ascii=False, sort_keys=True) def varion_root() -> str: root = os.getenv("VARION_BASE_URL", "https://api.varion.tech").rstrip("/") return root[:-3] if root.endswith("/v1") else root def optimize_with_varion(request_body: dict[str, Any], varion_key: str) -> dict[str, Any]: result, _ = post_json( varion_root() + "/v1/test/optimize", {"request": request_body}, {"Authorization": f"Bearer {varion_key}"}, timeout=60, ) optimized = result.get("optimized_request") if not isinstance(optimized, dict): raise RuntimeError("Varion did not return an optimized request") return result def upload_metrics(result: dict[str, Any], varion_key: str) -> None: payload = { "provider": result["provider"], "model": result["model"], "verdict": result["verdict"], "original_input_tokens": result["original_input_tokens"], "optimized_input_tokens": result["optimized_input_tokens"], "original_output_tokens": result["original_output_tokens"], "optimized_output_tokens": result["optimized_output_tokens"], "original_latency_ms": result["original_latency_ms"], "optimized_latency_ms": result["optimized_latency_ms"], "text_similarity": result["text_similarity"], "tool_similarity": None, "usage_verified": True, } post_json(varion_root() + "/api/customer/local-test-result", payload, {"Authorization": f"Bearer {varion_key}"}, timeout=60) def run_test(args: argparse.Namespace) -> dict[str, Any]: request_body = load_request(args.provider, args.request) varion_key = read_secret("VARION_API_KEY", "Free Varion test API key") preview = optimize_with_varion(request_body, varion_key) optimized_body = preview["optimized_request"] url, provider_headers, usage_reader, provider_label = provider_config(args.provider) print(f"\nRunning original and real Varion-optimised requests directly against {provider_label}...") print("Provider key destination: provider only") print("Request content destination: Varion optimise-only endpoint; prompt content is not stored") original_response, original_latency = post_json(url, request_body, provider_headers) optimized_response, optimized_latency = post_json(url, optimized_body, provider_headers) original_input, original_output = usage_reader(original_response) optimized_input, optimized_output = usage_reader(optimized_response) saved = max(0, original_input - optimized_input) reduction = round(saved / original_input * 100, 2) if original_input else 0.0 similarity = round(SequenceMatcher(None, answer_text(original_response), answer_text(optimized_response)).ratio() * 100, 2) result = { "provider": args.provider, "model": str(request_body.get("model") or "unknown"), "verdict": "TOKENS_SAVED" if saved else "NO_MEASURED_SAVINGS", "original_input_tokens": original_input, "optimized_input_tokens": optimized_input, "original_output_tokens": original_output, "optimized_output_tokens": optimized_output, "saved_input_tokens": saved, "reduction_percent": reduction, "original_latency_ms": original_latency, "optimized_latency_ms": optimized_latency, "text_similarity": similarity, "actions": preview.get("actions") or [], "provider_key_sent_to_varion": False, "prompt_stored_by_varion": False, } print("\nRESULT") print(json.dumps(result, indent=2, ensure_ascii=False)) if args.upload: upload_metrics(result, varion_key) print("\nMetrics-only dashboard upload succeeded.") return result def main() -> None: try: run_test(parse_options()) except Exception as exc: print(f"\nTEST FAILED: {exc}") print("No provider key was sent to or stored by Varion.") if not is_notebook(): raise SystemExit(1) from exc if __name__ == "__main__": main()