Add a new request to the queue. Args: task: Request A dictionary representing the request. sampling_params: A dictionary representing the sampling parameters. Returns: None
(self, task)
| 280 | return request["prompt_token_ids"] |
| 281 | |
| 282 | async def add_requests(self, task): |
| 283 | """ |
| 284 | Add a new request to the queue. |
| 285 | |
| 286 | Args: |
| 287 | task: Request A dictionary representing the request. |
| 288 | sampling_params: A dictionary representing the sampling parameters. |
| 289 | |
| 290 | Returns: |
| 291 | None |
| 292 | """ |
| 293 | |
| 294 | task["metrics"]["preprocess_start_time"] = time.time() |
| 295 | request_id = task.get("request_id").split("_")[0] |
| 296 | tracing.trace_slice_start(tracing.TraceSpanName.PREPROCESSING, request_id) |
| 297 | trace_print(LoggingEventName.PREPROCESSING_START, task["request_id"], task.get("user", "")) |
| 298 | try: |
| 299 | chat_template_kwargs = task.get("chat_template_kwargs") or {} |
| 300 | chat_template_kwargs.update({"chat_template": task.get("chat_template")}) |
| 301 | reasoning_effort = task.get("reasoning_effort") |
| 302 | if reasoning_effort is not None: |
| 303 | chat_template_kwargs["reasoning_effort"] = reasoning_effort |
| 304 | task["chat_template_kwargs"] = chat_template_kwargs |
| 305 | self.process_messages(task.get("messages", [])) |
| 306 | if inspect.iscoroutinefunction(self.data_processor.process_request_dict): |
| 307 | await self.data_processor.process_request_dict(task, self.max_model_len) |
| 308 | else: |
| 309 | self.data_processor.process_request_dict(task, self.max_model_len) |
| 310 | |
| 311 | task["prompt_token_ids_len"] = len(task["prompt_token_ids"]) |
| 312 | input_ids_len = task["prompt_token_ids_len"] |
| 313 | task["need_prefill_tokens"] = task["prompt_token_ids_len"] |
| 314 | |
| 315 | task["max_tokens"] = min(self.max_model_len - input_ids_len, task.get("max_tokens")) |
| 316 | min_tokens = task.get("min_tokens", 1) |
| 317 | if "messages" in task: |
| 318 | task["messages"] = None |
| 319 | api_server_logger.info(f"task['max_tokens']:{task['max_tokens']}") |
| 320 | main_process_metrics.request_params_max_tokens.observe(task["max_tokens"]) |
| 321 | main_process_metrics.prompt_tokens_total.inc(input_ids_len) |
| 322 | main_process_metrics.request_prompt_tokens.observe(input_ids_len) |
| 323 | except Exception as e: |
| 324 | api_server_logger.error(f"add_requests error: {e}, {str(traceback.format_exc())}") |
| 325 | raise EngineError(str(e), error_code=400) |
| 326 | |
| 327 | if input_ids_len + min_tokens >= self.max_model_len: |
| 328 | error_msg = ( |
| 329 | f"Input text is too long, input_ids_len ({input_ids_len}) " |
| 330 | f"+ min_tokens({min_tokens}) >= max_model_len({self.max_model_len})" |
| 331 | ) |
| 332 | api_server_logger.error(error_msg) |
| 333 | raise EngineError(error_msg, error_code=400) |
| 334 | |
| 335 | if input_ids_len > self.max_model_len: |
| 336 | error_msg = ( |
| 337 | f"Length of input token({input_ids_len}) exceeds the limit max_model_len({self.max_model_len})." |
| 338 | ) |
| 339 | api_server_logger.error(error_msg) |
no test coverage detected