Habiba A. Elbehairy commited on
Commit ·
2113508
1
Parent(s): 32f6522
fiix
Browse files
app.py
CHANGED
|
@@ -24,80 +24,11 @@ class MultitaskCodeSimilarityModel(nn.Module):
|
|
| 24 |
self.encoder = AutoModel.from_pretrained(model_name, config=self.config)
|
| 25 |
self.classifier = nn.Linear(self.config.hidden_size, num_labels)
|
| 26 |
|
| 27 |
-
|
| 28 |
-
self.decoder_embedding = nn.Linear(self.config.hidden_size, self.config.hidden_size)
|
| 29 |
-
self.decoder = nn.GRU(
|
| 30 |
-
input_size=self.config.hidden_size,
|
| 31 |
-
hidden_size=self.config.hidden_size,
|
| 32 |
-
batch_first=True
|
| 33 |
-
)
|
| 34 |
-
self.explanation_head = nn.Linear(self.config.hidden_size, len(tokenizer))
|
| 35 |
-
|
| 36 |
-
def forward(self, input_ids, attention_mask, explanation_ids=None, explanation_mask=None):
|
| 37 |
outputs = self.encoder(input_ids=input_ids, attention_mask=attention_mask)
|
| 38 |
pooled = outputs.last_hidden_state[:, 0]
|
| 39 |
logits = self.classifier(pooled)
|
| 40 |
-
|
| 41 |
-
explanation_logits = None
|
| 42 |
-
if explanation_ids is not None:
|
| 43 |
-
batch_size = input_ids.size(0)
|
| 44 |
-
seq_length = explanation_ids.size(1)
|
| 45 |
-
|
| 46 |
-
# Initialize decoder with pooled representation
|
| 47 |
-
decoder_input = self.decoder_embedding(pooled).unsqueeze(1).expand(-1, seq_length, -1)
|
| 48 |
-
|
| 49 |
-
# Run decoder
|
| 50 |
-
decoder_outputs, _ = self.decoder(decoder_input)
|
| 51 |
-
|
| 52 |
-
# Generate logits for each position
|
| 53 |
-
explanation_logits = self.explanation_head(decoder_outputs)
|
| 54 |
-
|
| 55 |
-
return logits, explanation_logits
|
| 56 |
-
|
| 57 |
-
def generate_explanation(self, input_ids, attention_mask, max_length=128):
|
| 58 |
-
"""Generate explanation text for inference"""
|
| 59 |
-
device = input_ids.device
|
| 60 |
-
|
| 61 |
-
# Get encoding
|
| 62 |
-
outputs = self.encoder(input_ids=input_ids, attention_mask=attention_mask)
|
| 63 |
-
pooled = outputs.last_hidden_state[:, 0]
|
| 64 |
-
|
| 65 |
-
# First token (usually [CLS] or <s>)
|
| 66 |
-
bos_token_id = self.tokenizer.bos_token_id if self.tokenizer.bos_token_id is not None else self.tokenizer.cls_token_id
|
| 67 |
-
current_token_id = torch.full((pooled.size(0), 1), bos_token_id, dtype=torch.long, device=device)
|
| 68 |
-
|
| 69 |
-
generated_ids = [current_token_id]
|
| 70 |
-
|
| 71 |
-
# Initial hidden state
|
| 72 |
-
hidden = pooled.unsqueeze(0) # Add seq dimension for GRU
|
| 73 |
-
|
| 74 |
-
for _ in range(max_length - 1):
|
| 75 |
-
# Get decoder input from current token
|
| 76 |
-
decoder_input = self.decoder_embedding(pooled).unsqueeze(1)
|
| 77 |
-
|
| 78 |
-
# Run decoder one step
|
| 79 |
-
decoder_output, hidden = self.decoder(decoder_input, hidden)
|
| 80 |
-
|
| 81 |
-
# Get next token probabilities
|
| 82 |
-
next_token_logits = self.explanation_head(decoder_output.squeeze(1))
|
| 83 |
-
next_token_id = torch.argmax(next_token_logits, dim=-1, keepdim=True)
|
| 84 |
-
|
| 85 |
-
# Stop if we predict EOS
|
| 86 |
-
if (next_token_id == self.tokenizer.eos_token_id).all():
|
| 87 |
-
break
|
| 88 |
-
|
| 89 |
-
generated_ids.append(next_token_id)
|
| 90 |
-
|
| 91 |
-
# Concatenate all generated tokens
|
| 92 |
-
all_tokens = torch.cat(generated_ids, dim=1)
|
| 93 |
-
|
| 94 |
-
# Convert to text
|
| 95 |
-
explanations = []
|
| 96 |
-
for tokens in all_tokens:
|
| 97 |
-
explanation = self.tokenizer.decode(tokens, skip_special_tokens=True)
|
| 98 |
-
explanations.append(explanation)
|
| 99 |
-
|
| 100 |
-
return explanations
|
| 101 |
|
| 102 |
# Load model and tokenizer
|
| 103 |
try:
|
|
@@ -188,24 +119,16 @@ async def predict(data: SimilarityInput):
|
|
| 188 |
# Get prediction from the model
|
| 189 |
with torch.no_grad():
|
| 190 |
# Check if using custom model or fallback
|
| 191 |
-
if hasattr(model, '
|
| 192 |
# Our custom model
|
| 193 |
-
logits
|
| 194 |
input_ids=inputs["input_ids"],
|
| 195 |
attention_mask=inputs["attention_mask"]
|
| 196 |
)
|
| 197 |
-
|
| 198 |
-
# Generate explanation
|
| 199 |
-
explanations = model.generate_explanation(
|
| 200 |
-
input_ids=inputs["input_ids"],
|
| 201 |
-
attention_mask=inputs["attention_mask"]
|
| 202 |
-
)
|
| 203 |
-
explanation = explanations[0] if explanations else ""
|
| 204 |
else:
|
| 205 |
# Fallback to standard model
|
| 206 |
outputs = model(**inputs)
|
| 207 |
logits = outputs.logits
|
| 208 |
-
explanation = ""
|
| 209 |
|
| 210 |
# Process results
|
| 211 |
probs = torch.softmax(logits, dim=-1)[0].cpu().tolist()
|
|
@@ -214,16 +137,6 @@ async def predict(data: SimilarityInput):
|
|
| 214 |
# Map prediction to class name
|
| 215 |
classification = label_to_class.get(prediction, "Unknown")
|
| 216 |
|
| 217 |
-
# Generate explanations contextually if not available from model
|
| 218 |
-
if not explanation or explanation.strip() == "":
|
| 219 |
-
# Template explanations based on classification
|
| 220 |
-
if classification == "Duplicate":
|
| 221 |
-
explanation = f"Tests {data.test_case_1.name} and {data.test_case_2.name} are duplicates because they both check the output formatting of their respective methods using the same approach of redirecting stdout to a buffer and verifying the exact output string."
|
| 222 |
-
elif classification == "Redundant":
|
| 223 |
-
explanation = f"Tests {data.test_case_1.name} and {data.test_case_2.name} are redundant because they test similar functionality (output formatting) using the same testing technique (capturing stdout) but on different methods."
|
| 224 |
-
elif classification == "Distinct":
|
| 225 |
-
explanation = f"Tests {data.test_case_1.name} and {data.test_case_2.name} are distinct because they test completely different functionality of the BankApp class: one tests listClients() while the other tests deposit()."
|
| 226 |
-
|
| 227 |
return {
|
| 228 |
"pair_id": data.pair_id,
|
| 229 |
"test_case_1_name": data.test_case_1.name,
|
|
@@ -231,7 +144,6 @@ async def predict(data: SimilarityInput):
|
|
| 231 |
"similarity": {
|
| 232 |
"score": prediction,
|
| 233 |
"classification": classification,
|
| 234 |
-
"explanation": explanation
|
| 235 |
},
|
| 236 |
"probabilities": probs
|
| 237 |
}
|
|
|
|
| 24 |
self.encoder = AutoModel.from_pretrained(model_name, config=self.config)
|
| 25 |
self.classifier = nn.Linear(self.config.hidden_size, num_labels)
|
| 26 |
|
| 27 |
+
def forward(self, input_ids, attention_mask):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 28 |
outputs = self.encoder(input_ids=input_ids, attention_mask=attention_mask)
|
| 29 |
pooled = outputs.last_hidden_state[:, 0]
|
| 30 |
logits = self.classifier(pooled)
|
| 31 |
+
return logits
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 32 |
|
| 33 |
# Load model and tokenizer
|
| 34 |
try:
|
|
|
|
| 119 |
# Get prediction from the model
|
| 120 |
with torch.no_grad():
|
| 121 |
# Check if using custom model or fallback
|
| 122 |
+
if hasattr(model, 'forward'):
|
| 123 |
# Our custom model
|
| 124 |
+
logits = model(
|
| 125 |
input_ids=inputs["input_ids"],
|
| 126 |
attention_mask=inputs["attention_mask"]
|
| 127 |
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 128 |
else:
|
| 129 |
# Fallback to standard model
|
| 130 |
outputs = model(**inputs)
|
| 131 |
logits = outputs.logits
|
|
|
|
| 132 |
|
| 133 |
# Process results
|
| 134 |
probs = torch.softmax(logits, dim=-1)[0].cpu().tolist()
|
|
|
|
| 137 |
# Map prediction to class name
|
| 138 |
classification = label_to_class.get(prediction, "Unknown")
|
| 139 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 140 |
return {
|
| 141 |
"pair_id": data.pair_id,
|
| 142 |
"test_case_1_name": data.test_case_1.name,
|
|
|
|
| 144 |
"similarity": {
|
| 145 |
"score": prediction,
|
| 146 |
"classification": classification,
|
|
|
|
| 147 |
},
|
| 148 |
"probabilities": probs
|
| 149 |
}
|