Kaggle API Reference¶

from llamatelemetry.kaggle import (
    KaggleEnvironment, quick_setup,
    ServerPreset, TensorSplitMode, PresetConfig, get_preset_config,
    KaggleSecrets, auto_load_secrets, setup_huggingface_auth, setup_graphistry_auth,
    GPUContext, rapids_gpu, llm_gpu, single_gpu, split_gpu_session,
    KagglePipelineConfig, load_grafana_otlp_env_from_kaggle,
    start_server_from_preset, setup_otel_and_client,
)

class KaggleEnvironment:
    gpu_count: int
    gpu_names: List[str]
    total_vram_gb: float
    vram_per_gpu_gb: List[float]
    compute_capability: str
    cuda_version: str
    hf_token: Optional[str]
    graphistry_key_id: Optional[str]
    graphistry_key_secret: Optional[str]
    preset: Optional[ServerPreset]
    telemetry_enabled: bool
    graphistry_enabled: bool
    rapids_gpu_id: int
    llm_gpu_ids: List[int]

@classmethod
def setup(
    cls,
    enable_telemetry: bool = True,
    enable_graphistry: bool = False,
    auto_load_secrets: bool = True,
    split_gpu_mode: bool = True,
    verbose: bool = True,
) -> KaggleEnvironment

env = KaggleEnvironment.setup()
print(f"GPUs: {env.gpu_count}x {env.gpu_names[0]}")
print(f"Total VRAM: {env.total_vram_gb:.1f} GB")
print(f"Preset: {env.preset.name}")

def create_engine(
    self,
    model_name_or_path: str,
    preset: Optional[ServerPreset] = None,
    auto_start: bool = True,
    verbose: bool = True,
    **kwargs,
) -> InferenceEngine

env = KaggleEnvironment.setup()
engine = env.create_engine("gemma-3-4b-Q4_K_M")
result = engine.infer("Hello, world!")
print(result.text)

def rapids_context(self) -> GPUContext

with env.rapids_context():
    import cudf, cugraph
    df = cudf.DataFrame({"x": [1, 2, 3]})
    # All RAPIDS ops on GPU 1

def llm_context(self) -> GPUContext

def download_model(
    self,
    repo_id: str,
    filename: str,
    local_dir: Optional[Path] = None,
) -> Path

def quick_setup(**kwargs) -> KaggleEnvironment

class ServerPreset(Enum):
    AUTO             # Auto-detect optimal settings
    KAGGLE_DUAL_T4   # 2x T4, 30GB total, split-GPU
    KAGGLE_SINGLE_T4 # 1x T4, 15GB
    COLAB_T4         # Google Colab T4
    COLAB_A100       # Google Colab A100
    LOCAL_3090       # RTX 3090 24GB
    LOCAL_4090       # RTX 4090 24GB
    CPU_ONLY         # No GPU

class TensorSplitMode(Enum):
    NONE        = "none"       # Single GPU
    EQUAL       = "equal"      # Split equally across all GPUs
    BALANCED    = "balanced"   # Auto-balance based on VRAM
    CUSTOM      = "custom"     # User-defined split ratios
    DUAL_50_50  = "0.5,0.5"
    DUAL_60_40  = "0.6,0.4"
    DUAL_70_30  = "0.7,0.3"
    DUAL_40_60  = "0.4,0.6"
    DUAL_30_70  = "0.3,0.7"

def to_string(self) -> Optional[str]

def to_list(self) -> Optional[List[float]]

@dataclass
class PresetConfig:
    name: str
    server_url: str = "http://127.0.0.1:8080"
    port: int = 8080
    host: str = "127.0.0.1"
    gpu_layers: int = 99
    tensor_split: Optional[List[float]] = None
    split_mode: TensorSplitMode = TensorSplitMode.NONE
    main_gpu: int = 0
    ctx_size: int = 4096
    batch_size: int = 512
    ubatch_size: int = 128
    flash_attention: bool = True
    use_mmap: bool = True
    n_parallel: int = 1
    enable_metrics: bool = True
    enable_props: bool = True
    enable_slots: bool = True

