pretrained.rwkv

Defines a simple API for using the RWKV model.

This code is adapted from the minimimal implementation here, adapted to be fine-tunable.

from rwkv.model import pretrained_rwkv

model = pretrained_rwkv("7B")
predictor = model.predictor()

for token in predictor.generate("The quick brown fox jumped over the"):
    print(token)

Using the tokenizer requires installing the tokenizers library:

pip install tokenizers

Additionally, using the training mode CUDA kernel requires installing triton:

pip install triton

The choices for the model key are:

• "169m"

• "430m"

• "1.5b"

• "3b"

• "7b"

• "14b"

pretrained.rwkv.cast_pretrained_rwkv_key(s: str) → Literal['169m', '430m', '1.5b', '3b', '7b', '14b']

class pretrained.rwkv.ModelArgs(url: str, sha256: str, emb_dim: int, num_layers: int)

Bases: object

url: str

sha256: str

emb_dim: int

num_layers: int

pretrained.rwkv.supports_triton() → bool

class pretrained.rwkv.WkvWithEps(*args, **kwargs)

Bases: Function

static forward(ctx: FunctionCtx, w: Tensor, u: Tensor, k: Tensor, v: Tensor, state: Tensor) → tuple[torch.Tensor, torch.Tensor]

This function is to be overridden by all subclasses. There are two ways to define forward:

Usage 1 (Combined forward and ctx):

@staticmethod
def forward(ctx: Any, *args: Any, **kwargs: Any) -> Any:
    pass

• It must accept a context ctx as the first argument, followed by any number of arguments (tensors or other types).

• See combining-forward-context for more details

Usage 2 (Separate forward and ctx):

@staticmethod
def forward(*args: Any, **kwargs: Any) -> Any:
    pass

@staticmethod
def setup_context(ctx: Any, inputs: Tuple[Any, ...], output: Any) -> None:
    pass

• The forward no longer accepts a ctx argument.

• Instead, you must also override the torch.autograd.Function.setup_context() staticmethod to handle setting up the ctx object. output is the output of the forward, inputs are a Tuple of inputs to the forward.

• See extending-autograd for more details

The context can be used to store arbitrary data that can be then retrieved during the backward pass. Tensors should not be stored directly on ctx (though this is not currently enforced for backward compatibility). Instead, tensors should be saved either with ctx.save_for_backward() if they are intended to be used in backward (equivalently, vjp) or ctx.save_for_forward() if they are intended to be used for in jvp.

static backward(ctx: FunctionCtx, grad_wkv: Tensor, grad_state: Tensor) → tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]

Defines a formula for differentiating the operation with backward mode automatic differentiation (alias to the vjp function).

This function is to be overridden by all subclasses.

It must accept a context ctx as the first argument, followed by as many outputs as the forward() returned (None will be passed in for non tensor outputs of the forward function), and it should return as many tensors, as there were inputs to forward(). Each argument is the gradient w.r.t the given output, and each returned value should be the gradient w.r.t. the corresponding input. If an input is not a Tensor or is a Tensor not requiring grads, you can just pass None as a gradient for that input.

The context can be used to retrieve tensors saved during the forward pass. It also has an attribute ctx.needs_input_grad as a tuple of booleans representing whether each input needs gradient. E.g., backward() will have ctx.needs_input_grad[0] = True if the first input to forward() needs gradient computed w.r.t. the output.

pretrained.rwkv.initial_state_with_eps(emb_dim: int) → Tensor

pretrained.rwkv.wkv_with_eps(w: Tensor, u: Tensor, k: Tensor, v: Tensor, state: Tensor) → tuple[torch.Tensor, torch.Tensor]

Runs the core WKV computation.

Parameters:

• w – The decay tensor, with shape (D)

• u – The output multiplier tensor, with shape (D)

• k – The K tensor, with shape (B, T, D)

• v – The V tensor, with shape (B, T, D)

• state – The state tensor, with shape (B, 3, T, D), consisting of the alpha, beta and eps tensors, each with shape (B, 1, T, D)

Returns:

The WKV tensor, with shape (B, T, D), and the next state, with shape (B, 3, 1, D), consisting of the next alpha, beta and eps tensors, each with shape (B, 1, 1, D)

class pretrained.rwkv.WkvLogSpace(*args, **kwargs)

Bases: Function

static forward(ctx: FunctionCtx, w: Tensor, u: Tensor, k: Tensor, v: Tensor, state: Tensor) → tuple[torch.Tensor, torch.Tensor]

This function is to be overridden by all subclasses. There are two ways to define forward:

Usage 1 (Combined forward and ctx):

@staticmethod
def forward(ctx: Any, *args: Any, **kwargs: Any) -> Any:
    pass

• It must accept a context ctx as the first argument, followed by any number of arguments (tensors or other types).

• See combining-forward-context for more details

Usage 2 (Separate forward and ctx):

@staticmethod
def forward(*args: Any, **kwargs: Any) -> Any:
    pass

@staticmethod
def setup_context(ctx: Any, inputs: Tuple[Any, ...], output: Any) -> None:
    pass

