Skip to content

RAG & Embeddings Reference

This section documents the Retrieval-Augmented Generation (RAG) pipeline, document chunk representation, vector store backends, and embedding generators.

DocumentChunk

edgevoiceai.rag.chunk

Document chunk representation for Voice AI RAG pipeline.

DocumentChunk dataclass

Standardized knowledge chunk retrieved from any vector database.

Source code in edgevoiceai/rag/chunk.py 
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
@dataclass(slots=True)
class DocumentChunk:
    """
    Standardized knowledge chunk retrieved from any vector database.
    """
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    title: str = ""
    text: str = ""
    score: float = 0.0
    metadata: dict[str, Any] = field(default_factory=dict)
    audio_url: str | None = None
    vector_store: str = "Unknown"

    def to_dict(self) -> dict[str, Any]:
        """Converts chunk to dictionary representation."""
        return {
            "id": self.id,
            "title": self.title,
            "text": self.text,
            "score": self.score,
            "metadata": self.metadata,
            "audio_url": self.audio_url,
            "vector_store": self.vector_store,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> DocumentChunk:
        """Constructs a DocumentChunk from dictionary data."""
        return cls(
            id=str(data.get("id", str(uuid.uuid4()))),
            title=str(data.get("title", "")),
            text=str(data.get("text", "")),
            score=float(data.get("score", 0.0)),
            metadata=dict(data.get("metadata", {})),
            audio_url=data.get("audio_url"),
            vector_store=str(data.get("vector_store", "Unknown")),
        )

from_dict classmethod 

from_dict(data)

Constructs a DocumentChunk from dictionary data.

Source code in edgevoiceai/rag/chunk.py 
37
38
39
40
41
42
43
44
45
46
47
48
@classmethod
def from_dict(cls, data: dict[str, Any]) -> DocumentChunk:
    """Constructs a DocumentChunk from dictionary data."""
    return cls(
        id=str(data.get("id", str(uuid.uuid4()))),
        title=str(data.get("title", "")),
        text=str(data.get("text", "")),
        score=float(data.get("score", 0.0)),
        metadata=dict(data.get("metadata", {})),
        audio_url=data.get("audio_url"),
        vector_store=str(data.get("vector_store", "Unknown")),
    )

to_dict 

to_dict()

Converts chunk to dictionary representation.

Source code in edgevoiceai/rag/chunk.py 
25
26
27
28
29
30
31
32
33
34
35
def to_dict(self) -> dict[str, Any]:
    """Converts chunk to dictionary representation."""
    return {
        "id": self.id,
        "title": self.title,
        "text": self.text,
        "score": self.score,
        "metadata": self.metadata,
        "audio_url": self.audio_url,
        "vector_store": self.vector_store,
    }

RAGRetriever

edgevoiceai.rag.retriever

RAG Retriever for Voice AI Framework. Orchestrates dense embedding generation, vector store lookup, and automated fallback to local/in-memory knowledge when networks are offline.

RAGRetriever

Modular RAG retrieval engine. Connects any embedding model with primary and fallback vector stores.

Source code in edgevoiceai/rag/retriever.py 
 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
136
137
138
class RAGRetriever:
    """
    Modular RAG retrieval engine.
    Connects any embedding model with primary and fallback vector stores.
    """

    def __init__(
        self,
        embedding_model: BaseEmbeddingModel,
        vector_store: BaseVectorStore,
        fallback_store: BaseVectorStore | None = None,
        default_limit: int = 3,
        default_score_threshold: float = 0.45,
        query_normalizer: Callable[[str], str] | None = None,
    ) -> None:
        self.embedding_model = embedding_model
        self.vector_store = vector_store
        self.fallback_store = fallback_store
        self.default_limit = default_limit
        self.default_score_threshold = default_score_threshold
        self.query_normalizer = query_normalizer

    async def retrieve(
        self,
        query: str,
        limit: int | None = None,
        score_threshold: float | None = None,
        filter_metadata: dict[str, Any] | None = None,
    ) -> list[DocumentChunk]:
        """
        Retrieves relevant document chunks for a natural language voice query.
        Falls back to fallback_store if primary store is unreachable or yields no matches.
        """
        clean_query = (query or "").strip()
        if not clean_query:
            return []

        if self.query_normalizer is not None:
            try:
                clean_query = self.query_normalizer(clean_query)
            except Exception as norm_err:
                print(f"⚠️  [RAGRetriever] Query normalizer error: {norm_err}")

        lim = limit if limit is not None else self.default_limit
        thresh = score_threshold if score_threshold is not None else self.default_score_threshold

        # 1. Generate query embedding vector
        query_vector: list[float] = []
        try:
            query_vector = await self.embedding_model.embed_query(clean_query)
        except Exception as emb_err:
            print(f"⚠️  [RAGRetriever] Embedding generation error: {emb_err}")

        # 2. Search primary vector store
        matches: list[DocumentChunk] = []
        if query_vector:
            try:
                matches = await self.vector_store.search(
                    query_vector=query_vector,
                    limit=lim,
                    score_threshold=thresh,
                    filter_metadata=filter_metadata,
                )
            except Exception as store_err:
                print(f"⚠️  [RAGRetriever] Primary store error ({self.vector_store.name}): {store_err}")

        # 3. Fallback to secondary store if primary returned no matches
        if not matches and self.fallback_store is not None:
            print(f"ℹ️  [RAGRetriever] Querying fallback store: {self.fallback_store.name}")
            try:
                if query_vector:
                    matches = await self.fallback_store.search(
                        query_vector=query_vector,
                        limit=lim,
                        score_threshold=thresh,
                        filter_metadata=filter_metadata,
                    )
                # If still no matches and fallback is InMemoryVectorStore, try keyword search
                if not matches and isinstance(self.fallback_store, InMemoryVectorStore):
                    matches = self.fallback_store.search_keyword(clean_query, limit=lim)
            except Exception as fb_err:
                print(f"⚠️  [RAGRetriever] Fallback store error ({self.fallback_store.name}): {fb_err}")

        return matches

    def build_context_text(
        self,
        matches: list[DocumentChunk],
        header: str | None = None,
        footer: str | None = None,
    ) -> str:
        """Formats retrieved chunks into a standardized context block for LLM prompts."""
        if not matches:
            return ""
        blocks = [f"- [{m.title}]: {m.text}" for m in matches if m.text]
        content = "\n\n".join(blocks)
        parts = []
        if header:
            parts.append(header)
        parts.append(content)
        if footer:
            parts.append(footer)
        return "\n\n".join(parts)

    async def __call__(
        self,
        query: str,
        limit: int | None = None,
        score_threshold: float | None = None,
        filter_metadata: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> list[dict[str, Any]]:
        """
        Callable interface directly compatible with KnowledgeBaseTool's retriever_func.
        Returns list of dicts.
        """
        chunks = await self.retrieve(
            query=query,
            limit=limit,
            score_threshold=score_threshold,
            filter_metadata=filter_metadata,
        )
        return [c.to_dict() for c in chunks]

__call__ async 

__call__(query, limit=None, score_threshold=None, filter_metadata=None, **kwargs)

Callable interface directly compatible with KnowledgeBaseTool's retriever_func. Returns list of dicts.

Source code in edgevoiceai/rag/retriever.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
async def __call__(
    self,
    query: str,
    limit: int | None = None,
    score_threshold: float | None = None,
    filter_metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[dict[str, Any]]:
    """
    Callable interface directly compatible with KnowledgeBaseTool's retriever_func.
    Returns list of dicts.
    """
    chunks = await self.retrieve(
        query=query,
        limit=limit,
        score_threshold=score_threshold,
        filter_metadata=filter_metadata,
    )
    return [c.to_dict() for c in chunks]

build_context_text 

build_context_text(matches, header=None, footer=None)

Formats retrieved chunks into a standardized context block for LLM prompts.

Source code in edgevoiceai/rag/retriever.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def build_context_text(
    self,
    matches: list[DocumentChunk],
    header: str | None = None,
    footer: str | None = None,
) -> str:
    """Formats retrieved chunks into a standardized context block for LLM prompts."""
    if not matches:
        return ""
    blocks = [f"- [{m.title}]: {m.text}" for m in matches if m.text]
    content = "\n\n".join(blocks)
    parts = []
    if header:
        parts.append(header)
    parts.append(content)
    if footer:
        parts.append(footer)
    return "\n\n".join(parts)

retrieve async 

retrieve(query, limit=None, score_threshold=None, filter_metadata=None)

Retrieves relevant document chunks for a natural language voice query. Falls back to fallback_store if primary store is unreachable or yields no matches.

Source code in edgevoiceai/rag/retriever.py 
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
async def retrieve(
    self,
    query: str,
    limit: int | None = None,
    score_threshold: float | None = None,
    filter_metadata: dict[str, Any] | None = None,
) -> list[DocumentChunk]:
    """
    Retrieves relevant document chunks for a natural language voice query.
    Falls back to fallback_store if primary store is unreachable or yields no matches.
    """
    clean_query = (query or "").strip()
    if not clean_query:
        return []

    if self.query_normalizer is not None:
        try:
            clean_query = self.query_normalizer(clean_query)
        except Exception as norm_err:
            print(f"⚠️  [RAGRetriever] Query normalizer error: {norm_err}")

    lim = limit if limit is not None else self.default_limit
    thresh = score_threshold if score_threshold is not None else self.default_score_threshold

    # 1. Generate query embedding vector
    query_vector: list[float] = []
    try:
        query_vector = await self.embedding_model.embed_query(clean_query)
    except Exception as emb_err:
        print(f"⚠️  [RAGRetriever] Embedding generation error: {emb_err}")

    # 2. Search primary vector store
    matches: list[DocumentChunk] = []
    if query_vector:
        try:
            matches = await self.vector_store.search(
                query_vector=query_vector,
                limit=lim,
                score_threshold=thresh,
                filter_metadata=filter_metadata,
            )
        except Exception as store_err:
            print(f"⚠️  [RAGRetriever] Primary store error ({self.vector_store.name}): {store_err}")

    # 3. Fallback to secondary store if primary returned no matches
    if not matches and self.fallback_store is not None:
        print(f"ℹ️  [RAGRetriever] Querying fallback store: {self.fallback_store.name}")
        try:
            if query_vector:
                matches = await self.fallback_store.search(
                    query_vector=query_vector,
                    limit=lim,
                    score_threshold=thresh,
                    filter_metadata=filter_metadata,
                )
            # If still no matches and fallback is InMemoryVectorStore, try keyword search
            if not matches and isinstance(self.fallback_store, InMemoryVectorStore):
                matches = self.fallback_store.search_keyword(clean_query, limit=lim)
        except Exception as fb_err:
            print(f"⚠️  [RAGRetriever] Fallback store error ({self.fallback_store.name}): {fb_err}")

    return matches

Vector Stores

BaseVectorStore

edgevoiceai.rag.stores.base

Base interface for Vector Database Stores.

BaseVectorStore

Bases: ABC

Abstract Base Class for vector storage and semantic retrieval backends.

Source code in edgevoiceai/rag/stores/base.py 
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
class BaseVectorStore(ABC):
    """
    Abstract Base Class for vector storage and semantic retrieval backends.
    """

    @property
    @abstractmethod
    def name(self) -> str:
        """Name of the vector database backend (e.g. In-Memory, Qdrant, Pinecone)."""
        pass

    @abstractmethod
    async def search(
        self,
        query_vector: list[float],
        limit: int = 3,
        score_threshold: float = 0.5,
        filter_metadata: dict[str, Any] | None = None,
    ) -> list[DocumentChunk]:
        """
        Searches the vector store using dense query vectors.
        Returns matched DocumentChunks ranked by descending similarity score.
        """
        pass

    @abstractmethod
    async def upsert(
        self,
        chunks: list[DocumentChunk],
        vectors: list[list[float]],
    ) -> None:
        """Stores or updates document chunks with their dense vector embeddings."""
        pass

    @abstractmethod
    async def is_available(self) -> bool:
        """Returns True if the backend is reachable and ready to serve queries."""
        pass

    async def get_stats(self) -> dict[str, Any]:
        """Returns collection/index statistics (point counts, dimensions, status)."""
        available = await self.is_available()
        return {
            "status": "online" if available else "offline",
            "name": self.name,
            "points_count": 0,
            "vectors_count": 0,
        }

    async def delete_points(self, point_ids: list[str]) -> int:
        """Deletes specific point IDs from the vector store."""
        return 0

    async def delete_by_filter(self, filter_metadata: dict[str, Any]) -> bool:
        """Deletes documents matching the specified metadata filter."""
        return False

    async def init_collection(self) -> bool:
        """Ensures the collection or index exists, creating it if required."""
        return True

name abstractmethod property 

name

Name of the vector database backend (e.g. In-Memory, Qdrant, Pinecone).

delete_by_filter async 

delete_by_filter(filter_metadata)

Deletes documents matching the specified metadata filter.

Source code in edgevoiceai/rag/stores/base.py 
65
66
67
async def delete_by_filter(self, filter_metadata: dict[str, Any]) -> bool:
    """Deletes documents matching the specified metadata filter."""
    return False

delete_points async 

delete_points(point_ids)

Deletes specific point IDs from the vector store.

Source code in edgevoiceai/rag/stores/base.py 
61
62
63
async def delete_points(self, point_ids: list[str]) -> int:
    """Deletes specific point IDs from the vector store."""
    return 0

get_stats async 

get_stats()

Returns collection/index statistics (point counts, dimensions, status).

Source code in edgevoiceai/rag/stores/base.py 
51
52
53
54
55
56
57
58
59
async def get_stats(self) -> dict[str, Any]:
    """Returns collection/index statistics (point counts, dimensions, status)."""
    available = await self.is_available()
    return {
        "status": "online" if available else "offline",
        "name": self.name,
        "points_count": 0,
        "vectors_count": 0,
    }

init_collection async 

init_collection()

Ensures the collection or index exists, creating it if required.

Source code in edgevoiceai/rag/stores/base.py 
69
70
71
async def init_collection(self) -> bool:
    """Ensures the collection or index exists, creating it if required."""
    return True

is_available abstractmethod async 

is_available()

Returns True if the backend is reachable and ready to serve queries.

Source code in edgevoiceai/rag/stores/base.py 
46
47
48
49
@abstractmethod
async def is_available(self) -> bool:
    """Returns True if the backend is reachable and ready to serve queries."""
    pass

search abstractmethod async 

search(query_vector, limit=3, score_threshold=0.5, filter_metadata=None)

Searches the vector store using dense query vectors. Returns matched DocumentChunks ranked by descending similarity score.

Source code in edgevoiceai/rag/stores/base.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
@abstractmethod
async def search(
    self,
    query_vector: list[float],
    limit: int = 3,
    score_threshold: float = 0.5,
    filter_metadata: dict[str, Any] | None = None,
) -> list[DocumentChunk]:
    """
    Searches the vector store using dense query vectors.
    Returns matched DocumentChunks ranked by descending similarity score.
    """
    pass

upsert abstractmethod async 

upsert(chunks, vectors)

Stores or updates document chunks with their dense vector embeddings.

Source code in edgevoiceai/rag/stores/base.py 
37
38
39
40
41
42
43
44
@abstractmethod
async def upsert(
    self,
    chunks: list[DocumentChunk],
    vectors: list[list[float]],
) -> None:
    """Stores or updates document chunks with their dense vector embeddings."""
    pass

InMemoryVectorStore

edgevoiceai.rag.stores.memory

In-Memory & Local File Vector Store. Provides pure-Python vector search with cosine similarity and keyword fallback. Zero external services or Docker containers required. Ideal for edge devices and offline testing.

InMemoryVectorStore

Bases: BaseVectorStore

In-memory vector store with optional JSON persistence.

Source code in edgevoiceai/rag/stores/memory.py 
 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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
class InMemoryVectorStore(BaseVectorStore):
    """
    In-memory vector store with optional JSON persistence.
    """

    def __init__(self, name: str = "In-Memory Store") -> None:
        self._name = name
        self._chunks: list[DocumentChunk] = []
        self._vectors: list[list[float]] = []

    @property
    def name(self) -> str:
        return self._name

    async def is_available(self) -> bool:
        return True

    async def get_stats(self) -> dict[str, Any]:
        """Returns statistics of in-memory store."""
        return {
            "status": "online",
            "name": self.name,
            "points_count": len(self._chunks),
            "vectors_count": len(self._vectors),
        }

    async def delete_points(self, point_ids: list[str]) -> int:
        """Deletes points by ID."""
        id_set = set(point_ids)
        new_chunks = []
        new_vectors = []
        deleted = 0
        for c, v in zip(self._chunks, self._vectors):
            if c.id in id_set:
                deleted += 1
            else:
                new_chunks.append(c)
                new_vectors.append(v)
        self._chunks = new_chunks
        self._vectors = new_vectors
        return deleted

    async def delete_by_filter(self, filter_metadata: dict[str, Any]) -> bool:
        """Deletes chunks matching filter."""
        new_chunks = []
        new_vectors = []
        for c, v in zip(self._chunks, self._vectors):
            match = True
            for k, val in filter_metadata.items():
                if c.metadata.get(k) != val:
                    match = False
                    break
            if not match:
                new_chunks.append(c)
                new_vectors.append(v)
        self._chunks = new_chunks
        self._vectors = new_vectors
        return True

    async def upsert(
        self,
        chunks: list[DocumentChunk],
        vectors: list[list[float]],
    ) -> None:
        """Stores chunks with their corresponding vectors."""
        for c, v in zip(chunks, vectors):
            c.vector_store = self.name
            self._chunks.append(c)
            self._vectors.append(v)

    async def search(
        self,
        query_vector: list[float],
        limit: int = 3,
        score_threshold: float = 0.5,
        filter_metadata: dict[str, Any] | None = None,
    ) -> list[DocumentChunk]:
        """Performs cosine similarity search against stored vectors."""
        if not self._chunks:
            return []

        # If query vector is empty or all zeros, return empty (or let keyword search handle)
        if not query_vector or all(v == 0.0 for v in query_vector):
            return []

        results: list[tuple[float, DocumentChunk]] = []

        for chunk, vec in zip(self._chunks, self._vectors):
            # Check metadata filter
            if filter_metadata:
                match = True
                for k, v in filter_metadata.items():
                    if chunk.metadata.get(k) != v:
                        match = False
                        break
                if not match:
                    continue

            score = cosine_similarity(query_vector, vec)
            if score >= score_threshold:
                match_chunk = DocumentChunk(
                    id=chunk.id,
                    title=chunk.title,
                    text=chunk.text,
                    score=round(score, 4),
                    metadata=chunk.metadata,
                    audio_url=chunk.audio_url,
                    vector_store=self.name,
                )
                results.append((score, match_chunk))

        # Sort descending by score
        results.sort(key=lambda x: x[0], reverse=True)
        return [chunk for _, chunk in results[:limit]]

    def search_keyword(
        self,
        query: str,
        limit: int = 3,
    ) -> list[DocumentChunk]:
        """
        Fast token-based lexical search fallback when embeddings are unavailable.
        """
        if not query or not self._chunks:
            return []

        words = [w.lower() for w in query.strip().split() if len(w) > 1]
        if not words:
            return []

        scored: list[tuple[int, DocumentChunk]] = []
        for chunk in self._chunks:
            haystack = f"{chunk.title} {chunk.text}".lower()
            match_count = sum(1 for w in words if w in haystack)
            if match_count > 0:
                scored.append((match_count, chunk))

        scored.sort(key=lambda x: x[0], reverse=True)
        return [
            DocumentChunk(
                id=c.id,
                title=c.title,
                text=c.text,
                score=round(score / max(len(words), 1), 2),
                metadata=c.metadata,
                audio_url=c.audio_url,
                vector_store=f"{self.name} (Keyword)",
            )
            for score, c in scored[:limit]
        ]

    @classmethod
    def from_json_file(cls, filepath: str) -> InMemoryVectorStore:
        """
        Loads document chunks and vectors from a structured JSON file.
        Accepts list of dictionaries with {title, text, metadata, audio_url, vector?}.
        """
        store = cls(name=f"JSON Store ({os.path.basename(filepath)})")
        if not os.path.isfile(filepath):
            return store

        try:
            with open(filepath, "r", encoding="utf-8") as f:
                data = json.load(f)

            items = data if isinstance(data, list) else data.get("records", [])
            for item in items:
                chunk = DocumentChunk(
                    title=item.get("title", ""),
                    text=item.get("text", "") or item.get("content", ""),
                    metadata=item.get("metadata", {}),
                    audio_url=item.get("audio_url"),
                    vector_store=store.name,
                )
                vec = item.get("vector") or item.get("embedding") or []
                store._chunks.append(chunk)
                store._vectors.append(vec if isinstance(vec, list) else [])
        except Exception as exc:
            print(f"⚠️  [InMemoryVectorStore] Error loading {filepath}: {exc}")

        return store

delete_by_filter async 

delete_by_filter(filter_metadata)

Deletes chunks matching filter.

Source code in edgevoiceai/rag/stores/memory.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
async def delete_by_filter(self, filter_metadata: dict[str, Any]) -> bool:
    """Deletes chunks matching filter."""
    new_chunks = []
    new_vectors = []
    for c, v in zip(self._chunks, self._vectors):
        match = True
        for k, val in filter_metadata.items():
            if c.metadata.get(k) != val:
                match = False
                break
        if not match:
            new_chunks.append(c)
            new_vectors.append(v)
    self._chunks = new_chunks
    self._vectors = new_vectors
    return True

delete_points async 

delete_points(point_ids)

Deletes points by ID.

Source code in edgevoiceai/rag/stores/memory.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
async def delete_points(self, point_ids: list[str]) -> int:
    """Deletes points by ID."""
    id_set = set(point_ids)
    new_chunks = []
    new_vectors = []
    deleted = 0
    for c, v in zip(self._chunks, self._vectors):
        if c.id in id_set:
            deleted += 1
        else:
            new_chunks.append(c)
            new_vectors.append(v)
    self._chunks = new_chunks
    self._vectors = new_vectors
    return deleted

from_json_file classmethod 

from_json_file(filepath)

Loads document chunks and vectors from a structured JSON file. Accepts list of dictionaries with {title, text, metadata, audio_url, vector?}.

Source code in edgevoiceai/rag/stores/memory.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
@classmethod
def from_json_file(cls, filepath: str) -> InMemoryVectorStore:
    """
    Loads document chunks and vectors from a structured JSON file.
    Accepts list of dictionaries with {title, text, metadata, audio_url, vector?}.
    """
    store = cls(name=f"JSON Store ({os.path.basename(filepath)})")
    if not os.path.isfile(filepath):
        return store

    try:
        with open(filepath, "r", encoding="utf-8") as f:
            data = json.load(f)

        items = data if isinstance(data, list) else data.get("records", [])
        for item in items:
            chunk = DocumentChunk(
                title=item.get("title", ""),
                text=item.get("text", "") or item.get("content", ""),
                metadata=item.get("metadata", {}),
                audio_url=item.get("audio_url"),
                vector_store=store.name,
            )
            vec = item.get("vector") or item.get("embedding") or []
            store._chunks.append(chunk)
            store._vectors.append(vec if isinstance(vec, list) else [])
    except Exception as exc:
        print(f"⚠️  [InMemoryVectorStore] Error loading {filepath}: {exc}")

    return store

get_stats async 

get_stats()

Returns statistics of in-memory store.

Source code in edgevoiceai/rag/stores/memory.py 
46
47
48
49
50
51
52
53
async def get_stats(self) -> dict[str, Any]:
    """Returns statistics of in-memory store."""
    return {
        "status": "online",
        "name": self.name,
        "points_count": len(self._chunks),
        "vectors_count": len(self._vectors),
    }

search async 

search(query_vector, limit=3, score_threshold=0.5, filter_metadata=None)

Performs cosine similarity search against stored vectors.

Source code in edgevoiceai/rag/stores/memory.py 
 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
136
137
138
139
140
141
142
async def search(
    self,
    query_vector: list[float],
    limit: int = 3,
    score_threshold: float = 0.5,
    filter_metadata: dict[str, Any] | None = None,
) -> list[DocumentChunk]:
    """Performs cosine similarity search against stored vectors."""
    if not self._chunks:
        return []

    # If query vector is empty or all zeros, return empty (or let keyword search handle)
    if not query_vector or all(v == 0.0 for v in query_vector):
        return []

    results: list[tuple[float, DocumentChunk]] = []

    for chunk, vec in zip(self._chunks, self._vectors):
        # Check metadata filter
        if filter_metadata:
            match = True
            for k, v in filter_metadata.items():
                if chunk.metadata.get(k) != v:
                    match = False
                    break
            if not match:
                continue

        score = cosine_similarity(query_vector, vec)
        if score >= score_threshold:
            match_chunk = DocumentChunk(
                id=chunk.id,
                title=chunk.title,
                text=chunk.text,
                score=round(score, 4),
                metadata=chunk.metadata,
                audio_url=chunk.audio_url,
                vector_store=self.name,
            )
            results.append((score, match_chunk))

    # Sort descending by score
    results.sort(key=lambda x: x[0], reverse=True)
    return [chunk for _, chunk in results[:limit]]

search_keyword 

search_keyword(query, limit=3)

Fast token-based lexical search fallback when embeddings are unavailable.

Source code in edgevoiceai/rag/stores/memory.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def search_keyword(
    self,
    query: str,
    limit: int = 3,
) -> list[DocumentChunk]:
    """
    Fast token-based lexical search fallback when embeddings are unavailable.
    """
    if not query or not self._chunks:
        return []

    words = [w.lower() for w in query.strip().split() if len(w) > 1]
    if not words:
        return []

    scored: list[tuple[int, DocumentChunk]] = []
    for chunk in self._chunks:
        haystack = f"{chunk.title} {chunk.text}".lower()
        match_count = sum(1 for w in words if w in haystack)
        if match_count > 0:
            scored.append((match_count, chunk))

    scored.sort(key=lambda x: x[0], reverse=True)
    return [
        DocumentChunk(
            id=c.id,
            title=c.title,
            text=c.text,
            score=round(score / max(len(words), 1), 2),
            metadata=c.metadata,
            audio_url=c.audio_url,
            vector_store=f"{self.name} (Keyword)",
        )
        for score, c in scored[:limit]
    ]

upsert async 

upsert(chunks, vectors)

Stores chunks with their corresponding vectors.

Source code in edgevoiceai/rag/stores/memory.py 
88
89
90
91
92
93
94
95
96
97
async def upsert(
    self,
    chunks: list[DocumentChunk],
    vectors: list[list[float]],
) -> None:
    """Stores chunks with their corresponding vectors."""
    for c, v in zip(chunks, vectors):
        c.vector_store = self.name
        self._chunks.append(c)
        self._vectors.append(v)

QdrantVectorStore

edgevoiceai.rag.stores.qdrant

Qdrant Vector Database Store Adapter. Supports local, on-premise, and cloud Qdrant instances.

QdrantVectorStore

Bases: BaseVectorStore

Qdrant vector database store adapter.

Source code in edgevoiceai/rag/stores/qdrant.py 
 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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
class QdrantVectorStore(BaseVectorStore):
    """
    Qdrant vector database store adapter.
    """

    def __init__(
        self,
        collection_name: str = "knowledge_base",
        url: str | None = None,
        host: str = "localhost",
        port: int = 6333,
        api_key: str | None = None,
    ) -> None:
        self.collection_name = collection_name
        self.url = url or os.environ.get("QDRANT_URL", f"http://{host}:{port}")
        self.api_key = api_key or os.environ.get("QDRANT_API_KEY")
        self._client: Any = None

    @property
    def name(self) -> str:
        return f"Qdrant ({self.collection_name})"

    def _get_client(self) -> Any:
        if not QDRANT_AVAILABLE or AsyncQdrantClient is None:
            return None
        if self._client is None:
            self._client = AsyncQdrantClient(url=self.url, api_key=self.api_key, timeout=5, check_compatibility=False)
        return self._client

    async def is_available(self) -> bool:
        """Checks if Qdrant server is reachable and collection exists."""
        client = self._get_client()
        if client is None:
            return False
        try:
            collections = await client.get_collections()
            return any(c.name == self.collection_name for c in collections.collections)
        except Exception:
            return False

    async def init_collection(self, vector_size: int = 768, distance: str = "Cosine") -> bool:
        """Ensures the collection exists with the required vector dimension and distance metric."""
        client = self._get_client()
        if client is None or qmodels is None:
            return False
        try:
            collections = await client.get_collections()
            existing = [c.name for c in collections.collections]
            if self.collection_name not in existing:
                dist_metric = qmodels.Distance.COSINE
                if distance.lower() == "euclid":
                    dist_metric = qmodels.Distance.EUCLID
                elif distance.lower() == "dot":
                    dist_metric = qmodels.Distance.DOT

                await client.create_collection(
                    collection_name=self.collection_name,
                    vectors_config=qmodels.VectorParams(
                        size=vector_size,
                        distance=dist_metric,
                    ),
                )
            return True
        except Exception as exc:
            print(f"⚠️  [QdrantVectorStore] Collection init skipped/error: {exc}")
            return False

    async def get_stats(self) -> dict[str, Any]:
        """Retrieves collection statistics (point counts, indexed vectors, dimension)."""
        client = self._get_client()
        if client is None or not await self.is_available():
            return {
                "status": "offline",
                "collection_name": self.collection_name,
                "points_count": 0,
                "vectors_count": 0,
            }
        try:
            info = await client.get_collection(collection_name=self.collection_name)
            points = getattr(info, "points_count", 0) or 0
            vectors = getattr(info, "indexed_vectors_count", points) or points
            vector_size = 768
            distance = "Cosine"
            try:
                if hasattr(info.config.params, "vectors"):
                    v_params = info.config.params.vectors
                    if hasattr(v_params, "size"):
                        vector_size = v_params.size
                    if hasattr(v_params, "distance"):
                        distance = str(v_params.distance.value if hasattr(v_params.distance, "value") else v_params.distance)
            except Exception:
                pass

            return {
                "status": "online",
                "collection_name": self.collection_name,
                "points_count": points,
                "vectors_count": vectors,
                "vector_size": vector_size,
                "distance": distance,
            }
        except Exception as exc:
            return {"status": "error", "error": str(exc), "points_count": 0}

    async def delete_points(self, point_ids: list[str]) -> int:
        """Deletes specific point IDs from Qdrant."""
        if not point_ids or qmodels is None:
            return 0
        client = self._get_client()
        if client is None:
            return 0
        try:
            await client.delete(
                collection_name=self.collection_name,
                points_selector=qmodels.PointIdsList(points=point_ids),
            )
            return len(point_ids)
        except Exception as exc:
            print(f"⚠️  [QdrantVectorStore] Delete points error: {exc}")
            return 0

    async def delete_by_filter(self, filter_metadata: dict[str, Any]) -> bool:
        """Deletes all points matching the given metadata filter."""
        if not filter_metadata or qmodels is None:
            return False
        client = self._get_client()
        if client is None:
            return False
        try:
            conditions: list[Any] = [
                qmodels.FieldCondition(key=k, match=qmodels.MatchValue(value=v))
                for k, v in filter_metadata.items()
            ]
            await client.delete(
                collection_name=self.collection_name,
                points_selector=qmodels.FilterSelector(
                    filter=qmodels.Filter(must=conditions)
                ),
            )
            return True
        except Exception as exc:
            print(f"⚠️  [QdrantVectorStore] Delete by filter error: {exc}")
            return False

    async def search(
        self,
        query_vector: list[float],
        limit: int = 3,
        score_threshold: float = 0.5,
        filter_metadata: dict[str, Any] | None = None,
    ) -> list[DocumentChunk]:
        """Executes vector similarity search on Qdrant collection."""
        client = self._get_client()
        if client is None:
            return []

        q_filter = None
        if filter_metadata and qmodels is not None:
            must_conditions: list[Any] = []
            should_conditions: list[Any] = []
            for k, v in filter_metadata.items():
                if isinstance(v, list):
                    for item in v:
                        should_conditions.append(qmodels.FieldCondition(key=k, match=qmodels.MatchValue(value=item)))
                else:
                    must_conditions.append(qmodels.FieldCondition(key=k, match=qmodels.MatchValue(value=v)))

            if must_conditions and should_conditions:
                q_filter = qmodels.Filter(must=must_conditions, should=should_conditions)
            elif should_conditions:
                q_filter = qmodels.Filter(should=should_conditions)
            elif must_conditions:
                q_filter = qmodels.Filter(must=must_conditions)

        try:
            # Use query_points API
            results = await client.query_points(
                collection_name=self.collection_name,
                query=query_vector,
                limit=limit,
                score_threshold=score_threshold,
                query_filter=q_filter,
                with_payload=True,
            )

            chunks: list[DocumentChunk] = []
            for hit in results.points:
                payload = hit.payload or {}
                meta = dict(payload.get("metadata", {}))
                for k, v in payload.items():
                    if k not in ("title", "text", "metadata", "audio_url"):
                        meta[k] = v
                chunk = DocumentChunk(
                    id=str(hit.id),
                    title=str(payload.get("title", "")),
                    text=str(payload.get("text", "")),
                    score=float(hit.score if hit.score is not None else 0.0),
                    metadata=meta,
                    audio_url=payload.get("audio_url"),
                    vector_store=self.name,
                )
                chunks.append(chunk)

            return chunks
        except Exception as exc:
            print(f"⚠️  [QdrantVectorStore] Search error on {self.collection_name}: {exc}")
            return []

    async def upsert(
        self,
        chunks: list[DocumentChunk],
        vectors: list[list[float]],
    ) -> None:
        """Upserts document chunks and embeddings into Qdrant in batches."""
        client = self._get_client()
        if client is None or qmodels is None:
            raise RuntimeError("Qdrant client not available for upsert.")

        await self.init_collection()

        points = [
            qmodels.PointStruct(
                id=c.id,
                vector=v,
                payload={
                    "title": c.title,
                    "text": c.text,
                    "audio_url": c.audio_url or "",
                    "metadata": c.metadata,
                    **c.metadata,
                },
            )
            for c, v in zip(chunks, vectors)
        ]

        batch_size = 50
        for i in range(0, len(points), batch_size):
            batch = points[i : i + batch_size]
            await client.upsert(collection_name=self.collection_name, points=batch)

delete_by_filter async 

delete_by_filter(filter_metadata)

Deletes all points matching the given metadata filter.

Source code in edgevoiceai/rag/stores/qdrant.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
async def delete_by_filter(self, filter_metadata: dict[str, Any]) -> bool:
    """Deletes all points matching the given metadata filter."""
    if not filter_metadata or qmodels is None:
        return False
    client = self._get_client()
    if client is None:
        return False
    try:
        conditions: list[Any] = [
            qmodels.FieldCondition(key=k, match=qmodels.MatchValue(value=v))
            for k, v in filter_metadata.items()
        ]
        await client.delete(
            collection_name=self.collection_name,
            points_selector=qmodels.FilterSelector(
                filter=qmodels.Filter(must=conditions)
            ),
        )
        return True
    except Exception as exc:
        print(f"⚠️  [QdrantVectorStore] Delete by filter error: {exc}")
        return False

delete_points async 

delete_points(point_ids)

Deletes specific point IDs from Qdrant.

Source code in edgevoiceai/rag/stores/qdrant.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
async def delete_points(self, point_ids: list[str]) -> int:
    """Deletes specific point IDs from Qdrant."""
    if not point_ids or qmodels is None:
        return 0
    client = self._get_client()
    if client is None:
        return 0
    try:
        await client.delete(
            collection_name=self.collection_name,
            points_selector=qmodels.PointIdsList(points=point_ids),
        )
        return len(point_ids)
    except Exception as exc:
        print(f"⚠️  [QdrantVectorStore] Delete points error: {exc}")
        return 0

get_stats async 

get_stats()

Retrieves collection statistics (point counts, indexed vectors, dimension).

Source code in edgevoiceai/rag/stores/qdrant.py 
 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
async def get_stats(self) -> dict[str, Any]:
    """Retrieves collection statistics (point counts, indexed vectors, dimension)."""
    client = self._get_client()
    if client is None or not await self.is_available():
        return {
            "status": "offline",
            "collection_name": self.collection_name,
            "points_count": 0,
            "vectors_count": 0,
        }
    try:
        info = await client.get_collection(collection_name=self.collection_name)
        points = getattr(info, "points_count", 0) or 0
        vectors = getattr(info, "indexed_vectors_count", points) or points
        vector_size = 768
        distance = "Cosine"
        try:
            if hasattr(info.config.params, "vectors"):
                v_params = info.config.params.vectors
                if hasattr(v_params, "size"):
                    vector_size = v_params.size
                if hasattr(v_params, "distance"):
                    distance = str(v_params.distance.value if hasattr(v_params.distance, "value") else v_params.distance)
        except Exception:
            pass

        return {
            "status": "online",
            "collection_name": self.collection_name,
            "points_count": points,
            "vectors_count": vectors,
            "vector_size": vector_size,
            "distance": distance,
        }
    except Exception as exc:
        return {"status": "error", "error": str(exc), "points_count": 0}

init_collection async 

init_collection(vector_size=768, distance='Cosine')

Ensures the collection exists with the required vector dimension and distance metric.

Source code in edgevoiceai/rag/stores/qdrant.py 
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
async def init_collection(self, vector_size: int = 768, distance: str = "Cosine") -> bool:
    """Ensures the collection exists with the required vector dimension and distance metric."""
    client = self._get_client()
    if client is None or qmodels is None:
        return False
    try:
        collections = await client.get_collections()
        existing = [c.name for c in collections.collections]
        if self.collection_name not in existing:
            dist_metric = qmodels.Distance.COSINE
            if distance.lower() == "euclid":
                dist_metric = qmodels.Distance.EUCLID
            elif distance.lower() == "dot":
                dist_metric = qmodels.Distance.DOT

            await client.create_collection(
                collection_name=self.collection_name,
                vectors_config=qmodels.VectorParams(
                    size=vector_size,
                    distance=dist_metric,
                ),
            )
        return True
    except Exception as exc:
        print(f"⚠️  [QdrantVectorStore] Collection init skipped/error: {exc}")
        return False

is_available async 

is_available()

Checks if Qdrant server is reachable and collection exists.

Source code in edgevoiceai/rag/stores/qdrant.py 
52
53
54
55
56
57
58
59
60
61
async def is_available(self) -> bool:
    """Checks if Qdrant server is reachable and collection exists."""
    client = self._get_client()
    if client is None:
        return False
    try:
        collections = await client.get_collections()
        return any(c.name == self.collection_name for c in collections.collections)
    except Exception:
        return False

search async 

search(query_vector, limit=3, score_threshold=0.5, filter_metadata=None)

Executes vector similarity search on Qdrant collection.

Source code in edgevoiceai/rag/stores/qdrant.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
async def search(
    self,
    query_vector: list[float],
    limit: int = 3,
    score_threshold: float = 0.5,
    filter_metadata: dict[str, Any] | None = None,
) -> list[DocumentChunk]:
    """Executes vector similarity search on Qdrant collection."""
    client = self._get_client()
    if client is None:
        return []

    q_filter = None
    if filter_metadata and qmodels is not None:
        must_conditions: list[Any] = []
        should_conditions: list[Any] = []
        for k, v in filter_metadata.items():
            if isinstance(v, list):
                for item in v:
                    should_conditions.append(qmodels.FieldCondition(key=k, match=qmodels.MatchValue(value=item)))
            else:
                must_conditions.append(qmodels.FieldCondition(key=k, match=qmodels.MatchValue(value=v)))

        if must_conditions and should_conditions:
            q_filter = qmodels.Filter(must=must_conditions, should=should_conditions)
        elif should_conditions:
            q_filter = qmodels.Filter(should=should_conditions)
        elif must_conditions:
            q_filter = qmodels.Filter(must=must_conditions)

    try:
        # Use query_points API
        results = await client.query_points(
            collection_name=self.collection_name,
            query=query_vector,
            limit=limit,
            score_threshold=score_threshold,
            query_filter=q_filter,
            with_payload=True,
        )

        chunks: list[DocumentChunk] = []
        for hit in results.points:
            payload = hit.payload or {}
            meta = dict(payload.get("metadata", {}))
            for k, v in payload.items():
                if k not in ("title", "text", "metadata", "audio_url"):
                    meta[k] = v
            chunk = DocumentChunk(
                id=str(hit.id),
                title=str(payload.get("title", "")),
                text=str(payload.get("text", "")),
                score=float(hit.score if hit.score is not None else 0.0),
                metadata=meta,
                audio_url=payload.get("audio_url"),
                vector_store=self.name,
            )
            chunks.append(chunk)

        return chunks
    except Exception as exc:
        print(f"⚠️  [QdrantVectorStore] Search error on {self.collection_name}: {exc}")
        return []

upsert async 

upsert(chunks, vectors)

Upserts document chunks and embeddings into Qdrant in batches.

Source code in edgevoiceai/rag/stores/qdrant.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
async def upsert(
    self,
    chunks: list[DocumentChunk],
    vectors: list[list[float]],
) -> None:
    """Upserts document chunks and embeddings into Qdrant in batches."""
    client = self._get_client()
    if client is None or qmodels is None:
        raise RuntimeError("Qdrant client not available for upsert.")

    await self.init_collection()

    points = [
        qmodels.PointStruct(
            id=c.id,
            vector=v,
            payload={
                "title": c.title,
                "text": c.text,
                "audio_url": c.audio_url or "",
                "metadata": c.metadata,
                **c.metadata,
            },
        )
        for c, v in zip(chunks, vectors)
    ]

    batch_size = 50
    for i in range(0, len(points), batch_size):
        batch = points[i : i + batch_size]
        await client.upsert(collection_name=self.collection_name, points=batch)

PineconeVectorStore

edgevoiceai.rag.stores.pinecone

Pinecone Serverless Cloud Vector Database Store Adapter.

PineconeVectorStore

Bases: BaseVectorStore

Pinecone serverless cloud vector database store adapter.

Source code in edgevoiceai/rag/stores/pinecone.py 
 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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
class PineconeVectorStore(BaseVectorStore):
    """
    Pinecone serverless cloud vector database store adapter.
    """

    def __init__(
        self,
        index_name: str = "knowledge-base",
        api_key: str | None = None,
        namespace: str = "",
    ) -> None:
        self.index_name = index_name
        self.api_key = api_key or os.environ.get("PINECONE_API_KEY", "")
        self.namespace = namespace
        self._pc: Any = None
        self._index: Any = None
        self._avail_cache_value: bool | None = None
        self._avail_cache_ts: float = 0.0
        self._cache_ttl: float = 300.0

    @property
    def name(self) -> str:
        return f"Pinecone ({self.index_name})"

    def _get_index(self) -> Any:
        if not PINECONE_AVAILABLE or Pinecone is None or not self.api_key:
            return None
        if self._index is None:
            # pyright: ignore[reportCallIssue]
            self._pc = Pinecone(api_key=self.api_key)  # type: ignore[call-arg]
            self._index = self._pc.Index(self.index_name)
        return self._index

    def invalidate_cache(self) -> None:
        """Clears availability cache."""
        self._avail_cache_value = None
        self._avail_cache_ts = 0.0

    async def is_available(self) -> bool:
        """Verifies connection to Pinecone cloud index with cached result."""
        if not PINECONE_AVAILABLE or Pinecone is None or not self.api_key:
            return False

        import time
        now = time.monotonic()
        if self._avail_cache_value is not None and (now - self._avail_cache_ts) < self._cache_ttl:
            return self._avail_cache_value

        def _check() -> bool:
            try:
                idx = self._get_index()
                if idx is None:
                    return False
                idx.describe_index_stats()
                return True
            except Exception:
                return False

        res = await asyncio.to_thread(_check)
        self._avail_cache_value = res
        self._avail_cache_ts = now
        return res

    async def get_stats(self) -> dict[str, Any]:
        """Retrieves Pinecone index statistics."""
        if not self.api_key:
            return {"status": "offline", "index_name": self.index_name, "points_count": 0}

        def _do_stats() -> dict[str, Any]:
            idx = self._get_index()
            if idx is None:
                return {"status": "offline", "index_name": self.index_name, "points_count": 0}
            try:
                stats = idx.describe_index_stats()
                ns = stats.get("namespaces", {}).get(self.namespace, {})
                return {
                    "status": "online",
                    "index_name": self.index_name,
                    "namespace": self.namespace,
                    "vector_size": getattr(stats, "dimension", 768),
                    "points_count": ns.get("vector_count", 0),
                    "vectors_count": ns.get("vector_count", 0),
                    "total_vectors": stats.get("total_vector_count", 0),
                }
            except Exception as exc:
                return {"status": "error", "error": str(exc), "points_count": 0}

        return await asyncio.to_thread(_do_stats)

    async def delete_points(self, point_ids: list[str]) -> int:
        """Deletes specific vector IDs from Pinecone in batches."""
        if not point_ids or not self.api_key:
            return 0

        def _do_delete() -> int:
            idx = self._get_index()
            if idx is None:
                return 0
            batch_size = 100
            for i in range(0, len(point_ids), batch_size):
                batch = point_ids[i : i + batch_size]
                idx.delete(ids=batch, namespace=self.namespace)
            return len(point_ids)

        try:
            return await asyncio.to_thread(_do_delete)
        except Exception as exc:
            print(f"⚠️  [PineconeVectorStore] Delete points error: {exc}")
            return 0

    async def delete_by_filter(self, filter_metadata: dict[str, Any]) -> bool:
        """Deletes all vectors in Pinecone matching metadata filter."""
        if not filter_metadata or not self.api_key:
            return False

        def _do_delete_filter() -> bool:
            idx = self._get_index()
            if idx is None:
                return False
            idx.delete(filter=filter_metadata, namespace=self.namespace)
            return True

        try:
            return await asyncio.to_thread(_do_delete_filter)
        except Exception as exc:
            print(f"⚠️  [PineconeVectorStore] Delete by filter error: {exc}")
            return False

    async def search(
        self,
        query_vector: list[float],
        limit: int = 3,
        score_threshold: float = 0.5,
        filter_metadata: dict[str, Any] | None = None,
    ) -> list[DocumentChunk]:
        """Queries Pinecone index asynchronously via thread offload."""
        if not self.api_key:
            return []

        def _do_query() -> list[DocumentChunk]:
            idx = self._get_index()
            if idx is None:
                return []

            kwargs: dict[str, Any] = {
                "vector": query_vector,
                "top_k": max(1, int(limit)),
                "include_metadata": True,
            }
            if self.namespace:
                kwargs["namespace"] = self.namespace
            if filter_metadata:
                kwargs["filter"] = filter_metadata

            results = idx.query(**kwargs)
            chunks: list[DocumentChunk] = []

            for match in getattr(results, "matches", []):
                score = float(match.get("score", 0.0) if isinstance(match, dict) else getattr(match, "score", 0.0))
                if score < score_threshold:
                    continue

                meta = match.get("metadata", {}) if isinstance(match, dict) else getattr(match, "metadata", {}) or {}
                chunk_id = str(match.get("id", "") if isinstance(match, dict) else getattr(match, "id", ""))

                chunks.append(
                    DocumentChunk(
                        id=chunk_id,
                        title=str(meta.get("title", "")),
                        text=str(meta.get("text", "")),
                        score=score,
                        metadata=meta,
                        audio_url=meta.get("audio_url"),
                        vector_store=self.name,
                    )
                )

            return chunks

        try:
            return await asyncio.to_thread(_do_query)
        except Exception as exc:
            print(f"⚠️  [PineconeVectorStore] Search error on {self.index_name}: {exc}")
            return []

    async def upsert(
        self,
        chunks: list[DocumentChunk],
        vectors: list[list[float]],
    ) -> None:
        """Upserts vectors and metadata into Pinecone."""
        def _do_upsert() -> None:
            idx = self._get_index()
            if idx is None:
                raise RuntimeError("Pinecone index not initialized.")

            records = [
                {
                    "id": c.id,
                    "values": v,
                    "metadata": {
                        "title": c.title,
                        "text": c.text,
                        "audio_url": c.audio_url or "",
                        **c.metadata,
                    },
                }
                for c, v in zip(chunks, vectors)
            ]
            batch_size = 100
            for i in range(0, len(records), batch_size):
                batch = records[i : i + batch_size]
                idx.upsert(vectors=batch, namespace=self.namespace)

        await asyncio.to_thread(_do_upsert)

delete_by_filter async 

delete_by_filter(filter_metadata)

Deletes all vectors in Pinecone matching metadata filter.

Source code in edgevoiceai/rag/stores/pinecone.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
async def delete_by_filter(self, filter_metadata: dict[str, Any]) -> bool:
    """Deletes all vectors in Pinecone matching metadata filter."""
    if not filter_metadata or not self.api_key:
        return False

    def _do_delete_filter() -> bool:
        idx = self._get_index()
        if idx is None:
            return False
        idx.delete(filter=filter_metadata, namespace=self.namespace)
        return True

    try:
        return await asyncio.to_thread(_do_delete_filter)
    except Exception as exc:
        print(f"⚠️  [PineconeVectorStore] Delete by filter error: {exc}")
        return False

delete_points async 

delete_points(point_ids)

Deletes specific vector IDs from Pinecone in batches.

Source code in edgevoiceai/rag/stores/pinecone.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
async def delete_points(self, point_ids: list[str]) -> int:
    """Deletes specific vector IDs from Pinecone in batches."""
    if not point_ids or not self.api_key:
        return 0

    def _do_delete() -> int:
        idx = self._get_index()
        if idx is None:
            return 0
        batch_size = 100
        for i in range(0, len(point_ids), batch_size):
            batch = point_ids[i : i + batch_size]
            idx.delete(ids=batch, namespace=self.namespace)
        return len(point_ids)

    try:
        return await asyncio.to_thread(_do_delete)
    except Exception as exc:
        print(f"⚠️  [PineconeVectorStore] Delete points error: {exc}")
        return 0

get_stats async 

get_stats()

Retrieves Pinecone index statistics.

Source code in edgevoiceai/rag/stores/pinecone.py 
 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
async def get_stats(self) -> dict[str, Any]:
    """Retrieves Pinecone index statistics."""
    if not self.api_key:
        return {"status": "offline", "index_name": self.index_name, "points_count": 0}

    def _do_stats() -> dict[str, Any]:
        idx = self._get_index()
        if idx is None:
            return {"status": "offline", "index_name": self.index_name, "points_count": 0}
        try:
            stats = idx.describe_index_stats()
            ns = stats.get("namespaces", {}).get(self.namespace, {})
            return {
                "status": "online",
                "index_name": self.index_name,
                "namespace": self.namespace,
                "vector_size": getattr(stats, "dimension", 768),
                "points_count": ns.get("vector_count", 0),
                "vectors_count": ns.get("vector_count", 0),
                "total_vectors": stats.get("total_vector_count", 0),
            }
        except Exception as exc:
            return {"status": "error", "error": str(exc), "points_count": 0}

    return await asyncio.to_thread(_do_stats)

invalidate_cache 

invalidate_cache()

Clears availability cache.

Source code in edgevoiceai/rag/stores/pinecone.py 
55
56
57
58
def invalidate_cache(self) -> None:
    """Clears availability cache."""
    self._avail_cache_value = None
    self._avail_cache_ts = 0.0

is_available async 

is_available()

Verifies connection to Pinecone cloud index with cached result.

Source code in edgevoiceai/rag/stores/pinecone.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
async def is_available(self) -> bool:
    """Verifies connection to Pinecone cloud index with cached result."""
    if not PINECONE_AVAILABLE or Pinecone is None or not self.api_key:
        return False

    import time
    now = time.monotonic()
    if self._avail_cache_value is not None and (now - self._avail_cache_ts) < self._cache_ttl:
        return self._avail_cache_value

    def _check() -> bool:
        try:
            idx = self._get_index()
            if idx is None:
                return False
            idx.describe_index_stats()
            return True
        except Exception:
            return False

    res = await asyncio.to_thread(_check)
    self._avail_cache_value = res
    self._avail_cache_ts = now
    return res

search async 

search(query_vector, limit=3, score_threshold=0.5, filter_metadata=None)

Queries Pinecone index asynchronously via thread offload.

Source code in edgevoiceai/rag/stores/pinecone.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
async def search(
    self,
    query_vector: list[float],
    limit: int = 3,
    score_threshold: float = 0.5,
    filter_metadata: dict[str, Any] | None = None,
) -> list[DocumentChunk]:
    """Queries Pinecone index asynchronously via thread offload."""
    if not self.api_key:
        return []

    def _do_query() -> list[DocumentChunk]:
        idx = self._get_index()
        if idx is None:
            return []

        kwargs: dict[str, Any] = {
            "vector": query_vector,
            "top_k": max(1, int(limit)),
            "include_metadata": True,
        }
        if self.namespace:
            kwargs["namespace"] = self.namespace
        if filter_metadata:
            kwargs["filter"] = filter_metadata

        results = idx.query(**kwargs)
        chunks: list[DocumentChunk] = []

        for match in getattr(results, "matches", []):
            score = float(match.get("score", 0.0) if isinstance(match, dict) else getattr(match, "score", 0.0))
            if score < score_threshold:
                continue

            meta = match.get("metadata", {}) if isinstance(match, dict) else getattr(match, "metadata", {}) or {}
            chunk_id = str(match.get("id", "") if isinstance(match, dict) else getattr(match, "id", ""))

            chunks.append(
                DocumentChunk(
                    id=chunk_id,
                    title=str(meta.get("title", "")),
                    text=str(meta.get("text", "")),
                    score=score,
                    metadata=meta,
                    audio_url=meta.get("audio_url"),
                    vector_store=self.name,
                )
            )

        return chunks

    try:
        return await asyncio.to_thread(_do_query)
    except Exception as exc:
        print(f"⚠️  [PineconeVectorStore] Search error on {self.index_name}: {exc}")
        return []

upsert async 

upsert(chunks, vectors)

Upserts vectors and metadata into Pinecone.

Source code in edgevoiceai/rag/stores/pinecone.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
async def upsert(
    self,
    chunks: list[DocumentChunk],
    vectors: list[list[float]],
) -> None:
    """Upserts vectors and metadata into Pinecone."""
    def _do_upsert() -> None:
        idx = self._get_index()
        if idx is None:
            raise RuntimeError("Pinecone index not initialized.")

        records = [
            {
                "id": c.id,
                "values": v,
                "metadata": {
                    "title": c.title,
                    "text": c.text,
                    "audio_url": c.audio_url or "",
                    **c.metadata,
                },
            }
            for c, v in zip(chunks, vectors)
        ]
        batch_size = 100
        for i in range(0, len(records), batch_size):
            batch = records[i : i + batch_size]
            idx.upsert(vectors=batch, namespace=self.namespace)

    await asyncio.to_thread(_do_upsert)

Embedding Services

BaseEmbeddingModel

edgevoiceai.rag.embeddings.base

Base interface for RAG embedding models.

BaseEmbeddingModel

Bases: ABC

Abstract Base Class for text embedding generators.

Source code in edgevoiceai/rag/embeddings/base.py 
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
class BaseEmbeddingModel(ABC):
    """
    Abstract Base Class for text embedding generators.
    """

    @abstractmethod
    async def embed_text(self, text: str) -> list[float]:
        """Generates a dense vector embedding for a single text document."""
        pass

    @abstractmethod
    async def embed_query(self, query: str) -> list[float]:
        """Generates a dense vector embedding for a search query."""
        pass

    async def embed_texts(self, texts: list[str]) -> list[list[float]]:
        """Batch generates embeddings for multiple texts."""
        embeddings: list[list[float]] = []
        for t in texts:
            emb = await self.embed_text(t)
            embeddings.append(emb)
        return embeddings

    async def embed_batch(self, texts: list[str], batch_size: int = 20) -> list[list[float]]:
        """Alias for batched text embeddings generation."""
        return await self.embed_texts(texts)

embed_batch async 

embed_batch(texts, batch_size=20)

Alias for batched text embeddings generation.

Source code in edgevoiceai/rag/embeddings/base.py 
33
34
35
async def embed_batch(self, texts: list[str], batch_size: int = 20) -> list[list[float]]:
    """Alias for batched text embeddings generation."""
    return await self.embed_texts(texts)

embed_query abstractmethod async 

embed_query(query)

Generates a dense vector embedding for a search query.

Source code in edgevoiceai/rag/embeddings/base.py 
20
21
22
23
@abstractmethod
async def embed_query(self, query: str) -> list[float]:
    """Generates a dense vector embedding for a search query."""
    pass

embed_text abstractmethod async 

embed_text(text)

Generates a dense vector embedding for a single text document.

Source code in edgevoiceai/rag/embeddings/base.py 
15
16
17
18
@abstractmethod
async def embed_text(self, text: str) -> list[float]:
    """Generates a dense vector embedding for a single text document."""
    pass

embed_texts async 

embed_texts(texts)

Batch generates embeddings for multiple texts.

Source code in edgevoiceai/rag/embeddings/base.py 
25
26
27
28
29
30
31
async def embed_texts(self, texts: list[str]) -> list[list[float]]:
    """Batch generates embeddings for multiple texts."""
    embeddings: list[list[float]] = []
    for t in texts:
        emb = await self.embed_text(t)
        embeddings.append(emb)
    return embeddings

GeminiEmbeddings

edgevoiceai.rag.embeddings.gemini

Gemini Embeddings Generator. Produces high-quality 768-dimensional dense vectors using Google Gemini API.

GeminiEmbeddings

Bases: BaseEmbeddingModel

Asynchronous embedding generator powered by Google Gemini API. Reuses persistent HTTP connections to eliminate TLS handshake latency on edge devices.

Source code in edgevoiceai/rag/embeddings/gemini.py 
 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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
class GeminiEmbeddings(BaseEmbeddingModel):
    """
    Asynchronous embedding generator powered by Google Gemini API.
    Reuses persistent HTTP connections to eliminate TLS handshake latency on edge devices.
    """

    def __init__(
        self,
        api_key: str | None = None,
        model: str = "gemini-embedding-001",
        dimension: int = 768,
        timeout: float = 25.0,
    ) -> None:
        self.api_key = (api_key or os.environ.get("GEMINI_API_KEY", "")).strip()
        self.model = model
        self.dimension = dimension
        self.timeout = timeout
        self._client: Any = None
        self._cache: dict[str, list[float]] = {}

    def _get_client(self) -> Any:
        if not HTTPX_AVAILABLE or httpx is None:
            raise RuntimeError("httpx is required for GeminiEmbeddings. Install with: pip install httpx")
        if self._client is None or self._client.is_closed:
            self._client = httpx.AsyncClient(
                timeout=self.timeout,
                http2=False,
                limits=httpx.Limits(max_keepalive_connections=5, max_connections=10),
            )
        return self._client

    async def embed_text(self, text: str) -> list[float]:
        """Generates embedding vector for a given text."""
        clean = (text or "").strip()
        if not clean:
            return [0.0] * self.dimension

        if clean in self._cache:
            return self._cache[clean]

        if not self.api_key:
            raise ValueError("Gemini API key is required for GeminiEmbeddings.")

        client = self._get_client()
        url = (
            f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:embedContent"
            f"?key={self.api_key}"
        )
        payload = {
            "model": f"models/{self.model}",
            "content": {"parts": [{"text": clean}]},
            "outputDimensionality": self.dimension,
        }

        # Up to 2 retries on rate limits
        for attempt in range(2):
            try:
                resp = await client.post(url, json=payload)
                if resp.status_code == 200:
                    data = resp.json()
                    values = data.get("embedding", {}).get("values", [])
                    if values:
                        self._cache[clean] = values
                        if len(self._cache) > 200:
                            self._cache.pop(next(iter(self._cache)))
                        return values
                elif resp.status_code == 429 and attempt == 0:
                    await asyncio.sleep(1.0)
                    continue
                else:
                    raise RuntimeError(f"Gemini embedding API error {resp.status_code}: {resp.text}")
            except Exception as exc:
                if attempt == 1:
                    raise exc
                await asyncio.sleep(0.5)

        return [0.0] * 768

    async def embed_query(self, query: str) -> list[float]:
        """Generates embedding vector for a search query."""
        return await self.embed_text(query)

    async def embed_batch(self, texts: list[str], batch_size: int = 20) -> list[list[float]]:
        """
        Generates dense vector embeddings for multiple texts using Gemini batchEmbedContents API.
        Includes automatic batch chunking, rate-limiting delays, and exponential backoff on HTTP 429.
        """
        if not texts:
            return []

        if not self.api_key:
            raise ValueError("Gemini API key is required for GeminiEmbeddings.")

        client = self._get_client()
        url = (
            f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:batchEmbedContents"
            f"?key={self.api_key}"
        )
        all_embeddings: list[list[float]] = []
        total_batches = (len(texts) + batch_size - 1) // batch_size

        for b_idx, i in enumerate(range(0, len(texts), batch_size)):
            chunk = texts[i : i + batch_size]
            requests_payload = [
                {
                    "model": f"models/{self.model}",
                    "content": {"parts": [{"text": t.strip() if t.strip() else " "}]},
                    "outputDimensionality": 768,
                }
                for t in chunk
            ]

            if b_idx > 0:
                await asyncio.sleep(2.0)

            max_retries = 6
            batch_success = False

            for attempt in range(max_retries):
                try:
                    resp = await client.post(url, json={"requests": requests_payload})
                    resp.raise_for_status()
                    data = resp.json()
                    embeddings_list = data.get("embeddings", [])
                    for emb in embeddings_list:
                        all_embeddings.append(emb.get("values", [0.0] * 768))
                    batch_success = True
                    break
                except HTTPStatusError as exc:
                    resp_obj = getattr(exc, "response", None)
                    status_code = getattr(resp_obj, "status_code", 0) if resp_obj is not None else 0
                    if status_code == 429 and attempt < max_retries - 1:
                        headers = getattr(resp_obj, "headers", {}) or {}
                        retry_after = headers.get("retry-after")
                        if retry_after:
                            try:
                                wait_time = float(retry_after) + 1.0
                            except ValueError:
                                wait_time = (2 ** attempt) * 3.0 + 3.0
                        else:
                            wait_time = (2 ** attempt) * 3.0 + 3.0
                        print(f"⏳ [GeminiEmbeddings] Rate limit 429 on batch {b_idx + 1}/{total_batches}, waiting {wait_time:.1f}s...")
                        await asyncio.sleep(wait_time)
                        continue
                    raise exc
                except Exception as exc:
                    if attempt < max_retries - 1:
                        wait_time = (2 ** attempt) * 2.5
                        print(f"⚠️ [GeminiEmbeddings] Network error on batch {b_idx + 1}/{total_batches} ({exc}), retrying in {wait_time:.1f}s...")
                        await asyncio.sleep(wait_time)
                        continue
                    raise exc

            if not batch_success:
                raise RuntimeError(f"Failed to generate embeddings for batch {b_idx + 1}/{total_batches} after {max_retries} retries.")

        return all_embeddings

    async def close(self) -> None:
        """Closes the underlying HTTP client."""
        if self._client and not self._client.is_closed:
            await self._client.aclose()
            self._client = None

close async 

close()

Closes the underlying HTTP client.

Source code in edgevoiceai/rag/embeddings/gemini.py 
182
183
184
185
186
async def close(self) -> None:
    """Closes the underlying HTTP client."""
    if self._client and not self._client.is_closed:
        await self._client.aclose()
        self._client = None

embed_batch async 

embed_batch(texts, batch_size=20)

Generates dense vector embeddings for multiple texts using Gemini batchEmbedContents API. Includes automatic batch chunking, rate-limiting delays, and exponential backoff on HTTP 429.

Source code in edgevoiceai/rag/embeddings/gemini.py 
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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
async def embed_batch(self, texts: list[str], batch_size: int = 20) -> list[list[float]]:
    """
    Generates dense vector embeddings for multiple texts using Gemini batchEmbedContents API.
    Includes automatic batch chunking, rate-limiting delays, and exponential backoff on HTTP 429.
    """
    if not texts:
        return []

    if not self.api_key:
        raise ValueError("Gemini API key is required for GeminiEmbeddings.")

    client = self._get_client()
    url = (
        f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:batchEmbedContents"
        f"?key={self.api_key}"
    )
    all_embeddings: list[list[float]] = []
    total_batches = (len(texts) + batch_size - 1) // batch_size

    for b_idx, i in enumerate(range(0, len(texts), batch_size)):
        chunk = texts[i : i + batch_size]
        requests_payload = [
            {
                "model": f"models/{self.model}",
                "content": {"parts": [{"text": t.strip() if t.strip() else " "}]},
                "outputDimensionality": 768,
            }
            for t in chunk
        ]

        if b_idx > 0:
            await asyncio.sleep(2.0)

        max_retries = 6
        batch_success = False

        for attempt in range(max_retries):
            try:
                resp = await client.post(url, json={"requests": requests_payload})
                resp.raise_for_status()
                data = resp.json()
                embeddings_list = data.get("embeddings", [])
                for emb in embeddings_list:
                    all_embeddings.append(emb.get("values", [0.0] * 768))
                batch_success = True
                break
            except HTTPStatusError as exc:
                resp_obj = getattr(exc, "response", None)
                status_code = getattr(resp_obj, "status_code", 0) if resp_obj is not None else 0
                if status_code == 429 and attempt < max_retries - 1:
                    headers = getattr(resp_obj, "headers", {}) or {}
                    retry_after = headers.get("retry-after")
                    if retry_after:
                        try:
                            wait_time = float(retry_after) + 1.0
                        except ValueError:
                            wait_time = (2 ** attempt) * 3.0 + 3.0
                    else:
                        wait_time = (2 ** attempt) * 3.0 + 3.0
                    print(f"⏳ [GeminiEmbeddings] Rate limit 429 on batch {b_idx + 1}/{total_batches}, waiting {wait_time:.1f}s...")
                    await asyncio.sleep(wait_time)
                    continue
                raise exc
            except Exception as exc:
                if attempt < max_retries - 1:
                    wait_time = (2 ** attempt) * 2.5
                    print(f"⚠️ [GeminiEmbeddings] Network error on batch {b_idx + 1}/{total_batches} ({exc}), retrying in {wait_time:.1f}s...")
                    await asyncio.sleep(wait_time)
                    continue
                raise exc

        if not batch_success:
            raise RuntimeError(f"Failed to generate embeddings for batch {b_idx + 1}/{total_batches} after {max_retries} retries.")

    return all_embeddings

embed_query async 

embed_query(query)

Generates embedding vector for a search query.

Source code in edgevoiceai/rag/embeddings/gemini.py 
102
103
104
async def embed_query(self, query: str) -> list[float]:
    """Generates embedding vector for a search query."""
    return await self.embed_text(query)

embed_text async 

embed_text(text)

Generates embedding vector for a given text.

Source code in edgevoiceai/rag/embeddings/gemini.py 
 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
async def embed_text(self, text: str) -> list[float]:
    """Generates embedding vector for a given text."""
    clean = (text or "").strip()
    if not clean:
        return [0.0] * self.dimension

    if clean in self._cache:
        return self._cache[clean]

    if not self.api_key:
        raise ValueError("Gemini API key is required for GeminiEmbeddings.")

    client = self._get_client()
    url = (
        f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:embedContent"
        f"?key={self.api_key}"
    )
    payload = {
        "model": f"models/{self.model}",
        "content": {"parts": [{"text": clean}]},
        "outputDimensionality": self.dimension,
    }

    # Up to 2 retries on rate limits
    for attempt in range(2):
        try:
            resp = await client.post(url, json=payload)
            if resp.status_code == 200:
                data = resp.json()
                values = data.get("embedding", {}).get("values", [])
                if values:
                    self._cache[clean] = values
                    if len(self._cache) > 200:
                        self._cache.pop(next(iter(self._cache)))
                    return values
            elif resp.status_code == 429 and attempt == 0:
                await asyncio.sleep(1.0)
                continue
            else:
                raise RuntimeError(f"Gemini embedding API error {resp.status_code}: {resp.text}")
        except Exception as exc:
            if attempt == 1:
                raise exc
            await asyncio.sleep(0.5)

    return [0.0] * 768