LLM Arena

arena.py

import requests
import json
import time
import os
from typing import List, Dict, Any, Optional, Tuple
from datetime import datetime

class AIModel:
    def __init__(
        self, model_name: str, system_prompt: str = "", temperature: float = 0.7
    ):
        self.model_name = model_name
        self.system_prompt = system_prompt
        self.temperature = temperature
        self.conversation_history = []
        self.total_time = 0.0
        self.response_count = 0

    def generate_response_stream(self, message: str) -> tuple[str, float]:
        url = "http://localhost:11434/api/chat"

        messages = self.conversation_history.copy()
        messages.append({"role": "user", "content": message})

        payload = {
            "model": self.model_name,
            "messages": messages,
            "stream": True,
            "temperature": self.temperature,
        }

        if self.system_prompt:
            payload["system"] = self.system_prompt

        start_time = time.time()
        full_response = ""

        try:
            with requests.post(url, json=payload, stream=True) as response:
                response.raise_for_status()

                for line in response.iter_lines():
                    if line:
                        chunk = json.loads(line)
                        if "message" in chunk and "content" in chunk["message"]:
                            content = chunk["message"]["content"]
                            print(content, end="", flush=True)
                            full_response += content

                end_time = time.time()
                elapsed_time = end_time - start_time

                self.conversation_history.append({"role": "user", "content": message})
                self.conversation_history.append(
                    {"role": "assistant", "content": full_response}
                )

                self.total_time += elapsed_time
                self.response_count += 1

                return full_response, elapsed_time

        except Exception as e:
            end_time = time.time()
            elapsed_time = end_time - start_time
            error_msg = f"Error: Could not generate response from {self.model_name}."
            print(error_msg)
            return error_msg, elapsed_time


def update_title_with_speeds(model1: AIModel, model2: AIModel) -> None:
    if model1.response_count == 0 or model2.response_count == 0:
        os.system("title AI Conversation: Not enough data to compare speeds")
        return

    model1_avg = model1.total_time / model1.response_count
    model2_avg = model2.total_time / model2.response_count

    faster_model = model1.model_name if model1_avg < model2_avg else model2.model_name

    title = f"AI Conversation: {model1.model_name} ({model1_avg:.2f}s) vs {model2.model_name} ({model2_avg:.2f}s) - {faster_model} is faster!"
    os.system(f"title {title}")


def run_conversation(
    model1: AIModel,
    model2: AIModel,
    initial_message: str,
    num_turns: int = 5,
    delay_seconds: float = 0.5,
) -> List[Tuple[str, str]]:
    conversation = []
    current_message = initial_message

    print(
        f"\n--- Starting conversation between {model1.model_name} and {model2.model_name} ---\n"
    )

    print(f"Initial message: {initial_message}\n")

    for turn in range(num_turns):
        print(f"--- Turn {turn+1}/{num_turns} ---")

        # Model 1's turn
        print(f"\n{model1.model_name}: ", end="", flush=True)
        response1, elapsed1 = model1.generate_response_stream(current_message)
        print(f" [took {elapsed1:.2f}s]\n")
        conversation.append((model1.model_name, response1))

        update_title_with_speeds(model1, model2)
        time.sleep(delay_seconds)

        # Model 2's turn
        print(f"{model2.model_name}: ", end="", flush=True)
        response2, elapsed2 = model2.generate_response_stream(response1)
        print(f" [took {elapsed2:.2f}s]\n")
        conversation.append((model2.model_name, response2))

        update_title_with_speeds(model1, model2)
        current_message = response2
        time.sleep(delay_seconds)

    print("\n--- Conversation ended ---\n")
    print(
        f"{model1.model_name} average response time: {model1.total_time / model1.response_count:.2f}s"
    )
    print(
        f"{model2.model_name} average response time: {model2.total_time / model2.response_count:.2f}s"
    )
    faster_model = (
        model1.model_name
        if (model1.total_time / model1.response_count)
        < (model2.total_time / model2.response_count)
        else model2.model_name
    )
    print(f"Faster model: {faster_model}")
    return conversation


def save_conversation(
    conversation: List[Tuple[str, str]], filename: str = "ai_conversation.txt"
) -> None:
    with open(filename, "w", encoding="utf-8") as f:
        f.write("AI-to-AI Conversation\n")
        f.write("====================\n\n")

        for speaker, message in conversation:
            f.write(f"{speaker}: {message}\n\n")

    print(f"Conversation saved to {filename}")


def main():
    # Hardcoded configuration values
    model1_name = "llama3.2"
    model2_name = "phi4-mini:latest"
    system1 = "Sen türkçe konuşan bir yapay zeka asistanısın. Kısa ve öz cevaplar vermeyi seviyorsun. Türkçe konuşuyorsun."
    system2 = "Sen türkçe konuşan bir yapay zeka asistanısın. Kısa ve öz cevaplar vermeyi seviyorsun. Türkçe konuşuyorsun."
    initial_message = "Merhaba"
    num_turns = 100
    delay_seconds = 0.1
    output_file = "ai_conversation.txt"
    temp1 = 0.25
    temp2 = 0.25

    # Initialize models
    model1 = AIModel(model1_name, system1, temp1)
    model2 = AIModel(model2_name, system2, temp2)

    # Set initial window title
    os.system(f"title AI Conversation: {model1_name} vs {model2_name}")

    # Run the conversation
    conversation = run_conversation(
        model1, model2, initial_message, num_turns, delay_seconds
    )

    # Save the conversation
    save_conversation(conversation, output_file)


if __name__ == "__main__":
    main()