Skip to content

vector_memory

VectorMemory

Source code in cat/memory/vector_memory.py 
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
class VectorMemory:
    local_vector_db = None

    def __init__(
        self,
        embedder_name=None,
        embedder_size=None,
    ) -> None:
        # connects to Qdrant and creates self.vector_db attribute
        self.connect_to_vector_memory()

        # Create vector collections
        # - Episodic memory will contain user and eventually cat utterances
        # - Declarative memory will contain uploaded documents' content
        # - Procedural memory will contain tools and knowledge on how to do things
        self.collections = {}
        for collection_name in ["episodic", "declarative", "procedural"]:
            # Instantiate collection
            collection = VectorMemoryCollection(
                client=self.vector_db,
                collection_name=collection_name,
                embedder_name=embedder_name,
                embedder_size=embedder_size,
            )

            # Update dictionary containing all collections
            # Useful for cross-searching and to create/use collections from plugins
            self.collections[collection_name] = collection

            # Have the collection as an instance attribute
            # (i.e. do things like cat.memory.vectors.declarative.something())
            setattr(self, collection_name, collection)

    def connect_to_vector_memory(self) -> None:
        db_path = "cat/data/local_vector_memory/"
        qdrant_host = get_env("CCAT_QDRANT_HOST")

        if not qdrant_host:
            log.info(f"Qdrant path: {db_path}")
            # Qdrant local vector DB client

            # reconnect only if it's the first boot and not a reload
            if VectorMemory.local_vector_db is None:
                VectorMemory.local_vector_db = QdrantClient(
                    path=db_path, force_disable_check_same_thread=True
                )

            self.vector_db = VectorMemory.local_vector_db
        else:
            # Qdrant remote or in other container
            qdrant_port = int(get_env("CCAT_QDRANT_PORT"))
            qdrant_https = is_https(qdrant_host)
            qdrant_host = extract_domain_from_url(qdrant_host)
            qdrant_api_key = get_env("CCAT_QDRANT_API_KEY")

            try:
                s = socket.socket()
                s.connect((qdrant_host, qdrant_port))
            except Exception:
                log.error(f"QDrant does not respond to {qdrant_host}:{qdrant_port}")
                sys.exit()
            finally:
                s.close()

            # Qdrant vector DB client
            self.vector_db = QdrantClient(
                host=qdrant_host,
                port=qdrant_port,
                https=qdrant_https,
                api_key=qdrant_api_key,
            )

    def delete_collection(self, collection_name: str):
        """Delete specific vector collection"""

        return self.vector_db.delete_collection(collection_name)

    def get_collection(self, collection_name: str):
        """Get collection info"""

        return self.vector_db.get_collection(collection_name)

delete_collection(collection_name)

Delete specific vector collection

Source code in cat/memory/vector_memory.py 
86
87
88
89
def delete_collection(self, collection_name: str):
    """Delete specific vector collection"""

    return self.vector_db.delete_collection(collection_name)

get_collection(collection_name)

Get collection info

Source code in cat/memory/vector_memory.py 
91
92
93
94
def get_collection(self, collection_name: str):
    """Get collection info"""

    return self.vector_db.get_collection(collection_name)