Skip to content

RequestQueue Classes

langbatch.request_queues.RequestQueue

Bases: ABC

RequestQueue is an abstract class for request queues. Implementations should provide a way to add and retrieve requests.

Used in BatchDispatcher to get requests.

Usage: 

request_queue = InMemoryRequestQueue()
request_queue.add_requests([
    [
        {"role": "user", "content": "How can I learn Python?"}
    ],
    [
        {"role": "user", "content": "Who is the first president of the United States?"},
        {"role": "assistant", "content": "George Washington"},
        {"role": "user", "content": "Second?"}
    ]
])

batch_dispatcher = BatchDispatcher(
    batch_handler=batch_handler,
    queue=request_queue
)

asyncio.create_task(batch_dispatcher.run())

Source code in langbatch\request_queues.py 
 7
 8
 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
class RequestQueue(ABC):
    """
    RequestQueue is an abstract class for request queues.
    Implementations should provide a way to add and retrieve requests.

    Used in `BatchDispatcher` to get requests.

    Usage:
    ```python
    request_queue = InMemoryRequestQueue()
    request_queue.add_requests([
        [
            {"role": "user", "content": "How can I learn Python?"}
        ],
        [
            {"role": "user", "content": "Who is the first president of the United States?"},
            {"role": "assistant", "content": "George Washington"},
            {"role": "user", "content": "Second?"}
        ]
    ])

    batch_dispatcher = BatchDispatcher(
        batch_handler=batch_handler,
        queue=request_queue
    )

    asyncio.create_task(batch_dispatcher.run())
    ```
    """
    @abstractmethod
    def add_requests(self, requests: List[Any]):
        """
        Add requests to the queue
        """
        pass

    @abstractmethod
    def get_requests(self, count: int) -> List[Any]:
        """
        Get requests from the queue
        """
        pass

    @abstractmethod
    def __len__(self):
        pass

add_requests abstractmethod 

add_requests(requests: List[Any])

Add requests to the queue

Source code in langbatch\request_queues.py 
36
37
38
39
40
41
@abstractmethod
def add_requests(self, requests: List[Any]):
    """
    Add requests to the queue
    """
    pass

get_requests abstractmethod 

get_requests(count: int) -> List[Any]

Get requests from the queue

Source code in langbatch\request_queues.py 
43
44
45
46
47
48
@abstractmethod
def get_requests(self, count: int) -> List[Any]:
    """
    Get requests from the queue
    """
    pass

Info

Please make sure to pass the the correct type of requests to the queue as per the type of Batch you are going to use. For Example, if you are using ChatCompletionBatch as your batch type, then you should pass the requests in the same format as the ChatCompletionBatch.create() expects. For EmbeddingBatch, refer EmbeddingBatch.create()

langbatch.request_queues.InMemoryRequestQueue

Bases: RequestQueue

Source code in langbatch\request_queues.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class InMemoryRequestQueue(RequestQueue):
    def __init__(self):
        self.queue = deque()

    def add_requests(self, requests: List[Any]):
        self.queue.extend(requests)
        logging.info(f"Added {len(requests)} requests to queue")

    def get_requests(self, count: int) -> List[Any]:
        if count > len(self.queue):
            count = len(self.queue)
        return [self.queue.popleft() for _ in range(count)]

    def __len__(self):
        return len(self.queue)

add_requests 

add_requests(requests: List[Any])
Source code in langbatch\request_queues.py 
58
59
60
def add_requests(self, requests: List[Any]):
    self.queue.extend(requests)
    logging.info(f"Added {len(requests)} requests to queue")

get_requests 

get_requests(count: int) -> List[Any]
Source code in langbatch\request_queues.py 
62
63
64
65
def get_requests(self, count: int) -> List[Any]:
    if count > len(self.queue):
        count = len(self.queue)
    return [self.queue.popleft() for _ in range(count)]

langbatch.request_queues.RedisRequestQueue

