sequence.py 31.8 KB
Newer Older
1
"""Sequence and its related classes."""
2
import copy
Woosuk Kwon's avatar
Woosuk Kwon committed
3
import enum
4
from abc import ABC, abstractmethod
5
from dataclasses import dataclass, field
6
from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Union
Woosuk Kwon's avatar
Woosuk Kwon committed
7

8
9
import torch

Woosuk Kwon's avatar
Woosuk Kwon committed
10
from vllm.block import LogicalTokenBlock
11
from vllm.inputs import LLMInputs
12
from vllm.lora.request import LoRARequest
13
from vllm.pooling_params import PoolingParams
14
from vllm.sampling_params import SamplingParams
Woosuk Kwon's avatar
Woosuk Kwon committed
15

16
if TYPE_CHECKING:
17
    from vllm.multimodal import MultiModalData
18
19
    from vllm.spec_decode.metrics import SpecDecodeWorkerMetrics

20
21
22

@dataclass
class Logprob:
23
24
25
26
27
28
29
    """Infos for supporting OpenAI compatible logprobs and token ranks.

    Attributes:
        logprob: The logprob of chosen token
        rank: The vocab rank of chosen token (>=1)
        decoded_token: The decoded chosen token index
    """
30
    logprob: float
31
    rank: Optional[int] = None
32
33
34
    decoded_token: Optional[str] = None


35
36
# {token_id -> logprob} per each sequence group. None if the corresponding
# sequence group doesn't require prompt logprob.
37
PromptLogprobs = List[Optional[Dict[int, Logprob]]]
38
# {token_id -> logprob} for each sequence group.
39
SampleLogprobs = List[Dict[int, Logprob]]
40

Woosuk Kwon's avatar
Woosuk Kwon committed
41
42

class SequenceStatus(enum.Enum):
43
    """Status of a sequence."""
44
    WAITING = enum.auto()
Woosuk Kwon's avatar
Woosuk Kwon committed
45
    RUNNING = enum.auto()
Woosuk Kwon's avatar
Woosuk Kwon committed
46
    SWAPPED = enum.auto()
Zhuohan Li's avatar
Zhuohan Li committed
47
48
    FINISHED_STOPPED = enum.auto()
    FINISHED_LENGTH_CAPPED = enum.auto()
49
    FINISHED_ABORTED = enum.auto()
Lily Liu's avatar
Lily Liu committed
50
    FINISHED_IGNORED = enum.auto()
Zhuohan Li's avatar
Zhuohan Li committed
51
52
53
54
55
56

    @staticmethod
    def is_finished(status: "SequenceStatus") -> bool:
        return status in [
            SequenceStatus.FINISHED_STOPPED,
            SequenceStatus.FINISHED_LENGTH_CAPPED,
57
            SequenceStatus.FINISHED_ABORTED,
58
            SequenceStatus.FINISHED_IGNORED,
Zhuohan Li's avatar
Zhuohan Li committed
59
60
61
62
63
64
65
66
        ]

    @staticmethod
    def get_finished_reason(status: "SequenceStatus") -> Union[str, None]:
        if status == SequenceStatus.FINISHED_STOPPED:
            finish_reason = "stop"
        elif status == SequenceStatus.FINISHED_LENGTH_CAPPED:
            finish_reason = "length"
67
68
        elif status == SequenceStatus.FINISHED_ABORTED:
            finish_reason = "abort"
Lily Liu's avatar
Lily Liu committed
69
        elif status == SequenceStatus.FINISHED_IGNORED:
70
71
72
            # The ignored sequences are the sequences whose prompt lengths
            # are longer than the model's length cap. Therefore, the stop
            # reason should also be "length" as in OpenAI API.
Lily Liu's avatar
Lily Liu committed
73
            finish_reason = "length"
Zhuohan Li's avatar
Zhuohan Li committed
74
75
76
        else:
            finish_reason = None
        return finish_reason
Woosuk Kwon's avatar
Woosuk Kwon committed
77

78

79
80
81
82
83
class SequenceStage(enum.Enum):
    PREFILL = enum.auto()
    DECODE = enum.auto()


84
85
86
87
@dataclass
class RequestMetrics:
    """Metrics associated with a request.

88
    Attributes:
89
90
91
92
93
94
95
96
97
98
99
100
101
102
        arrival_time: The time when the request arrived.
        first_scheduled_time: The time when the request was first scheduled.
        first_token_time: The time when the first token was generated.
        time_in_queue: The time the request spent in the queue.
        finished_time: The time when the request was finished.
    """
    arrival_time: float
    last_token_time: float
    first_scheduled_time: Optional[float]
    first_token_time: Optional[float]
    time_in_queue: Optional[float]
    finished_time: Optional[float] = None


