diff --git a/examples/inference/gpt/gpt_dynamic_inference.py b/examples/inference/gpt/gpt_dynamic_inference.py
index 0fb0e2a8d04..a3b82a69e68 100644
--- a/examples/inference/gpt/gpt_dynamic_inference.py
+++ b/examples/inference/gpt/gpt_dynamic_inference.py
@@ -1,4 +1,4 @@
-# Copyright (c) 2025, NVIDIA CORPORATION. All rights reserved.
+# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 
 import hashlib
 import json
@@ -11,7 +11,7 @@
 from collections import defaultdict
 from functools import partial
 from tqdm import tqdm
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 import torch
 from tqdm import tqdm
@@ -117,8 +117,11 @@ def get_model() -> MegatronModule:
     return model
 
 
-def get_inference_context(requests: List[Request], sampling_params: SamplingParams, 
-                          calculate_max_sequence_length_from_requests: bool =True):
+def get_inference_context(
+    requests: List[Request],
+    sampling_params: Optional[SamplingParams] = None,
+    calculate_max_sequence_length_from_requests: bool = True
+):
     """The inference context manages the KV cache and other inference state."""
 
     args = get_args()
@@ -199,19 +202,28 @@ def get_inference_controller(
 
 
 def run_inference(
-    requests: List[Request], sampling_params: SamplingParams, engine: DynamicInferenceEngine
+    requests: List[Request],
+    engine: DynamicInferenceEngine,
+    sampling_params: Optional[SamplingParams] = None,
 ) -> List[Dict[str, float]]:
     """Add requests to engine and generate tokens.
 
     Args:
         requests (List[Request]): Requests that are to be added and processed.
-        sampling_params (SamplingParams): Sampling params for the logits.
         engine (DynamicInferenceEngine): Inference engine that manages generating tokens.
+        sampling_params (SamplingParams): Deprecated as of megatron-core 0.16.
 
     Return:
         A dictionary of step times with `prefill` and `decode` keys.
     """
 
+    if sampling_params is not None and torch.distributed.get_rank() == 0:
+        warnings.warn(
+            "The `sampling_params` argument is deprecated. "
+            "Sampling parameters are specified per request.",
+            DeprecationWarning,
+        )
+
     args = get_args()
 
     # Initialize request arrival times.
@@ -244,7 +256,7 @@ def _add_request():
         engine.add_request(
             num_requests_added,
             _request.prompt_text,
-            sampling_params.num_tokens_to_generate,
+            _request.sampling_params,
         )
         _request.time_start = get_curr_time()
         _request.state = "started"
@@ -271,7 +283,7 @@ def _add_request():
 
         # Step inference engine (i.e., generate a token for each active request).
         # Before step, we haven't done the scheduling, so we cannot know the is_decode_only
-        result = engine.step_modern(sampling_params, verbose=True)
+        result = engine.step_modern(verbose=True)
         # After step, we lost track of last iteration's is_decode_only, so we need to get it from the engine
         is_decode_only = engine.is_decode_only 
         step_id += 1
@@ -301,7 +313,7 @@ def _add_request():
                 request.output_text = finished_request.generated_text
                 request.state = "finished"
                 request.request_id = finished_request.request_id
-                if sampling_params.return_log_probs:
+                if finished_request.sampling_params.return_log_probs:
                     request.log_probs = (
                         finished_request.prompt_log_probs + finished_request.generated_log_probs
                     )
@@ -349,11 +361,12 @@ def main():
         top_p=args.top_p,
         return_log_probs=args.return_log_probs,
         num_tokens_to_generate=args.num_tokens_to_generate,
+        termination_id=args.termination_id if args.termination_id is not None else tokenizer.eod,
     )
 
     # Requests, context, conroller.
     model = get_model()
-    requests = build_requests(args, tokenizer)
+    requests = build_requests(args, tokenizer, sampling_params)
     context = get_inference_context(requests, sampling_params)
     controller = get_inference_controller(model, context)
 
@@ -371,7 +384,6 @@ def main():
     engine = DynamicInferenceEngine(
         controller,
         context,
-        termination_id=args.termination_id if args.termination_id is not None else tokenizer.eod,
         enable_cuda_graph=args.cuda_graph_impl == "local",
         random_seed=args.seed,
         track_paused_request_events=args.inference_dynamic_batching_track_paused_request_events,
@@ -387,7 +399,7 @@ def main():
     throughputs = []
     for _ in range(args.inference_repeat_n):
         t = get_curr_time()
-        result = run_inference(requests, sampling_params, engine)
+        result = run_inference(requests, engine)
         step_times = result["step_times"]
         add_times = result["add_times"]
         output_times = result["output_times"]
@@ -458,7 +470,7 @@ def escape_str(s):
                         "cuda_graph_request_count_map" : result["cuda_graph_request_count_map"],
                         "step_count" : engine.step_count,
                     }
-                    if sampling_params.return_log_probs:
+                    if req.sampling_params.return_log_probs:
                         response_logprobs = req.log_probs
                         result_dict["logprobs"] = response_logprobs
                     json_results[req.request_id] = result_dict
diff --git a/examples/inference/gpt/gpt_dynamic_inference_with_coordinator.py b/examples/inference/gpt/gpt_dynamic_inference_with_coordinator.py
index 31243497cd4..7b5de5c21f2 100644
--- a/examples/inference/gpt/gpt_dynamic_inference_with_coordinator.py
+++ b/examples/inference/gpt/gpt_dynamic_inference_with_coordinator.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+
 from megatron.core.inference.inference_client import InferenceClient
 from examples.inference.gpt.utils import add_common_inference_args
 import asyncio
@@ -18,13 +20,24 @@
 from megatron.training.arguments import parse_args
 from megatron.core import parallel_state
 
