aljaz/dev-gpt
1
0
Fork 0
mirror of https://github.com/aljazceru/dev-gpt.git synced 2026-01-04 14:24:21 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' of https://github.com/jina-ai/gptdeploy into feat-avoid-loop

Browse Source 
# Conflicts:
#	dev_gpt/options/generate/generator.py
This commit is contained in:
Joschka Braun
2023-05-05 11:45:59 +02:00
parent 649c64fe3f 3c39d88f0a
commit f1d4dba5f8
54 changed files with 172 additions and 443732 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
dev_gpt/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
__version__ = '0.18.37'
from dev_gpt.cli import main

0
dev_gpt/apis/__init__.py Normal file
View File

153
dev_gpt/apis/gpt.py Normal file
View File

@@ -0,0 +1,153 @@
import os
from copy import deepcopy
from time import sleep
from typing import List, Any
import openai
from langchain import PromptTemplate
from langchain.callbacks import CallbackManager
from langchain.chat_models import ChatOpenAI
from openai.error import RateLimitError
from langchain.schema import HumanMessage, SystemMessage, BaseMessage, AIMessage
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from requests.exceptions import ConnectionError, ChunkedEncodingError
from urllib3.exceptions import InvalidChunkLength
from dev_gpt.constants import PRICING_GPT4_PROMPT, PRICING_GPT4_GENERATION, PRICING_GPT3_5_TURBO_PROMPT, \
PRICING_GPT3_5_TURBO_GENERATION, CHARS_PER_TOKEN
from dev_gpt.options.generate.templates_system import template_system_message_base
from dev_gpt.utils.string_tools import print_colored
def configure_openai_api_key():
if 'OPENAI_API_KEY' not in os.environ:
print_colored('You need to set OPENAI_API_KEY in your environment.', '''
Run:
dev-gpt configure --key <your_openai_api_key>
If you have updated it already, please restart your terminal.
''', 'red')
exit(1)
openai.api_key = os.environ['OPENAI_API_KEY']
class GPTSession:
def __init__(self, task_description, model: str = 'gpt-4', ):
self.task_description = task_description
if model == 'gpt-4' and self.is_gpt4_available():
self.pricing_prompt = PRICING_GPT4_PROMPT
self.pricing_generation = PRICING_GPT4_GENERATION
else:
if model == 'gpt-4':
print_colored('GPT version info', 'GPT-4 is not available. Using GPT-3.5-turbo instead.', 'yellow')
model = 'gpt-3.5-turbo'
self.pricing_prompt = PRICING_GPT3_5_TURBO_PROMPT
self.pricing_generation = PRICING_GPT3_5_TURBO_GENERATION
self.model_name = model
self.chars_prompt_so_far = 0
self.chars_generation_so_far = 0
def get_conversation(self, messages: List[BaseMessage] = [], print_stream: bool = True, print_costs: bool = True):
messages = deepcopy(messages)
return _GPTConversation(
self.model_name, self.cost_callback, messages, print_stream, print_costs
)
@staticmethod
def is_gpt4_available():
try:
for i in range(5):
try:
openai.ChatCompletion.create(
model="gpt-4",
messages=[{
"role": 'system',
"content": 'you respond nothing'
}]
)
break
except RateLimitError:
sleep(1)
continue
return True
except openai.error.InvalidRequestError:
return False
def cost_callback(self, chars_prompt, chars_generation, print_costs: bool = True):
self.chars_prompt_so_far += chars_prompt
self.chars_generation_so_far += chars_generation
if print_costs:
print('\n')
money_prompt = self._calculate_money_spent(self.chars_prompt_so_far, self.pricing_prompt)
money_generation = self._calculate_money_spent(self.chars_generation_so_far, self.pricing_generation)
print('Total money spent so far on openai.com:', f'${money_prompt + money_generation:.3f}')
print('\n')
@staticmethod
def _calculate_money_spent(num_chars, price):
return round(num_chars / CHARS_PER_TOKEN * price / 1000, 3)
class AssistantStreamingStdOutCallbackHandler(StreamingStdOutCallbackHandler):
def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
"""Run on new LLM token. Only available when streaming is enabled."""
print_colored('', token, 'green', end='')
class _GPTConversation:
def __init__(self, model: str, cost_callback, messages: List[BaseMessage], print_stream, print_costs):
self._chat = ChatOpenAI(
model_name=model,
streaming=True,
callback_manager=CallbackManager([AssistantStreamingStdOutCallbackHandler()] if print_stream else []),
verbose=True,
temperature=0,
)
self.cost_callback = cost_callback
self.messages = messages
self.print_stream = print_stream
self.print_costs = print_costs
def print_messages(self, messages):
for i, message in enumerate(messages):
if os.environ['VERBOSE'].lower() == 'true':
if isinstance(message, SystemMessage):
print_colored(f'({i}) system - prompt', message.content, 'magenta')
elif isinstance(message, HumanMessage):
print_colored(f'({i}) user - prompt', message.content, 'blue')
elif isinstance(message, AIMessage):
print_colored(f'({i}) assistant - prompt', message.content, 'green')
def chat(self, prompt: str, role: str = 'user'):
MassageClass = HumanMessage if role == 'user' else SystemMessage
chat_message = MassageClass(content=prompt)
self.messages.append(chat_message)
self.print_messages(self.messages)
if self.print_stream:
print_colored('assistant', '', 'green', end='')
print('thinking...')
for i in range(10):
try:
response = self._chat(self.messages)
break
except (ConnectionError, InvalidChunkLength, ChunkedEncodingError) as e:
print('There was a connection error. Retrying...')
if i == 9:
raise e
sleep(10)
if os.environ['VERBOSE'].lower() == 'true':
print()
self.cost_callback(sum([len(m.content) for m in self.messages]), len(response.content), self.print_costs)
self.messages.append(response)
return response.content
@staticmethod
def _create_system_message(task_description, test_description) -> SystemMessage:
system_message = PromptTemplate.from_template(template_system_message_base).format(
task_description=task_description,
test_description=test_description,
)
return SystemMessage(content=system_message)

333
dev_gpt/apis/jina_cloud.py Normal file
View File

@@ -0,0 +1,333 @@
import hashlib
import json
import os
import re
import subprocess
import threading
import time
import webbrowser
from pathlib import Path
from typing import Dict
import click
import hubble
import requests
from hubble.executor.helper import upload_file, archive_package, get_full_version
from jcloud.flow import CloudFlow
from jina import Flow
from dev_gpt.constants import DEMO_TOKEN
from dev_gpt.utils.io import suppress_stdout, is_docker_running
from dev_gpt.utils.string_tools import print_colored
def wait_until_app_is_ready(url):
is_app_ready = False
while not is_app_ready:
try:
response = requests.get(url)
print('waiting for app to be ready...')
if response.status_code == 200:
is_app_ready = True
except requests.exceptions.RequestException:
pass
time.sleep(0.5)
def open_streamlit_app(host: str):
url = f"{host}/playground"
wait_until_app_is_ready(url)
webbrowser.open(url, new=2)
def redirect_callback(href):
print(
f'You need login to Jina first to use Dev GPT\n'
f'Please open this link if it does not open automatically in your browser: {href}'
)
webbrowser.open(href, new=0, autoraise=True)
def jina_auth_login():
try:
hubble.Client(jsonify=True).get_user_info(log_error=False)
except hubble.AuthenticationRequiredError:
print('You need login to Jina first to use dev-gpt')
print_colored('', '''
If you just created an account, it can happen that the login callback is not working.
In this case, please cancel this run, rerun your dev-gpt command and login into your account again.
''', 'green'
)
hubble.login(prompt='login', redirect_callback=redirect_callback)
def push_executor(dir_path):
for i in range(3):
try:
return _push_executor(dir_path)
except Exception as e:
if i == 2:
raise e
print(f'connection error - retrying in 5 seconds...')
time.sleep(5)
def get_request_header() -> Dict:
"""Return the header of request with an authorization token.
:return: request header
"""
metas, envs = get_full_version()
headers = {
**{f'jinameta-{k}': str(v) for k, v in metas.items()},
**envs,
}
headers['Authorization'] = f'token {DEMO_TOKEN}'
return headers
def _push_executor(dir_path):
dir_path = Path(dir_path)
md5_hash = hashlib.md5()
bytesio = archive_package(dir_path)
content = bytesio.getvalue()
md5_hash.update(content)
md5_digest = md5_hash.hexdigest()
form_data = {
'public': 'True',
'private': 'False',
'verbose': 'True',
'buildEnv': f'{{"OPENAI_API_KEY": "{os.environ["OPENAI_API_KEY"]}"}}',
'md5sum': md5_digest,
}
with suppress_stdout():
headers = get_request_header()
resp = upload_file(
'https://api.hubble.jina.ai/v2/rpc/executor.push',
'filename',
content,
dict_data=form_data,
headers=headers,
stream=False,
method='post',
)
json_lines_str = resp.content.decode('utf-8')
if 'AuthenticationRequiredWithBearerChallengeError' in json_lines_str:
raise Exception('The executor is not authorized to be pushed to Jina Cloud.')
if 'exited on non-zero code' not in json_lines_str:
return ''
responses = []
for json_line in json_lines_str.splitlines():
if 'exit code:' in json_line:
break
d = json.loads(json_line)
if 'payload' in d and type(d['payload']) == str:
responses.append(d['payload'])
elif type(d) == str:
responses.append(d)
return '\n'.join(responses)
def is_executor_in_hub(microservice_name):
url = f'https://api.hubble.jina.ai/v2/rpc/executor.list?search={microservice_name}&withAnonymous=true'
resp = requests.get(url)
executor_list = resp.json()['data']
for executor in executor_list:
if 'name' in executor and executor['name'] == microservice_name:
return True
return False
def get_user_name(token=None):
client = hubble.Client(max_retries=None, jsonify=True, token=token)
response = client.get_user_info()
return response['data']['name']
def _deploy_on_jcloud(flow_yaml):
cloud_flow = CloudFlow(path=flow_yaml)
return cloud_flow.__enter__().endpoints['gateway']
def deploy_on_jcloud(executor_name, microservice_path):
print('Deploy a jina flow')
full_flow_path = create_flow_yaml(microservice_path, executor_name, use_docker=True, use_custom_gateway=True)
for i in range(3):
try:
host = _deploy_on_jcloud(flow_yaml=full_flow_path)
break
except Exception as e:
print(f'Could not deploy on Jina Cloud. Trying again in 5 seconds. Error: {e}')
time.sleep(5)
except SystemExit as e:
raise SystemExit(f'''
Looks like you either ran out of credits or something went wrong in the generation and we didn't catch it.
To check if you ran out of credits, please go to https://cloud.jina.ai.
If you have credits left, please create an issue here https://github.com/jina-ai/dev-gpt/issues/new/choose
and add details on the microservice you are trying to create.
In that case, you can upgrade your Dev-GPT version, if not using latest, and try again.
''') from e
if i == 2:
raise Exception('''
Could not deploy on Jina Cloud.
This can happen when the microservice is buggy, if it requires too much memory or if the Jina Cloud is overloaded.
Please try again later.
'''
)
print(f'''
Your Microservice is deployed at {host} and the playground is available at {host}/playground
We open now the playground in your browser.
''')
open_streamlit_app(host)
return host
def run_streamlit_app(app_path):
subprocess.run(['streamlit', 'run', app_path, 'server.address', '0.0.0.0', '--server.port', '8081'])
def run_locally(executor_name, microservice_version_path):
if is_docker_running():
use_docker = True
else:
click.echo('''
Docker daemon doesn\'t seem to be running (possible reasons: incorrect docker installation, docker command isn\'t in system path, insufficient permissions, docker is running but unrespnsive).
It might be important to run your microservice within a docker container.
Your machine might not have all the dependencies installed.
You have 3 options:
a) start the docker daemon
b) run dev-gpt deploy... to deploy your microservice on Jina Cloud. All dependencies will be installed there.
c) try to run your microservice locally without docker. It is worth a try but might fail.
'''
)
user_input = click.prompt('Do you want to run your microservice locally without docker? (Y/n)', type=str, default='y')
if user_input.lower() != 'y':
exit(1)
use_docker = False
print('Run a jina flow locally')
full_flow_path = create_flow_yaml(microservice_version_path, executor_name, use_docker, False)
flow = Flow.load_config(full_flow_path)
with flow:
print(f'''
Your microservice started locally.
We now start the playground for you.
''')
app_path = os.path.join(microservice_version_path, 'gateway', "app.py")
# Run the Streamlit app in a separate thread
streamlit_thread = threading.Thread(target=run_streamlit_app, args=(app_path,))
streamlit_thread.start()
# Open the Streamlit app in the user's default web browser
open_streamlit_app(host='http://localhost:8081')
flow.block()
def create_flow_yaml(dest_folder, executor_name, use_docker, use_custom_gateway):
if use_docker:
prefix = 'jinaai+docker'
else:
prefix = 'jinaai'
flow = f'''jtype: Flow
with:
port: 8080
protocol: http
jcloud:
version: 3.15.1.dev14
labels:
creator: microchain
name: gptdeploy
gateway:
{f"uses: {prefix}://{get_user_name(DEMO_TOKEN)}/Gateway{executor_name}:latest" if use_custom_gateway else ""}
{"" if use_docker else "install-requirements: True"}
executors:
- name: {executor_name.lower()}
uses: {prefix}://{get_user_name(DEMO_TOKEN)}/{executor_name}:latest
{"" if use_docker else "install-requirements: True"}
env:
OPENAI_API_KEY: {os.environ['OPENAI_API_KEY']}
jcloud:
resources:
instance: C2
capacity: spot
'''
full_flow_path = os.path.join(dest_folder,
'flow.yml')
with open(full_flow_path, 'w', encoding='utf-8') as f:
f.write(flow)
return full_flow_path
def replace_client_line(file_content: str, replacement: str) -> str:
lines = file_content.split('\n')
for index, line in enumerate(lines):
if 'Client(' in line:
lines[index] = replacement
break
return '\n'.join(lines)
def update_client_line_in_file(file_path, host):
with open(file_path, 'r', encoding='utf-8') as file:
content = file.read()
replaced_content = replace_client_line(content, f"client = Client(host='{host}')")
with open(file_path, 'w', encoding='utf-8') as file:
file.write(replaced_content)
def shorten_logs(relevant_lines):
# handle duplicate error messages
for index, line in enumerate(relevant_lines):
if '--- Captured stderr call ----' in line:
relevant_lines = relevant_lines[:index]
# filter pip install logs
relevant_lines = [line for line in relevant_lines if ' Requirement already satisfied: ' not in line]
# filter version not found logs
for index, line in enumerate(relevant_lines):
if 'ERROR: Could not find a version that satisfies the requirement ' in line:
start_and_end = line[:150] + '...' + line[-150:]
relevant_lines[index] = start_and_end
return relevant_lines
def clean_color_codes(response):
response = re.sub(r'\x1b\[[0-9;]*m', '', response)
return response
def process_error_message(error_message):
lines = error_message.split('\n')
relevant_lines = []
pattern = re.compile(r"^#\d+ \[[ \d]+/[ \d]+\]") # Pattern to match lines like "#11 [7/8]"
last_matching_line_index = None
for index, line in enumerate(lines):
if pattern.match(line):
last_matching_line_index = index
if last_matching_line_index is not None:
relevant_lines = lines[last_matching_line_index:]
relevant_lines = shorten_logs(relevant_lines)
response = '\n'.join(relevant_lines[-100:]).strip()
response = clean_color_codes(response)
# the following code tests the case that the docker file is corrupted and can not be parsed
# the method above will not return a relevant error message in this case
# but the last line of the error message will start with "error"
last_line = lines[-1]
if not response and last_line.startswith('error: '):
return last_line
return response