103
class SequenceData:
104
105
106
107
    """Data associated with a sequence.

    Args:
        prompt_token_ids: The token IDs of the prompt.
108
109
        output_token_ids: The token IDs of the output. Set to an empty list if
            None.
110
111
112
113
114
115

    Attributes:
        prompt_token_ids: The token IDs of the prompt.
        output_token_ids: The token IDs of the output.
        cumulative_logprob: The cumulative log probability of the output.
    """
116
117
118
119

    def __init__(
        self,
        prompt_token_ids: List[int],
120
        output_token_ids: Optional[List[int]] = None,
121
    ) -> None:
122
123
124
        if output_token_ids is None:
            output_token_ids = []

125
        self.prompt_token_ids = prompt_token_ids
126
        self._prompt_token_ids_tuple = tuple(prompt_token_ids)
127
        self.output_token_ids = output_token_ids
128
        self.cumulative_logprob = 0.0
129
130
        # The number of tokens that are computed (that run against the model).
        self._num_computed_tokens = 0
131
        self._stage: SequenceStage = SequenceStage.PREFILL
132

133
    def append_token_id(self, token_id: int, logprob: float) -> None:
134
135
        self.output_token_ids.append(token_id)
        self.cumulative_logprob += logprob
136
137
138
139

    def get_len(self) -> int:
        return len(self.output_token_ids) + len(self.prompt_token_ids)

140
141
142
    def get_prompt_len(self) -> int:
        return len(self.prompt_token_ids)

143
144
145
    def get_output_len(self) -> int:
        return len(self.output_token_ids)

146
147
148
    def get_token_ids(self) -> List[int]:
        return self.prompt_token_ids + self.output_token_ids

149
150
151
152
153
154
155
156
157
158
159
    def get_prefix_token_ids(
            self, num_tokens: int
    ) -> Tuple[Tuple[int, ...], Optional[Tuple[int, ...]]]:
        """Get prefix tokens, and make the return value hashable"""
        prompt_length = len(self.prompt_token_ids)
        if num_tokens > prompt_length:
            return (self._prompt_token_ids_tuple,
                    tuple(self.output_token_ids[:num_tokens - prompt_length]))
        else:
            return (self._prompt_token_ids_tuple[:num_tokens], None)

160
161
162
163
    def get_num_computed_tokens(self) -> int:
        """Return the number of prefill tokens that are already computed."""
        return self._num_computed_tokens

164
    def update_num_computed_tokens(self, num_new_computed_tokens: int):
165
166
        """Update number of tokens computed so far."""
        self._num_computed_tokens += num_new_computed_tokens
167
168
169
170
171
        assert self._num_computed_tokens <= self.get_len(), (
            self._num_computed_tokens, self.get_len())
        # If all tokens are computed, it means it is in decoding phase.
        if self.get_num_uncomputed_tokens() == 0:
            self._stage = SequenceStage.DECODE
172

173
    def reset_state_for_recompute(self) -> None:
174
175
176
177
178
        """Reset the number of computed tokens from this sequence. It is
        supposed to be called when a sequence needs to be started from
        the beginning again (e.g., sequence is preempted).
        """
        self._num_computed_tokens = 0
179
        self._stage = SequenceStage.PREFILL
180
181

    def get_num_uncomputed_tokens(self) -> int:
Uranus's avatar
Uranus committed
182
        """Return the number of prefill tokens that are not computed."""
183
184
185
186
187
        # we use `get_len()` which includes prompt_len + output_len instead
        # of prompt_len here. This is because during recompute we need to
        # prefill for both prompt and output.
        return self.get_len() - self.get_num_computed_tokens()

188
189
190
191
192
    def get_last_token_id(self) -> int:
        if not self.output_token_ids:
            return self.prompt_token_ids[-1]
        return self.output_token_ids[-1]

193
    def get_prompt_token_ids(self) -> List[int]:
194
195
        return self.prompt_token_ids

196
    def get_output_token_ids(self) -> List[int]:
197
198
        return self.output_token_ids

199
200
201
202
    @property
    def stage(self) -> SequenceStage:
        return self._stage

203
204
205
    def __repr__(self) -> str:
        return (f"SequenceData("
                f"prompt_token_ids={self.prompt_token_ids}, "
206
207
                f"output_token_ids={self.output_token_ids}, "
                f"cumulative_logprob={self.cumulative_logprob})")
208
209


Woosuk Kwon's avatar
Woosuk Kwon committed
210
class Sequence:
211
212
213
214
    """Stores the data, status, and block information of a sequence.

    Args:
        seq_id: The ID of the sequence.
215
        inputs: The inputs of the sequence.
216
217
        block_size: The block size of the sequence. Should be the same as the
            block size used by the block manager and cache engine.
218
        lora_request: LoRA request.
219
    """
