### Install aiobatch

Source: https://context7.com/hoveychen/aiobatch/llms.txt

Install the library via pip.

```bash
pip install aiobatch
```

--------------------------------

### AsyncBatchScheduler for Asynchronous Batching

Source: https://context7.com/hoveychen/aiobatch/llms.txt

Use AsyncBatchScheduler for fine-grained control over asynchronous batch processing. It requires manual starting and stopping and is suitable for applications needing direct access to the scheduler lifecycle.

```python
from aiobatch import AsyncBatchScheduler
from aiobatch.task import AsyncTask
import torch
import asyncio

async def run_inference():
    # Create scheduler with batch_size=4, 100ms timeout
    scheduler = AsyncBatchScheduler(batch_size=4, timeout=0.1, is_tensor=True)

    def model_inference(x):
        """CPU-bound model inference run in thread pool."""
        return torch.sigmoid(x)

    # Start scheduler with executor function
    await scheduler.start(model_inference)

    async def process_input(input_tensor):
        task = AsyncTask((input_tensor,), {}, is_tensor=True)
        await scheduler.submit(task)
        return await task

    # Submit multiple requests concurrently
    inputs = [torch.randn(1, 32) for _ in range(6)]
    tasks = [process_input(inp) for inp in inputs]
    results = await asyncio.gather(*tasks)

    scheduler.stop()

    for i, result in enumerate(results):
        print(f"Input {i}: output shape {result.shape}")

asyncio.run(run_inference())
```

--------------------------------

### async_queuing - Asynchronous Scheduler Factory

Source: https://context7.com/hoveychen/aiobatch/llms.txt

Creates an asynchronous scheduler for use in async/await contexts, ideal for concurrent applications.

```APIDOC
## @aiobatch.async_queuing

### Description
Creates an asynchronous scheduler for use in async/await contexts. Ideal for web servers and concurrent applications where multiple requests arrive simultaneously.

### Parameters
- **batch_size** (int) - Required - The maximum number of requests to collect before processing.
- **timeout** (float) - Required - The maximum time in seconds to wait before flushing the current batch.
- **is_tensor** (bool) - Optional - Whether to use tensor-based batching.
```

--------------------------------

### queuing - Synchronous Scheduler Factory

Source: https://context7.com/hoveychen/aiobatch/llms.txt

Creates a synchronous scheduler that queues inference requests and processes them in batches using a decorator.

```APIDOC
## @aiobatch.queuing

### Description
Creates a synchronous scheduler that queues inference requests and processes them in batches. Useful for sequential or batched processing of model inference.

### Parameters
- **batch_size** (int) - Required - The maximum number of requests to collect before processing.
- **timeout** (float) - Required - The maximum time in seconds to wait before flushing the current batch.
- **is_tensor** (bool) - Optional - Whether to use tensor-based batching (concatenating PyTorch/NumPy arrays).
```

--------------------------------

### Asynchronous Scheduler with async_queuing

Source: https://context7.com/hoveychen/aiobatch/llms.txt

Use the async_queuing decorator for async/await contexts, suitable for concurrent web server environments.

```python
import aiobatch
import torch
import asyncio

@aiobatch.async_queuing(batch_size=8, timeout=0.05, is_tensor=True)
def model_forward(embeddings):
    """Async-compatible model inference."""
    # Your PyTorch model inference here
    return torch.softmax(embeddings, dim=-1)

async def handle_request(request_id, data):
    """Simulate handling a web request."""
    result = await model_forward(data)
    return {"id": request_id, "prediction": result}

async def main():
    # Simulate multiple concurrent requests
    requests = [
        torch.randn(1, 10) for _ in range(10)
    ]

    # All requests are batched automatically
    tasks = [handle_request(i, req) for i, req in enumerate(requests)]
    results = await asyncio.gather(*tasks)

    for r in results:
        print(f"Request {r['id']}: prediction shape {r['prediction'].shape}")

asyncio.run(main())
```

--------------------------------

### Synchronous Scheduler with queuing

Source: https://context7.com/hoveychen/aiobatch/llms.txt

Use the queuing decorator to batch inference requests synchronously. Set is_tensor=True for PyTorch/NumPy batching.

