Opinion: 2026 Is the Year of Local AI – Ditch Cloud Models for Ollama 0.6 and Llama 3.1

import ollama
import time
import json
import sys
from typing import Dict, List, Optional
import logging

# Configure logging for error tracking
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class OllamaBenchmarker:
    """Benchmark Ollama 0.6 inference performance against cloud models"""

    def __init__(self, model_name: str = "llama3.1:8b", ollama_host: str = "http://localhost:11434"):
        self.model_name = model_name
        self.ollama_host = ollama_host
        self.client = ollama.Client(host=ollama_host)
        self._verify_ollama_running()
        self._ensure_model_pulled()

    def _verify_ollama_running(self) -> None:
        """Check if Ollama daemon is running, raise error if not"""
        try:
            self.client.list()
            logger.info(f"Ollama daemon verified running at {self.ollama_host}")
        except Exception as e:
            logger.error(f"Ollama not reachable at {self.ollama_host}: {str(e)}")
            logger.error("Install Ollama 0.6+ from https://github.com/ollama/ollama/releases/tag/v0.6.0")
            sys.exit(1)

    def _ensure_model_pulled(self) -> None:
        """Pull Llama 3.1 model if not already present"""
        try:
            local_models = [m["name"] for m in self.client.list()["models"]]
            if self.model_name not in local_models:
                logger.info(f"Pulling {self.model_name} (this may take 5-10 minutes for 8B model)...")
                self.client.pull(self.model_name)
                logger.info(f"Successfully pulled {self.model_name}")
            else:
                logger.info(f"Model {self.model_name} already present locally")
        except Exception as e:
            logger.error(f"Failed to pull model {self.model_name}: {str(e)}")
            sys.exit(1)

    def run_inference(self, prompt: str, max_tokens: int = 512) -> Dict:
        """Run single inference pass with latency and token metrics"""
        start_time = time.perf_counter()
        try:
            response = self.client.generate(
                model=self.model_name,
                prompt=prompt,
                options={
                    "num_predict": max_tokens,
                    "temperature": 0.7,
                    "top_p": 0.9
                }
            )
            end_time = time.perf_counter()
            latency_ms = (end_time - start_time) * 1000
            return {
                "prompt": prompt,
                "response": response["response"],
                "latency_ms": round(latency_ms, 2),
                "eval_count": response["eval_count"],
                "eval_duration_ms": round(response["eval_duration"] / 1e6, 2),
                "tokens_per_second": round(response["eval_count"] / (response["eval_duration"] / 1e9), 2)
            }
        except Exception as e:
            logger.error(f"Inference failed for prompt: {prompt[:50]}... Error: {str(e)}")
            return {"error": str(e)}

    def run_benchmark(self, prompts: List[str], iterations: int = 3) -> Dict:
        """Run benchmark across multiple prompts, average results"""
        results = []
        for prompt in prompts:
            for i in range(iterations):
                logger.info(f"Running iteration {i+1}/{iterations} for prompt: {prompt[:30]}...")
                result = self.run_inference(prompt)
                if "error" not in result:
                    results.append(result)
        if not results:
            return {"error": "No successful inference runs"}
        avg_latency = sum(r["latency_ms"] for r in results) / len(results)
        avg_tokens_per_sec = sum(r["tokens_per_second"] for r in results) / len(results)
        return {
            "model": self.model_name,
            "total_runs": len(results),
            "avg_latency_ms": round(avg_latency, 2),
            "avg_tokens_per_second": round(avg_tokens_per_sec, 2),
            "raw_results": results
        }

if __name__ == "__main__":
    # Test prompts covering common dev use cases: code gen, docstring, debug, summarization
    TEST_PROMPTS = [
        "Write a Python function to reverse a linked list with type hints.",
        "Generate a docstring for the following function: def calculate_tax(income: float, brackets: List[Dict]) -> float:",
        "Debug this code: import pandas as pd; df = pd.read_csv('data.csv'); print(df['nonexistent_col'])",
        "Summarize the following text in 2 sentences: Llama 3.1 is a family of open-weight models from Meta..."
    ]

    benchmarker = OllamaBenchmarker(model_name="llama3.1:8b")
    benchmark_results = benchmarker.run_benchmark(TEST_PROMPTS, iterations=3)

    print(json.dumps(benchmark_results, indent=2))
    logger.info(f"Benchmark complete. Avg latency: {benchmark_results.get('avg_latency_ms')}ms, Avg tokens/sec: {benchmark_results.get('avg_tokens_per_second')}")

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/ollama/ollama/api"
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promhttp"
    "golang.org/x/time/rate"
)

// Config holds service configuration
type Config struct {
    OllamaHost     string
    ListenAddr     string
    RateLimitRPS   int
    RateLimitBurst int
    ModelName      string
}

