| 418 | |
| 419 | |
| 420 | async def run_batch( |
| 421 | args: argparse.Namespace, |
| 422 | ) -> None: |
| 423 | |
| 424 | # Setup engine and handlers |
| 425 | engine_client, chat_handler = await setup_engine_and_handlers(args) |
| 426 | |
| 427 | concurrency = getattr(args, "max_concurrency", 512) |
| 428 | workers = getattr(args, "workers", 1) |
| 429 | max_concurrency = (concurrency + workers - 1) // workers |
| 430 | semaphore = asyncio.Semaphore(max_concurrency) |
| 431 | |
| 432 | console_logger.info(f"concurrency: {concurrency}, workers: {workers}, max_concurrency: {max_concurrency}") |
| 433 | |
| 434 | tracker = BatchProgressTracker() |
| 435 | console_logger.info("Reading batch from %s...", args.input_file) |
| 436 | |
| 437 | # Submit all requests in the file to the engine "concurrently". |
| 438 | response_futures: list[Awaitable[BatchRequestOutput]] = [] |
| 439 | for request_json in (await read_file(args.input_file)).strip().split("\n"): |
| 440 | # Skip empty lines. |
| 441 | request_json = request_json.strip() |
| 442 | if not request_json: |
| 443 | continue |
| 444 | |
| 445 | request = BatchRequestInput.model_validate_json(request_json) |
| 446 | |
| 447 | # Determine the type of request and run it. |
| 448 | if request.url == "/v1/chat/completions": |
| 449 | chat_handler_fn = chat_handler.create_chat_completion if chat_handler is not None else None |
| 450 | if chat_handler_fn is None: |
| 451 | response_futures.append( |
| 452 | make_async_error_request_output( |
| 453 | request, |
| 454 | error_msg="The model does not support Chat Completions API", |
| 455 | ) |
| 456 | ) |
| 457 | continue |
| 458 | |
| 459 | response_futures.append(run_request(chat_handler_fn, request, tracker, semaphore)) |
| 460 | tracker.submitted() |
| 461 | else: |
| 462 | response_futures.append( |
| 463 | make_async_error_request_output( |
| 464 | request, |
| 465 | error_msg=f"URL {request.url} was used. " |
| 466 | "Supported endpoints: /v1/chat/completions" |
| 467 | "See fastdeploy/entrypoints/openai/api_server.py for supported " |
| 468 | "/v1/chat/completions versions.", |
| 469 | ) |
| 470 | ) |
| 471 | |
| 472 | with tracker.pbar(): |
| 473 | responses = await asyncio.gather(*response_futures) |
| 474 | |
| 475 | success_count = sum(1 for r in responses if r.error is None) |
| 476 | error_count = len(responses) - success_count |
| 477 | console_logger.info(f"Batch processing completed: {success_count} success, {error_count} errors") |