```python
import aiobatch
import torch
import numpy as np

# Create a batch scheduler with batch_size=4 and 100ms timeout
@aiobatch.queuing(batch_size=4, timeout=0.1, is_tensor=True)
def predict(input_tensor):
    """Model inference function that processes batched tensors."""
    # Simulated model inference - in practice, this would be your PyTorch/ONNX model
    return input_tensor * 2

# Each call submits a task; tasks are automatically batched
result1 = predict(torch.randn(1, 128))  # Single sample
result2 = predict(torch.randn(2, 128))  # Two samples batched together

print(f"Result 1 shape: {result1.shape}")  # torch.Size([1, 128])
print(f"Result 2 shape: {result2.shape}")  # torch.Size([2, 128])
```

--------------------------------

### Manual BatchScheduler Control

Source: https://context7.com/hoveychen/aiobatch/llms.txt

Use the BatchScheduler class for fine-grained control over the scheduler lifecycle and manual task submission.

```python
from aiobatch import BatchScheduler
import torch
import threading

# Initialize scheduler with batch_size=4, 50ms timeout, tensor mode
scheduler = BatchScheduler(batch_size=4, timeout=0.05, is_tensor=True)

def inference_fn(input_data):
    """The actual model inference function."""
    # Simulate model processing
    return input_data.mean(dim=-1, keepdim=True)

# Start the scheduler with the inference function
scheduler.start(inference_fn)

def submit_request(data):
    """Submit inference request from any thread."""
    from aiobatch.task import Task
    task = Task((data,), {}, is_tensor=True)
    scheduler.submit(task)
    return task.result()  # Blocks until batch is processed

# Submit from multiple threads
results = []
threads = []
for i in range(8):
    t = threading.Thread(
        target=lambda: results.append(submit_request(torch.randn(1, 64)))
    )
    threads.append(t)
    t.start()

for t in threads:
    t.join()

# Clean shutdown
scheduler.stop()
print(f"Processed {len(results)} requests")
```

--------------------------------

### Task - Synchronous Task Wrapper

Source: https://context7.com/hoveychen/aiobatch/llms.txt

Use the Task class to wrap synchronous inference requests. It provides a Future-like interface for managing arguments and retrieving results.

```python
from aiobatch.task import Task
import torch

# Create task with positional and keyword arguments
input_tensor = torch.randn(2, 128)
task = Task(
    args=(input_tensor,),
    kwargs={"threshold": 0.5},
    is_tensor=True
)

# Access task properties
print(f"Sample size: {task.sample_size}")  # 2 (batch dimension of tensor)
print(f"Tensor arg indices: {task.tensor_args_idx}")  # [0]
print(f"Args: {task.args[0].shape}")  # torch.Size([2, 128])

# Set result (normally done by scheduler)
result = torch.softmax(input_tensor, dim=-1)
task.set_result(result)

# Retrieve result (blocks if not yet set)
output = task.result()
print(f"Output shape: {output.shape}")  # torch.Size([2, 128])
```

--------------------------------

### BatchScheduler - Synchronous Batch Scheduler Class

Source: https://context7.com/hoveychen/aiobatch/llms.txt

Low-level synchronous batch scheduler with manual lifecycle control for fine-grained management.

```APIDOC
## BatchScheduler

### Description
Low-level synchronous batch scheduler class providing manual control over starting, stopping, and task submission.

### Methods
- **start(inference_fn)** - Starts the scheduler with the provided inference function.
- **submit(task)** - Submits an inference task to the scheduler.
- **stop()** - Performs a clean shutdown of the scheduler.
```

--------------------------------

### AsyncTask - Asynchronous Task Wrapper

Source: https://context7.com/hoveychen/aiobatch/llms.txt

The AsyncTask class wraps asynchronous inference requests, supporting await syntax for non-blocking result retrieval. It's suitable for asyncio-based applications.

```python
from aiobatch.task import AsyncTask
import torch
import asyncio

async def demo_async_task():
    # Create async task
    input_data = torch.randn(3, 64)
    task = AsyncTask(
        args=(input_data,),
        kwargs={},
        is_tensor=True
    )

    print(f"Sample size: {task.sample_size}")  # 3
    print(f"Tensor indices: {task.tensor_args_idx}")  # [0]

    # Simulate async result setting (normally done by scheduler)
    async def set_result_later():
        await asyncio.sleep(0.01)
        task.set_result(input_data.mean(dim=-1))

    asyncio.create_task(set_result_later())

    # Await the result
    result = await task
    print(f"Result shape: {result.shape}")  # torch.Size([3])

asyncio.run(demo_async_task())
```

--------------------------------

### Unbatching Results

Source: https://context7.com/hoveychen/aiobatch/llms.txt

