Skip to content

base

BatchAbandoned

Bases: RuntimeError

The caller holding a batch died before it could be dispatched.

Source code in genlm/backend/llm/base.py 
12
13
class BatchAbandoned(RuntimeError):
    """The caller holding a batch died before it could be dispatched."""

batch_abandoned(exc)

The failure handed to callers whose batch leader died.

Never the cause itself: a CancelledError given to a caller who never asked for one leaves their task cancelled and skips their except Exception.

Source code in genlm/backend/llm/base.py 
16
17
18
19
20
21
22
23
24
def batch_abandoned(exc):
    """The failure handed to callers whose batch leader died.

    Never the cause itself: a `CancelledError` given to a caller who never asked
    for one leaves their task cancelled and skips their `except Exception`.
    """
    abandoned = BatchAbandoned(f"batch leader did not survive it: {exc!r}")
    abandoned.__cause__ = exc
    return abandoned

fail_futures(entries, exc)

Fail each entry's future (its last element) with exc.

Source code in genlm/backend/llm/base.py 
27
28
29
30
31
32
def fail_futures(entries, exc):
    """Fail each entry's future (its last element) with `exc`."""
    for entry in entries:
        future = entry[-1]
        if not future.done():
            future.set_exception(exc)

AsyncLM

Bases: ABC

Abstract base class for asynchronous language models.

This class provides an interface for language models that can generate token probabilities asynchronously. It handles tokenization and vocabulary management.

Parameters:

Name Type Description Default
tokenizer

A Hugging Face tokenizer instance compatible with the language model

 required
