169 lines
8.4 KiB
Python
169 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
||
"""DEGENERATION TEST 1 v2 — Chat-template token-ID diff using official encoding.
|
||
|
||
Uses the official DeepSeek V4 encoding reference from encoding/encoding_dsv4.py
|
||
to build the canonical prompt, then diffs against our hand-rolled construction.
|
||
|
||
Official format (from DeepSeek-V4-Pro/encoding/README.md):
|
||
Thinking mode: <BOS>{system}<｜User｜>{msg}<｜Assistant｜>ately{reasoning}heroically{response}<EOS>
|
||
Chat mode: <BOS>{system}<｜User｜>{msg}<｜Assistant｜>heroically{response}<EOS>
|
||
|
||
Key differences from our hand-rolled:
|
||
1. No \n\n between User token and content
|
||
2. System prompt goes directly after BOS (no User token for system)
|
||
"""
|
||
import os, sys
|
||
|
||
CHECKPOINT_DIR = os.environ.get("CHECKPOINT_DIR", "/root/nvidia-meeting/DeepSeek-V4-Pro-NVFP4")
|
||
PROMPT = os.environ.get("TEST_PROMPT", "The capital of France is")
|
||
|
||
THINK_START, THINK_END = 128821, 128822
|
||
USER_TOKEN, ASSISTANT_TOKEN = 128803, 128804
|
||
|
||
def main():
|
||
from transformers import AutoTokenizer
|
||
print("=" * 70)
|
||
print("DEGENERATION TEST 1 v2 — Official encoding diff")
|
||
print("=" * 70)
|
||
|
||
tokenizer = AutoTokenizer.from_pretrained(CHECKPOINT_DIR)
|
||
bos = tokenizer.bos_token_id or 0
|
||
|
||
# === 1. Hand-rolled (current single_shot_inference.py) ===
|
||
input_ids = [bos, USER_TOKEN]
|
||
input_ids += tokenizer.encode('\n\n' + PROMPT, add_special_tokens=False)
|
||
input_ids.append(ASSISTANT_TOKEN)
|
||
input_ids.append(THINK_START)
|
||
|
||
print(f"\n1. HAND-ROLLED ({len(input_ids)} tokens):")
|
||
for i, tid in enumerate(input_ids):
|
||
print(f" [{i:3d}] id={tid:>7d} {repr(tokenizer.decode([tid]))}")
|
||
print(f" Full: {repr(tokenizer.decode(input_ids))}")
|
||
|
||
# === 2. Official encoding (thinking mode, no system prompt) ===
|
||
# Format: <BOS><｜User｜>{msg}<｜Assistant｜>ately
|
||
# NO \n\n between User token and message
|
||
canonical_thinking = [bos, USER_TOKEN]
|
||
canonical_thinking += tokenizer.encode(PROMPT, add_special_tokens=False)
|
||
canonical_thinking.append(ASSISTANT_TOKEN)
|
||
canonical_thinking.append(THINK_START)
|
||
|
||
print(f"\n2. OFFICIAL (thinking, no \\n\\n) ({len(canonical_thinking)} tokens):")
|
||
for i, tid in enumerate(canonical_thinking):
|
||
print(f" [{i:3d}] id={tid:>7d} {repr(tokenizer.decode([tid]))}")
|
||
print(f" Full: {repr(tokenizer.decode(canonical_thinking))}")
|
||
|
||
# === 3. Official encoding (chat mode — THINK_END closes thinking) ===
|
||
canonical_chat = [bos, USER_TOKEN]
|
||
canonical_chat += tokenizer.encode(PROMPT, add_special_tokens=False)
|
||
canonical_chat.append(ASSISTANT_TOKEN)
|
||
canonical_chat.append(THINK_END)
|
||
|
||
print(f"\n3. OFFICIAL (chat mode, THINK_END) ({len(canonical_chat)} tokens):")
|
||
for i, tid in enumerate(canonical_chat):
|
||
print(f" [{i:3d}] id={tid:>7d} {repr(tokenizer.decode([tid]))}")
|
||
print(f" Full: {repr(tokenizer.decode(canonical_chat))}")
|
||
|
||
# === 4. Official encoding with system prompt ===
|
||
# Format: <BOS>{system}<｜User｜>{msg}<｜Assistant｜>ately
|
||
system_prompt = "You are a helpful assistant."
|
||
canonical_sys = tokenizer.encode(system_prompt, add_special_tokens=False)
|
||
canonical_sys_thinking = [bos] + canonical_sys + [USER_TOKEN]
|
||
canonical_sys_thinking += tokenizer.encode(PROMPT, add_special_tokens=False)
|
||
canonical_sys_thinking.append(ASSISTANT_TOKEN)
|
||
canonical_sys_thinking.append(THINK_START)
|
||
|
||
print(f"\n4. OFFICIAL (thinking + system prompt) ({len(canonical_sys_thinking)} tokens):")
|
||
for i, tid in enumerate(canonical_sys_thinking):
|
||
print(f" [{i:3d}] id={tid:>7d} {repr(tokenizer.decode([tid]))}")
|
||
print(f" Full: {repr(tokenizer.decode(canonical_sys_thinking))}")
|
||
|
||
# === 5. Diff ===
|
||
print(f"\n{'='*70}")
|
||
print("DIFF: hand-rolled vs official (thinking, no \\n\\n)")
|
||
print(f"{'='*70}")
|
||
if input_ids == canonical_thinking:
|
||
print(" IDENTICAL")
|
||
else:
|
||
print(f" DIFFERENT: hand_rolled={len(input_ids)} tokens, canonical={len(canonical_thinking)} tokens")
|
||
min_len = min(len(input_ids), len(canonical_thinking))
|
||
for i in range(min_len):
|
||
if input_ids[i] != canonical_thinking[i]:
|
||
print(f" First diff at position {i}:")
|
||
print(f" hand_rolled[{i}] = {input_ids[i]} ({repr(tokenizer.decode([input_ids[i]]))})")
|
||
print(f" canonical[{i}] = {canonical_thinking[i]} ({repr(tokenizer.decode([canonical_thinking[i]]))})")
|
||
# Show context
|
||
for j in range(max(0,i-2), min(len(input_ids), i+3)):
|
||
hr = input_ids[j] if j < len(input_ids) else "—"
|
||
cn = canonical_thinking[j] if j < len(canonical_thinking) else "—"
|
||
mark = " <<<" if j == i else ""
|
||
print(f" [{j}] hand={hr} canon={cn}{mark}")
|
||
break
|
||
else:
|
||
if len(input_ids) != len(canonical_thinking):
|
||
print(f" Same prefix but different lengths: {len(input_ids)} vs {len(canonical_thinking)}")
|
||
longer = input_ids if len(input_ids) > len(canonical_thinking) else canonical_thinking
|
||
shorter_len = min(len(input_ids), len(canonical_thinking))
|
||
label = "hand_rolled" if len(input_ids) > len(canonical_thinking) else "canonical"
|
||
for j in range(shorter_len, len(longer)):
|
||
print(f" Extra in {label}: [{j}] = {longer[j]} ({repr(tokenizer.decode([longer[j]]))})")
|
||
|
||
# === 6. The key question: does the \n\n matter? ===
|
||
# Check what token 271 decodes to (it's our \n\n)
|
||
print(f"\n{'='*70}")
|
||
print("ANALYSIS")
|
||
print(f"{'='*70}")
|
||
# The only difference should be the \n\n token (id 271) in hand-rolled
|
||
# Check if tokenizer encodes PROMPT differently with/without \n\n prefix
|
||
enc_with_prefix = tokenizer.encode('\n\n' + PROMPT, add_special_tokens=False)
|
||
enc_no_prefix = tokenizer.encode(PROMPT, add_special_tokens=False)
|
||
print(f" encode('\\n\\n' + PROMPT) = {enc_with_prefix} ({len(enc_with_prefix)} tokens)")
|
||
print(f" encode(PROMPT) = {enc_no_prefix} ({len(enc_no_prefix)} tokens)")
|
||
if len(enc_with_prefix) > len(enc_no_prefix):
|
||
diff_tokens = enc_with_prefix[:len(enc_with_prefix) - len(enc_no_prefix)]
|
||
print(f" Extra tokens from \\n\\n: {diff_tokens}")
|
||
for t in diff_tokens:
|
||
print(f" id={t}: {repr(tokenizer.decode([t]))}")
|
||
# Check if the remaining tokens match
|
||
if enc_with_prefix[len(diff_tokens):] == enc_no_prefix:
|
||
print(f" Remaining tokens MATCH — \\n\\n only adds prefix tokens")
|
||
else:
|
||
print(f" WARNING: remaining tokens DIFFER — \\n\\n changes tokenization!")
|
||
print(f" with prefix tail: {enc_with_prefix[len(diff_tokens):]}")
|
||
print(f" without prefix: {enc_no_prefix}")
|
||
|
||
# === 7. What does SGLang use? ===
|
||
# From the SGLang docs: --reasoning-parser deepseek-v4 and SGLANG_DEFAULT_THINKING=1
|
||
# This should use the same encoding. Let's check the raw tokenizer.json for special tokens
|
||
print(f"\n{'='*70}")
|
||
print("SPECIAL TOKEN INVENTORY")
|
||
print(f"{'='*70}")
|
||
if hasattr(tokenizer, 'added_tokens_decoder'):
|
||
for tid_str, tok in sorted(tokenizer.added_tokens_decoder.items(), key=lambda x: int(x[0])):
|
||
tid = int(tid_str)
|
||
s = str(tok)
|
||
if tid >= 128000 or any(x in s.lower() for x in ['think', 'user', 'assistant', 'end', 'sentence', 'dsml']):
|
||
print(f" id={tid:>7d}: {repr(s)}")
|
||
if hasattr(tokenizer, 'special_tokens_map'):
|
||
for k, v in tokenizer.special_tokens_map.items():
|
||
tid = tokenizer.convert_tokens_to_ids(v) if isinstance(v, str) else '—'
|
||
print(f" special: {k} = {repr(v)} (id={tid})")
|
||
|
||
# === 8. Verdict ===
|
||
print(f"\n{'='*70}")
|
||
print("VERDICT")
|
||
print(f"{'='*70}")
|
||
if input_ids == canonical_thinking:
|
||
print(" Hand-rolled matches official thinking-mode encoding.")
|
||
print(" Prompt is CORRECT per the official spec.")
|
||
print(" Degeneration is NOT caused by prompt format → look at Test 2.")
|
||
else:
|
||
print(" Hand-rolled DIFFERS from official encoding!")
|
||
print(" This is likely contributing to degenerate output.")
|
||
print(" FIX: Use canonical_thinking encoding in single_shot_inference.py.")
|
||
print(f" Also try: canonical_chat (THINK_END after Assistant) for non-reasoning mode.")
|
||
print(f" Also try: canonical_sys_thinking (with system prompt).")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|