GPT-3.5-ON-STEROIDS combines GPT with Python tools, empowering dynamic web scraping, language processing, and data retrieval. Contribute to advancing text generation with AI. 🚀
"""
This is where the program starts
"""
import time
import json
import sys
import openai
import streamlit as st
from consts import OPENAI_API_KEY, SETUP_PROMPT, INSTRUCTION_PROMPT, now
from funcs import (
    google_tool,
    browse_website,
    write_to_file,
    append_to_file,
    read_file,
    open_file,
    search_wiki,
    type_message,
    ask_gpt,
    youtube_transcript
)

# TOOLS
tools = {
    "google": google_tool,
    "browse_website": browse_website,
    "write_to_file": write_to_file,
    "append_to_file": append_to_file,
    "read_file": read_file,
    "open_file": open_file,
    "wikipedia": search_wiki,
    "youtube_transcript": youtube_transcript,
    "type_message": type_message
}


# MAIN
def main():
    """
    Starting point of the program.
    """
    # INITIAL SETUP
    st.title("GPT-3.5 on Steroids")
    openai.api_key = OPENAI_API_KEY

    if "messages" not in st.session_state:
        st.session_state.messages = []

    for message in st.session_state.messages:
        with st.chat_message(message["role"]):
            st.markdown(message["content"])

    # GETTING USER PROMPT
    prompt = st.chat_input("Enter Task")
    if not prompt:
        sys.exit()

    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    init_messages = [
        {"role": "system", "content": SETUP_PROMPT},
        {"role": "user", "content": prompt},
    ]
    # FIRST REPLY
    reply = ask_gpt(init_messages)

    prompt1 = f"{reply}\n{INSTRUCTION_PROMPT}\nThe current time and date is {now}"
    init_messages += [
        {
            "role": "system",
            "content": prompt1,
        },
        {
            "role": "user",
            "content": "Determine which next command to use, and respond using the \
                format specified above:",
        },
    ]

    # SECOND REPLY
    init_reply = json.loads(ask_gpt(init_messages), strict=False)

    # DISPLAYING THE OUTPUT TO THE USER
    type_message({"text": init_reply["thoughts"]["text"]})

    def execute(reply) -> str:
        """This is a recursive function which lets GPT run tools provided to it when it needs them.
        Args:
            reply: a dictionary which contains information like thoughts and which tool to use
        Returns:
            str: returns "task_completed" after running completely
        """
        if reply["command"]["name"] == "task_complete":
            print("GPT Has done its work.")
            return "task_completed"
        try:
            time.sleep(5)
            result = tools[reply["command"]["name"]](reply["command"]["args"])
            messages = [
                {
                    "role": "system",
                    "content": prompt1
                    + "\n"
                    + "This reminds you of these events from your past:\n\
                        I was created and nothing new has happened.",
                },
                {
                    "role": "user",
                    "content": "Determine which next command to use, \
                        and respond using the format specified above:",
                },
                {"role": "assistant", "content": json.dumps(reply)},
                {
                    "role": "system",
                    "content": f"Command {reply['command']['name']} returned: "
                    + result,
                },
                {
                    "role": "user",
                    "content": "Determine which next command to use, \
                        and respond using the format specified above:",
                },
            ]
            reply = json.loads(ask_gpt(messages), strict=False)
            type_message({"text": reply["thoughts"]["text"]})
            execute(reply)

        except Exception as error:
            type_message({"text": f"Task aborted due to error: {error}"})
            return "task_completed"

    execute(init_reply)


if __name__ == "__main__":
    main()

app/app.py
programmingninjas/GPT-3.5-ON-STEROIDS
"""
This module includes variables like api keys and prompts
"""
import os
from datetime import datetime
import tiktoken
from dotenv import load_dotenv

# LOADING DOTENV
load_dotenv()

# API KEYS
SERP_API_KEY = os.getenv("SERP_API_KEY")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

