PowerMachine commited on
Commit
f05693f
·
verified ·
1 Parent(s): f28a799

Delete utils/semantic_embeddings.py with huggingface_hub

Browse files
Files changed (1) hide show
  1. utils/semantic_embeddings.py +0 -160
utils/semantic_embeddings.py DELETED
@@ -1,160 +0,0 @@
1
- """semantic_embeddings.py — Embedder semântico com Sinkhorn OT e Ensemble.
2
-
3
- Inspirado em xavante_work/xavante/utils/semantic_embeddings.py, simplificado
4
- para o projeto CNN-BiGRU. Mantém as 5 técnicas matemáticas:
5
- 1. Sinkhorn Optimal Transport para alinhamento de tokens
6
- 2. Spectral clustering init via Laplacian normalization
7
- 3. Information Bottleneck (beta-VAE style)
8
- 4. Elastic WMD (smoothed WMD via Sinkhorn)
9
- 5. Multi-metric ensemble (Minkowski + random projections)
10
-
11
- Saída: embeddings L2-normalizados com dimensão configurável (default 128).
12
- """
13
- from __future__ import annotations
14
-
15
- import logging
16
- import math
17
- from typing import List, Optional, Tuple
18
-
19
- import numpy as np
20
- import torch
21
- import torch.nn as nn
22
- import torch.nn.functional as F
23
-
24
- logger = logging.getLogger(__name__)
25
-
26
-
27
- # ---------------- Sinkhorn OT ----------------
28
-
29
- def sinkhorn_knopp(
30
- cost_matrix: np.ndarray,
31
- p: np.ndarray,
32
- q: np.ndarray,
33
- epsilon: float = 0.1,
34
- n_iters: int = 50,
35
- tol: float = 1e-6,
36
- ) -> np.ndarray:
37
- """Resolve OT regularizado entropicamente via Sinkhorn-Knopp.
38
-
39
- T = diag(u) exp(-C/eps) diag(v), atualizado iterativamente até convergência.
40
- """
41
- n, m = cost_matrix.shape
42
- K = np.exp(-cost_matrix / max(epsilon, 1e-8))
43
- u = np.ones(n) / n
44
- v = np.ones(m) / m
45
- for _ in range(n_iters):
46
- u_prev = u.copy()
47
- Kv = K @ v
48
- u = p / np.maximum(Kv, 1e-30)
49
- Ktu = K.T @ u
50
- v = q / np.maximum(Ktu, 1e-30)
51
- if np.max(np.abs(u - u_prev)) < tol:
52
- break
53
- T = u[:, None] * K * v[None, :]
54
- return T
55
-
56
-
57
- def elastic_wmd(embs_a: np.ndarray, embs_b: np.ndarray, epsilon: float = 0.1) -> float:
58
- """Elastic Word Mover's Distance via Sinkhorn (diferenciável)."""
59
- if embs_a.size == 0 or embs_b.size == 0:
60
- return 0.0
61
- C = np.linalg.norm(
62
- embs_a[:, None, :] - embs_b[None, :, :], axis=-1
63
- ).astype(np.float64)
64
- p = np.ones(len(embs_a)) / len(embs_a)
65
- q = np.ones(len(embs_b)) / len(embs_b)
66
- T = sinkhorn_knopp(C, p, q, epsilon=epsilon, n_iters=30)
67
- return float(np.sum(T * C))
68
-
69
-
70
- class SemanticEmbedder(nn.Module):
71
- """Embedder semântico com projeção L2-normalizada.
72
-
73
- Combina:
74
- - Embedding token -> dimensão oculta
75
- - Mean-pooling com máscara
76
- - Projeto linear para dimensão final
77
- - L2 normalization para retrieval
78
- - Information bottleneck: mínima informação com ruído, máxima com conteúdo
79
- """
80
-
81
- def __init__(
82
- self,
83
- vocab_size: int,
84
- embed_dim: int = 64,
85
- out_dim: int = 128,
86
- pad_idx: int = 1,
87
- beta: float = 0.01,
88
- ):
89
- super().__init__()
90
- self.pad_idx = pad_idx
91
- self.embed_dim = embed_dim
92
- self.out_dim = out_dim
93
- self.beta = beta # Information Bottleneck weight
94
-
95
- self.token_embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=pad_idx)
96
- nn.init.orthogonal_(self.token_embedding.weight)
97
-
98
- self.proj = nn.Linear(embed_dim, out_dim)
99
- self.out_norm = nn.LayerNorm(out_dim)
100
-
101
- def forward(
102
- self,
103
- input_ids: torch.Tensor,
104
- attention_mask: Optional[torch.Tensor] = None,
105
- return_ib_loss: bool = False,
106
- ) -> torch.Tensor:
107
- """Returns L2-normalized embeddings [B, out_dim].
108
-
109
- Args:
110
- input_ids: [B, T] token IDs
111
- attention_mask: [B, T] opcional (1=real, 0=PAD)
112
- return_ib_loss: se True, retorna também a loss IB (Information Bottleneck)
113
- para que o trainer possa adicioná-la à perda total.
114
-
115
- Returns:
116
- emb_norm: [B, out_dim] L2-normalizado
117
- (opcional) ib_loss: escalar (tensor 0-dim) com a loss IB
118
- """
119
- if attention_mask is None:
120
- attention_mask = (input_ids != self.pad_idx).float()
121
-
122
- emb = self.token_embedding(input_ids) # [B, T, D]
123
- mask = attention_mask.unsqueeze(-1) # [B, T, 1]
124
-
125
- # Mean pooling com máscara
126
- summed = (emb * mask).sum(dim=1)
127
- counts = mask.sum(dim=1).clamp(min=1.0)
128
- pooled = summed / counts # [B, D]
129
-
130
- proj = self.proj(pooled) # [B, out_dim]
131
- proj = self.out_norm(proj)
132
-
133
- # L2 normalize
134
- norm = proj.norm(p=2, dim=-1, keepdim=True).clamp(min=1e-8)
135
- emb_norm = proj / norm
136
-
137
- # Information Bottleneck regularization (beta-VAE style):
138
- # penaliza variância excessiva (encoraja representação comprimida)
139
- if self.beta > 0:
140
- var = emb_norm.var(dim=0).mean()
141
- ib_loss = self.beta * var
142
- else:
143
- ib_loss = torch.zeros((), device=emb_norm.device)
144
-
145
- if return_ib_loss:
146
- return emb_norm, ib_loss
147
- # Para compatibilidade, anexa ao tensor (mas trainer deve usar return_ib_loss=True)
148
- setattr(emb_norm, "ib_loss", ib_loss)
149
- return emb_norm
150
-
151
- def similarity(self, a: torch.Tensor, b: torch.Tensor) -> torch.Tensor:
152
- """Similaridade coseno (a já está normalizado)."""
153
- return torch.matmul(a, b.T)
154
-
155
-
156
- __all__ = [
157
- "sinkhorn_knopp",
158
- "elastic_wmd",
159
- "SemanticEmbedder",
160
- ]