Woosuk Kwon's avatar
Woosuk Kwon committed
220
221
222
223

    def __init__(
        self,
        seq_id: int,
224
        inputs: LLMInputs,
Woosuk Kwon's avatar
Woosuk Kwon committed
225
        block_size: int,
Cade Daniel's avatar
Cade Daniel committed
226
        eos_token_id: Optional[int] = None,
227
        lora_request: Optional[LoRARequest] = None,
Woosuk Kwon's avatar
Woosuk Kwon committed
228
229
    ) -> None:
        self.seq_id = seq_id
230
        self.inputs = inputs
Woosuk Kwon's avatar
Woosuk Kwon committed
231
        self.block_size = block_size
232
        self.eos_token_id = eos_token_id
233
        self.lora_request = lora_request
Woosuk Kwon's avatar
Woosuk Kwon committed
234

235
        self.data = SequenceData(self.prompt_token_ids)
236
        self.output_logprobs: SampleLogprobs = []
237
        self.output_text = ""
238

Woosuk Kwon's avatar
Woosuk Kwon committed
239
        self.logical_token_blocks: List[LogicalTokenBlock] = []
240
        # Initialize the logical token blocks with the prompt token ids.
241
        self._append_tokens_to_blocks(self.prompt_token_ids)
242
        self.status = SequenceStatus.WAITING
243
        self.stop_reason: Union[int, str, None] = None
Woosuk Kwon's avatar
Woosuk Kwon committed
244

245
246
247
248
249
250
        # Used for incremental detokenization
        self.prefix_offset = 0
        self.read_offset = 0
        # Input + output tokens
        self.tokens: Optional[List[str]] = None

251
252
    @property
    def prompt(self) -> Optional[str]:
253
        return self.inputs.get("prompt")
254
255
256
257
258
259
260

    @property
    def prompt_token_ids(self) -> List[int]:
        return self.inputs["prompt_token_ids"]

    @property
    def multi_modal_data(self) -> Optional["MultiModalData"]:
261
        return self.inputs.get("multi_modal_data")
262

263
264
265
266
    @property
    def lora_int_id(self) -> int:
        return self.lora_request.lora_int_id if self.lora_request else 0

267
268
269
270
271
272
    def get_output_text_to_return(self, buffer_length: int):
        # We return the full output text if the sequence is finished.
        truncate = buffer_length and not self.is_finished()
        return self.output_text[:-buffer_length] if truncate else (
            self.output_text)

273
    def hash_of_block(self, logical_idx: int) -> int:
274
275
        # TODO This can produce incorrect hash when block size > prompt size

276
        # Compute the number of tokens in the sequence
277
278
        # TODO: The current hashing function is O(L^2). We should optimize
        # this in the future.
279
        num_tokens = self.num_hashed_tokens_of_block(logical_idx)
280
281
        hashed_tokens = self.data.get_prefix_token_ids(num_tokens)
        return hash((hashed_tokens, self.lora_int_id))
282
283
284
285

    def num_hashed_tokens_of_block(self, logical_idx: int):
        return logical_idx * self.block_size + self.block_size

286
287
    def reset_state_for_recompute(self):
        """Reset the sequence states for recomputation."""
288
        self.data.reset_state_for_recompute()
289

290
    def _append_logical_block(self) -> None:
Woosuk Kwon's avatar
Woosuk Kwon committed
291
292
293
294
295
296
        block = LogicalTokenBlock(
            block_number=len(self.logical_token_blocks),
            block_size=self.block_size,
        )
        self.logical_token_blocks.append(block)

297
    def _append_tokens_to_blocks(self, token_ids: List[int]) -> None:
298
299
        cursor = 0
        while cursor < len(token_ids):
Woosuk Kwon's avatar
Woosuk Kwon committed
300
            if not self.logical_token_blocks:
301
                self._append_logical_block()
Woosuk Kwon's avatar
Woosuk Kwon committed
302
303
304

            last_block = self.logical_token_blocks[-1]
            if last_block.is_full():
305
                self._append_logical_block()
Woosuk Kwon's avatar
Woosuk Kwon committed
306
307
308
                last_block = self.logical_token_blocks[-1]

            num_empty_slots = last_block.get_num_empty_slots()
309
310
311
            last_block.append_tokens(token_ids[cursor:cursor +
                                               num_empty_slots])
            cursor += num_empty_slots
Woosuk Kwon's avatar
Woosuk Kwon committed
312

313
314
315
    def append_token_id(
        self,
        token_id: int,
316
        logprobs: Dict[int, Logprob],
317
    ) -> None:
318
        assert token_id in logprobs
319
        self._append_tokens_to_blocks([token_id])
320
        self.output_logprobs.append(logprobs)
321
        self.data.append_token_id(token_id, logprobs[token_id].logprob)
322

Woosuk Kwon's avatar
Woosuk Kwon committed
323
    def get_len(self) -> int:
324
        return self.data.get_len()
Woosuk Kwon's avatar
Woosuk Kwon committed
325

326
327
328
    def get_prompt_len(self) -> int:
        return self.data.get_prompt_len()

329
330
331
    def get_output_len(self) -> int:
        return self.data.get_output_len()

Woosuk Kwon's avatar
Woosuk Kwon committed
332
    def get_token_ids(self) -> List[int]:
333
        return self.data.get_token_ids()
Woosuk Kwon's avatar
Woosuk Kwon committed
334

335
336
337
    def get_prompt_token_ids(self) -> List[int]:
        return self.data.get_prompt_token_ids()

338
    def get_last_token_id(self) -> int:
339
        return self.data.get_last_token_id()
340

341
342
343
344
345
346
    def get_output_token_ids(self) -> List[int]:
        return self.data.output_token_ids

    def get_cumulative_logprob(self) -> float:
        return self.data.cumulative_logprob

347
    def get_beam_search_score(self,
348
                              length_penalty: float = 1.0,
349
350
351
352
353
354
355
356
357
358
                              seq_len: Optional[int] = None,
                              eos_token_id: Optional[int] = None) -> float:
        """Calculate the beam search score with length penalty.

        Adapted from

        https://github.com/huggingface/transformers/blob/ccb92be23def445f2afdea94c31286f84b89eb5b/src/transformers/generation/beam_search.py#L938
        """
        if seq_len is None:
            seq_len = self.get_len()
359
            # NOTE: HF implementation does not count the EOS token
360
361
362
363
364
365
            # towards the length, we align with that here for testing.
            if (eos_token_id is not None
                    and self.get_last_token_id() == eos_token_id):
                seq_len -= 1
        return self.get_cumulative_logprob() / (seq_len**length_penalty)

366
367
368
    def is_finished(self) -> bool:
        return SequenceStatus.is_finished(self.status)

369
370
371
372
    def fork(self, new_seq_id: int) -> "Sequence":
        new_seq = copy.deepcopy(self)
        new_seq.seq_id = new_seq_id
        return new_seq
373

374
375
376
377
    def get_num_new_tokens(self) -> int:
        """Get the number of new tokens to be computed.

        Returns:
Uranus's avatar
Uranus committed
378
379
            The new number of tokens to be computed. I.e., 1 for decode, or
            the remaining prompt size for prefill.
380
381
382
383
384
385
386
387
        """
        if self.data.stage == SequenceStage.DECODE:
            return 1
        return self.data.get_num_uncomputed_tokens()

    def is_prefill(self) -> bool:
        return self.data.stage == SequenceStage.PREFILL

Woosuk Kwon's avatar
Woosuk Kwon committed
388
    def __repr__(self) -> str:
389
390
391
        return (f"Sequence(seq_id={self.seq_id}, "
                f"status={self.status.name}, "
                f"num_blocks={len(self.logical_token_blocks)})")
Woosuk Kwon's avatar
Woosuk Kwon committed
392

Woosuk Kwon's avatar
Woosuk Kwon committed
393

Nick Hill's avatar
Nick Hill committed
394
395
396
397
398
@dataclass
class SequenceGroupState:
    """Mutable state tied to a specific sequence group"""

    # torch.Generator used in seeded sampling
399
    generator: Optional = None  # type: ignore
Nick Hill's avatar
Nick Hill committed
400
401


Woosuk Kwon's avatar
Woosuk Kwon committed
402
class SequenceGroup:
403
404
405
406
407
408
409
    """A group of sequences that are generated from the same prompt.

    Args:
        request_id: The ID of the request.
        seqs: The list of sequences.
        sampling_params: The sampling parameters used to generate the outputs.
        arrival_time: The arrival time of the request.
410
        lora_request: LoRA request.
411
412
413
414
        embeddings: The embeddings vectors of the prompt of the sequence group
            for an embedding model.
        pooling_params: The pooling parameters used to generate the pooling
            for an embedding model.
415
416
        encoder_seq: Optional, the single encoder sequence. Should be None
                     unless you are working with an encoder/decoder model.
417
    """