• The forward no longer accepts a ctx argument.

• Instead, you must also override the torch.autograd.Function.setup_context() staticmethod to handle setting up the ctx object. output is the output of the forward, inputs are a Tuple of inputs to the forward.

• See extending-autograd for more details

The context can be used to store arbitrary data that can be then retrieved during the backward pass. Tensors should not be stored directly on ctx (though this is not currently enforced for backward compatibility). Instead, tensors should be saved either with ctx.save_for_backward() if they are intended to be used in backward (equivalently, vjp) or ctx.save_for_forward() if they are intended to be used for in jvp.

static backward(ctx: FunctionCtx, grad_wkv: Tensor, grad_state: Tensor) → tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]

Defines a formula for differentiating the operation with backward mode automatic differentiation (alias to the vjp function).

This function is to be overridden by all subclasses.

It must accept a context ctx as the first argument, followed by as many outputs as the forward() returned (None will be passed in for non tensor outputs of the forward function), and it should return as many tensors, as there were inputs to forward(). Each argument is the gradient w.r.t the given output, and each returned value should be the gradient w.r.t. the corresponding input. If an input is not a Tensor or is a Tensor not requiring grads, you can just pass None as a gradient for that input.

The context can be used to retrieve tensors saved during the forward pass. It also has an attribute ctx.needs_input_grad as a tuple of booleans representing whether each input needs gradient. E.g., backward() will have ctx.needs_input_grad[0] = True if the first input to forward() needs gradient computed w.r.t. the output.

pretrained.rwkv.initial_state_log_space(emb_dim: int) → Tensor

pretrained.rwkv.wkv_log_space(w: Tensor, u: Tensor, k: Tensor, v: Tensor, state: Tensor) → tuple[torch.Tensor, torch.Tensor]

Runs the core WKV computation.

Parameters:

• w – The decay tensor, with shape (D)

• u – The output multiplier tensor, with shape (D)

• k – The K tensor, with shape (B, T, D)

• v – The V tensor, with shape (B, T, D)

• state – The state tensor, with shape (B, 3, D), consisting of the alpha plus, alpha minus and beta tensors, each with shape (B, 1, D)

Returns:

The WKV tensor, with shape (B, T, D), and the next state, with shape (B, 2, D), consisting of the next alpha plus, alpha minus and beta tensors, each with shape (B, 1, D)

pretrained.rwkv.get_wkv_fn(key: Literal['eps', 'log']) → Callable[[Tensor, Tensor, Tensor, Tensor, Tensor], tuple[torch.Tensor, torch.Tensor]]

pretrained.rwkv.get_wkv_fn_cuda(key: Literal['eps', 'log']) → Callable[[Tensor, Tensor, Tensor, Tensor, Tensor], tuple[torch.Tensor, torch.Tensor]]

pretrained.rwkv.get_default_wkv_fn_key() → Literal['eps', 'log']

class pretrained.rwkv.Attention(dim: int, lora_rank: int | None = None, lora_alpha: float = 1.0, lora_dropout: float = 0.0, freeze: bool = False, wkv_key: Literal['eps', 'log'] | None = None)

Bases: Module

Initializes internal Module state, shared by both nn.Module and ScriptModule.

init_x: Tensor

init_state: Tensor

time_shift(last_x: Tensor, x: Tensor) → Tensor

forward(x: Tensor, state: tuple[torch.Tensor, torch.Tensor] | None) → tuple[torch.Tensor, tuple[torch.Tensor, torch.Tensor]]

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class pretrained.rwkv.FeedForward(dim: int, ffn_dim: int, lora_rank: int | None = None, lora_alpha: float = 1.0, lora_dropout: float = 0.0, freeze: bool = False)

Bases: Module

Initializes internal Module state, shared by both nn.Module and ScriptModule.

init_state: Tensor

time_shift(last_x: Tensor, x: Tensor) → Tensor

forward(x: Tensor, state: Tensor | None = None) → tuple[torch.Tensor, torch.Tensor]

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class pretrained.rwkv.Block(emb_dim: int, pre_norm: bool, lora_rank: int | None = None, lora_alpha: float = 1.0, lora_dropout: float = 0.0, lora_attn: bool = True, lora_ffn: bool = True, freeze_layer_norm: bool = False, freeze_attn: bool = False, freeze_ffn: bool = False, use_checkpointing: bool = False, wkv_key: Literal['eps', 'log'] | None = None)

Bases: Module

Initializes internal Module state, shared by both nn.Module and ScriptModule.

run_attn(x: Tensor, state: tuple[tuple[torch.Tensor, torch.Tensor], torch.Tensor] | None = None) → tuple[torch.Tensor, tuple[torch.Tensor, torch.Tensor]]

run_ffn(x: Tensor, state: tuple[tuple[torch.Tensor, torch.Tensor], torch.Tensor] | None = None) → tuple[torch.Tensor, torch.Tensor]

