opendan/rootfs/knowledge_pipelines/Mia/query.py

import os
import logging
import json
from aios import *

class EmbeddingEnvironment(SimpleEnvironment):
    def __init__(self, workspace: str) -> None:
        super().__init__(workspace)
        self.path = os.path.join(AIStorage.get_instance().get_myai_dir(), "knowledge/indices/embedding")
        self._default_text_model = "all-MiniLM-L6-v2"
        self._default_image_model = "clip-ViT-B-32"

        query_param = {
            "tokens": "key words to query", 
            "types": "prefered knowledge types, one or more of [text, image]",
            "index": "index of query result"
        }
        self.add_ai_function(SimpleAIFunction("query_knowledge", 
                                            "vector query content from local knowledge base",
                                            self._query, 
                                            query_param))
        
    def __get_vector_store(self, model_name: str) -> ChromaVectorStore:
        return ChromaVectorStore(self.path, model_name)
    
    async def query_objects(self, tokens: str, types: list[str], topk: int) -> [ObjectID]:
        texts = []
        if "text" in types:
            vector = await ComputeKernel.get_instance().do_text_embedding(tokens, self._default_text_model)
            texts = await self.__get_vector_store(self._default_text_model).query(vector, topk)
        images = []
        if "image" in types:
            vector = await ComputeKernel.get_instance().do_text_embedding(tokens, self._default_image_model)
            images = await self.__get_vector_store(self._default_image_model).query(vector, topk)
        return texts + images
    
        
    def tokens_from_objects(self, object_ids: [ObjectID]) -> list[str]:
        results = dict()
        for object_id in object_ids:
            parents = KnowledgeStore().get_relation_store().get_related_root_objects(object_id)
            # last parent is the root object
            root_object_id = parents[0] if parents else object_id
            logging.info(f"object_id: {str(object_id)} root_object_id: {str(root_object_id)}")
            if str(root_object_id) in results:
                results[str(root_object_id)].append(object_id)
            else:
                results[str(root_object_id)] = [root_object_id, object_id]
        content = ""
        result_desc = []
        for result in results.values():
            # first element in result is the root object
            root_object_id = result[0]
            if root_object_id.get_object_type() == ObjectType.Email:
                email = KnowledgeStore().load_object(root_object_id)
                desc = email.get_desc()
                desc["type"] = "email"
                desc["contents"] = []
                result_desc.append(desc)
                upper_list = desc["contents"]
                result = result[1:]
            else:
                upper_list = result_desc
            
            for object_id in result:
                if object_id.get_object_type() == ObjectType.Chunk:
                    upper_list.append({"type": "text", "content": KnowledgeStore().get_chunk_reader().get_chunk(object_id).read().decode("utf-8")})
                if object_id.get_object_type() == ObjectType.Image:
                    # image = self.load_object(object_id)
                    desc = dict()
                    desc["id"] = str(object_id)
                    desc["type"] = "image"
                    upper_list.append(desc)
                if object_id.get_object_type() == ObjectType.Video:
                    video = KnowledgeStore().load_object(object_id)
                    desc = video.get_desc()
                    desc["type"] = "video"
                    upper_list.append(desc)
                else:
                    pass
        content += json.dumps(result_desc)
        content += ".\n"  

        return content
    
    async def _query(self, tokens: str, types: list[str] = ["text"], index: str=0):
        index = int(index)
        object_ids = await self.query_objects(tokens, types, 4)
        if len(object_ids) <= index:
            return "*** I have no more information for your reference.\n"
        else:
            content = "*** I have provided the following known information for your reference with json format:\n"
            return content + self.tokens_from_objects(object_ids[index:index+1])

def init(workspace: str) -> EmbeddingEnvironment:
    return EmbeddingEnvironment(workspace)