llama_cpp_standalone.py

#!/usr/bin/env python3
"""
llama.cpp Python Standalone - Simple wrapper for llama.cpp server binary.

Author: Gregor Koch (@cronos3k)
License: MIT

Use this when llama-cpp-python doesn't support your model architecture yet.
Provides OpenAI-compatible API via the official llama.cpp server.

Example:
    server = LlamaCppServer("/path/to/llama-server")
    server.start(model_path="model.gguf", port=8080)
    # Use with OpenAI client at http://localhost:8080/v1
    server.stop()
"""

import subprocess
import time
import signal
import atexit
from pathlib import Path
from typing import Optional, List
import requests


class LlamaCppServer:
    """
    Manages llama.cpp server binary lifecycle from Python.

    This wrapper launches the official llama.cpp server and provides
    a simple Python interface for starting, stopping, and health checking.
    """

    def __init__(self, binary_path: str = "llama-server"):
        """
        Initialize server manager.

        Args:
            binary_path: Path to llama-server binary. Can be absolute path
                        or command name if in PATH.
        """
        self.binary_path = binary_path
        self.process: Optional[subprocess.Popen] = None
        self.port: Optional[int] = None

        # Verify binary exists
        if not self._check_binary():
            raise FileNotFoundError(
                f"llama-server binary not found at: {binary_path}\n"
                "Build it with: scripts/build_llama_cpp.sh"
            )

        # Ensure cleanup on exit
        atexit.register(self.stop)

    def _check_binary(self) -> bool:
        """Check if llama-server binary exists and is executable."""
        try:
            result = subprocess.run(
                [self.binary_path, "--help"],
                capture_output=True,
                timeout=5
            )
            return result.returncode == 0
        except (FileNotFoundError, subprocess.TimeoutExpired):
            return False

    def start(
        self,
        model_path: str,
        port: int = 8080,
        host: str = "0.0.0.0",
        n_gpu_layers: int = -1,
        n_ctx: int = 4096,
        n_batch: int = 512,
        mmproj_path: Optional[str] = None,
        extra_args: Optional[List[str]] = None,
        timeout: int = 60
    ) -> bool:
        """
        Start llama.cpp server with specified configuration.

        Args:
            model_path: Path to GGUF model file
            port: Port to listen on (default: 8080)
            host: Host to bind to (default: 0.0.0.0)
            n_gpu_layers: GPU layers to offload. -1 = all (default: -1)
            n_ctx: Context window size (default: 4096)
            n_batch: Batch size for prompt processing (default: 512)
            mmproj_path: Path to multimodal projector for vision models
            extra_args: Additional arguments to pass to server
            timeout: Seconds to wait for server startup (default: 60)

        Returns:
            True if server started successfully

        Raises:
            FileNotFoundError: If model file doesn't exist
            RuntimeError: If server fails to start
            TimeoutError: If server doesn't respond within timeout
        """
        if self.process is not None:
            raise RuntimeError("Server already running. Call stop() first.")

        # Verify model exists
        if not Path(model_path).exists():
            raise FileNotFoundError(f"Model not found: {model_path}")

        if mmproj_path and not Path(mmproj_path).exists():
            raise FileNotFoundError(f"Multimodal projector not found: {mmproj_path}")

        # Build command
        cmd = [
            self.binary_path,
            "-m", model_path,
            "--port", str(port),
            "--host", host,
            "--n-gpu-layers", str(n_gpu_layers),
            "-c", str(n_ctx),
            "-b", str(n_batch),
        ]

        if mmproj_path:
            cmd.extend(["--mmproj", mmproj_path])

        if extra_args:
            cmd.extend(extra_args)

        print(f"Starting llama.cpp server...")
        print(f"Command: {' '.join(cmd)}")

        # Start process
        self.process = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1
        )
        self.port = port

        # Wait for server to be ready
        print(f"Waiting for server on port {port}...")
        start_time = time.time()

        while time.time() - start_time < timeout:
            # Check if process crashed
            if self.process.poll() is not None:
                stdout = self.process.stdout.read() if self.process.stdout else ""
                raise RuntimeError(
                    f"Server process exited with code {self.process.returncode}\n"
                    f"Output: {stdout}"
                )

            # Check health
            if self.check_health():
                print(f"✓ Server ready at http://{host}:{port}")
                print(f"✓ OpenAI-compatible API: http://{host}:{port}/v1")
                return True

            time.sleep(1)

        # Timeout - kill process
        self.stop()
        raise TimeoutError(f"Server did not start within {timeout} seconds")

    def stop(self) -> bool:
        """
        Stop the llama.cpp server gracefully.

        Returns:
            True if stopped successfully
        """
        if self.process is None:
            return True

        print("Stopping llama.cpp server...")

        try:
            # Try graceful shutdown first
            self.process.terminate()
            self.process.wait(timeout=10)
            print("✓ Server stopped")
        except subprocess.TimeoutExpired:
            # Force kill if needed
            print("Force killing server...")
            self.process.kill()
            self.process.wait()
            print("✓ Server killed")
        finally:
            self.process = None
            self.port = None

        return True

    def check_health(self) -> bool:
        """
        Check if server is responding.

        Returns:
            True if server is healthy
        """
        if self.port is None:
            return False

        try:
            response = requests.get(
                f"http://localhost:{self.port}/health",
                timeout=2
            )
            return response.status_code == 200
        except (requests.RequestException, Exception):
            return False

    def get_endpoint(self) -> str:
        """
        Get the OpenAI-compatible API endpoint URL.

        Returns:
            API endpoint URL (e.g., "http://localhost:8080/v1")

        Raises:
            RuntimeError: If server is not running
        """
        if self.port is None:
            raise RuntimeError("Server not running")

        return f"http://localhost:{self.port}/v1"

    def __enter__(self):
        """Context manager support."""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager cleanup."""
        self.stop()


def main():
    """Example usage."""
    import sys

    if len(sys.argv) < 2:
        print("Usage: python llama_cpp_standalone.py <model.gguf> [port]")
        sys.exit(1)

    model_path = sys.argv[1]
    port = int(sys.argv[2]) if len(sys.argv) > 2 else 8080

    # Example: Start server and keep it running
    server = LlamaCppServer()
    server.start(model_path=model_path, port=port)

    print("\nServer is running. Press Ctrl+C to stop...")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\nShutting down...")
        server.stop()


if __name__ == "__main__":
    main()