BrandAnalysis/agent/glm.py

from config import load_config
from utils import image_to_base, pil_image_to_base64, load_image_from_cos
from zhipuai import ZhipuAI
class Glm:
    _instance = None
    def __new__(cls):
        if not cls._instance:
            cls._instance = super(Glm, cls).__new__(cls)
            cls._instance._initialized = False
        return cls._instance
    
    def __init__(self):
        if not self._initialized:
            self.cfg = load_config()['glm']
            self.client = ZhipuAI(api_key=self.cfg["api_key"])
            self._initialized = True
            
        self.model_name = "glm-4-plus"
    
    def text_response(self, query):
        """文字问答"""
        resonse = self.client.chat.completions.create(
            model=self.model_name,
            messages=[
                {"role": "user", "content": query}
            ],
            response_format= {
                'type': 'json_object'
            }
        )
        return resonse.choices[0].message
    
    def images_response(self, query, image_url_list):
        """多图像单轮问答"""
        content = [
            {
                "type": "text",
                "text": query
            }
        ]
        for image_url in image_url_list:
            # image = load_image_from_cos(image_url)
            # image_base = pil_image_to_base64(image)
            content.append(
                {
                    "type": "image_url",
                    "image_url": {"url": image_url}
                }
            )
        response = self.client.chat.completions.create(
            model="glm-4v-plus-0111",
            messages=[
                {
                    "role": "user",
                    "content": content
                }
            ],
            response_format={
                "type": "json_object"
            }
        )
        
        return response.choices[0].message
    
    def multi_images_similarity_judge(self, image_url_list1, image_url_list2, query):
        """多组图像中的产品图像相似度对比"""
        content1 = [
            {
                "text": '这是第一组图像，商品信息中的产品图像',
                "type": "text"
            }
        ]
        
        for image_url in image_url_list1:
            image = load_image_from_cos(image_url)
            image_base = pil_image_to_base64(image)
            content1.append(
                {
                    "type": "image_url",
                    "image_url": {"url": image_base}
                }
            )
        
        content2 = [
            {
                "text": query,
                "type": "text"
            }
        ]
        
        for image_url in image_url_list2:
            image = load_image_from_cos(image_url)
            image_base = pil_image_to_base64(image)
            content2.append(
                {
                    "type": "image_url",
                    "image_url": {"url": image_base}
                }
            )
            
        response = self.client.chat.completions.create(
            model=self.model_name,
            messages=[
                {
                    "content": content1,
                    "role": "user"
                },
                {
                    "content": content2,
                    "role": "user"
                }
            ],
            response_format = {
                "type": "json_object"
            }
        )
        
        return response.choices[0].message
    
    def image_response(self, query, image_url1, image_url2):
        """单图像问答"""
        image1 = load_image_from_cos(image_url1)
        image1_base = pil_image_to_base64(image1)
        
        image2 = load_image_from_cos(image_url2)
        image2_base = pil_image_to_base64(image2)
        response = self.client.chat.completions.create(
            model="glm-4v-plus-0111",
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": query
                        },
                        {
                            "type": "image_url",
                            "image_url": {"url": image1_base}
                        },
                        {
                            "type": "image_url",
                            "image_url": {"url": image2_base}
                        }
                    ]
                }
            ],
            response_format = {
                "type": "json_object"
            }
        )
        
        return response.choices[0].message
    
    def multi_epoch_image_response(self, logo_path, image_url, query):
        """多轮图像问答"""
        image_base = image_to_base(logo_path)
        response = self.client.chat.completions.create(
            model=self.model_name,
            messages=[
                {
                    "content": [
                        {
                            "image_url": {"url": image_base},
                            "type": "image_url"
                        },
                        {
                            "text": "这是李宁的logo",
                            "type": "text"
                        }
                    ],
                    "role": "user"
                },
                {
                    "content": [
                        {
                            "image_url": {"url": image_url},
                            "type": "image_url"
                        },
                        {
                            "text": query,
                            "type": "text"
                        }
                    ],
                    "role": "user"
                }
            ],
            response_format = {
                "type": "json_object"
            }
        )
        
        return response.choices[0].message
        
    
    def set_modelname(self, modelname):
        self.model_name = modelname