This snippet shows how to unbatch results from a batch inference operation back into individual task results.

```python
batcher.unbatch(batch_results, tasks)
print(f"Task results: {task1.result()}, {task2.result()}, {task3.result()}")
```

--------------------------------

### AsyncSingleScheduler for Async Sequential Execution

Source: https://context7.com/hoveychen/aiobatch/llms.txt

Utilize AsyncSingleScheduler to serialize model access within asynchronous applications. Submissions are processed one after another, even when submitted concurrently.

```python
from aiobatch import AsyncSingleScheduler
from aiobatch.task import AsyncTask
import torch
import asyncio

async def main():
    scheduler = AsyncSingleScheduler()

    def inference(x):
        return x.sum()

    await scheduler.start(inference)

    async def submit_task(value):
        task = AsyncTask((torch.tensor([value]),), {})
        await scheduler.submit(task)
        return await task

    # Requests processed sequentially despite concurrent submission
    results = await asyncio.gather(*[submit_task(i) for i in range(5)])
    print(f"Results: {results}")  # [tensor(0), tensor(1), tensor(2), ...]

    scheduler.stop()

asyncio.run(main())
```

--------------------------------

### TensorBatcher for PyTorch/NumPy Batching

Source: https://context7.com/hoveychen/aiobatch/llms.txt

TensorBatcher is an internal utility for concatenating and splitting PyTorch tensors or NumPy arrays. It supports fixed batch sizes and zero-padding, useful for models requiring consistent input dimensions.

```python
from aiobatch.batcher import TensorBatcher
from aiobatch.task import Task
import torch

# Create batcher with fixed batch size (for models requiring fixed dimensions)
batcher = TensorBatcher(fixed_batch_size=4)

# Simulate tasks with tensor arguments
task1 = Task((torch.randn(1, 64),), {}, is_tensor=True)
task2 = Task((torch.randn(2, 64),), {}, is_tensor=True)
tasks = [task1, task2]

# Batch tensors together (concatenates along dim 0, pads to fixed size)
batched_args, batched_kwargs = batcher.batch(tasks)
print(f"Batched shape: {batched_args[0].shape}")  # torch.Size([4, 64]) - padded

# Simulate model output
model_output = batched_args[0] * 2

# Unbatch results back to individual tasks
batcher.unbatch(model_output, tasks)

# Each task now has its result
result1 = task1.result()
result2 = task2.result()
print(f"Task 1 result shape: {result1.shape}")  # torch.Size([1, 64])
print(f"Task 2 result shape: {result2.shape}")  # torch.Size([2, 64])
```

--------------------------------

### SingleScheduler for Sequential Request Processing

Source: https://context7.com/hoveychen/aiobatch/llms.txt

Employ SingleScheduler for synchronous, sequential processing of requests without batching. It's ideal for serializing access to non-thread-safe models.

```python
from aiobatch import SingleScheduler
import torch
import threading

scheduler = SingleScheduler()

def model_predict(x, temperature=1.0):
    """Non-thread-safe model inference."""
    return torch.softmax(x / temperature, dim=-1)

scheduler.start(model_predict)

def worker(worker_id):
    from aiobatch.task import Task
    data = torch.randn(1, 10)
    task = Task((data,), {"temperature": 0.5})
    scheduler.submit(task)
    result = task.result()
    print(f"Worker {worker_id}: {result.shape}")

# Safe concurrent access - requests are serialized
threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

scheduler.stop()
```

--------------------------------

### ListBatcher for Non-Tensor Argument Batching

Source: https://context7.com/hoveychen/aiobatch/llms.txt

ListBatcher is an internal utility for batching non-tensor arguments as lists. It is suitable for data types like strings or other objects that should not be concatenated, preserving their individual nature within batches.

```python
from aiobatch.batcher import ListBatcher
from aiobatch.task import Task

batcher = ListBatcher()

# Tasks with non-tensor arguments
task1 = Task(("hello",), {"option": True})
task2 = Task(("world",), {"option": True})
task3 = Task(("test",), {"option": True})
tasks = [task1, task2, task3]

# Batch as lists
batched_args, batched_kwargs = batcher.batch(tasks)
print(f"Batched args: {batched_args}")  # [['hello', 'world', 'test']]
print(f"Batched kwargs: {batched_kwargs}")  # {'option': [True, True, True]}

# Simulate batch processing returning per-item results
batch_results = ["HELLO", "WORLD", "TEST"]
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.