From 1902cd2d0323749819c3164d1332f0bb8e654812 Mon Sep 17 00:00:00 2001 From: Aditya Gupta Date: Fri, 3 Jul 2026 13:07:41 +0530 Subject: [PATCH] refactor: Concurrent batch processing --- app/worker.py | 32 ++++++++++++++++++++++++++------ docs/assets/overview_diagram.svg | 0 2 files changed, 26 insertions(+), 6 deletions(-) create mode 100644 docs/assets/overview_diagram.svg diff --git a/app/worker.py b/app/worker.py index 79a7794..94c6b3a 100644 --- a/app/worker.py +++ b/app/worker.py @@ -362,14 +362,34 @@ async def _process_job_async(job_id: str, transactions: List[Dict[str, Any]]): ) try: client = genai.Client(api_key=api_key) - # Batch transactions in chunks of 50 + # Split into batches of 50 and fire all concurrently batch_size = 50 - for i in range(0, len(uncategorized_txns), batch_size): - batch = uncategorized_txns[i : i + batch_size] - batch_results = await classify_transactions_batch( - client, batch, allowed_categories + batches = [ + uncategorized_txns[i : i + batch_size] + for i in range(0, len(uncategorized_txns), batch_size) + ] + logger.info( + f"Dispatching {len(batches)} classification batch(es) concurrently..." + ) + with benchmark( + "LLM Concurrent Batch Classification (all batches)" + ): + batch_results_list = await asyncio.gather( + *[ + classify_transactions_batch( + client, batch, allowed_categories + ) + for batch in batches + ], + return_exceptions=True, ) - llm_results.update(batch_results) + for batch_result in batch_results_list: + if isinstance(batch_result, Exception): + logger.error( + f"A classification batch failed: {batch_result}" + ) + else: + llm_results.update(batch_result) except Exception as e: logger.error( f"Failed to initialize GenAI Client or classify: {e}" diff --git a/docs/assets/overview_diagram.svg b/docs/assets/overview_diagram.svg new file mode 100644 index 0000000..e69de29