API Reference for inference¶

@abstractmethod
async def generate(
    self,
    prompt: str,
    model: str,
    adapter_id: str | None = None,
    max_tokens: int = 4096,
    system_prompt: str | None = None,
    temperature: float | None = None,
    top_p: float | None = None,
    repetition_penalty: float | None = None,
) -> GenerationResult:
    """Generate text from a prompt.

    Args:
        prompt: The user-facing input prompt.
        model: The model identifier to use for generation.
        adapter_id: Optional LoRA adapter to apply during generation.
            If None, uses the base model directly.
        max_tokens: Maximum number of tokens to generate.
        system_prompt: Optional system-level instruction. Providers that
            support chat templates will format this as a system message.

        temperature: Sampling temperature override.
        top_p: Nucleus sampling threshold override.
        repetition_penalty: Repetition penalty override.

    Returns:
        A GenerationResult containing the generated text and metadata.

    Example:
        >>> result = await provider.generate("def hello", model="Qwen2.5-Coder-7B")
        >>> print(result.text)
    """
    ...

@abstractmethod
async def load_adapter(self, adapter_id: str, adapter_path: str) -> None:
    """Load a LoRA adapter into the inference server.

    Args:
        adapter_id: Unique name for the adapter (used as lora_name in vLLM).
        adapter_path: Filesystem path to the adapter weights directory.

    Raises:
        UnsupportedOperationError: If the provider does not support adapters.
        httpx.HTTPStatusError: If the server returns an error response.

    Example:
        >>> await provider.load_adapter("adapter-001", "/models/adapter-001")
    """
    ...

@abstractmethod
async def unload_adapter(self, adapter_id: str) -> None:
    """Unload a previously loaded LoRA adapter.

    Args:
        adapter_id: The adapter name to remove from the server.

    Raises:
        UnsupportedOperationError: If the provider does not support adapters.
        httpx.HTTPStatusError: If the server returns an error response.

    Example:
        >>> await provider.unload_adapter("adapter-001")
    """
    ...

@abstractmethod
async def list_adapters(self) -> list[str]:
    """List all currently loaded LoRA adapters.

    Returns:
        Sorted list of adapter IDs currently available for inference.
        Returns an empty list if no adapters are loaded or the provider
        does not support adapters.

    Example:
        >>> adapters = await provider.list_adapters()
        >>> print(adapters)  # ["adapter-001", "adapter-002"]
    """
    ...

async def generate(
    self,
    prompt: str,
    model: str,
    adapter_id: str | None = None,
    max_tokens: int = 4096,
    system_prompt: str | None = None,
    temperature: float | None = None,
    top_p: float | None = None,
    repetition_penalty: float | None = None,
) -> GenerationResult:
    """Generate text using llama-cpp-python with optional LoRA adapter.

    Args:
        prompt: The user-facing input prompt.
        model: Ignored (model is set at construction via model_path).
        adapter_id: LoRA adapter ID to apply. Must be loaded via
            load_adapter() before use.
        max_tokens: Maximum tokens to generate.
        system_prompt: Optional system-level instruction. Prepended to
            the prompt for llama.cpp (no native chat template support).
        temperature: Sampling temperature override.
        top_p: Nucleus sampling threshold override.
        repetition_penalty: Repetition penalty override.

    Returns:
        GenerationResult with generated text and metadata.

    Raises:
        ValueError: If adapter_id is provided but has not been loaded.
    """
    lora_path: str | None = None
    if adapter_id:
        if adapter_id not in self._loaded_adapters:
            raise ValueError(
                f"Adapter '{adapter_id}' has not been loaded. "
                "Call load_adapter() first."
            )
        lora_path = self._loaded_adapters[adapter_id]

    self._load_model_if_needed(lora_path=lora_path)

    full_prompt = f"{system_prompt}\n\n{prompt}" if system_prompt else prompt
    response = self._llm(  # type: ignore[union-attr]
        full_prompt,
        max_tokens=max_tokens,
        stop=_STOP_SEQUENCES,
    )

    text = response["choices"][0]["text"]
    token_count = response["usage"]["total_tokens"]
    finish_reason = response["choices"][0].get("finish_reason", "stop")

    return GenerationResult(
        text=text,
        model=Path(self._model_path).stem,
        adapter_id=adapter_id,
        token_count=token_count,
        finish_reason=finish_reason,
    )