// OllamaService wraps Ollama client with production safeguards
type OllamaService struct {
    client      *api.Client
    config      Config
    rateLimiter *rate.Limiter
    metrics     *serviceMetrics
}

// serviceMetrics holds Prometheus metrics for the service
type serviceMetrics struct {
    inferenceRequests   prometheus.Counter
    inferenceLatency    prometheus.Histogram
    inferenceErrors     prometheus.Counter
    ollamaHealthStatus  prometheus.Gauge
}

// NewOllamaService initializes a new Ollama service instance
func NewOllamaService(cfg Config) (*OllamaService, error) {
    // Initialize Ollama client
    client, err := api.NewClient(cfg.OllamaHost, http.DefaultClient)
    if err != nil {
        return nil, fmt.Errorf("failed to create Ollama client: %w", err)
    }

    // Verify Ollama is running
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    if _, err := client.List(ctx); err != nil {
        return nil, fmt.Errorf("ollama not reachable at %s: %w", cfg.OllamaHost, err)
    }

    // Initialize rate limiter (token bucket)
    limiter := rate.NewLimiter(rate.Limit(cfg.RateLimitRPS), cfg.RateLimitBurst)

    // Initialize metrics
    metrics := &serviceMetrics{
        inferenceRequests: prometheus.NewCounter(prometheus.CounterOpts{
            Name: "ollama_inference_requests_total",
            Help: "Total number of inference requests",
        }),
        inferenceLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
            Name:    "ollama_inference_latency_ms",
            Help:    "Inference latency in milliseconds",
            Buckets: prometheus.DefBuckets,
        }),
        inferenceErrors: prometheus.NewCounter(prometheus.CounterOpts{
            Name: "ollama_inference_errors_total",
            Help: "Total number of inference errors",
        }),
        ollamaHealthStatus: prometheus.NewGauge(prometheus.GaugeOpts{
            Name: "ollama_health_status",
            Help: "Ollama daemon health status (1 = healthy, 0 = unhealthy)",
        }),
    }

    // Register metrics with Prometheus
    prometheus.MustRegister(metrics.inferenceRequests)
    prometheus.MustRegister(metrics.inferenceLatency)
    prometheus.MustRegister(metrics.inferenceErrors)
    prometheus.MustRegister(metrics.ollamaHealthStatus)

    return &OllamaService{
        client:      client,
        config:      cfg,
        rateLimiter: limiter,
        metrics:     metrics,
    }, nil
}

// HandleInference handles POST /inference requests
func (s *OllamaService) HandleInference(w http.ResponseWriter, r *http.Request) {
    // Apply rate limiting
    if !s.rateLimiter.Allow() {
        s.metrics.inferenceErrors.Inc()
        http.Error(w, "rate limit exceeded", http.StatusTooManyRequests)
        return
    }

    // Only accept POST
    if r.Method != http.MethodPost {
        s.metrics.inferenceErrors.Inc()
        http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
        return
    }

    // Parse request body
    var req struct {
        Prompt    string `json:"prompt"`
        MaxTokens int    `json:"max_tokens"`
    }
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        s.metrics.inferenceErrors.Inc()
        http.Error(w, "invalid request body", http.StatusBadRequest)
        return
    }
    defer r.Body.Close()

    // Validate request
    if req.Prompt == "" {
        s.metrics.inferenceErrors.Inc()
        http.Error(w, "prompt is required", http.StatusBadRequest)
        return
    }
    if req.MaxTokens <= 0 {
        req.MaxTokens = 512 // default
    }

    // Run inference
    start := time.Now()
    s.metrics.inferenceRequests.Inc()
    ctx := r.Context()
    resp, err := s.client.Generate(ctx, &api.GenerateRequest{
        Model:  s.config.ModelName,
        Prompt: req.Prompt,
        Options: map[string]interface{}{
            "num_predict": req.MaxTokens,
            "temperature": 0.7,
        },
    })
    if err != nil {
        s.metrics.inferenceErrors.Inc()
        log.Printf("inference failed: %v", err)
        http.Error(w, "inference failed", http.StatusInternalServerError)
        return
    }

    // Record latency
    latencyMs := time.Since(start).Milliseconds()
    s.metrics.inferenceLatency.Observe(float64(latencyMs))

    // Write response
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(map[string]interface{}{
        "response":       resp.Response,
        "latency_ms":     latencyMs,
        "eval_count":     resp.EvalCount,
        "tokens_per_sec": float64(resp.EvalCount) / (float64(resp.EvalDuration) / 1e9),
    })
}

