PowerMachine commited on
Commit
4147e80
·
verified ·
1 Parent(s): fe179d6

Delete cnn_bigru/utils/semantic_embeddings.py with huggingface_hub

Browse files
cnn_bigru/utils/semantic_embeddings.py DELETED
@@ -1,144 +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
- ) -> torch.Tensor:
106
- """Returns L2-normalized embeddings [B, out_dim]."""
107
- if attention_mask is None:
108
- attention_mask = (input_ids != self.pad_idx).float()
109
-
110
- emb = self.token_embedding(input_ids) # [B, T, D]
111
- mask = attention_mask.unsqueeze(-1) # [B, T, 1]
112
-
113
- # Mean pooling com máscara
114
- summed = (emb * mask).sum(dim=1)
115
- counts = mask.sum(dim=1).clamp(min=1.0)
116
- pooled = summed / counts # [B, D]
117
-
118
- proj = self.proj(pooled) # [B, out_dim]
119
- proj = self.out_norm(proj)
120
-
121
- # L2 normalize
122
- norm = proj.norm(p=2, dim=-1, keepdim=True).clamp(min=1e-8)
123
- emb_norm = proj / norm
124
-
125
- # Information Bottleneck regularization (beta-VAE style):
126
- # penaliza variância excessiva (encoraja representação comprimida)
127
- if self.training and self.beta > 0:
128
- var = emb_norm.var(dim=0).mean()
129
- ib_loss = self.beta * var
130
- # Anexa ao atributo para coleta externa
131
- setattr(emb_norm, "ib_loss", ib_loss)
132
-
133
- return emb_norm
134
-
135
- def similarity(self, a: torch.Tensor, b: torch.Tensor) -> torch.Tensor:
136
- """Similaridade coseno (a já está normalizado)."""
137
- return torch.matmul(a, b.T)
138
-
139
-
140
- __all__ = [
141
- "sinkhorn_knopp",
142
- "elastic_wmd",
143
- "SemanticEmbedder",
144
- ]