File size: 2,664 Bytes
20555bc
 
 
 
 
2b38366
 
 
20555bc
c24b9a6
 
28af5bc
20555bc
 
156199c
 
20555bc
 
c24b9a6
28af5bc
20555bc
 
 
c24b9a6
20555bc
 
156199c
28af5bc
20555bc
c24b9a6
 
 
 
 
 
 
 
 
 
 
 
2b38366
 
 
c24b9a6
20555bc
 
 
 
28af5bc
156199c
20555bc
 
28af5bc
20555bc
c24b9a6
20555bc
 
28af5bc
20555bc
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
from langchain.embeddings.openai import OpenAIEmbeddings

from ..common.utils import OPENAI_API_KEY, PINECONE_NAMESPACE, PINECONE_INDEX_NAME
from .pinecone_engine import (
    init_pinecone,
    get_pinecone_index_namespace,
    delete_pinecone,
    update_pinecone,
)
from ..model.basic_model import DataStatus
from ..model.image_model import ImageModel
from ..model.req_model import ReqModel


def get_embeddings(setting: ReqModel):
    return OpenAIEmbeddings(openai_api_key=setting.openai_key)


# def embed_image_text(image_text: str, image_name: str, uuid: str) -> str:
def embed_image_text(image: ImageModel, setting: ReqModel) -> str:
    prompt_template = f"""
        This is the text about the image.
        ###
        {image.image_text}
        """

    embed_image = get_embeddings(setting=setting).embed_query(prompt_template)
    index = init_pinecone(index_name=PINECONE_INDEX_NAME, setting=setting)

    """create | update | delete in pinecone"""
    pinecone_namespace = get_pinecone_index_namespace(image.uuid)
    try:
        if image.status == DataStatus.CREATED:
            """add a data in pinecone"""
            upsert_response = index.upsert(
                vectors=[{"id": image.image_name, "values": embed_image}],
                namespace=pinecone_namespace,
            )
        elif image.status == DataStatus.DELETED:
            delete_pinecone(namespace=pinecone_namespace, key=image.image_name)
        elif image.status == DataStatus.UPDATED:
            update_pinecone(
                namespace=pinecone_namespace, key=image.image_name, value=embed_image
            )
    except Exception as e:
        return "fail to embed image text"
    return "success to embed image text"


def query_image_text(image_content, message, setting: ReqModel):
    embed_image = get_embeddings(setting=setting).embed_query(
        get_prompt_image_with_message(image_content, message)
    )
    index = init_pinecone(index_name=PINECONE_INDEX_NAME, setting=setting)
    relatedness_data = index.query(
        vector=embed_image,
        top_k=3,
        include_values=False,
        namespace=get_pinecone_index_namespace(setting.uuid),
    )
    if len(relatedness_data["matches"]) > 0:
        return relatedness_data["matches"][0]["id"]
    return ""


def get_prompt_image_with_message(image_content, message):
    prompt_template = f"""
                This is the text about the image.
                ###
                {image_content}
                ###
                This message is the detailed description of the image.
                ### 
                {message}
                """

    return prompt_template