mirror of
https://github.com/modelscope/FunASR
synced 2025-09-15 14:48:36 +08:00
131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
from typing import Any
|
|
from typing import List
|
|
from typing import Tuple
|
|
|
|
import torch
|
|
import torch.nn as nn
|
|
|
|
from funasr.modules.embedding import PositionalEncoding
|
|
from funasr.modules.embedding import SinusoidalPositionEncoder
|
|
#from funasr.models.encoder.transformer_encoder import TransformerEncoder as Encoder
|
|
from funasr.punctuation.sanm_encoder import SANMEncoder as Encoder
|
|
#from funasr.modules.mask import subsequent_n_mask
|
|
from funasr.punctuation.abs_model import AbsPunctuation
|
|
|
|
|
|
class TargetDelayTransformer(AbsPunctuation):
|
|
|
|
def __init__(
|
|
self,
|
|
vocab_size: int,
|
|
punc_size: int,
|
|
pos_enc: str = None,
|
|
embed_unit: int = 128,
|
|
att_unit: int = 256,
|
|
head: int = 2,
|
|
unit: int = 1024,
|
|
layer: int = 4,
|
|
dropout_rate: float = 0.5,
|
|
):
|
|
super().__init__()
|
|
if pos_enc == "sinusoidal":
|
|
# pos_enc_class = PositionalEncoding
|
|
pos_enc_class = SinusoidalPositionEncoder
|
|
elif pos_enc is None:
|
|
|
|
def pos_enc_class(*args, **kwargs):
|
|
return nn.Sequential() # indentity
|
|
|
|
else:
|
|
raise ValueError(f"unknown pos-enc option: {pos_enc}")
|
|
|
|
self.embed = nn.Embedding(vocab_size, embed_unit)
|
|
self.encoder = Encoder(
|
|
input_size=embed_unit,
|
|
output_size=att_unit,
|
|
attention_heads=head,
|
|
linear_units=unit,
|
|
num_blocks=layer,
|
|
dropout_rate=dropout_rate,
|
|
input_layer="pe",
|
|
# pos_enc_class=pos_enc_class,
|
|
padding_idx=0,
|
|
)
|
|
self.decoder = nn.Linear(att_unit, punc_size)
|
|
|
|
|
|
# def _target_mask(self, ys_in_pad):
|
|
# ys_mask = ys_in_pad != 0
|
|
# m = subsequent_n_mask(ys_mask.size(-1), 5, device=ys_mask.device).unsqueeze(0)
|
|
# return ys_mask.unsqueeze(-2) & m
|
|
|
|
def forward(self, input: torch.Tensor, text_lengths: torch.Tensor) -> Tuple[torch.Tensor, None]:
|
|
"""Compute loss value from buffer sequences.
|
|
|
|
Args:
|
|
input (torch.Tensor): Input ids. (batch, len)
|
|
hidden (torch.Tensor): Target ids. (batch, len)
|
|
|
|
"""
|
|
x = self.embed(input)
|
|
# mask = self._target_mask(input)
|
|
h, _, _ = self.encoder(x, text_lengths)
|
|
y = self.decoder(h)
|
|
return y, None
|
|
|
|
def with_vad(self):
|
|
return False
|
|
|
|
def score(self, y: torch.Tensor, state: Any, x: torch.Tensor) -> Tuple[torch.Tensor, Any]:
|
|
"""Score new token.
|
|
|
|
Args:
|
|
y (torch.Tensor): 1D torch.int64 prefix tokens.
|
|
state: Scorer state for prefix tokens
|
|
x (torch.Tensor): encoder feature that generates ys.
|
|
|
|
Returns:
|
|
tuple[torch.Tensor, Any]: Tuple of
|
|
torch.float32 scores for next token (vocab_size)
|
|
and next state for ys
|
|
|
|
"""
|
|
y = y.unsqueeze(0)
|
|
h, _, cache = self.encoder.forward_one_step(self.embed(y), self._target_mask(y), cache=state)
|
|
h = self.decoder(h[:, -1])
|
|
logp = h.log_softmax(dim=-1).squeeze(0)
|
|
return logp, cache
|
|
|
|
def batch_score(self, ys: torch.Tensor, states: List[Any], xs: torch.Tensor) -> Tuple[torch.Tensor, List[Any]]:
|
|
"""Score new token batch.
|
|
|
|
Args:
|
|
ys (torch.Tensor): torch.int64 prefix tokens (n_batch, ylen).
|
|
states (List[Any]): Scorer states for prefix tokens.
|
|
xs (torch.Tensor):
|
|
The encoder feature that generates ys (n_batch, xlen, n_feat).
|
|
|
|
Returns:
|
|
tuple[torch.Tensor, List[Any]]: Tuple of
|
|
batchfied scores for next token with shape of `(n_batch, vocab_size)`
|
|
and next state list for ys.
|
|
|
|
"""
|
|
# merge states
|
|
n_batch = len(ys)
|
|
n_layers = len(self.encoder.encoders)
|
|
if states[0] is None:
|
|
batch_state = None
|
|
else:
|
|
# transpose state of [batch, layer] into [layer, batch]
|
|
batch_state = [torch.stack([states[b][i] for b in range(n_batch)]) for i in range(n_layers)]
|
|
|
|
# batch decoding
|
|
h, _, states = self.encoder.forward_one_step(self.embed(ys), self._target_mask(ys), cache=batch_state)
|
|
h = self.decoder(h[:, -1])
|
|
logp = h.log_softmax(dim=-1)
|
|
|
|
# transpose state of [layer, batch] into [batch, layer]
|
|
state_list = [[states[i][b] for i in range(n_layers)] for b in range(n_batch)]
|
|
return logp, state_list
|