405 lines
15 KiB
Markdown
405 lines
15 KiB
Markdown
---
|
|
library_name: transformers
|
|
license: mit
|
|
datasets:
|
|
- Zhengping/UNLI
|
|
- Zhengping/UNLI-style-synthetic
|
|
language:
|
|
- en
|
|
metrics:
|
|
- pearsonr
|
|
- spearmanr
|
|
- accuracy
|
|
base_model:
|
|
- Qwen/Qwen2.5-14B-Instruct
|
|
---
|
|
|
|
# Model Card for Model ID
|
|
|
|
<!-- Provide a quick summary of what the model is/does. -->
|
|
|
|
|
|
|
|
## Model Details
|
|
|
|
### Model Description
|
|
|
|
<!-- Provide a longer summary of what this model is. -->
|
|
|
|
This is the model card of a 🤗 transformers model that has been pushed on the Hub. This model card has been automatically generated.
|
|
|
|
- **Developed by:** Liaoyaqi Wang, Zhengping Jiang, Anqi Liu, Benjamin Van Durme
|
|
- **Model type:** Decoding-based Regression Model (Classification)
|
|
- **Language(s) (NLP):** `en`
|
|
- **License:** mit
|
|
- **Finetuned from model [optional]:** Qwen/Qwen2.5-14B-Instruct
|
|
|
|
### Model Sources [optional]
|
|
|
|
<!-- Provide the basic links for the model. -->
|
|
|
|
- **Repository:** [Decoding-based Regression](https://github.com/zipJiang/decoding-based-regression.git)
|
|
- **Paper [optional]:** [Always Tell Me The Odds: Fine-grained Conditional Probability Estimation](https://arxiv.org/pdf/2505.01595)
|
|
|
|
## Uses
|
|
|
|
<!-- Address questions around how the model is intended to be used, including the foreseeable users of the model and those affected by the model. -->
|
|
|
|
### Direct Use
|
|
|
|
<!-- This section is for the model use without fine-tuning or plugging into a larger ecosystem/app. -->
|
|
|
|
```python
|
|
import enum
|
|
import transformers
|
|
import torch
|
|
from transformers.pipelines import PIPELINE_REGISTRY
|
|
from transformers import (
|
|
pipeline,
|
|
Pipeline,
|
|
TextGenerationPipeline,
|
|
PreTrainedTokenizer,
|
|
AutoModelForCausalLM,
|
|
PreTrainedTokenizer
|
|
)
|
|
from transformers.pipelines.text_generation import Chat, ReturnType
|
|
from typing import (
|
|
Dict,
|
|
Callable,
|
|
Tuple,
|
|
List,
|
|
)
|
|
|
|
|
|
class LevelToScorePipeline(TextGenerationPipeline):
|
|
|
|
def __init__(
|
|
self,
|
|
level_to_score_func: Callable[[Tuple[torch.FloatTensor], PreTrainedTokenizer], Tuple[List[float], List[List[float]]]],
|
|
*args,
|
|
**kwargs
|
|
):
|
|
super().__init__(*args, **kwargs)
|
|
self._level_to_score_func = level_to_score_func
|
|
|
|
def preprocess(
|
|
self,
|
|
prompt_text,
|
|
prefix="",
|
|
handle_long_generation=None,
|
|
add_special_tokens=None,
|
|
truncation=None,
|
|
padding=None,
|
|
max_length=None,
|
|
continue_final_message=None,
|
|
**generate_kwargs,
|
|
):
|
|
# Only set non-None tokenizer kwargs, so as to rely on the tokenizer's defaults
|
|
tokenizer_kwargs = {
|
|
"add_special_tokens": add_special_tokens,
|
|
"truncation": truncation,
|
|
"padding": padding,
|
|
"max_length": max_length,
|
|
}
|
|
tokenizer_kwargs = {key: value for key, value in tokenizer_kwargs.items() if value is not None}
|
|
|
|
if isinstance(prompt_text, Chat):
|
|
tokenizer_kwargs.pop("add_special_tokens", None) # ignore add_special_tokens on chats
|
|
# If the user passes a chat that ends in an assistant message, we treat it as a prefill by default
|
|
# because very few models support multiple separate, consecutive assistant messages
|
|
if continue_final_message is None:
|
|
continue_final_message = prompt_text.messages[-1]["role"] == "assistant"
|
|
inputs = self.tokenizer.apply_chat_template(
|
|
prompt_text.messages,
|
|
add_generation_prompt=not continue_final_message,
|
|
continue_final_message=continue_final_message,
|
|
return_dict=True,
|
|
return_tensors=self.framework,
|
|
**tokenizer_kwargs,
|
|
)
|
|
else:
|
|
inputs = self.tokenizer(prefix + prompt_text, return_tensors=self.framework, **tokenizer_kwargs)
|
|
|
|
inputs["prompt_text"] = prompt_text
|
|
|
|
if handle_long_generation == "hole":
|
|
cur_len = inputs["input_ids"].shape[-1]
|
|
if "max_new_tokens" in generate_kwargs:
|
|
new_tokens = generate_kwargs["max_new_tokens"]
|
|
else:
|
|
new_tokens = generate_kwargs.get("max_length", self.generation_config.max_length) - cur_len
|
|
if new_tokens < 0:
|
|
raise ValueError("We cannot infer how many new tokens are expected")
|
|
if cur_len + new_tokens > self.tokenizer.model_max_length:
|
|
keep_length = self.tokenizer.model_max_length - new_tokens
|
|
if keep_length <= 0:
|
|
raise ValueError(
|
|
"We cannot use `hole` to handle this generation the number of desired tokens exceeds the"
|
|
" models max length"
|
|
)
|
|
|
|
inputs["input_ids"] = inputs["input_ids"][:, -keep_length:]
|
|
if "attention_mask" in inputs:
|
|
inputs["attention_mask"] = inputs["attention_mask"][:, -keep_length:]
|
|
|
|
return inputs
|
|
|
|
def _forward(self, model_inputs, **generate_kwargs):
|
|
input_ids = model_inputs["input_ids"]
|
|
attention_mask = model_inputs.get("attention_mask", None)
|
|
# Allow empty prompts
|
|
if input_ids.shape[1] == 0:
|
|
input_ids = None
|
|
attention_mask = None
|
|
in_b = 1
|
|
else:
|
|
in_b = input_ids.shape[0]
|
|
prompt_text = model_inputs.pop("prompt_text")
|
|
|
|
# If there is a prefix, we may need to adjust the generation length. Do so without permanently modifying
|
|
# generate_kwargs, as some of the parameterization may come from the initialization of the pipeline.
|
|
prefix_length = generate_kwargs.pop("prefix_length", 0)
|
|
if prefix_length > 0:
|
|
has_max_new_tokens = "max_new_tokens" in generate_kwargs or (
|
|
"generation_config" in generate_kwargs
|
|
and generate_kwargs["generation_config"].max_new_tokens is not None
|
|
)
|
|
if not has_max_new_tokens:
|
|
generate_kwargs["max_length"] = generate_kwargs.get("max_length") or self.generation_config.max_length
|
|
generate_kwargs["max_length"] += prefix_length
|
|
has_min_new_tokens = "min_new_tokens" in generate_kwargs or (
|
|
"generation_config" in generate_kwargs
|
|
and generate_kwargs["generation_config"].min_new_tokens is not None
|
|
)
|
|
if not has_min_new_tokens and "min_length" in generate_kwargs:
|
|
generate_kwargs["min_length"] += prefix_length
|
|
|
|
# User-defined `generation_config` passed to the pipeline call take precedence
|
|
if "generation_config" not in generate_kwargs:
|
|
generate_kwargs["generation_config"] = self.generation_config
|
|
|
|
generate_kwargs["output_scores"] = not generate_kwargs.get("do_sample", False)
|
|
generate_kwargs["return_dict_in_generate"] = True
|
|
|
|
generated_sequence = self.model.generate(input_ids=input_ids, attention_mask=attention_mask, **generate_kwargs)
|
|
|
|
logits = None
|
|
|
|
# TODO: check good default
|
|
if generate_kwargs.get("return_scores", True):
|
|
assert not generate_kwargs.get("do_sample", False), "return_logits=True is only supported for do_sample=False"
|
|
|
|
# Proceed to process logits and convert to score average.
|
|
# next_token_logits is [batch_size, vocab_size]
|
|
# raw_logits is a tuple of ([next_token_logits, past_key_values])
|
|
|
|
logits = generated_sequence.scores
|
|
|
|
out_b = generated_sequence.sequences.shape[0]
|
|
if self.framework == "pt":
|
|
generated_sequence = generated_sequence.sequences.reshape(in_b, out_b // in_b, *generated_sequence.sequences.shape[1:])
|
|
# elif self.framework == "tf":
|
|
# generated_sequence = tf.reshape(generated_sequence, (in_b, out_b // in_b, *generated_sequence.shape[1:]))
|
|
return {"generated_sequence": generated_sequence, "input_ids": input_ids, "prompt_text": prompt_text, "logits": logits}
|
|
|
|
def postprocess(
|
|
self,
|
|
model_outputs,
|
|
return_type=ReturnType.FULL_TEXT,
|
|
clean_up_tokenization_spaces=True,
|
|
continue_final_message=None,
|
|
):
|
|
generated_sequence = model_outputs["generated_sequence"][0]
|
|
input_ids = model_outputs["input_ids"]
|
|
prompt_text = model_outputs["prompt_text"]
|
|
logits = model_outputs["logits"]
|
|
|
|
#TODO: This is now making many assumptions about how the logits are ordered,
|
|
# Should think about how to make this explicit
|
|
scores, selective_logits = self._level_to_score_func(logits, self.tokenizer)
|
|
|
|
generated_sequence = generated_sequence.numpy().tolist()
|
|
records = []
|
|
for sequence in generated_sequence:
|
|
if return_type == ReturnType.TENSORS:
|
|
record = {"generated_token_ids": sequence}
|
|
elif return_type in {ReturnType.NEW_TEXT, ReturnType.FULL_TEXT}:
|
|
# Decode text
|
|
text = self.tokenizer.decode(
|
|
sequence,
|
|
skip_special_tokens=True,
|
|
clean_up_tokenization_spaces=clean_up_tokenization_spaces,
|
|
)
|
|
|
|
# Remove PADDING prompt of the sequence if XLNet or Transfo-XL model is used
|
|
if input_ids is None:
|
|
prompt_length = 0
|
|
else:
|
|
prompt_length = len(
|
|
self.tokenizer.decode(
|
|
input_ids[0],
|
|
skip_special_tokens=True,
|
|
clean_up_tokenization_spaces=clean_up_tokenization_spaces,
|
|
)
|
|
)
|
|
|
|
all_text = text[prompt_length:]
|
|
if return_type == ReturnType.FULL_TEXT:
|
|
if isinstance(prompt_text, str):
|
|
all_text = prompt_text + all_text
|
|
elif isinstance(prompt_text, Chat):
|
|
if continue_final_message is None:
|
|
# If the user passes a chat ending in an assistant message, we treat it as a prefill by
|
|
# default because very few models support multiple separate, consecutive assistant messages
|
|
continue_final_message = prompt_text.messages[-1]["role"] == "assistant"
|
|
if continue_final_message:
|
|
# With assistant prefill, concat onto the end of the last message
|
|
all_text = list(prompt_text.messages)[:-1] + [
|
|
{
|
|
"role": prompt_text.messages[-1]["role"],
|
|
"content": prompt_text.messages[-1]["content"] + all_text,
|
|
}
|
|
]
|
|
else:
|
|
# When we're not starting from a prefill, the output is a new assistant message
|
|
all_text = list(prompt_text.messages) + [{"role": "assistant", "content": all_text}]
|
|
record = {
|
|
"generated_text": all_text,
|
|
"score": scores[0],
|
|
"selective_logits": selective_logits[0]
|
|
}
|
|
records.append(record)
|
|
|
|
return records
|
|
|
|
|
|
class SingleLabelRankDict:
|
|
def __init__(
|
|
self,
|
|
rank_dict: Dict[Text, Any]
|
|
):
|
|
self._rank_dict = rank_dict
|
|
|
|
def __len__(self) -> int:
|
|
return len(self._rank_dict)
|
|
|
|
def get_rank_dict(self, tokenizer: PreTrainedTokenizer) -> Dict[int, Any]:
|
|
return {tokenizer.convert_tokens_to_ids([token])[0]: value for token, value in self._rank_dict.items()}
|
|
|
|
def to_tokenizer(self, tokenizer: PreTrainedTokenizer) -> PreTrainedTokenizer:
|
|
"""Augment tokenizer vocab with `rank_dict` IN-PLACE.
|
|
"""
|
|
vocabs: List[Text] = self._rank_dict.keys()
|
|
new_vocab = [vocab for vocab in vocabs if vocab not in tokenizer.get_vocab()]
|
|
tokenizer.add_tokens(new_vocab)
|
|
return tokenizer
|
|
def from_tokenizer(cls, tokenizer: PreTrainedTokenizer) -> "SingleLabelRankDict":
|
|
vocab = tokenizer.get_vocab()
|
|
rank_dict = {}
|
|
pattern = re.compile(r" <\|label_level_(\d+)\|>")
|
|
|
|
for token in vocab.keys():
|
|
match = pattern.match(token)
|
|
if match:
|
|
value = int(match.group(1))
|
|
# normalized_value = value / (len(vocab) - 1)
|
|
rank_dict[token] = value
|
|
|
|
# normalize rank_values
|
|
num_levels = max(rank_dict.values()) + 1
|
|
for token in rank_dict.keys():
|
|
rank_dict[token] = 1. / num_levels * (rank_dict[token] + 0.5)
|
|
|
|
return cls(rank_dict=rank_dict)
|
|
|
|
|
|
model = transformers.AutoModelForCausalLM.from_pretrained(
|
|
"Zhengping/conditional-probability-regression",
|
|
torch_dtype="auto",
|
|
attn_implementation="flash_attention_2",
|
|
)
|
|
tokenizer = transformers.AutoTokenizer.from_pretrained(
|
|
"Zhengping/conditional-probability-regression",
|
|
)
|
|
|
|
rank_dict = SingleLabelRankDict.from_tokenizer(tokenizer)
|
|
|
|
PIPELINE_REGISTRY.register_pipeline(
|
|
"level-to-score",
|
|
pipeline_class=LevelToScorePipeline,
|
|
pt_model=AutoModelForCausalLM
|
|
)
|
|
|
|
# This allows fine-grained labeling, the greedy decoding gives a coarse score,
|
|
# one can also attach their own level-to-score function to the pipeline, e.g. using UNLI
|
|
# label transformation to get it more binarized
|
|
def _level_to_score_func(
|
|
logits: Tuple[torch.FloatTensor],
|
|
tokenizer: PreTrainedTokenizer
|
|
) -> Tuple[List[float], List[float]]:
|
|
""" """
|
|
logits = logits[0]
|
|
num_labels = len(rank_dict)
|
|
considering_ids = tokenizer.convert_tokens_to_ids([f" <|label_level_{i}|>" for i in range(num_labels)])
|
|
selective_logits = torch.index_select(logits, 1, torch.tensor(considering_ids, device=logits.device))
|
|
step_size = 1 / num_labels
|
|
expectation = torch.tensor([[i * step_size + 1 / 2 * step_size for i in range(num_labels)]], device=selective_logits.device)
|
|
scores = torch.softmax(selective_logits, dim=-1) @ expectation.T
|
|
scores = scores.squeeze(-1).tolist()
|
|
return scores, selective_logits.tolist()
|
|
|
|
pipe = pipeline(
|
|
"level-to-score",
|
|
model=model,
|
|
max_new_tokens=2,
|
|
tokenizer=tokenizer,
|
|
device=0,
|
|
level_to_score_func=_level_to_score_func,
|
|
torch_dtype=torch.bfloat16,
|
|
)
|
|
|
|
template = UNLITemplate()
|
|
|
|
premise = "Sam is sleeping."
|
|
hypothesis = "Sam is awake."
|
|
|
|
inputs = [
|
|
{
|
|
"role": "user",
|
|
"content": "### Question: Given the premise \"{premise}\", how likely is it that the hypothesis \"{hypothesis}\" is true?\n\n".format(
|
|
premise=premise,
|
|
hypothesis=hypothesis
|
|
)
|
|
},
|
|
{
|
|
"role": "assitant",
|
|
"content": "### Answer:"
|
|
}
|
|
]
|
|
|
|
result = pipe(inputs)
|
|
print(result)
|
|
```
|
|
|
|
|
|
## Use with vLLM
|
|
|
|
`TODO`
|
|
|
|
|
|
#### Summary
|
|
|
|
LLM-based Fine-grained Conditional Probability Estimation
|
|
|
|
## Citation [optional]
|
|
|
|
```bibtex
|
|
@article{wang2025always,
|
|
title={Always Tell Me The Odds: Fine-grained Conditional Probability Estimation},
|
|
author={Wang, Liaoyaqi and Jiang, Zhengping and Liu, Anqi and Van Durme, Benjamin},
|
|
journal={arXiv preprint arXiv:2505.01595},
|
|
year={2025}
|
|
}
|
|
```
|
|
|
|
<!-- If there is a paper or blog post introducing the model, the APA and Bibtex information for that should go in this section. --> |