feat: chain of thought

src/gpt.py

@@ -4,7 +4,8 @@ from time import sleep
 import openai
 from openai.error import RateLimitError, Timeout
 
-from src.utils.string import print_colored
+from src.utils.io import timeout_generator_wrapper
+from src.utils.string_tools import print_colored
 
 openai.api_key = os.environ['OPENAI_API_KEY']
 
@@ -13,7 +14,7 @@ def get_response(system_definition, user_query):
     print_colored('user_query', user_query, 'blue')
     for i in range(10):
         try:
-            response = openai.ChatCompletion.create(
+            response_generator = openai.ChatCompletion.create(
                 temperature=0,
                 max_tokens=5_000,
                 model="gpt-4",
@@ -32,15 +33,17 @@ def get_response(system_definition, user_query):
 
                 ]
             )
+            response_generator_with_timeout = timeout_generator_wrapper(response_generator, 5)
+
             complete_string = ''
-            for chunk in response:
+            for chunk in response_generator_with_timeout:
                 delta = chunk['choices'][0]['delta']
                 if 'content' in delta:
                     content = delta['content']
                     print_colored('' if complete_string else 'Agent response:', content, 'green', end='')
                     complete_string += content
             return complete_string
-        except (RateLimitError, Timeout) as e:
+        except (RateLimitError, Timeout, ConnectionError) as e:
             print(e)
             print('retrying')
             sleep(3)

src/jina_cloud.py

@@ -1,4 +1,3 @@
-import asyncio
 import os
 from multiprocessing.connection import Client
 
@@ -9,8 +8,8 @@ from jina import Flow
 from src.constants import FLOW_URL_PLACEHOLDER
 
 
-def push_executor():
-    cmd = 'jina hub push executor/. --verbose'
+def push_executor(dir_path):
+    cmd = f'jina hub push {dir_path}/. --verbose'
     os.system(cmd)
 
 def get_user_name():
@@ -25,7 +24,7 @@ def deploy_on_jcloud(flow_yaml):
 
 
 
-def deploy_flow(executor_name, do_validation):
+def deploy_flow(executor_name, do_validation, dest_folder):
     flow = f'''
 jtype: Flow
 with:
@@ -47,7 +46,8 @@ executors:
         instance: C4
         capacity: spot
 '''
-    full_flow_path = os.path.join('executor', 'flow.yml')
+    full_flow_path = os.path.join(dest_folder,
+                     'flow.yml')
     with open(full_flow_path, 'w') as f:
         f.write(flow)
 

src/prompt_system.py

@@ -6,14 +6,16 @@ executor_example = "Here is an example of how an executor can be defined. It alw
 # this executor takes ... as input and returns ... as output
 # it processes each document in the following way: ...
 from jina import Executor, requests, DocumentArray, Document
-class MyExecutor(Executor):
+class MyInfoExecutor(Executor):
     def __init__(self, **kwargs):
         super().__init__()
 
     @requests
     def foo(self, docs: DocumentArray, **kwargs) => DocumentArray:
         for d in docs:
-            d.text = 'hello world'"
+            d.load_uri_to_blob()
+            d.tags['my_info'] = {'byte_length': len(d.blob)}
+            d.blob = None
         return docs
 '''
 "An executor gets a DocumentArray as input and returns a DocumentArray as output. "

src/prompt_tasks.py

@@ -73,7 +73,7 @@ def docker_file_task():
         "It is important to make sure that all libs are installed that are required by the python packages. "
         "Usually libraries are installed with apt-get. "
         "Add the config.yml file to the Dockerfile. "
-        "The base image of the Dockerfile is FROM jinaai/jina:3.14.2-dev18-py310-standard. "
+        "The base image of the Dockerfile is FROM jinaai/jina:3.14.1-py39-standard. "
         'The entrypoint is ENTRYPOINT ["jina", "executor", "--uses", "config.yml"] '
         "The Dockerfile runs the test during the build process. ",
         DOCKER_FILE_TAG,

src/utils/io.py

@@ -1,8 +1,43 @@
 import os
 import shutil
-
+import concurrent.futures
+import concurrent.futures
+from typing import Generator
 
 def recreate_folder(folder_path):
     if os.path.exists(folder_path) and os.path.isdir(folder_path):
         shutil.rmtree(folder_path)
     os.makedirs(folder_path)
+
+
+class GenerationTimeoutError(Exception):
+    pass
+
+def timeout_generator_wrapper(generator, timeout):
+    def generator_func():
+        for item in generator:
+            yield item
+
+    def wrapper() -> Generator:
+        gen = generator_func()
+        while True:
+            try:
+                with concurrent.futures.ThreadPoolExecutor() as executor:
+                    future = executor.submit(next, gen)
+                    yield future.result(timeout=timeout)
+            except StopIteration:
+                break
+            except concurrent.futures.TimeoutError:
+                raise GenerationTimeoutError(f"Generation took longer than {timeout} seconds")
+
+    return wrapper()
+
+# def my_generator():
+#     for i in range(10):
+#         sleep(3)
+#         yield 1
+#
+#
+# my_generator_with_timeout = timeout_generator_wrapper(my_generator, 2.9)
+# for item in my_generator_with_timeout():
+#     print(item)