Woosuk Kwon's avatar
Woosuk Kwon committed
418
419
420

    def __init__(
        self,
421
        request_id: str,
Woosuk Kwon's avatar
Woosuk Kwon committed
422
        seqs: List[Sequence],
423
        arrival_time: float,
424
        sampling_params: Optional[SamplingParams] = None,
425
        lora_request: Optional[LoRARequest] = None,
426
427
        embeddings: Optional[List[float]] = None,
        pooling_params: Optional[PoolingParams] = None,
428
        encoder_seq: Optional[Sequence] = None,
Woosuk Kwon's avatar
Woosuk Kwon committed
429
    ) -> None:
430
        self.request_id = request_id
431
        self.seqs_dict = {seq.seq_id: seq for seq in seqs}
432
        self.sampling_params = sampling_params
433
434
435
436
437
        self.metrics = RequestMetrics(arrival_time=arrival_time,
                                      last_token_time=arrival_time,
                                      first_scheduled_time=None,
                                      first_token_time=None,
                                      time_in_queue=None)
438
        self.lora_request = lora_request
439
        self.prompt_logprobs: Optional[PromptLogprobs] = None
Nick Hill's avatar
Nick Hill committed
440
        self.state = SequenceGroupState()
441
442
        self.embeddings = embeddings
        self.pooling_params = pooling_params
443
        self.encoder_seq = encoder_seq
444
445

    @property
446
    def prompt(self) -> Optional[str]:
447
448
449
450
451
452
453
454
        # All sequences in the group should have the same prompt.
        # We use the prompt of an arbitrary sequence.
        return next(iter(self.seqs_dict.values())).prompt

    @property
    def prompt_token_ids(self) -> List[int]:
        # All sequences in the group should have the same prompt.
        # We use the prompt of an arbitrary sequence.
455
456
457
        return next(iter(self.seqs_dict.values())).prompt_token_ids

    @property
458
    def multi_modal_data(self) -> Optional["MultiModalData"]:
459
460
461
        # All sequences in the group should have the same multi-modal data.
        # We use the multi-modal data of an arbitrary sequence.
        return next(iter(self.seqs_dict.values())).multi_modal_data
Woosuk Kwon's avatar
Woosuk Kwon committed
462

463
464
465
466
    @property
    def lora_int_id(self) -> int:
        return self.lora_request.lora_int_id if self.lora_request else 0

467
468
469
470
471
472
473
474
475
    def get_last_latency(self, now: float) -> Optional[float]:
        """Sets the last token time for Request level timings."""
        # If still in prefill phase, raise Error.
        if self.is_prefill():
            raise ValueError(
                "seq_group.get_last_latency() should not be called "
                "if the seq_group is in prefill phase.")

        # Otherwise return token latency.
476
477
        latency = now - self.metrics.last_token_time
        self.metrics.last_token_time = now
478
479
        return latency

480
481
    def maybe_set_first_token_time(self, time: float) -> None:
        """Sets the first token time for Request level timings."""
482
483
484
485
486
487
        # Note: in a case where a sequence_group is swapped and
        #   recomputed, the time between iterations is counted
        #   in TPOT, rather than recalculating TTFT (since from the )
        #   POV of the user, there is simply a long generation delay.
        if (self.metrics.first_token_time is None
                and self.get_seqs()[0].get_output_len() == 1):
488
489
490
            self.metrics.first_token_time = time

    def maybe_set_first_scheduled_time(self, time: float) -> None:
491
492
        """Sets the first scheduled time and time in queue for Request
        level timings."""
493
494
495
496
497
498
499
500
        if self.metrics.first_scheduled_time is None:
            self.metrics.first_scheduled_time = time
            self.metrics.time_in_queue = time - self.metrics.arrival_time

    def set_finished_time(self, time: Optional[float]) -> None:
        """Sets the finished time for Request level timings."""
        self.metrics.finished_time = time

501
502
503
    def get_max_num_running_seqs(self) -> int:
        """The maximum number of sequences running in parallel in the remaining
        lifetime of the request."""
504
        if self.sampling_params and self.sampling_params.use_beam_search:
505
506
507
508
            # For beam search, maximally there will always be `best_of` beam
            # candidates running in the future.
            return self.sampling_params.best_of
        else:
509
510
            if (self.sampling_params
                    and self.sampling_params.best_of > self.num_seqs()):
511
512
513
514
515
                # At prompt stage, the sequence group is not yet filled up
                # and only have one sequence running. However, in the
                # generation stage, we will have `best_of` sequences running.
                return self.sampling_params.best_of
            # At sampling stages, return the number of actual sequences
516
517
            # that are not finished yet.
            return self.num_unfinished_seqs()
518

519
520
521
522
    def get_seqs(
        self,
        status: Optional[SequenceStatus] = None,
    ) -> List[Sequence]:
523
524
525
        return list(self.seqs_dict.values()) if status is None else [
            seq for seq in self.seqs_dict.values() if seq.status == status
        ]
526