forward(x: Tensor, state: tuple[tuple[torch.Tensor, torch.Tensor], torch.Tensor] | None = None) → tuple[torch.Tensor, tuple[tuple[torch.Tensor, torch.Tensor], torch.Tensor]]

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class pretrained.rwkv.RwkvStack(emb_dim: int, num_layers: int, use_checkpointing: bool = False, wkv_key: Literal['eps', 'log'] | None = None)

Bases: Module

Defines a stack of RWKV modules.

Parameters:

• emb_dim – The number of embedding dimensions in each block

• num_layers – The number of layers in the stack

• use_checkpointing – Whether to use checkpointing

• wkv_key – The WKV algorithm to use

Inputs:

x: The input tensor, with shape (B, T, D) state: The previous state

Outputs:

The output tensor, with shape (B, T, D), and the next state

Initializes internal Module state, shared by both nn.Module and ScriptModule.

forward(x: Tensor, state: list[tuple[tuple[torch.Tensor, torch.Tensor], torch.Tensor]] | None = None) → tuple[torch.Tensor, list[tuple[tuple[torch.Tensor, torch.Tensor], torch.Tensor]]]

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

class pretrained.rwkv.Rwkv(emb_dim: int, num_tokens: int, num_layers: int, lora_rank: int | None = None, lora_alpha: float = 1.0, lora_dropout: float = 0.0, lora_embeddings: bool = True, lora_linear: bool = True, lora_top_k_blocks: int | None = None, lora_attn: bool = True, lora_ffn: bool = True, freeze_non_lora: bool = False, freeze_layer_norm: bool | None = None, freeze_attn: bool | None = None, freeze_ffn: bool | None = None, use_checkpointing: bool = False, wkv_key: Literal['eps', 'log'] | None = None)

Bases: Module

Initializes internal Module state, shared by both nn.Module and ScriptModule.

tensor_to(x: Tensor) → Tensor

forward(tokens: Tensor, states_in: list[tuple[tuple[torch.Tensor, torch.Tensor], torch.Tensor]] | None = None, return_logits: bool = False) → tuple[torch.Tensor, list[tuple[tuple[torch.Tensor, torch.Tensor], torch.Tensor]]]

Defines the computation performed at every call.

Should be overridden by all subclasses.

Note

Although the recipe for forward pass needs to be defined within this function, one should call the Module instance afterwards instead of this since the former takes care of running the registered hooks while the latter silently ignores them.

predictor() → RwkvPredictor

pretrained.rwkv.get_tokenizer() → Any

class pretrained.rwkv.RwkvPredictor(rwkv_model: Rwkv)

Bases: object

Provides an API for sampling from the RWKV model.

Parameters:

rwkv_model – The RWKV model to use for sampling.

sample_probs(probs: Tensor, temperature: float = 1.0, top_p: float = 0.85) → Tensor

generate(prompt: str | Tensor, max_len: int = 256, temperature: float = 1.0, top_p: float = 0.85, end_toks: Sequence[int] | None = None, end_strs: Sequence[str] | None = None) → Iterator[str]

pretrained.rwkv.pretrained_rwkv(key: Literal['169m', '430m', '1.5b', '3b', '7b', '14b'], *, device: base_device | None = None, lora_rank: int | None = None, lora_alpha: float = 1.0, lora_dropout: float = 0.0, lora_embeddings: bool = True, lora_linear: bool = True, lora_top_k_blocks: int | None = None, lora_attn: bool = True, lora_ffn: bool = True, freeze_non_lora: bool = False, freeze_layer_norm: bool | None = None, freeze_attn: bool | None = None, freeze_ffn: bool | None = None, use_checkpointing: bool = False, empty: bool = False, wkv_key: Literal['eps', 'log'] | None = None) → Rwkv

Returns a pretrained RWKV model.

Parameters:

• key – The key of the pretrained model to load.

• device – The device to load the model onto. If None, the model will be loaded onto the device returned by detect_device().

• lora_rank – The rank of the LoRA decomposition to use.

• lora_alpha – The alpha parameter of the LoRA decomposition.

• lora_dropout – The dropout rate to use in the LoRA decomposition.

• lora_embeddings – Whether to use LoRA for the embedding matrices.

• lora_linear – Whether to use LoRA for the linear layers.

• lora_top_k_blocks – The number of top-k blocks to use in the LoRA decomposition.

• lora_attn – Whether to use LoRA for the attention layers.

• lora_ffn – Whether to use LoRA for the feed-forward layers.

• freeze_non_lora – Whether to freeze the non-LoRA parameters. This value will override the other freeze parameters if they are None.

• freeze_layer_norm – Whether to freeze the layer normalization parameters.

• freeze_attn – Whether to freeze the attention parameters.

• freeze_ffn – Whether to freeze the feed-forward parameters.

• use_checkpointing – Whether to use checkpointing to reduce memory usage.

• empty – Whether to return an empty model with the same structure as the pretrained model.

• wkv_key – The choice of WKV function to use. They are mathematically equivalent, but with different behaviors regarding numerical stability.

Returns:

The pretrained RWKV model.

pretrained.rwkv.test_rwkv_adhoc() → None