Source code in genlm/backend/llm/base.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
class AsyncLM(ABC):
    """Abstract base class for asynchronous language models.

    This class provides an interface for language models that can generate token probabilities
    asynchronously. It handles tokenization and vocabulary management.

    Args:
        tokenizer: A Hugging Face tokenizer instance compatible with the language model
    """

    def __init__(self, tokenizer):
        self.tokenizer = tokenizer
        self.byte_vocab, self.str_vocab = decode_vocab(self.tokenizer)
        self._batch_queue = []
        self._batch_armed = False

    async def _join_batch(self, entries, *, timeout=0.0):
        """Join the batch of concurrent queries on this event loop.

        The first caller leads: it holds the batch open until an event-loop pass adds
        no new query, then receives the drained batch; every other caller receives
        `None`. The leader must evaluate the batch in its own coroutine, never a
        background task. If the leader dies first, every other queued future fails
        with `BatchAbandoned`.

        Args:
            entries (list): Queries to add to the batch, each a tuple ending in its future.
            timeout (float, optional): Seconds to wait once for late callers.
                Defaults to 0.0.

        Returns:
            (list | None): The drained batch for the leader, `None` for every other caller.
        """
        self._batch_queue.extend(entries)
        if self._batch_armed:
            return None
        self._batch_armed = True
        try:
            waited = not timeout
            while True:
                n = len(self._batch_queue)
                await asyncio.sleep(0)
                if len(self._batch_queue) > n:
                    continue
                if waited:
                    break
                waited = True
                await asyncio.sleep(timeout)
            batch, self._batch_queue = self._batch_queue, []
            return batch
        except BaseException as exc:
            batch, self._batch_queue = self._batch_queue, []
            # Skip this caller's own entries: it is unwinding, so an exception set
            # on them would go unretrieved.
            mine = {id(e) for e in entries}
            fail_futures([e for e in batch if id(e) not in mine], batch_abandoned(exc))
            raise
        finally:
            self._batch_armed = False

    @abstractmethod
    async def next_token_logprobs(self, token_ids, lora_name=None):
        """Request log probabilities of next token asynchronously.

        Args:
            token_ids (list[int]): A list of token IDs representing the prompt.
            lora_name (str, optional): Name of the LoRA adapter to use. Defaults to None (the base model).

        Returns:
            (torch.Tensor): Normalized log probability tensor.
        """
        pass

    @abstractmethod
    def next_token_logprobs_sync(self, token_ids, lora_name=None):
        """Request log probabilities of next token synchronously.

        Args:
            token_ids (list[int]): A list of token IDs representing the prompt.
            lora_name (str, optional): Name of the LoRA adapter to use. Defaults to None (the base model).

        Returns:
            (torch.Tensor): Normalized log probability tensor.
        """
        pass

    async def batch_next_token_logprobs(self, token_ids_list, lora_name=None):
        """Batch request log probabilities for multiple token sequences asynchronously.

        Args:
            token_ids_list (list[list[int]]): A list of token ID lists.
            lora_name (str, optional): Name of the LoRA adapter to use. Defaults to None (the base model).

        Returns:
            (torch.Tensor): A tensor of log probability tensors.
        """
        logprobs = await asyncio.gather(
            *[
                self.next_token_logprobs(token_ids, lora_name=lora_name)
                for token_ids in token_ids_list
            ]
        )

        return torch.stack(logprobs)

    def batch_next_token_logprobs_sync(self, token_ids_list, lora_name=None):
        """Batch request log probabilities for multiple token sequences synchronously.

        Args:
            token_ids_list (list[list[int]]): A list of token ID lists.
            lora_name (str, optional): Name of the LoRA adapter to use. Defaults to None (the base model).

        Returns:
            (torch.Tensor): A tensor of log probability tensors.
        """
        return torch.stack(
            [
                self.next_token_logprobs_sync(token_ids, lora_name=lora_name)
                for token_ids in token_ids_list
            ]
        )

    def add_new_lora(self, lora_path, lora_name):
        """Load a LoRA adapter into the base model.

        Loading under an existing name replaces that adapter. Select it per call with `lora_name`.

        Args:
            lora_path (str): Path to the adapter weights directory or identifier in HuggingFace's model hub.
            lora_name (str): Name to assign to the loaded adapter.

        """
        raise NotImplementedError(
            "add_new_lora must be implemented by subclasses"
        )  # pragma: no cover

    def remove_lora(self, lora_name):
        """Remove a previously loaded LoRA adapter.

        Args:
            lora_name (str): Name of the adapter to remove.

        """
        raise NotImplementedError(
            "remove_lora must be implemented by subclasses"
        )  # pragma: no cover

    def set_lora(self, lora_path, lora_name):
        """Unsupported: adapter selection is per-request, via ``lora_name=``."""
        raise RuntimeError(
            "set_lora() was removed: there is no active-adapter global anymore. "
            "Pass lora_name= per call (next_token_logprobs/sample/...)."
        )

    def clear_lora(self):
        """Unsupported: ``lora_name=None`` (the default) is the base model."""
        raise RuntimeError(
            "clear_lora() was removed: omit lora_name (None = base model)."
        )

    def clear_cache(self):
        """Clear any caches used by the language model. No-op in base class."""
        pass  # pragma: no cover

    async def sample(
        self,
        prompt_token_ids,
        max_tokens,
        eos_token_ids,
        temperature=1.0,
        seed=None,
        lora_name=None,
    ):
        """Sample from the language model.

        Args:
            prompt_token_ids (list[int]): The token IDs of the prompt.
            eos_token_ids (list[int]): The token IDs of the end-of-sequence tokens.
            temperature (float, optional): The temperature to use to rescale the logits. Defaults to 1.0.
            max_tokens (int): The maximum number of tokens to generate.
            seed (int, optional): The seed for the random number generator. Defaults to None.
            lora_name (str, optional): Name of the LoRA adapter to use. Defaults to None (the base model).

        Returns:
            (list[int]): The sampled token IDs.
        """
        if seed is not None:
            generator = torch.Generator()
            generator.manual_seed(seed)
        else:
            generator = None

        generated_token_ids = []
        for _ in range(max_tokens):
            logprobs = await self.next_token_logprobs(
                prompt_token_ids + generated_token_ids, lora_name=lora_name
            )
            probs = torch.softmax(logprobs / temperature, dim=-1)
            next_token_id = torch.multinomial(
                probs.cpu() if seed is not None else probs,
                num_samples=1,
                generator=generator,
            ).item()
            if next_token_id in eos_token_ids:
                break
            generated_token_ids.append(next_token_id)

        return generated_token_ids

    async def batch_sample(
        self,
        prompt_token_ids_list,
        max_tokens,
        eos_token_ids,
        temperature=1.0,
        seed=None,
        lora_name=None,
    ):
        """Batch sample from the language model.

        Args:
            prompt_token_ids_list (list[list[int]]): The token IDs of the prompts.
            max_tokens (int): The maximum number of tokens to generate.
            eos_token_ids (list[int]): The token IDs of the end-of-sequence token.
            temperature (float): The temperature to use for the logits.
            seed (int, optional): The seed for the random number generator. Defaults to None.
            lora_name (str, optional): Name of the LoRA adapter to use. Defaults to None (the base model).

        Returns:
            (list[list[int]]): The sampled token IDs.
        """
        return await asyncio.gather(
            *[
                self.sample(
                    prompt_token_ids=prompt_token_ids,
                    max_tokens=max_tokens,
                    eos_token_ids=eos_token_ids,
                    temperature=temperature,
                    seed=seed,
                    lora_name=lora_name,
                )
                for prompt_token_ids in prompt_token_ids_list
            ]
        )