-async def main(engine: DynamicInferenceEngine, requests: List[Request], sampling_params: SamplingParams, port: int):
+async def main(
+    engine: DynamicInferenceEngine,
+    requests: List[Request],
+    port: int,
+    sampling_params: SamplingParams | None = None,
+):
+    if sampling_params is not None:
+        warnings.warn(
+            "The `sampling_params` argument is deprecated. "
+            "Sampling parameters are specified per request.",
+            DeprecationWarning,
+        )
     # once you call engine.start_listening_to_data_parallel_coordinator,
     # the engine will start accepting requests from the data parallel coordinator.
     # and processing them in an asyncio coroutine. 
-    await engine.start_listening_to_data_parallel_coordinator(sampling_params, 
-                                                inference_coordinator_port=port,
-                                                launch_inference_coordinator=True)   
+    await engine.start_listening_to_data_parallel_coordinator( 
+        inference_coordinator_port=port, launch_inference_coordinator=True
+    )
     # if you want to use your own inference coordinator - 
     # 1. set launch_inference_coordinator to False
     # 2. setup a router socket at tcp://MASTER_ADDR:PORT
@@ -50,8 +63,7 @@ async def main(engine: DynamicInferenceEngine, requests: List[Request], sampling
                 # These add-request calls will queue up the request on a zmq socket and return
                 # instantaneously. They will return an asyncio future which can be awaited for
                 # request completion.
-                futures.append(client.add_request(request.prompt_text, 
-                                                        sampling_params))
+                futures.append(client.add_request(request.prompt_text, request.sampling_params))
                 num_requests_added += 1
                 #tbar.update(1)
             if num_requests_added == num_requests_total:
@@ -74,7 +86,7 @@ async def main(engine: DynamicInferenceEngine, requests: List[Request], sampling
                     "generated_tokens": req.generated_tokens,
                     "latency": req.latency, #InferenceClient populates this field in the returned future.
                 }
-                if sampling_params.return_log_probs:
+                if req.sampling_params["return_log_probs"]:
                     result_dict["logprobs"] = req.prompt_log_probs + req.generated_log_probs
                 json_results[req.request_id] = result_dict
             with open(args.output_path, "w") as fp:
@@ -115,13 +127,13 @@ async def main(engine: DynamicInferenceEngine, requests: List[Request], sampling
             top_p=args.top_p,
             return_log_probs=args.return_log_probs,
             num_tokens_to_generate=args.num_tokens_to_generate,
+            termination_id=args.termination_id if args.termination_id is not None else tokenizer.eod,
         )
 
         # Requests, context, conroller.
         model = get_model()
-        requests = build_requests(args, tokenizer) if dist.get_rank() == 0 else None
+        requests = build_requests(args, tokenizer, sampling_params) if dist.get_rank() == 0 else None
 
-        
         context = get_inference_context(None, 
                                         None,
                                         calculate_max_sequence_length_from_requests=False)
@@ -132,7 +144,6 @@ async def main(engine: DynamicInferenceEngine, requests: List[Request], sampling
         engine = DynamicInferenceEngine(
             controller,
             context,
-            termination_id=tokenizer.eod,
             enable_cuda_graph=args.cuda_graph_impl == "local",
             random_seed=args.seed,
             enable_chunked_prefill=not args.disable_chunked_prefill
@@ -147,6 +158,5 @@ async def main(engine: DynamicInferenceEngine, requests: List[Request], sampling
         
         asyncio.run(main(engine, 
                         requests,
-                        sampling_params,
                         args.inference_coordinator_port))
 
diff --git a/examples/inference/gpt/utils.py b/examples/inference/gpt/utils.py
index 2c314f52d34..e7745537991 100644
--- a/examples/inference/gpt/utils.py
+++ b/examples/inference/gpt/utils.py
@@ -1,4 +1,4 @@
-# Copyright (c) 2025, NVIDIA CORPORATION. All rights reserved.
+# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 
 import json
 import itertools
@@ -13,6 +13,8 @@
 from megatron.core.inference.contexts import DynamicInferenceContext
 from megatron.core.transformer.module import MegatronModule
 
+from megatron.core.inference.sampling_params import SamplingParams
+
 
 
 def add_common_inference_args(parser: ArgumentParser) -> ArgumentParser:
@@ -130,6 +132,16 @@ def add_common_inference_args(parser: ArgumentParser) -> ArgumentParser:
     return parser
 
 
+def get_default_sampling_params(termination_id: int = None):
+    return SamplingParams(
+        temperature=1.0,
+        top_k=1,
+        top_p=0.0,
+        return_log_probs=False,
+        num_tokens_to_generate=30,
+        termination_id = termination_id,
+    )
+
 def get_curr_time() -> float:
     """Get synchronized time across ranks."""
     curr_time = torch.cuda.LongTensor([time.time_ns()])
@@ -153,7 +165,7 @@ class Request:
         tokenizer (Any): Tokenizer for tokenizing the prompt.
     """
 
-    def __init__(self, prompt_text: str, time_offset: float, tokenizer: Any):
+    def __init__(self, prompt_text: str, time_offset: float, tokenizer: Any, sampling_params: SamplingParams = None):
         self.prompt_text = prompt_text
         self.prompt_tokens = tokenizer.tokenize(prompt_text)
         self.output_text = None
@@ -163,6 +175,7 @@ def __init__(self, prompt_text: str, time_offset: float, tokenizer: Any):
         self.time_start = None
         self.time_end = None
         self.state = "not-started"
+        self.sampling_params: SamplingParams = sampling_params if sampling_params is not None else get_default_sampling_params(tokenizer.eod)
 
     def __str__(self) -> str:
         return "state '%s'; toffset %.1e; prompt len %d; output len %d; '%s'" % (
@@ -216,10 +229,12 @@ def arrival(r):
     return time_offsets
 
 
-def get_cli_requests(args: Namespace, tokenizer: Any) -> list[Request]:
+def get_cli_requests(
+        args: Namespace, tokenizer: Any, sampling_params: Optional[SamplingParams] = None
+) -> list[Request]:
 
     # Get time offsets.
-    time_offsets = get_time_offsets(
+    t_offsets = get_time_offsets(
         args.seed,
         args.incoming_requests_per_step,
         args.incoming_requests_per_sec,
@@ -227,11 +242,13 @@ def get_cli_requests(args: Namespace, tokenizer: Any) -> list[Request]:
     )
 
     # Init requests.
-    requests = [Request(p, t, tokenizer) for p,t in zip(args.prompts, time_offsets)]
+    requests = [Request(p, t, tokenizer, sampling_params) for p,t in zip(args.prompts, t_offsets)]
     return requests
 
 
-def get_synthetic_requests(args: Namespace, tokenizer: Any) -> list[Request]:
+def get_synthetic_requests(
+    args: Namespace, tokenizer: Any, sampling_params: Optional[SamplingParams] = None
+) -> list[Request]:
     """Get example requests."""
 
     # Get time offsets.
@@ -244,14 +261,16 @@ def get_synthetic_requests(args: Namespace, tokenizer: Any) -> list[Request]:
 
     # Init requests.
     requests = [
-        Request("hi " * random.randint(*args.num_tokens_to_prompt), t, tokenizer)
+        Request("hi " * random.randint(*args.num_tokens_to_prompt), t, tokenizer, sampling_params)
         for t in time_offsets
     ]
 
     return requests
 
 
-def get_requests_from_file(args: Namespace, tokenizer: Any) -> list[Request]:
+def get_requests_from_file(
+    args: Namespace, tokenizer: Any, sampling_params: Optional[SamplingParams] = None
+) -> list[Request]:
     """Get requests from a file."""
     if not args.prompt_file:
         raise ValueError("Prompt file is required to read requests from a file.")
@@ -275,23 +294,25 @@ def get_requests_from_file(args: Namespace, tokenizer: Any) -> list[Request]:
 
     # Init requests.
     requests = [
-        Request(p, t, tokenizer)
+        Request(p, t, tokenizer, sampling_params)
         for p, t in tqdm(zip(prompts, time_offsets), "init requests", total=len(prompts))
     ]
 
     return requests
 
 
-def build_requests(args: Namespace, tokenizer: Any) -> list[Request]:
+def build_requests(
+    args: Namespace, tokenizer: Any, sampling_params: Optional[SamplingParams] = None
+) -> list[Request]:
     # Check if we have any prompts (from command line or JSONL)
     if args.prompts:
         if args.prompt_file:
             raise ValueError("Cannot use both --prompts and --prompt-file")
-        return get_cli_requests(args, tokenizer)
+        return get_cli_requests(args, tokenizer, sampling_params)
     elif args.prompt_file:
-        return get_requests_from_file(args, tokenizer)
+        return get_requests_from_file(args, tokenizer, sampling_params)
     else:
-        return get_synthetic_requests(args, tokenizer)
+        return get_synthetic_requests(args, tokenizer, sampling_params)
 
 
 def get_model_size_str(model):
diff --git a/megatron/core/inference/engines/dynamic_engine.py b/megatron/core/inference/engines/dynamic_engine.py
index b536e66d3da..23a122f1c5f 100644
--- a/megatron/core/inference/engines/dynamic_engine.py
+++ b/megatron/core/inference/engines/dynamic_engine.py
@@ -84,21 +84,23 @@ class DynamicInferenceEngine(AbstractEngine):
             outputs and detokenizer the output tokens.
         inference_context (DynamicInferenceContext): Context for managing in-flight
             batching and a dynamic block-level KV cache (similar to paged attention).
-        termination_id (int): Token ID to mark end-of-sequence.
         random_seed (Optional[int]): Use a random seed if you want deterministic
             results. Defaults to None.
+        static_sampling (bool): If True, all requests are assumed to have the same
+            sampling parameters. This avoids needing to loop through all requests and
+            their sampling parameters every generation step, improving latency.
     """
 
     def __init__(
         self,
         controller: TextGenerationController,
         context: DynamicInferenceContext,
-        termination_id: int,
         enable_cuda_graph: Optional[bool] = None,
         random_seed: Optional[int] = None,
         *,
         track_paused_request_events: bool = False,
         enable_chunked_prefill: bool = True,
+        static_sampling: bool = False,
     ):
 
         if enable_cuda_graph is not None:
@@ -114,15 +116,11 @@ def __init__(
         assert isinstance(
             context, DynamicInferenceContext
         ), f"context must be a DynamicInferenceContext, got {type(context)}"
-        assert isinstance(
-            termination_id, int
-        ), f"termination_id must be an int, got {type(termination_id)}"
         assert isinstance(random_seed, int), f"random_seed must be an int, got {type(random_seed)}"
 
         self.request_counter = Counter()
         self.controller = controller
         self.context = context
-        self.termination_id = termination_id
         self.random_seed = random_seed
         self.track_paused_request_events = track_paused_request_events
         self.step_count = 0
@@ -137,6 +135,7 @@ def __init__(
         self.paused = False
         self.stopped = False
         self.enable_chunked_prefill = enable_chunked_prefill
+        self.static_sampling = static_sampling
 
         # Initialize the asyncio loop if it has not already been initialized.
         # TODO: Start the engine loop here.
@@ -248,10 +247,7 @@ def __init__(
             self.capture_stats = capture_stats
 
     async def start_listening_to_data_parallel_coordinator(
-        self,
-        sampling_params: SamplingParams,
-        inference_coordinator_port: int,
-        launch_inference_coordinator: bool = True,
+        self, inference_coordinator_port: int, launch_inference_coordinator: bool = True
     ):
         """Initializes ZMQ communication to connect the engine with an inference coordinator.
 
@@ -276,8 +272,6 @@ async def start_listening_to_data_parallel_coordinator(
         (`self.run_engine`) as a background asyncio task.
 
         Args:
-            sampling_params (SamplingParams): The default sampling parameters to be
-                used for inference, passed to the engine's main loop.
             inference_coordinator_port (int): The network port where the central
                 `InferenceCoordinator` is or will be listening.
             launch_inference_coordinator (bool, optional): If True, the global rank 0
@@ -371,9 +365,7 @@ async def start_listening_to_data_parallel_coordinator(
             logging.info("Inference co-ordinator is ready to receive requests!")
 
         # Finally run the engine infinite loop
-        self.engine_loop_task = asyncio.create_task(
-            self.run_engine_with_coordinator(sampling_params)
-        )
+        self.engine_loop_task = asyncio.create_task(self.run_engine_with_coordinator())
 
     async def _notify_cond_for_new_request(self):
         """Helper function to notify condition variable when a new request is added."""
@@ -435,24 +427,19 @@ def add_request(
         self,
         request_id: int,
         prompt: Union[str, List[int], Tensor],
-        num_tokens_to_generate: Optional[int] = None,
-        num_tokens_total: Optional[int] = None,
+        sampling_params: Optional[SamplingParams] = None,
     ) -> asyncio.Future[DynamicInferenceRequest]:
         """Add request to inference context.
 
         Args:
             request_id (int): Unique ID of request.
             prompt (Union[str, Tensor]): Prompt as either a text string or token IDs.
-            num_tokens_to_generate (Optional[int]): Number of output tokens to generate.
-            num_tokens_total (Optional[int]): Limit on total number of tokens (prompt + generated).
+            sampling_params (Optional[SamplingParams]): Sampling parameters for the request.
 
         Return:
             Returns an asyncio `Future[DynamicInferenceRequest]` for the user to wait on.
         """
 
-        sampling_params = SamplingParams(
-            num_tokens_to_generate=num_tokens_to_generate, num_tokens_total=num_tokens_total
-        )
         prompt_str = None
         # Tokenize prompt if text.
         if isinstance(prompt, str):
@@ -616,6 +603,37 @@ def schedule_non_chunked_prefill(self):
             else:
                 break
 
+    def get_active_sampling_map(self) -> List[Tuple[SamplingParams, List[int]]]:
+        """Gets a map of sampling methods to active requests indices in the context."""
+        # Get all active request IDs.
+        active_request_ids = self.context.request_ids[
+            self.context.paused_request_count : self.context.total_request_count
+        ].tolist()
+        if self.static_sampling:
+            return [(next(iter(self.requests.values())).sampling_params, active_request_ids)]
+
+        # Get a map from request_id to context array index.
+        context_id_map = {r: i for i, r in enumerate(active_request_ids)}
+
+        # Create map of sampling methods to context array indices.
+        sampling_map: List[Tuple[SamplingParams, List[int]]] = []
+        for request_id, request in self.requests.items():
+            if request_id not in context_id_map:
+                continue
+            context_id = context_id_map[request_id]
+            sp = request.sampling_params
+
+            # Look for a pre-existing group with these sampling parameters.
+            for sampling, indices in sampling_map:
+                if sampling == sp:
+                    indices.append(context_id)
+                    break
+            # If no group exists, create a new one.
+            else:
+                sampling_map.append((sp, [context_id]))
+
+        return sampling_map
+
     def schedule_chunked_prefill(self):
         """
         This function schedules chunked prefill requests.
@@ -678,7 +696,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
+        self, *, verbose: Optional[bool] = False
     ) -> Tuple[List[DynamicInferenceRequest], List[DynamicInferenceRequest], float]:
         """
         Wrapper for controller.generate_output_tokens_dynamic_batch(), to
@@ -711,9 +729,8 @@ async def async_step(
         # save the is_decode_only AFTER scheduling, BEFORE update
         self.is_decode_only = is_decode_only
         self.step_start_event.record()
-        result = await self.controller.async_generate_output_tokens_dynamic_batch(
-            sampling_params, self.termination_id
-        )
+        sampling_map = self.get_active_sampling_map()
+        result = await self.controller.async_generate_output_tokens_dynamic_batch(sampling_map)
         self.step_end_event.record()
         self.step_end_event.synchronize()
         step_time = self.step_start_event.elapsed_time(self.step_end_event) / 1e3
@@ -804,12 +821,10 @@ async def async_step(
         }
 
     def step_modern(
-        self, sampling_params: SamplingParams, *, verbose: Optional[bool] = False
+        self, *, verbose: Optional[bool] = False
     ) -> Tuple[List[DynamicInferenceRequest], List[DynamicInferenceRequest], float]:
         """Synchronous wrapper for `self.async_step`."""
-        return self._loop.run_until_complete(
-            self.async_step(sampling_params=sampling_params, verbose=verbose)
-        )
+        return self._loop.run_until_complete(self.async_step(verbose=verbose))
 
     def step_legacy(
         self, sampling_params: SamplingParams, *, verbose: Optional[bool] = False
@@ -836,16 +851,11 @@ def generate(
 
         for prompt in prompts:
             request_id = int(next(self.request_counter))
-            _ = self.add_request(
-                request_id,
-                prompt,
-                sampling_params.num_tokens_to_generate,
-                sampling_params.num_tokens_total,
-            )
+            _ = self.add_request(request_id, prompt, sampling_params)
 
         finished_requests_list = []
         while self.has_unfinished_requests():
-            result = self.step_modern(sampling_params)
+            result = self.step_modern()
             finished_requests_list.extend(result["finished_requests"])
 
         # Ensure requests are returned in the same order they were passed in.
@@ -922,12 +932,7 @@ def schedule_requests(self) -> int:
             if header == Headers.SUBMIT_REQUEST:
                 request_id, prompt, sampling_params = data[1:]
                 sampling_params = SamplingParams.deserialize(sampling_params)
-                self.add_request(
-                    request_id,
-                    prompt,
-                    sampling_params.num_tokens_to_generate,
-                    sampling_params.num_tokens_total,
-                )
+                self.add_request(request_id, prompt, sampling_params)
             elif header == Headers.PAUSE:
                 self.paused = True
             elif header == Headers.STOP:
@@ -953,7 +958,7 @@ def stop(self):
         self.zmq_context.term()
         parallel_state.destroy_model_parallel()
 
-    async def run_engine(self, sampling_params: SamplingParams, *, verbose: Optional[bool] = False):
+    async def run_engine(self, *, verbose: Optional[bool] = False):
         """Continually steps the engine asynchronously."""
         try:
             while True:
@@ -964,13 +969,11 @@ async def run_engine(self, sampling_params: SamplingParams, *, verbose: Optional
                         or self.waiting_request_ids
                     )
 
-                await self.async_step(sampling_params=sampling_params, verbose=verbose)
+                await self.async_step(verbose=verbose)
         except asyncio.CancelledError:
             pass
 
-    async def run_engine_with_coordinator(
-        self, sampling_params: SamplingParams, *, verbose: Optional[bool] = False
-    ):
+    async def run_engine_with_coordinator(self, *, verbose: Optional[bool] = False):
         """Continually steps the engine asynchronously."""
         try:
             while True:
@@ -1002,9 +1005,7 @@ async def run_engine_with_coordinator(
                     await asyncio.sleep(0.02)
                     continue
 
-                engine_output = await self.async_step(
-                    sampling_params=sampling_params, verbose=verbose
-                )
+                engine_output = await self.async_step(verbose=verbose)
 
                 is_tp0_and_pp0 = (
                     parallel_state.get_tensor_model_parallel_rank() == 0
diff --git a/megatron/core/inference/engines/static_engine.py b/megatron/core/inference/engines/static_engine.py
index 6464aa315fd..d084528b8f2 100644
--- a/megatron/core/inference/engines/static_engine.py
+++ b/megatron/core/inference/engines/static_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 warnings
@@ -101,7 +101,6 @@ def __init__(
                     buffer_size_gb=buffer_size_gb,
                     num_cuda_graphs=1,
                 )
-                termination_id = self.controller.tokenizer.eod
                 self.controller.inference_wrapped_model.inference_context = dynamic_context
                 self.controller.inference_wrapped_model.prep_model_for_inference()
 
@@ -109,8 +108,8 @@ def __init__(
                     controller=self.controller,
                     random_seed=self.random_seed,
                     context=dynamic_context,
-                    termination_id=termination_id,
                     enable_cuda_graph=True,
+                    static_sampling=True,
                 )
         except Exception as e:
             # Get exception details for better debugging
diff --git a/megatron/core/inference/inference_request.py b/megatron/core/inference/inference_request.py
index 9fb1f33edd7..21ff7786d6a 100644
--- a/megatron/core/inference/inference_request.py
+++ b/megatron/core/inference/inference_request.py
@@ -229,6 +229,7 @@ class DynamicInferenceRequest(InferenceRequest):
     finished_chunk_token_count = 0
 
     def __post_init__(self):
+        self.sampling_params = copy.deepcopy(self.sampling_params)
         if self.prompt_tokens is not None:
             self.remaining_prompt_tokens = copy.deepcopy(self.prompt_tokens)
 
diff --git a/megatron/core/inference/sampling_params.py b/megatron/core/inference/sampling_params.py
index 443c23dc23e..a64e2e56775 100644
--- a/megatron/core/inference/sampling_params.py
+++ b/megatron/core/inference/sampling_params.py
@@ -1,4 +1,5 @@
-# Copyright (c) 2024, NVIDIA CORPORATION. All rights reserved.
+# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+
 from dataclasses import dataclass
 from typing import Optional
 
@@ -23,6 +24,7 @@ class SamplingParams:
     return_segments: bool = False  # Whether to return individually detokenized tokens
     num_tokens_to_generate: int = 30
     num_tokens_total: Optional[int] = None  # Cannot set both this and num_tokens_to_generate
+    termination_id: Optional[int] = None
     top_n_logprobs: int = 0
     return_prompt_top_n_logprobs: bool = False
     add_BOS: bool = False
diff --git a/megatron/core/inference/text_generation_controllers/text_generation_controller.py b/megatron/core/inference/text_generation_controllers/text_generation_controller.py
index e82397b854f..e877e785ff7 100644
--- a/megatron/core/inference/text_generation_controllers/text_generation_controller.py
+++ b/megatron/core/inference/text_generation_controllers/text_generation_controller.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 concurrent
@@ -319,6 +319,60 @@ def modify_logits_for_top_p_filtering(logits, top_p):
 
         return sampled_logits
 
+    def sample_from_dynamic_logits(
+        self,
+        last_token_logits: torch.Tensor,
+        active_sampling_map: List[Tuple[SamplingParams, List[int]]],
+        vocab_size: Optional[int] = None,
+        generation_started: Optional[torch.Tensor] = None,
+        top_n_logprobs_dict: Dict[int, List[Dict[str, float]]] = None,
+        logits: Optional[torch.Tensor] = None,
+        **kwargs,
+    ) -> torch.Tensor:
+        """Samples the logits to generate outputs
+
+        Given the logits of the last token, this function samples it
+        according to the parameters defined in active_sampling_map
+        and returns the samples. If sampling parameters top_n_logprobs > 0
+        at each step it also updates the top_n_logprobs dict.
+
+        Args:
+            last_token_logits (torch.Tensor): The last token logits. A tensor of
+                size [batch_size, vocab_size]
+            active_sampling_map (List[Tuple[SamplingParams, List[int]]]): A list of tuples
+                matching each unique set of sampling params to the context array indices
+                of the corresponding active requests.
+            vocab_size (int): Obtained from the tokenizer. Defaults to None
+            generation_started (torch.Tensor): A boolean tensor of shape [batch_size]. True
+                            indicates the prompt at that index has started generating tokens.
+            top_n_logprobs_dict (top_n_logprobs_dict): The dict to be updated
+
+        Returns:
+            sampled_logits (torch.Tensor): 1D tensor with [batch_size] elements
+            termination_id (torch.Tensor): Tensor of shape [batch_size] with termination ids
+            top_n_logprobs_this_step (torch.return_types.topk): a topk tensor with values as logits
+                and indices as the top k elements. None if sampling params top_n_logprobs is 0.
+        """
+        batch_size = last_token_logits.size(0)
+        new_sample = torch.zeros(batch_size, dtype=torch.int64, device=last_token_logits.device)
+        termination_id = torch.zeros_like(new_sample, dtype=torch.int64)
+
+        for sampling_params, mask in active_sampling_map:
+            # Filter out indices that are out of bounds for the current batch
+            valid_mask = [i for i in mask if i < batch_size]
+            if valid_mask:
+                new_sample[valid_mask] = self.sample_from_logits(
+                    last_token_logits[valid_mask],
+                    sampling_params=sampling_params,
+                    vocab_size=vocab_size,
+                )
+                if sampling_params.termination_id is not None:
+                    termination_id[valid_mask] = sampling_params.termination_id
+                else:
+                    termination_id[valid_mask] = self.tokenizer.eod
+
+        return new_sample, termination_id
+
     def update_generation_status(
         self,
         updated_prompts_tokens: torch.Tensor,
@@ -414,12 +468,14 @@ def unpad_input_prompt_tokens(
 
     @torch.inference_mode()
     async def async_generate_output_tokens_dynamic_batch(
-        self, sampling_params: SamplingParams, termination_id: int
+        self, active_sampling_map: List[Tuple[SamplingParams, List[int]]]
     ) -> Optional[Tuple[Tensor, Tensor, Tensor, Tensor]]:
         """Forward step the model and update the inference context.
 
         Args:
-            sampling_params (SamplingParams): Parameters for sampling logits.
+            active_sampling_map (List[Tuple[SamplingParams, List[int]]]): A list of tuples
+                matching each unique set of sampling params to the context array indices
+                of the corresponding active requests.
 
         Return:
             (Optional[Tuple[Tensor, Tensor, Tensor, Tensor]]) Current request IDs,
@@ -431,14 +487,17 @@ async def async_generate_output_tokens_dynamic_batch(
         unwrapped_model = unwrap_model(self.inference_wrapped_model.model)
 
         materialize_only_last_token_logits = context.materialize_only_last_token_logits
-        if sampling_params.return_log_probs:
-            skip_prompt_log_probs_for_dynamic_inference = getattr(
-                sampling_params, "skip_prompt_log_probs_for_dynamic_inference", False
-            )
-            assert (
-                skip_prompt_log_probs_for_dynamic_inference
-                or materialize_only_last_token_logits is False
-            ), "Materialize only last token logits must be false for returning log probs"
+        return_log_probs = False
+        for sampling_params, mask in active_sampling_map:
+            if sampling_params.return_log_probs:
+                skip_prompt_log_probs_for_dynamic_inference = getattr(
+                    sampling_params, "skip_prompt_log_probs_for_dynamic_inference", False
+                )
+                assert (
+                    skip_prompt_log_probs_for_dynamic_inference
+                    or materialize_only_last_token_logits is False
+                ), "Materialize only last token logits must be false for returning log probs"
+                return_log_probs = True
 
         # No tokens?
         if context.active_token_count == 0:
@@ -525,8 +584,8 @@ async def async_generate_output_tokens_dynamic_batch(
         # Use padded vocab size because tokenizer vocab size might not include padding
         # to nearest power of 2.
         vocab_size = inference_wrapper_config.padded_vocab_size
-        new_sample = self.sample_from_logits(
-            last_token_logits, sampling_params, vocab_size=vocab_size
+        new_sample, termination_id = self.sample_from_dynamic_logits(
+            last_token_logits, active_sampling_map, vocab_size=vocab_size
         )
 
         # Active sequence lengths.
@@ -538,6 +597,7 @@ async def async_generate_output_tokens_dynamic_batch(
         max_sequence_lengths = context.get_max_sequence_lengths()
 
         # Request finished if termination_id or length >= max_sequence_length.
+        # Note: termination_id tensor has per-request termination IDs from mixed sampling
         active_request_mask = (new_sample != termination_id).byte() & torch.less(
             active_sequence_lengths, max_sequence_lengths
         ).byte()
@@ -550,7 +610,7 @@ async def async_generate_output_tokens_dynamic_batch(
         new_sample_copy = new_sample.clone()
 
         log_probs = None
-        if sampling_params.return_log_probs:
+        if return_log_probs:
             log_probs = context.calculate_log_probs(
                 logits, new_sample_copy, only_last_token_logits=materialize_only_last_token_logits
             )
@@ -569,12 +629,12 @@ async def async_generate_output_tokens_dynamic_batch(
 
     @torch.inference_mode()
     def generate_output_tokens_dynamic_batch(
-        self, sampling_params: SamplingParams, termination_id: int
+        self, active_sampling_map: List[Tuple[SamplingParams, List[int]]]
     ) -> Optional[Tuple[Tensor, Tensor, Tensor]]:
         """Synchronous wrapper for `self.async_generate_output_tokens_dynamic_batch."""
         loop = get_asyncio_loop()
         return loop.run_until_complete(
-            self.async_generate_output_tokens_dynamic_batch(sampling_params, termination_id)
+            self.async_generate_output_tokens_dynamic_batch(active_sampling_map)
         )
 
     def _update_top_n_logprobs_dict(
diff --git a/tests/unit_tests/inference/engines/test_dynamic_engine.py b/tests/unit_tests/inference/engines/test_dynamic_engine.py
index 2d87b3c6adb..2227367469b 100644
--- a/tests/unit_tests/inference/engines/test_dynamic_engine.py
+++ b/tests/unit_tests/inference/engines/test_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 random
@@ -154,6 +154,9 @@ def _build_requests(cls, test_config: DynamicEngineTestConfig) -> List[DynamicIn
             # Sampling params.
             sampling_params = SamplingParams(
                 num_tokens_to_generate=num_tokens_to_generate,
+                termination_id=(
+                    -1 if test_config.use_fixed_output_lengths else test_config.vocab_size - 1
+                ),
                 return_log_probs=test_config.return_log_probs,
             )
             if not hasattr(sampling_params, "num_tokens_total"):
@@ -161,10 +164,10 @@ def _build_requests(cls, test_config: DynamicEngineTestConfig) -> List[DynamicIn
                 sampling_params.add_attributes({"num_tokens_total": num_tokens_total})
             else:
                 sampling_params.num_tokens_total = num_tokens_total
+
+            config_entry = test_config.skip_prompt_log_probs_for_dynamic_inference
             sampling_params.add_attributes(
-                {
-                    "skip_prompt_log_probs_for_dynamic_inference": test_config.skip_prompt_log_probs_for_dynamic_inference
-                }
+                {"skip_prompt_log_probs_for_dynamic_inference": config_entry}
             )
 
             # Request.
@@ -316,9 +319,6 @@ def _build_test_env(cls, test_config):
         engine = DynamicInferenceEngine(
             text_generation_controller,
             inference_context,
-            termination_id=(
-                -1 if test_config.use_fixed_output_lengths else test_config.vocab_size - 1
-            ),
             random_seed=test_config.random_seed,
             enable_cuda_graph=transformer_config.cuda_graph_impl == "local",
         )
@@ -342,7 +342,7 @@ def _run_step(cls, env):
         # the only thing that differs between requests is num_tokens_to_generate,
         # and engine.async_step() doesn't use this sampling param's
         # num_tokens_to_generate.
-        result = env.engine.step_modern(env.requests[0].sampling_params, verbose=False)
+        result = env.engine.step_modern(verbose=False)
         finished_requests = result["finished_requests"]
 
     @classmethod
@@ -724,11 +724,7 @@ async def test_run_engine(self):
         test_config = DynamicEngineTestConfig(use_fixed_output_lengths=True)
         env = self._build_test_env(test_config)
 
-        # It's safe to use request 0's sampling params here because all sampling
-        # params are identical as long as use_fixed_output_lengths == False.
-        engine_task = asyncio.create_task(
-            env.engine.run_engine(sampling_params=env.requests[0].sampling_params, verbose=False)
-        )
+        engine_task = asyncio.create_task(env.engine.run_engine(verbose=False))
 
         request_completion_futures: Dict[int, asyncio.Future[DynamicInferenceRequest]] = {}
 
diff --git a/tests/unit_tests/inference/text_generation_controllers/test_simple_text_generation_controller.py b/tests/unit_tests/inference/text_generation_controllers/test_simple_text_generation_controller.py
index c7f9d3214f5..f23a9782646 100644
--- a/tests/unit_tests/inference/text_generation_controllers/test_simple_text_generation_controller.py
+++ b/tests/unit_tests/inference/text_generation_controllers/test_simple_text_generation_controller.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+
 import copy
 import os
 import random
@@ -49,6 +51,7 @@ def setup_model(
         fp8: bool = False,
         tensor_model_parallel_size: int = 2,
         pipeline_model_parallel_size: int = 1,
+        batch_size: int = 4,
         static: bool = True,
         use_training_random_init: bool = False,
     ):
@@ -62,7 +65,7 @@ def setup_model(
             _set_random_seed(123, inference_rng_tracker=True)
         else:
             model_parallel_cuda_manual_seed(123, inference_rng_tracker=True)
-        self.batch_size = 4
+        self.batch_size = batch_size
         self.hidden_size = 12
         self.vocab_size = 100
         self.sequence_length = 60 if fp8 else 64  # Test padding for fp8
@@ -225,6 +228,45 @@ def detokenize(self, inp, skip_special_tokens=False):
             sampled_logits >= expected_min_value
         ), f"The sampled logits should all be greater than {expected_min_value} but its {sampled_logits}"
 
+    def test_sample_from_dynamic_logits(self):
+        batch_size = 12
+        self.setup_model(torch.float32, batch_size=batch_size, static=False)
+        self.mock_tokenizer.eod = self.vocab_size
+
+        active_sampling_map: List[Tuple[SamplingParams, List[int]]] = [
+            (SamplingParams(top_k=3), [0, 3, 2]),
+            (SamplingParams(top_p=0.8), [4, 1, 7]),
+            (SamplingParams(top_k=5), [11, 5, 8]),
+            # (SamplingParams(top_k=5, top_p=0.7), [11, 5, 8]), # uncomment for FlashInfer sampling
+            (SamplingParams(temperature=2.0), [9, 6, 10]),
+        ]
+        rev_sampling_map: List[SamplingParams] = [None] * batch_size
+        for sampling_params, indices in active_sampling_map:
+            for idx in indices:
+                rev_sampling_map[idx] = sampling_params
+
+        last_token_logits = torch.arange(0, self.vocab_size).repeat(batch_size, 1).float().cuda()
+        sampled_logits, _ = self.text_generation_controller.sample_from_dynamic_logits(
+            last_token_logits, active_sampling_map, vocab_size=self.vocab_size
+        )
+        top_k_values = torch.Tensor([s.top_k for s in rev_sampling_map]).cuda().unsqueeze(1)
+        top_k_values[top_k_values == 0] = self.vocab_size
+        top_p_values = torch.Tensor([s.top_p for s in rev_sampling_map]).cuda().unsqueeze(1)
+        temp_values = torch.Tensor([s.temperature for s in rev_sampling_map]).cuda().unsqueeze(1)
+        vocab_indices = torch.arange(self.vocab_size).cuda()
+
+        assert torch.all(
+            sampled_logits >= self.vocab_size - top_k_values
+        ), f"The sampled logits should all be greater than {self.vocab_size - top_k_values} but its {sampled_logits}"
+        l = last_token_logits[0]
+        sampled_l = l.div(temp_values).softmax(dim=-1)
+        top_k_mask = vocab_indices.unsqueeze(0) < (self.vocab_size - top_k_values)
+        sampled_l.masked_fill_(top_k_mask, 0.0)
+        expected_min_values = sampled_l[sampled_l.cumsum(dim=-1) > top_p_values].amax(dim=-1)
+        assert torch.all(
+            sampled_logits >= expected_min_values
+        ), f"The sampled logits should all be greater than {expected_min_values} but its {sampled_logits}"
+
     @pytest.mark.parametrize("dtype", [torch.float32, torch.bfloat16])
     @pytest.mark.parametrize(
         "symmetric_ar_type",
@@ -731,10 +773,11 @@ def test_sampled_tokens_match_with_parallelism(self, static, tp_size, pp_size):
                         ),
                     )
                 )
-            sampling_params = SamplingParams(top_k=10, return_log_probs=True)
+            sampling_params = SamplingParams(top_k=10, return_log_probs=True, termination_id=-1)
+            sampling_map = [(sampling_params, list(range(len(active_requests))))]
             while context.has_unfinished_requests():
                 result = self.text_generation_controller.generate_output_tokens_dynamic_batch(
-                    sampling_params=sampling_params, termination_id=-1
+                    active_sampling_map=sampling_map
                 )
                 new_tokens = result["sample"]
                 assert len(new_tokens) == len(active_requests)
diff --git a/tests/unit_tests/transformer/test_cuda_graphs.py b/tests/unit_tests/transformer/test_cuda_graphs.py
index 685e3674374..b4764aa7e18 100644
--- a/tests/unit_tests/transformer/test_cuda_graphs.py
+++ b/tests/unit_tests/transformer/test_cuda_graphs.py
@@ -1,4 +1,4 @@
-# Copyright (c) 2025, NVIDIA CORPORATION. All rights reserved.
+# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 
 import os
 import random
@@ -572,7 +572,9 @@ def capture_cuda_graphs(self, cuda_graph_capture_freeze_gc: bool) -> None:
         )
 
         # Sampling params.
-        sampling_params = SamplingParams(num_tokens_to_generate=num_tokens_to_generate)
+        sampling_params = SamplingParams(
+            num_tokens_to_generate=num_tokens_to_generate, termination_id=vocab_size - 1
+        )
 
         # GPT model.
         model = GPTModel(
@@ -630,10 +632,7 @@ def capture_cuda_graphs(self, cuda_graph_capture_freeze_gc: bool) -> None:
 
         # Inference engine.
         engine = DynamicInferenceEngine(
-            text_generation_controller,
-            context,
-            termination_id=vocab_size - 1,
-            random_seed=random_seed,
+            text_generation_controller, context, random_seed=random_seed
         )
 
         return engine.capture_stats
diff --git a/tools/run_inference_performance_test.py b/tools/run_inference_performance_test.py
index c5318b9fbbf..2f2adabc0ab 100644
--- a/tools/run_inference_performance_test.py
+++ b/tools/run_inference_performance_test.py
@@ -1,3 +1,5 @@
+# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+
 import os
 from megatron.core.inference.model_inference_wrappers.inference_wrapper_config import (
     InferenceWrapperConfig,
@@ -233,7 +235,6 @@ def generate_dynamic(
     args: argparse.Namespace,
     inference_requests: List[InferenceRequest],
     inference_engine: DynamicInferenceEngine,
-    sampling_params: SamplingParams,
 ):
     global REQUEST_ID
     for request in inference_requests:
@@ -241,13 +242,13 @@ def generate_dynamic(
         REQUEST_ID += 1
         prompt_tokens = request.prompt_tokens
         inference_engine.add_request(
-            request_id, prompt_tokens, num_tokens_to_generate=args.num_tokens_to_generate
+            request_id, prompt_tokens, request.inference_parameters,
         )
 
     start_time = time.perf_counter()
     all_finished_requests = []
     while inference_engine.has_unfinished_requests():
-        result = inference_engine.step(sampling_params, verbose=False)
+        result = inference_engine.step(verbose=False)
         finished_requests = result["finished_requests"]
         for request in finished_requests:
             req_id = request.request_id
@@ -340,7 +341,7 @@ def main():
         if args.engine_type == "static":
             inference_engine.generate(prompts=["warmup"], sampling_params=warmup_sampling_params)
         elif args.engine_type == "dynamic":
-            generate_dynamic(args, requests, inference_engine, sampling_params)
+            generate_dynamic(args, requests, inference_engine)
 
     if args.benchmark_profile:
         torch.cuda.cudart().cudaProfilerStart()
@@ -361,7 +362,7 @@ def main():
             )
         elif args.engine_type == "dynamic":
             results: List[InferenceRequest] = generate_dynamic(
-                args, requests, inference_engine, sampling_params
+                args, requests, inference_engine,
             )
     end_time = time.perf_counter()
     latency = end_time - start_time