Bases: RequestQueue

RedisRequestQueue is a request queue that uses a Redis list to store requests.

Usage: 

import os
import redis

REDIS_URL = os.environ.get('REDIS_URL')
redis_client = redis.from_url(REDIS_URL)

request_queue = RedisRequestQueue(redis_client, queue_name='turbo_requests')
request_queue.add_requests([
    [
        {"role": "user", "content": "How can I learn Python?"}
    ],
    [
        {"role": "user", "content": "Who is the first president of the United States?"},
        {"role": "assistant", "content": "George Washington"},
        {"role": "user", "content": "Second?"}
    ]
])

Source code in langbatch\request_queues.py 
 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
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class RedisRequestQueue(RequestQueue):
    """
    RedisRequestQueue is a request queue that uses a Redis list to store requests.

    Usage:
    ```python
    import os
    import redis

    REDIS_URL = os.environ.get('REDIS_URL')
    redis_client = redis.from_url(REDIS_URL)

    request_queue = RedisRequestQueue(redis_client, queue_name='turbo_requests')
    request_queue.add_requests([
        [
            {"role": "user", "content": "How can I learn Python?"}
        ],
        [
            {"role": "user", "content": "Who is the first president of the United States?"},
            {"role": "assistant", "content": "George Washington"},
            {"role": "user", "content": "Second?"}
        ]
    ])
    ```
    """
    def __init__(self, redis_client: Any, queue_name: str = 'request_queue'):
        try:
            import redis
            if not isinstance(redis_client, redis.Redis):
                raise TypeError("redis_client must be an instance of redis.Redis")
        except ImportError:
            raise ImportError("redis package is required for RedisRequestQueue. Run: pip install langbatch[redis]")

        self.redis_client = redis_client
        self.queue_name = queue_name

    def add_requests(self, requests: List[Any]):
        count = len(requests)
        for request in requests:
            self.redis_client.rpush(self.queue_name, str(json.dumps(request)))
        logging.debug(f"Added {count} requests to queue.")

    def get_requests(self, count: int) -> List[Any]:
        size = len(self)
        if count > size:
            count = size

        if count == 0:
            return []

        items = self.redis_client.lpop(self.queue_name, count=count)
        if items is None:
            return []

        results = [json.loads(item.decode('utf-8')) for item in items]

        logging.debug(f"Retrieved {len(results)} requests from queue.")
        return results

    def __len__(self):
        length = self.redis_client.llen(self.queue_name)
        return length

__init__ 

__init__(redis_client: Any, queue_name: str = 'request_queue')
Source code in langbatch\request_queues.py 
 95
 96
 97
 98
 99
100
101
102
103
104
def __init__(self, redis_client: Any, queue_name: str = 'request_queue'):
    try:
        import redis
        if not isinstance(redis_client, redis.Redis):
            raise TypeError("redis_client must be an instance of redis.Redis")
    except ImportError:
        raise ImportError("redis package is required for RedisRequestQueue. Run: pip install langbatch[redis]")

    self.redis_client = redis_client
    self.queue_name = queue_name

add_requests 

add_requests(requests: List[Any])
Source code in langbatch\request_queues.py 
106
107
108
109
110
def add_requests(self, requests: List[Any]):
    count = len(requests)
    for request in requests:
        self.redis_client.rpush(self.queue_name, str(json.dumps(request)))
    logging.debug(f"Added {count} requests to queue.")

get_requests 

get_requests(count: int) -> List[Any]
Source code in langbatch\request_queues.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def get_requests(self, count: int) -> List[Any]:
    size = len(self)
    if count > size:
        count = size

    if count == 0:
        return []

    items = self.redis_client.lpop(self.queue_name, count=count)
    if items is None:
        return []

    results = [json.loads(item.decode('utf-8')) for item in items]

    logging.debug(f"Retrieved {len(results)} requests from queue.")
    return results