next_token_logprobs(token_ids, lora_name=None) abstractmethod async

Request log probabilities of next token asynchronously.

Parameters:

Name Type Description Default
token_ids list[int]

A list of token IDs representing the prompt.

 required
lora_name str

Name of the LoRA adapter to use. Defaults to None (the base model).

 None

Returns:

Type Description
Tensor

Normalized log probability tensor.
Source code in genlm/backend/llm/base.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@abstractmethod
async def next_token_logprobs(self, token_ids, lora_name=None):
    """Request log probabilities of next token asynchronously.

    Args:
        token_ids (list[int]): A list of token IDs representing the prompt.
        lora_name (str, optional): Name of the LoRA adapter to use. Defaults to None (the base model).

    Returns:
        (torch.Tensor): Normalized log probability tensor.
    """
    pass

next_token_logprobs_sync(token_ids, lora_name=None) abstractmethod

Request log probabilities of next token synchronously.

Parameters:

Name Type Description Default
token_ids list[int]

A list of token IDs representing the prompt.

 required
lora_name str

Name of the LoRA adapter to use. Defaults to None (the base model).

 None

Returns:

Type Description
Tensor

Normalized log probability tensor.
Source code in genlm/backend/llm/base.py 
108
109
110
111
112
113
114
115
116
117
118
119
@abstractmethod
def next_token_logprobs_sync(self, token_ids, lora_name=None):
    """Request log probabilities of next token synchronously.

    Args:
        token_ids (list[int]): A list of token IDs representing the prompt.
        lora_name (str, optional): Name of the LoRA adapter to use. Defaults to None (the base model).

    Returns:
        (torch.Tensor): Normalized log probability tensor.
    """
    pass

batch_next_token_logprobs(token_ids_list, lora_name=None) async

Batch request log probabilities for multiple token sequences asynchronously.

Parameters:

Name Type Description Default
token_ids_list list[list[int]]

A list of token ID lists.

 required
lora_name str

Name of the LoRA adapter to use. Defaults to None (the base model).

 None

Returns:

Type Description
Tensor

A tensor of log probability tensors.
Source code in genlm/backend/llm/base.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
async def batch_next_token_logprobs(self, token_ids_list, lora_name=None):
    """Batch request log probabilities for multiple token sequences asynchronously.

    Args:
        token_ids_list (list[list[int]]): A list of token ID lists.
        lora_name (str, optional): Name of the LoRA adapter to use. Defaults to None (the base model).

    Returns:
        (torch.Tensor): A tensor of log probability tensors.
    """
    logprobs = await asyncio.gather(
        *[
            self.next_token_logprobs(token_ids, lora_name=lora_name)
            for token_ids in token_ids_list
        ]
    )

    return torch.stack(logprobs)

batch_next_token_logprobs_sync(token_ids_list, lora_name=None)

Batch request log probabilities for multiple token sequences synchronously.

Parameters:

Name Type Description Default
token_ids_list list[list[int]]

A list of token ID lists.

 required
lora_name str

Name of the LoRA adapter to use. Defaults to None (the base model).

 None

Returns:

Type Description
Tensor

