serving.py 26.4 KB
Newer Older
1
# SPDX-License-Identifier: Apache-2.0
2
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
import json
4
from collections.abc import AsyncGenerator, Mapping
5
from typing import Any, Final, cast
6

7
import torch
8
from fastapi import Request
9
10
from fastapi.responses import Response
from typing_extensions import assert_never, override
11

12
from vllm.engine.protocol import EngineClient
13
from vllm.entrypoints.chat_utils import ChatTemplateContentFormatOption
14
from vllm.entrypoints.logger import RequestLogger
15
16
17
18
19
20
21
22
23
24
from vllm.entrypoints.openai.protocol import (
    ErrorResponse,
    UsageInfo,
)
from vllm.entrypoints.openai.serving_engine import (
    EmbeddingServeContext,
    OpenAIServing,
    ServeContext,
    TextTokensPrompt,
)
25
from vllm.entrypoints.openai.serving_models import OpenAIServingModels
26
27
28
29
30
31
32
33
from vllm.entrypoints.pooling.embed.protocol import (
    EmbeddingBytesResponse,
    EmbeddingChatRequest,
    EmbeddingCompletionRequest,
    EmbeddingRequest,
    EmbeddingResponse,
    EmbeddingResponseData,
)
34
from vllm.entrypoints.renderer import RenderConfig
35
from vllm.inputs.data import TokensPrompt as EngineTokensPrompt
36
from vllm.logger import init_logger
37
38
39
40
41
42
from vllm.outputs import (
    EmbeddingRequestOutput,
    PoolingOutput,
    PoolingRequestOutput,
    RequestOutput,
)
43
from vllm.pooling_params import PoolingParams
44
45
from vllm.utils.async_utils import merge_async_iterators
from vllm.utils.collection_utils import chunk_list
46
47
48
49
50
51
52
from vllm.utils.serial_utils import (
    EmbedDType,
    EncodingFormat,
    Endianness,
    encode_pooling_bytes,
    encode_pooling_output,
)
53
54
55
56

logger = init_logger(__name__)


57
class EmbeddingMixin(OpenAIServing):
58
59
60
61
62
63
64
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        pooler_config = self.model_config.pooler_config

        # Avoid repeated attribute lookups
        self.supports_chunked_processing = bool(
65
66
67
68
69
70
71
            pooler_config and pooler_config.enable_chunked_processing
        )
        self.max_embed_len = (
            pooler_config.max_embed_len
            if pooler_config and pooler_config.max_embed_len
            else None
        )
72

73
    @override
74
    async def _preprocess(
75
        self,
76
        ctx: ServeContext,
77
    ) -> ErrorResponse | None:
78
        ctx = cast(EmbeddingServeContext, ctx)
79
        try:
80
            ctx.lora_request = self._maybe_get_adapters(ctx.request)
81

82
            tokenizer = await self.engine_client.get_tokenizer()
83
            renderer = self._get_renderer(tokenizer)
84

85
            if isinstance(ctx.request, EmbeddingChatRequest):
86
87
                (
                    _,
88
                    _,
89
                    ctx.engine_prompts,
90
                ) = await self._preprocess_chat(
91
                    ctx.request,
92
                    tokenizer,
93
                    ctx.request.messages,
94
95
                    chat_template=ctx.request.chat_template or ctx.chat_template,
                    chat_template_content_format=ctx.chat_template_content_format,
96
                    add_generation_prompt=ctx.request.add_generation_prompt,
97
                    continue_final_message=False,
98
                    add_special_tokens=ctx.request.add_special_tokens,
99
100
                )
            else:
101
102
                ctx.engine_prompts = await renderer.render_prompt(
                    prompt_or_prompts=ctx.request.input,
103
                    config=self._build_render_config(ctx.request),
104
                )
105
            return None
106
        except (ValueError, TypeError) as e:
107
108
            logger.exception("Error in preprocessing prompt inputs")
            return self.create_error_response(str(e))
109

110
    def _build_render_config(self, request: EmbeddingCompletionRequest) -> RenderConfig:
111
112
113
114
115
116
117
118
119
        # Set max_length based on chunked processing capability
        if self._should_use_chunked_processing(request):
            max_length = None
        else:
            max_length = self.max_embed_len or self.max_model_len

        return RenderConfig(
            max_length=max_length,
            truncate_prompt_tokens=request.truncate_prompt_tokens,
120
121
            add_special_tokens=request.add_special_tokens,
        )
122

123
    @override
124
    def _build_response(
125
        self,
126
        ctx: ServeContext,
127
    ) -> EmbeddingResponse | Response | ErrorResponse:
128
        final_res_batch_checked = cast(list[PoolingRequestOutput], ctx.final_res_batch)
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
        encoding_format: EncodingFormat = ctx.request.encoding_format
        embed_dtype: EmbedDType = ctx.request.embed_dtype
        endianness: Endianness = ctx.request.endianness

        def encode_float_base64():
            items: list[EmbeddingResponseData] = []
            num_prompt_tokens = 0

            for idx, final_res in enumerate(final_res_batch_checked):
                item = EmbeddingResponseData(
                    index=idx,
                    embedding=encode_pooling_output(
                        final_res,
                        encoding_format=encoding_format,
                        embed_dtype=embed_dtype,
                        endianness=endianness,
                    ),
                )
                prompt_token_ids = final_res.prompt_token_ids

                items.append(item)
                num_prompt_tokens += len(prompt_token_ids)

            usage = UsageInfo(
                prompt_tokens=num_prompt_tokens,
                total_tokens=num_prompt_tokens,
156
157
            )

158
159
160
161
162
163
164
            return EmbeddingResponse(
                id=ctx.request_id,
                created=ctx.created_time,
                model=ctx.model_name,
                data=items,
                usage=usage,
            )
165

166
167
168
169
170
171
        def encode_bytes():
            body, items, usage = encode_pooling_bytes(
                pooling_outputs=final_res_batch_checked,
                embed_dtype=embed_dtype,
                endianness=endianness,
            )
172

173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
            metadata = {
                "id": ctx.request_id,
                "created": ctx.created_time,
                "model": ctx.model_name,
                "data": items,
                "usage": usage,
            }
            return EmbeddingBytesResponse(
                body=body,
                metadata=json.dumps(metadata),
            )

        if encoding_format == "float" or encoding_format == "base64":
            return encode_float_base64()
        elif encoding_format == "bytes":
            return encode_bytes()
        else:
            assert_never(encoding_format)
191

192
193
194
195
196
197
    def _get_max_position_embeddings(self) -> int:
        """Get the model's effective maximum sequence length for chunking."""
        return self.model_config.max_model_len

    def _should_use_chunked_processing(self, request) -> bool:
        """Check if chunked processing should be used for this request."""
198
199
200
201
        return (
            isinstance(request, (EmbeddingCompletionRequest, EmbeddingChatRequest))
            and self.supports_chunked_processing
        )
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218

    async def _process_chunked_request(
        self,
        ctx: EmbeddingServeContext,
        original_prompt: TextTokensPrompt,
        pooling_params,
        trace_headers,
        prompt_idx: int,
    ) -> list[AsyncGenerator[PoolingRequestOutput, None]]:
        """Process a single prompt using chunked processing."""
        generators: list[AsyncGenerator[PoolingRequestOutput, None]] = []
        token_ids = original_prompt["prompt_token_ids"]

        # Split into chunks using max_position_embeddings
        max_pos_embeddings = self._get_max_position_embeddings()
        # Process all chunks for MEAN aggregation
        for chunk_idx, chunk_tokens in enumerate(
219
220
            chunk_list(token_ids, max_pos_embeddings)
        ):
221
            # Create a request ID for this chunk
222
            chunk_request_id = f"{ctx.request_id}-prompt-{prompt_idx}-chunk-{chunk_idx}"
223
224

            # Create engine prompt for this chunk
225
            chunk_engine_prompt = EngineTokensPrompt(prompt_token_ids=chunk_tokens)
226
227
228
229

            # Create chunk request prompt for logging
            chunk_text = ""
            chunk_request_prompt = TextTokensPrompt(
230
231
                prompt=chunk_text, prompt_token_ids=chunk_tokens
            )
232
233

            # Log the chunk
234
235
236
237
238
239
            self._log_inputs(
                chunk_request_id,
                chunk_request_prompt,
                params=pooling_params,
                lora_request=ctx.lora_request,
            )
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

            # Create generator for this chunk and wrap it to return indices
            original_generator = self.engine_client.encode(
                chunk_engine_prompt,
                pooling_params,
                chunk_request_id,
                lora_request=ctx.lora_request,
                trace_headers=trace_headers,
                priority=getattr(ctx.request, "priority", 0),
            )

            generators.append(original_generator)

        return generators

    def _validate_input(
        self,
        request,
        input_ids: list[int],
        input_text: str,
    ) -> TextTokensPrompt:
        """Override to support chunked processing for embedding requests."""
        token_num = len(input_ids)

        # Note: EmbeddingRequest doesn't have max_tokens
265
        if isinstance(request, (EmbeddingCompletionRequest, EmbeddingChatRequest)):
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
            # Check if chunked processing is enabled for pooling models
            enable_chunked = self._should_use_chunked_processing(request)

            # Use max_position_embeddings for chunked processing decisions
            max_pos_embeddings = self._get_max_position_embeddings()

            # Determine the effective max length for validation
            if self.max_embed_len is not None:
                # Use max_embed_len for validation instead of max_model_len
                length_type = "maximum embedding input length"
                max_length_value = self.max_embed_len
            else:
                # Fall back to max_model_len validation (original behavior)
                length_type = "maximum context length"
                max_length_value = self.max_model_len

            validation_error_msg = (
                "This model's {length_type} is {max_length_value} tokens. "
                "However, you requested {token_num} tokens in the input for "
285
286
                "embedding generation. Please reduce the length of the input."
            )
287
288
289
290
291

            chunked_processing_error_msg = (
                "This model's {length_type} is {max_length_value} tokens. "
                "However, you requested {token_num} tokens in the input for "
                "embedding generation. Please reduce the length of the input "
292
293
                "or enable chunked processing."
            )
294
295
296
297
298
299
300

            # Check if input exceeds max length
            if token_num > max_length_value:
                raise ValueError(
                    validation_error_msg.format(
                        length_type=length_type,
                        max_length_value=max_length_value,
301
302
303
                        token_num=token_num,
                    )
                )
304
305
306
307
308
309
310
311

            # Check for chunked processing
            # when exceeding max_position_embeddings
            if token_num > max_pos_embeddings:
                if enable_chunked:
                    # Allow long inputs when chunked processing is enabled
                    logger.info(
                        "Input length %s exceeds max_position_embeddings "
312
313
314
315
                        "%s, will use chunked processing",
                        token_num,
                        max_pos_embeddings,
                    )
316
317
318
319
320
                else:
                    raise ValueError(
                        chunked_processing_error_msg.format(
                            length_type="maximum position embeddings length",
                            max_length_value=max_pos_embeddings,
321
322
323
                            token_num=token_num,
                        )
                    )
324

325
            return TextTokensPrompt(prompt=input_text, prompt_token_ids=input_ids)
326
327
328
329
330
331

        # For other request types, use the parent's implementation
        return super()._validate_input(request, input_ids, input_text)

    def _is_text_tokens_prompt(self, prompt) -> bool:
        """Check if a prompt is a TextTokensPrompt (has prompt_token_ids)."""
332
333
334
335
336
        return (
            isinstance(prompt, dict)
            and "prompt_token_ids" in prompt
            and "prompt_embeds" not in prompt
        )
337
338
339
340

    async def _create_single_prompt_generator(
        self,
        ctx: EmbeddingServeContext,
341
        engine_prompt: EngineTokensPrompt,
342
        pooling_params: PoolingParams,
343
        trace_headers: Mapping[str, str] | None,
344
        prompt_index: int,
345
    ) -> AsyncGenerator[RequestOutput | PoolingRequestOutput, None]:
346
347
348
        """Create a generator for a single prompt using standard processing."""
        request_id_item = f"{ctx.request_id}-{prompt_index}"

349
350
351
352
353
354
        self._log_inputs(
            request_id_item,
            engine_prompt,
            params=pooling_params,
            lora_request=ctx.lora_request,
        )
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369

        # Return the original generator without wrapping
        return self.engine_client.encode(
            engine_prompt,
            pooling_params,
            request_id_item,
            lora_request=ctx.lora_request,
            trace_headers=trace_headers,
            priority=getattr(ctx.request, "priority", 0),
        )

    @override
    async def _prepare_generators(
        self,
        ctx: ServeContext,
370
    ) -> ErrorResponse | None:
371
372
373
374
375
376
377
378
379
380
381
        """Override to support chunked processing."""
        ctx = cast(EmbeddingServeContext, ctx)

        # Check if we should use chunked processing
        use_chunked = self._should_use_chunked_processing(ctx.request)

        # If no chunked processing needed, delegate to parent class
        if not use_chunked:
            return await super()._prepare_generators(ctx)

        # Custom logic for chunked processing
382
        generators: list[
383
            AsyncGenerator[RequestOutput | PoolingRequestOutput, None]
384
        ] = []
385
386

        try:
387
388
389
390
391
            trace_headers = (
                None
                if ctx.raw_request is None
                else await self._get_trace_headers(ctx.raw_request.headers)
            )
392
393
394
395
396
397
398
399
400
401
402
403

            pooling_params = self._create_pooling_params(ctx)
            if isinstance(pooling_params, ErrorResponse):
                return pooling_params

            # Verify and set the task for pooling params
            try:
                pooling_params.verify("embed", self.model_config)
            except ValueError as e:
                return self.create_error_response(str(e))

            if ctx.engine_prompts is None:
404
                return self.create_error_response("Engine prompts not available")
405
406
407
408
409

            max_pos_embeddings = self._get_max_position_embeddings()

            for i, engine_prompt in enumerate(ctx.engine_prompts):
                # Check if this specific prompt needs chunked processing
410
                if self._is_text_tokens_prompt(engine_prompt):
411
412
                    # Cast to TextTokensPrompt since we've verified
                    # prompt_token_ids
413
                    text_tokens_prompt = cast(TextTokensPrompt, engine_prompt)
414
                    if len(text_tokens_prompt["prompt_token_ids"]) > max_pos_embeddings:
415
416
                        # Use chunked processing for this prompt
                        chunk_generators = await self._process_chunked_request(
417
418
                            ctx, text_tokens_prompt, pooling_params, trace_headers, i
                        )
419
420
421
422
423
                        generators.extend(chunk_generators)
                        continue

                # Normal processing for short prompts or non-token prompts
                generator = await self._create_single_prompt_generator(
424
425
                    ctx, engine_prompt, pooling_params, trace_headers, i
                )
426
427
428
429
430
431
432
433
434
435
436
437
438
439
                generators.append(generator)

            ctx.result_generator = merge_async_iterators(*generators)

            return None

        except Exception as e:
            # TODO: Use a vllm-specific Validation Error
            return self.create_error_response(str(e))

    @override
    async def _collect_batch(
        self,
        ctx: ServeContext,
440
    ) -> ErrorResponse | None:
441
442
        """Collect and aggregate batch results
        with support for chunked processing.
443
444

        For chunked requests, performs online aggregation to
445
446
447
448
449
450
        minimize memory usage.
        For regular requests, collects results normally.
        """
        ctx = cast(EmbeddingServeContext, ctx)
        try:
            if ctx.engine_prompts is None:
451
                return self.create_error_response("Engine prompts not available")
452
453
454
455
456
457
458
459

            # Check if we used chunked processing
            use_chunked = self._should_use_chunked_processing(ctx.request)

            if not use_chunked:
                return await super()._collect_batch(ctx=ctx)

            if ctx.result_generator is None:
460
                return self.create_error_response("Result generator not available")
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480

            # Online aggregation for chunked requests to
            # minimize memory usage
            # Track aggregation state for each prompt
            prompt_aggregators: dict[int, dict[str, Any]] = {}
            short_prompts_results: dict[int, PoolingRequestOutput] = {}

            async for result_idx, result in ctx.result_generator:
                if "-chunk-" in result.request_id:
                    # Extract prompt_idx from chunked request_id
                    parts = result.request_id.split("-")
                    try:
                        prompt_idx = int(parts[parts.index("prompt") + 1])
                    except (ValueError, IndexError):
                        # Fallback: extract from result_idx if parsing fails
                        prompt_idx = result_idx

                    # Initialize aggregator for this prompt if needed
                    if prompt_idx not in prompt_aggregators:
                        prompt_aggregators[prompt_idx] = {
481
482
483
484
                            "weighted_sum": None,
                            "total_weight": 0,
                            "chunk_count": 0,
                            "request_id": result.request_id.split("-chunk-")[0],
485
486
487
488
489
490
491
492
493
494
495
                        }

                    aggregator = prompt_aggregators[prompt_idx]

                    # MEAN pooling with online weighted averaging
                    # Ensure result is PoolingRequestOutput
                    # for embedding processing
                    if not isinstance(result, PoolingRequestOutput):
                        return self.create_error_response(
                            f"Expected PoolingRequestOutput for "
                            f"chunked embedding, got "
496
497
                            f"{type(result).__name__}"
                        )
498
499
500

                    # Handle both PoolingOutput and
                    # EmbeddingOutput types
501
                    if hasattr(result.outputs, "data"):
502
503
                        # PoolingOutput case
                        embedding_data = result.outputs.data
504
                    elif hasattr(result.outputs, "embedding"):
505
506
507
508
509
                        # EmbeddingOutput case -
                        # convert embedding list to tensor
                        embedding_data = result.outputs.embedding
                    else:
                        return self.create_error_response(
510
511
                            f"Unsupported output type: {type(result.outputs).__name__}"
                        )
512
513

                    if not isinstance(embedding_data, torch.Tensor):
514
515
516
                        embedding_data = torch.tensor(
                            embedding_data, dtype=torch.float32
                        )
517
518
519

                    if result.prompt_token_ids is None:
                        return self.create_error_response(
520
521
                            "prompt_token_ids cannot be None for chunked processing"
                        )
522
523
                    weight = len(result.prompt_token_ids)

524
                    weighted_embedding = embedding_data.to(dtype=torch.float32) * weight
525

526
                    if aggregator["weighted_sum"] is None:
527
                        # First chunk
528
                        aggregator["weighted_sum"] = weighted_embedding
529
530
                    else:
                        # Accumulate
531
                        aggregator["weighted_sum"] += weighted_embedding
532

533
534
                    aggregator["total_weight"] += weight
                    aggregator["chunk_count"] += 1
535
536
537
538
539
540
541
542
543
544
                else:
                    # Non-chunked result - extract prompt_idx from request_id
                    parts = result.request_id.split("-")
                    try:
                        # Last part should be prompt index
                        prompt_idx = int(parts[-1])
                    except (ValueError, IndexError):
                        prompt_idx = result_idx  # Fallback to result_idx

                    short_prompts_results[prompt_idx] = cast(
545
546
                        PoolingRequestOutput, result
                    )
547
548

            # Finalize aggregated results
549
            final_res_batch: list[PoolingRequestOutput | EmbeddingRequestOutput] = []
550
551
552
553
554
555
556
            num_prompts = len(ctx.engine_prompts)

            for prompt_idx in range(num_prompts):
                if prompt_idx in prompt_aggregators:
                    # Finalize MEAN aggregation for this chunked prompt
                    aggregator = prompt_aggregators[prompt_idx]

557
558
                    weighted_sum = aggregator["weighted_sum"]
                    total_weight = aggregator["total_weight"]
559

560
561
562
563
564
565
                    if (
                        weighted_sum is not None
                        and isinstance(weighted_sum, torch.Tensor)
                        and isinstance(total_weight, (int, float))
                        and total_weight > 0
                    ):
566
567
568
569
570
                        # Compute final mean embedding
                        final_embedding = weighted_sum / total_weight

                        # Create a PoolingRequestOutput
                        # for the aggregated result
571
                        pooling_output_data = PoolingOutput(data=final_embedding)
572
573

                        # Get original prompt token IDs for this prompt
574
                        original_prompt = ctx.engine_prompts[prompt_idx]
575
576
                        if not self._is_text_tokens_prompt(original_prompt):
                            return self.create_error_response(
577
578
                                f"Chunked prompt {prompt_idx} is not a TextTokensPrompt"
                            )
579

580
581
582
                        original_token_ids = cast(TextTokensPrompt, original_prompt)[
                            "prompt_token_ids"
                        ]
583
584

                        pooling_request_output = PoolingRequestOutput(
585
                            request_id=aggregator["request_id"],
586
587
                            prompt_token_ids=original_token_ids,
                            outputs=pooling_output_data,
588
                            num_cached_tokens=0,
589
590
                            finished=True,
                        )
591
592
593
594

                        final_res_batch.append(pooling_request_output)
                    else:
                        return self.create_error_response(
595
596
                            f"Failed to aggregate chunks for prompt {prompt_idx}"
                        )
597
598
                elif prompt_idx in short_prompts_results:
                    final_res_batch.append(
599
600
                        cast(PoolingRequestOutput, short_prompts_results[prompt_idx])
                    )
601
602
                else:
                    return self.create_error_response(
603
604
                        f"Result not found for prompt {prompt_idx}"
                    )
605
606

            ctx.final_res_batch = cast(
607
                list[RequestOutput | PoolingRequestOutput], final_res_batch
608
            )
609
610
611
612
613
614

            return None

        except Exception as e:
            return self.create_error_response(str(e))

615
616
617
618
619
620
621
622
623

class OpenAIServingEmbedding(EmbeddingMixin):
    request_id_prefix = "embd"

    def __init__(
        self,
        engine_client: EngineClient,
        models: OpenAIServingModels,
        *,
624
625
        request_logger: RequestLogger | None,
        chat_template: str | None,
626
        chat_template_content_format: ChatTemplateContentFormatOption,
627
        trust_request_chat_template: bool = False,
628
        log_error_stack: bool = False,
629
    ) -> None:
630
631
632
633
634
635
        super().__init__(
            engine_client=engine_client,
            models=models,
            request_logger=request_logger,
            log_error_stack=log_error_stack,
        )
636
637
638

        self.chat_template = chat_template
        self.chat_template_content_format: Final = chat_template_content_format
639
        self.trust_request_chat_template = trust_request_chat_template
640
641
642
643

    async def create_embedding(
        self,
        request: EmbeddingRequest,
644
645
        raw_request: Request | None = None,
    ) -> EmbeddingResponse | ErrorResponse:
646
647
648
649
650
651
        """
        Embedding API similar to OpenAI's API.

        See https://platform.openai.com/docs/api-reference/embeddings/create
        for the API specification. This API mimics the OpenAI Embedding API.
        """
652
        model_name = self.models.model_name()
653
654
        request_id = (
            f"{self.request_id_prefix}-"
655
656
            f"{self._base_request_id(raw_request, request.request_id)}"
        )
657
658
659
660
661
662
663
664
665
666
667
668

        ctx = EmbeddingServeContext(
            request=request,
            raw_request=raw_request,
            model_name=model_name,
            request_id=request_id,
            chat_template=self.chat_template,
            chat_template_content_format=self.chat_template_content_format,
        )

        return await super().handle(ctx)  # type: ignore

669
670
671
672
    @override
    def _create_pooling_params(
        self,
        ctx: ServeContext[EmbeddingRequest],
673
    ) -> PoolingParams | ErrorResponse:
674
675
676
        pooling_params = super()._create_pooling_params(ctx)
        if isinstance(pooling_params, ErrorResponse):
            return pooling_params
677
678

        try:
679
            pooling_params.verify("embed", self.model_config)
680
681
682
        except ValueError as e:
            return self.create_error_response(str(e))

683
        return pooling_params
684
685
686
687

    async def _preprocess(
        self,
        ctx: ServeContext,
688
    ) -> ErrorResponse | None:
689
690
691
692
693
694
695
696
697
        if isinstance(ctx.request, EmbeddingChatRequest):
            error_check_ret = self._validate_chat_template(
                request_chat_template=ctx.request.chat_template,
                chat_template_kwargs=ctx.request.chat_template_kwargs,
                trust_request_chat_template=self.trust_request_chat_template,
            )
            if error_check_ret is not None:
                return error_check_ret
        return await super()._preprocess(ctx)