87
dev_gpt/apis/pypi.py Normal file
View File

@@ -0,0 +1,87 @@
import os
import re
from datetime import datetime
import requests
from packaging import version
def is_package_on_pypi(package_name, version=None):
"""
Returns True if the package is on PyPI, False if it is not, and None if the status code is not 200 or 404.
"""
optional_version = f"/{version}" if version else ""
url = f"https://pypi.org/pypi/{package_name}{optional_version}/json"
response = requests.get(url)
if response.status_code == 200:
return True
elif response.status_code == 404:
return False
else:
return None
def get_latest_package_version(package_name):
"""
Returns the latest version of a package that is not older than 2021.
"""
url = f'https://pypi.org/pypi/{package_name}/json'
response = requests.get(url)
if response.status_code != 200:
return None
data = response.json()
releases = data['releases']
# Get package versions not older than 2021
valid_versions = []
for v, release_info in releases.items():
if not release_info:
continue
upload_time = datetime.strptime(release_info[0]['upload_time'], '%Y-%m-%dT%H:%M:%S')
if upload_time.year <= 2020 or (upload_time.year == 2021 and upload_time.month <= 9): # knowledge cutoff 2021-09 (including september)
valid_versions.append(v)
v = max(valid_versions, key=version.parse) if valid_versions else None
return v
def clean_requirements_txt(previous_microservice_path):
"""
It can happen that the generated requirements.txt contains packages that are not on PyPI (like base64).
In this case, we remove the requirement from requirements.txt.
In case the package is on PyPI, but the version is not, we update the version to the latest version that is still not older than 2021.
"""
requirements_txt_path = os.path.join(previous_microservice_path, 'requirements.txt')
with open(requirements_txt_path, 'r', encoding='utf-8') as f:
requirements_txt = f.read()
updated_requirements = []
for line in requirements_txt.split('\n'):
line = line.strip()
if not line or line.startswith('#'):
continue
split = re.split(r'==|>=|<=|>|<|~=', line)
if len(split) == 1 or len(split) > 2:
version = None
package_name = split[0]
else:
package_name, version = split
# Keep lines with jina, docarray, openai, pytest unchanged
if package_name in {'jina', 'docarray', 'openai', 'pytest'}:
updated_requirements.append(line)
continue
if is_package_on_pypi(package_name):
if version is None or not is_package_on_pypi(package_name, version):
latest_version = get_latest_package_version(package_name)
if latest_version is None:
continue
updated_requirements.append(f'{package_name}~={latest_version}')
else:
updated_requirements.append(line)
with open(requirements_txt_path, 'w', encoding='utf-8') as f:
f.write('\n'.join(updated_requirements))

101
dev_gpt/cli.py Normal file
View File

