sockeye/inference.py [109:206]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - return max_input_len, get_max_output_length BeamHistory = Dict[str, List] Tokens = List[str] TokenIds = List[List[int]] # each token id may contain multiple factors SentenceId = Union[int, str] @dataclass class TranslatorInput: """ Object required by Translator.translate(). If not None, `pass_through_dict` is an arbitrary dictionary instantiated from a JSON object via `make_input_from_dict()`, and it contains extra fields found in an input JSON object. If `--output-type json` is selected, all such fields that are not fields used or changed by Sockeye will be included in the output JSON object. This provides a mechanism for passing fields through the call to Sockeye. """ sentence_id: SentenceId tokens: Tokens factors: Optional[List[Tokens]] = None restrict_lexicon: Optional[lexicon.TopKLexicon] = None constraints: Optional[List[Tokens]] = None avoid_list: Optional[List[Tokens]] = None pass_through_dict: Optional[Dict] = None def __str__(self): return 'TranslatorInput(%s, %s, factors=%s, constraints=%s, avoid=%s)' \ % (self.sentence_id, self.tokens, self.factors, self.constraints, self.avoid_list) def __len__(self): return len(self.tokens) @property def num_factors(self) -> int: """ Returns the number of factors of this instance. """ return 1 + (0 if not self.factors else len(self.factors)) def chunks(self, chunk_size: int) -> Generator['TranslatorInput', None, None]: """ Takes a TranslatorInput (itself) and yields TranslatorInputs for chunks of size chunk_size. :param chunk_size: The maximum size of a chunk. :return: A generator of TranslatorInputs, one for each chunk created. """ if len(self.tokens) > chunk_size and self.constraints is not None: logger.warning( 'Input %s has length (%d) that exceeds max input length (%d), ' 'triggering internal splitting. Placing all target-side constraints ' 'with the first chunk, which is probably wrong.', self.sentence_id, len(self.tokens), chunk_size) for chunk_id, i in enumerate(range(0, len(self), chunk_size)): factors = [factor[i:i + chunk_size] for factor in self.factors] if self.factors is not None else None # Constrained decoding is not supported for chunked TranslatorInputs. As a fall-back, constraints are # assigned to the first chunk constraints = self.constraints if chunk_id == 0 else None pass_through_dict = copy.deepcopy(self.pass_through_dict) \ if (chunk_id == 0 and self.pass_through_dict is not None) else None yield TranslatorInput(sentence_id=self.sentence_id, tokens=self.tokens[i:i + chunk_size], factors=factors, restrict_lexicon=self.restrict_lexicon, constraints=constraints, avoid_list=self.avoid_list, pass_through_dict=pass_through_dict) def with_eos(self) -> 'TranslatorInput': """ :return: A new translator input with EOS appended to the tokens and factors. """ return TranslatorInput(sentence_id=self.sentence_id, tokens=self.tokens + [C.EOS_SYMBOL], factors=[factor + [C.EOS_SYMBOL] for factor in self.factors] if self.factors is not None else None, restrict_lexicon=self.restrict_lexicon, constraints=self.constraints, avoid_list=self.avoid_list, pass_through_dict=self.pass_through_dict) class BadTranslatorInput(TranslatorInput): def __init__(self, sentence_id: SentenceId, tokens: Tokens) -> None: super().__init__(sentence_id=sentence_id, tokens=tokens, factors=None) def _bad_input(sentence_id: SentenceId, reason: str = '') -> BadTranslatorInput: logger.warning("Bad input (%s): '%s'. Will return empty output.", sentence_id, reason.strip()) return BadTranslatorInput(sentence_id=sentence_id, tokens=[]) def make_input_from_plain_string(sentence_id: SentenceId, string: str) -> TranslatorInput: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - sockeye/inference_pt.py [110:207]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - return max_input_len, get_max_output_length BeamHistory = Dict[str, List] Tokens = List[str] TokenIds = List[List[int]] # each token id may contain multiple factors SentenceId = Union[int, str] @dataclass class TranslatorInput: """ Object required by Translator.translate(). If not None, `pass_through_dict` is an arbitrary dictionary instantiated from a JSON object via `make_input_from_dict()`, and it contains extra fields found in an input JSON object. If `--output-type json` is selected, all such fields that are not fields used or changed by Sockeye will be included in the output JSON object. This provides a mechanism for passing fields through the call to Sockeye. """ sentence_id: SentenceId tokens: Tokens factors: Optional[List[Tokens]] = None restrict_lexicon: Optional[lexicon.TopKLexicon] = None constraints: Optional[List[Tokens]] = None avoid_list: Optional[List[Tokens]] = None pass_through_dict: Optional[Dict] = None def __str__(self): return 'TranslatorInput(%s, %s, factors=%s, constraints=%s, avoid=%s)' \ % (self.sentence_id, self.tokens, self.factors, self.constraints, self.avoid_list) def __len__(self): return len(self.tokens) @property def num_factors(self) -> int: """ Returns the number of factors of this instance. """ return 1 + (0 if not self.factors else len(self.factors)) def chunks(self, chunk_size: int) -> Generator['TranslatorInput', None, None]: """ Takes a TranslatorInput (itself) and yields TranslatorInputs for chunks of size chunk_size. :param chunk_size: The maximum size of a chunk. :return: A generator of TranslatorInputs, one for each chunk created. """ if len(self.tokens) > chunk_size and self.constraints is not None: logger.warning( 'Input %s has length (%d) that exceeds max input length (%d), ' 'triggering internal splitting. Placing all target-side constraints ' 'with the first chunk, which is probably wrong.', self.sentence_id, len(self.tokens), chunk_size) for chunk_id, i in enumerate(range(0, len(self), chunk_size)): factors = [factor[i:i + chunk_size] for factor in self.factors] if self.factors is not None else None # Constrained decoding is not supported for chunked TranslatorInputs. As a fall-back, constraints are # assigned to the first chunk constraints = self.constraints if chunk_id == 0 else None pass_through_dict = copy.deepcopy(self.pass_through_dict) \ if (chunk_id == 0 and self.pass_through_dict is not None) else None yield TranslatorInput(sentence_id=self.sentence_id, tokens=self.tokens[i:i + chunk_size], factors=factors, restrict_lexicon=self.restrict_lexicon, constraints=constraints, avoid_list=self.avoid_list, pass_through_dict=pass_through_dict) def with_eos(self) -> 'TranslatorInput': """ :return: A new translator input with EOS appended to the tokens and factors. """ return TranslatorInput(sentence_id=self.sentence_id, tokens=self.tokens + [C.EOS_SYMBOL], factors=[factor + [C.EOS_SYMBOL] for factor in self.factors] if self.factors is not None else None, restrict_lexicon=self.restrict_lexicon, constraints=self.constraints, avoid_list=self.avoid_list, pass_through_dict=self.pass_through_dict) class BadTranslatorInput(TranslatorInput): def __init__(self, sentence_id: SentenceId, tokens: Tokens) -> None: super().__init__(sentence_id=sentence_id, tokens=tokens, factors=None) def _bad_input(sentence_id: SentenceId, reason: str = '') -> BadTranslatorInput: logger.warning("Bad input (%s): '%s'. Will return empty output.", sentence_id, reason.strip()) return BadTranslatorInput(sentence_id=sentence_id, tokens=[]) def make_input_from_plain_string(sentence_id: SentenceId, string: str) -> TranslatorInput: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -