diff --git a/examples/inference/gpt/gpt_dynamic_inference_12b.sh b/examples/inference/gpt/gpt_dynamic_inference_12b.sh index f0ed226987e..a16fe5176d5 100644 --- a/examples/inference/gpt/gpt_dynamic_inference_12b.sh +++ b/examples/inference/gpt/gpt_dynamic_inference_12b.sh @@ -33,6 +33,7 @@ export CUDA_DEVICE_MAX_CONNECTIONS=1 : ${CUDA_GRAPH_SHARE_IO_BUFFERS=1} # Miscellaneous. +: ${USE_COORDINATOR=0} : ${ENGINE=dynamic} : ${EXTRA_ARGS=""} # NSIGHT_PREFIX=/path/to/nsight/profile @@ -85,7 +86,7 @@ ARGS=" \ " # Cuda graphs. -if [ "${CUDA_GRAPH_IMPL}" = "local" ]; then +if [ "${NUM_CUDA_GRAPHS}" != "0" ]; then ARGS+=" \ --cuda-graph-impl local \ --inference-dynamic-batching-num-cuda-graphs ${NUM_CUDA_GRAPHS} \ @@ -108,7 +109,12 @@ else fi # Command. -CMD="python -m examples.inference.gpt.gpt_${ENGINE}_inference ${ARGS}" +if [[ "${USE_COORDINATOR}" == "0" ]]; then + CMD="python -m examples.inference.gpt.gpt_${ENGINE}_inference ${ARGS}" +else + CMD="python -um examples.inference.gpt.gpt_${ENGINE}_inference_with_coordinator ${ARGS}" +fi + if [[ -v NSIGHT_PREFIX ]]; then CMD="nsys profile -s none -t nvtx,cuda --cudabacktrace=all --cuda-graph-trace=node --python-backtrace=cuda --wait all -o ${NSIGHT_PREFIX} --force-overwrite true --capture-range=cudaProfilerApi --capture-range-end=stop ${CMD}" fi diff --git a/examples/inference/gpt/gpt_dynamic_inference_357m.sh b/examples/inference/gpt/gpt_dynamic_inference_357m.sh index 46211a4adcf..c095371714f 100644 --- a/examples/inference/gpt/gpt_dynamic_inference_357m.sh +++ b/examples/inference/gpt/gpt_dynamic_inference_357m.sh @@ -34,6 +34,7 @@ export CUDA_DEVICE_MAX_CONNECTIONS=1 : ${CUDA_GRAPH_SHARE_IO_BUFFERS=1} # Miscellaneous. +: ${USE_COORDINATOR=0} : ${ENGINE=dynamic} : ${EXTRA_ARGS=""} # NSIGHT_PREFIX=/path/to/nsight/profile @@ -71,7 +72,7 @@ ARGS=" \ " # Cuda graphs. -if [ "${CUDA_GRAPH_IMPL}" = "local" ]; then +if [ "${NUM_CUDA_GRAPHS}" != "0" ]; then ARGS+=" \ --cuda-graph-impl local \ --inference-dynamic-batching-num-cuda-graphs ${NUM_CUDA_GRAPHS} \ @@ -94,7 +95,12 @@ else fi # Command. -CMD="python -m examples.inference.gpt.gpt_${ENGINE}_inference ${ARGS}" +if [[ "${USE_COORDINATOR}" == "0" ]]; then + CMD="python -m examples.inference.gpt.gpt_${ENGINE}_inference ${ARGS}" +else + CMD="python -um examples.inference.gpt.gpt_${ENGINE}_inference_with_coordinator ${ARGS}" +fi + if [[ -v NSIGHT_PREFIX ]]; then CMD="nsys profile -s none -t nvtx,cuda --cudabacktrace=all --cuda-graph-trace=node --python-backtrace=cuda --wait all -o ${NSIGHT_PREFIX} --force-overwrite true --capture-range=cudaProfilerApi --capture-range-end=stop ${CMD}" fi diff --git a/megatron/core/inference/contexts/dynamic_context.py b/megatron/core/inference/contexts/dynamic_context.py index d304226f34a..db4866d1142 100644 --- a/megatron/core/inference/contexts/dynamic_context.py +++ b/megatron/core/inference/contexts/dynamic_context.py @@ -1,4 +1,4 @@ -# Copyright (c) 2025, NVIDIA CORPORATION. All rights reserved. +# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. import math import warnings @@ -59,8 +59,10 @@ def __init__( self, request_id: Optional[int], message: Optional[str] = None, *, is_transient: bool = True ): request_str = '--' if request_id is None else str(request_id) - message = "" if message is None else f" | {message}" - super().__init__(f"request {request_str}{message}") + _message = "" if message is None else f" | {message}" + super().__init__(f"request {request_str}{_message}") + self.request_id = request_id + self.message = message self.is_transient = is_transient @@ -102,6 +104,50 @@ def __init__(self, max_request_count, active_request_count): ) +class ContextErrorFactory: + """Factory class for serializing/deserializing context errors.""" + + @classmethod + def serialize(cls, error: ContextOverflowError) -> dict: + """Serialize error. + + Args: + error (ContextOverflowError): Error. + + Returns: + (dict) Serialized error data. + """ + assert isinstance(error, ContextOverflowError) + return { + "type": type(error).__name__, + "request_id": error.request_id, + "message": error.message, + "is_transient": error.is_transient, + } + + @classmethod + def deserialize(cls, obj: dict) -> ContextOverflowError: + """Deserialize error. + + Args: + obj (dict): Serialized error data. + + Returns: + (ContextOverflowError) Deserialized error. + """ + error_cls = { + "ContextOverflowError": ContextOverflowError, + "RequestOverflowError": RequestOverflowError, + "TokenOverflowError": TokenOverflowError, + "MaxSequenceLengthOverflowError": MaxSequenceLengthOverflowError, + "BlockOverflowError": BlockOverflowError, + "ActiveRequestCountOverflowError": ActiveRequestCountOverflowError, + }[obj["type"]] + error = ContextOverflowError(**{k: v for k, v in obj.items() if k != "type"}) + error.__class__ = error_cls # todo (@lmcafe): better/safer alternative? + return error + + class WarmupEngineMode(Enum): """Enumeration for warmup engine modes used during cuda graph capture.""" diff --git a/megatron/core/inference/data_parallel_inference_coordinator.py b/megatron/core/inference/data_parallel_inference_coordinator.py index 101acb31e1f..ea0560183d8 100644 --- a/megatron/core/inference/data_parallel_inference_coordinator.py +++ b/megatron/core/inference/data_parallel_inference_coordinator.py @@ -1,14 +1,13 @@ -# Copyright (c) 2025, NVIDIA CORPORATION. All rights reserved. +# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. import logging from collections import deque -from itertools import cycle, repeat -from typing import List, Tuple +from itertools import cycle +from multiprocessing import Event import torch from megatron.core.inference.headers import Headers -from megatron.core.inference.inference_request import DynamicInferenceRequest try: import zmq @@ -48,7 +47,6 @@ class DataParallelInferenceCoordinator: from a client to all connected data parallel ranks. Attributes: - tokenizer: The tokenizer object for encoding prompts. router_socket (zmq.Socket): The central ZMQ ROUTER socket for all communication. data_parallel_size (int): The number of data parallel workers to expect. identities_of_data_parallel_ranks (deque): A deque holding the ZMQ @@ -58,11 +56,9 @@ class DataParallelInferenceCoordinator: request_id_to_client_request_id (dict): Maps server-side request IDs to the original request ID provided by the client. next_request_id (int): A counter for generating unique server-side request IDs. - requests (dict): A store for active `DynamicInferenceRequest` objects, keyed by - server-side request ID. """ - def __init__(self, tokenizer, inference_coordinator_port: int, data_parallel_size: int): + def __init__(self, inference_coordinator_port: int, data_parallel_size: int): """ Initializes the inference coordinator. @@ -71,8 +67,6 @@ def __init__(self, tokenizer, inference_coordinator_port: int, data_parallel_siz ranks to connect before proceeding. Args: - tokenizer: An object with `tokenize`, `detokenize`, and `bos` attributes, - used for processing text. inference_coordinator_port (int): The TCP port number to bind the server to. data_parallel_size (int): The number of TP-coordinator workers that are expected to connect. @@ -86,7 +80,6 @@ def __init__(self, tokenizer, inference_coordinator_port: int, data_parallel_siz "pip install msgpack" ) self.context = zmq.Context() - self.tokenizer = tokenizer # This is the central router socket # 1. data parallel ranks connect to this socket to register themselves @@ -114,7 +107,6 @@ def __init__(self, tokenizer, inference_coordinator_port: int, data_parallel_siz self.request_id_to_client_request_id = {} self.next_request_id = 0 - self.requests = {} def get_next_data_parallel_rank(self): """ @@ -125,121 +117,6 @@ def get_next_data_parallel_rank(self): """ return next(self.data_parallel_rank_iterator) - def tokenize_prompt( - self, prompt: str, add_BOS: bool = False - ) -> Tuple[torch.Tensor, torch.Tensor]: - """Utility to tokenize the input prompts - - Args: - prompt (str): The input prompt - - Returns: - torch.Tensor: Returns the tokenized prompt - """ - prompt_tokens = self.tokenizer.tokenize(prompt) - - if add_BOS: - prompt_tokens = [self.tokenizer.bos] + prompt_tokens - - return prompt_tokens - - def postprocess( - self, - request_ids: List[int], - finished_request_ids: List[int], - generated_tokens: List[int], - log_probs: List[int], - chunked_prefill_request_id: int = -1, - materialize_only_last_token_logits: bool = True, - ): - """ - Processes replies from the engine, appending tokens and handling finished requests. - - For each generated token, this method appends it to the corresponding active - request. If a request is marked as finished, it detokenizes the full - sequence, sends the final result back to the original client, and cleans - up the request state. - - Args: - request_ids (List[int]): A list of request IDs that have new tokens. - finished_request_ids (List[int]): A list of request IDs that have completed - generation in this step. - generated_tokens (List[int]): The list of new tokens, one for each ID in - `request_ids`. - log_probs (List[int]): Log probabilities for each token. - chunked_prefill_request_id (int): The request ID currently undergoing chunked prefill, - -1 if no chunked prefill is active. - """ - # Todo [Siddharth]: This is duplicated logic from the engine. - # We should refactor this to avoid duplication. - log_probs_iter = log_probs if log_probs else repeat(None) - for request_id, token, request_log_probs in zip( - request_ids, generated_tokens, log_probs_iter - ): - request: DynamicInferenceRequest = self.requests[request_id] - # Handle chunked prefill similar to the engine logic - if chunked_prefill_request_id == -1 or request_id != chunked_prefill_request_id: - request.generated_tokens.append(token) - - if request_log_probs is not None: - if not request.prompt_log_probs: - request.prompt_log_probs = [] - if not request.generated_log_probs: - request.generated_log_probs = [] - # If the request log probs span > 1 token we are in prefill - if len(request_log_probs) > 1: - request.prompt_log_probs.extend(request_log_probs) - else: - if ( - # If it is a chunked prefill request - len(request.prompt_log_probs) > 0 - # And we are missing the last token for prefill - and len(request.prompt_log_probs) < len(request.prompt_tokens) - # And we need to track full prefill - and not materialize_only_last_token_logits - ): - assert ( - len(request.prompt_log_probs) == len(request.prompt_tokens) - 1 - ), "Prompt log probs length is not equal to prompt tokens length - 1" - request.prompt_log_probs.extend(request_log_probs) - else: - request.generated_log_probs.extend(request_log_probs) - else: - # This is the chunked prefill request, handle log probs but don't append tokens - if request_log_probs is not None: - if materialize_only_last_token_logits: - # Here we discard intermediate log probs, - # as we only materialize the last token log probs - request.prompt_log_probs = [] - request.generated_log_probs = [] - else: - # Otherwise, we gather log probs for all tokens - if not request.prompt_log_probs: - request.prompt_log_probs = [] - request.prompt_log_probs.extend(request_log_probs) - request.generated_log_probs = [] - - if finished_request_ids: - for fid in finished_request_ids: - if fid == chunked_prefill_request_id: - continue # skip chunked prefill request, this is not a finished request - request = self.requests.pop(fid) - request.generated_length = len(request.generated_tokens) - request.generated_text = self.tokenizer.detokenize(request.generated_tokens) - - client_identity = self.request_id_to_client_id[fid] - client_request_identity = self.request_id_to_client_request_id[fid] - del self.request_id_to_client_id[fid] - del self.request_id_to_client_request_id[fid] - self.router_socket.send_multipart( - [ - client_identity, - msgpack.packb( - [client_request_identity, request.serializable()], use_bin_type=True - ), - ] - ) - def start(self): """ Starts the main event loop for the coordinator. @@ -291,27 +168,20 @@ def start(self): self.request_id_to_client_id[request_id] = sender_identity self.request_id_to_client_request_id[request_id] = client_request_id - # tokenize the prompt if it is a string. - if isinstance(prompt, str): - prompt_tokens = self.tokenize_prompt(prompt) + # Serialize prompt. + if isinstance(prompt, (str, list)): + pass + elif isinstance(prompt, torch.Tensor): + prompt = prompt.tolist() else: - prompt_tokens = prompt # no error handling here as it is done in the engine. - - self.requests[request_id] = DynamicInferenceRequest( - request_id=request_id, prompt=prompt, prompt_tokens=prompt_tokens - ) + raise Exception("specialize for <%s> prompt." % type(prompt).__name__) next_data_parallel_rank_identity = self.get_next_data_parallel_rank() self.router_socket.send_multipart( [ next_data_parallel_rank_identity, msgpack.packb( - [ - Headers.SUBMIT_REQUEST.value, - request_id, - prompt_tokens, - sampling_params, - ], + [Headers.SUBMIT_REQUEST.value, request_id, prompt, sampling_params], use_bin_type=True, ), ] @@ -328,26 +198,27 @@ def start(self): elif header == Headers.ENGINE_REPLY: # This is the output of a single engine step on some data parallel rank. assert sender_identity in self.identities_of_data_parallel_ranks - ( - request_ids, - finished_request_ids, - generated_tokens, - logprobs, - chunked_prefill_request_id, - materialize_only_last_token_logits, - ) = deserialized_payload[1:] - self.postprocess( - request_ids, - finished_request_ids, - generated_tokens, - logprobs, - chunked_prefill_request_id, - materialize_only_last_token_logits, - ) + finished_requests = deserialized_payload[1] + + for finished_request in finished_requests: + fid = finished_request["request_id"] + client_identity = self.request_id_to_client_id[fid] + client_request_identity = self.request_id_to_client_request_id[fid] + del self.request_id_to_client_id[fid] + del self.request_id_to_client_request_id[fid] + + self.router_socket.send_multipart( + [ + client_identity, + msgpack.packb( + [client_request_identity, finished_request], use_bin_type=True + ), + ] + ) @classmethod def entrypoint( - cls, ready_event, tokenizer, inference_coordinator_port: int, data_parallel_size: int + cls, ready_event: Event, inference_coordinator_port: int, data_parallel_size: int ): """ Class method to instantiate and run the coordinator, for use in a separate process. @@ -356,14 +227,12 @@ def entrypoint( that it is fully initialized and listening, and then starts the main event loop. Args: - ready_event: A threading or multiprocessing event object that is set() + ready_event (Event): A threading or multiprocessing event object that is set() once the coordinator is ready to accept connections. - tokenizer: The tokenizer object. inference_coordinator_port (int): The port to bind to. data_parallel_size (int): The number of expected TP-coordinators. """ - tokenizer = tokenizer - coordinator = cls(tokenizer, inference_coordinator_port, data_parallel_size) + coordinator = cls(inference_coordinator_port, data_parallel_size) ready_event.set() try: coordinator.start() diff --git a/megatron/core/inference/engines/dynamic_engine.py b/megatron/core/inference/engines/dynamic_engine.py index 64c6fca7603..b536e66d3da 100644 --- a/megatron/core/inference/engines/dynamic_engine.py +++ b/megatron/core/inference/engines/dynamic_engine.py @@ -1,4 +1,4 @@ -# Copyright (c) 2025, NVIDIA CORPORATION. All rights reserved. +# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. import asyncio import logging @@ -306,7 +306,6 @@ async def start_listening_to_data_parallel_coordinator( target=DataParallelInferenceCoordinator.entrypoint, args=( coordinator_ready_event, - self.controller.tokenizer, inference_coordinator_port, parallel_state.get_data_parallel_world_size(), ), @@ -557,6 +556,10 @@ def post_process_requests( request.generated_length = len(request.generated_tokens) request.status = Status.COMPLETED finished_request = self.requests.pop(request_id) + if finished_request.prompt is None: + finished_request.prompt = self.controller.tokenizer.detokenize( + finished_request.prompt_tokens.tolist() + ) finished_request.generated_length = len(finished_request.generated_tokens) finished_requests.append(finished_request) finished_request.generated_text = self.controller.tokenizer.detokenize( @@ -675,11 +678,7 @@ def schedule_chunked_prefill(self): # Note that we do not need to continue check the queue, as the tokens are full async def async_step( - self, - sampling_params: SamplingParams, - *, - verbose: Optional[bool] = False, - post_process_requests_locally: bool = True, + self, sampling_params: SamplingParams, *, verbose: Optional[bool] = False ) -> Tuple[List[DynamicInferenceRequest], List[DynamicInferenceRequest], float]: """ Wrapper for controller.generate_output_tokens_dynamic_batch(), to @@ -738,16 +737,11 @@ async def async_step( [self.requests[i].add_event_finish() for i in finished_request_ids.tolist()] # Add finished events. - if post_process_requests_locally: - (active_requests, finished_requests) = self.post_process_requests( - active_request_ids, finished_request_ids, step_time, sample, log_probs - ) - else: - return active_request_ids, finished_request_ids, sample, log_probs + (active_requests, finished_requests) = self.post_process_requests( + active_request_ids, finished_request_ids, step_time, sample, log_probs + ) else: - if not post_process_requests_locally: - return None active_requests: List[DynamicInferenceRequest] = [] finished_requests: List[DynamicInferenceRequest] = [] @@ -1009,33 +1003,22 @@ async def run_engine_with_coordinator( continue engine_output = await self.async_step( - sampling_params=sampling_params, - verbose=verbose, - post_process_requests_locally=False, + sampling_params=sampling_params, verbose=verbose ) is_tp0_and_pp0 = ( parallel_state.get_tensor_model_parallel_rank() == 0 and parallel_state.get_pipeline_model_parallel_rank() == 0 ) - if is_tp0_and_pp0 and engine_output is not None: - # return the engine output to the coordinator. The coordinator will take - # care of the post-processing. - request_ids, finished_request_ids, sample, logprobs = engine_output - # Include chunked prefill request id, use -1 if None - chunked_prefill_id = self.context.chunked_prefill_request_id - materialize_only_last_token_logits = ( - self.context.materialize_only_last_token_logits - ) + if ( + is_tp0_and_pp0 + and engine_output is not None + and engine_output["finished_requests"] + ): payload = msgpack.packb( [ Headers.ENGINE_REPLY.value, - request_ids.tolist(), - finished_request_ids.tolist(), - sample.tolist(), - logprobs, - chunked_prefill_id, - materialize_only_last_token_logits, + [r.serializable() for r in engine_output["finished_requests"]], ], use_bin_type=True, ) diff --git a/megatron/core/inference/inference_client.py b/megatron/core/inference/inference_client.py index fc6480c8472..59b9144a207 100644 --- a/megatron/core/inference/inference_client.py +++ b/megatron/core/inference/inference_client.py @@ -1,6 +1,7 @@ -# Copyright (c) 2025, NVIDIA CORPORATION. All rights reserved. +# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. import asyncio +import logging import os import time from typing import List, Union @@ -123,7 +124,7 @@ async def _listen_for_completed_requests(self): request_id ) completion_future = self.completion_futures.pop(request_id) - completion_future.set_result(DynamicInferenceRequest(**reply)) + completion_future.set_result(DynamicInferenceRequest.deserialize(reply)) except zmq.Again: await asyncio.sleep(0.005) continue @@ -150,7 +151,7 @@ async def start(self): the initial handshake and spawns the `listen_for_completed_requests` coroutine. """ - print("Client: Connecting to InferenceCoordinator...") + logging.info("Client: Connecting to InferenceCoordinator...") self._connect_with_inference_coordinator() self.listener_task = asyncio.create_task(self._listen_for_completed_requests()) diff --git a/megatron/core/inference/inference_request.py b/megatron/core/inference/inference_request.py index d4b956e58e2..9fb1f33edd7 100644 --- a/megatron/core/inference/inference_request.py +++ b/megatron/core/inference/inference_request.py @@ -1,6 +1,7 @@ -# Copyright (c) 2025, NVIDIA CORPORATION. All rights reserved. +# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. import copy +import io import time import warnings from dataclasses import asdict, dataclass, field @@ -12,6 +13,22 @@ from megatron.core.inference.sampling_params import SamplingParams +def serialize_tensor(tensor): + """Serialize tensor to bytes.""" + buffer = io.BytesIO() + torch.save(tensor, buffer) + buffer.seek(0) + tensor_bytes = buffer.read() + return tensor_bytes + + +def deserialize_tensor(tensor_bytes): + """Deserialize tensor from bytes.""" + buffer = io.BytesIO(tensor_bytes) + tensor = torch.load(buffer) + return tensor + + # class syntax class Status(Enum): """Enum for status""" @@ -66,7 +83,39 @@ def serializable(self): dict: A dictionary representation of the instance suitable for serialization. """ - return asdict(self) + # Dataclass to dict. + obj = asdict(self) + obj["status"] = self.status.name if self.status else None + + # Serialize tensors. + obj = { + k: (("tensor", serialize_tensor(v)) if isinstance(v, torch.Tensor) else v) + for k, v in obj.items() + } + + return obj + + @classmethod + def deserialize(cls, obj: dict) -> "InferenceRequest": + """Deserialize request. + + Args: + obj (dict): Serialized request data. + + Returns: + (InferenceRequest) Deserialized request. + """ + + # Initialize request. + request = cls(**obj) + request.status = None if obj["status"] is None else Status[obj["status"]] + + # Deserialize tensors. + for k, v in obj.items(): + if isinstance(v, list) and len(v) == 2 and v[0] == "tensor": + setattr(request, k, deserialize_tensor(v[1])) + + return request class DynamicInferenceEventType(Enum): @@ -101,8 +150,8 @@ class DynamicInferenceEvent: def __post_init__(self): # Timestamp. - assert self.timestamp is None, "timestamp automatically set." - self.timestamp = time.time() + if self.timestamp is None: + self.timestamp = time.time() # Validate type. assert isinstance(self.type, DynamicInferenceEventType) @@ -120,6 +169,47 @@ def __str__(self): payload_str = "" if self.payload is None else f", {type(self.payload).__name__}" return f"[{self.timestamp:.3f}] {self.type.name}{payload_str}" + def serialize(self): + """ + Converts the instance into a serializable dictionary. + Returns: + dict: A dictionary representation of the instance suitable for serialization. + """ + + # Dataclass to dict. + obj = asdict(self) + obj["type"] = self.type.name + + # Serialize payload. + if self.payload: + from .contexts.dynamic_context import ContextErrorFactory # avoid circular import. + + obj["payload"] = ContextErrorFactory.serialize(self.payload) + + return obj + + @classmethod + def deserialize(cls, obj: dict) -> "DynamicInferenceEvent": + """Deserialize event. + + Args: + obj (dict): Serialized event data. + + Returns: + (DynamicInferenceEvent) Deserialized event. + """ + + # Initialize event. + event = cls(**{**obj, "type": DynamicInferenceEventType[obj["type"]]}) + + # Deserialize payload. + if obj["payload"]: + from .contexts.dynamic_context import ContextErrorFactory # avoid circular import. + + event.payload = ContextErrorFactory.deserialize(obj["payload"]) + + return event + @dataclass(kw_only=True) class DynamicInferenceRequest(InferenceRequest): @@ -162,6 +252,30 @@ def __str__(self): ) ) + def serializable(self): + """ + Converts the instance into a serializable dictionary. + Returns: + dict: A dictionary representation of the instance suitable for serialization. + """ + obj = super().serializable() + obj["events"] = [e.serialize() for e in self.events] + return obj + + @classmethod + def deserialize(cls, obj: dict) -> "DynamicInferenceRequest": + """Deserialize request. + + Args: + obj (dict): Serialized request data. + + Returns: + (DynamicInferenceRequest) Deserialized request. + """ + request = super().deserialize(obj) + request.events = [DynamicInferenceEvent.deserialize(e) for e in obj["events"]] + return request + def add_event(self, type: DynamicInferenceEventType, payload: Optional[Any] = None) -> None: """Add event.""" self.events.append(DynamicInferenceEvent(type=type, payload=payload))