forked from openai/openai-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathengine.py
More file actions
50 lines (42 loc) · 1.63 KB
/
Copy pathengine.py
File metadata and controls
50 lines (42 loc) · 1.63 KB
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
import time
import warnings
from openai import util
from openai.api_resources.abstract import ListableAPIResource, UpdateableAPIResource
from openai.error import TryAgain
class Engine(ListableAPIResource, UpdateableAPIResource):
OBJECT_NAME = "engines"
def generate(self, timeout=None, **params):
start = time.time()
while True:
try:
return self.request(
"post",
self.instance_url() + "/generate",
params,
stream=params.get("stream"),
plain_old_data=True,
)
except TryAgain as e:
if timeout is not None and time.time() > start + timeout:
raise
util.log_info("Waiting for model to warm up", error=e)
async def agenerate(self, timeout=None, **params):
start = time.time()
while True:
try:
return await self.arequest(
"post",
self.instance_url() + "/generate",
params,
stream=params.get("stream"),
plain_old_data=True,
)
except TryAgain as e:
if timeout is not None and time.time() > start + timeout:
raise
util.log_info("Waiting for model to warm up", error=e)
def embeddings(self, **params):
warnings.warn(
"Engine.embeddings is deprecated, use Embedding.create", DeprecationWarning
)
return self.request("post", self.instance_url() + "/embeddings", params)