How to Build a High-Concurrency Module for Local Models?

Request1 \
Request2  \
Request3   →  Queue → Batch → one model.encode()
Request4  /
Request5 /

├── inference_engine/          High-concurrency batch request processing module
│   ├── queue_manager.py       # Queue management
│   ├── batch_scheduler.py     # Batch scheduling
│   ├── worker.py              # Worker executor
│   ├── task.py                # Request encapsulation
│   └── config.py              # Configuration (batch size, etc.)

class InferenceTask:
    def __init__(self, data, future):
        self.data = data
        self.future = future "future": ...
}

import asyncio

class QueueManager:
    def __init__(self):
        self.queue = asyncio.Queue()

    async def put(self, task):
        await self.queue.put(task)

    async def get_batch(self, max_batch_size, timeout=0.01):
        .....
        return batch

import asyncio

class BatchScheduler:
    def __init__(self, queue_manager, process_fn, batch_size=32):
        self.queue_manager = queue_manager
        self.process_fn = process_fn
        self.batch_size = batch_size

    async def start(self):
        while True:
            batch = await self.queue_manager.get_batch(self.batch_size)

            if not batch:
                continue

            inputs = [task.data for task in batch]

            # 👉 Core: unified processing
            results = self.process_fn(inputs)

            for task, result in zip(batch, results):
                task.future.set_result(result)

import asyncio

class InferenceWorker:
    def __init__(self, scheduler):
        self.scheduler = scheduler

    def start(self):
        asyncio.create_task(self.scheduler.start())

BATCH_SIZE = 32
TIMEOUT = 0.01