1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
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
|
import logging
import os
from typing import Any
import tiktoken
from openai import AsyncOpenAI, AuthenticationError, OpenAI
from openai._types import NOT_GIVEN
from core.base import (
ChunkSearchResult,
EmbeddingConfig,
EmbeddingProvider,
EmbeddingPurpose,
)
logger = logging.getLogger()
class OpenAIEmbeddingProvider(EmbeddingProvider):
MODEL_TO_TOKENIZER = {
"text-embedding-ada-002": "cl100k_base",
"text-embedding-3-small": "cl100k_base",
"text-embedding-3-large": "cl100k_base",
}
MODEL_TO_DIMENSIONS = {
"text-embedding-ada-002": [1536],
"text-embedding-3-small": [512, 1536],
"text-embedding-3-large": [256, 1024, 3072],
}
def __init__(self, config: EmbeddingConfig):
super().__init__(config)
if not config.provider:
raise ValueError(
"Must set provider in order to initialize OpenAIEmbeddingProvider."
)
if config.provider != "openai":
raise ValueError(
"OpenAIEmbeddingProvider must be initialized with provider `openai`."
)
if not os.getenv("OPENAI_API_KEY"):
raise ValueError(
"Must set OPENAI_API_KEY in order to initialize OpenAIEmbeddingProvider."
)
self.client = OpenAI()
self.async_client = AsyncOpenAI()
if config.rerank_model:
raise ValueError(
"OpenAIEmbeddingProvider does not support separate reranking."
)
if config.base_model and "openai/" in config.base_model:
self.base_model = config.base_model.split("/")[-1]
else:
self.base_model = config.base_model
self.base_dimension = config.base_dimension
if not self.base_model:
raise ValueError(
"Must set base_model in order to initialize OpenAIEmbeddingProvider."
)
if self.base_model not in OpenAIEmbeddingProvider.MODEL_TO_TOKENIZER:
raise ValueError(
f"OpenAI embedding model {self.base_model} not supported."
)
if self.base_dimension:
if (
self.base_dimension
not in OpenAIEmbeddingProvider.MODEL_TO_DIMENSIONS[
self.base_model
]
):
raise ValueError(
f"Dimensions {self.base_dimension} for {self.base_model} are not supported"
)
else:
# If base_dimension is not set, use the largest available dimension for the model
self.base_dimension = max(
OpenAIEmbeddingProvider.MODEL_TO_DIMENSIONS[self.base_model]
)
def _get_dimensions(self):
return (
NOT_GIVEN
if self.base_model == "text-embedding-ada-002"
else self.base_dimension
or OpenAIEmbeddingProvider.MODEL_TO_DIMENSIONS[self.base_model][-1]
)
def _get_embedding_kwargs(self, **kwargs):
return {
"model": self.base_model,
"dimensions": self._get_dimensions(),
} | kwargs
async def _execute_task(self, task: dict[str, Any]) -> list[list[float]]:
texts = task["texts"]
kwargs = self._get_embedding_kwargs(**task.get("kwargs", {}))
try:
response = await self.async_client.embeddings.create(
input=texts,
**kwargs,
)
return [data.embedding for data in response.data]
except AuthenticationError as e:
raise ValueError(
"Invalid OpenAI API key provided. Please check your OPENAI_API_KEY environment variable."
) from e
except Exception as e:
error_msg = f"Error getting embeddings: {str(e)}"
logger.error(error_msg)
raise ValueError(error_msg) from e
def _execute_task_sync(self, task: dict[str, Any]) -> list[list[float]]:
texts = task["texts"]
kwargs = self._get_embedding_kwargs(**task.get("kwargs", {}))
try:
response = self.client.embeddings.create(
input=texts,
**kwargs,
)
return [data.embedding for data in response.data]
except AuthenticationError as e:
raise ValueError(
"Invalid OpenAI API key provided. Please check your OPENAI_API_KEY environment variable."
) from e
except Exception as e:
error_msg = f"Error getting embeddings: {str(e)}"
logger.error(error_msg)
raise ValueError(error_msg) from e
async def async_get_embedding(
self,
text: str,
stage: EmbeddingProvider.Step = EmbeddingProvider.Step.BASE,
purpose: EmbeddingPurpose = EmbeddingPurpose.INDEX,
**kwargs,
) -> list[float]:
if stage != EmbeddingProvider.Step.BASE:
raise ValueError(
"OpenAIEmbeddingProvider only supports search stage."
)
task = {
"texts": [text],
"stage": stage,
"purpose": purpose,
"kwargs": kwargs,
}
result = await self._execute_with_backoff_async(task)
return result[0]
def get_embedding(
self,
text: str,
stage: EmbeddingProvider.Step = EmbeddingProvider.Step.BASE,
purpose: EmbeddingPurpose = EmbeddingPurpose.INDEX,
**kwargs,
) -> list[float]:
if stage != EmbeddingProvider.Step.BASE:
raise ValueError(
"OpenAIEmbeddingProvider only supports search stage."
)
task = {
"texts": [text],
"stage": stage,
"purpose": purpose,
"kwargs": kwargs,
}
result = self._execute_with_backoff_sync(task)
return result[0]
async def async_get_embeddings(
self,
texts: list[str],
stage: EmbeddingProvider.Step = EmbeddingProvider.Step.BASE,
purpose: EmbeddingPurpose = EmbeddingPurpose.INDEX,
**kwargs,
) -> list[list[float]]:
if stage != EmbeddingProvider.Step.BASE:
raise ValueError(
"OpenAIEmbeddingProvider only supports search stage."
)
task = {
"texts": texts,
"stage": stage,
"purpose": purpose,
"kwargs": kwargs,
}
return await self._execute_with_backoff_async(task)
def get_embeddings(
self,
texts: list[str],
stage: EmbeddingProvider.Step = EmbeddingProvider.Step.BASE,
purpose: EmbeddingPurpose = EmbeddingPurpose.INDEX,
**kwargs,
) -> list[list[float]]:
if stage != EmbeddingProvider.Step.BASE:
raise ValueError(
"OpenAIEmbeddingProvider only supports search stage."
)
task = {
"texts": texts,
"stage": stage,
"purpose": purpose,
"kwargs": kwargs,
}
return self._execute_with_backoff_sync(task)
def rerank(
self,
query: str,
results: list[ChunkSearchResult],
stage: EmbeddingProvider.Step = EmbeddingProvider.Step.RERANK,
limit: int = 10,
):
return results[:limit]
async def arerank(
self,
query: str,
results: list[ChunkSearchResult],
stage: EmbeddingProvider.Step = EmbeddingProvider.Step.RERANK,
limit: int = 10,
):
return results[:limit]
def tokenize_string(self, text: str, model: str) -> list[int]:
if model not in OpenAIEmbeddingProvider.MODEL_TO_TOKENIZER:
raise ValueError(f"OpenAI embedding model {model} not supported.")
encoding = tiktoken.get_encoding(
OpenAIEmbeddingProvider.MODEL_TO_TOKENIZER[model]
)
return encoding.encode(text)
|
