amucfg.py: Automating AI Model Metadata for Aider with OpenRouter

amucfg.py

import requests
import json
import os
import time
import yaml
import argparse
import re
from datetime import datetime

OPENROUTER_API_BASE = "https://openrouter.ai/api/v1"
MODEL_CACHE_FILE = "openrouter_models.json"
ENDPOINT_CACHE_DIR = "endpoints_cache"
MY_MODELS_FILE = "my_models.json"
AIDER_MODEL_SETTINGS_FILE = ".aider.model.settings.yml"
AIDER_METADATA_FILE = "aider_model_metadata.json"

MY_MODEL_SLUGS = {
    "google/gemini-2.0-flash-001": ["coding"],
    "google/gemini-flash-1.5": ["bulk"],
    "anthropic/claude-3.5-sonnet": ["coding"],
    "anthropic/claude-3.5-haiku": ["bulk"],
}

MODEL_SERIES = ["GPT", "Claude", "Gemini", "DeepSeek", "Mistral", "Llama3"]
MODEL_VENDORS = ["meta-llama", "openai", "google", "anthropic", "mistral", "deepseek"]
DEFAULT_PROVIDERS = ["DeepSeek", "OpenAI", "Anthropic", "Google", "Mistral"]
DEFAULT_CONFIG_DIR = os.path.expanduser("~/.ai-model-utils")
DEFAULT_AIDER_CONFIG_DIR = os.path.expanduser("~/")

def fetch_openrouter_models(config_dir):
    try:
        response = requests.get(f"{OPENROUTER_API_BASE}/models")
        response.raise_for_status()
        models = response.json().get("data", [])
        save_json_data(models, MODEL_CACHE_FILE, config_dir)  # Cache the fetched models
        return models
    except requests.exceptions.RequestException as e:
        print(f"Error fetching models: {e}")
        return None

def load_cached_models(config_dir):
    filepath = os.path.join(config_dir, MODEL_CACHE_FILE)
    if os.path.exists(filepath):
        with open(filepath, "r") as f:
            return json.load(f)
    return None

def update_model_cache(config_dir, cache_time=86400):
    filepath = os.path.join(config_dir, MODEL_CACHE_FILE)
    if os.path.exists(filepath):
        modified_time = os.path.getmtime(filepath)
        if time.time() - modified_time < cache_time:
            return load_cached_models(config_dir)
    return fetch_openrouter_models(config_dir)

def fetch_model_endpoints(author, slug, config_dir):
    cache_file = os.path.join(config_dir, ENDPOINT_CACHE_DIR, f"{author}_{slug}.json")
    if os.path.exists(cache_file):
        with open(cache_file, "r") as f:
            return json.load(f)

    try:
        response = requests.get(f"{OPENROUTER_API_BASE}/models/{author}/{slug}/endpoints")
        response.raise_for_status()
        endpoints = response.json()
        if not os.path.exists(os.path.join(config_dir, ENDPOINT_CACHE_DIR)):
            os.makedirs(os.path.join(config_dir, ENDPOINT_CACHE_DIR))
        with open(cache_file, "w") as f:
            json.dump(endpoints, f, indent=4)
        return endpoints
    except requests.exceptions.RequestException as e:
        print(f"Error fetching endpoints for {author}/{slug}: {e}")
        return None

def create_my_models_json(models):
    filtered_models = []

    for model in models:
        # Split vendor / model_slug from id
        vendor, module_slug = model["id"].split("/", 1)
        # Filter by Model Vendor
        if vendor in MODEL_VENDORS and model["architecture"]["tokenizer"] in MODEL_SERIES:
            # 1. General Timestamp Filtering (Keep recent models)
            if model["created"] > 1700000000:  # Adjust this timestamp as needed

                model_slug = model["id"]

                vendor, model_name = model_slug.split("/", 1) #Split on the / for vendor/model_name
                match = re.match(r"^(.*?)(?:-([^-:]*))?(?::(.*))?$", model_name) #Regex on model_name
                if match:
                    series = match.group(1)
                    version = match.group(2)
                    tags = match.group(3) if match.group(3) else None
                else:
                    series = model["name"]
                    version = None
                    tags = None

                print(f"Processing model: {model['name']}")
                print(f"Series: {series}, Version: {version}, Tags: {tags}")
                exclude = False

                # Simplified Exclusion Logic (Exclude if ANY tags are present)
                if tags:
                    exclude = True

                if not exclude:
                    filtered_model = {
                        "id": model["id"],
                        "name": model["name"],
                        "created": model["created"],
                        "created_date": datetime.fromtimestamp(model["created"]).strftime("%Y-%m-%d"),
                        "architecture": model["architecture"],
                        "providers": [],
                        "series": series,
                        "version": version,
                        "tags": tags,
                    }
                    filtered_models.append(filtered_model)

    filtered_models.sort(key=lambda x: x["id"])

    return filtered_models