A tensor of log probability tensors.
Source code in genlm/backend/llm/base.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def batch_next_token_logprobs_sync(self, token_ids_list, lora_name=None):
    """Batch request log probabilities for multiple token sequences synchronously.

    Args:
        token_ids_list (list[list[int]]): A list of token ID lists.
        lora_name (str, optional): Name of the LoRA adapter to use. Defaults to None (the base model).

    Returns:
        (torch.Tensor): A tensor of log probability tensors.
    """
    return torch.stack(
        [
            self.next_token_logprobs_sync(token_ids, lora_name=lora_name)
            for token_ids in token_ids_list
        ]
    )

add_new_lora(lora_path, lora_name)

Load a LoRA adapter into the base model.

Loading under an existing name replaces that adapter. Select it per call with lora_name.

Parameters:

Name Type Description Default
lora_path str

Path to the adapter weights directory or identifier in HuggingFace's model hub.

 required
lora_name str

Name to assign to the loaded adapter.

 required
Source code in genlm/backend/llm/base.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
def add_new_lora(self, lora_path, lora_name):
    """Load a LoRA adapter into the base model.

    Loading under an existing name replaces that adapter. Select it per call with `lora_name`.

    Args:
        lora_path (str): Path to the adapter weights directory or identifier in HuggingFace's model hub.
        lora_name (str): Name to assign to the loaded adapter.

    """
    raise NotImplementedError(
        "add_new_lora must be implemented by subclasses"
    )  # pragma: no cover

remove_lora(lora_name)

Remove a previously loaded LoRA adapter.

Parameters:

Name Type Description Default
lora_name str

Name of the adapter to remove.

 required
Source code in genlm/backend/llm/base.py 
171
172
173
174
175
176
177
178
179
180
def remove_lora(self, lora_name):
    """Remove a previously loaded LoRA adapter.

    Args:
        lora_name (str): Name of the adapter to remove.

    """
    raise NotImplementedError(
        "remove_lora must be implemented by subclasses"
    )  # pragma: no cover

set_lora(lora_path, lora_name)

Unsupported: adapter selection is per-request, via lora_name=.

Source code in genlm/backend/llm/base.py 
182
183
184
185
186
187
def set_lora(self, lora_path, lora_name):
    """Unsupported: adapter selection is per-request, via ``lora_name=``."""
    raise RuntimeError(
        "set_lora() was removed: there is no active-adapter global anymore. "
        "Pass lora_name= per call (next_token_logprobs/sample/...)."
    )

clear_lora()

Unsupported: lora_name=None (the default) is the base model.

Source code in genlm/backend/llm/base.py 
189
190
191
192
193
def clear_lora(self):
    """Unsupported: ``lora_name=None`` (the default) is the base model."""
    raise RuntimeError(
        "clear_lora() was removed: omit lora_name (None = base model)."
    )

clear_cache()

Clear any caches used by the language model. No-op in base class.

Source code in genlm/backend/llm/base.py 
195
196
197
def clear_cache(self):
    """Clear any caches used by the language model. No-op in base class."""
    pass  # pragma: no cover

sample(prompt_token_ids, max_tokens, eos_token_ids, temperature=1.0, seed=None, lora_name=None) async

Sample from the language model.

Parameters:

Name Type Description Default
prompt_token_ids list[int]

The token IDs of the prompt.

 required
eos_token_ids list[int]

The token IDs of the end-of-sequence tokens.

 required
temperature float

The temperature to use to rescale the logits. Defaults to 1.0.

 1.0
max_tokens int

The maximum number of tokens to generate.

 required
seed int

The seed for the random number generator. Defaults to None.

 None
lora_name str

Name of the LoRA adapter to use. Defaults to None (the base model).

 None

Returns:

Type Description
list[int]

