Skip to content

BatchQueue Classes

langbatch.batch_queues.BatchQueue

Bases: ABC

Abstract class for batch queue. Implementations should provide a way to save and load batch queues.

Used in BatchHandler to save and load the batch queues.

Usage: 

import asyncio

# Using default FileBatchQueue
file_batch_queue = FileBatchQueue("batch_queue.json")
batch_handler = BatchHandler(
    batch_process_func=process_batch,
    batch_type=OpenAIChatCompletionBatch,
    batch_queue=file_batch_queue
)
asyncio.create_task(batch_handler.run())

# With custom batch queue
class MyCustomBatchQueue(BatchQueue):
    def save(self, queue: Dict[str, List[str]]):
        # Custom save logic

    def load(self) -> Dict[str, List[str]]:
        # Custom load logic

custom_batch_queue = MyCustomBatchQueue()
batch_handler = BatchHandler(
    batch_process_func=process_batch,
    batch_type=OpenAIChatCompletionBatch,
    batch_queue=custom_batch_queue
)
asyncio.create_task(batch_handler.run())

Source code in langbatch\batch_queues.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class BatchQueue(ABC):
    """
    Abstract class for batch queue.
    Implementations should provide a way to save and load batch queues.

    Used in BatchHandler to save and load the batch queues.

    Usage:
    ```python
    import asyncio

    # Using default FileBatchQueue
    file_batch_queue = FileBatchQueue("batch_queue.json")
    batch_handler = BatchHandler(
        batch_process_func=process_batch,
        batch_type=OpenAIChatCompletionBatch,
        batch_queue=file_batch_queue
    )
    asyncio.create_task(batch_handler.run())

    # With custom batch queue
    class MyCustomBatchQueue(BatchQueue):
        def save(self, queue: Dict[str, List[str]]):
            # Custom save logic

        def load(self) -> Dict[str, List[str]]:
            # Custom load logic

    custom_batch_queue = MyCustomBatchQueue()
    batch_handler = BatchHandler(
        batch_process_func=process_batch,
        batch_type=OpenAIChatCompletionBatch,
        batch_queue=custom_batch_queue
    )
    asyncio.create_task(batch_handler.run())
    ```
    """
    @abstractmethod
    def save(self, queue: Dict[str, List[str]]):
        """
        Save the batch queue.
        """
        pass

    @abstractmethod
    def load(self) -> Dict[str, List[str]]:
        """
        Load the batch queue.
        """
        pass

save abstractmethod 

save(queue: Dict[str, List[str]])

Save the batch queue.

Source code in langbatch\batch_queues.py 
46
47
48
49
50
51
@abstractmethod
def save(self, queue: Dict[str, List[str]]):
    """
    Save the batch queue.
    """
    pass

load abstractmethod 

load() -> Dict[str, List[str]]

Load the batch queue.

Source code in langbatch\batch_queues.py 
53
54
55
56
57
58
@abstractmethod
def load(self) -> Dict[str, List[str]]:
    """
    Load the batch queue.
    """
    pass

langbatch.batch_queues.FileBatchQueue

Bases: BatchQueue

Batch queue that saves the queue to a file.

Usage: 

queue = FileBatchQueue("batch_queue.json")

batch_handler = BatchHandler(
    batch_process_func=process_batch,
    batch_type=OpenAIChatCompletionBatch,
    batch_queue=queue
)

asyncio.create_task(batch_handler.run())

Source code in langbatch\batch_queues.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class FileBatchQueue(BatchQueue):
    """
    Batch queue that saves the queue to a file.

    Usage:
    ```python
    queue = FileBatchQueue("batch_queue.json")

    batch_handler = BatchHandler(
        batch_process_func=process_batch,
        batch_type=OpenAIChatCompletionBatch,
        batch_queue=queue
    )

    asyncio.create_task(batch_handler.run())
    ```
    """
    def __init__(self, file_path: str):
        self.file_path = file_path

    def save(self, queue: Dict[str, List[str]]):
        try:
            with open(self.file_path, 'w') as f:
                json.dump(queue, f)
        except IOError as e:
            logger.error(f"Error saving queue to file: {e}")
            raise

    def load(self) -> Dict[str, List[str]]:
        try:
            if os.path.exists(self.file_path):
                with open(self.file_path, 'r') as f:
                    return json.load(f)
            return {"pending": [], "processing": []}
        except IOError as e:
            logger.error(f"Error loading queue from file: {e}")
            raise

file_path instance-attribute 

file_path = file_path

__init__ 

__init__(file_path: str)
Source code in langbatch\batch_queues.py 
77
78
def __init__(self, file_path: str):
    self.file_path = file_path

save 

save(queue: Dict[str, List[str]])
Source code in langbatch\batch_queues.py 
80
81
82
83
84
85
86
def save(self, queue: Dict[str, List[str]]):
    try:
        with open(self.file_path, 'w') as f:
            json.dump(queue, f)
    except IOError as e:
        logger.error(f"Error saving queue to file: {e}")
        raise

load 

load() -> Dict[str, List[str]]
Source code in langbatch\batch_queues.py 
88
89
90
91
92
93
94
95
96
def load(self) -> Dict[str, List[str]]:
    try:
        if os.path.exists(self.file_path):
            with open(self.file_path, 'r') as f:
                return json.load(f)
        return {"pending": [], "processing": []}
    except IOError as e:
        logger.error(f"Error loading queue from file: {e}")
        raise