第61日: 指示データフォーマット、スキーマとトークン境界アライメント

def normalize_to_chatml(record):
messages = []
if 'instruction' in record and 'output' in record:
if record.get('input', '').strip():
content = f"{record['instruction']}\n\nInput:\n{record['input']}"
else:
content = record['instruction']
messages.append({'role': 'user', 'content': content})
messages.append({'role': 'assistant', 'content': record['output']})
elif 'conversations' in record:
role_map = {'human': 'user', 'gpt': 'assistant', 'system': 'system'}
for turn in record['conversations']:
messages.append({'role': role_map.get(turn['from'], turn['from']), 'content': turn['value']})
return messages

from transformers import AutoTokenizer
def verify_special_tokens(tokenizer_name_or_path='gpt2'):
tokenizer = AutoTokenizer.from_pretrained(tokenizer_name_or_path)
num_added = tokenizer.add_special_tokens({'additional_special_tokens': ['<|im_start|>', '<|im_end|>']})
encoded = tokenizer.encode('<|im_start|>user\nHello<|im_end|>', add_special_tokens=False)
start_id = tokenizer.convert_tokens_to_ids('<|im_start|>')
end_id = tokenizer.convert_tokens_to_ids('<|im_end|>')
assert encoded[0] == start_id, f'Expected start token {start_id}, got {encoded[0]}'
assert encoded[-1] == end_id, f'Expected end token {end_id}, got {encoded[-1]}'
return True

def truncate_dialogue(messages, max_tokens, tokenizer):
system_msg = [m for m in messages if m['role'] == 'system']
history = [m for m in messages if m['role'] != 'system']
while True:
candidate = system_msg + history
rendered = tokenizer.apply_chat_template(candidate, tokenize=False)
tokens = tokenizer.encode(rendered)
if len(tokens) <= max_tokens or len(history) < 2:
return candidate
history = history[2:]

class ChatMLDatasetValidator:
def __init__(self, tokenizer):
self.tokenizer = tokenizer
def validate_record(self, messages):
if not messages:
return False, 'Empty messages'
roles = [m['role'] for m in messages]
if roles[0] == 'system':
turns = roles[1:]
else:
turns = roles
if not turns or turns[0] != 'user' or turns[-1] != 'assistant':
return False, 'Must start with user and end with assistant'
for i in range(1, len(turns)):
if turns[i] == turns[i-1]:
return False, f'Consecutive duplicate role: {turns[i]}'
return True, 'Valid'
def process_batch(self, batch):
valid, filtered = [], []
for item in batch:
ok, reason = self.validate_record(item)
if ok:
valid.append(item)
else:
filtered.append((item, reason))
return valid, filtered