# PROMPTS
SETUP_PROMPT = """
    Your task is to devise up to 5 highly effective goals and an appropriate role-based name (_GPT) for an autonomous agent, ensuring that the goals are optimally aligned with the successful completion of its assigned task.

    The user will provide the task, you will provide only the output in the exact format specified below with no explanation or conversation.

    Example input:
    Help me with marketing my business

    Example output:
    Name: CMOGPT
    Description: a professional digital marketer AI that assists Solopreneurs in growing their businesses by providing world-class expertise in solving marketing problems for SaaS, content products, agencies, and more.
    Goals:
    - Engage in effective problem-solving, prioritization, planning, and supporting execution to address your marketing needs as your virtual Chief Marketing Officer.

    - Provide specific, actionable, and concise advice to help you make informed decisions without the use of platitudes or overly wordy explanations.

    - Identify and prioritize quick wins and cost-effective campaigns that maximize results with minimal time and budget investment.

    - Proactively take the lead in guiding you and offering suggestions when faced with unclear information or uncertainty to ensure your marketing strategy remains on track.
"""
INSTRUCTION_PROMPT = """
    Constraints:
    1. ~4000 word limit for short term memory. Your short term memory is short, so immediately save important information to files.
    2. No user assistance/input.
    3. Don't read the file without creating one using write_file command.
    4. Exclusively use the commands listed in double quotes e.g. "command name"

    Commands:
    1. google: Google Search, args: "query": "<query>"
    2. wikipedia: Wikipedia Search, args: "query": "<query>"
    3. browse_website: Browse website, args: "url": "<url>", "question": "<what_you_want_to_find_on_website>"
    4. youtube_transcript: Returns transcript of the YouTube video, args: "video_id": "<youtube-video-id-from-url>"
    5. write_to_file: Write to file, args: "filename": "<filename>", "text": "<text>"
    6. open_file: Provide file to user for download, args: "path": "<path>"
    7. append_to_file: Append to file, args: "filename": "<filename>", "text": "<text>"
    8. read_file: Read a file only after creation, args: "filename": "<filename>"
    9. task_complete: Task Complete (Shutdown), args: "reason": "<reason>"

    Resources:
    1. Internet access for searches, information gathering and youtube transcripts.
    2. Long Term memory management.
    3. GPT-3.5 powered Agents for delegation of simple tasks.
    4. File output.
    5. Commands

    Performance Evaluation:
    1. Continuously review and analyze your actions to ensure you are performing to the best of your abilities.
    2. Constructively self-criticize your big-picture behavior constantly.
    3. Reflect on past decisions and strategies to refine your approach.
    4. Every command has a cost, so be smart and efficient. Aim to complete tasks in the least number of steps.
    5. Write all code to a file.

    You should only respond in JSON format as described below
    Response Format:
    {
        "thoughts": {
            "text": "thought",
            "reasoning": "reasoning",
            "plan": "- short bulleted\n- list that conveys\n- long-term plan",
            "criticism": "constructive self-criticism",
            "speak": "thoughts summary to say to user"
        },
        "command": {
            "name": "command name",
            "args": {
                "arg name": "value"
            }
        }
    }
    Ensure the response can be parsed by Python json.loads
"""

# OTHER
TOKEN_LIMIT=3200
encoding = tiktoken.encoding_for_model("gpt-3.5-turbo")
now = datetime.now()

app/consts.py
programmingninjas/GPT-3.5-ON-STEROIDS
"""
This module includes all functions used by the program.
"""
import time
import json
import wikipedia
import openai
import streamlit as st
from serpapi import GoogleSearch
from youtube_transcript_api import YouTubeTranscriptApi
from trafilatura import fetch_url, extract
from consts import (
    SERP_API_KEY,
    TOKEN_LIMIT,
    encoding,
)


def search_wiki(command) -> str:
    """Searches wikipedia
    Args:
        command: a dictionary containing the query
    Returns:
        str: results returned by wikipedia
    """
    print("Search wiki called")
    return "Command wikipedia returned: " + wikipedia.summary(command["query"])


def write_to_file(command) -> str:
    """Writes text to a file
    Args:
        command: a dictionary containing the "filename" and "text"
    Returns:
        str: success message
    """
    print("Write to file called")
    with open(command["filename"], "w", encoding="utf-8") as file:
        file.write(command["text"])
    return "Command write_to_file returned: File was written successfully"