def to_load_kwargs(self) -> Dict[str, Any]

def to_server_kwargs(self) -> Dict[str, Any]

def get_preset_config(preset: ServerPreset) -> PresetConfig

config = get_preset_config(ServerPreset.KAGGLE_DUAL_T4)
print(config.ctx_size)       # 8192
print(config.tensor_split)   # [0.5, 0.5]
print(config.n_parallel)     # 4

class KaggleSecrets:
    KNOWN_SECRETS: List[str]  # HF_TOKEN, OTLP_ENDPOINT, Graphistry keys, etc.

def get(self, name: str, default: Optional[str] = None) -> Optional[str]

def set_in_env(self, name: str) -> bool

def list_available(self) -> List[str]

secrets = KaggleSecrets()
hf_token = secrets.get("HF_TOKEN")
print(secrets.list_available())

def auto_load_secrets(
    set_env: bool = True,
    secrets_to_load: Optional[List[str]] = None,
) -> Dict[str, Optional[str]]

def setup_huggingface_auth() -> bool

def setup_graphistry_auth() -> bool

class GPUContext:
    gpu_ids: Optional[List[int]]
    restore_on_exit: bool
    visible_devices: str       # property
    is_active: bool            # property

with GPUContext(gpu_ids=[1]) as ctx:
    import cudf
    df = cudf.DataFrame({"a": [1, 2, 3]})
    print(ctx.visible_devices)  # "1"
# CUDA_VISIBLE_DEVICES restored automatically

def set_devices(self, gpu_ids: List[int]) -> None

@contextmanager
def rapids_gpu(gpu_id: int = 1) -> GPUContext

@contextmanager
def llm_gpu(gpu_ids: Optional[List[int]] = None) -> GPUContext

@contextmanager
def single_gpu(gpu_id: int = 0) -> GPUContext

@contextmanager
def split_gpu_session(llm_gpu: int = 0, graph_gpu: int = 1) -> Dict[str, Any]

with split_gpu_session(llm_gpu=0, graph_gpu=1) as session:
    server_kwargs = session["llm_server_kwargs"]
    # {"main_gpu": 0, "tensor_split": "1.0,0.0"}

def get_current_gpu_context() -> Optional[List[int]]

def set_gpu_for_rapids(gpu_id: int = 1) -> str

def reset_gpu_context(value: Optional[str] = None) -> None

@dataclass
class KagglePipelineConfig:
    service_name: str = "llamatelemetry"
    service_version: str = "0.1.1"
    otlp_endpoint: Optional[str] = None
    enable_graphistry: bool = False
    graphistry_server: Optional[str] = None
    enable_llama_metrics: bool = True
    llama_metrics_interval: float = 5.0

def load_grafana_otlp_env_from_kaggle(
    *,
    endpoint_secret: str = "GRAFANA_OTLP_ENDPOINT",
    headers_secret: str = "GRAFANA_OTLP_HEADERS",
    token_secret: str = "GRAFANA_OTLP_TOKEN",
) -> Dict[str, str]

def start_server_from_preset(
    model_path: str,
    preset: ServerPreset,
    *,
    extra_args: Optional[Dict[str, Any]] = None,
) -> ServerManager

mgr = start_server_from_preset("/path/to/model.gguf", ServerPreset.KAGGLE_DUAL_T4)
# Server is running and ready at http://127.0.0.1:8080

def setup_otel_and_client(
    base_url: str,
    cfg: KagglePipelineConfig,
) -> Dict[str, Any]

cfg = KagglePipelineConfig(otlp_endpoint="https://otlp.grafana.net/otlp")
resources = setup_otel_and_client("http://127.0.0.1:8080", cfg)
client = resources["client"]
response = client.chat_completion(messages=[{"role": "user", "content": "Hello"}])