(self, token: interface.Token)
| 22 | self.total_len = 0 |
| 23 | |
| 24 | def add(self, token: interface.Token): |
| 25 | self.tokens.append(token) |
| 26 | self.token_lens.append(len(token)) |
| 27 | self.total_len += len(token) |
| 28 | while True: |
| 29 | if not self.tokens: |
| 30 | break |
| 31 | head_len = self.token_lens[0] |
| 32 | if self.total_len - head_len >= self.longest_stop_len: |
| 33 | token = self.tokens.popleft() |
| 34 | self.token_lens.popleft() |
| 35 | self.total_len -= head_len |
| 36 | yield token |
| 37 | else: |
| 38 | break |
| 39 | |
| 40 | def finish(self, reason: Literal["length", "eos_token", "stop_sequence"]) -> Iterable[interface.Token]: |
| 41 | if reason == "stop_sequence": |
no outgoing calls
no test coverage detected