#!/usr/bin/env python3 """Compare three LLMs: DeepSeek V4 Flash, DeepSeek V4 Pro, Aillm.ru. Usage: DEEPSEEK_API_KEY=sk-... python tools/compare_llm.py """ import os import sys import time from pathlib import Path import requests import yaml TEST_DATA = { "vin": "WVWZZZ1KZAW123456", "dtc_codes": [ {"code": "P0301", "status": "stored", "description": "Misfire cylinder 1"}, {"code": "P0171", "status": "stored", "description": "System too lean bank 1"}, {"code": "P0420", "status": "pending", "description": "Catalyst efficiency low bank 1"}, ], "parameters": [ {"pid_code": "0105", "name": "coolant_temp", "value": 87.0, "unit": "C"}, {"pid_code": "010C", "name": "rpm", "value": 680, "unit": "rpm"}, {"pid_code": "010D", "name": "speed", "value": 0, "unit": "km/h"}, {"pid_code": "010B", "name": "map", "value": 35, "unit": "kPa"}, {"pid_code": "010F", "name": "iat", "value": 31, "unit": "C"}, {"pid_code": "0104", "name": "engine_load", "value": 18.0, "unit": "%"}, {"pid_code": "0106", "name": "stft_b1", "value": -8.6, "unit": "%"}, {"pid_code": "0107", "name": "ltft_b1", "value": -12.5, "unit": "%"}, {"pid_code": "0111", "name": "throttle_pos", "value": 12, "unit": "%"}, ], } SYSTEM_PROMPT = ( "You are an auto diagnostic expert with 20 years of experience. " "Analyze OBD2 error codes and ECU parameters. Give a DEEP, DETAILED analysis.\n\n" "RULES:\n" "1. Do NOT limit yourself to a brief summary - give FULL analysis of each error and parameter.\n" "2. For each error: explain what it means, ALL possible causes (common to rare), " "which parameters confirm/refute each version.\n" "3. Analyze RELATIONSHIPS between errors and parameters.\n" "4. Give confidence percentages for EACH conclusion.\n" "5. If data is insufficient - list SPECIFIC PIDs to read additionally, explain why.\n" "6. Suggest action plan: what to check FIRST (most likely and cheapest), THEN what.\n" "7. For each action: HOW to check, WHAT to look for, normal/deviant values.\n" "8. Add 'If not helped' section - plan B for each item.\n" "9. NEVER give categorical commands 'replace part X' without 100% confidence. " "Write 'check X before replacing Y'.\n" "10. Write in Russian, accessible but TECHNICALLY PRECISE. Use tables where appropriate.\n\n" "OUTPUT FORMAT:\n" "## Diagnosis (detailed)\n...\n## Error analysis\n...\n## Parameter analysis\n...\n" "## Relationships\n...\n## Action plan (by priority)\n...\n" "## What data is missing\n...\n## Confidence\n..." ) def build_user_prompt(data: dict) -> str: lines = [f"**VIN:** {data['vin']}", ""] if data["dtc_codes"]: lines.append("**Error codes:**") for d in data["dtc_codes"]: lines.append(f"- {d['code']} ({d['status']}): {d.get('description', '')}") lines.append("") if data["parameters"]: lines.append("**ECU parameters:**") for p in data["parameters"]: lines.append(f"- {p['name']}: {p['value']} {p['unit']}") lines.append("") lines.append("Conduct a full diagnosis.") return "\n".join(lines) def call_llm(api_key: str, base_url: str, model: str, messages: list[dict]) -> dict: start = time.time() resp = requests.post( f"{base_url.rstrip('/')}/chat/completions", headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, json={ "model": model, "messages": messages, "temperature": 0.3, "max_tokens": 32000, }, timeout=300, ) elapsed = time.time() - start resp.raise_for_status() data = resp.json() usage = data.get("usage", {}) return { "content": data["choices"][0]["message"]["content"], "elapsed": elapsed, "prompt_tokens": usage.get("prompt_tokens", "?"), "completion_tokens": usage.get("completion_tokens", "?"), "total_tokens": usage.get("total_tokens", "?"), } def print_answer(label: str, result: dict | None): if result is None: return print() border = "=" * 20 print(f"{border} {label} {border}") print() content = result["content"] print(content) print() def main(): config_path = Path(__file__).parent.parent / "config.yaml" with open(config_path) as f: cfg = yaml.safe_load(f) aillm_key = cfg["llm"]["api_key"] aillm_model = cfg["llm"].get("model", "gpt-oss-120b") aillm_url = cfg["llm"].get("base_url", "https://api.aillm.ru/v1") ds_key = os.environ.get("DEEPSEEK_API_KEY") if not ds_key: print("ERROR: DEEPSEEK_API_KEY not set") sys.exit(1) ds_url = "https://api.deepseek.com/v1" user_text = build_user_prompt(TEST_DATA) messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_text}, ] # Print the prompt being sent print("=" * 70) print("PROMPT BEING SENT") print("=" * 70) print() print("--- SYSTEM ---") print(SYSTEM_PROMPT[:300] + "...") print() print("--- USER ---") print(user_text) print() models = [ ("DeepSeek V4 Flash", "deepseek-v4-flash", ds_key, ds_url), ("DeepSeek V4 Pro", "deepseek-v4-pro", ds_key, ds_url), ("Aillm.ru", aillm_model, aillm_key, aillm_url), ] results = {} for label, model, key, url in models: print("-" * 70) print(f" Calling {label} ({model})...", end=" ", flush=True) try: r = call_llm(key, url, model, messages) results[label] = r print(f"DONE {r['elapsed']:.1f}s " f"in={r['prompt_tokens']} out={r['completion_tokens']}") except Exception as e: print(f"ERROR: {e}") results[label] = None print() print("=" * 70) print() for label in [m[0] for m in models]: print_answer(label, results[label]) # Summary table print("=" * 70) print("SUMMARY") print("=" * 70) print() print(f"{'Model':<25} {'Time':>8} {'Input':>8} {'Output':>8} {'Cost':>12}") print("-" * 65) for label, _, key, url in models: r = results[label] if r is None: print(f"{label:<25} {'FAIL':>8}") else: if "DeepSeek" in label: cost = (r["prompt_tokens"] / 1_000_000 * 0.14 + r["completion_tokens"] / 1_000_000 * 0.28) cost_str = f"${cost:.6f}" else: cost_str = "?" print(f"{label:<25} {r['elapsed']:>7.1f}s {r['prompt_tokens']:>8} " f"{r['completion_tokens']:>8} {cost_str:>12}") print() if __name__ == "__main__": main()