The sampled token IDs.
Source code in genlm/backend/llm/base.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
async def sample(
    self,
    prompt_token_ids,
    max_tokens,
    eos_token_ids,
    temperature=1.0,
    seed=None,
    lora_name=None,
):
    """Sample from the language model.

    Args:
        prompt_token_ids (list[int]): The token IDs of the prompt.
        eos_token_ids (list[int]): The token IDs of the end-of-sequence tokens.
        temperature (float, optional): The temperature to use to rescale the logits. Defaults to 1.0.
        max_tokens (int): The maximum number of tokens to generate.
        seed (int, optional): The seed for the random number generator. Defaults to None.
        lora_name (str, optional): Name of the LoRA adapter to use. Defaults to None (the base model).

    Returns:
        (list[int]): The sampled token IDs.
    """
    if seed is not None:
        generator = torch.Generator()
        generator.manual_seed(seed)
    else:
        generator = None

    generated_token_ids = []
    for _ in range(max_tokens):
        logprobs = await self.next_token_logprobs(
            prompt_token_ids + generated_token_ids, lora_name=lora_name
        )
        probs = torch.softmax(logprobs / temperature, dim=-1)
        next_token_id = torch.multinomial(
            probs.cpu() if seed is not None else probs,
            num_samples=1,
            generator=generator,
        ).item()
        if next_token_id in eos_token_ids:
            break
        generated_token_ids.append(next_token_id)

    return generated_token_ids

batch_sample(prompt_token_ids_list, max_tokens, eos_token_ids, temperature=1.0, seed=None, lora_name=None) async

Batch sample from the language model.

Parameters:

Name Type Description Default
prompt_token_ids_list list[list[int]]

The token IDs of the prompts.

 required
max_tokens int

The maximum number of tokens to generate.

 required
eos_token_ids list[int]

The token IDs of the end-of-sequence token.

 required
temperature float

The temperature to use for the logits.

 1.0
seed int

The seed for the random number generator. Defaults to None.

 None
lora_name str

Name of the LoRA adapter to use. Defaults to None (the base model).

 None

Returns:

Type Description
list[list[int]]

The sampled token IDs.
Source code in genlm/backend/llm/base.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
async def batch_sample(
    self,
    prompt_token_ids_list,
    max_tokens,
    eos_token_ids,
    temperature=1.0,
    seed=None,
    lora_name=None,
):
    """Batch sample from the language model.

    Args:
        prompt_token_ids_list (list[list[int]]): The token IDs of the prompts.
        max_tokens (int): The maximum number of tokens to generate.
        eos_token_ids (list[int]): The token IDs of the end-of-sequence token.
        temperature (float): The temperature to use for the logits.
        seed (int, optional): The seed for the random number generator. Defaults to None.
        lora_name (str, optional): Name of the LoRA adapter to use. Defaults to None (the base model).

    Returns:
        (list[list[int]]): The sampled token IDs.
    """
    return await asyncio.gather(
        *[
            self.sample(
                prompt_token_ids=prompt_token_ids,
                max_tokens=max_tokens,
                eos_token_ids=eos_token_ids,
                temperature=temperature,
                seed=seed,
                lora_name=lora_name,
            )
            for prompt_token_ids in prompt_token_ids_list
        ]
    )

MockAsyncLM

Bases: AsyncLM

Mock implementation of AsyncLM used for testing.

Source code in genlm/backend/llm/base.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
class MockAsyncLM(AsyncLM):
    """Mock implementation of AsyncLM used for testing."""

    def __init__(self, tokenizer):
        """Initialize a `MockAsyncLM` instance.

        Args:
            tokenizer: Hugging Face tokenizer instance
        """
        super().__init__(tokenizer)
        self._rng = np.random.RandomState(42)

    @classmethod
    def from_name(cls, model_name, **kwargs):
        """Create a MockAsyncLM instance over the vocabulary of the model's tokenizer.

        Args:
            model_name (str): Name of pretrained model to load tokenizer from
            **kwargs: Additional arguments passed to `MockAsyncLM` constructor

        Returns:
            (MockAsyncLM): `MockAsyncLM` instance initialized with tokenizer from `model_name`
        """
        from transformers import AutoTokenizer

        return cls(AutoTokenizer.from_pretrained(model_name), **kwargs)

    async def next_token_logprobs(self, token_ids, lora_name=None):
        """Get next token log probabilities asynchronously.

        Args:
            token_ids (list[int]): Input token IDs.
            lora_name (str, optional): Must be None; the mock has no adapters.

        Returns:
            (torch.Tensor): Normalized log probability tensor.
        """
        if lora_name is not None:
            raise ValueError(UNKNOWN_ADAPTER.format(lora_name))
        return self._get_logprobs(token_ids)

    def next_token_logprobs_sync(self, token_ids, lora_name=None):
        """Get next token log probabilities synchronously.

        Args:
            token_ids (list[int]): Input token IDs.
            lora_name (str, optional): Must be None; the mock has no adapters.

        Returns:
            (torch.Tensor): Normalized log probability tensor.
        """
        if lora_name is not None:
            raise ValueError(UNKNOWN_ADAPTER.format(lora_name))
        return self._get_logprobs(token_ids)

    def _get_logprobs(self, token_ids):
        """Generate random but deterministic log probabilities for given tokens.

        Uses token_ids to seed the random generator, ensuring same inputs produce same outputs.

        Args:
            token_ids (list[int]): Input token IDs.

        Returns:
            (torch.Tensor): Normalized log probability tensor.
        """
        seed = sum([(i + 1) * t for i, t in enumerate(token_ids)])
        self._rng.seed(seed)
        logits = torch.from_numpy(
            self._rng.rand(len(self.byte_vocab)).astype(np.float32)
        )
        return torch.log_softmax(logits, dim=-1)

