[TRTLLM-8686][chore] Standardize LlmArgs initialization and parsing around Pydantic

CODING_GUIDELINES.md

@@ -404,6 +404,41 @@ foo.SomeClass()
 1. For interfaces that may be used outside a file, prefer docstrings over comments.
 2. Comments should be reserved for code within a function, or interfaces that are local to a file.
 
+#### Pydantic Guidelines
+
+When defining any user-facing configuration classes (particularly `LlmArgs` or any class used in its fields), always use Pydantic classes rather than dataclasses or vanilla classes.
+
+**Model Structure:**
+- Inherit from `StrictBaseModel` (which sets `extra="forbid"`) to catch typos in field names
+- Use [discriminated unions](https://docs.pydantic.dev/latest/concepts/unions/#discriminated-unions) when a field needs to accept one of several possible config classes (e.g. `speculative_config` accepts any of `EagleDecodingConfig`, `MedusaDecodingConfig`, etc.)
+
+**Field Definitions:**
+- Add descriptions to all user-facing fields via `Field(description="...")`. Avoid using comments for descriptions.
+- Avoid `dict`, `object`, `Any` as field types - use properly typed alternatives
+- Avoid defining mutable defaults directly; use `default_factory` instead.
+  - Good: `Field(default_factory=list)`, `Field(default_factory=dict)`, `Field(default_factory=MyClass)`
+  - Bad: `Field(default=[])`, `Field(default={})`, `Field(default=MyClass())` directly
+- Use `Literal["value1", "value2"]` instead of `str` when a field should only accept certain values
+- Prefer `PositiveInt`, `NonNegativeInt`, `NonNegativeFloat`, `PositiveFloat`, `Field(gt=0)`, `Field(ge=0)`, etc. for numeric constraints instead of defining custom validators
+- Use `Field(min_length=1)` to enforce minimum length of a list
+
+**Validation:**
+- Use `@field_validator` and `@model_validator` instead of manual `validate()` or `is_valid()` methods
+- Raise `ValueError` instead of using assertions
+- Co-locate validation logic within the class itself rather than in a parent class, unless it depends on fields in the parent
+
+**Serialization:**
+- Avoid defining `to_dict()` methods - prefer Pydantic's built-in `model_dump()` to convert to a dictionary.
+  - Note: you can override `model_dump()` to customize its behavior, but avoid doing so unless absolutely necessary.
+  - Good: `MyModel.model_dump()`
+  - Bad: `MyModel.to_dict()`
+- Avoid defining `from_dict()` methods - prefer constructing the class directly from arguments.
+  - Good: `MyModel(**kwargs)`
+  - Bad: `MyModel.from_dict(kwargs)`
+
+TODO:
+- unit tests: e.g. for get_llm_args_from_cli_params, other tests you removed, etc
+
 #### Docstring Syntax
 ##### Classes and Functions
 Use the [Google style](https://google.github.io/styleguide/pyguide.html), which can be parsed by Sphinx.

examples/llm-eval/lm-eval-harness/lm_eval_tensorrt_llm.py

@@ -56,7 +56,7 @@ def __init__(
         free_gpu_memory_fraction: float = 0.9,
         trust_remote_code: bool = False,
         use_cuda_graph: bool = True,
-        backend: str = 'trt',
+        backend: str = 'tensorrt',
         max_context_length: Optional[int] = None,
         moe_expert_parallel_size: Optional[int] = None,
         moe_backend: Optional[str] = "TRTLLM",

tensorrt_llm/_torch/auto_deploy/llm.py

@@ -1,11 +1,12 @@
 import types
-from typing import Any, Dict, List, Optional, Tuple
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 import torch
 
 from ...executor.result import CompletionOutput
 from ...inputs.registry import DefaultInputProcessor, ExtraProcessedInputs
-from ...llmapi.llm import RequestOutput, _TorchLLM
+from ...llmapi.llm import BaseLLM, RequestOutput, _TorchLLM
 from ...llmapi.tokenizer import TokenizerBase, TransformersTokenizer, tokenizer_factory
 from ...sampling_params import SamplingParams
 from .distributed import common as dist_ad
@@ -112,9 +113,9 @@ def factory(self) -> ModelFactory:
             self._factory = self.args.create_factory()
         return self._factory
 
-    def __init__(self, *args, **kwargs):
-        kwargs["backend"] = "_autodeploy"
-        super().__init__(*args, **kwargs)
+    def __init__(self, model: Union[str, Path], **kwargs: Any) -> None:
+        args = LlmArgs(model=model, **kwargs)
+        BaseLLM.__init__(args)
 
     def _try_load_tokenizer(self) -> Optional[TokenizerBase]:
         if self.args.skip_tokenizer_init:
@@ -159,7 +160,7 @@ class DemoLLM(LLM):
     """
 
     def __init__(self, **kwargs):
-        self.args: LlmArgs = LlmArgs.from_kwargs(**kwargs)
+        self.args: LlmArgs = LlmArgs(**kwargs)
 
         self.mpi_session = None
         self.runtime_context = None

tensorrt_llm/_torch/auto_deploy/llm_args.py

@@ -1,3 +1,4 @@
+import functools
 from importlib.resources import files
 from pathlib import Path
 from typing import Any, Dict, List, Literal, Optional, Type, Union
@@ -7,6 +8,7 @@
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 from ...llmapi.llm_args import (
+    BaseLlmArgs,
     BuildConfig,
     EagleDecodingConfig,
     SamplerType,
@@ -28,16 +30,26 @@ def _get_config_dict() -> SettingsConfigDict:
     )
 
 
+@functools.cache
+def _get_dummy_llm_args(llm_args_cls: Type[BaseLlmArgs]) -> BaseLlmArgs:
+    """Create a cached dummy LlmArgs instance to retrieve post-validation default values."""
+    return llm_args_cls(model="dummy")
+
+
 def _check_for_default_value_only(
-    cls: Type[BaseSettings], value: Any, info: ValidationInfo, msg: str
+    llm_args_cls: Type[BaseSettings], value: Any, info: ValidationInfo, msg: str
 ) -> Any:
-    """Check if the value is the default value for the field.
+    """Raise ValueError if value differs from the field's default.
 
-    If the value is not the default value, raise a ValueError.
+    Compares against both the raw field default and the post-validation default
+    (from a dummy instance) to handle cases where validators modify defaults.
     """
+    dummy_llm_args = _get_dummy_llm_args(llm_args_cls)
     field_name = info.field_name
     assert field_name is not None, "field_name should be set for validated field."
-    if value != cls.model_fields[field_name].get_default(call_default_factory=True):
+    field_default = llm_args_cls.model_fields[field_name].get_default(call_default_factory=True)
+    post_validation_default = getattr(dummy_llm_args, field_name)
+    if value not in {field_default, post_validation_default}:
         raise ValueError(msg)
     return value
 

tensorrt_llm/_torch/speculative/save_hidden_state.py

@@ -47,7 +47,7 @@ def _process_request(self, request: LlmRequest, resource_manager) -> None:
                 "input_ids": input_ids,
                 "hidden_state": hidden_states,
             }
-            if len(self.spec_config.eagle3_layers_to_capture) > 1:
+            if self.spec_config.num_capture_layers > 1:
                 if self.spec_config._last_hidden_in_save:
                     out_dict[
                         "aux_hidden_states"] = resource_manager.hidden_states[:num_tokens, :].cpu(

tensorrt_llm/bench/benchmark/__init__.py

@@ -4,11 +4,8 @@
 
 from pydantic import AliasChoices, BaseModel, Field
 
-from tensorrt_llm import LLM as PyTorchLLM
-from tensorrt_llm._tensorrt_engine import LLM
 from tensorrt_llm.bench.benchmark.utils.processes import IterationWriter
 from tensorrt_llm.bench.build.build import get_model_config
-from tensorrt_llm.bench.dataclasses.configuration import RuntimeConfig
 from tensorrt_llm.bench.dataclasses.general import BenchmarkEnvironment
 from tensorrt_llm.logger import logger
 
@@ -21,7 +18,6 @@ class GeneralExecSettings(BaseModel):
     backend: str = Field(
         default="pytorch",
         description="The backend to use when running benchmarking")
-    beam_width: int = Field(default=1, description="Number of search beams")
     model_path: Optional[Path] = Field(default=None,
                                        description="Path to model checkpoint")
     concurrency: int = Field(
@@ -36,15 +32,8 @@ class GeneralExecSettings(BaseModel):
         default=-1, description="End-of-sequence token ID, -1 to disable EOS")
     iteration_log: Optional[Path] = Field(
         default=None, description="Path where iteration logging is written")
-    kv_cache_percent: float = Field(
-        default=0.90,
-        validation_alias=AliasChoices("kv_cache_percent",
-                                      "kv_cache_free_gpu_mem_fraction"),
-        description="Percentage of memory for KV Cache after model load")
     max_input_len: int = Field(default=4096,
                                description="Maximum input sequence length")
-    max_seq_len: Optional[int] = Field(default=None,
-                                       description="Maximum sequence length")
     modality: Optional[str] = Field(
         default=None, description="Modality of multimodal requests")
     model: Optional[str] = Field(default=None, description="Model name or path")
@@ -75,54 +64,6 @@ def checkpoint_path(self) -> Path:
         return self.model_path or self.model
 
 
-def ignore_trt_only_args(kwargs: dict, backend: str):
-    """Ignore TensorRT-only arguments for non-TensorRT backends.
-
-    Args:
-        kwargs: Dictionary of keyword arguments to be passed to the LLM constructor.
-        backend: The backend type (e.g., "pytorch", "_autodeploy").
-    """
-    trt_only_args = [
-        "batching_type",
-        "normalize_log_probs",
-        "extended_runtime_perf_knob_config",
-    ]
-    for arg in trt_only_args:
-        if kwargs.pop(arg, None):
-            logger.warning(f"Ignore {arg} for {backend} backend.")
-
-
-def get_llm(runtime_config: RuntimeConfig, kwargs: dict):
-    """Create and return an appropriate LLM instance based on the backend configuration.
-
-    Args:
-        runtime_config: Runtime configuration containing backend selection and settings.
-        kwargs: Additional keyword arguments to pass to the LLM constructor.
-
-    Returns:
-        An instance of the appropriate LLM class for the specified backend.
-    """
-    llm_cls = LLM
-
-    if runtime_config.backend != None:
-        ignore_trt_only_args(kwargs, runtime_config.backend)
-
-    if runtime_config.iteration_log is not None:
-        kwargs["enable_iter_perf_stats"] = True
-
-    if runtime_config.backend == 'pytorch':
-        llm_cls = PyTorchLLM
-
-    elif runtime_config.backend == "_autodeploy":
-        from tensorrt_llm._torch.auto_deploy import LLM as AutoDeployLLM
-
-        kwargs["world_size"] = kwargs.pop("tensor_parallel_size", None)
-        llm_cls = AutoDeployLLM
-
-    llm = llm_cls(**kwargs)
-    return llm
-
-
 def get_general_cli_options(
         params: Dict, bench_env: BenchmarkEnvironment) -> GeneralExecSettings:
     """Get general execution settings from command line parameters.

tensorrt_llm/bench/benchmark/low_latency.py

@@ -8,27 +8,23 @@
 import yaml
 from click_option_group import (MutuallyExclusiveOptionGroup, OptionGroup,
                                 optgroup)
-from huggingface_hub import snapshot_download
 
 from tensorrt_llm.bench.benchmark import (generate_json_report,
-                                          get_general_cli_options, get_llm)
+                                          get_general_cli_options)
 from tensorrt_llm.bench.benchmark.utils.asynchronous import async_benchmark
-from tensorrt_llm.bench.benchmark.utils.general import generate_warmup_dataset
-from tensorrt_llm.bench.dataclasses.configuration import RuntimeConfig
+from tensorrt_llm.bench.benchmark.utils.general import (
+    ALL_SUPPORTED_BACKENDS, generate_warmup_dataset,
+    get_exec_settings_for_backend, update_sampler_args_with_extra_options)
 from tensorrt_llm.bench.dataclasses.general import BenchmarkEnvironment
 from tensorrt_llm.bench.dataclasses.reporting import ReportUtility
-from tensorrt_llm.llmapi import CapacitySchedulerPolicy
-from tensorrt_llm.models.modeling_utils import SpeculativeDecodingMode
-
-# isort: off
-from tensorrt_llm.bench.benchmark.utils.general import (
-    get_settings_from_engine, get_settings,
-    update_sampler_args_with_extra_options, ALL_SUPPORTED_BACKENDS)
-# isort: on
 from tensorrt_llm.bench.utils.data import (create_dataset_from_stream,
                                            initialize_tokenizer,
                                            update_metadata_for_multimodal)
+from tensorrt_llm.llmapi import CapacitySchedulerPolicy
+from tensorrt_llm.llmapi.llm_args import CudaGraphConfig
+from tensorrt_llm.llmapi.llm_create import create_llm_from_llm_args
 from tensorrt_llm.logger import logger
+from tensorrt_llm.models.modeling_utils import SpeculativeDecodingMode
 from tensorrt_llm.sampling_params import SamplingParams
 
 
@@ -222,67 +218,25 @@ def latency_command(
         #       The accurate table for multimodal models will be logged after the benchmark is done.
         logger.info(metadata.get_summary_for_print())
 
-    # Engine configuration parsing for PyTorch backend
-    kwargs = {}
-    if options.backend and options.backend.lower(
-    ) in ALL_SUPPORTED_BACKENDS and options.backend.lower() != "tensorrt":
-        if bench_env.checkpoint_path is None:
-            snapshot_download(options.model, revision=bench_env.revision)
-
-        exec_settings = get_settings(params, metadata, bench_env.model,
-                                     bench_env.checkpoint_path)
-        kwargs_max_sql = options.max_seq_len or metadata.max_sequence_length
-        logger.info(f"Setting PyTorch max sequence length to {kwargs_max_sql}")
-        kwargs["max_seq_len"] = kwargs_max_sql
-    elif options.backend.lower() == "tensorrt":
-        assert options.max_seq_len is None, (
-            "max_seq_len is not a runtime parameter for C++ backend")
-        exec_settings, build_cfg = get_settings_from_engine(options.engine_dir)
-        engine_max_seq_len = build_cfg["max_seq_len"]
+    # Engine configuration parsing
+    exec_settings = get_exec_settings_for_backend(params, metadata, options,
+                                                  bench_env)
 
-        if metadata.max_sequence_length > engine_max_seq_len:
-            raise RuntimeError(
-                f"Engine supports a max sequence of {engine_max_seq_len}. Provided "
-                "dataset contains a maximum sequence of "
-                f"{metadata.max_sequence_length}. Please rebuild a new engine to"
-                "support this dataset.")
-    else:
-        raise click.BadParameter(
-            f"{options.backend} is not a known backend, check help for available options.",
-            param_hint="backend")
+    exec_settings.revision = bench_env.revision
 
-    exec_settings["model"] = options.model
-    exec_settings["revision"] = bench_env.revision
-    engine_tokens = exec_settings["settings_config"]["max_num_tokens"]
-
-    # Update configuration with runtime options
-    exec_settings["settings_config"][
-        "kv_cache_percent"] = options.kv_cache_percent
-    exec_settings["settings_config"]["max_batch_size"] = 1
-    exec_settings["settings_config"]["max_num_tokens"] = engine_tokens
-    exec_settings["settings_config"]["beam_width"] = options.beam_width
-    exec_settings["settings_config"]["chunking"] = False
-    exec_settings["settings_config"][
-        "scheduler_policy"] = CapacitySchedulerPolicy.GUARANTEED_NO_EVICT
-
-    # Performance options
-    exec_settings["performance_options"]["cuda_graphs"] = True
-    exec_settings["performance_options"]["multi_block_mode"] = True
-
-    exec_settings["extra_llm_api_options"] = params.get("extra_llm_api_options")
+    # Update llm_args with runtime options
+    llm_args = exec_settings.llm_args
+    llm_args.max_batch_size = 1
+    llm_args.cuda_graph_config = CudaGraphConfig(max_batch_size=1)
+    llm_args.enable_chunked_prefill = False
+    llm_args.scheduler_config.capacity_scheduler_policy = CapacitySchedulerPolicy.GUARANTEED_NO_EVICT
 
     # Decoding Options
     if medusa_choices is not None:
         with open(medusa_choices, "r") as medusa_yml:
-            exec_settings["decoding_config"]["medusa_choices"] = \
-                yaml.load(medusa_yml, Loader=yaml.SafeLoader)
-
-    # Construct the runtime configuration dataclass.
-    runtime_config = RuntimeConfig(**exec_settings)
-
-    llm = None
-    kwargs = kwargs | runtime_config.get_llm_args()
-    kwargs['backend'] = options.backend
+            medusa_choices_data = yaml.load(medusa_yml, Loader=yaml.SafeLoader)
+            if llm_args.decoding_config is not None:
+                llm_args.decoding_config.medusa_choices = medusa_choices_data
 
     # Set environment variables for setting runtime options.
     default_env_overrides = {
@@ -292,23 +246,22 @@ def latency_command(
         "TRTLLM_ENABLE_PDL": "1",
     }
     # Update defaults with existing overrides (user preference takes priority)
-    default_env_overrides.update(kwargs.get("env_overrides", {}))
-    kwargs["env_overrides"] = default_env_overrides
+    llm_args.env_overrides = default_env_overrides | llm_args.env_overrides
 
+    llm = None
     try:
         logger.info("Setting up latency benchmark.")
+        llm = create_llm_from_llm_args(llm_args)
 
-        llm = get_llm(runtime_config, kwargs)
-
-        ignore_eos = True if runtime_config.decoding_config.decoding_mode == SpeculativeDecodingMode.NONE else False
+        ignore_eos = True if llm_args.decoding_config.decoding_mode == SpeculativeDecodingMode.NONE else False
         eos_id = tokenizer.eos_token_id if not ignore_eos else -1
         pad_id = tokenizer.pad_token_id if not ignore_eos else -1
 
         sampler_args = {
             "end_id": eos_id,
             "pad_id": pad_id,
-            "n": options.beam_width,
-            "use_beam_search": options.beam_width > 1
+            "n": llm_args.max_beam_width,
+            "use_beam_search": llm_args.max_beam_width > 1
         }
 
         sampler_args = update_sampler_args_with_extra_options(
@@ -354,8 +307,8 @@ def latency_command(
             # For multimodal models, we need to update the metadata with the correct input lengths
             metadata = update_metadata_for_multimodal(metadata, statistics)
 
-        report_utility = ReportUtility(statistics, metadata, runtime_config,
-                                       logger, kwargs, True)
+        report_utility = ReportUtility(statistics, metadata, exec_settings,
+                                       logger, llm_args, True)
         # Generate reports for statistics, output tokens, and request info.
         generate_json_report(options.report_json,
                              report_utility.get_statistics_dict)