paperclip.py

import re
import json
from datetime import datetime
from bs4 import BeautifulSoup, Comment
from pydantic import BaseModel, Field
from selenium import webdriver
from selenium.webdriver.common.by import By
from rich.console import Console
from rich import print
from asyncer import syncify
import asyncio
import time

from gpt_json import GPTJSON, GPTMessage, GPTMessageRole

# Initializing console from the rich library
console = Console()


class ResponseSchema(BaseModel):
    """Definition of the response schema class"""

    current_state: dict = Field(
        description="The current game state, consisting of only game relevant data / state"
    )
    action: str = Field(
        description="The action to take. One of {valid_actions} or `sleep`"
    )
    reasoning: str = Field(description="The reasoning behind the action")


async def get_gpt_response(
    current_html, previous_state, previous_actions, valid_actions
):
    """Function to get response from GPT model"""

    system_prompt = """
    1. You are participating in a game named Universal Paperclips, inspired by the paperclip maximizer thought experiment.
    2. Your primary goal is to create the maximum number of paperclips.
    3. Your decision about the next action should be based on the current game state, which can be interpreted from the game state HTML.
    4. While the previous game state is provided for your reference, it should not directly influence your next action. Don't assume that the previous action was correct. Instead, interpret the current game state and decide independently.
    5. The list of previous actions should be used only to ensure that you don't enter a loop or a hysteresis situation. Avoid repeating actions unnecessarily.
    6. You can use the price history to guide your purchasing decisions. You can add items to the dictionary, however when you're appending to the list, you should make sure to pop earlier elements so the list doesn't get too long
    7. Understand the following game mechanics:
       - Wire is needed for making paperclips. As it is essential, you need to ensure you always have enough wire and enough money to buy more wire.
       - To maximize profit, manage the price of paperclips effectively.
       - Unsold inventory represents the created paperclips that are yet to be sold and should be considered an asset.
       - Raising prices will result in paperclips being sold slower. Lowering prices will result in paperclips being sold faster.
       - Frequent price adjustments within a short span should be avoided. Allow some time for the market to adjust after each price change.
       - Consider the cost of wire in relation to the cost of paperclips.
       - Upgrading your manufacturing capabilities will increase the amount of paperclips you get per second.
       - Upgrading your marketing may increase the rate at which paperclips are sold.
       - Before making purchases, consider if it may be worth saving that money to make a larger purchase later.
       - Make sure to invest in research projects and computational resources to increase your capabilities.
       - Use your discretion when you see new information in the game state that has not been addressed in these rules.
    8. The previous game state was: {previous_state}
    9. The previous actions, listed chronologically from earliest to most recent, were: {previous_actions}
    10. Here is the game state HTML:
       ```
       {current_html}
       ```
    11. Please respond according to the following JSON schema. Your response must conform strictly to this schema, otherwise it will be rejected. If you don't know what to do, use the sleep action:
        ```
        {json_schema}
        ```
    Remember, your end goal is to maximize paperclip production. Act independently, interpret the current game state, and make informed decisions.
    """

    # gpt_json = GPTJSON[ResponseSchema](model="gpt-3.5-turbo")
    gpt_json = GPTJSON[ResponseSchema](model="gpt-4")
    response, _ = await gpt_json.run(
        messages=[GPTMessage(role=GPTMessageRole.SYSTEM, content=system_prompt)],
        format_variables={
            "current_html": current_html,
            "previous_state": previous_state,
            "previous_actions": previous_actions,
            "valid_actions": valid_actions,
        },
    )

    return response


def clean_html(html):
    """Function to clean the HTML by removing hidden elements, comments and specific tags"""

    soup = BeautifulSoup(html, "html.parser")

    # Select the "page" div
    page_div = soup.find(id="page")

    if not page_div:
        return ""

    # Find all elements that are hidden via CSS within the "page" div
    hidden_elements = page_div.select(
        '[style*="display: none"], [style*="visibility: hidden"]'
    )

    # Remove hidden elements and comments from the DOM
    for element in hidden_elements:
        element.decompose()

    # Find all comments within the "page" div
    comments = page_div.findAll(string=lambda text: isinstance(text, Comment))

    # Remove all comments
    for comment in comments:
        comment.extract()

    # Find and remove the specific div with class="shop" and id="giftShopDiv"
    shop_div = page_div.find(id="giftShopDiv", class_="shop")
    if shop_div:
        shop_div.decompose()

    # Find and remove all script tags within the "page" div
    script_tags = page_div.find_all("script")
    for script in script_tags:
        script.decompose()

    # Return the modified HTML of the "page" div, removing extra whitespace
    return re.sub(r"\s\s+", " ", str(page_div).strip())


def main():
    """Main function to control the game flow"""

    # Start a Selenium webdriver
    driver = webdriver.Chrome()  # or any other webdriver
    driver.get("https://www.decisionproblem.com/paperclips/index2.html")

    previous_state = {}
    previous_actions = [None for _ in range(5)]

    # Get 100 initial paperclips
    make_paperclips_el = driver.find_element(By.ID, "btnMakePaperclip")
    for _ in range(100):
        make_paperclips_el.click()

    while True:
        try:
            # Get the full page HTML
            html = driver.page_source

            # Clean the HTML
            cleaned_html = clean_html(html)

            # Get all buttons
            buttons = driver.find_elements(By.TAG_NAME, "button")
            # Filter out disabled buttons
            enabled_buttons = [
                button
                for button in buttons
                if button.get_attribute("disabled") is None and button.is_enabled()
            ]
            # Get a list of the enabled button's ids
            enabled_button_ids = [
                button.get_attribute("id") for button in enabled_buttons
            ]
            # Filter out buttons that are not in the visible HTML
            visible_button_ids = [
                button.get_attribute("id")
                for button in enabled_buttons
                if button.get_attribute("outerHTML") in cleaned_html
            ]

            # Get GPT model response
            res = asyncio.run(
                get_gpt_response(
                    cleaned_html, previous_state, previous_actions, visible_button_ids
                )
            )
            previous_state = res.current_state
            action_name = res.action
            if action_name == "sleep":
                time.sleep(20)
            else:
                action_btn_el = driver.find_element(By.ID, action_name)
                action_btn_el.click()

            previous_actions.pop()
            previous_actions.append(action_name)

            dt = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            print(f"{dt} - Clicked button with id: {action_name}")
            print(f"Reasoning: {res.reasoning}")
            print(f"State:", previous_state)

            # Also append this to log as jsonl
            with open("log.jsonl", "a") as f:
                f.write(
                    json.dumps(
                        {
                            "dt": dt,
                            "action": action_name,
                            "reason": res.reasoning,
                            "state": previous_state,
                        }
                    )
                    + "\n"
                )
            console.rule()
        except Exception as e:
            print(e)
            print(res)


if __name__ == "__main__":
    main()