__init__(tokenizer)

Initialize a MockAsyncLM instance.

Parameters:

Name Type Description Default
tokenizer

Hugging Face tokenizer instance

 required
Source code in genlm/backend/llm/base.py 
284
285
286
287
288
289
290
291
def __init__(self, tokenizer):
    """Initialize a `MockAsyncLM` instance.

    Args:
        tokenizer: Hugging Face tokenizer instance
    """
    super().__init__(tokenizer)
    self._rng = np.random.RandomState(42)

from_name(model_name, **kwargs) classmethod

Create a MockAsyncLM instance over the vocabulary of the model's tokenizer.

Parameters:

Name Type Description Default
model_name str

Name of pretrained model to load tokenizer from

 required
**kwargs

Additional arguments passed to MockAsyncLM constructor

 {}

Returns:

Type Description
MockAsyncLM

MockAsyncLM instance initialized with tokenizer from model_name
Source code in genlm/backend/llm/base.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
@classmethod
def from_name(cls, model_name, **kwargs):
    """Create a MockAsyncLM instance over the vocabulary of the model's tokenizer.

    Args:
        model_name (str): Name of pretrained model to load tokenizer from
        **kwargs: Additional arguments passed to `MockAsyncLM` constructor

    Returns:
        (MockAsyncLM): `MockAsyncLM` instance initialized with tokenizer from `model_name`
    """
    from transformers import AutoTokenizer

    return cls(AutoTokenizer.from_pretrained(model_name), **kwargs)

next_token_logprobs(token_ids, lora_name=None) async

Get next token log probabilities asynchronously.

Parameters:

Name Type Description Default
token_ids list[int]

Input token IDs.

 required
lora_name str

Must be None; the mock has no adapters.

 None

Returns:

Type Description
Tensor

Normalized log probability tensor.
Source code in genlm/backend/llm/base.py 
308
309
310
311
312
313
314
315
316
317
318
319
320
async def next_token_logprobs(self, token_ids, lora_name=None):
    """Get next token log probabilities asynchronously.

    Args:
        token_ids (list[int]): Input token IDs.
        lora_name (str, optional): Must be None; the mock has no adapters.

    Returns:
        (torch.Tensor): Normalized log probability tensor.
    """
    if lora_name is not None:
        raise ValueError(UNKNOWN_ADAPTER.format(lora_name))
    return self._get_logprobs(token_ids)

next_token_logprobs_sync(token_ids, lora_name=None)

Get next token log probabilities synchronously.

Parameters:

Name Type Description Default
token_ids list[int]

Input token IDs.

 required
lora_name str

Must be None; the mock has no adapters.

 None

Returns:

Type Description
Tensor

Normalized log probability tensor.
Source code in genlm/backend/llm/base.py 
322
323
324
325
326
327
328
329
330
331
332
333
334
def next_token_logprobs_sync(self, token_ids, lora_name=None):
    """Get next token log probabilities synchronously.

    Args:
        token_ids (list[int]): Input token IDs.
        lora_name (str, optional): Must be None; the mock has no adapters.

    Returns:
        (torch.Tensor): Normalized log probability tensor.
    """
    if lora_name is not None:
        raise ValueError(UNKNOWN_ADAPTER.format(lora_name))
    return self._get_logprobs(token_ids)