527
528
529
530
531
532
    def is_encoder_decoder(self) -> bool:
        return self.encoder_seq is not None

    def get_encoder_seq(self) -> Optional[Sequence]:
        return self.encoder_seq

533
534
535
536
537
    def get_unfinished_seqs(self) -> List[Sequence]:
        return [
            seq for seq in self.seqs_dict.values() if not seq.is_finished()
        ]

538
539
    def get_finished_seqs(self) -> List[Sequence]:
        return [seq for seq in self.seqs_dict.values() if seq.is_finished()]
540

541
542
543
    def update_num_computed_tokens(self, num_new_computed_tokens: int):
        """Update number of tokens computed so far."""
        for seq in self.seqs_dict.values():
544
545
            if not seq.is_finished():
                seq.data.update_num_computed_tokens(num_new_computed_tokens)
546
547

    def get_num_uncomputed_tokens(self) -> int:
548
549
        num_uncomputed_tokens = 0
        for seq in self.get_seqs():
550
551
            if not seq.is_finished():
                num_uncomputed_tokens += seq.data.get_num_uncomputed_tokens()
552
        return num_uncomputed_tokens
553

554
    def num_seqs(self, status: Optional[SequenceStatus] = None) -> int:
555
556
557
558
559
        # Optimization. We don't need to call get_seqs if we don't need to
        # filter by states.
        if status is None:
            return len(self.seqs_dict)

560
        return len(self.get_seqs(status))
561

562
563
564
565
566
567
    def num_unfinished_seqs(self) -> int:
        return len(self.get_unfinished_seqs())

    def num_finished_seqs(self) -> int:
        return len(self.get_finished_seqs())

568
    def find(self, seq_id: int) -> Sequence:
569
570
571
572
573
574
575
576
577
578
579
580
581
        if seq_id not in self.seqs_dict:
            raise ValueError(f"Sequence {seq_id} not found.")
        return self.seqs_dict[seq_id]

    def add(self, seq: Sequence) -> None:
        if seq.seq_id in self.seqs_dict:
            raise ValueError(f"Sequence {seq.seq_id} already exists.")
        self.seqs_dict[seq.seq_id] = seq

    def remove(self, seq_id: int) -> None:
        if seq_id not in self.seqs_dict:
            raise ValueError(f"Sequence {seq_id} not found.")
        del self.seqs_dict[seq_id]
Woosuk Kwon's avatar
Woosuk Kwon committed
582

Woosuk Kwon's avatar
Woosuk Kwon committed
583
    def is_finished(self) -> bool:
584
        return all(seq.is_finished() for seq in self.get_seqs())
Woosuk Kwon's avatar
Woosuk Kwon committed
585

586
    def is_prefill(self) -> bool:
587
        # Every sequence should be in the same stage.
588
589
        return self.get_seqs()[0].is_prefill()

Woosuk Kwon's avatar
Woosuk Kwon committed
590
    def __repr__(self) -> str:
591
592
        return (f"SequenceGroup(request_id={self.request_id}, "
                f"sampling_params={self.sampling_params}, "
593
                f"num_seqs={len(self.seqs_dict)})")
594
595


596
class SequenceGroupMetadata:
597
    """Metadata for a sequence group. Used to create `AttentionMetadata`.
598
599
600
601
602
603
604
605

    Args:
        request_id: The ID of the request.
        is_prompt: Whether the request is at prompt stage.
        seq_data: The sequence data. (Seq id -> sequence data)
        sampling_params: The sampling parameters used to generate the outputs.
        block_tables: The block tables. (Seq id -> list of physical block
            numbers)
606
607
608
        do_sample: True if sampling is required. Sampling is not required when
            e.g., prefill is chunked, and the current iteration only computes
            query tokens for prefill, we don't need sampling.
609
610
        token_chunk_size: The number of tokens to be processed (per sequence).
            None if chunking is not required.
611
        lora_request: LoRA request.
612
613
614
        computed_block_nums: The block numbers that are already computed,
            used in prefix caching.
        state: Internal state tied to this sequence group.
615
        multi_modal_data: Multi modal data.
616
617
618
619
620
621
622
623
624
        encoder_seq_data: Optional sequence data for encoder prompt
                          (SequenceGroup.encoder_seq). Should be None 
                          unless you are working with an encoder/decoder
                          model.
        cross_block_table: Optional cross-attention block table associated
                           with the encoder prompt
                           (SequenceGroup.encoder_seq). Should be None
                           unless you are working with an encoder/decoder
                           model.
625
    """