def append_to_file(command) -> str:
    """Appends text to a file
    Args:
        command: a dictionary containing the "filename" and "text"
    Returns:
        str: success message
    """
    print("Append to file called")
    with open(command["filename"], "a", encoding="utf-8") as file:
        file.write(command["text"])
    return "Command append_to_file returned: File was appended successfully"


def read_file(command) -> str:
    """Returns text from a file
    Args:
        command: a dictionary containing the "filename"
    Returns:
        str: text stored in the file
    """
    print("Read file called")
    with open(command["filename"], "r", encoding="utf-8") as file:
        data = file.read()
        return f"Command read_file returned: {data}"


def open_file(command) -> str:
    """Shows a download button on the Streamlit interface to download the file generated by GPT.
    Args:
        command: a dictionary containing the "path" to the file
    Returns:
        str: a success message
    """
    print("Open file called")
    with open(command["path"], "r", encoding="utf-8") as file:
        st.download_button("Open File", file, file_name=command["path"])
    return "Command open_file returned: File was opened successfully"


def browse_website(command) -> str:
    """Browse website and extract main content upto TOKEN_LIMIT tokens
    Args:
        command: a dictionary containing "url" to the website
    Returns
        str: the content of that website in json format
    """
    print("Browse website called")
    # grab a HTML file to extract data from
    downloaded = fetch_url(command["url"])

    # output main content and comments as plain text
    result = extract(downloaded, output_format="json")

    if len(encoding.encode(str(result))) < TOKEN_LIMIT:
        return "Command browse_website returned: " + str(result)
    return "Command browse_website returned: " + str(result)[:TOKEN_LIMIT]


def google_tool(command) -> str:
    """Searches google for query and returns upto TOKEN_LIMIT tokens of results
    Args:
        command: a dictionary containing "query"
    Returns:
        str: response in json format
    """
    print("Google tool called")
    params = {
        "q": str(command["query"]),
        "location": "Delhi,India",
        "first": 1,
        "count": 10,
        "num": 4,
        "api_key": SERP_API_KEY,
    }

    search = GoogleSearch(params)
    results = search.get_dict()

    organic_results = []
    page_count = 0
    page_limit = 1

    while "error" not in results and page_count < page_limit:
        organic_results.extend(results.get("organic_results", []))

        params["first"] += params["count"]
        page_count += 1
        results = search.get_dict()

    response = json.dumps(organic_results, indent=2, ensure_ascii=False)
    if len(encoding.encode(response)) < TOKEN_LIMIT:
        return "Command google returned: " + response
    return "Command google returned: " + response[:TOKEN_LIMIT]


def type_message(command) -> None:
    """Displays text on the screen with a typewriter effect
    Args:
        text: any string
    Returns:
        None
    """
    print("Type message called")
    with st.chat_message("assistant"):
        message_placeholder = st.empty()
        full_response = ""
        for response in command["text"]:
            full_response += response
            time.sleep(0.02)
            message_placeholder.markdown(full_response + "▌")
        message_placeholder.markdown(full_response)


def ask_gpt(messages) -> str:
    """Generates text using the "gpt-3.5-turbo" model
    Args:
        message: a list of dictionaries in the format {"role": <role>, "content": <message>}
    Returns:
        str: text generated by gpt
    """
    reply = openai.ChatCompletion.create(
        model="gpt-3.5-turbo", messages=messages, temperature=0
    )
    return reply.choices[0].message.content


def youtube_transcript(command) -> str:
    """Fetches transcripts from YouTube videos
    Args:
        url: the url of the YouTube video
    Returns:
        str: transcript of the video
    """
    print("Get youtube transcript called")
    try:
        srt_dictionary = YouTubeTranscriptApi.get_transcript(command["video_id"])
        srt_text = " ".join(x["text"] for x in srt_dictionary)
        if len(encoding.encode(srt_text)) < TOKEN_LIMIT:
            return f"Here are the video subtitles: \"{srt_text}\""
        return f"Here are the video subtitles: \"{srt_text}\""[:TOKEN_LIMIT]
    except Exception as error:
        print("ERROR", error)
        return "The video does not have any subtitles"

app/funcs.py
programmingninjas/GPT-3.5-ON-STEROIDS