async def load_adapter(self, adapter_id: str, adapter_path: str) -> None:
    """Register a LoRA adapter for use during generation.

    The adapter is applied on next generate() call by reloading the model
    with the LoRA path. llama-cpp-python applies LoRA at model load time.

    Args:
        adapter_id: Unique name for the adapter.
        adapter_path: Filesystem path to the LoRA adapter weights (GGUF format).
    """
    self._loaded_adapters[adapter_id] = adapter_path
    logger.info("Registered adapter %s -> %s", adapter_id, adapter_path)

async def unload_adapter(self, adapter_id: str) -> None:
    """Remove a registered LoRA adapter.

    If the currently active adapter is unloaded, the model will reload
    without it on the next generate() call.

    Args:
        adapter_id: The adapter name to remove.
    """
    if adapter_id in self._loaded_adapters:
        was_active = self._current_lora == self._loaded_adapters[adapter_id]
        del self._loaded_adapters[adapter_id]
        if was_active:
            self._current_lora = None  # Force reload without adapter
        logger.info("Unloaded adapter %s", adapter_id)

async def list_adapters(self) -> list[str]:
    """List all registered LoRA adapter IDs.

    Returns:
        Sorted list of registered adapter IDs.
    """
    return sorted(self._loaded_adapters.keys())

async def generate(
    self,
    prompt: str,
    model: str,
    adapter_id: str | None = None,
    max_tokens: int = 4096,
    system_prompt: str | None = None,
    temperature: float | None = None,
    top_p: float | None = None,
    repetition_penalty: float | None = None,
) -> GenerationResult:
    """Generate text from a prompt using the base Ollama model.

    If adapter_id is provided, a warning is logged and it is ignored —
    Ollama does not support LoRA adapters. The base model is always used.

    Args:
        prompt: The user-facing input prompt.
        model: The Ollama model identifier (e.g. "qwen2.5-coder:7b").
        adapter_id: Ignored. If provided, a warning is logged.
        max_tokens: Maximum number of tokens to generate.
        system_prompt: Optional system-level instruction.
        temperature: Sampling temperature override.
        top_p: Nucleus sampling threshold override.
        repetition_penalty: Repetition penalty override.

    Returns:
        GenerationResult with adapter_id=None (Ollama has no adapter concept).

    Example:
        >>> result = await provider.generate("def fib", model="qwen2.5-coder:7b")
        >>> print(result.text)
    """
    if adapter_id is not None:
        logger.warning(
            "OllamaProvider ignoring adapter_id=%s; "
            "Ollama does not support LoRA adapters.",
            adapter_id,
        )

    logger.debug("generate: model=%s max_tokens=%d", model, max_tokens)

    messages: list[dict[str, str]] = [{"role": "user", "content": prompt}]
    if system_prompt:
        messages.insert(0, {"role": "system", "content": system_prompt})

    response = await self._client.chat.completions.create(
        model=model,
        messages=messages,  # type: ignore[arg-type]
        max_tokens=max_tokens,
    )

    choice = response.choices[0]
    return GenerationResult(
        text=choice.message.content or "",
        model=response.model,
        adapter_id=None,
        token_count=response.usage.total_tokens if response.usage else 0,
        finish_reason=choice.finish_reason or "stop",
    )

async def load_adapter(self, adapter_id: str, adapter_path: str) -> None:
    """Not supported by Ollama. Always raises UnsupportedOperationError.

    Args:
        adapter_id: Unused.
        adapter_path: Unused.

    Raises:
        UnsupportedOperationError: Always — Ollama does not support
            LoRA adapter loading. Use VLLMProvider for adapter operations.

    Example:
        >>> await provider.load_adapter("adapter-001", "/models/adapter-001")
        # Raises UnsupportedOperationError
    """
    raise UnsupportedOperationError(
        "OllamaProvider does not support LoRA adapter loading. "
        "Use VLLMProvider for adapter operations."
    )

async def unload_adapter(self, adapter_id: str) -> None:
    """Not supported by Ollama. Always raises UnsupportedOperationError.

    Args:
        adapter_id: Unused.

    Raises:
        UnsupportedOperationError: Always — Ollama does not support
            LoRA adapter unloading.

    Example:
        >>> await provider.unload_adapter("adapter-001")
        # Raises UnsupportedOperationError
    """
    raise UnsupportedOperationError(
        "OllamaProvider does not support LoRA adapter unloading."
    )

async def list_adapters(self) -> list[str]:
    """Return an empty list — Ollama has no adapter concept.

    Returns:
        Always returns an empty list.

    Example:
        >>> adapters = await provider.list_adapters()
        >>> print(adapters)  # []
    """
    return []

