File size: 4,190 Bytes
9064b67
 
 
 
 
 
 
 
 
 
 
 
2de5e80
0b33796
 
 
 
 
59c15ca
7accea0
 
 
 
9064b67
2de5e80
9064b67
2de5e80
9064b67
 
 
 
2de5e80
9064b67
2de5e80
9064b67
 
 
 
 
 
 
 
2de5e80
9064b67
2de5e80
9064b67
 
 
 
 
 
 
2de5e80
9064b67
2de5e80
9064b67
 
 
 
 
 
 
 
2de5e80
9064b67
2de5e80
9064b67
2de5e80
9064b67
 
 
 
 
 
 
 
2de5e80
9064b67
 
 
2de5e80
9064b67
 
 
 
 
 
2de5e80
9064b67
2de5e80
9064b67
 
 
 
2de5e80
9064b67
 
 
 
2de5e80
9064b67
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
2de5e80
 
 
1057e6a
 
 
 
5e0acb9
c40c2ce
c1641f3
1057e6a
 
 
 
 
 
7accea0
1057e6a
 
 
 
 
 
 
 
 
 
 
5e0acb9
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
import gradio as gr
import openai, os, time

from openai import OpenAI
from utils import show_json

_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

_assistant, _thread = None, None

def create_assistant(client):
    assistant = client.beta.assistants.create(
        name="Python Code Generator",
        instructions=(
                         "You are a Python programming language expert that "
                         "generates Pylint-compliant code and explains it. "
                         "Only execute code when explicitly asked to."
                     ),
        model="gpt-4o",
        tools=[
                  {"type": "code_interpreter"}, 
                  {"type": "retrieval"},
              ],
    )
    
    show_json("assistant", assistant)
    
    return assistant
    
def create_thread(client):
    thread = client.beta.threads.create()
    
    show_json("thread", thread)
    
    return thread

def create_message(client, thread, msg):
    message = client.beta.threads.messages.create(
        role="user",
        thread_id=thread.id,
        content=msg,
    )
    
    show_json("message", message)
    
    return message

def create_run(client, assistant, thread):
    run = client.beta.threads.runs.create(
        assistant_id=assistant.id,
        thread_id=thread.id,
    )
    
    show_json("run", run)
    
    return run

def wait_on_run(client, thread, run):
    while run.status == "queued" or run.status == "in_progress":
        run = client.beta.threads.runs.retrieve(
            thread_id=thread.id,
            run_id=run.id,
        )
    
        time.sleep(0.25)
    
    show_json("run", run)
    
    return run

def list_run_steps(client, thread, run):
    run_steps = client.beta.threads.runs.steps.list(
        thread_id=thread.id,
        run_id=run.id,
        order="asc",
    )
    
    for step in run_steps.data:
        step_details = step.step_details
        show_json("step_details", step_details)
    
    return run_steps
    
def list_messages(client, thread):
    messages = client.beta.threads.messages.list(
        thread_id=thread.id
    )
    
    show_json("messages", messages)
    
    return messages
    
def extract_content_values(data):
    content_values = []
    
    for item in data.data:
        for content in item.content:
            if content.type == "text":
                content_values.append(content.text.value)
    
    return content_values

def chat(message, history, openai_api_key):
    global _client, _assistant, _thread     
       
    if _assistant == None:
        _assistant = create_assistant(_client)

    if _thread == None:
        _thread = create_thread(_client)
        
    create_message(_client, _thread, message)

    run = create_run(_client, _assistant, _thread)
    run = wait_on_run(_client, _thread, run)

    list_run_steps(_client, _thread, run)
    
    messages = list_messages(_client, _thread)

    content_values = extract_content_values(messages)
    
    return content_values[0]

def vote(data: gr.LikeData):
    print("voted")

gr.ChatInterface(
        fn=chat,
        chatbot=gr.Chatbot(height=300),
        textbox=gr.Textbox(placeholder="Ask anything", container=False, scale=7),
        title="Python Code Generator",
        description="Generate, explain, fix, optimize, document, test, help, ... Can execute code when asked to.",
        clear_btn="Clear",
        retry_btn="Retry",
        undo_btn="Undo",
        multimodal=True,
        examples=[
                  ["Generate: NumPy/Pandas/Matplotlib & yfinance trading app", "sk-<BringYourOwn>"],
                  ["Explain: r'^(?=.*[A-Z])(?=.*[a-z])(?=.*[0-9])(?=.*[\\W]).{8,}$'", "sk-<BringYourOwn>"],
                  ["Fix: x = [5, 2, 1, 3, 4]; print(x.sort())", "sk-<BringYourOwn>"],
                  ["Optimize: x = []; for i in range(0, 10000): x.append(i)", "sk-<BringYourOwn>"],
                  ["Execute: Code to generate the first 20 fibbonaci numbers", "sk-<BringYourOwn>"],
                 ],
        cache_examples=False,
        additional_inputs=[
            gr.Textbox("sk-", label="OpenAI API Key", type = "password"),
        ],
    ).launch()