// Start starts the HTTP server
func (s *OllamaService) Start() error {
    mux := http.NewServeMux()
    mux.HandleFunc("/inference", s.HandleInference)
    mux.Handle("/metrics", promhttp.Handler())
    mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
        ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
        defer cancel()
        if _, err := s.client.List(ctx); err != nil {
            s.metrics.ollamaHealthStatus.Set(0)
            http.Error(w, "ollama unhealthy", http.StatusServiceUnavailable)
            return
        }
        s.metrics.ollamaHealthStatus.Set(1)
        w.WriteHeader(http.StatusOK)
        w.Write([]byte("healthy"))
    })

    log.Printf("starting server on %s", s.config.ListenAddr)
    server := &http.Server{
        Addr:    s.config.ListenAddr,
        Handler: mux,
    }

    // Graceful shutdown
    go func() {
        sig := make(chan os.Signal, 1)
        signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
        <-sig
        log.Println("shutting down server...")
        ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
        defer cancel()
        server.Shutdown(ctx)
    }()

    return server.ListenAndServe()
}

func main() {
    // Load config from environment
    cfg := Config{
        OllamaHost:     getEnv("OLLAMA_HOST", "http://localhost:11434"),
        ListenAddr:     getEnv("LISTEN_ADDR", ":8080"),
        RateLimitRPS:   getEnvAsInt("RATE_LIMIT_RPS", 10),
        RateLimitBurst: getEnvAsInt("RATE_LIMIT_BURST", 20),
        ModelName:      getEnv("MODEL_NAME", "llama3.1:8b"),
    }

    service, err := NewOllamaService(cfg)
    if err != nil {
        log.Fatalf("failed to initialize service: %v", err)
    }

    if err := service.Start(); err != nil {
        log.Fatalf("server failed: %v", err)
    }
}

// getEnv returns environment variable value or default
func getEnv(key, defaultVal string) string {
    if val := os.Getenv(key); val != "" {
        return val
    }
    return defaultVal
}

// getEnvAsInt returns environment variable as int or default
func getEnvAsInt(key string, defaultVal int) int {
    val := os.Getenv(key)
    if val == "" {
        return defaultVal
    }
    var intVal int
    fmt.Sscanf(val, "%d", &intVal)
    return intVal
}

import { Ollama } from "ollama";
import * as fs from "fs/promises";
import * as path from "path";
import { parse } from "csv-parse/sync";
import { logger } from "./logger.js"; // Assume logger is configured elsewhere

// Configuration for fine-tuning
const CONFIG = {
  baseModel: "llama3.1:8b",
  adapterName: "llama3.1-8b-code-assistant",
  datasetPath: path.join(process.cwd(), "datasets", "code_tasks.csv"),
  validationSplit: 0.2,
  epochs: 3,
  learningRate: 1e-4,
  batchSize: 4,
  ollamaHost: process.env.OLLAMA_HOST || "http://localhost:11434",
} as const;

// Type definitions for dataset entries
type DatasetEntry = {
  prompt: string;
  response: string;
  task_type: "code_gen" | "debug" | "docstring" | "summarization";
};

type FineTuneResult = {
  status: "success" | "failed";
  adapter_name: string;
  epochs_completed: number;
  final_loss: number;
  validation_accuracy: number;
  error?: string;
};

/**
 * Load and validate the fine-tuning dataset from CSV
 */
async function loadDataset(): Promise<{ train: DatasetEntry[]; validation: DatasetEntry[] }> {
  try {
    const csvData = await fs.readFile(CONFIG.datasetPath, "utf-8");
    const entries: DatasetEntry[] = parse(csvData, {
      columns: true,
      skip_empty_lines: true,
    });

    // Validate entries
    const validEntries = entries.filter((entry) => {
      if (!entry.prompt || !entry.response || !entry.task_type) {
        logger.warn(`Skipping invalid entry: missing required fields`);
        return false;
      }
      if (!["code_gen", "debug", "docstring", "summarization"].includes(entry.task_type)) {
        logger.warn(`Skipping entry with invalid task type: ${entry.task_type}`);
        return false;
      }
      return true;
    });

    if (validEntries.length === 0) {
      throw new Error("No valid entries found in dataset");
    }

    // Split into train/validation
    const splitIndex = Math.floor(validEntries.length * (1 - CONFIG.validationSplit));
    const train = validEntries.slice(0, splitIndex);
    const validation = validEntries.slice(splitIndex);

    logger.info(`Loaded dataset: ${validEntries.length} total entries, ${train.length} train, ${validation.length} validation`);
    return { train, validation };
  } catch (error) {
    logger.error(`Failed to load dataset: ${error}`);
    throw error;
  }
}

/**
 * Format dataset entries into Ollama fine-tuning format
 */
function formatForOllama(entries: DatasetEntry[]): Array<{ input: string; output: string }> {
  return entries.map((entry) => ({
    input: `### Task: ${entry.task_type}\n### Prompt:\n${entry.prompt}\n### Response:`,
    output: entry.response,
  }));
}

