aphrodite_engine.py 59 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471
  1. import time
  2. from contextlib import contextmanager
  3. from typing import TYPE_CHECKING, Any, ClassVar, Dict, Iterable, List, Optional
  4. from typing import Sequence as GenericSequence
  5. from typing import Tuple, Type, TypeVar, Union
  6. from loguru import logger
  7. from transformers import PreTrainedTokenizer
  8. from typing_extensions import assert_never
  9. import aphrodite.common.envs as envs
  10. from aphrodite.common.config import (CacheConfig, DecodingConfig, DeviceConfig,
  11. EngineConfig, LoadConfig, LoRAConfig,
  12. ModelConfig, ParallelConfig,
  13. PromptAdapterConfig, SchedulerConfig,
  14. SpeculativeConfig)
  15. from aphrodite.common.logger import setup_logger
  16. from aphrodite.common.outputs import (EmbeddingRequestOutput, RequestOutput,
  17. RequestOutputFactory)
  18. from aphrodite.common.pooling_params import PoolingParams
  19. from aphrodite.common.sampling_params import SamplingParams
  20. from aphrodite.common.sequence import (EmbeddingSequenceGroupOutput,
  21. ExecuteModelRequest, PoolerOutput,
  22. SamplerOutput, Sequence, SequenceGroup,
  23. SequenceGroupMetadata, SequenceStatus)
  24. from aphrodite.common.utils import Counter, Device
  25. from aphrodite.engine.args_tools import EngineArgs
  26. from aphrodite.engine.metrics_types import StatLoggerBase, Stats
  27. from aphrodite.engine.output_processor.interfaces import (
  28. SequenceGroupOutputProcessor)
  29. from aphrodite.engine.output_processor.stop_checker import StopChecker
  30. from aphrodite.engine.output_processor.util import (
  31. create_output_by_sequence_group)
  32. from aphrodite.executor.executor_base import ExecutorBase
  33. from aphrodite.executor.ray_utils import initialize_ray_cluster
  34. from aphrodite.inputs import (INPUT_REGISTRY, EncoderDecoderLLMInputs,
  35. InputRegistry, LLMInputs, PromptInputs,
  36. SingletonPromptInputs)
  37. from aphrodite.inputs.parse import is_explicit_encoder_decoder_prompt
  38. from aphrodite.lora.request import LoRARequest
  39. from aphrodite.multimodal import MultiModalDataDict
  40. from aphrodite.processing.scheduler import (ScheduledSequenceGroup, Scheduler,
  41. SchedulerOutputs)
  42. from aphrodite.prompt_adapter.request import PromptAdapterRequest
  43. from aphrodite.transformers_utils.config import try_get_generation_config
  44. from aphrodite.transformers_utils.detokenizer import Detokenizer
  45. from aphrodite.transformers_utils.tokenizer_group import (
  46. BaseTokenizerGroup, init_tokenizer_from_configs)
  47. from aphrodite.version import __version__ as APHRODITE_VERSION
  48. _LOCAL_LOGGING_INTERVAL_SEC = 5
  49. APHRODITE_USE_RAY_SPMD_WORKER = envs.APHRODITE_USE_RAY_SPMD_WORKER
  50. def _load_generation_config_dict(model_config: ModelConfig) -> Dict[str, Any]:
  51. config = try_get_generation_config(
  52. model_config.model,
  53. trust_remote_code=model_config.trust_remote_code,
  54. revision=model_config.revision,
  55. )
  56. if config is None:
  57. return {}
  58. return config.to_diff_dict()
  59. _O = TypeVar("_O", RequestOutput, EmbeddingRequestOutput)
  60. PromptComponents = Tuple[Optional[str], List[int],
  61. Optional[MultiModalDataDict]]
  62. DecoderPromptComponents = Tuple[Optional[str], Optional[List[int]],
  63. Optional[MultiModalDataDict]]
  64. class AphroditeEngine:
  65. """An LLM engine that receives requests and generates texts.
  66. This is the main class for the Aphrodite engine. It receives requests
  67. from clients and generates texts from the LLM. It includes a tokenizer, a
  68. language model (possibly distributed across multiple GPUs), and GPU memory
  69. space allocated for intermediate states (aka KV cache). This class utilizes
  70. iteration-level scheduling and efficient memory management to maximize the
  71. serving throughput.
  72. The `LLM` class wraps this class for offline batched inference and the
  73. `AsyncAphrodite` class wraps this class for online serving.
  74. NOTE: The config arguments are derived from the `EngineArgs` class. For the
  75. comprehensive list of arguments, see `EngineArgs`.
  76. Args:
  77. model_config: The configuration related to the LLM model.
  78. cache_config: The configuration related to the KV cache memory
  79. management.
  80. parallel_config: The configuration related to distributed execution.
  81. scheduler_config: The configuration related to the request scheduler.
  82. device_config: The configuration related to the device.
  83. lora_config (Optional): The configuration related to serving multi-LoRA.
  84. speculative_config (Optional): The configuration related to speculative
  85. decoding.
  86. executor_class: The model executor class for managing distributed
  87. execution.
  88. prompt_adapter_config (Optional): The configuration related to serving
  89. prompt adapters.
  90. log_stats: Whether to log statistics.
  91. """
  92. DO_VALIDATE_OUTPUT: ClassVar[bool] = False
  93. """A flag to toggle whether to validate the type of request output."""
  94. @classmethod
  95. @contextmanager
  96. def enable_output_validation(cls):
  97. cls.DO_VALIDATE_OUTPUT = True
  98. yield
  99. cls.DO_VALIDATE_OUTPUT = False
  100. @classmethod
  101. def validate_output(
  102. cls,
  103. output: object,
  104. output_type: Type[_O],
  105. ) -> _O:
  106. do_validate = cls.DO_VALIDATE_OUTPUT
  107. if ((TYPE_CHECKING or do_validate)
  108. and not isinstance(output, output_type)):
  109. raise TypeError(f"Expected output of type {output_type}, "
  110. f"but found type {type(output)}")
  111. return output
  112. @classmethod
  113. def validate_outputs(
  114. cls,
  115. outputs: GenericSequence[object],
  116. output_type: Type[_O],
  117. ) -> List[_O]:
  118. do_validate = cls.DO_VALIDATE_OUTPUT
  119. outputs_: List[_O]
  120. if TYPE_CHECKING or do_validate:
  121. outputs_ = []
  122. for output in outputs:
  123. if not isinstance(output, output_type):
  124. raise TypeError(f"Expected output of type {output_type}, "
  125. f"but found type {type(output)}")
  126. outputs_.append(output)
  127. else:
  128. outputs_ = outputs
  129. return outputs_
  130. tokenizer: Optional[BaseTokenizerGroup]
  131. def __init__(
  132. self,
  133. model_config: ModelConfig,
  134. cache_config: CacheConfig,
  135. parallel_config: ParallelConfig,
  136. scheduler_config: SchedulerConfig,
  137. device_config: DeviceConfig,
  138. load_config: LoadConfig,
  139. lora_config: Optional[LoRAConfig],
  140. speculative_config: Optional[SpeculativeConfig],
  141. decoding_config: Optional[DecodingConfig],
  142. prompt_adapter_config: Optional[PromptAdapterConfig],
  143. executor_class: Type[ExecutorBase],
  144. log_stats: bool,
  145. stat_loggers: Optional[Dict[str, StatLoggerBase]] = None,
  146. input_registry: InputRegistry = INPUT_REGISTRY,
  147. ) -> None:
  148. try:
  149. import aphrodite.commit_id
  150. commit_id = True
  151. except ImportError:
  152. commit_id = False
  153. config_dict = {
  154. "Model": model_config.model,
  155. "Speculative Config": speculative_config,
  156. "DataType": model_config.dtype,
  157. "Model Load Format": load_config.load_format,
  158. "Tensor Parallel Size": parallel_config.tensor_parallel_size,
  159. "Pipeline Parallel Size": parallel_config.pipeline_parallel_size,
  160. "Disable Custom All-Reduce":
  161. parallel_config.disable_custom_all_reduce,
  162. "Quantization Format": model_config.quantization,
  163. "Context Length": model_config.max_model_len,
  164. "Enforce Eager Mode": model_config.enforce_eager,
  165. "Prefix Caching": cache_config.enable_prefix_caching,
  166. "KV Cache DataType": cache_config.cache_dtype,
  167. "Device": device_config.device,
  168. "Rope Scaling": model_config.rope_scaling,
  169. "Guided Decoding Backend": decoding_config
  170. }
  171. logger.info("-" * 85)
  172. if not commit_id:
  173. logger.info(
  174. f"Initializing Aphrodite Engine (v{APHRODITE_VERSION}) "
  175. "with the following config:")
  176. else:
  177. logger.info(f"Initializing Aphrodite Engine (v{APHRODITE_VERSION} "
  178. f"commit {aphrodite.__short_commit__}) with the "
  179. "following config:")
  180. for key, value in config_dict.items():
  181. if value is not None and not ((key == "Model Load Format" or key ==\
  182. "KV Cache DataType") and value == \
  183. "auto"):
  184. logger.info(f"{key} = {value!r}")
  185. logger.info("-" * 85)
  186. # TODO: Print more configs in debug mode.
  187. from aphrodite.plugins import load_general_plugins
  188. load_general_plugins()
  189. self.model_config = model_config
  190. self.cache_config = cache_config
  191. self.lora_config = lora_config
  192. self.parallel_config = parallel_config
  193. self.scheduler_config = scheduler_config
  194. self.device_config = device_config
  195. self.speculative_config = speculative_config
  196. self.load_config = load_config
  197. self.decoding_config = decoding_config or DecodingConfig()
  198. self.prompt_adapter_config = prompt_adapter_config
  199. self.log_stats = log_stats
  200. if not self.model_config.skip_tokenizer_init:
  201. self.tokenizer = self._init_tokenizer()
  202. self.detokenizer = Detokenizer(self.tokenizer)
  203. tokenizer_group = self.get_tokenizer_group()
  204. else:
  205. self.tokenizer = None
  206. self.detokenizer = None
  207. tokenizer_group = None
  208. # Ensure that the function doesn't contain a reference to self,
  209. # to avoid engine GC issues
  210. def get_tokenizer_for_seq(sequence: Sequence) -> PreTrainedTokenizer:
  211. assert tokenizer_group, ("tokenizer_group cannot be None, "
  212. "make sure skip_tokenizer_init is False")
  213. return tokenizer_group.get_lora_tokenizer(sequence.lora_request)
  214. self.seq_counter = Counter()
  215. self.generation_config_fields = _load_generation_config_dict(
  216. model_config)
  217. self.input_registry = input_registry
  218. self.input_processor = input_registry.create_input_processor(
  219. model_config)
  220. self.model_executor = executor_class(
  221. model_config=model_config,
  222. cache_config=cache_config,
  223. parallel_config=parallel_config,
  224. scheduler_config=scheduler_config,
  225. device_config=device_config,
  226. lora_config=lora_config,
  227. speculative_config=speculative_config,
  228. load_config=load_config,
  229. prompt_adapter_config=prompt_adapter_config,
  230. )
  231. if not self.model_config.embedding_mode:
  232. self._initialize_kv_caches()
  233. if self.tokenizer:
  234. # Ping the tokenizer to ensure liveness if it runs in a
  235. # different process.
  236. self.tokenizer.ping()
  237. # Create the scheduler.
  238. # NOTE: the cache_config here have been updated with the numbers of
  239. # GPU and CPU blocks, which are profiled in the distributed executor.
  240. self.scheduler = [
  241. Scheduler(scheduler_config, cache_config, lora_config,
  242. parallel_config.pipeline_parallel_size)
  243. for _ in range(parallel_config.pipeline_parallel_size)
  244. ]
  245. # Metric Logging.
  246. if self.log_stats:
  247. if stat_loggers is not None:
  248. self.stat_loggers = stat_loggers
  249. else:
  250. # Lazy import for prometheus multiprocessing.
  251. # We need to set PROMETHEUS_MULTIPROC_DIR environment variable
  252. # before prometheus_client is imported.
  253. # See https://prometheus.github.io/client_python/multiprocess/
  254. from aphrodite.engine.metrics import (LoggingStatLogger,
  255. PrometheusStatLogger)
  256. self.stat_loggers = {
  257. "logging":
  258. LoggingStatLogger(
  259. local_interval=_LOCAL_LOGGING_INTERVAL_SEC),
  260. "prometheus":
  261. PrometheusStatLogger(
  262. local_interval=_LOCAL_LOGGING_INTERVAL_SEC,
  263. labels=dict(model_name=model_config.served_model_name),
  264. max_model_len=self.model_config.max_model_len),
  265. }
  266. self.stat_loggers["prometheus"].info("cache_config",
  267. self.cache_config)
  268. # Create sequence output processor, e.g. for beam search or
  269. # speculative decoding.
  270. self.output_processor = (
  271. SequenceGroupOutputProcessor.create_output_processor(
  272. self.scheduler_config,
  273. self.detokenizer,
  274. self.scheduler,
  275. self.seq_counter,
  276. get_tokenizer_for_seq,
  277. stop_checker=StopChecker(
  278. self.scheduler_config.max_model_len,
  279. get_tokenizer_for_seq,
  280. ),
  281. ))
  282. def _initialize_kv_caches(self) -> None:
  283. """Initialize the KV cache in the worker(s).
  284. The workers will determine the number of blocks in both the GPU cache
  285. and the swap CPU cache.
  286. """
  287. num_gpu_blocks, num_cpu_blocks = (
  288. self.model_executor.determine_num_available_blocks())
  289. if self.cache_config.num_gpu_blocks_override is not None:
  290. num_gpu_blocks_override = self.cache_config.num_gpu_blocks_override
  291. logger.info(f"Overriding {num_gpu_blocks=} with "
  292. f"{num_gpu_blocks_override=}")
  293. num_gpu_blocks = num_gpu_blocks_override
  294. self.cache_config.num_gpu_blocks = num_gpu_blocks
  295. self.cache_config.num_cpu_blocks = num_cpu_blocks
  296. self.model_executor.initialize_cache(num_gpu_blocks, num_cpu_blocks)
  297. @classmethod
  298. def _get_executor_cls(cls,
  299. engine_config: EngineConfig) -> Type[ExecutorBase]:
  300. distributed_executor_backend = (
  301. engine_config.parallel_config.distributed_executor_backend)
  302. # Initialize the cluster and specify the executor class.
  303. if isinstance(distributed_executor_backend, type):
  304. if not issubclass(distributed_executor_backend, ExecutorBase):
  305. raise TypeError(
  306. "distributed_executor_backend must be a subclass of "
  307. f"ExecutorBase. Got {distributed_executor_backend}.")
  308. if distributed_executor_backend.uses_ray: # type: ignore
  309. initialize_ray_cluster(engine_config.parallel_config)
  310. executor_class = distributed_executor_backend
  311. elif engine_config.device_config.device_type == "neuron":
  312. from aphrodite.executor.neuron_executor import NeuronExecutor
  313. executor_class = NeuronExecutor
  314. elif engine_config.device_config.device_type == "tpu":
  315. if distributed_executor_backend == "ray":
  316. initialize_ray_cluster(engine_config.parallel_config)
  317. from aphrodite.executor.ray_tpu_executor import RayTPUExecutor
  318. executor_class = RayTPUExecutor
  319. else:
  320. assert distributed_executor_backend is None
  321. from aphrodite.executor.tpu_executor import TPUExecutor
  322. executor_class = TPUExecutor
  323. elif engine_config.device_config.device_type == "cpu":
  324. from aphrodite.executor.cpu_executor import CPUExecutor
  325. executor_class = CPUExecutor
  326. elif engine_config.device_config.device_type == "openvino":
  327. from aphrodite.executor.openvino_executor import OpenVINOExecutor
  328. executor_class = OpenVINOExecutor
  329. elif engine_config.device_config.device_type == "xpu":
  330. if distributed_executor_backend == "ray":
  331. initialize_ray_cluster(engine_config.parallel_config)
  332. from aphrodite.executor.ray_xpu_executor import RayXPUExecutor
  333. executor_class = RayXPUExecutor
  334. else:
  335. from aphrodite.executor.xpu_executor import XPUExecutor
  336. executor_class = XPUExecutor
  337. elif distributed_executor_backend == "ray":
  338. initialize_ray_cluster(engine_config.parallel_config)
  339. from aphrodite.executor.ray_gpu_executor import RayGPUExecutor
  340. executor_class = RayGPUExecutor
  341. elif distributed_executor_backend == "mp":
  342. from aphrodite.executor.multiproc_gpu_executor import (
  343. MultiprocessingGPUExecutor)
  344. assert not APHRODITE_USE_RAY_SPMD_WORKER, (
  345. "multiprocessing distributed executor backend does not "
  346. "support APHRODITE_USE_RAY_SPMD_WORKER=1")
  347. executor_class = MultiprocessingGPUExecutor
  348. else:
  349. from aphrodite.executor.gpu_executor import GPUExecutor
  350. executor_class = GPUExecutor
  351. return executor_class
  352. @classmethod
  353. def from_engine_args(
  354. cls,
  355. engine_args: EngineArgs,
  356. stat_loggers: Optional[Dict[str, StatLoggerBase]] = None,
  357. ) -> "AphroditeEngine":
  358. """Creates an LLM engine from the engine arguments."""
  359. # Create the engine configs.
  360. engine_config = engine_args.create_engine_config()
  361. executor_class = cls._get_executor_cls(engine_config)
  362. # Create the LLM engine.
  363. engine = cls(
  364. **engine_config.to_dict(),
  365. executor_class=executor_class,
  366. log_stats=not engine_args.disable_log_stats,
  367. stat_loggers=stat_loggers,
  368. )
  369. return engine
  370. def __reduce__(self):
  371. # This is to ensure that the AphroditeEngine is not referenced in
  372. # the closure used to initialize Ray worker actors
  373. raise RuntimeError("AphroditeEngine should not be pickled!")
  374. def __del__(self):
  375. # Shutdown the model executor when engine is garbage collected.
  376. # Use getattr since __init__ can fail before the field is set
  377. if model_executor := getattr(self, "model_executor", None):
  378. model_executor.shutdown()
  379. MISSING_TOKENIZER_GROUP_MSG = ("Unable to get tokenizer because "
  380. "skip_tokenizer_init is True")
  381. def get_tokenizer_group(
  382. self,
  383. fail_msg: str = MISSING_TOKENIZER_GROUP_MSG) -> BaseTokenizerGroup:
  384. if self.tokenizer is None:
  385. raise ValueError(fail_msg)
  386. return self.tokenizer
  387. def get_tokenizer(
  388. self,
  389. lora_request: Optional[LoRARequest] = None
  390. ) -> "PreTrainedTokenizer":
  391. return self.get_tokenizer_group().get_lora_tokenizer(lora_request)
  392. def _init_tokenizer(self) -> BaseTokenizerGroup:
  393. return init_tokenizer_from_configs(
  394. model_config=self.model_config,
  395. scheduler_config=self.scheduler_config,
  396. parallel_config=self.parallel_config,
  397. enable_lora=bool(self.lora_config))
  398. def _verify_args(self) -> None:
  399. self.model_config.verify_with_parallel_config(self.parallel_config)
  400. self.cache_config.verify_with_parallel_config(self.parallel_config)
  401. if self.lora_config:
  402. self.lora_config.verify_with_model_config(self.model_config)
  403. self.lora_config.verify_with_scheduler_config(
  404. self.scheduler_config)
  405. if self.prompt_adapter_config:
  406. self.prompt_adapter_config.verify_with_model_config(
  407. self.model_config)
  408. def _get_bos_token_id(self,
  409. lora_request: Optional[LoRARequest] = None
  410. ) -> Optional[int]:
  411. if self.tokenizer is None:
  412. logger.warning("Using None for BOS token id because tokenizer "
  413. "is not initialized")
  414. return None
  415. return self.tokenizer.get_lora_tokenizer(lora_request).bos_token_id
  416. def _get_eos_token_id(self,
  417. lora_request: Optional[LoRARequest] = None
  418. ) -> Optional[int]:
  419. if self.tokenizer is None:
  420. logger.warning("Using None for EOS token id because tokenizer "
  421. "is not initialized")
  422. return None
  423. return self.tokenizer.get_lora_tokenizer(lora_request).eos_token_id
  424. def _get_decoder_start_token_id(self) -> Optional[int]:
  425. '''
  426. Obtain the decoder start token id employed by an encoder/decoder
  427. model. Returns None for non-encoder/decoder models or if the
  428. model config is unavailable.
  429. '''
  430. if not self.is_encoder_decoder_model():
  431. logger.warning("Using None for decoder start token id because "
  432. "this is not an encoder/decoder model.")
  433. return None
  434. if (self.model_config is None or self.model_config.hf_config is None):
  435. logger.warning("Using None for decoder start token id because "
  436. "model config is not available.")
  437. return None
  438. dec_start_token_id = getattr(self.model_config.hf_config,
  439. 'decoder_start_token_id', None)
  440. if dec_start_token_id is None:
  441. logger.warning("Falling back on <BOS> for decoder start token id "
  442. "because decoder start token id is not available.")
  443. dec_start_token_id = self._get_bos_token_id()
  444. return dec_start_token_id
  445. def _add_processed_request(
  446. self,
  447. request_id: str,
  448. processed_inputs: Union[LLMInputs, EncoderDecoderLLMInputs],
  449. params: Union[SamplingParams, PoolingParams],
  450. arrival_time: float,
  451. lora_request: Optional[LoRARequest],
  452. prompt_adapter_request: Optional[PromptAdapterRequest] = None,
  453. ) -> None:
  454. # Create the sequences.
  455. block_size = self.cache_config.block_size
  456. seq_id = next(self.seq_counter)
  457. eos_token_id = self._get_eos_token_id(lora_request)
  458. seq = Sequence(seq_id, processed_inputs, block_size, eos_token_id,
  459. lora_request, prompt_adapter_request)
  460. encoder_seq = None
  461. if 'encoder_prompt_token_ids' in processed_inputs:
  462. encoder_seq = Sequence(seq_id,
  463. processed_inputs,
  464. block_size,
  465. eos_token_id,
  466. lora_request,
  467. prompt_adapter_request,
  468. from_decoder_prompt=False)
  469. # Create a SequenceGroup based on SamplingParams or PoolingParams
  470. if isinstance(params, SamplingParams):
  471. seq_group = self._create_sequence_group_with_sampling(
  472. request_id,
  473. seq,
  474. params,
  475. arrival_time=arrival_time,
  476. lora_request=lora_request,
  477. prompt_adapter_request=prompt_adapter_request,
  478. encoder_seq=encoder_seq,
  479. )
  480. elif isinstance(params, PoolingParams):
  481. seq_group = self._create_sequence_group_with_pooling(
  482. request_id,
  483. seq,
  484. params,
  485. arrival_time=arrival_time,
  486. lora_request=lora_request,
  487. prompt_adapter_request=prompt_adapter_request,
  488. encoder_seq=encoder_seq,
  489. )
  490. else:
  491. raise ValueError(
  492. "Either SamplingParams or PoolingParams must be provided.")
  493. # Add the sequence group to the scheduler with least unfinished seqs.
  494. costs = [
  495. scheduler.get_num_unfinished_seq_groups()
  496. for scheduler in self.scheduler
  497. ]
  498. min_cost_scheduler = self.scheduler[costs.index(min(costs))]
  499. min_cost_scheduler.add_seq_group(seq_group)
  500. def stop_remote_worker_execution_loop(self) -> None:
  501. self.model_executor.stop_remote_worker_execution_loop()
  502. _LLMInputComponentsType = Tuple[str, List[int]]
  503. def _prepare_decoder_input_ids_for_generation(
  504. self,
  505. decoder_input_ids: Optional[List[int]],
  506. ) -> List[int]:
  507. """
  508. Prepares `decoder_input_ids` for generation with encoder-decoder models.
  509. Based on
  510. https://github.com/huggingface/transformers/blob/
  511. 4037a2b5b1278736e566aec12e169100275545ea/
  512. src/transformers/generation/utils.py
  513. specifically GenerationMixin._prepare_decoder_input_ids_for_generation()
  514. Arguments:
  515. * decoder_input_ids: input token ids to preprocess
  516. Returns:
  517. * Processed token list
  518. """
  519. decoder_start_token_id = self._get_decoder_start_token_id()
  520. assert decoder_start_token_id is not None
  521. if decoder_input_ids is None:
  522. # no decoder prompt input ->
  523. # use decoder_start_token_id as decoder_input_ids
  524. decoder_input_ids = self._get_default_enc_dec_decoder_prompt()
  525. if (len(decoder_input_ids) == 0
  526. or decoder_input_ids[0] != decoder_start_token_id):
  527. decoder_input_ids = [decoder_start_token_id] + decoder_input_ids
  528. return decoder_input_ids
  529. def _tokenize_prompt(
  530. self,
  531. prompt: str,
  532. request_id: str,
  533. lora_request: Optional[LoRARequest],
  534. ) -> List[int]:
  535. '''
  536. Wrapper around application of the model's tokenizer.
  537. Arguments:
  538. * prompt
  539. * request_id
  540. * lora_request
  541. Returns:
  542. * prompt token ids
  543. '''
  544. tokenizer = self.get_tokenizer_group("prompts must be None if "
  545. "skip_tokenizer_init is True")
  546. return tokenizer.encode(request_id=request_id,
  547. prompt=prompt,
  548. lora_request=lora_request)
  549. def _extract_prompt_components(
  550. self,
  551. inputs: SingletonPromptInputs,
  552. request_id: str,
  553. lora_request: Optional[LoRARequest] = None,
  554. ) -> PromptComponents:
  555. '''
  556. Extract the components of any single encoder or decoder input prompt.
  557. Arguments:
  558. * request_id
  559. * inputs: single encoder or decoder input prompt
  560. * lora_request: this is only valid for decoder prompts
  561. Returns:
  562. * prompt
  563. * prompt_token_ids
  564. * multi_modal_data
  565. '''
  566. if isinstance(inputs, str):
  567. prompt = inputs
  568. prompt_token_ids = self._tokenize_prompt(
  569. prompt,
  570. request_id=request_id,
  571. lora_request=lora_request,
  572. )
  573. multi_modal_data = None
  574. elif isinstance(inputs, dict):
  575. if "prompt_token_ids" in inputs:
  576. prompt = None
  577. prompt_token_ids = inputs["prompt_token_ids"]
  578. else:
  579. # NOTE: This extra assignment is required to pass mypy
  580. prompt = parsed_prompt = inputs["prompt"]
  581. prompt_token_ids = self._tokenize_prompt(
  582. parsed_prompt,
  583. request_id=request_id,
  584. lora_request=lora_request,
  585. )
  586. multi_modal_data = inputs.get("multi_modal_data")
  587. else:
  588. assert_never(inputs)
  589. return prompt, prompt_token_ids, multi_modal_data
  590. def _apply_prompt_adapter(
  591. self,
  592. prompt_token_ids: List[int],
  593. prompt_adapter_request: Optional[PromptAdapterRequest],
  594. ) -> List[int]:
  595. if prompt_adapter_request:
  596. prompt_token_ids = (
  597. [0] * prompt_adapter_request.prompt_adapter_num_virtual_tokens
  598. + prompt_token_ids)
  599. return prompt_token_ids
  600. def _get_default_enc_dec_decoder_prompt(self) -> List[int]:
  601. '''
  602. Specifically for encoder/decoder models:
  603. generate a default decoder prompt for when
  604. the user specifies only the encoder prompt.
  605. Encoder/decoder models utilize the decoder
  606. prompt in different ways; as new models are
  607. added, it is intended that this function
  608. will be extended to produce differing
  609. default decoder prompts, depending on the
  610. model variety.
  611. Absent a special case, the default behavior
  612. of this method is to mirror the behavior of
  613. the HuggingFace (HF) GenerationMixin for a None
  614. decoder prompt, which is to employ a logit processor
  615. setting to force the first decoded token to be <BOS>.
  616. Here, this behavior is approximated by having the
  617. "default" decoder prompt be <BOS>.
  618. However, it is possible that in the future
  619. other models may have different or more
  620. complex logic for the default decoder prompt.
  621. This motivates having a special helper method
  622. for default decoder prompts.
  623. Returns:
  624. * prompt_token_ids
  625. '''
  626. bos_token_id = self._get_bos_token_id()
  627. assert bos_token_id is not None
  628. return [bos_token_id]
  629. def _build_enc_dec_llm_inputs(
  630. self,
  631. encoder_comps: PromptComponents,
  632. decoder_comps: DecoderPromptComponents,
  633. ) -> EncoderDecoderLLMInputs:
  634. encoder_prompt, encoder_prompt_ids, encoder_mm_data = encoder_comps
  635. decoder_prompt, decoder_prompt_ids, decoder_mm_data = decoder_comps
  636. if encoder_mm_data is not None or decoder_mm_data is not None:
  637. raise ValueError("Multi-modal encoder-decoder models are "
  638. "not supported yet")
  639. decoder_prompt_ids = (
  640. self._prepare_decoder_input_ids_for_generation(decoder_prompt_ids))
  641. return EncoderDecoderLLMInputs(
  642. prompt_token_ids=decoder_prompt_ids,
  643. prompt=decoder_prompt,
  644. encoder_prompt_token_ids=encoder_prompt_ids,
  645. encoder_prompt=encoder_prompt,
  646. )
  647. def _process_encoder_decoder_prompt(
  648. self,
  649. inputs: PromptInputs,
  650. request_id: str,
  651. ) -> EncoderDecoderLLMInputs:
  652. '''
  653. For encoder/decoder models only:
  654. Process an input prompt into an
  655. :class:`EncoderDecoderLLMInputs` instance.
  656. There are two types of input prompts:
  657. singleton prompts which carry only the
  658. encoder prompt, and explicit encoder/decoder
  659. prompts which carry both the encoder and the
  660. decoder prompts as member variables.
  661. This function handles the following scenarios:
  662. * Singleton encoder prompt: extract encoder prompt
  663. token ids & infer default decoder prompt token ids
  664. * Explicit encoder/decoder prompt: extract encoder
  665. and decoder prompt token ids
  666. Note that for Explicit encoder/decoder prompts,
  667. each sub-prompt (encoder or decoder prompt) can
  668. have any possible singleton type; thus this
  669. method relies on helper functions to obtain
  670. token ids for the sub-prompts.
  671. Arguments:
  672. * inputs: an input prompt
  673. * request_id
  674. Returns:
  675. * :class:`EncoderDecoderLLMInputs` instance
  676. '''
  677. encoder_comps: PromptComponents
  678. decoder_comps: DecoderPromptComponents
  679. if is_explicit_encoder_decoder_prompt(inputs):
  680. encoder_comps = self._extract_prompt_components(
  681. inputs["encoder_prompt"],
  682. request_id=request_id,
  683. )
  684. if (decoder_input := inputs["decoder_prompt"]) is None:
  685. decoder_comps = None, None, None
  686. else:
  687. decoder_comps = self._extract_prompt_components(
  688. decoder_input,
  689. request_id=request_id,
  690. )
  691. else:
  692. encoder_comps = self._extract_prompt_components(
  693. inputs,
  694. request_id=request_id,
  695. )
  696. decoder_comps = None, None, None
  697. return self._build_enc_dec_llm_inputs(encoder_comps, decoder_comps)
  698. def _build_decoder_only_llm_inputs(
  699. self,
  700. prompt_comps: PromptComponents,
  701. prompt_adapter_request: Optional[PromptAdapterRequest],
  702. ) -> LLMInputs:
  703. prompt, prompt_token_ids, multi_modal_data = prompt_comps
  704. prompt_token_ids = self._apply_prompt_adapter(
  705. prompt_token_ids, prompt_adapter_request=prompt_adapter_request)
  706. return LLMInputs(prompt_token_ids=prompt_token_ids,
  707. prompt=prompt,
  708. multi_modal_data=multi_modal_data)
  709. def _process_decoder_only_prompt(
  710. self,
  711. inputs: SingletonPromptInputs,
  712. request_id: str,
  713. lora_request: Optional[LoRARequest] = None,
  714. prompt_adapter_request: Optional[PromptAdapterRequest] = None,
  715. ) -> LLMInputs:
  716. '''
  717. For decoder-only models:
  718. Process an input prompt into an :class:`LLMInputs` instance.
  719. Arguments:
  720. * inputs: input prompt
  721. * request_id
  722. * lora_request
  723. * prompt_adapter_request
  724. Returns:
  725. * :class:`LLMInputs` instance
  726. '''
  727. prompt_comps = self._extract_prompt_components(
  728. inputs,
  729. request_id=request_id,
  730. lora_request=lora_request,
  731. )
  732. return self._build_decoder_only_llm_inputs(
  733. prompt_comps,
  734. prompt_adapter_request=prompt_adapter_request,
  735. )
  736. def process_model_inputs(
  737. self,
  738. inputs: PromptInputs,
  739. request_id: str,
  740. lora_request: Optional[LoRARequest] = None,
  741. prompt_adapter_request: Optional[PromptAdapterRequest] = None,
  742. ) -> Union[LLMInputs, EncoderDecoderLLMInputs]:
  743. if self.is_encoder_decoder_model():
  744. # Encoder-decoder model requires special mapping of
  745. # input prompts to encoder & decoder
  746. model_inputs = self._process_encoder_decoder_prompt(
  747. inputs,
  748. request_id=request_id,
  749. )
  750. else:
  751. if is_explicit_encoder_decoder_prompt(inputs):
  752. raise ValueError("Cannot pass encoder-decoder prompt "
  753. "to decoder-only models")
  754. # Decoder-only operation
  755. model_inputs = self._process_decoder_only_prompt(
  756. inputs,
  757. request_id=request_id,
  758. lora_request=lora_request,
  759. prompt_adapter_request=prompt_adapter_request,
  760. )
  761. return self.input_processor(model_inputs)
  762. def add_request(
  763. self,
  764. request_id: str,
  765. inputs: PromptInputs,
  766. params: Union[SamplingParams, PoolingParams],
  767. arrival_time: Optional[float] = None,
  768. lora_request: Optional[LoRARequest] = None,
  769. prompt_adapter_request: Optional[PromptAdapterRequest] = None,
  770. ) -> None:
  771. """Add a request to the engine's request pool.
  772. The request is added to the request pool and will be processed by the
  773. scheduler as `engine.step()` is called. The exact scheduling policy is
  774. determined by the scheduler.
  775. Args:
  776. request_id: The unique ID of the request.
  777. prompt: The prompt string. Can be None if prompt_token_ids is
  778. provided.
  779. params: Parameters for sampling or pooling. SamplingParams
  780. for text generation. PoolingParams for pooling.
  781. prompt_token_ids: The token IDs of the prompt. If None, we
  782. use the tokenizer to convert the prompts to token IDs.
  783. arrival_time: The arrival time of the request. If None, we use
  784. the current monotonic time.
  785. multi_modal_data: Multi modal data per request.
  786. Details:
  787. - Set arrival_time to the current time if it is None.
  788. - Set prompt_token_ids to the encoded prompt if it is None.
  789. - Create `best_of` number of :class:`~aphrodite.common.sequence`
  790. objects.
  791. - Create a :class:`~aphrodite.common.sequenceGroup` object
  792. from the list of :class:`~aphrodite.common.sequence`.
  793. - Add the :class:`~aphrodite.common.sequenceGroup` object to the
  794. scheduler.
  795. Example:
  796. >>> # initialize engine
  797. >>> engine = AphroditeEngine.from_engine_args(engine_args)
  798. >>> # set request arguments
  799. >>> example_prompt = "Who is the president of the United States?"
  800. >>> sampling_params = SamplingParams(temperature=0.0)
  801. >>> request_id = 0
  802. >>>
  803. >>> # add the request to the engine
  804. >>> engine.add_request(
  805. >>> str(request_id),
  806. >>> example_prompt,
  807. >>> SamplingParams(temperature=0.0))
  808. >>> # continue the request processing
  809. >>> ...
  810. """
  811. if lora_request is not None and not self.lora_config:
  812. raise ValueError(f"Got lora_request {lora_request} but LoRA is "
  813. "not enabled!")
  814. if arrival_time is None:
  815. arrival_time = time.time()
  816. processed_inputs = self.process_model_inputs(
  817. inputs,
  818. request_id=request_id,
  819. lora_request=lora_request,
  820. prompt_adapter_request=prompt_adapter_request,
  821. )
  822. self._add_processed_request(
  823. request_id=request_id,
  824. processed_inputs=processed_inputs,
  825. params=params,
  826. arrival_time=arrival_time,
  827. lora_request=lora_request,
  828. prompt_adapter_request=prompt_adapter_request,
  829. )
  830. def _create_sequence_group_with_sampling(
  831. self,
  832. request_id: str,
  833. seq: Sequence,
  834. sampling_params: SamplingParams,
  835. arrival_time: float,
  836. lora_request: Optional[LoRARequest],
  837. prompt_adapter_request: Optional[PromptAdapterRequest] = None,
  838. encoder_seq: Optional[Sequence] = None,
  839. ) -> SequenceGroup:
  840. """Creates a SequenceGroup with SamplingParams."""
  841. max_logprobs = self.get_model_config().max_logprobs
  842. if (sampling_params.logprobs
  843. and sampling_params.logprobs > max_logprobs) or (
  844. sampling_params.prompt_logprobs
  845. and sampling_params.prompt_logprobs > max_logprobs):
  846. raise ValueError(f"Cannot request more than "
  847. f"{max_logprobs} logprobs.")
  848. # Defensive copy of SamplingParams, which are used by the sampler,
  849. # this doesn't deep-copy LogitsProcessor objects
  850. sampling_params = sampling_params.clone()
  851. sampling_params.update_from_generation_config(
  852. self.generation_config_fields, seq.eos_token_id)
  853. # Create the sequence group.
  854. seq_group = SequenceGroup(
  855. request_id=request_id,
  856. seqs=[seq],
  857. arrival_time=arrival_time,
  858. sampling_params=sampling_params,
  859. lora_request=lora_request,
  860. prompt_adapter_request=prompt_adapter_request,
  861. encoder_seq=encoder_seq)
  862. return seq_group
  863. def _create_sequence_group_with_pooling(
  864. self,
  865. request_id: str,
  866. seq: Sequence,
  867. pooling_params: PoolingParams,
  868. arrival_time: float,
  869. lora_request: Optional[LoRARequest],
  870. prompt_adapter_request: Optional[PromptAdapterRequest] = None,
  871. encoder_seq: Optional[Sequence] = None,
  872. ) -> SequenceGroup:
  873. """Creates a SequenceGroup with PoolingParams."""
  874. # Defensive copy of PoolingParams, which are used by the pooler
  875. pooling_params = pooling_params.clone()
  876. # Create the sequence group.
  877. seq_group = SequenceGroup(
  878. request_id=request_id,
  879. seqs=[seq],
  880. arrival_time=arrival_time,
  881. lora_request=lora_request,
  882. pooling_params=pooling_params,
  883. prompt_adapter_request=prompt_adapter_request,
  884. encoder_seq=encoder_seq)
  885. return seq_group
  886. def abort_request(self, request_id: Union[str, Iterable[str]]) -> None:
  887. """Aborts a request(s) with the given ID.
  888. Args:
  889. request_id: The ID(s) of the request to abort.
  890. Details:
  891. - Refer to the
  892. :meth:`~aphrodite.processing.scheduler.Scheduler.abort_seq_group`
  893. from class :class:`~aphrodite.processing.scheduler.Scheduler`.
  894. Example:
  895. >>> # initialize engine and add a request with request_id
  896. >>> request_id = str(0)
  897. >>> # abort the request
  898. >>> engine.abort_request(request_id)
  899. """
  900. for scheduler in self.scheduler:
  901. scheduler.abort_seq_group(request_id)
  902. def get_model_config(self) -> ModelConfig:
  903. """Gets the model configuration."""
  904. return self.model_config
  905. def get_parallel_config(self) -> ParallelConfig:
  906. """Gets the parallel configuration."""
  907. return self.parallel_config
  908. def get_decoding_config(self) -> DecodingConfig:
  909. """Gets the decoding configuration."""
  910. return self.decoding_config
  911. def get_scheduler_config(self) -> SchedulerConfig:
  912. """Gets the scheduler configuration."""
  913. return self.scheduler_config
  914. def get_lora_config(self) -> LoRAConfig:
  915. """Gets the LoRA configuration."""
  916. return self.lora_config
  917. def get_num_unfinished_requests(self) -> int:
  918. """Gets the number of unfinished requests."""
  919. return sum(scheduler.get_num_unfinished_seq_groups()
  920. for scheduler in self.scheduler)
  921. def has_unfinished_requests(self) -> bool:
  922. """Returns True if there are unfinished requests."""
  923. return any(scheduler.has_unfinished_seqs()
  924. for scheduler in self.scheduler)
  925. def has_unfinished_requests_for_virtual_engine(
  926. self, virtual_engine: int) -> bool:
  927. """
  928. Returns True if there are unfinished requests for the virtual engine.
  929. """
  930. return self.scheduler[virtual_engine].has_unfinished_seqs()
  931. def _process_sequence_group_outputs(
  932. self,
  933. seq_group: SequenceGroup,
  934. outputs: List[EmbeddingSequenceGroupOutput],
  935. ) -> None:
  936. seq_group.embeddings = outputs[0].embeddings
  937. for seq in seq_group.get_seqs():
  938. seq.status = SequenceStatus.FINISHED_STOPPED
  939. return
  940. def _process_model_outputs(
  941. self,
  942. output: GenericSequence[Union[SamplerOutput, PoolerOutput]],
  943. scheduled_seq_groups: List[ScheduledSequenceGroup],
  944. ignored_seq_groups: List[SequenceGroup],
  945. seq_group_metadata_list: List[SequenceGroupMetadata],
  946. ) -> List[Union[RequestOutput, EmbeddingRequestOutput]]:
  947. """Apply the model output to the sequences in the scheduled seq groups.
  948. Returns RequestOutputs that can be returned to the client.
  949. """
  950. now = time.time()
  951. # Organize outputs by [sequence group][step] instead of
  952. # [step][sequence group].
  953. output_by_sequence_group = create_output_by_sequence_group(
  954. output, num_seq_groups=len(scheduled_seq_groups))
  955. # Update the scheduled sequence groups with the model outputs.
  956. for scheduled_seq_group, outputs, seq_group_meta in zip(
  957. scheduled_seq_groups, output_by_sequence_group,
  958. seq_group_metadata_list):
  959. seq_group = scheduled_seq_group.seq_group
  960. seq_group.update_num_computed_tokens(
  961. scheduled_seq_group.token_chunk_size)
  962. if self.model_config.embedding_mode:
  963. self._process_sequence_group_outputs(seq_group, outputs)
  964. continue
  965. self.output_processor.process_prompt_logprob(seq_group, outputs)
  966. if seq_group_meta.do_sample:
  967. self.output_processor.process_outputs(seq_group, outputs)
  968. # Free the finished sequence groups.
  969. for scheduler in self.scheduler:
  970. scheduler.free_finished_seq_groups()
  971. # Create the outputs.
  972. request_outputs: List[Union[RequestOutput,
  973. EmbeddingRequestOutput]] = []
  974. for scheduled_seq_group in scheduled_seq_groups:
  975. seq_group = scheduled_seq_group.seq_group
  976. seq_group.maybe_set_first_token_time(now)
  977. request_output = RequestOutputFactory.create(seq_group)
  978. request_outputs.append(request_output)
  979. for seq_group in ignored_seq_groups:
  980. request_output = RequestOutputFactory.create(seq_group)
  981. request_outputs.append(request_output)
  982. return request_outputs
  983. def step(self) -> List[Union[RequestOutput, EmbeddingRequestOutput]]:
  984. """Performs one decoding iteration and returns newly generated results.
  985. .. figure:: https://i.imgur.com/sv2HssD.png
  986. :alt: Overview of the step function
  987. :align: center
  988. Overview of the step function.
  989. Details:
  990. - Step 1: Schedules the sequences to be executed in the next
  991. iteration and the token blocks to be swapped in/out/copy.
  992. - Depending on the scheduling policy,
  993. sequences may be `preempted/reordered`.
  994. - A Sequence Group (SG) refer to a group of sequences
  995. that are generated from the same prompt.
  996. - Step 2: Calls the distributed executor to execute the model.
  997. - Step 3: Processes the model output. This mainly includes:
  998. - Decodes the relevant outputs.
  999. - Updates the scheduled sequence groups with model outputs
  1000. based on its `sampling parameters` (`use_beam_search` or not).
  1001. - Frees the finished sequence groups.
  1002. - Finally, it creates and returns the newly generated results.
  1003. Example:
  1004. >>> # Please see the example/ folder for more detailed examples.
  1005. >>>
  1006. >>> # initialize engine and request arguments
  1007. >>> engine = AphroditeEngine.from_engine_args(engine_args)
  1008. >>> example_inputs = [(0, "What is LLM?",
  1009. >>> SamplingParams(temperature=0.0))]
  1010. >>>
  1011. >>> # Start the engine with an event loop
  1012. >>> while True:
  1013. >>> if example_inputs:
  1014. >>> req_id, prompt, sampling_params = example_inputs.pop(0)
  1015. >>> engine.add_request(str(req_id), prompt, sampling_params)
  1016. >>>
  1017. >>> # continue the request processing
  1018. >>> request_outputs = engine.step()
  1019. >>> for request_output in request_outputs:
  1020. >>> if request_output.finished:
  1021. >>> # return or show the request output
  1022. >>>
  1023. >>> if not (engine.has_unfinished_requests() or example_inputs):
  1024. >>> break
  1025. """
  1026. if self.parallel_config.pipeline_parallel_size > 1:
  1027. raise NotImplementedError(
  1028. "Pipeline parallelism is only supported through AsyncAphrodite "
  1029. "as performance will be severely degraded otherwise.")
  1030. if self.scheduler_config.num_scheduler_steps > 1:
  1031. raise NotImplementedError(
  1032. "Multiple scheduler steps (multi-step) are only supported "
  1033. "through AsyncAphrodite.")
  1034. seq_group_metadata_list, scheduler_outputs = self.scheduler[
  1035. 0].schedule()
  1036. if not scheduler_outputs.is_empty():
  1037. finished_requests_ids = self.scheduler[
  1038. 0].get_and_reset_finished_requests_ids()
  1039. execute_model_req = ExecuteModelRequest(
  1040. seq_group_metadata_list=seq_group_metadata_list,
  1041. blocks_to_swap_in=scheduler_outputs.blocks_to_swap_in,
  1042. blocks_to_swap_out=scheduler_outputs.blocks_to_swap_out,
  1043. blocks_to_copy=scheduler_outputs.blocks_to_copy,
  1044. num_lookahead_slots=scheduler_outputs.num_lookahead_slots,
  1045. running_queue_size=scheduler_outputs.running_queue_size,
  1046. finished_requests_ids=finished_requests_ids,
  1047. )
  1048. output = self.model_executor.execute_model(
  1049. execute_model_req=execute_model_req)
  1050. else:
  1051. output = []
  1052. request_outputs = self._process_model_outputs(
  1053. output, scheduler_outputs.scheduled_seq_groups,
  1054. scheduler_outputs.ignored_seq_groups, seq_group_metadata_list)
  1055. # Log stats.
  1056. self.do_log_stats(scheduler_outputs, output)
  1057. if not self.has_unfinished_requests():
  1058. # Stop the execute model loop in parallel workers until there are
  1059. # more requests to process. This avoids waiting indefinitely in
  1060. # torch.distributed ops which may otherwise timeout, and unblocks
  1061. # the RPC thread in the workers so that they can process any other
  1062. # queued control plane messages, such as add/remove lora adapters.
  1063. self.model_executor.stop_remote_worker_execution_loop()
  1064. return request_outputs
  1065. def add_logger(self, logger_name: str, logger: StatLoggerBase) -> None:
  1066. if logger_name in self.stat_loggers:
  1067. raise KeyError(f"Logger with name {logger_name} already exists.")
  1068. self.stat_loggers[logger_name] = logger
  1069. def remove_logger(self, logger_name: str) -> None:
  1070. if logger_name not in self.stat_loggers:
  1071. raise KeyError(f"Logger with name {logger_name} does not exist.")
  1072. del self.stat_loggers[logger_name]
  1073. def do_log_stats(
  1074. self,
  1075. scheduler_outputs: Optional[SchedulerOutputs] = None,
  1076. model_output: Optional[List[SamplerOutput]] = None) -> None:
  1077. """Forced log when no requests active."""
  1078. if self.log_stats:
  1079. stats = self._get_stats(scheduler_outputs, model_output)
  1080. for loggers in self.stat_loggers.values():
  1081. loggers.log(stats)
  1082. def _get_stats(
  1083. self,
  1084. scheduler_outputs: Optional[SchedulerOutputs],
  1085. model_output: Optional[List[SamplerOutput]] = None) -> Stats:
  1086. """Get Stats to be Logged to Prometheus.
  1087. Args:
  1088. scheduler_outputs: Optional, used to populate metrics related to
  1089. the scheduled batch,
  1090. model_output: Optional, used to emit speculative decoding metrics
  1091. which are created by the workers.
  1092. """
  1093. now = time.time()
  1094. # System State
  1095. # Scheduler State
  1096. num_running_sys = sum(
  1097. len(scheduler.running) for scheduler in self.scheduler)
  1098. num_swapped_sys = sum(
  1099. len(scheduler.swapped) for scheduler in self.scheduler)
  1100. num_waiting_sys = sum(
  1101. len(scheduler.waiting) for scheduler in self.scheduler)
  1102. # KV Cache Usage in %
  1103. num_total_gpu = self.cache_config.num_gpu_blocks
  1104. gpu_cache_usage_sys = 0.
  1105. if num_total_gpu is not None:
  1106. num_free_gpu = sum(
  1107. scheduler.block_manager.get_num_free_gpu_blocks()
  1108. for scheduler in self.scheduler)
  1109. if not self.model_config.is_attention_free():
  1110. gpu_cache_usage_sys = 1.0 - (num_free_gpu / num_total_gpu)
  1111. else:
  1112. gpu_cache_usage_sys = 0.0
  1113. num_total_cpu = self.cache_config.num_cpu_blocks
  1114. cpu_cache_usage_sys = 0.
  1115. if num_total_cpu is not None and num_total_cpu > 0:
  1116. num_free_cpu = sum(
  1117. scheduler.block_manager.get_num_free_cpu_blocks()
  1118. for scheduler in self.scheduler)
  1119. if not self.model_config.is_attention_free():
  1120. cpu_cache_usage_sys = 1.0 - (num_free_cpu / num_total_cpu)
  1121. else:
  1122. cpu_cache_usage_sys = 0.0
  1123. # Prefix Cache Hit Rate. Note that we always use
  1124. # the cache hit rate of the first virtual engine.
  1125. cpu_prefix_cache_hit_rate = self.scheduler[
  1126. 0].get_prefix_cache_hit_rate(Device.CPU)
  1127. gpu_prefix_cache_hit_rate = self.scheduler[
  1128. 0].get_prefix_cache_hit_rate(Device.GPU)
  1129. # Iteration stats
  1130. num_prompt_tokens_iter = 0
  1131. num_generation_tokens_iter = 0
  1132. time_to_first_tokens_iter: List[float] = []
  1133. time_per_output_tokens_iter: List[float] = []
  1134. num_preemption_iter = (0 if scheduler_outputs is None else
  1135. scheduler_outputs.preempted)
  1136. # Request stats
  1137. # Latency
  1138. time_e2e_requests: List[float] = []
  1139. # Metadata
  1140. num_prompt_tokens_requests: List[int] = []
  1141. num_generation_tokens_requests: List[int] = []
  1142. best_of_requests: List[int] = []
  1143. n_requests: List[int] = []
  1144. finished_reason_requests: List[str] = []
  1145. # NOTE: This loop assumes prefill seq_groups are before
  1146. # decode seq_groups in scheduled_seq_groups.
  1147. if scheduler_outputs is not None:
  1148. num_generation_tokens_from_prefill_groups = 0.
  1149. # NOTE: if scheduler_outputs.num_prefill_groups > 0 and
  1150. # the len of scheduler_outputs.scheduled_seq_groups is !=
  1151. # scheduler_outputs.num_prefill_groups, this means that
  1152. # chunked prefills have been detected.
  1153. for idx, scheduled_seq_group in enumerate(
  1154. scheduler_outputs.scheduled_seq_groups):
  1155. group_was_prefill = idx < scheduler_outputs.num_prefill_groups
  1156. seq_group = scheduled_seq_group.seq_group
  1157. # NOTE: a seq_group that completed all of its prefill tokens
  1158. # in the last iteration will have seq_group.is_prefill() = False
  1159. # with group_was_prefill = True
  1160. if group_was_prefill:
  1161. # Number of prompt tokens.
  1162. num_prompt_tokens_iter += (
  1163. scheduled_seq_group.token_chunk_size)
  1164. # If the seq_group just finished the prefill state
  1165. # get TTFT.
  1166. if not seq_group.is_prefill():
  1167. latency = seq_group.get_last_latency(now)
  1168. time_to_first_tokens_iter.append(latency)
  1169. # One generation token per finished prefill.
  1170. num_generation_tokens_from_prefill_groups += (
  1171. seq_group.num_seqs())
  1172. else:
  1173. # TPOTs.
  1174. latency = seq_group.get_last_latency(now)
  1175. time_per_output_tokens_iter.append(latency)
  1176. # Because of chunked prefill, we can have a single sequence
  1177. # group that does multiple prompt_runs. To prevent logging
  1178. # the same metadata more than once per request, we standardize
  1179. # on logging request level information for finished requests,
  1180. # which can only happen once.
  1181. if seq_group.is_finished():
  1182. # Latency timings
  1183. time_e2e_requests.append(now -
  1184. seq_group.metrics.arrival_time)
  1185. # Metadata
  1186. num_prompt_tokens_requests.append(
  1187. len(seq_group.prompt_token_ids))
  1188. num_generation_tokens_requests.extend([
  1189. seq.get_output_len()
  1190. for seq in seq_group.get_finished_seqs()
  1191. ])
  1192. if seq_group.sampling_params is not None:
  1193. best_of_requests.append(
  1194. seq_group.sampling_params.best_of)
  1195. n_requests.append(seq_group.sampling_params.n)
  1196. finished_reason_requests.extend([
  1197. SequenceStatus.get_finished_reason(seq.status)
  1198. for seq in seq_group.get_finished_seqs()
  1199. ])
  1200. # Number of generation tokens.
  1201. # num_batched_tokens equals the number of prompt_tokens plus the
  1202. # number of decode_tokens in a single iteration. So,
  1203. # num_generation_tokens = num_batched_tokens - num_prompt_tokens
  1204. # + num_generation_tokens_from_prefill_groups (since we generate
  1205. # one token on prefills on iters where the prefill finishes).
  1206. num_generation_tokens_iter = (
  1207. scheduler_outputs.num_batched_tokens - num_prompt_tokens_iter +
  1208. num_generation_tokens_from_prefill_groups)
  1209. # Spec decode, if enabled, emits specialized metrics from the worker in
  1210. # sampler output.
  1211. if model_output and (model_output[0].spec_decode_worker_metrics
  1212. is not None):
  1213. spec_decode_metrics = model_output[0].spec_decode_worker_metrics
  1214. else:
  1215. spec_decode_metrics = None
  1216. return Stats(
  1217. now=now,
  1218. # System stats
  1219. # Scheduler State
  1220. num_running_sys=num_running_sys,
  1221. num_swapped_sys=num_swapped_sys,
  1222. num_waiting_sys=num_waiting_sys,
  1223. # KV Cache Usage in %
  1224. gpu_cache_usage_sys=gpu_cache_usage_sys,
  1225. cpu_cache_usage_sys=cpu_cache_usage_sys,
  1226. # Prefix Cache Hit Rate
  1227. cpu_prefix_cache_hit_rate=cpu_prefix_cache_hit_rate,
  1228. gpu_prefix_cache_hit_rate=gpu_prefix_cache_hit_rate,
  1229. # Iteration stats
  1230. num_prompt_tokens_iter=num_prompt_tokens_iter,
  1231. num_generation_tokens_iter=num_generation_tokens_iter,
  1232. time_to_first_tokens_iter=time_to_first_tokens_iter,
  1233. time_per_output_tokens_iter=time_per_output_tokens_iter,
  1234. spec_decode_metrics=spec_decode_metrics,
  1235. num_preemption_iter=num_preemption_iter,
  1236. # Request stats
  1237. # Latency
  1238. time_e2e_requests=time_e2e_requests,
  1239. # Metadata
  1240. num_prompt_tokens_requests=num_prompt_tokens_requests,
  1241. num_generation_tokens_requests=num_generation_tokens_requests,
  1242. best_of_requests=best_of_requests,
  1243. n_requests=n_requests,
  1244. finished_reason_requests=finished_reason_requests,
  1245. )
  1246. def add_lora(self, lora_request: LoRARequest) -> bool:
  1247. return self.model_executor.add_lora(lora_request)
  1248. def remove_lora(self, lora_id: int) -> bool:
  1249. return self.model_executor.remove_lora(lora_id)
  1250. def list_loras(self) -> List[int]:
  1251. return self.model_executor.list_loras()
  1252. def pin_lora(self, lora_id: int) -> bool:
  1253. return self.model_executor.pin_lora(lora_id)
  1254. def add_prompt_adapter(
  1255. self, prompt_adapter_request: PromptAdapterRequest) -> bool:
  1256. return self.model_executor.add_prompt_adapter(prompt_adapter_request)
  1257. def remove_prompt_adapter(self, prompt_adapter_id: int) -> bool:
  1258. return self.model_executor.remove_prompt_adapter(prompt_adapter_id)
  1259. def list_prompt_adapters(self) -> List[int]:
  1260. return self.model_executor.list_prompt_adapters()
  1261. def check_health(self) -> None:
  1262. if self.tokenizer:
  1263. self.tokenizer.check_health()
  1264. self.model_executor.check_health()
  1265. def is_encoder_decoder_model(self):
  1266. return self.model_config.is_encoder_decoder_model
  1267. def is_embedding_model(self):
  1268. return self.model_config.is_embedding_model
  1269. setup_logger()