File size: 5,432 Bytes
ad022d3
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7d9878f
1
2
3
4
5
6
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
53
54
55
56
57
58
59
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
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
132
133
134
135
"""Base interface for client making requests/call to visual language model provider API"""

from abc import ABC, abstractmethod
from typing import List, Optional, Dict, Union, Iterator
import requests
import json
from utility import isBase64, encode_image, encode_image_from_path_or_url, lvlm_inference

class BaseClient(ABC):
    def __init__(self,

                 hostname: str = "127.0.0.1",

                 port: int = 8090,

                 timeout: int = 60,

                 url: Optional[str] = None):
        self.connection_url = f"http://{hostname}:{port}" if url is None else url
        self.timeout = timeout
        # self.headers = {'Content-Type': 'application/x-www-form-urlencoded'}
        self.headers = {'Content-Type': 'application/json'}

    def root(self):
        """Request for showing welcome message"""
        connection_route = f"{self.connection_url}/"
        return requests.get(connection_route)

    @abstractmethod
    def generate(self, 

                 prompt: str,

                 image: str,

                 **kwargs

        ) -> str:
        """Send request to visual language model API 

        and return generated text that was returned by the visual language model API



        Use this method when you want to call visual language model API to generate text without streaming



        Args:

            prompt: A prompt.

            image: A string that can be either path to image or base64 of an image.

            **kwargs: Arbitrary additional keyword arguments. 

                These are usually passed to the model provider API call as hyperparameter for generation.



        Returns:

            Text returned from visual language model provider API call

        """


    def generate_stream(

            self, 

            prompt: str, 

            image: str, 

            **kwargs

    ) -> Iterator[str]:
        """Send request to visual language model API 

        and return an iterator of streaming text that were returned from the visual language model API call



        Use this method when you want to call visual language model API to stream generated text.



        Args:

            prompt: A prompt.

            image: A string that can be either path to image or base64 of an image.

            **kwargs: Arbitrary additional keyword arguments. 

                These are usually passed to the model provider API call as hyperparameter for generation.



        Returns:

            Iterator of text streamed from visual language model provider API call

        """
        raise NotImplementedError()
    
    def generate_batch(

            self, 

            prompt: List[str], 

            image: List[str], 

            **kwargs

    ) -> List[str]:
        """Send a request to visual language model API for multi-batch generation 

        and return a list of generated text that was returned by the visual language model API



        Use this method when you want to call visual language model API to multi-batch generate text.

        Multi-batch generation does not support streaming.



        Args:

            prompt: List of prompts.

            image: List of strings; each of which can be either path to image or base64 of an image.

            **kwargs: Arbitrary additional keyword arguments. 

                These are usually passed to the model provider API call as hyperparameter for generation.



        Returns:

            List of texts returned from visual language model provider API call

        """
        raise NotImplementedError()
    
class PredictionGuardClient(BaseClient):

    generate_kwargs = ['max_tokens', 
                       'temperature',
                       'top_p', 
                       'top_k']

    def filter_accepted_genkwargs(self, kwargs):
        gen_args = {}
        if "generate_kwargs" in kwargs and isinstance(kwargs["generate_kwargs"], dict):
            gen_args = {k:kwargs["generate_kwargs"][k] 
                        for k in self.generate_kwargs
                        if k in kwargs["generate_kwargs"]}
        return gen_args
            
    def generate(self, 

                 prompt: str,

                 image: str,

                 **kwargs

        ) -> str:
        """Send request to PredictionGuard's API 

        and return generated text that was returned by LLAVA model



        Use this method when you want to call LLAVA model API to generate text without streaming



        Args:

            prompt: A prompt.

            image: A string that can be either path/URL to image or base64 of an image.

            **kwargs: Arbitrary additional keyword arguments. 

                These are usually passed to the model provider API call as hyperparameter for generation.



        Returns:

            Text returned from visual language model provider API call

        """

        assert image is not None and len(image) != "", "the input image cannot be None, it must be either base64-encoded image or path/URL to image"
        if isBase64(image):
            base64_image = image
        else: # this is path to image or URL to image
            base64_image = encode_image_from_path_or_url(image)

        args = self.filter_accepted_genkwargs(kwargs)
        return lvlm_inference(prompt=prompt, image=base64_image, **args)