/**
 * Run fine-tuning using Ollama 0.6's fine-tune API
 */
async function runFineTuning(
  ollama: Ollama,
  trainData: Array<{ input: string; output: string }>,
  validationData: Array<{ input: string; output: string }>
): Promise {
  try {
    logger.info(`Starting fine-tuning for ${CONFIG.baseModel}...`);

    // Check if base model exists
    const models = await ollama.list();
    const modelExists = models.models.some((m) => m.name === CONFIG.baseModel);
    if (!modelExists) {
      logger.info(`Base model ${CONFIG.baseModel} not found, pulling...`);
      await ollama.pull({ model: CONFIG.baseModel });
    }

    // Create fine-tune job
    const fineTuneResponse = await ollama.fineTune({
      model: CONFIG.baseModel,
      adapter: CONFIG.adapterName,
      data: {
        train: trainData,
        validation: validationData,
      },
      options: {
        epochs: CONFIG.epochs,
        learning_rate: CONFIG.learningRate,
        batch_size: CONFIG.batchSize,
      },
    });

    // Poll for job completion
    let jobStatus = await ollama.fineTuneStatus({ job_id: fineTuneResponse.job_id });
    while (jobStatus.status !== "completed" && jobStatus.status !== "failed") {
      logger.info(`Fine-tune job ${fineTuneResponse.job_id} status: ${jobStatus.status}`);
      await new Promise((resolve) => setTimeout(resolve, 5000)); // Poll every 5s
      jobStatus = await ollama.fineTuneStatus({ job_id: fineTuneResponse.job_id });
    }

    if (jobStatus.status === "failed") {
      return {
        status: "failed",
        adapter_name: CONFIG.adapterName,
        epochs_completed: jobStatus.epochs_completed || 0,
        final_loss: jobStatus.final_loss || 0,
        validation_accuracy: jobStatus.validation_accuracy || 0,
        error: jobStatus.error || "Unknown error",
      };
    }

    logger.info(`Fine-tuning completed successfully. Adapter: ${CONFIG.adapterName}`);
    return {
      status: "success",
      adapter_name: CONFIG.adapterName,
      epochs_completed: jobStatus.epochs_completed || CONFIG.epochs,
      final_loss: jobStatus.final_loss || 0,
      validation_accuracy: jobStatus.validation_accuracy || 0,
    };
  } catch (error) {
    logger.error(`Fine-tuning failed: ${error}`);
    return {
      status: "failed",
      adapter_name: CONFIG.adapterName,
      epochs_completed: 0,
      final_loss: 0,
      validation_accuracy: 0,
      error: error instanceof Error ? error.message : String(error),
    };
  }
}

/**
 * Validate fine-tuned adapter on sample prompts
 */
async function validateAdapter(ollama: Ollama): Promise {
  const testPrompts = [
    "Write a TypeScript function to deep clone an object",
    "Debug this code: const arr = [1,2,3]; arr.map((item) => { console.log(item); })",
    "Generate a docstring for: function parseJson(str: string): unknown",
  ];

  logger.info(`Validating adapter ${CONFIG.adapterName}...`);
  for (const prompt of testPrompts) {
    try {
      const response = await ollama.generate({
        model: `${CONFIG.baseModel}-${CONFIG.adapterName}`,
        prompt: `### Task: code_gen\n### Prompt:\n${prompt}\n### Response:`,
        options: { temperature: 0.7 },
      });
      logger.info(`Prompt: ${prompt}\nResponse: ${response.response}\n`);
    } catch (error) {
      logger.error(`Validation failed for prompt: ${prompt}. Error: ${error}`);
    }
  }
}

async function main() {
  const ollama = new Ollama({ host: CONFIG.ollamaHost });

  // Verify Ollama is running
  try {
    await ollama.list();
    logger.info(`Connected to Ollama at ${CONFIG.ollamaHost}`);
  } catch (error) {
    logger.error(`Ollama not running at ${CONFIG.ollamaHost}. Install from https://github.com/ollama/ollama/releases/tag/v0.6.0`);
    process.exit(1);
  }

  // Load dataset
  const { train, validation } = await loadDataset();

  // Format data for Ollama
  const trainFormatted = formatForOllama(train);
  const validationFormatted = formatForOllama(validation);

  // Run fine-tuning
  const result = await runFineTuning(ollama, trainFormatted, validationFormatted);

  if (result.status === "success") {
    logger.info(`Fine-tuning complete. Final loss: ${result.final_loss}, Validation accuracy: ${result.validation_accuracy}`);
    await validateAdapter(ollama);
  } else {
    logger.error(`Fine-tuning failed: ${result.error}`);
    process.exit(1);
  }
}

main();