def create_aider_metadata_json(my_models, config_dir):
    aider_metadata = {}
    for model in my_models:
        model_slug = model["id"]
        # Check if the model slug is a key in MY_MODEL_SLUGS
        if model_slug not in MY_MODEL_SLUGS.keys():
            continue

        author, slug = model_slug.split("/", 1)
        model_endpoints = fetch_model_endpoints(author, slug, config_dir)
# Retrieve the endpoints from the model_endpoints dictionary,
# accessing the 'data' key and then the 'endpoints' key within the 'data' dictionary.
# If either of these keys do not exist, an empty dictionary will be returned by default.
        endpoints = model_endpoints.get("data", {}).get("endpoints", {})
# The retrieved endpoints are stored in the 'endpoints' variable for further use.
# The following line is commented out, but it would assign the endpoints to the 'endpoints' key within the 'model' dictionary.
# model["endpoints"] = endpoints

        # Initialize default values
        max_tokens = 1000000
        max_input_tokens = 1048576
        max_output_tokens = 8192
        input_cost_per_token = 0
        output_cost_per_token = 0
        litellm_provider = "openrouter"
        mode = "chat"

        model_key = f"{litellm_provider}/{model_slug}"
        print(f"Processing model: {model_key}")
        # Process endpoints to extract relevant information
        if endpoints:
            # Assuming you want to use the first endpoint for now
            endpoint = endpoints[0]

            max_tokens = endpoint.get("context_length", max_tokens)
            max_completion_tokens = endpoint.get("max_completion_tokens", max_output_tokens)
            max_prompt_tokens = endpoint.get("max_prompt_tokens", max_input_tokens)

            pricing = endpoint.get("pricing", {})

            input_cost_per_token = pricing.get("prompt")
            output_cost_per_token = pricing.get("completion")

        aider_metadata[model_key] = {
            "max_tokens": max_tokens,
            "max_input_tokens": max_prompt_tokens,
            "max_output_tokens": max_completion_tokens,
            "input_cost_per_token": input_cost_per_token,
            "output_cost_per_token": output_cost_per_token,
            "litellm_provider": "openrouter",
            "mode": mode
        }

    filepath = os.path.join(config_dir, AIDER_METADATA_FILE)
    with open(filepath, "w") as f:
        json.dump(aider_metadata, f, indent=4)
    print("Aider metadata JSON created.")

def get_config_dir(args):
    if args.config_dir:
        return args.config_dir
    else:
        return DEFAULT_CONFIG_DIR

def ensure_config_dir(config_dir):
    if not os.path.exists(config_dir):
        os.makedirs(config_dir)
        print(f"Created configuration directory: {config_dir}")

def save_json_data(data, filename, config_dir):
    filepath = os.path.join(config_dir, filename)
    with open(filepath, "w") as f:
        json.dump(data, f, indent=4)
    print(f"{filename} saved to {filepath}")

def load_json_data(filename, config_dir):
    filepath = os.path.join(config_dir, filename)
    if os.path.exists(filepath):
        with open(filepath, "r") as f:
            return json.load(f)
    return None

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="AI Model Utilities")
    parser.add_argument("--config-dir", help="Path to the configuration directory")
    args = parser.parse_args()

    config_dir = get_config_dir(args)
    ensure_config_dir(config_dir)

    print(f"Using config directory: {config_dir}")
    # Load cached models or fetch if not available
    models = load_cached_models(config_dir)
    if not models:
        models = fetch_openrouter_models(config_dir)

    # Now with OpenRouter models, we need to create My Models JSON
    if models:
        my_models = create_my_models_json(models)

        if my_models:
            save_json_data(my_models, MY_MODELS_FILE, config_dir)
            create_aider_metadata_json(my_models, config_dir)

BUG Fix: Aider model config now uses model_key instead of model_slug so openrouter/ is added to model_slug