| """Strict model-owned repetition, with no implicit switch to single-read mode.""" |
| from .prompt_dyno import state_messages, options_of, render_value, DEFAULT_QUESTION, INSTRUCTION, ANSWER, render |
|
|
| READ_AGAIN = 'Read the same input again before answering:\n\n' |
|
|
|
|
| class LengthError(ValueError): |
| pass |
|
|
|
|
| def render_twice(processor, state, question, labels): |
| messages, media = state_messages(state) |
| if media: raise ValueError('This release is text-only') |
| keys, texts = options_of(question) |
| head = (render_value(question['instructions']) |
| if question.get('instructions') is not None else DEFAULT_QUESTION) |
| options = '\n'.join(f'{labels[i]}. {value}' for i, value in enumerate(texts)) |
| ask = INSTRUCTION + f'\n\nQuestion: {head}\nOptions:\n{options}' |
| first = [dict(m) for m in messages] + [{'role': 'user', 'content': ask}] |
| again = [dict(m) for m in first if m.get('role') != 'system'] |
| if again[0]['role'] == 'user': |
| content = again[0].get('content') |
| content = ([{'type': 'text', 'text': READ_AGAIN}] + content |
| if isinstance(content, list) else READ_AGAIN + (content or '')) |
| again[0] = {**again[0], 'content': content} |
| else: again.insert(0, {'role': 'user', 'content': READ_AGAIN.rstrip()}) |
| text = processor.apply_chat_template(first + again, tokenize=False, |
| add_generation_prompt=True, enable_thinking=False) |
| return text + ANSWER, keys |
|
|
|
|
| def encode(processor, labels, label_ids, request, mode='default', limit=16384): |
| if mode not in ('default', 'flash'): raise ValueError('mode must be default or flash') |
| if not isinstance(request, dict) or 'state' not in request: |
| raise ValueError('Pass a structured decision request with state and question(s)') |
| if request.get('images'): raise ValueError('This release is text-only') |
| questions = request.get('questions') |
| if questions is None: questions = {'q': request['question']} |
| if not isinstance(questions, dict) or not questions: raise ValueError('Empty questions') |
| _, once, media = render(processor, request['state'], questions, labels) |
| if media: raise ValueError('This release is text-only') |
| output = [] |
| for key, question in questions.items(): |
| text, keys = (render_twice(processor, request['state'], question, labels) |
| if mode == 'default' else once[key]) |
| if keys != once[key][1]: raise ValueError('Option order changed') |
| if len(keys) > len(label_ids): raise ValueError('Unsupported option count') |
| ids = processor.tokenizer.encode(text, add_special_tokens=False) |
| if not 1 <= len(ids) <= limit: |
| raise LengthError(f'{mode} rendered sequence has {len(ids)} tokens; limit {limit}. ' |
| 'No truncation or automatic flash fallback. Shorten input or explicitly request mode="flash".') |
| |
| for label, token in zip(labels[:len(keys)], label_ids[:len(keys)]): |
| if processor.tokenizer.encode(text+' '+label, add_special_tokens=False) != ids+[token]: |
| raise ValueError('Answer label is not one token at the actual prompt boundary') |
| output.append({'qid': key, 'ids': ids, 'keys': keys, 'kind': question['type']}) |
| return output |
|
|