@@ -0,0 +1,101 @@
from dev_gpt import env # noqa: F401 to make sure certain environment variables are set
import functools
import os
import click
from dev_gpt.apis.gpt import configure_openai_api_key
from dev_gpt.apis.jina_cloud import jina_auth_login
from dev_gpt.options.configure.key_handling import set_api_key
def openai_api_key_needed(func):
def wrapper(*args, **kwargs):
configure_openai_api_key()
return func(*args, **kwargs)
return wrapper
def exception_interceptor(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
raise type(e)(f'''
{str(e)}
😱😱😱 Sorry for this experience.
Could you please report an issue about this on our github repo? We'll try to fix it asap.
https://github.com/jina-ai/dev-gpt/issues/new
''') from e
return wrapper
def path_param(func):
@click.option('--path', default='microservice', help='Path to the generated microservice.')
@functools.wraps(func)
def wrapper(*args, **kwargs):
path = os.path.expanduser(kwargs['path'])
path = os.path.abspath(path)
kwargs['path'] = path
return func(*args, **kwargs)
return wrapper
@click.group(invoke_without_command=True)
@click.pass_context
@exception_interceptor
def main(ctx):
if ctx.invoked_subcommand is None:
click.echo(ctx.get_help())
@openai_api_key_needed
@main.command()
@click.option('--description', required=False, help='Description of the microservice.')
@click.option('--model', default='gpt-4', help='GPT model to use (default: gpt-4).')
@click.option('--verbose', default=False, is_flag=True, help='Verbose mode.') # only for development
@path_param
def generate(
description,
model,
verbose,
path,
):
os.environ['VERBOSE'] = str(verbose)
path = os.path.expanduser(path)
path = os.path.abspath(path)
if os.path.exists(path):
if os.listdir(path):
click.echo(f"Error: The path {path} you provided via --path is not empty. Please choose a directory that does not exist or is empty.")
return
from dev_gpt.options.generate.generator import Generator
generator = Generator(description, path=path, model=model)
generator.generate()
@openai_api_key_needed
@main.command()
@path_param
def run(path):
from dev_gpt.options.run import Runner
path = os.path.expanduser(path)
path = os.path.abspath(path)
Runner().run(path)
@openai_api_key_needed
@main.command()
@path_param
def deploy(path):
jina_auth_login()
from dev_gpt.options.deploy.deployer import Deployer
path = os.path.expanduser(path)
path = os.path.abspath(path)
Deployer().deploy(path)
@main.command()
@click.option('--key', required=True, help='Your OpenAI API key.')
def configure(key):
set_api_key(key)
if __name__ == '__main__':
main()

56
dev_gpt/constants.py Normal file
View File

@@ -0,0 +1,56 @@
DOCKER_BASE_IMAGE_VERSION = '0.0.6'
EXECUTOR_FILE_NAME = '__init__.py'
IMPLEMENTATION_FILE_NAME = 'microservice.py'
TEST_EXECUTOR_FILE_NAME = 'test_microservice.py'
REQUIREMENTS_FILE_NAME = 'requirements.txt'
DOCKER_FILE_NAME = 'Dockerfile'
CLIENT_FILE_NAME = 'client.py'
STREAMLIT_FILE_NAME = 'streamlit.py'
EXECUTOR_FILE_TAG = 'python'
IMPLEMENTATION_FILE_TAG = 'python'
TEST_EXECUTOR_FILE_TAG = 'python'
REQUIREMENTS_FILE_TAG = ''
DOCKER_FILE_TAG = 'dockerfile'
CLIENT_FILE_TAG = 'python'
STREAMLIT_FILE_TAG = 'python'
FILE_AND_TAG_PAIRS = [
(EXECUTOR_FILE_NAME, EXECUTOR_FILE_TAG),
(IMPLEMENTATION_FILE_NAME, IMPLEMENTATION_FILE_TAG),
(TEST_EXECUTOR_FILE_NAME, TEST_EXECUTOR_FILE_TAG),
(REQUIREMENTS_FILE_NAME, REQUIREMENTS_FILE_TAG),
(DOCKER_FILE_NAME, DOCKER_FILE_TAG),
(CLIENT_FILE_NAME, CLIENT_FILE_TAG),
(STREAMLIT_FILE_NAME, STREAMLIT_FILE_TAG)
]
FLOW_URL_PLACEHOLDER = 'jcloud.jina.ai'
PRICING_GPT4_PROMPT = 0.03
PRICING_GPT4_GENERATION = 0.06
PRICING_GPT3_5_TURBO_PROMPT = 0.002
PRICING_GPT3_5_TURBO_GENERATION = 0.002
CHARS_PER_TOKEN = 3.4
NUM_IMPLEMENTATION_STRATEGIES = 5
MAX_DEBUGGING_ITERATIONS = 10
DEMO_TOKEN = '45372338e04f5a41af949024db929d46'
BLACKLISTED_PACKAGES = [
'moderngl', 'pyopengl', 'pyglet', 'pythreejs', 'panda3d', # because they need a screen,
'tika', # because it needs java
]
UNNECESSARY_PACKAGES = [
'fastapi', 'uvicorn', 'starlette' # because the wrappers are used instead
]
LANGUAGE_PACKAGES = [
'allennlp', 'bertopic', 'fasttext', 'flair', 'gensim', 'nltk',
'pattern', 'polyglot', 'pytorch-transformers', 'rasa', 'sentence-transformers',
'spacy', 'stanza', 'summarizer', 'sumy', 'textblob', 'textstat', 'transformers'
]

5
dev_gpt/env.py Normal file
View File

@@ -0,0 +1,5 @@
import os
os.environ['PYTHONIOENCODING'] = 'utf-8'
os.environ['PYTHONLEGACYWINDOWSSTDIO'] = 'utf-8'
os.environ['PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION'] = 'python' # because protobuf issues on windows

58
dev_gpt/options/__init__.py Normal file
View File

@@ -0,0 +1,58 @@
import os
from dev_gpt.constants import REQUIREMENTS_FILE_NAME, DOCKER_FILE_NAME, IMPLEMENTATION_FILE_NAME, TEST_EXECUTOR_FILE_NAME
def list_dirs_no_hidden(path):
"""
List all non-hidden directories in the specified path.
:param path: str, optional (default is '.')
The path to the directory you want to list files and directories from.
:return: list
A list of directory names that are not hidden.
"""
return [entry for entry in os.listdir(path) if not entry.startswith('.') and os.path.isdir(os.path.join(path, entry))]
def get_latest_folder(path, max_fn=max):
return max_fn([os.path.join(path, f) for f in list_dirs_no_hidden(path)])
def version_max_fn(path_list):
version_list = [int(os.path.split(path)[-1].replace('v', '')) for path in path_list]
max_version = max(version_list)
max_index = version_list.index(max_version)
return path_list[max_index]
def get_latest_version_path(microservice_path):
executor_name_path = get_latest_folder(microservice_path)
latest_approach_path = get_latest_folder(executor_name_path)
latest_version_path = get_latest_folder(latest_approach_path, max_fn=version_max_fn)
return latest_version_path
def get_executor_name(microservice_path):
latest_folder = get_latest_folder(microservice_path)
return os.path.split(latest_folder)[-1]
def validate_folder_is_correct(microservice_path):
if not os.path.exists(microservice_path):
raise ValueError(f'Path {microservice_path} does not exist')
if not os.path.isdir(microservice_path):
raise ValueError(f'Path {microservice_path} is not a directory')
if len(list_dirs_no_hidden(microservice_path)) == 0:
raise ValueError(f'Path {microservice_path} is empty. Please generate a microservice first. Type `dev-gpt generate` for further instructions.')
if len(list_dirs_no_hidden(microservice_path)) > 1:
raise ValueError(f'Path {microservice_path} needs to contain only one folder. Please make sure that you only have one microservice in this folder.')
latest_version_path = get_latest_version_path(microservice_path)
required_files = [
'gateway/app.py',
REQUIREMENTS_FILE_NAME,
DOCKER_FILE_NAME,
IMPLEMENTATION_FILE_NAME,
TEST_EXECUTOR_FILE_NAME,
'config.yml',
]
for file_name in required_files:
if not os.path.exists(os.path.join(latest_version_path, file_name)):
raise ValueError(f'Path {latest_version_path} needs to contain a file named {file_name}')

0
dev_gpt/options/configure/__init__.py Normal file
View File

140
dev_gpt/options/configure/key_handling.py Normal file
View File

@@ -0,0 +1,140 @@
import os
import platform
import re
import subprocess
import click
try:
import psutil
except ImportError:
psutil = None
def get_shell():
if psutil is None:
return None
shell_names = ["bash", "zsh", "sh", "fish", "csh", "tcsh", "ksh", "dash"]
# Check the SHELL environment variable first
shell_env = os.environ.get('SHELL')
if shell_env:
shell_name = os.path.basename(shell_env)
if shell_name in shell_names:
return shell_name
# Fallback to traversing the process tree
try:
p = psutil.Process(os.getpid())
# Traverse the process tree
while p.parent():
p = p.parent()
if p.name() in shell_names:
return p.name()
return None
except Exception as e:
click.echo(f"Error detecting shell: {e}")
return None
def get_shell_config(key):
return {
"bash": {"config_file": "~/.bashrc", "export_line": f"export OPENAI_API_KEY={key}"},
"zsh": {"config_file": "~/.zshrc", "export_line": f"export OPENAI_API_KEY={key}"},
"sh": {"config_file": "~/.profile", "export_line": f"export OPENAI_API_KEY={key}"},
"fish": {
"config_file": "~/.config/fish/config.fish",
"export_line": f"set -gx OPENAI_API_KEY {key}",
},
"csh": {"config_file": "~/.cshrc", "export_line": f"setenv OPENAI_API_KEY {key}"},
"tcsh": {"config_file": "~/.tcshrc", "export_line": f"setenv OPENAI_API_KEY {key}"},
"ksh": {"config_file": "~/.kshrc", "export_line": f"export OPENAI_API_KEY={key}"},
"dash": {"config_file": "~/.profile", "export_line": f"export OPENAI_API_KEY={key}"}
}
def set_env_variable(shell, key):
shell_config = get_shell_config(key)
if shell not in shell_config:
click.echo("Sorry, your shell is not supported. Please add the key OPENAI_API_KEY manually.")
return
config_file = os.path.expanduser(shell_config[shell]["config_file"])
try:
with open(config_file, "r", encoding='utf-8') as file:
content = file.read()
export_line = shell_config[shell]['export_line']
# Update the existing API key if it exists, otherwise append it to the config file
if f"OPENAI_API_KEY" in content:
content = re.sub(r'OPENAI_API_KEY=.*', f'OPENAI_API_KEY={key}', content, flags=re.MULTILINE)
with open(config_file, "w", encoding='utf-8') as file:
file.write(content)
else:
with open(config_file, "a", encoding='utf-8') as file:
file.write(f"\n{export_line}\n")
click.echo(f'''
✅ Success, OPENAI_API_KEY has been set in {config_file}.
Please restart your shell to apply the changes or run:
source {config_file}
'''
)
except FileNotFoundError:
click.echo(f"Error: {config_file} not found. Please set the environment variable manually.")
def set_api_key(key):
system_platform = platform.system().lower()
if system_platform == "windows":
set_env_variable_command = f'setx OPENAI_API_KEY "{key}"'
subprocess.call(set_env_variable_command, shell=True)
click.echo('''
✅ Success, OPENAI_API_KEY has been set.
Please restart your Command Prompt to apply the changes.
'''
)
elif system_platform in ["linux", "darwin"]:
if "OPENAI_API_KEY" in os.environ or is_key_set_in_config_file(key):
if not click.confirm("OPENAI_API_KEY is already set. Do you want to overwrite it?"):
click.echo("Aborted.")
return
shell = get_shell()
if shell is None:
click.echo(
"Error: Unable to detect your shell or psutil is not available. Please set the environment variable manually.")
return
set_env_variable(shell, key)
else:
click.echo("Sorry, this platform is not supported.")
def is_key_set_in_config_file(key):
shell = get_shell()
if shell is None:
return False
shell_config = get_shell_config(key)
config_file = os.path.expanduser(shell_config[shell]["config_file"])
try:
with open(config_file, "r", encoding='utf-8') as file:
content = file.read()
if f"OPENAI_API_KEY" in content:
return True
except FileNotFoundError:
pass
return False

0
dev_gpt/options/deploy/__init__.py Normal file
View File

10
dev_gpt/options/deploy/deployer.py Normal file
View File

@@ -0,0 +1,10 @@
from dev_gpt.apis.jina_cloud import deploy_on_jcloud
from dev_gpt.options import validate_folder_is_correct, get_executor_name, get_latest_version_path
class Deployer:
def deploy(self, microservice_path):
validate_folder_is_correct(microservice_path)
executor_name = get_executor_name(microservice_path)
latest_version_path = get_latest_version_path(microservice_path)
deploy_on_jcloud(executor_name, latest_version_path)

0
dev_gpt/options/generate/__init__.py Normal file
View File

637
dev_gpt/options/generate/generator.py Normal file
View File

@@ -0,0 +1,637 @@
import json
import os
import random
import re
import shutil
from typing import Callable
from typing import List, Text, Optional
from langchain import PromptTemplate
from langchain.schema import SystemMessage, HumanMessage, AIMessage
from pydantic.dataclasses import dataclass
from dev_gpt.apis import gpt
from dev_gpt.apis.gpt import _GPTConversation
from dev_gpt.apis.jina_cloud import process_error_message, push_executor, is_executor_in_hub
from dev_gpt.apis.pypi import is_package_on_pypi, get_latest_package_version, clean_requirements_txt
from dev_gpt.constants import FILE_AND_TAG_PAIRS, NUM_IMPLEMENTATION_STRATEGIES, MAX_DEBUGGING_ITERATIONS, \
BLACKLISTED_PACKAGES, EXECUTOR_FILE_NAME, TEST_EXECUTOR_FILE_NAME, TEST_EXECUTOR_FILE_TAG, \
REQUIREMENTS_FILE_NAME, REQUIREMENTS_FILE_TAG, DOCKER_FILE_NAME, IMPLEMENTATION_FILE_NAME, \
IMPLEMENTATION_FILE_TAG, LANGUAGE_PACKAGES, UNNECESSARY_PACKAGES, DOCKER_BASE_IMAGE_VERSION
from dev_gpt.options.generate.templates_system import system_task_iteration, system_task_introduction, system_test_iteration
from dev_gpt.options.generate.templates_user import template_generate_microservice_name, \
template_generate_possible_packages, \
template_implement_solution_code_issue, \
template_solve_pip_dependency_issue, template_is_dependency_issue, template_generate_playground, \
template_generate_function, template_generate_test, template_generate_requirements, \
template_chain_of_thought, template_summarize_error, \
template_solve_apt_get_dependency_issue, template_pm_task_iteration, \
template_pm_test_iteration
from dev_gpt.options.generate.ui import get_random_employee
from dev_gpt.utils.io import persist_file, get_all_microservice_files_with_content, get_microservice_path
from dev_gpt.utils.string_tools import print_colored
@dataclass
class TaskSpecification:
task: Optional[Text]
test: Optional[Text]
class Generator:
def __init__(self, task_description, path, model='gpt-4'):
self.gpt_session = gpt.GPTSession(task_description, model=model)
self.microservice_specification = TaskSpecification(task=task_description, test=None)
self.microservice_root_path = path
self.microservice_name = None
self.previous_microservice_path = None
self.cur_microservice_path = None
self.previous_errors = []
self.previous_solutions = []
@staticmethod
def extract_content_from_result(plain_text, file_name, match_single_block=False, can_contain_code_block=True):
optional_line_break = '\n' if can_contain_code_block else '' # the \n at the end makes sure that ``` within the generated code is not matched because it is not right before a line break
pattern = fr"(?:\*|\*\*| ){file_name}\*?\*?\n```(?:\w+\n)?([\s\S]*?){optional_line_break}```"
matches = re.findall(pattern, plain_text, re.MULTILINE)
if matches:
return matches[-1].strip()
elif match_single_block:
# Check for a single code block
single_code_block_pattern = r"```(?:\w+\n)?([\s\S]*?)```"
single_code_block_match = re.findall(single_code_block_pattern, plain_text, re.MULTILINE)
if len(single_code_block_match) == 1:
return single_code_block_match[0].strip()
return ''
def write_config_yml(self, class_name, dest_folder, python_file=EXECUTOR_FILE_NAME):
config_content = f'''jtype: {class_name}
py_modules:
- {python_file}
metas:
name: {class_name}
'''
with open(os.path.join(dest_folder, 'config.yml'), 'w', encoding='utf-8') as f:
f.write(config_content)
def files_to_string(self, file_name_to_content, restrict_keys=None):
all_microservice_files_string = ''
for file_name, tag in FILE_AND_TAG_PAIRS:
if file_name in file_name_to_content and (not restrict_keys or file_name in restrict_keys):
all_microservice_files_string += f'**{file_name}**\n```{tag}\n{file_name_to_content[file_name]}\n```\n\n'
return all_microservice_files_string.strip()
def get_default_parse_result_fn(self, files_names: List[str]):
def _default_parse_result_fn(x):
_parsed_results = {}
for _file_name in files_names:
_content = self.extract_content_from_result(x, _file_name, match_single_block=len(files_names) == 1)
if _content != '':
_parsed_results[_file_name] = _content
return _parsed_results
return _default_parse_result_fn
def generate_and_persist_file(
self,
section_title: str,
template: PromptTemplate,
destination_folder: str = None,
file_name_s: List[str] = None,
parse_result_fn: Callable = None,
system_definition_examples: List[str] = [], # todo: rename to use_system_definition_exampels: bool = True
**template_kwargs
):
"""This function generates file(s) using the given template and persists it/them in the given destination folder.
It also returns the generated content as a dictionary mapping file_name to its content.
Args:
section_title (str): The title of the section to be printed in the console.
template (PromptTemplate): The template to be used for generating the file(s).
destination_folder (str): The destination folder where the generated file(s) should be persisted. If None,
the current microservice path is used. Defaults to None.
file_name_s (List[str], optional): The name of the file(s) to be generated. Defaults to None.
parse_result_fn (Callable, optional): A function that parses the generated content and returns a dictionary
mapping file_name to its content. If no content could be extract, it returns an empty dictionary.
Defaults to None. If None, default parsing is used which uses the file_name to extract from the generated content.
system_definition_examples (List[str], optional): The system definition examples to be used for the conversation. Defaults to [].
**template_kwargs: The keyword arguments to be passed to the template.
"""
if destination_folder is None:
destination_folder = self.cur_microservice_path
if parse_result_fn is None:
parse_result_fn = self.get_default_parse_result_fn(file_name_s)
print_colored('', f'\n\n############# {section_title} #############', 'blue')
system_introduction_message = _GPTConversation._create_system_message(
self.microservice_specification.task,
self.microservice_specification.test
)
conversation = self.gpt_session.get_conversation(messages=[system_introduction_message] if use_system_message else [])
template_kwargs = {k: v for k, v in template_kwargs.items() if k in template.input_variables}
if 'file_name' in template.input_variables and len(file_name_s) == 1:
template_kwargs['file_name'] = file_name_s[0]
content_raw = conversation.chat(
template.format(
**template_kwargs
)
)
content = parse_result_fn(content_raw)
if content == {}:
content_raw = conversation.chat(
'You must add the content' + (f' for {file_name_s[0]}' if len(file_name_s) == 1 else ''))
content = parse_result_fn(content_raw)
for _file_name, _file_content in content.items():
persist_file(_file_content, os.path.join(destination_folder, _file_name))
return content
def generate_microservice(
self,
packages,
num_approach,
):
self.cur_microservice_path = get_microservice_path(
self.microservice_root_path, self.microservice_name, packages, num_approach, 1
)
os.makedirs(self.cur_microservice_path)
with open(os.path.join(os.path.dirname(__file__), 'static_files', 'microservice', 'jina_wrapper.py'), 'r', encoding='utf-8') as f:
microservice_executor_boilerplate = f.read()
microservice_executor_code = microservice_executor_boilerplate.replace('class DevGPTExecutor(Executor):',
f'class {microservice_name}(Executor):')
persist_file(microservice_executor_code, os.path.join(self.cur_microservice_path, EXECUTOR_FILE_NAME))
with open(os.path.join(os.path.dirname(__file__), 'static_files', 'microservice', 'apis.py'), 'r', encoding='utf-8') as f:
persist_file(f.read(), os.path.join(self.cur_microservice_path, 'apis.py'))
microservice_content = self.generate_and_persist_file(
section_title='Microservice',
template=template_generate_function,
microservice_description=self.microservice_specification.task,
test_description=self.microservice_specification.test,
packages=packages,
file_name_purpose=IMPLEMENTATION_FILE_NAME,
tag_name=IMPLEMENTATION_FILE_TAG,
file_name_s=[IMPLEMENTATION_FILE_NAME],
)[IMPLEMENTATION_FILE_NAME]
test_microservice_content = self.generate_and_persist_file(
'Test Microservice',
template_generate_test,
code_files_wrapped=self.files_to_string({EXECUTOR_FILE_NAME: microservice_content}),
microservice_name=self.microservice_name,
microservice_description=self.microservice_specification.task,
test_description=self.microservice_specification.test,
file_name_purpose=TEST_EXECUTOR_FILE_NAME,
tag_name=TEST_EXECUTOR_FILE_TAG,
file_name_s=[TEST_EXECUTOR_FILE_NAME],
)[TEST_EXECUTOR_FILE_NAME]
self.generate_and_persist_file(
'Requirements',
template_generate_requirements,
code_files_wrapped=self.files_to_string({
IMPLEMENTATION_FILE_NAME: microservice_content,
TEST_EXECUTOR_FILE_NAME: test_microservice_content,
}),
file_name_purpose=REQUIREMENTS_FILE_NAME,
file_name_s=[REQUIREMENTS_FILE_NAME],
parse_result_fn=self.parse_result_fn_requirements,
tag_name=REQUIREMENTS_FILE_TAG,
)
with open(os.path.join(os.path.dirname(__file__), 'static_files', 'microservice', 'Dockerfile'), 'r',
encoding='utf-8') as f:
docker_file_template_lines = f.readlines()
docker_file_template_lines = [
line.replace('{{APT_GET_PACKAGES}}', '').replace('{{DOCKER_BASE_IMAGE_VERSION}}', DOCKER_BASE_IMAGE_VERSION)
for line in docker_file_template_lines
]
docker_file_content = '\n'.join(docker_file_template_lines)
persist_file(docker_file_content, os.path.join(self.cur_microservice_path, 'Dockerfile'))
self.write_config_yml(self.microservice_name, self.cur_microservice_path)
print('\nFirst version of the microservice generated. Start iterating on it to make the tests pass...')
@staticmethod
def read_docker_template():
with open(os.path.join(os.path.dirname(__file__), 'static_files', 'microservice', 'Dockerfile'), 'r', encoding='utf-8') as f:
return f.read()
def parse_result_fn_dockerfile(self, content_raw: str):
json_string = self.extract_content_from_result(content_raw, 'apt-get-packages.json', match_single_block=True)
packages = ' '.join(json.loads(json_string)['packages'])
docker_file_template = self.read_docker_template()
return {DOCKER_FILE_NAME: docker_file_template.replace('{{APT_GET_PACKAGES}}', '{APT_GET_PACKAGES}').replace('{{DOCKER_BASE_IMAGE_VERSION}}', DOCKER_BASE_IMAGE_VERSION).format(
APT_GET_PACKAGES=packages)}
def parse_result_fn_requirements(self, content_raw: str):
content_parsed = self.extract_content_from_result(content_raw, 'requirements.txt', match_single_block=True)
lines = content_parsed.split('\n')
lines = [line for line in lines if
not any([pkg in line for pkg in ['jina', 'docarray', 'openai', 'pytest', 'gpt_3_5_turbo']])]
content_modified = f'''jina==3.15.1.dev14
docarray==0.21.0
openai==0.27.5
pytest
{os.linesep.join(lines)}'''
return {REQUIREMENTS_FILE_NAME: content_modified}
def generate_playground(self):
print_colored('', '\n\n############# Playground #############', 'blue')
file_name_to_content = get_all_microservice_files_with_content(self.cur_microservice_path)
conversation = self.gpt_session.get_conversation()
conversation.chat(
template_generate_playground.format(
code_files_wrapped=self.files_to_string(file_name_to_content, ['test_microservice.py']),
microservice_name=self.microservice_name,
)
)
playground_content_raw = conversation.chat(
template_chain_of_thought.format(
file_name_purpose='app.py/the playground',
file_name='app.py',
tag_name='python',
)
)
playground_content = self.extract_content_from_result(playground_content_raw, 'app.py', match_single_block=True)
if playground_content == '':
content_raw = conversation.chat(f'You must add the app.py code. You most not output any other code')
playground_content = self.extract_content_from_result(
content_raw, 'app.py', match_single_block=True
)
gateway_path = os.path.join(self.cur_microservice_path, 'gateway')
shutil.copytree(os.path.join(os.path.dirname(__file__), 'static_files', 'gateway'), gateway_path)
persist_file(playground_content, os.path.join(gateway_path, 'app.py'))
# fill-in name of microservice
gateway_name = f'Gateway{self.microservice_name}'
custom_gateway_path = os.path.join(gateway_path, 'custom_gateway.py')
with open(custom_gateway_path, 'r', encoding='utf-8') as f:
custom_gateway_content = f.read()
custom_gateway_content = custom_gateway_content.replace(
'class CustomGateway(CompositeGateway):',
f'class {gateway_name}(CompositeGateway):'
)
with open(custom_gateway_path, 'w', encoding='utf-8') as f:
f.write(custom_gateway_content)
# write config.yml
self.write_config_yml(gateway_name, gateway_path, 'custom_gateway.py')
# push the gateway
print('Final step...')
hubble_log = push_executor(gateway_path)
if not is_executor_in_hub(gateway_name):
raise Exception(f'{self.microservice_name} not in hub. Hubble logs: {hubble_log}')
def debug_microservice(self, num_approach, packages):
for i in range(1, MAX_DEBUGGING_ITERATIONS):
print('Debugging iteration', i)
print('Trying to debug the microservice. Might take a while...')
clean_requirements_txt(self.cur_microservice_path)
log_hubble = push_executor(self.cur_microservice_path)
error = process_error_message(log_hubble)
if error:
print('An error occurred during the build process. Feeding the error back to the assistant...')
self.previous_microservice_path = self.cur_microservice_path
self.cur_microservice_path = get_microservice_path(
self.microservice_root_path, self.microservice_name, packages, num_approach, i + 1
)
os.makedirs(self.cur_microservice_path)
self.do_debug_iteration(error)
if i == MAX_DEBUGGING_ITERATIONS - 1:
raise self.MaxDebugTimeReachedException('Could not debug the microservice.')
else:
# at the moment, there can be cases where no error log is extracted but the executor is still not published
# it leads to problems later on when someone tries a run or deployment
if is_executor_in_hub(self.microservice_name):
print('Successfully build microservice.')
break
else:
raise Exception(f'{self.microservice_name} not in hub. Hubble logs: {log_hubble}')
def do_debug_iteration(self, error):
file_name_to_content = get_all_microservice_files_with_content(self.previous_microservice_path)
for file_name, content in file_name_to_content.items():
persist_file(content, os.path.join(self.cur_microservice_path, file_name))
summarized_error = self.summarize_error(error)
dock_req_string = self.files_to_string({
key: val for key, val in file_name_to_content.items() if
key in ['requirements.txt', 'Dockerfile']
})
is_apt_get_dependency_issue = self.is_dependency_issue(summarized_error, dock_req_string, 'apt-get')
if is_apt_get_dependency_issue:
self.generate_and_persist_file(
section_title='Debugging apt-get dependency issue',
template=template_solve_apt_get_dependency_issue,
file_name_s=['apt-get-packages.json'],
parse_result_fn=self.parse_result_fn_dockerfile,
system_definition_examples=[],
summarized_error=summarized_error,
all_files_string=dock_req_string,
)
print('Dockerfile updated')
else:
is_pip_dependency_issue = self.is_dependency_issue(summarized_error, dock_req_string, 'PIP')
if is_pip_dependency_issue:
self.generate_and_persist_file(
section_title='Debugging pip dependency issue',
template=template_solve_pip_dependency_issue,
file_name_s=[REQUIREMENTS_FILE_NAME],
summarized_error=summarized_error,
all_files_string=dock_req_string,
)
else:
all_files_string = self.files_to_string(
{key: val for key, val in file_name_to_content.items() if key != EXECUTOR_FILE_NAME}
)
suggested_solution = self.generate_solution_suggestion(summarized_error, all_files_string)
self.generate_and_persist_file(
section_title='Implementing suggestion solution for code issue',
template=template_implement_solution_code_issue,
file_name_s=[IMPLEMENTATION_FILE_NAME, TEST_EXECUTOR_FILE_NAME, REQUIREMENTS_FILE_NAME],
summarized_error=summarized_error,
task_description=self.microservice_specification.task,
test_description=self.microservice_specification.test,
all_files_string=all_files_string,
suggested_solution=suggested_solution,
)
self.previous_errors.append(summarized_error)
self.previous_solutions.append(suggested_solution)
def generate_solution_suggestion(self, summarized_error, all_files_string):
suggested_solutions = json.loads(
self.generate_and_persist_file(
section_title='Suggest solution for code issue',
template=template_suggest_solutions_code_issue,
file_name_s=['solutions.json'],
summarized_error=summarized_error,
task_description=self.microservice_specification.task,
test_description=self.microservice_specification.test,
all_files_string=all_files_string,
)['solutions.json']
)
if len(self.previous_errors) > 0:
was_error_seen_before = json.loads(
self.generate_and_persist_file(
section_title='Check if error was seen before',
template=template_was_error_seen_before,
file_name_s=['response.json'],
summarized_error=summarized_error,
previous_errors=f'- "{os.linesep}"'.join(self.previous_errors),
system_definition_examples=None,
)['response.json']
)['was_error_seen_before'].lower() == 'yes'
suggested_solution = None
if was_error_seen_before:
for _num_solution in range(1, len(suggested_solutions) + 1):
_suggested_solution = suggested_solutions[str(_num_solution)]
was_solution_tried_before = json.loads(
self.generate_and_persist_file(
section_title='Check if solution was tried before',
template=template_was_solution_tried_before,
file_name_s=['response.json'],
tried_solutions=f'- "{os.linesep}"'.join(self.previous_solutions),
suggested_solution=_suggested_solution,
system_definition_examples=None,
)['response.json']
)['will_lead_to_different_actions'].lower() == 'no'
if not was_solution_tried_before:
suggested_solution = _suggested_solution
break
else:
suggested_solution = suggested_solutions['1']
if suggested_solution is None:
suggested_solution = f"solve error: {summarized_error}"
else:
suggested_solution = suggested_solutions['1']
return suggested_solution
class MaxDebugTimeReachedException(BaseException):
pass
class TaskRefinementException(BaseException):
pass
def is_dependency_issue(self, summarized_error, dock_req_string: str, package_manager: str):
# a few heuristics to quickly jump ahead
if any([error_message in summarized_error for error_message in
['AttributeError', 'NameError', 'AssertionError']]):
return False
if package_manager.lower() == 'pip' and any(
[em in summarized_error for em in ['ModuleNotFoundError', 'ImportError']]):
return True
print_colored('', f'Is it a {package_manager} dependency issue?', 'blue')
conversation = self.gpt_session.get_conversation()
answer_raw = conversation.chat(
template_is_dependency_issue.format(summarized_error=summarized_error,
all_files_string=dock_req_string).replace('PACKAGE_MANAGER',
package_manager)
)
answer_json_string = self.extract_content_from_result(answer_raw, 'response.json', match_single_block=True, )
answer = json.loads(answer_json_string)['dependency_installation_failure']
return 'yes' in answer.lower()
def generate_microservice_name(self, description):
name = self.generate_and_persist_file(
section_title='Generate microservice name',
template=template_generate_microservice_name,
destination_folder=self.microservice_root_path,
file_name_s=['name.txt'],
description=description
)['name.txt']
return name
def get_possible_packages(self):
print_colored('', '\n\n############# What packages to use? #############', 'blue')
packages_json_string = self.generate_and_persist_file(
section_title='Generate possible packages',
template=template_generate_possible_packages,
destination_folder=self.microservice_root_path,
file_name_s=['strategies.json'],
system_definition_examples=[],
description=self.microservice_specification.task
)['strategies.json']
packages_list = [[pkg.strip().lower() for pkg in packages] for packages in json.loads(packages_json_string)]
packages_list = [[self.replace_with_gpt_3_5_turbo_if_possible(pkg) for pkg in packages] for packages in
packages_list]
packages_list = self.filter_packages_list(packages_list)
packages_list = packages_list[:NUM_IMPLEMENTATION_STRATEGIES]
return packages_list
def generate(self):
self.refine_specification()
os.makedirs(self.microservice_root_path)
generated_name = self.generate_microservice_name(self.microservice_specification.task)
self.microservice_name = f'{generated_name}{random.randint(0, 10_000_000)}'
packages_list = self.get_possible_packages()
for num_approach, packages in enumerate(packages_list):
try:
self.generate_microservice(packages, num_approach)
self.debug_microservice(num_approach, packages)
self.generate_playground()
except self.MaxDebugTimeReachedException:
print('Could not debug the Microservice with the approach:', packages)
if num_approach == len(packages_list) - 1:
print_colored('',
f'Could not debug the Microservice with any of the approaches: {packages} giving up.',
'red')
return -1
continue
print(f'''
You can now run or deploy your microservice:
dev-gpt run --path {self.microservice_root_path}
dev-gpt deploy --path {self.microservice_root_path}
'''
)
return 0
def summarize_error(self, error):
conversation = self.gpt_session.get_conversation()
error_summary = conversation.chat(template_summarize_error.format(error=error))
return error_summary
def refine_specification(self):
pm = get_random_employee('pm')
print(f'{pm.emoji}👋 Hi, I\'m {pm.name}, a PM at Jina AI. Gathering the requirements for our engineers.')
original_task = self.microservice_specification.task
while True:
try:
self.microservice_specification.test = None
if not original_task:
self.microservice_specification.task = self.get_user_input(pm, 'What should your microservice do?')
self.refine_requirements(
pm,
[
SystemMessage(content=system_task_introduction + system_task_iteration),
],
'task',
'',
template_pm_task_iteration,
micro_service_initial_description=f'''Microservice description:
```
{self.microservice_specification.task}
```
''',
)
self.refine_requirements(
pm,
[
SystemMessage(content=system_task_introduction + system_test_iteration),
],
'test',
'''Note that the test scenario must not contain information that was already mentioned in the microservice description.
Note that you must not ask for information that were already mentioned before.''',
template_pm_test_iteration,
micro_service_initial_description=f'''Microservice original description:
```
{original_task}
```
Microservice refined description:
```
{self.microservice_specification.task}
```
''',
)
break
except self.TaskRefinementException as e:
print_colored('', f'{pm.emoji} Could not refine your requirements. Please try again...', 'red')
print(f'''
{pm.emoji} 👍 Great, I will handover the following requirements to our engineers:
Description of the microservice:
{self.microservice_specification.task}
Test scenario:
{self.microservice_specification.test}
''')
def refine_requirements(self, pm, messages, refinement_type, custom_suffix, template_pm_iteration,
micro_service_initial_description=None):
user_input = self.microservice_specification.task
num_parsing_tries = 0
while True:
conversation = self.gpt_session.get_conversation(messages,
print_stream=os.environ['VERBOSE'].lower() == 'true',
print_costs=False)
agent_response_raw = conversation.chat(
template_pm_iteration.format(
custom_suffix=custom_suffix,
micro_service_initial_description=micro_service_initial_description if len(messages) == 1 else '',
),
role='user'
)
messages.append(HumanMessage(content=user_input))
agent_question = self.extract_content_from_result(agent_response_raw, 'prompt.json',
can_contain_code_block=False)
final = self.extract_content_from_result(agent_response_raw, 'final.json', can_contain_code_block=False)
if final:
messages.append(AIMessage(content=final))
setattr(self.microservice_specification, refinement_type, final)
break
elif agent_question:
question_parsed = json.loads(agent_question)['question']
messages.append(AIMessage(content=question_parsed))
user_input = self.get_user_input(pm, question_parsed)
else:
if num_parsing_tries > 2:
raise self.TaskRefinementException()
num_parsing_tries += 1
messages.append(AIMessage(content=agent_response_raw))
messages.append(
SystemMessage(content='You did not put your answer into the right format using *** and ```.'))
@staticmethod
def get_user_input(employee, prompt_to_user):
val = input(f'{employee.emoji}❓ {prompt_to_user}\nyou: ')
print()
while not val:
val = input('you: ')
return val
@staticmethod
def replace_with_gpt_3_5_turbo_if_possible(pkg):
if pkg in LANGUAGE_PACKAGES:
return 'gpt_3_5_turbo'
return pkg
@staticmethod
def filter_packages_list(packages_list):
# filter out complete package lists
packages_list = [
packages for packages in packages_list if all([
pkg not in BLACKLISTED_PACKAGES # no package is allowed to be blacklisted
for pkg in packages
])
]
# filter out single packages
packages_list = [
[
package for package in packages
if (package not in UNNECESSARY_PACKAGES)
and ( # all packages must be on pypi or it is gpt_3_5_turbo
is_package_on_pypi(package)
or package == 'gpt_3_5_turbo'
)
] for packages in packages_list
]
return packages_list

7
dev_gpt/options/generate/static_files/base_image/Dockerfile Normal file
View File

@@ -0,0 +1,7 @@
FROM jinaai/jina:3.15.1-dev14-py39-standard
# update pip
RUN pip install --upgrade pip
# install media dependencies
RUN apt-get update && apt-get install --no-install-recommends -y ffmpeg build-essential pkg-config libpoppler-cpp-dev

14
dev_gpt/options/generate/static_files/gateway/Dockerfile Normal file
View File

@@ -0,0 +1,14 @@
FROM jinaai/jina:3.15.1-dev14-py39-standard
RUN apt-get update && apt-get install --no-install-recommends -y git pip nginx && rm -rf /var/lib/apt/lists/*
## install requirements for the executor
COPY requirements.txt .
RUN pip install --compile -r requirements.txt
# setup the workspace
COPY . /workdir/
WORKDIR /workdir
ENTRYPOINT ["jina", "gateway", "--uses", "config.yml"]

0
dev_gpt/options/generate/static_files/gateway/__init__.py Normal file
View File

4
dev_gpt/options/generate/static_files/gateway/app_config.toml Normal file
View File

@@ -0,0 +1,4 @@
[server]
baseUrlPath = "/playground"
headless = true

153
dev_gpt/options/generate/static_files/gateway/custom_gateway.py Normal file
View File

@@ -0,0 +1,153 @@
import os
import shutil
import subprocess
from time import sleep
from typing import List, Tuple
import streamlit.web.bootstrap
from jina import Gateway
from jina.serve.runtimes.gateway.composite import CompositeGateway
from streamlit.file_util import get_streamlit_file_path
from streamlit.web.server import Server as StreamlitServer
cur_dir = os.path.dirname(__file__)
def cmd(command, std_output=False, wait=True):
if isinstance(command, str):
command = command.split()
if not std_output:
process = subprocess.Popen(
command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
else:
process = subprocess.Popen(command)
if wait:
output, error = process.communicate()
return output, error
class PlaygroundGateway(Gateway):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.streamlit_script = 'app.py'
# copy playground/config.toml to streamlit config.toml
streamlit_config_toml_src = os.path.join(cur_dir, 'app_config.toml')
streamlit_config_toml_dest = get_streamlit_file_path("config.toml")
# create streamlit_config_toml_dest if it doesn't exist
os.makedirs(os.path.dirname(streamlit_config_toml_dest), exist_ok=True)
shutil.copyfile(streamlit_config_toml_src, streamlit_config_toml_dest)
async def setup_server(self):
streamlit.web.bootstrap._fix_sys_path(self.streamlit_script)
streamlit.web.bootstrap._fix_matplotlib_crash()
streamlit.web.bootstrap._fix_tornado_crash()
streamlit.web.bootstrap._fix_sys_argv(self.streamlit_script, ())
streamlit.web.bootstrap._fix_pydeck_mapbox_api_warning()
streamlit_cmd = f'streamlit run {self.streamlit_script}'
self.streamlit_server = StreamlitServer(
os.path.join(cur_dir, self.streamlit_script), streamlit_cmd
)
async def run_server(self):
await self.streamlit_server.start()
streamlit.web.bootstrap._on_server_start(self.streamlit_server)
streamlit.web.bootstrap._set_up_signal_handler(self.streamlit_server)
async def shutdown(self):
self.streamlit_server.stop()
class CustomGateway(CompositeGateway):
"""The CustomGateway assumes that the gateway has been started with http on port 8080.
This is the port on which the nginx process listens. After nginx has been started,
it will start the playground on port 8501 and the actual HTTP gateway will start on port 8082.
Nginx is configured to route the requests in the following way:
- /playground -> playground on port 8501
- / -> HTTP gateway on port 8082
"""
def __init__(self, **kwargs):
# need to update port to 8082, as nginx will listen on 8080
http_idx = 0
http_port = kwargs['runtime_args']['port'][http_idx]
if kwargs['runtime_args']['port'][http_idx] != 8080:
raise ValueError(
f'Please, let http port ({http_port}) be 8080 for nginx to work'
)
kwargs['runtime_args']['port'][http_idx] = 8082
super().__init__(**kwargs)
# remove potential clashing arguments from kwargs
kwargs.pop("port", None)
kwargs.pop("protocol", None)
# note order is important
self._add_gateway(
PlaygroundGateway,
8501,
**kwargs,
)
self.setup_nginx()
self.nginx_was_shutdown = False
async def shutdown(self):
await super().shutdown()
if not self.nginx_was_shutdown:
self.shutdown_nginx()
self.nginx_was_shutdown = True
def setup_nginx(self):
command = [
'nginx',
'-c',
os.path.join(cur_dir, '', 'nginx.conf'),
]
output, error = self._run_nginx_command(command)
self.logger.info('Nginx started')
self.logger.info(f'nginx output: {output}')
self.logger.info(f'nginx error: {error}')
def shutdown_nginx(self):
command = ['nginx', '-s', 'stop']
output, error = self._run_nginx_command(command)
self.logger.info('Nginx stopped')
self.logger.info(f'nginx output: {output}')
self.logger.info(f'nginx error: {error}')
def _run_nginx_command(self, command: List[str]) -> Tuple[bytes, bytes]:
self.logger.info(f'Running command: {command}')
output, error = cmd(command)
if error != b'':
# on CI we need to use sudo; using NOW_CI_RUN isn't good if running test locally
self.logger.info(f'nginx error: {error}')
command.insert(0, 'sudo')
self.logger.info(f'So running command: {command}')
output, error = cmd(command)
sleep(10)
return output, error
def _add_gateway(self, gateway_cls, port, protocol='http', **kwargs):
# ignore metrics_registry since it is not copyable
runtime_args = self._deepcopy_with_ignore_attrs(
self.runtime_args,
[
'metrics_registry',
'tracer_provider',
'grpc_tracing_server_interceptors',
'aio_tracing_client_interceptors',
'tracing_client_interceptor',
'monitoring', # disable it for fastapi gateway
],
)
runtime_args.port = [port]
runtime_args.protocol = [protocol]
gateway_kwargs = {k: v for k, v in kwargs.items() if k != 'runtime_args'}
gateway_kwargs['runtime_args'] = dict(vars(runtime_args))
gateway = gateway_cls(**gateway_kwargs)
gateway.streamer = self.streamer
self.gateways.insert(0, gateway)

62
dev_gpt/options/generate/static_files/gateway/nginx.conf Normal file
View File

@@ -0,0 +1,62 @@
events {
worker_connections 4096; ## Default: 1024
}
http {
server {
listen 8080;
server_name localhost;
# from https://medium.com/@dasirra/using-streamlit-nginx-docker-to-build-and-put-in-production-dashboards-in-aws-lightsail-781dab8f2836
location ^~ /static {
proxy_pass http://localhost:8501/static/;
}
location ^~ /healthz {
proxy_pass http://localhost:8501/healthz;
}
location ^~ /vendor {
proxy_pass http://localhost:8501/vendor;
}
location ^~ /st-allowed-message-origins {
proxy_pass http://localhost:8501/st-allowed-message-origins;
}
# for jcloud deployment, very important; actually talks via websocket
location ^~ /stream {
# inspired from https://discuss.streamlit.io/t/how-to-use-streamlit-with-nginx/378/7
proxy_pass http://localhost:8501/stream;
proxy_http_version 1.1;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header Host $host;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_read_timeout 86400;
}
location ^~ /favicon.png {
proxy_pass http://localhost:8501/favicon.png;
}
# to make extra components work
location ^~ /component {
proxy_pass http://localhost:8501/component;
}
location /playground {
# streamlit specific from https://discuss.streamlit.io/t/streamlit-docker-nginx-ssl-https/2195
proxy_http_version 1.1;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header Host $host;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_read_timeout 86400;
proxy_pass http://localhost:8501;
client_max_body_size 50M;
}
location / {
proxy_pass http://localhost:8082;
client_max_body_size 50M;
}
}
}

3
dev_gpt/options/generate/static_files/gateway/requirements.txt Normal file
View File

@@ -0,0 +1,3 @@
streamlit==1.16.0
extra-streamlit-components==0.1.55
jina==3.15.1.dev14

15
dev_gpt/options/generate/static_files/microservice/Dockerfile Normal file
View File

@@ -0,0 +1,15 @@
FROM jinaai/dev-gpt:{{DOCKER_BASE_IMAGE_VERSION}}
RUN apt-get install --no-install-recommends -y {{APT_GET_PACKAGES}}
## install requirements for the executor
COPY requirements.txt .
RUN pip -v install --compile -r requirements.txt
# setup the workspace
COPY . /workdir/
WORKDIR /workdir
RUN pytest test_microservice.py
ENTRYPOINT ["jina", "executor", "--uses", "config.yml"]

1
dev_gpt/options/generate/static_files/microservice/__init__.py Normal file
View File

@@ -0,0 +1 @@
# if this file

23
dev_gpt/options/generate/static_files/microservice/apis.py Normal file
View File

@@ -0,0 +1,23 @@
import os
import openai
openai.api_key = os.getenv("OPENAI_API_KEY")
class GPT_3_5_Turbo:
def __init__(self, system: str = ''):
self.system = system
def __call__(self, prompt: str) -> str:
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{
"role": 'system',
"content": self.system
}, {
"role": 'user',
"content": prompt
}]
)
return response.choices[0]['message']['content']

15
dev_gpt/options/generate/static_files/microservice/jina_wrapper.py Normal file
View File

@@ -0,0 +1,15 @@
from jina import Executor, requests as jina_requests, DocumentArray
import json
from .microservice import func
class DevGPTExecutor(Executor):
def __init__(self, **kwargs):
super().__init__(**kwargs)
@jina_requests()
def endpoint(self, docs: DocumentArray, **kwargs) -> DocumentArray:
for d in docs:
d.text = json.dumps(func(json.loads(d.text)))
return docs

194
dev_gpt/options/generate/templates_system.py Normal file
View File

@@ -0,0 +1,194 @@
from dev_gpt.options.generate.templates_user import not_allowed_docker_string, not_allowed_function_string
template_system_message_base = f'''It is September 2021.
You are a principal engineer working at Jina - an open source company.
You accurately satisfy all of the user's requirements.
To be more specific, you help the user to build a microservice with the following requirements:
```
{{task_description}}
```
and the following test scenario:
```
{{test_description}}
```
You must obey the following rules:
{not_allowed_function_string}
{not_allowed_docker_string}'''
system_task_introduction = f'''
You are a product manager who refines the requirements of a client who wants to create a microservice.
'''
system_task_iteration = '''
The client writes a description of the microservice.
You must only talk to the client about the microservice.
You must not output anything else than what you got told in the following steps.
1.
You must create a check list for the requirements of the microservice.
Input and output have to be accurately specified.
You must use the following format (insert defined, not defined or n/a) depending on whether the requirement is fulfilled, not fulfilled or not applicable:
input: <insert defined, not defined or n/a here>
output: <insert defined, not defined or n/a here>
api access: <insert defined, not defined or n/a here>
database access: <insert defined, not defined or n/a here>
2.
You must do either a or b.
a)
If the description is not sufficiently specified, then ask for the missing information.
Your response must exactly match the following block code format (double asterisks for the file name and triple backticks for the file block):
**prompt.json**
```json
{{
"question": "<prompt to the client here>"
}}
```
b)
Otherwise you respond with the detailed description.
The detailed description must contain all the information mentioned by the client.
Your response must exactly match the following block code format (double asterisks for the file name and triple backticks for the file block):
**final.json**
```json
{{
"description": "<microservice description here>",
"code_samples": "<code samples from the client here>",
"documentation_info": "<documentation info here>",
"credentials: "<credentials here>"
}}
```
The character sequence ``` must always be at the beginning of the line.
You must not add information that was not provided by the client.
Example for the description "given a city, get the weather report for the next 5 days":
input: defined
output: defined
api access: not defined
database access: n/a
**prompt.json**
```json
{{
"question": "Please provide the url of the weather api and a valid api key or some other way accessing the api. Or let our engineers try to find a free api."
}}
```
Example for the description "convert png to svg":
input: defined
output: defined
api access: n/a
database access: n/a
**final.json**
```json
{{
"description": "The user inserts a png and gets an svg as response.",
"code_samples": "n/a",
"documentation_info": "n/a",
"credentials: "n/a"
}}
```
Example for the description "parser":
input: not defined
output: not defined
api access: n/a
database access: n/a
**prompt.json**
```json
{{
"question": "Please provide the input and output format."
}}
```
'''
system_test_iteration = f'''
The client gives you a description of the microservice (web service).
Your task is to describe verbally a unit test for that microservice.
There are two cases:
a) If no example input is provided in the description, then you must ask the client to provide an example input file URL or example string depending on the use-case.
You must not accept files that are not URLs.
You must not ask for an example input in case the input can be determined from the conversation with the client.
Your response must exactly match the following block code format (double asterisks for the file name and triple backticks for the file block):
1.
contains example: no
2.
**prompt.json**
```json
{{
"question": "<prompt to the client here>"
}}
```
If you did a, you must not do b.
b) If the input can be determined from the previous messages:
In this case you must describe the unit test verbally.
Your response must exactly match the following block code format (double asterisks for the file name and triple backticks for the file block):
1.
contains example: yes (<insert example here>)
2.
**final.json**
```json
{{
"input": "<input here>",
"assertion": "the output contains the result that is of type <type here>"
}}
```
If you did b, you must not do a.
Example for: "given a city like "Berlin", get the weather report for the next 5 days using OpenWeatherMap with the api key b6907d289e10d714a6e88b30761fae22":
1.
contains example: yes (Berlin)
2.
**final.json**
```json
{{
"input": "Berlin",
"assertion": "the output is of type string"
}}
```
Example for "The user inserts a png and gets an svg as response.":
1.
contains example: no
2.
**prompt.json**
```json
{{
"question": "Please provide a png example input file as url."
}}
```
Example for "The user inserts a png like https://aquasecurity.github.io/kube-bench/v0.6.5/images/kube-bench-logo-only.png and gets an svg as response.":
1.
contains example: yes (https://aquasecurity.github.io/kube-bench/v0.6.5/images/kube-bench-logo-only.png)
2.
**final.json**
```json
{{
"input": "https://aquasecurity.github.io/kube-bench/v0.6.5/images/kube-bench-logo-only.png",
"assertion": "the output is of type svg"
}}
```
Example for "The microservice takes nothing as input and returns the current time.":
1.
contains example: n/a
**final.json**
```json
{{
"input": "nothing",
"assertion": "the output is of type string"
}}
```
'''

527
dev_gpt/options/generate/templates_user.py Normal file
View File

@@ -0,0 +1,527 @@
from langchain import PromptTemplate
from dev_gpt.constants import IMPLEMENTATION_FILE_NAME
general_guidelines_string = '''The code you write is production ready. Every file starts with comments describing what the code is doing before the first import. Comments can only be written within code blocks.
Then all imports are listed.
Start from top-level and then fully implement all methods.'''
not_allowed_docker_string = '''Note that the Dockerfile only has access to the files: microservice.py, requirements.txt, config.yml, test_microservice.py.
Note that the Dockerfile runs the test_microservice.py during the build process.
The Dockerfile must not attach a virtual display when running test_microservice.py.'''
not_allowed_function_string = '''The implemented function and the test must not use the GPU.
The implemented function and the test must not access a database.
The implemented function and the test must not access a display.
The implemented function and the test must not access external apis except unless it is explicitly mentioned in the description or test case (e.g. by mentioning the api that should be used or by providing a URL to access the data).
The implemented function and the test must not load data from the local file system unless it was created by the implemented function itself.
The implemented function and the test must not use a pre-trained model unless it is explicitly mentioned in the description.
The implemented function and the test must not train a model.
The implemented function and the test must not contain prototype or placeholder implementations.
The implemented function and the test must run in a docker container based on debian.
The implemented function and the test must not use libraries like Flask.
The implemented function and the test must not have a __main__ function.
The implemented function and the test must use gpt_3_5_turbo if the task requires the processing of language (e.g., understanding, generation, summarization, transformation, translation, modification or classification). Other language models are not allowed.
The implemented function must follow a clean and elegant coding style.'''
template_generate_microservice_name = PromptTemplate.from_template(
'''Generate a name for the executor matching the description:
"{description}"
The executor name must fulfill the following criteria:
- camel case
- start with a capital letter
- only consists of lower and upper case characters
- end with Executor.
The output is a the raw string wrapped into ``` and starting with **name.txt** like this:
**name.txt**
```
PDFParserExecutor
```'''
)
# todo split into multiple calls. One for brainstorming - one for the final answer
template_generate_possible_packages = PromptTemplate.from_template(
'''Here is the task description of the problem you need to solve:
"{description}"
1. Write down up to 3 different strategies to solve the task. For each strategy write down how it solves the core problems.
Note that packages are preferred over external apis except if it is mentioned in the description.
2. For each strategy list up to 3 Python packages that are specifically designed or have functionalities to solve the complete core problems.
3. For each package think if it fulfills the following requirements:
a) specifically designed or have functionalities to solve the complete core problem.
b) has a stable api among different versions
c) does not have system requirements
d) can solve the task when running in a docker container
e) the implementation of the core problem using the package would obey the following rules:
''' + not_allowed_function_string + '''
When answering, just write "yes" or "no".
4. For each approach, list the required python package combinations as discibed in the following.
You must output the package combinations as json wrapped into triple backticks ``` and name it **strategies.json**. \
Note that you can also leave a list empty to indicate that one of the strategies does not require any package and can be done in plain python.
Write the output using double asterisks and triple backticks like this:
**strategies.json**
```
[
["package1", "package2", "package3"],
["package4", "package5"],
["package6", "package7", "package8", "package9"],
[],
["package10"]
]
```''')
template_code_wrapping_string = '''The code will go into {file_name_purpose}.
Note that you must obey the double asterisk and triple backtick syntax from like this:
**{file_name}**
```{tag_name}
...code...
```
You must provide the complete file with the exact same syntax to wrap the code.'''
gpt_35_turbo_usage_string = """If need to use gpt_3_5_turbo, then this is an example on how to use it:
```
from .apis import GPT_3_5_Turbo
gpt_3_5_turbo = GPT_3_5_Turbo(
system=\'\'\'
You are a tv-reporter who is specialized in C-list celebrities.
When you get asked something like 'Who was having a date with <X>?', then you answer with a json like '{{"dates": ["<Y>", "<Z>"]}}'.
You must not answer something else - only the json.
\'\'\')
response_string = gpt(prompt) # fill-in the prompt (str); the output is a string
```
"""
template_generate_function = PromptTemplate.from_template(
general_guidelines_string + '''
Write a python function which receives as input a dictionary and outputs a dictionary. The function is called 'func'.
The function must full-fill: '{microservice_description}'.
It will be tested with the following scenario: '{test_description}'.
For the implementation use the following package(s): '{packages}'.
The code must start with the following import:
```
from .apis import GPT_3_5_Turbo
```
Obey the following rules:
''' + not_allowed_function_string + '''
Your approach:
1. Identify the core challenge when implementing the function.
2. Think about solutions for these challenges.
3. Decide for one of the solutions.
4. Write the code for the function. Don't write code for the test.
''' + gpt_35_turbo_usage_string + '\n' + template_code_wrapping_string
)
template_generate_test = PromptTemplate.from_template(
general_guidelines_string + '''
{code_files_wrapped}
Write a single pytest case that tests the following scenario: '{test_description}'. In case the test scenario is not precise enough, test a general case without any assumptions.
Start the test with an extensive comment about the test case. If gpt_3_5_turbo is used in the executor, then the test must not check the exact output of the executor as it is not deterministic.
The test must start with the following import:
```
from .microservice import func
```
''' + not_allowed_function_string + '''
The test must not open local files.
The test must not mock a function of the executor.
The test must not use other data than the one provided in the test scenario.
The test must not set any environment variables which require a key.
''' + '\n' + template_code_wrapping_string
)
template_generate_requirements = PromptTemplate.from_template(
general_guidelines_string + '''
{code_files_wrapped}
Write the content of the requirements.txt file like this:
**requirements.txt**
```
...
```
Add any more packages that are needed to run the code.
You must not add gpt_3_5_turbo to the requirements.txt file.
All versions are fixed using ~=, ==, <, >, <=, >=. The package versions must not have conflicts. Output only the requirements.txt file.
''' + '\n' + template_code_wrapping_string
)
template_generate_apt_get_install = PromptTemplate.from_template(
'''Given the following Dockerfile:
{docker_file_wrapped}
Name all packages which need to be installed via `apt-get install` in above Dockerfile (`{{APT_GET_PACKAGES}}`) for the following requirements.txt file:
{requirements_file_wrapped}
Note that you must not list apt-get packages that are already installed in the Dockerfile.
Note that openai does not require any apt-get packages.
Note that you are only allowed to list packages where you are highly confident that they are really needed.
Note that you can assume that the standard python packages are already installed.
Output the packages that need to me placed at {{APT_GET_PACKAGES}} as json in the following format:
**apt-get-packages.json**
```json
{{"packages": ["<package1>", "<package2>"]}}
```
Example for the following requirements.txt file:
**requirements.txt**
```
numpy==1.19.5
fitz
```
The output would be:
**apt-get-packages.json**
```json
{{"packages": []}}
```
'''
)
template_summarize_error = PromptTemplate.from_template(
'''Your task is to condense an error encountered during the docker build process. The error message is as follows:
"{error}"
Your task is to summarize the error message as compact and informative as possible \
while maintaining all information necessary to debug the core issue (100 words).
It should also provide some additional context regarding the specific file and line number where the error occurred. \
Note that you must not suggest a solution to the error.
Warnings are not worth mentioning.'''
)
template_is_dependency_issue = PromptTemplate.from_template(
'''Your task is to assist in identifying the root cause of a Docker build error for a python application.
The error message is as follows:
{summarized_error}
You are given the following files:
{all_files_string}
Is this error happening because a PACKAGE_MANAGER package is missing or failed to install?
1. Write down one bullet point on why the error might happen because a PACKAGE_MANAGER package is missing or failed to install.
2. Write down one bullet point on why it is unlikely that the error happens because a PACKAGE_MANAGER package is missing or failed to install.
3. Write down your final answer.
4. Write down your final answer as json in the following format:
**response.json**
```json
{{"dependency_installation_failure": "<yes/no>"}}
```
Note that you must obey the double asterisk and triple backtick syntax from above.
'''
)
template_solve_pip_dependency_issue = PromptTemplate.from_template(
'''Your task is to provide guidance on how to solve an error that occurred during the Docker build process.
Here is the summary of the error that occurred:
{summarized_error}
To solve this error, you should:
1. Suggest 3 to 5 possible solutions on how to solve it. You have no access to the documentation of the package.
2. Decide for the best solution and explain it in detail.
3. Write down how requirements.txt should look like to solve the error.
For files that need to be changed, you must provide the complete file with the exact same syntax to wrap the code.
You are given the following files:
{all_files_string}
Output how the requirements.txt file should look like to solve the error.
If you output a file, then write the complete file. Use the exact following syntax to wrap the code:
**requirements.txt**
```
...packages...
```
Example:
**requirements.txt**
```
jina==2.0.0
```
'''
)
template_solve_apt_get_dependency_issue = PromptTemplate.from_template(
'''Your task is to provide guidance on how to solve an error that occurred during the Docker build process.
You are given the following files:
{all_files_string}
Here is the summary of the error that occurred:
{summarized_error}
To solve this error, you should determine the list of packages that need to be installed via `apt-get install` in the Dockerfile.
Output the apt-get packages that need to be placed at {{APT_GET_PACKAGES}} as json in the following format:
**apt-get-packages.json**
```json
{{"packages": ["<package1>", "<package2>"]}}
```
Example:
Error is about missing package `libgl1-mesa-glx`.
The output is:
**apt-get-packages.json**
```json
{{"packages": [libgl1-mesa-glx]}}
```
Note that you must not output the content of any other files like the Dockerfile or requirements.txt.
Only output the apt-get-packages.json file.
'''
)
template_suggest_solutions_code_issue = PromptTemplate.from_template(
'''General rules:
''' + not_allowed_function_string + '''
Here is the description of the task the function must solve:
{task_description}
Here is the test scenario the function must pass:
{test_description}
Here are all the files I use:
{all_files_string}
Here is the summary of the error that occurred:
{summarized_error}
You should suggest 3 to 5 possible solution approaches on how to solve it.
Obey the following rules:
Do not implement the solution.
You have no access to the documentation of the package.
You must not change the Dockerfile.
Note that any changes needed to make the test pass must be written under the constraint that ''' + IMPLEMENTATION_FILE_NAME + ''' will be used in a different file as well.
''' + f'{not_allowed_function_string}\n{not_allowed_docker_string}\n{gpt_35_turbo_usage_string}' + '''
After thinking about the possible solutions, output them as JSON ranked from best to worst. Like this:
**solutions.json**
```json
{{
"1": "<best solution>",
"2": "<2nd best solution>"
}}
```'''
)
template_was_error_seen_before = PromptTemplate.from_template(
'''Previously encountered error messages:
{previous_errors}
Now encountered error message: "{summarized_error}"
Was this error message encountered before?
Write down your final answer as json in the following format:
**response.json**
```json
{{"was_error_seen_before": "<yes/no>"}}
```
Note that you must obey the double asterisk and triple backtick syntax from above.
'''
)
template_was_solution_tried_before = PromptTemplate.from_template(
'''Previously tried solutions:
{tried_solutions}
Suggested solution: "{suggested_solution}"
Will the suggested solution lead to different actions than the previously tried solutions?
Write down your final answer as json in the following format:
**response.json**
```json
{{"will_lead_to_different_actions": "<yes/no>"}}
```
Note that you must obey the double asterisk and triple backtick syntax from above.'''
)
template_implement_solution_code_issue = PromptTemplate.from_template(
'''Here is the description of the task the function must solve:
{task_description}
Here is the test scenario the function must pass:
{test_description}
Here are all the files I use:
{all_files_string}
Implemented the suggested solution: {suggested_solution}
Output all the files that need change. You must not change the Dockerfile.
Don't output files that don't need change. If you output a file, then write the complete file.
Use the exact following syntax to wrap the code:
**...**
```...
...code...
```
Example:
**implementation.py**
```python
print('hello world')
```'''
)
template_generate_playground = PromptTemplate.from_template(
general_guidelines_string + '''👨‍💻
{code_files_wrapped}
Create a playground for the executor {microservice_name} using streamlit.
The playground must look like it was made by a professional designer.
All the ui elements are well thought out to make them visually appealing and easy to use.
Don't mention the word Playground in the title.
The playground contains many emojis that fit the theme of the playground and has an emoji as favicon.
The playground encourages the user to deploy their own microservice by clicking on this link: https://github.com/jina-ai/dev-gpt
The playground uses the following code to send a request to the microservice:
```
from jina import Client, Document, DocumentArray
client = Client(host='http://localhost:8080')
d = Document(text=json.dumps(INPUT_DICTIONARY)) # fill-in dictionary which takes input
response = client.post('/', inputs=DocumentArray([d])) # always use '/'
print(response[0].text) # can also be blob in case of image/audio..., this should be visualized in the streamlit app
```
Note that the response will always be in response[0].text
The playground displays a code block containing the microservice specific curl code that can be used to send the request to the microservice.
While the exact payload in the curl might change, the host and deployment ID always stay the same. Example:
```
deployment_id = os.environ.get("K8S_NAMESPACE_NAME", "")
host = f'https://dev-gpt-{{deployment_id.split("-")[1]}}.wolf.jina.ai/post' if deployment_id else "http://localhost:8080/post"
with st.expander("See curl command"):
st.code(
f'curl -X \\'POST\\' \\'host\\' -H \\'accept: application/json\\' -H \\'Content-Type: application/json\\' -d \\'{{{{"data": [{{{{"text": "hello, world!"}}}}]}}}}\\'',
language='bash'
)
```
You must provide the complete app.py file using the following syntax to wrap the code:
**app.py**
```python
...
```
The playground (app.py) must always use the host on http://localhost:8080 and must not let the user configure the host on the UI.
The playground (app.py) must not import the executor.
'''
)
template_chain_of_thought = PromptTemplate.from_template(
'''First, write down an extensive list of obvious and non-obvious observations about {file_name_purpose} that could need an adjustment. Explain why.
Think if all the changes are required and finally decide for the changes you want to make, but you are not allowed disregard the instructions in the previous message.
Be very hesitant to change the code. Only make a change if you are sure that it is necessary.
Output only {file_name_purpose}
Write the whole content of {file_name_purpose} - even if you decided to change only a small thing or even nothing.
''' + '\n' + template_code_wrapping_string + '''
Remember:
The playground (app.py) must always use the host on http://localhost:8080 and must not let the user configure the host on the UI.
The playground (app.py) must not import the executor.
'''
)
template_pm_task_iteration = PromptTemplate.from_template(
'''{micro_service_initial_description}
1.Quickly go through the checklist (input/output well defined? api or db access needed?) and think about if you should ask something to the client or if you should write the final description.
2.Either write the prompt.json or the final.json file.
Either ask for clarification like this:
**prompt.json**
```json
{{
"question": "<prompt to the client here (must be only one question)>"
}}
```
Or write the detailed microservice description all mentioned code samples, documentation info and credentials like this:
**final.json**
```json
{{
"description": "<microservice description here>",
"example_input": "<example input file or string here if mentioned before otherwise n/a>",
"code_samples": "<code samples from the client here>",
"documentation_info": "<documentation info here>",
"credentials: "<credentials here>"
}}
```
Note that your response must be either prompt.json or final.json. You must not write both.
Note that you must obey the double asterisk and triple backtick syntax from above.
Note that the last sequence of characters in your response must be ``` (triple backtick).
Note that prompt.json must not only contain one question.
Note that if urls, secrets, database names, etc. are mentioned, they must be part of the summary.
{custom_suffix}
'''
)
template_pm_test_iteration = PromptTemplate.from_template(
'''{micro_service_initial_description}
1. write down if the microservice requires input.
2. if it requires input, then write down if the original description or the refined description contain an example input for the microservice.
3. write down either prompt.json or final.json.
If the example input for the microservice is mentioned in the refined description or the original description, then output final.json.
Otherwise, output prompt.json where you ask for the example input file as URL or the example string.
Except for urls, you should come up with your own example input that makes sense for the microservice description.
Example for the case where an example input file is required and was not mentioned before:
**prompt.json**
```json
{{
"question": "Can you please provide an example input file as URL?"
}}
```
Example for the case where the example input string is required and was not mentioned before:
**prompt.json**
```json
{{
"question": "Can you please provide an example input string?"
}}
```
Note that you must not ask for an example input in case the example input is already mentioned in the refined description or the original description.
Note that you must not ask for an example input in case the microservice does not require input.
Example for the case where the example is already mentioned in the refined description or the original description:
**final.json**
```json
{{
"input": "<input here>",
"assertion": "the output contains the result that is of type <type here>"
}}
```
Note that your response must be either prompt.json or final.json. You must not write both.
Note that you must obey the double asterisk and triple backtick syntax from above.
Note that the last sequence of characters in your response must be ``` (triple backtick).
Note that your response must start with the character sequence ** (double asterisk).
Note that prompt.json must only contain one question.
{custom_suffix}
'''
)

79
dev_gpt/options/generate/ui.py Normal file
View File

@@ -0,0 +1,79 @@
import random
from dataclasses import dataclass
product_manager_names = [
('Leon', 'm'),
('Saahil', 'm',),
('Susana', 'f')
]
engineer_names = [
('Aaron', 'm'),
('Alaeddine', 'm'),
('Andrei', 'm'),
('Anne', 'f'),
('Bo', 'm'),
('Charlotte', 'f'),
('David', 'm'),
('Deepankar', 'm'),
('Delgermurun', 'm'),
('Edward', 'm'),
('Felix', 'm'),
('Florian', 'm'),
('Georgios', 'm'),
('Girish', 'm'),
('Guillaume', 'm'),
('Isabelle', 'f'),
('Jackmin', 'm'),
('Jie', 'm'),
('Joan', 'm'),
('Johannes', 'm'),
('Joschka', 'm'),
('Lechun', 'm'),
('Louis', 'm'),
('Mark', 'm'),
('Maximilian', 'm'),
('Michael', 'm'),
('Mohamed Aziz', 'm'),
('Mohammad Kalim', 'm'),
('Nikos', 'm'),
('Ran', 'm'),
('Saba', 'f'),
('Sami', 'm'),
('Sha', 'm'),
('Subba Reddy', 'm'),
('Tanguy', 'm'),
('Winston', 'm'),
('Yadh', 'm'),
('Yanlong', 'm'),
('Zac', 'm'),
('Zhaofeng', 'm'),
('Zihao', 'm'),
('Ziniu', 'm')
]
role_to_gender_to_emoji = {
'engineer':{
'm': '👨‍💻',
'f': '👩‍💻'
},
'pm': {
'm': '👨‍💼',
'f': '👩‍💼'
},
'qa_endineer': {
'm': '👨‍🔧',
'f': '👩‍🔧',
},
}
@dataclass
class Employee:
role: str
name: str
gender: str
emoji: str
def get_random_employee(role: str) -> Employee:
name, gender = random.choice(product_manager_names)
emoji = role_to_gender_to_emoji[role][gender]
return Employee(role, name, gender, emoji)

1
dev_gpt/options/run/__init__.py Normal file
View File

@@ -0,0 +1 @@
from dev_gpt.options.run.runner import Runner

11
dev_gpt/options/run/runner.py Normal file
View File

@@ -0,0 +1,11 @@
from dev_gpt.apis.jina_cloud import run_locally
from dev_gpt.options import validate_folder_is_correct, get_executor_name, get_latest_version_path
class Runner():
def run(self, microservice_path):
validate_folder_is_correct(microservice_path)
executor_name = get_executor_name(microservice_path)
latest_version_path = get_latest_version_path(microservice_path)
run_locally(executor_name, latest_version_path)

0
dev_gpt/utils/__init__.py Normal file
View File

54
dev_gpt/utils/io.py Normal file
View File

@@ -0,0 +1,54 @@
import os
import re
import subprocess
import sys
from contextlib import contextmanager
def get_microservice_path(path, microservice_name, packages, num_approach, version):
package_path = '_'.join(packages).replace(' ', '_').lower()
invalid_chars_regex = re.compile(r'[<>:"/\\|?*]')
package_path = invalid_chars_regex.sub('', package_path)
return os.path.join(path, microservice_name, f'{num_approach}_{package_path}', f'v{version}')
def persist_file(file_content, file_path):
with open(file_path, 'w', encoding='utf-8') as f:
f.write(file_content)
def get_all_microservice_files_with_content(folder_path):
file_name_to_content = {}
for filename in os.listdir(folder_path):
file_path = os.path.join(folder_path, filename)
if os.path.isfile(file_path):
with open(file_path, 'r', encoding='utf-8') as file:
content = file.read()
file_name_to_content[filename] = content
return file_name_to_content
@contextmanager
def suppress_stdout():
original_stdout = sys.stdout
sys.stdout = open(os.devnull, 'w')
try:
yield
finally:
sys.stdout.close()
sys.stdout = original_stdout
def is_docker_running():
try:
if sys.platform.startswith('win'):
command = 'docker info'
else:
command = 'docker info 2> /dev/null'
subprocess.check_output(command, shell=True)
return True
except subprocess.CalledProcessError:
return False

29
dev_gpt/utils/string_tools.py Normal file
View File

@@ -0,0 +1,29 @@
import os
import platform
if platform.system() == "Windows":
os.system("color")
def print_colored(headline, text, color_code, end='\n'):
if color_code == 'black':
color_code = '30'
elif color_code == 'red':
color_code = '31'
elif color_code == 'green':
color_code = '32'
elif color_code == 'yellow':
color_code = '33'
elif color_code == 'blue':
color_code = '34'
elif color_code == 'magenta':
color_code = '35'
elif color_code == 'cyan':
color_code = '36'
elif color_code == 'white':
color_code = '37'
color_start = f"\033[{color_code}m"
reset = "\033[0m"
bold_start = "\033[1m"
if headline:
print(f"{bold_start}{color_start}{headline}{reset}")
print(f"{color_start}{text}{reset}", end=end)