PowerMachine commited on
Commit
8c5ba73
·
verified ·
1 Parent(s): 078bbd2

Delete cnn_bigru/models/generator_verifier.py with huggingface_hub

Browse files
cnn_bigru/models/generator_verifier.py DELETED
@@ -1,344 +0,0 @@
1
- """generator_verifier.py — Gerador + Verificador + Anti-Alucinação.
2
-
3
- Implementa o pseudocódigo da seção 5-7 de dados.txt:
4
-
5
- - Generator: CNN-BiGRU encoder + GRU decoder com atenção Bahdanau
6
- - Verifier: CNN-BiGRU + classificador sigmoid (v ∈ [0,1])
7
- - Anti-hallucination layer: ativadores lógicos fuzzy de Łukasiewicz
8
- NOT(x) = 1 - sigmoid(x)
9
- AND(x,y) = relu(sigmoid(x) + sigmoid(y) - 1)
10
- OR(x,y) = min(1, sigmoid(x) + sigmoid(y))
11
- IMP(x,y) = min(1, 1 - sigmoid(x) + sigmoid(y))
12
- """
13
- from __future__ import annotations
14
-
15
- import logging
16
- import math
17
- from typing import Optional, Tuple
18
-
19
- import torch
20
- import torch.nn as nn
21
- import torch.nn.functional as F
22
-
23
- from .cooperative_bigru import CooperativeCNNBiGRU
24
-
25
- logger = logging.getLogger(__name__)
26
-
27
-
28
- # ============================================================================
29
- # Funções fuzzy de Łukasiewicz
30
- # ============================================================================
31
-
32
- def fuzzy_not(x: torch.Tensor) -> torch.Tensor:
33
- """NOT(x) = 1 - sigmoid(x)."""
34
- return 1.0 - torch.sigmoid(x)
35
-
36
-
37
- def fuzzy_and(x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
38
- """AND(x, y) = relu(sigmoid(x) + sigmoid(y) - 1)."""
39
- s = torch.sigmoid(x) + torch.sigmoid(y)
40
- return F.relu(s - 1.0)
41
-
42
-
43
- def fuzzy_or(x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
44
- """OR(x, y) = min(1, sigmoid(x) + sigmoid(y))."""
45
- s = torch.sigmoid(x) + torch.sigmoid(y)
46
- return torch.clamp(s, 0.0, 1.0)
47
-
48
-
49
- def fuzzy_imp(x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
50
- """IMP(x, y) = min(1, 1 - sigmoid(x) + sigmoid(y))."""
51
- s = 1.0 - torch.sigmoid(x) + torch.sigmoid(y)
52
- return torch.clamp(s, 0.0, 1.0)
53
-
54
-
55
- # ============================================================================
56
- # Camada Anti-Alucinação
57
- # ============================================================================
58
-
59
- class AntiHallucinationLayer(nn.Module):
60
- """Camada anti-alucinação com lógica fuzzy de Łukasiewicz.
61
-
62
- Recebe os logits [B, V] de um passo gerado e calcula um valor de verdade
63
- suave τ ∈ [0, 1] que indica o quão "lógico" é o passo.
64
-
65
- Implementação diferenciável:
66
- - Calcula sigmoid(logit) -> probabilidade soft
67
- - Aplica operadores fuzzy em pares de tokens mais prováveis
68
- - Combina resultados via IMP (implicação) e AND
69
- """
70
-
71
- def __init__(self, vocab_size: int, embed_dim: int = 32):
72
- super().__init__()
73
- self.proj = nn.Linear(vocab_size, embed_dim)
74
- self.combine = nn.Linear(embed_dim * 2, 1)
75
-
76
- def forward(self, logits: torch.Tensor) -> torch.Tensor:
77
- """
78
- Args:
79
- logits: [B, V] logits do passo gerado
80
-
81
- Returns:
82
- tau: [B, 1] valor de verdade suave em [0, 1]
83
- """
84
- # Probabilidades soft via sigmoid
85
- probs = torch.sigmoid(logits) # [B, V]
86
-
87
- # Top-2 tokens mais prováveis (diferenciável via topk)
88
- top2_vals, top2_idx = torch.topk(probs, k=min(2, probs.size(-1)), dim=-1) # [B, 2]
89
-
90
- # Logits dos top-2 tokens via gather
91
- top1_logit = logits.gather(1, top2_idx[:, 0:1]) # [B, 1]
92
- top2_logit = logits.gather(1, top2_idx[:, 1:2]) # [B, 1]
93
-
94
- # Operadores fuzzy de Łukasiewicz sobre os top-2 logits
95
- imp_val = fuzzy_imp(top1_logit, top2_logit) # [B, 1]
96
- and_val = fuzzy_and(top1_logit, top2_logit) # [B, 1]
97
- or_val = fuzzy_or(top1_logit, top2_logit) # [B, 1]
98
-
99
- # Projeção do vetor de logits completo para embedding
100
- proj = self.proj(logits) # [B, embed_dim]
101
-
102
- # Modulariza a projeção pelos valores fuzzy (broadcasting)
103
- fuzzy_factor = (imp_val + and_val + or_val) / 3.0 # [B, 1]
104
- modulated = proj * fuzzy_factor # [B, embed_dim]
105
-
106
- # Combina: [modulated, original_proj] -> [B, 2*embed_dim]
107
- combined = torch.cat((modulated, proj), dim=-1) # [B, 2*embed_dim]
108
- tau = torch.sigmoid(self.combine(combined)) # [B, 1]
109
- return tau
110
-
111
-
112
- # ============================================================================
113
- # Gerador (CNN-BiGRU encoder + GRU decoder com atenção)
114
- # ============================================================================
115
-
116
- class GeneratorCNNBiGRU(nn.Module):
117
- """Gerador: encoder CNN-BiGRU + decoder GRU com atenção Bahdanau.
118
-
119
- Args:
120
- vocab_size, embedding_dim, cnn_filters, gru_hidden: herdados do núcleo.
121
- max_proof_len: comprimento máximo da sequência gerada.
122
- """
123
-
124
- def __init__(
125
- self,
126
- vocab_size: int,
127
- embedding_dim: int = 64,
128
- cnn_filters: int = 64,
129
- gru_hidden: int = 64,
130
- n_heads: int = 4,
131
- dropout: float = 0.1,
132
- pad_idx: int = 1,
133
- max_proof_len: int = 32,
134
- ):
135
- super().__init__()
136
- self.pad_idx = pad_idx
137
- self.vocab_size = vocab_size
138
- self.gru_hidden = gru_hidden
139
- self.max_proof_len = max_proof_len
140
-
141
- # Encoder cooperativo
142
- self.encoder = CooperativeCNNBiGRU(
143
- vocab_size=vocab_size,
144
- embedding_dim=embedding_dim,
145
- cnn_filters=cnn_filters,
146
- gru_hidden=gru_hidden,
147
- n_heads=n_heads,
148
- dropout=dropout,
149
- pad_idx=pad_idx,
150
- )
151
-
152
- # Decoder GRU (unidirecional)
153
- self.dec_embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=pad_idx)
154
- self.dec_gru = nn.GRUCell(embedding_dim + gru_hidden * 2, gru_hidden)
155
-
156
- # Atenção Bahdanau
157
- self.attn_W = nn.Linear(gru_hidden * 2, gru_hidden, bias=False)
158
- self.attn_U = nn.Linear(gru_hidden, gru_hidden, bias=False)
159
- self.attn_v = nn.Linear(gru_hidden, 1, bias=False)
160
-
161
- # Projeção final
162
- self.out_proj = nn.Linear(gru_hidden + gru_hidden * 2 + embedding_dim, vocab_size)
163
-
164
- self.bos_id = 0 # placeholder, será sobrescrito
165
- self.eos_id = 2
166
-
167
- def _attention(
168
- self,
169
- dec_state: torch.Tensor, # [B, H]
170
- enc_out: torch.Tensor, # [B, T, 2H]
171
- enc_mask: torch.Tensor, # [B, T]
172
- ) -> torch.Tensor:
173
- """Atenção Bahdanau: retorna contexto [B, 2H]."""
174
- # score = v^T tanh(W enc + U dec)
175
- proj_enc = self.attn_W(enc_out) # [B, T, H]
176
- proj_dec = self.attn_U(dec_state).unsqueeze(1) # [B, 1, H]
177
- scores = self.attn_v(torch.tanh(proj_enc + proj_dec)).squeeze(-1) # [B, T]
178
- scores = scores.masked_fill(enc_mask == 0, -1e9)
179
- attn = F.softmax(scores, dim=-1) # [B, T]
180
- ctx = torch.bmm(attn.unsqueeze(1), enc_out).squeeze(1) # [B, 2H]
181
- return ctx
182
-
183
- def forward(
184
- self,
185
- input_ids_a: torch.Tensor,
186
- input_ids_b: torch.Tensor,
187
- target_proof: Optional[torch.Tensor] = None,
188
- bos_id: int = 0,
189
- eos_id: int = 2,
190
- max_len: Optional[int] = None,
191
- ) -> dict:
192
- """
193
- Args:
194
- input_ids_a, input_ids_b: [B, T] entradas (axiomas + conjectura)
195
- target_proof: [B, T_proof] para teacher forcing (treino)
196
- bos_id, eos_id: IDs dos tokens especiais
197
-
198
- Returns:
199
- dict com:
200
- logits: [B, T_proof, V]
201
- enc_out: [B, T, 2H]
202
- dec_state: [B, H]
203
- """
204
- self.bos_id = bos_id
205
- self.eos_id = eos_id
206
-
207
- # 1. Encoder
208
- enc = self.encoder(input_ids_a, input_ids_b, return_sequences=False)
209
- # Para o decoder, usamos o fused como representação final e repetimos ao longo do tempo
210
- fused = enc["fused"] # [B, 256]
211
- # enc_out "virtual": usamos a mask do stream A para o tempo
212
- mask_A = enc["mask_A"]
213
- T_enc = mask_A.size(1)
214
- enc_out = fused.unsqueeze(1).expand(-1, T_enc, -1) # [B, T, 256]
215
- # Para atenção, usamos a primeira metade como "enc_out" e a mask do stream A
216
- # Enc_out tem 2H = 256, então H = 128
217
- enc_mask = mask_A
218
-
219
- B = input_ids_a.size(0)
220
- device = input_ids_a.device
221
-
222
- # 2. Inicialização do decoder
223
- # Estado inicial: projeção do fused
224
- h = fused[:, :self.gru_hidden] # [B, H] (primeira metade do fused)
225
-
226
- # 3. Token inicial = BOS
227
- if target_proof is not None:
228
- # Teacher forcing
229
- T_dec = target_proof.size(1)
230
- inputs = target_proof # [B, T_dec]
231
- else:
232
- T_dec = max_len or self.max_proof_len
233
- inputs = None
234
-
235
- logits_list = []
236
- cur_token = torch.full((B, 1), bos_id, dtype=torch.long, device=device)
237
-
238
- for t in range(T_dec):
239
- if target_proof is not None:
240
- tok = target_proof[:, t] # [B]
241
- else:
242
- tok = cur_token.squeeze(1)
243
-
244
- emb = self.dec_embedding(tok) # [B, D]
245
- ctx = self._attention(h, enc_out, enc_mask) # [B, 2H]
246
- gru_input = torch.cat((emb, ctx), dim=-1) # [B, D+2H]
247
- h = self.dec_gru(gru_input, h) # [B, H]
248
-
249
- # Logits
250
- logit_input = torch.cat((h, ctx, emb), dim=-1) # [B, H+2H+D]
251
- logit = self.out_proj(logit_input) # [B, V]
252
- logits_list.append(logit)
253
-
254
- if target_proof is None:
255
- cur_token = logit.argmax(dim=-1, keepdim=True)
256
-
257
- logits = torch.stack(logits_list, dim=1) # [B, T_dec, V]
258
-
259
- return {
260
- "logits": logits,
261
- "enc_out": enc_out,
262
- "dec_state": h,
263
- }
264
-
265
-
266
- # ============================================================================
267
- # Verificador (CNN-BiGRU + classificador sigmoid)
268
- # ============================================================================
269
-
270
- class VerifierCNNBiGRU(nn.Module):
271
- """Verificador: CNN-BiGRU + classificador sigmoid (v ∈ [0,1]).
272
-
273
- Recebe premissas (concat de axiomas + conjectura) e um passo hipotético,
274
- e retorna uma probabilidade de o passo estar correto.
275
- """
276
-
277
- def __init__(
278
- self,
279
- vocab_size: int,
280
- embedding_dim: int = 64,
281
- cnn_filters: int = 64,
282
- gru_hidden: int = 64,
283
- n_heads: int = 4,
284
- dropout: float = 0.1,
285
- pad_idx: int = 1,
286
- ):
287
- super().__init__()
288
- self.pad_idx = pad_idx
289
- self.gru_hidden = gru_hidden
290
-
291
- self.encoder = CooperativeCNNBiGRU(
292
- vocab_size=vocab_size,
293
- embedding_dim=embedding_dim,
294
- cnn_filters=cnn_filters,
295
- gru_hidden=gru_hidden,
296
- n_heads=n_heads,
297
- dropout=dropout,
298
- pad_idx=pad_idx,
299
- )
300
-
301
- # Classificador sigmoid
302
- self.classifier = nn.Sequential(
303
- nn.Linear(gru_hidden * 2 * 2, gru_hidden), # 256*2 = 512 (fused A+B do premissa+passo)
304
- nn.ReLU(),
305
- nn.Dropout(dropout),
306
- nn.Linear(gru_hidden, 1),
307
- )
308
-
309
- def forward(
310
- self,
311
- premissas_a: torch.Tensor,
312
- premissas_b: torch.Tensor,
313
- passo_a: torch.Tensor,
314
- passo_b: torch.Tensor,
315
- ) -> torch.Tensor:
316
- """
317
- Args:
318
- premissas_a, premissas_b: [B, T] (contexto da prova até o momento)
319
- passo_a, passo_b: [B, T_p] (passo gerado a ser validado)
320
-
321
- Returns:
322
- v: [B, 1] probabilidade de o passo estar correto
323
- """
324
- # Encode premissas
325
- enc_pre = self.encoder(premissas_a, premissas_b)
326
- feat_pre = enc_pre["fused"] # [B, 256]
327
-
328
- # Encode passo
329
- enc_passo = self.encoder(passo_a, passo_b)
330
- feat_passo = enc_passo["fused"] # [B, 256]
331
-
332
- # Concatena e classifica
333
- combined = torch.cat((feat_pre, feat_passo), dim=-1) # [B, 512]
334
- logit = self.classifier(combined) # [B, 1]
335
- v = torch.sigmoid(logit) # [B, 1]
336
- return v
337
-
338
-
339
- __all__ = [
340
- "fuzzy_not", "fuzzy_and", "fuzzy_or", "fuzzy_imp",
341
- "AntiHallucinationLayer",
342
- "GeneratorCNNBiGRU",
343
- "VerifierCNNBiGRU",
344
- ]