async def generate(
    self,
    prompt: str,
    model: str,
    adapter_id: str | None = None,
    max_tokens: int = 4096,
    system_prompt: str | None = None,
    temperature: float | None = None,
    top_p: float | None = None,
    repetition_penalty: float | None = None,
) -> GenerationResult:
    """Generate text using transformers with optional PEFT adapter.

    Args:
        prompt: The user-facing input prompt.
        model: Ignored (model is set at construction).
        adapter_id: LoRA adapter ID to activate. Must be loaded via
            load_adapter() before use.
        max_tokens: Maximum tokens to generate.
        system_prompt: Optional system-level instruction prepended via
            the tokenizer's chat template when available.
        temperature: Sampling temperature (default from pipeline config).
        top_p: Nucleus sampling threshold (default from pipeline config).
        repetition_penalty: Repetition penalty (default 1.0 = off).

    Returns:
        GenerationResult with generated text and metadata.

    Raises:
        ValueError: If adapter_id is provided but has not been loaded.
    """
    import torch  # noqa: PLC0415

    self._load_model_if_needed()

    # Apply defaults from pipeline config
    if temperature is None:
        temperature = float(os.environ.get("RUNE_TEMPERATURE", "0.25"))
    if top_p is None:
        top_p = float(os.environ.get("RUNE_TOP_P", "0.9"))
    if repetition_penalty is None:
        repetition_penalty = float(
            os.environ.get("RUNE_REPETITION_PENALTY", "1.04")
        )

    # Validate adapter before switching
    if adapter_id and adapter_id not in self._loaded_adapters:
        raise ValueError(
            f"Adapter '{adapter_id}' has not been loaded. "
            "Call load_adapter() first."
        )

    # Switch adapter if needed
    if adapter_id and adapter_id != self._active_adapter:
        self._activate_adapter(adapter_id)
    elif not adapter_id and self._active_adapter:
        self._deactivate_adapter()

    # Build chat-formatted prompt via tokenizer's chat template
    formatted = self._format_prompt(prompt, system_prompt)
    inputs = self._tokenizer(
        formatted, return_tensors="pt", truncation=True, max_length=8192
    )
    inputs = {k: v.to(self._device) for k, v in inputs.items()}
    input_len = inputs["input_ids"].shape[1]

    gen_kwargs: dict[str, object] = {
        "max_new_tokens": max_tokens,
        "do_sample": temperature > 0,
        "temperature": max(temperature, 0.01),
        "top_p": top_p,
        "pad_token_id": self._tokenizer.pad_token_id,
    }
    if repetition_penalty > 1.0:
        gen_kwargs["repetition_penalty"] = repetition_penalty

    with torch.no_grad():
        outputs = self._model.generate(**inputs, **gen_kwargs)

    new_tokens = outputs[0][input_len:]
    text = self._tokenizer.decode(new_tokens, skip_special_tokens=True)
    total_tokens = outputs.shape[1]
    new_token_count = len(new_tokens)

    # Detect truncation: generated exactly max_tokens means cut off
    finish_reason = "length" if new_token_count >= max_tokens else "stop"

    return GenerationResult(
        text=text,
        model=self._model_name,
        adapter_id=self._active_adapter,
        token_count=total_tokens,
        finish_reason=finish_reason,
    )

async def load_adapter(self, adapter_id: str, adapter_path: str) -> None:
    """Register a PEFT adapter directory for use during generation.

    The adapter directory must contain adapter_model.safetensors and
    adapter_config.json as output by save_hypernetwork_adapter().

    Args:
        adapter_id: Unique name for the adapter.
        adapter_path: Path to the PEFT adapter directory.
    """
    self._loaded_adapters[adapter_id] = adapter_path
    logger.info("Registered adapter %s -> %s", adapter_id, adapter_path)

async def unload_adapter(self, adapter_id: str) -> None:
    """Remove a registered adapter, freeing GPU memory.

    Args:
        adapter_id: The adapter name to remove.
    """
    if adapter_id in self._loaded_adapters:
        if self._active_adapter == adapter_id:
            self._deactivate_adapter()
        # Delete from PeftModel to free GPU memory
        if self._is_peft_wrapped and adapter_id in self._model.peft_config:
            self._model.delete_adapter(adapter_id)
        del self._loaded_adapters[adapter_id]
        # If no adapters remain, revert to base model
        if not self._loaded_adapters and self._is_peft_wrapped:
            self._model = self._base_model
            self._is_peft_wrapped = False
            logger.info("All adapters removed, reverted to base model")
        else:
            logger.info("Unloaded adapter %s", adapter_id)

