File size: 4,048 Bytes
954544e 2cb6966 954544e 2cb6966 954544e 2cb6966 954544e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | from __future__ import annotations
import argparse
from pathlib import Path
import torch
from transformers import AutoTokenizer
from dot_rd.config import ModelConfig
from dot_rd.export import load_exported_core, load_inference_checkpoint
from dot_rd.model import DotRecurrentDepthModel
def _token_ids(value: object) -> list[int]:
if hasattr(value, "input_ids"):
value = value.input_ids
if isinstance(value, torch.Tensor):
value = value.tolist()
if isinstance(value, list) and value and isinstance(value[0], list):
value = value[0]
if not isinstance(value, list) or not all(isinstance(token, int) for token in value):
raise TypeError("chat template did not return a token id list")
return value
@torch.inference_mode()
def generate(
model: DotRecurrentDepthModel,
tokenizer: object,
prompt: str,
*,
max_new_tokens: int,
max_context_tokens: int,
) -> str:
messages = [
{"role": "system", "content": "You are Dot, a local reasoning model."},
{"role": "user", "content": prompt},
]
prompt_ids = _token_ids(
tokenizer.apply_chat_template(
messages,
tokenize=True,
add_generation_prompt=True,
enable_thinking=True,
)
)[-max_context_tokens:]
input_ids = torch.tensor([prompt_ids], dtype=torch.long, device="cuda")
attention_mask = torch.ones_like(input_ids)
generated: list[int] = []
eos_token_id = tokenizer.eos_token_id
if eos_token_id is None:
raise ValueError("Dot tokenizer has no EOS token")
for _ in range(max_new_tokens):
output = model(
input_ids=input_ids,
attention_mask=attention_mask,
use_cache=False,
logits_to_keep=1,
)
next_token = int(output.logits[:, -1].argmax(dim=-1).item())
if next_token == eos_token_id:
break
generated.append(next_token)
input_ids = torch.cat(
(input_ids, torch.tensor([[next_token]], device=input_ids.device)), dim=1
)
attention_mask = torch.cat(
(attention_mask, torch.ones((1, 1), dtype=torch.long, device=input_ids.device)),
dim=1,
)
suffix = tokenizer.decode(generated, skip_special_tokens=True).strip()
return suffix if suffix.startswith("<think>") else f"<think>\n{suffix}"
def main() -> None:
parser = argparse.ArgumentParser(description="Run Dot v0.4 with greedy decoding")
parser.add_argument("--model", default=".", help="local Dot repository path")
parser.add_argument(
"--checkpoint",
help="optional Dot inference-checkpoint directory containing manifest.json",
)
parser.add_argument("--prompt", required=True)
parser.add_argument("--max-new-tokens", type=int, default=256)
parser.add_argument("--max-context-tokens", type=int, default=4096)
args = parser.parse_args()
if not torch.cuda.is_available():
raise RuntimeError("the verified Dot v0.4 runtime requires CUDA")
model_path = str(Path(args.model).resolve())
tokenizer = AutoTokenizer.from_pretrained(model_path)
config = ModelConfig(
base_model=model_path,
insertion_after=15,
source_layers=(12, 13, 14, 15),
max_loops=8,
active_loops=4,
initial_loop_scale=0.01,
attention_implementation="sdpa",
)
model = DotRecurrentDepthModel.from_pretrained(
config,
dtype=torch.bfloat16,
device_map=None,
).to("cuda").eval()
manifest = (
load_inference_checkpoint(args.checkpoint, model)
if args.checkpoint
else load_exported_core(model_path, model)
)
print(
generate(
model,
tokenizer,
args.prompt,
max_new_tokens=args.max_new_tokens,
max_context_tokens=args.max_context_tokens,
)
)
print(f"\n[Dot release step {manifest['source_step']}]")
if __name__ == "__main__":
main()
|