626
627
628

    def __init__(
        self,
629
        request_id: str,
630
        is_prompt: bool,
631
        seq_data: Dict[int, SequenceData],
632
        sampling_params: SamplingParams,
633
        block_tables: Dict[int, List[int]],
634
        do_sample: bool = True,
635
        pooling_params: Optional[PoolingParams] = None,
636
        token_chunk_size: Optional[int] = None,
637
        lora_request: Optional[LoRARequest] = None,
638
        computed_block_nums: Optional[List[int]] = None,
Nick Hill's avatar
Nick Hill committed
639
        state: Optional[SequenceGroupState] = None,
640
        multi_modal_data: Optional["MultiModalData"] = None,
641
642
        encoder_seq_data: Optional[SequenceData] = None,
        cross_block_table: Optional[List[int]] = None,
643
    ) -> None:
644
        self.request_id = request_id
645
        self.is_prompt = is_prompt
646
        self.seq_data = seq_data
647
648
        self.sampling_params = sampling_params
        self.block_tables = block_tables
649
        self.pooling_params = pooling_params
650
        self.lora_request = lora_request
651
        self.computed_block_nums = computed_block_nums
652
        self.multi_modal_data = multi_modal_data
Nick Hill's avatar
Nick Hill committed
653
        self.state = SequenceGroupState() if state is None else state
654
655
        self.encoder_seq_data = encoder_seq_data
        self.cross_block_table = cross_block_table
656
        self._token_chunk_size = token_chunk_size
657
        self.do_sample = do_sample
658

659
660
661
662
663
664
        # The number of speculative tokens adopted in this request.
        # None means specuative decoding is not used.
        # Zero means speculative decoding is disabled for some reasons.
        # TODO: We should maintain this states out of the sequence group.
        self.num_speculative_tokens = None

665
666
667
668
669
        if self._token_chunk_size is None:
            if is_prompt:
                self._token_chunk_size = list(seq_data.values())[0].get_len()
            else:
                self._token_chunk_size = 1
670

671
672
673
674
    @property
    def lora_int_id(self) -> int:
        return self.lora_request.lora_int_id if self.lora_request else 0

675
    @property
676
    def token_chunk_size(self) -> int:
677
        """Return the number of tokens to be processed (chunk size)."""
678
        assert self._token_chunk_size is not None
679
680
        return self._token_chunk_size

681

Zhuohan Li's avatar
Zhuohan Li committed
682
class SequenceOutput:
683
684
685
686
687
688
689
690
691
    """The model output associated with a sequence.

    Args:
        parent_seq_id: The ID of the parent sequence (for forking in beam
            search).
        output_token: The output token ID.
        logprobs: The logprobs of the output token.
            (Token id -> logP(x_i+1 | x_0, ..., x_i))
    """
692
693
694
695
696

    def __init__(
        self,
        parent_seq_id: int,
        output_token: int,
697
        logprobs: Dict[int, Logprob],
698
699
700
701
702
703
    ) -> None:
        self.parent_seq_id = parent_seq_id
        self.output_token = output_token
        self.logprobs = logprobs

    def __repr__(self) -> str:
Zhuohan Li's avatar
Zhuohan Li committed
704
        return (f"SequenceOutput(parent_seq_id={self.parent_seq_id}, "
705
706
                f"output_token={self.output_token}, "
                f"logprobs={self.logprobs})")
Zhuohan Li's avatar
Zhuohan Li committed
707

708
    def __eq__(self, other: object) -> bool:
Zhuohan Li's avatar
Zhuohan Li committed
709
        if not isinstance(other, SequenceOutput):
Zhuohan Li's avatar
Zhuohan Li committed
710
            raise NotImplementedError()
711
712
713
714
        equal = (self.parent_seq_id == other.parent_seq_id
                 and self.output_token == other.output_token)
        log_probs_equal = other.logprobs == self.logprobs
        return equal and log_probs_equal
715
716


717
718
719
720
721
722
723
724
725
726
727
728
729
730
class SequenceGroupOutput(ABC):
    """The base class for model outputs associated with a sequence group."""

    @abstractmethod
    def __repr__(self) -> str:
        pass

    @abstractmethod
    def __eq__(self, other: object) -> bool:
        pass


class CompletionSequenceGroupOutput(SequenceGroupOutput):
    """The model output associated with a completion sequence group."""
731
732
733

    def __init__(
        self,
Zhuohan Li's avatar
Zhuohan Li committed
734
        samples: List[SequenceOutput],
735
736
737
        prompt_logprobs: Optional[PromptLogprobs],
    ) -> None:
        self.samples = samples
738
        # Prompt logprob for each prompt query token.
739
740
741
        self.prompt_logprobs = prompt_logprobs

    def __repr__(self) -> str:
742
        return (f"CompletionSequenceGroupOutput(samples={self.samples}, "
743
744
                f"prompt_logprobs={self.prompt_logprobs})")