async def list_adapters(self) -> list[str]:
    """List all registered adapter IDs.

    Returns:
        Sorted list of registered adapter IDs.
    """
    return sorted(self._loaded_adapters.keys())

async def generate(
    self,
    prompt: str,
    model: str,
    adapter_id: str | None = None,
    max_tokens: int = 4096,
    system_prompt: str | None = None,
    temperature: float | None = None,
    top_p: float | None = None,
    repetition_penalty: float | None = None,
) -> GenerationResult:
    """Generate text from a prompt, optionally using a loaded LoRA adapter.

    When adapter_id is provided, it is passed as the model parameter to
    the OpenAI API — this is how vLLM identifies and routes to loaded
    LoRA adapters (the adapter is referenced by its lora_name).

    Args:
        prompt: The user-facing input prompt.
        model: Base model identifier. Used as-is when no adapter is given.
        adapter_id: Name of a loaded LoRA adapter to apply. When set,
            this value replaces model in the API call.
        max_tokens: Maximum number of tokens to generate.
        system_prompt: Optional system-level instruction.
        temperature: Sampling temperature override.
        top_p: Nucleus sampling threshold override.
        repetition_penalty: Repetition penalty override.

    Returns:
        GenerationResult with the generated text and metadata.

    Example:
        >>> result = await provider.generate("def fib", model="Qwen2.5-Coder-7B")
        >>> print(result.text)
    """
    effective_model = adapter_id if adapter_id is not None else model
    logger.debug(
        "generate: model=%s adapter_id=%s max_tokens=%d",
        effective_model,
        adapter_id,
        max_tokens,
    )

    messages: list[dict[str, str]] = [{"role": "user", "content": prompt}]
    if system_prompt:
        messages.insert(0, {"role": "system", "content": system_prompt})

    response = await self._client.chat.completions.create(
        model=effective_model,
        messages=messages,  # type: ignore[arg-type]
        max_tokens=max_tokens,
    )

    choice = response.choices[0]
    return GenerationResult(
        text=choice.message.content or "",
        model=response.model,
        adapter_id=adapter_id,
        token_count=response.usage.total_tokens if response.usage else 0,
        finish_reason=choice.finish_reason or "stop",
    )

async def load_adapter(self, adapter_id: str, adapter_path: str) -> None:
    """Load a LoRA adapter into the vLLM server.

    Posts to vLLM's /v1/load_lora_adapter endpoint and adds the adapter
    to the internal tracking set on success.

    Args:
        adapter_id: Unique name for the adapter (used as lora_name).
        adapter_path: Filesystem path to the adapter weights directory.

    Raises:
        httpx.HTTPStatusError: If the vLLM server returns an error response.

    Example:
        >>> await provider.load_adapter("adapter-001", "/models/adapter-001")
    """
    url = f"{self._base_url.rstrip('/')}/load_lora_adapter"
    logger.debug("load_adapter: POST %s lora_name=%s", url, adapter_id)

    async with httpx.AsyncClient() as client:
        response = await client.post(
            url,
            json={"lora_name": adapter_id, "lora_path": adapter_path},
        )
        response.raise_for_status()

    self._loaded_adapters.add(adapter_id)
    logger.info("Adapter loaded: %s", adapter_id)

async def unload_adapter(self, adapter_id: str) -> None:
    """Unload a LoRA adapter from the vLLM server.

    Posts to vLLM's /v1/unload_lora_adapter endpoint and removes the
    adapter from the internal tracking set.

    Args:
        adapter_id: Name of the adapter to unload.

    Raises:
        httpx.HTTPStatusError: If the vLLM server returns an error response.

    Example:
        >>> await provider.unload_adapter("adapter-001")
    """
    url = f"{self._base_url.rstrip('/')}/unload_lora_adapter"
    logger.debug("unload_adapter: POST %s lora_name=%s", url, adapter_id)

    async with httpx.AsyncClient() as client:
        response = await client.post(
            url,
            json={"lora_name": adapter_id},
        )
        response.raise_for_status()

    self._loaded_adapters.discard(adapter_id)
    logger.info("Adapter unloaded: %s", adapter_id)

async def list_adapters(self) -> list[str]:
    """List all currently loaded LoRA adapters.

    Returns the internal tracking set rather than querying vLLM to avoid
    the unreliable list endpoint (vLLM bug #11761).

    Returns:
        Sorted list of adapter IDs currently tracked as loaded.

    Example:
        >>> adapters = await provider.list_adapters()
        >>> print(adapters)  # ["adapter-001", "adapter-002"]
    """
    return sorted(self._loaded_adapters)