(
engine: Union[Path, Engine],
executor_config: Optional[tllm.ExecutorConfig] = None,
batched_logits_processor: Optional[BatchedLogitsProcessor] = None,
model_world_size: int = 1,
world_size: int = 0,
mpi_session: Optional[MpiSession] = None,
reuse_mpi_comm: bool = False,
return_logits: bool = False,
postproc_worker_config: Optional[PostprocWorkerConfig] = None,
is_llm_executor: Optional[bool] = None,
hf_model_dir: Optional[Path] = None,
tokenizer: Optional[TokenizerBase] = None,
llm_args: Optional[BaseLlmArgs] = None,
**args,
)
| 423 | |
| 424 | @staticmethod |
| 425 | def create( |
| 426 | engine: Union[Path, Engine], |
| 427 | executor_config: Optional[tllm.ExecutorConfig] = None, |
| 428 | batched_logits_processor: Optional[BatchedLogitsProcessor] = None, |
| 429 | model_world_size: int = 1, |
| 430 | world_size: int = 0, |
| 431 | mpi_session: Optional[MpiSession] = None, |
| 432 | reuse_mpi_comm: bool = False, |
| 433 | return_logits: bool = False, |
| 434 | postproc_worker_config: Optional[PostprocWorkerConfig] = None, |
| 435 | is_llm_executor: Optional[bool] = None, |
| 436 | hf_model_dir: Optional[Path] = None, |
| 437 | tokenizer: Optional[TokenizerBase] = None, |
| 438 | llm_args: Optional[BaseLlmArgs] = None, |
| 439 | **args, |
| 440 | ) -> Union["GenerationExecutorProxy", "GenerationExecutorWorker"]: |
| 441 | if world_size == 0: |
| 442 | world_size = mpi_world_size() |
| 443 | |
| 444 | if world_size > 1 and world_size < model_world_size: |
| 445 | raise RuntimeError( |
| 446 | "Cannot instantiate Generator for engine built " |
| 447 | f"for {model_world_size} ranks, while currently running " |
| 448 | f"on {world_size} ranks.") |
| 449 | |
| 450 | postproc_worker_config = postproc_worker_config or PostprocWorkerConfig( |
| 451 | ) |
| 452 | |
| 453 | if postproc_worker_config.enabled: |
| 454 | logger_debug( |
| 455 | f"Using {postproc_worker_config.num_postprocess_workers} postprocess parallel processes.\n", |
| 456 | "green") |
| 457 | |
| 458 | worker_kwargs = { |
| 459 | "engine": engine, |
| 460 | "executor_config": executor_config, |
| 461 | "batched_logits_processor": batched_logits_processor, |
| 462 | "hf_model_dir": hf_model_dir, |
| 463 | "tokenizer": tokenizer, |
| 464 | "llm_args": llm_args, |
| 465 | } |
| 466 | |
| 467 | orchestrator_type = None if not isinstance( |
| 468 | llm_args, TorchLlmArgs) else llm_args.orchestrator_type |
| 469 | if orchestrator_type == "ray": |
| 470 | if llm_args and hasattr(llm_args, 'ray_worker_extension_cls'): |
| 471 | worker_kwargs[ |
| 472 | "ray_worker_extension_cls"] = llm_args.ray_worker_extension_cls |
| 473 | return GenerationExecutor._create_ray_executor( |
| 474 | worker_kwargs, |
| 475 | model_world_size, |
| 476 | postproc_worker_config, |
| 477 | is_llm_executor=is_llm_executor, |
| 478 | tp_size=args.get("tp_size", 1)) |
| 479 | elif orchestrator_type is not None and orchestrator_type != "rpc": |
| 480 | raise ValueError( |
| 481 | f"Unsupported orchestrator_type: {orchestrator_type}") |
| 482 |
no test coverage detected