745
    def __eq__(self, other: object) -> bool:
746
        if not isinstance(other, CompletionSequenceGroupOutput):
747
748
749
750
            raise NotImplementedError()
        return (self.samples == other.samples
                and self.prompt_logprobs == other.prompt_logprobs)

751

752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
class EmbeddingSequenceGroupOutput(SequenceGroupOutput):
    """The model output associated with an embedding sequence group."""

    def __init__(
        self,
        embeddings: List[float],
    ) -> None:
        self.embeddings = embeddings

    def __repr__(self) -> str:
        return (f"EmbeddingSequenceGroupOutput("
                f"embeddings_shape={len(self.embeddings)})")

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, EmbeddingSequenceGroupOutput):
            raise NotImplementedError()
        return self.embeddings == other.embeddings


771
772
773
774
775
@dataclass
class SamplerOutput:
    """For each sequence group, we generate a list of SequenceOutput object,
    each of which contains one possible candidate for the next token.

776
    This data structure implements methods, so it can be used like a list, but
777
778
779
    also has optional fields for device tensors.
    """

780
    outputs: List[CompletionSequenceGroupOutput]
781
782

    # On-device tensor containing probabilities of each token.
783
    sampled_token_probs: Optional[torch.Tensor] = None
784

785
786
787
    # On-device tensor containing the logprobs of each token.
    logprobs: Optional["torch.Tensor"] = None

788
    # On-device tensor containing the sampled token ids.
789
    sampled_token_ids: Optional[torch.Tensor] = None
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805

    # Spec decode metrics populated by workers.
    spec_decode_worker_metrics: Optional["SpecDecodeWorkerMetrics"] = None

    def __getitem__(self, idx: int):
        return self.outputs[idx]

    def __setitem__(self, idx: int, value):
        self.outputs[idx] = value

    def __len__(self):
        return len(self.outputs)

    def __eq__(self, other: object):
        return isinstance(other,
                          self.__class__) and self.outputs == other.outputs
806
807
808
809
810
811
812
813
814
815
816
817
818

    def __repr__(self) -> str:
        """Show the shape of a tensor instead of its values to reduce noise.
        """
        sampled_token_probs_repr = ("None" if self.sampled_token_probs is None
                                    else self.sampled_token_probs.shape)
        sampled_token_ids_repr = ("None" if self.sampled_token_ids is None else
                                  self.sampled_token_ids.shape)
        return (
            f"SamplerOutput(outputs={self.outputs}, "
            f"sampled_token_probs={sampled_token_probs_repr}, "
            f"sampled_token_ids={sampled_token_ids_repr}, "
            f"spec_decode_worker_metrics={self.spec_decode_worker_metrics})")
819
820


821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
@dataclass
class PoolerOutput:
    """The output from a pooling operation in the embedding model."""
    outputs: List[EmbeddingSequenceGroupOutput]

    spec_decode_worker_metrics: Optional["SpecDecodeWorkerMetrics"] = None

    def __getitem__(self, idx: int):
        return self.outputs[idx]

    def __setitem__(self, idx: int, value):
        self.outputs[idx] = value

    def __len__(self):
        return len(self.outputs)

    def __eq__(self, other: object):
        return isinstance(other,
                          self.__class__) and self.outputs == other.outputs


842
843
844
845
846
@dataclass
class ExecuteModelRequest:
    """The model execution request."""
    # The sequence group metadata list.
    seq_group_metadata_list: List[SequenceGroupMetadata]
847
848
849
850
    # Blocks to swap in. List of CPU -> GPU block number.
    blocks_to_swap_in: List[Tuple[int, int]] = field(default_factory=list)
    # Blocks to swap out. List of GPU -> CPU block number.
    blocks_to_swap_out: List[Tuple[int, int]] = field(default_factory=list)
851
852
    # Blocks to copy. Source to dest block.
    blocks_to_copy: List[Tuple[int, int]] = field(default_factory=list)
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
    # The number of slots for lookahead decoding.
    num_lookahead_slots: int = 0
    # The number of requests in the running queue.
    running_queue_size: int = 0

    def clone(
        self, seq_group_metadata_list: List[SequenceGroupMetadata]
    ) -> "ExecuteModelRequest":
        """Clone the request with a new sequence group metadata list."""
        return ExecuteModelRequest(
            seq_group_metadata_list=seq_group_metadata_list,
            blocks_to_swap_in=self.blocks_to_swap_in.copy(),
            blocks_to_swap_out=self.blocks_to_swap_out.copy(),
            blocks_to_copy=self.blocks_to_copy.copy(),
            num_lookahead_slots=self.num_lookahead_slots,
            running_queue_size=self.running_queue_size,
        )