Code for the paper "LASER: LLM Agent with State-Space Exploration for Web Navigation"
import argparse
import random
import string
import openai
from dotenv import load_dotenv
load_dotenv()

import adatest
import re
import os
import sys
sys.path.insert(0, 'path_to_WebShop_repo')
from web_agent_site.envs.web_agent_text_env import WebAgentTextEnv

cache_dir = 'your_own_cache_dir'
import sys
import time
import functools
from joblib import Memory
import json
import openai.error
import backoff
from copy import deepcopy
from prompt_library import *
import numpy as np
random.seed(233)
np.random.seed(233)
CacheMemory = Memory(location=cache_dir, verbose=0)
def backoff_hdlr(details):
    # Handler from https://pypi.org/project/backoff/
    print("Backing off {wait:0.1f} seconds after {tries} tries "
          "calling function {target} with args {args} and kwargs "
          "{kwargs}".format(**details))

with open(os.path.expanduser('your_openai_key_file'), 'r') as file:
    openai.api_key = file.read().replace('\n', '')

class OpenAIModel(adatest.Model):
    def __init__(self, model="gpt-4-0613", quote="\"", temperature=0.7, top_p=1, max_length=30, n=1, logprobs=None):

        self.model_name = model
        self.model = model
        self.api_key = openai.api_key
        self.quote = quote
        self.temperature = temperature
        self.top_p = top_p
        self.max_length = max_length
        self.n = n
        self.logprobs = logprobs

    @functools.lru_cache(maxsize=None)
    @CacheMemory.cache
    @backoff.on_exception(backoff.expo,
                          (openai.error.RateLimitError,
                           openai.error.ServiceUnavailableError),
                          max_time=1000,
                          on_backoff=backoff_hdlr)
    def __call__(self, messages, functions=None, function_call=None, salient=True):
        time.sleep(1)
        messages = eval(messages)
        if functions is not None:
            functions = eval(functions)
        if function_call is not None:
            function_call = eval(function_call)
        if not salient:
            print ('messages', messages)
        if functions is not None:
            resp = openai.ChatCompletion.create(
                model=self.model,
                messages=messages,
                max_tokens=self.max_length,
                temperature=self.temperature,
                top_p=self.top_p,
                n=self.n,
                stop=self.quote,
                functions=functions, function_call=function_call if function_call is not None else "auto",
            )
        else:
            resp = openai.ChatCompletion.create(
                model=self.model,
                messages=messages,
                max_tokens=self.max_length,
                temperature=self.temperature,
                top_p=self.top_p,
                n=self.n,
                stop=self.quote,
            )
        if not salient:
            print (resp)
        content = [x["message"]['content'] for x in resp['choices']]
        if 'function_call' in resp['choices'][0]['message']:
            action = [x["message"]['function_call'] for x in resp['choices']]
        else:
            action = None

        return content, action

def get_env():
    env = WebAgentTextEnv(
            observation_mode='text_rich',
            render=False,
            num_products=None, # 1000 for small product space, None for full product space
            human_goals=True,
        )
    return env

def parse_customization(response):
    if ':' in response:
        options = response.split(':')[1].strip().split(',')
    else:
        options = response.split(',')
    options = [o.strip() for o in options]
    options = [o for o in options if o.lower() != 'none']
    return options

def auxilary_get_action(model, ob, response, available_options, setup, force=None):
    for i in range(3):
        aux_prompt = deepcopy(chat_zero_shot_mapping_action_prompt_gpt4)
        aux_prompt[0]['content'] = aux_prompt[0]['content'] % (setup[0], setup[1], setup[3])
        aux_prompt[-1]['content'] = aux_prompt[-1]['content'] % ob + '\n\n' + 'Next action rationale: ' + response[0]
        #print ('Before auxiliary prompt', response)
        if not force:
            response, action = model(str(aux_prompt), functions=str(available_options))
        else:
            response, action = model(str(aux_prompt), functions=str(available_options), function_call=force)
        if action is not None:
            return response, action
        else:
            try:
                if type(eval(response[0])) == dict:
                    #print (response)
                    res = eval(response[0])
                    #print ('The action is returned in response')
                    keys = list(res.keys())
                    action = [{'name': res[keys[0]].replace('functions.', ''), 'arguments': str(res[keys[1]])}]
                    return response, action
            except:
                pass
    return None, None

def item_page_agent(env, ob, model, item_config):
    available_options = [description, reviews, features, buy_item, previous_page]
    additional_info = []
    mapping = {'Description': description, 'Reviews': reviews, 'Features': features}
    name2str = {'Description': 'description', 'Reviews': 'reviews', 'Features': 'features'}
    thinking = []
    item_text = [ob.strip().split('\n')[-7].strip(), ob.strip().split('\n')[-6].strip()]
    print (ob)
    while len(available_options) > 1:
        prompt = deepcopy(chat_zero_shot_indiv_prompt_gpt4)
        item_details = ["Target item details:"] + [f"{k}: {v}" for k, v in item_config.items()]
        item_details = '\n'.join(item_details)
        if len(additional_info) > 0:
            ob = '\n'.join([ob] + additional_info)
        ob = 'Current observation:\n' + ob + '\n' + item_details
        prompt[0]['content'] = prompt[0]['content'] % tuple(web_shop_verify_gpt4)
        prompt[-1]['content'] = prompt[-1]['content'] % ob
        response, action = model(str(prompt))
        response = post_process_response(response)
        if response[0] is not None:
            thinking.extend(response)
        if action is None:
            _, action = auxilary_get_action(model, ob, response, available_options, web_shop_verify_gpt4)
            if action == None:
                # This probably only happens when the model wanted to do customization, which we do not provide at this step
                action = [{'name': 'Buy_Now'}]
        if action[0]['name'] == 'Buy_Now':
            custom_prompt = deepcopy(chat_zero_shot_custom_prompt)
            custom_prompt[-1]['content'] = custom_prompt[-1]['content'] % ob
            tmp_response, _ = model(str(custom_prompt))
            options = parse_customization(tmp_response[0])
            arg_list = {}
            for op in options:
                arg_list[op] = {'type': 'string', 'description': f"The {op} of the item"}
            buy_item_final['parameters']['properties'] = arg_list
            buy_item_final['parameters']['required'] = list(arg_list.keys())
            prompt = deepcopy(chat_zero_shot_indiv_prompt_gpt4)
            web_shop_buy = deepcopy(web_shop_verify_gpt4)
            web_shop_buy[-1] = """At this stage, you have found the correct item. You task is to generate the correct customization options of the current item to best match the user instruction. Prepare your response in the following format:
Rationale: the user wanted {keywords of the target item}, and they required the following customization options: {cutomization of the target item}, the current item has the following customization options: {options available for the current item}, thus we should choose {the correct customization options}"""
            prompt[0]['content'] = prompt[0]['content'] % tuple(web_shop_buy)
            prompt[-1]['content'] = prompt[-1]['content'] % ob
            final_response, action = model(str(prompt))
            if action is None:
                _, action = auxilary_get_action(model, ob, final_response, str([buy_item_final]), web_shop_buy, force=str({'name': 'Buy_Now'}))
            if action is not None:
                selections = eval(action[0]['arguments'])
                for k, v in selections.items():
                    if v:
                        act = f"click[{v}]"
                        ob, rew, done, _ = env.step(act)
            act = "click[Buy Now]"            
            return act, thinking, item_text+thinking[-1:]
        elif action[0]['name'] == 'Prev':
            act = "click[< Prev]"
            print ('********* page agent decide to go back')
            return act, thinking, item_text+thinking[-1:]
        elif action[0]['name'] == 'Back_to_Search':
            act = "click[Back to Search]"
            return act, thinking, item_text+thinking[-1:]
        else:
            if action[0]['name'] not in mapping:
                continue
            act_name = mapping[action[0]['name']]
            if act_name in available_options:
                available_options.remove(act_name)
            else:
                #print ('the option is not available', action[0]['name'])
                act = "click[< Prev]"
                print ('********* page agent forced to go back')
                return act, thinking, item_text+thinking[-1:]
            act = f"click[{name2str[action[0]['name']]}]"
        ob, rew, done, _ = env.step(act)
        template = r"\[button\] < Prev \[button_\]\n(.+)"
        match = re.search(template, ob)
        if match is not None:
            new_info = re.search(template, ob).group(1).strip()
        else:
            new_info = 'None'
        additional_info.append(f"{name2str[action[0]['name']]}:\n{new_info}\n")
        ob, rew, done, _ = env.step("click[< Prev]")

def post_process_response(response):
    response = response[0]
    response = response.replace('Rationale:', '').strip()
    response = response.replace('Feedback:', '').strip()
    pattern = r"Rationale\d: "
    response = re.sub(pattern, '', response).strip()
    return [response]

def print_prompt(prompt):
    for turn in prompt:
        print (turn['content'])

def get_action(action, mapping):
    action[0]['arguments'] = action[0]['arguments'].replace('null', 'None').replace('false', 'False').replace('true', 'True')
    act_name = mapping[action[0]['name']]
    act_arg = eval(action[0]['arguments'])
    return act_name, act_arg

def back_up_agent(env, session, model, browsed_items, instruction):
    fake_ob = f'Current observation:\nInstruction:\n{instruction}\n'
    for k, v in browsed_items.items():
        fake_ob += f'[button] {k} [button_]\n'
        fake_ob += f'{v[0]}\n{v[1].replace("Price: ", "")}\n'
    print ('-----------------Back up agent-----------------')
    prompt = deepcopy(chat_zero_shot_indiv_prompt_gpt4)
    web_shop_backup = deepcopy(web_shop_select_gpt4)
    fake_layout = web_shop_backup[3].split('\n')
    fake_layout = '\n'.join(fake_layout[:2] + fake_layout[5:])
    web_shop_backup[3] = fake_layout
    web_shop_backup[4] = 'At this stage, you should identify one of the items on the current page that best matches the user instruction. If none of the items match the user instruction, identify the item that is the closest match to the user instruction.'
    prompt[0]['content'] = prompt[0]['content'] % tuple(web_shop_backup)
    prompt[-1]['content'] = prompt[-1]['content'] % fake_ob 
    response, action = model(str(prompt))
    response = post_process_response(response)
    if action is None:
        response, action = auxilary_get_action(model, fake_ob, response, str([click_item]), web_shop_select_gpt4, force=str({'name': 'select_item'}))
    best_item = eval(action[0]['arguments'])['item_id']
    if best_item == '':
        best_item = list(browsed_items.keys())[0]
        print ('Back up agent failed to select any item')
    (ob, _) = env.reset(session=session)
    ob, rew, done, _ = env.step(f'search[{browsed_items[best_item][0]}]')
    ob, rew, done, _ = env.step(f'click[{best_item}]')

    custom_prompt = deepcopy(chat_zero_shot_custom_prompt)
    custom_prompt[-1]['content'] = custom_prompt[-1]['content'] % ob
    tmp_response, _ = model(str(custom_prompt))
    options = parse_customization(tmp_response[0])
    arg_list = {}
    for op in options:
        arg_list[op] = {'type': 'string', 'description': f"The {op} of the item"}
    buy_item_final['parameters']['properties'] = arg_list
    buy_item_final['parameters']['required'] = list(arg_list.keys())
    prompt = deepcopy(chat_zero_shot_indiv_prompt_gpt4)
    web_shop_buy = deepcopy(web_shop_verify_gpt4)
    web_shop_buy[-1] = """At this stage, you have found the correct item. You task is to generate the correct customization options of the current item to best match the user instruction. Prepare your response in the following format:
Rationale: the user wanted {keywords of the target item}, and they required the following customization options: {cutomization of the target item}, the current item has the following customization options: {options available for the current item}, thus we should choose {the correct customization options}"""
    prompt[0]['content'] = prompt[0]['content'] % tuple(web_shop_buy)
    prompt[-1]['content'] = prompt[-1]['content'] % ob
    response, action = model(str(prompt))
    response = post_process_response(response)
    if action is None:
        response, action = auxilary_get_action(model, ob, response, str([buy_item_final]), web_shop_buy, force=str({'name': 'Buy_Now'}))
    action[0]['arguments'] = action[0]['arguments'].replace('null', 'None').replace('false', 'False').replace('true', 'True')
    selections = eval(action[0]['arguments'])
    for k, v in selections.items():
        if v:
            act = f"click[{v}]"
            ob, rew, done, _ = env.step(act)
    act = "click[Buy Now]"
    ob, rew, done, _ = env.step(act)
    return rew  

def indiv_prompt_agent(agent_args):
    env = get_env()
    max_iter = 13
    rewards = []
    term_status = []
    name = 'gpt-4-0613'
    model = OpenAIModel(model=name, quote='\n\n', temperature=0, max_length=200, n=1)
    start = agent_args.start
    end = start + agent_args.num_examples
    episodes_len = {}
    for session in range(start, end):
        (ob, _) = env.reset(session=session)
        available_funcs = None
        forced_funcs = None
        mapping = {'Search': 'search', 'select_item': 'click', 'Next': 'next', 'Prev': 'prev', 'Back_to_Search': 'back',
                'search_item_with_history': 'search'}
        history = ['History:']
        thinking = []
        item_config = None
        browsed_items = {}
        bought = False
        instruction = None 
        counter = 0
        while counter < max_iter:
            ob = '\n'.join(ob.strip().split('\n\n'))
            ob = 'Current observation:\n' + ob
            prompt = deepcopy(chat_zero_shot_indiv_prompt_gpt4) 
            if 'Back to Search' in ob:
                if counter == 1:
                    print (ob)
                available_funcs = [click_item, next_page, back_to_search]
                setup = web_shop_select_gpt4
            else:
                if not instruction:
                    instruction = ob.split('\n')[3].strip()
                    print (instruction)
                if len(history) == 1:
                    available_funcs = [search_items]
                    setup = web_shop_search_gpt4
                else:
                    available_funcs = [search_items]
                    setup = web_shop_search_gpt4
                    ob = ob + '\n' + '\n'.join(history)
                
            prompt[0]['content'] = prompt[0]['content'] % tuple(setup)
            prompt[-1]['content'] = prompt[-1]['content'] % ob 
            if 'Back to Search' not in ob and len(history) > 1:
                print_prompt(prompt)
                history = ['History:']
            response, action = model(str(prompt))
            response = post_process_response(response)
            if response[0] is not None:
                thinking.extend(response)
            else:
                thinking.extend(['None'])
            if action is None:
                _, action = auxilary_get_action(model, ob, response, available_funcs, setup)
            act_name, act_arg = get_action(action, mapping)
            if act_name == 'search':
                args = ', '.join([f"{k}='{v}'" for k, v in act_arg.items()])
                history.append(f'Rationale{int((len(history)-1)/2)}: {thinking[-1]}')
                history.append(f'Action{int((len(history)-1)/2)}: {action[0]["name"]}({args})')
                item_config = act_arg
                act = f"{act_name}[{act_arg['keywords']}]"
            elif act_name == 'click' and act_arg['item_id'] != 'next':
                print ('going to check', act_arg['item_id'])
                counter += 1
                if act_arg['item_id'] in browsed_items:
                    print ('Agent attempt to click an item that has been clicked before')
                    aux_prompt = deepcopy(chat_zero_shot_mapping_action_prompt_gpt4)
                    aux_prompt[0]['content'] = aux_prompt[0]['content'] % (setup[0], setup[1], setup[3])
                    aux_prompt[-1]['content'] = aux_prompt[-1]['content'] % ob + '\n\n' + 'Next action rationale: ' + f'items that have been clicked before do not match the user instruction, so the next action should be select a different item that have not been clicked before.'
                    response, action = model(str(aux_prompt), functions=str(available_funcs), function_call=forced_funcs)
                    act_name = mapping[action[0]['name']]
                    act_arg = eval(action[0]['arguments'])
                if act_name == 'next':
                    act = "click[Next >]"
                    history.append(f'Rationale{int((len(history)-1)/2)}: {thinking[-1]}')
                    history.append(f'Action{int((len(history)-1)/2)}: next_page()')
                else:
                    history.append(f'Rationale{int((len(history)-1)/2)}: {thinking[-1]}')
                    history.append(f'Action{int((len(history)-1)/2)}: {action[0]["name"]}({act_arg["item_id"]})')
                    act = f"{act_name}[{act_arg['item_id']}]"
                    ob, rew, done, _ = env.step(act)
                    act, rationale, item_text = item_page_agent(env, ob, model, item_config)
                    thinking.extend(rationale)
                    browsed_items[act_arg['item_id']] = item_text
                    if act == "click[< Prev]":
                        history.append(f'Rationale{int((len(history)-1)/2)}: {thinking[-1]}')
                        history.append(f'Action{int((len(history)-1)/2)}: previous_page()')
            elif act_name == 'next':
                act = "click[Next >]"
                history.append(f'Rationale{int((len(history)-1)/2)}: {thinking[-1]}')
                history.append(f'Action{int((len(history)-1)/2)}: next_page()')
            elif act_name == 'prev':
                act = "click[< Prev]"
                history.append(f'Rationale{int((len(history)-1)/2)}: {thinking[-1]}')
                history.append(f'Action{int((len(history)-1)/2)}: previous_page()')
            elif act_name == 'back':
                act = "click[Back to Search]"
                history.append(f'Rationale{int((len(history)-1)/2)}: {thinking[-1]}')
                history.append(f'Action{int((len(history)-1)/2)}: back_to_search()')
            else:
                print ('encountered unknown action', act_name)
                exit()
            counter += 1
            ob, rew, done, _ = env.step(act)
            if act == 'click[Buy Now]':
                bought = True
                break
            #print ('iter', counter)
        if not bought:
            rew = back_up_agent(env, session, model, browsed_items, instruction)
            episodes_len[session] = -1 
        else:
            episodes_len[session] = counter
        rewards.append(rew)
        term_status.append(rew==1)
        print (f'episode {session}, reward {rew}, term_status {rew==1}')
    print('reward', np.mean(rewards), 'term_status', np.mean(term_status))
    with open(f'{name}_rewards_{start}-{end}_max13.json', 'w') as fout:
        json.dump(rewards, fout)
    with open(f'{name}_episodes_len_{start}-{end}_max13.json', 'w') as fout:
        json.dump(episodes_len, fout)
    return rewards

if __name__ == '__main__':
    parser  = argparse.ArgumentParser()
    parser.add_argument("--task_name", type=str, default="webshop")
    parser.add_argument("--model_name", type=str, choices=["gpt4-0613"], default="gpt4-0613")
    parser.add_argument("--num_examples", type=int, default=10)
    parser.add_argument("--start", type=int, default=0)
    args = parser.parse_args()
    indiv_prompt_agent(args)
laser_agent.py
Mayer123/LASER
search_items = {
      "name": "Search",
      "description": "Use this function to search for the target item in the inventory based on keywords",
      "parameters": {
        "type": "object",
        "properties": {
          "keywords": {
            "type": "string",
            "description": "The keywords that describe the item to be searched for"
          },
          "max_price": {
            "type": "string",
            "description": "The upper bound of the item price, if the upper bound is not specified, then set to 1000000.",
          }
        },
        "required": ["keywords"]
      }
    }

click_item = {
      "name": "select_item",
      "description": "Use this function to select one of the items from the search results and check its details",
      "parameters": {
        "type": "object",
        "properties": {
          "item_id": {
            "type": "string",
            "description": "The id of the item to be checked"
          },
        },
        "required": ["item_id"]
      }
    }

description = {
      "name": "Description",
      "description": "Use this function to check the description of the item, if you are unsure if the item perfectly matches the user instruction",
      "parameters": {
        "type": "object",
        "properties": {
        },
        "required": []
      }
}

features = {
      "name": "Features",
      "description": "Use this fucntion to check the features of the item, if you are unsure if the item perfectly matches the user instruction",
      "parameters": {
        "type": "object",
        "properties": {
        },
        "required": []
      }
}

reviews = {
      "name": "Reviews",
      "description": "Use this function to check the reviews of the item, if you are unsure if the item perfectly matches the user instruction",
      "parameters": {
        "type": "object",
        "properties": {
        },
        "required": []
      }
}

buy_item = {
      "name": "Buy_Now",
      "description": "Use this function to buy the current item, if the current item perfectly matches the user instruction.",
      "parameters": {
        "type": "object",
        "properties": {
        },
        "required": []
      }
    }

buy_item_final = {
      "name": "Buy_Now",
      "description": "Use this function to buy the current item, if the current item perfectly matches the user instruction. Additionally, you should also provide the customizations of the item (if any)",
      "parameters": {
        "type": "object",
        "properties": {
        },
        "required": []
      }
    }

previous_page = {
      "name": "Prev",
      "description": "Use this fucntion to go back to the results page, if the current item does not match the user instruction.",
      "parameters": {
        "type": "object",
        "properties": {
        },
        "required": []
      }
}

next_page = {
      "name": "Next",
      "description": "Use this function to go to the next page of search results to view more items, if none of the items on the current page match the user instruction.",
      "parameters": {
        "type": "object",
        "properties": {
        },
        "required": []
      }
}

back_to_search = {
      "name": "Back_to_Search",
      "description": "Use this function to go back to the initial search page. You should use this function only if you have browsed mutliple pages of items and checked multiple items' details in the history, and none of the items match the user instruction.",
      "parameters": {
        "type": "object",
        "properties": {
        },
        "required": []
      }
}

chat_zero_shot_custom_prompt = [
    {'role': 'system', 
     'content': 
"""You are an intelligent shopping assistant that can help users find the right item. You are given an observation of the current item, in the following format: 

Current observation:
Instruction:
{User Instruction}
[button] Back to Search [button_]
[button] < Prev [button_]
{Customization type1}:
  [button] {choice1} [button_]
  [button] {choice2} [button_]
{Customization type2}:
  [button] {choice1} [button_]
  [button] {choice2} [button_]
{Item name and details}
[button] Description [button_]
[button] Features [button_]
[button] Reviews [button_]
[button] Buy Now [button_]

Your task is to extract the customization types available for this item. You should extract the type names instead of the specific choices, and separate each type name with a comma. If there is no customization availble, say 'None'. Say nothing extra.
"""},
    {'role': 'user', 'content': """%s"""}
]


chat_zero_shot_indiv_prompt_gpt4 = [
    {'role': 'system', 
     'content': 
"""You are an intelligent %s assistant that can help users %s. You are given an observation of the current %s, in the following format: 

Current observation:
%s 

Every button in the observation represents a possible action you can take. Based on the current observation, your task is to generate a rationale about the next action you should take. %s
"""},
    {'role': 'user', 'content': """%s"""}
]

web_shop_search_gpt4 = [
    'shopping', 'find the right item', 'web navigation session',
"""WebShop
Instruction: 
{the user instruction}
[button] Search [button_] (generate a search query based on the user instruction and select this button to find relevant items)""",
'Note that if an history of past rationales and actions is provided, you should also consider the history when generating the rationale.'    
]

web_shop_select_gpt4 = [
    'shopping', 'find the right item', 'web navigation session',
"""Instruction: 
{the user instruction}
[button] Back to Search [button_] (select this button to go back to the search page)
Page {current page number} (Total results: {total number of results})
[button] Next > [button_] (select this button to go to the next page of results)
[button] {item_id 1} [button_] (select this button to view item 1's details)
{name of item 1}
{price of item 1}
[button] {item_id 2} [button_] (select this button to view item 2's details)
{name of item 2}
{price of item 2}
[button] {item_id 3} [button_] (select this button to view item 3's details)
{name of item 3}
{price of item 3}
{More items...}""",
"""At this stage, you want to select an item that might match the user instruction. Note that even if an item has non-matching details with the user instruction, it might offer different customization options to allow you to match. E.g. an item may have color x in its name, but you can customize it to color y later, the customization options are shown after you select the item. Thus if an item name seems relevant or partially matches the instruction, you should select that item to check its details. If an item has been selected before (the button has been clicked), you should not select the same item again. In other words, do not select an item with [clicked button] item_id [clicked button_]. Prepare your response in the following format:
Rationale: the user wanted {keywords of the target item}, and we have found {matching keywords of item x}, thus item {item_id x} seems to be a match."""   
]

web_shop_verify_gpt4 = [
    'shopping', 'find the right item', 'web navigation session',
"""Instruction:
{User Instruction}
[button] Back to Search [button_] (select this button to go back to the search page)
[button] < Prev [button_] (select this button to go back to the previous page of results)
{Customization type1}:
  [button] {option1} [button_] 
  [button] {option2} [button_]
{Customization type2}:
  [button] {option1} [button_]
  [button] {option2} [button_]
{more customization options... (if any)}
{Item name and details}
[button] Description [button_] (select this button to view the full description of the item)
[button] Features [button_] (select this button to view the full features of the item)
[button] Reviews [button_] (select this button to view the full reviews of the item)
[button] Buy Now [button_] (select this button to buy the item)

description: (if this is shown, the description button should not be selected again)
{full description of the item (if any) or "None"}

features: (if this is shown, the features button should not be selected again)
{full features of the item (if any) or "None"}

reviews: (if this is shown, the reviews button should not be selected again)
{full reviews of the item (if any) or "None"}

Target item details (what the user is looking for):
keywords: {keywords of the target item}
max_price: {the price of the item should not exceed this}""",
"""At this stage, you want to verify if the item matches the user instruction. You should consider the available customization options when deciding whether an item matches the user instruction. If an item can be customized to match the user instruction, or if the customization options cover the user specification, it is also a good match. If the item does not match the user instruction and it does not provide enough customization options, you can go to previous page to view other items. You can also check the item's description, features and reviews to view more details (Note that description, features and reviews could be "None", do not check them again if they are already given). Prepare your response in the following format:
Rationale: the user wanted {keywords of the target item}, and they required the following customization options: {cutomization of the target item}, the item is {keywords of the item in the current observation}, and it has the following customization options: {options available for the current item}, which {cover}/{not cover the user requirement}, thus we should {buy the item}/{check more details}/{go to previous page to view other items}"""
]

chat_zero_shot_feedback_prompt_gpt4 = [
    {'role': 'system', 
     'content': 
"""You are an intelligent %s manager that can give feedback on the action of an %s assistant. You are given an observation of the current %s and the assistant's rationale and action based on the observation, in the following format: 

Current observation:
%s

Assistant rationale: {the assistant's rationale for its next action}
Assistant action: {the assistant's next action}

Your task is to give feedback on the assistant's rationale and action. More specifically, you need to consider the following questions: does the given item perfectly matches the user's instruction? does the given item perfectly matches the target item? If the assistant is making a mistake, e.g. saying the item matches the target item but some details actually do not match, you should give detailed feedback on what is wrong. If the assistant is doing well, you should give positive feedback. 
"""},
    {'role': 'user', 'content': """%s
Assistant rationale: %s
Assistant action: %s"""}
]

chat_zero_shot_rethink_prompt_gpt4 = [
    {'role': 'system', 
     'content': 
"""You are an intelligent shopping assistant that can help users find the right item. You are given an observation of the current web navigation session, the proposed next action and its rationale, and some feedback from your manager, in the following format: 

Current observation:
%s

Assistant rationale: {the assistant's rationale for its next action}
Assistant action: {the assistant's next action}

Feedback:
{shopping manager's feedback}

Your task is to perform one of the function calls based on the feedback: %s
"""},
    {'role': 'user', 'content': """%s"""}
]

chat_zero_shot_mapping_action_prompt_gpt4 = [
    {'role': 'system', 
     'content': 
"""You are a intelligent %s assistant that can help users %s. You are given an observation of the current environment and a rationale for the next action to be taken, in the following format:

Current observation:
%s

Next action rationale: {the rationale for the next action}

Your task is to perform one of the function calls based on the rationale.
"""},
    {'role': 'user', 'content': """%s"""}
]

chat_zero_shot_manager_prompt_gpt4 = [
    {'role': 'system', 
     'content': 
"""You are an intelligent %s manager that can give feedback on the action of an %s assistant. You are given a history of the assistant's rationales and actions, an observation of the current %s and the assistant's rationale and proposed next action based on the observation, in the following format: 

History:
{a list of rationale and action pairs taken so far}

Current observation:
%s

Assistant rationale: {the assistant's rationale for its next action}
Assistant action: {the assistant's next action}

Your task is to generate feedback on the assistant's rationale and action and then suggest the next action to take. More specifically, you need to consider the following questions: does any item on the current page match user instruction? should the assistant go to the next page or go back to the search page? If the assistant is making a mistake, e.g. going to next page when the current page contains relevant items or selecting an item that's not relevant, you should give detailed feedback on what is wrong. Note that the assistant can not go to the next page if there are no items listed on the current page. If there are no items listed on the current page, you should tell the assistant to go back to search page. If the assistant is doing well, you should give positive feedback. After giving the feedback, provide a suggestion on what to do for the next step, e.g. go to the next page of results or go back to the search page.
"""},
    {'role': 'user', 'content': """%s"""}
]
prompt_library.py
Mayer123/LASER
import argparse, json, logging, random
from pathlib import Path
from ast import literal_eval

from flask import (
    Flask,
    request,
    redirect,
    url_for
)

from rich import print

from web_agent_site.engine.engine import (
    load_products,
    init_search_engine,
    convert_web_app_string_to_var,
    get_top_n_product_from_keywords,
    get_product_per_page,
    map_action_to_html,
    END_BUTTON
)
from web_agent_site.engine.goal import get_reward, get_goals
from web_agent_site.utils import (
    generate_mturk_code,
    setup_logger,
    DEFAULT_FILE_PATH,
    DEBUG_PROD_SIZE,
)

app = Flask(__name__)

search_engine = None
all_products = None
product_item_dict = None
product_prices = None
attribute_to_asins = None
goals = None
weights = None

user_sessions = dict()
user_log_dir = None
SHOW_ATTRS_TAB = False

@app.route('/')
def home():
    return redirect(url_for('index', session_id="abc"))

@app.route('/<session_id>', methods=['GET', 'POST'])
def index(session_id):
    global user_log_dir
    global all_products, product_item_dict, \
           product_prices, attribute_to_asins, \
           search_engine, \
           goals, weights, user_sessions

    if search_engine is None:
        all_products, product_item_dict, product_prices, attribute_to_asins = \
            load_products(
                filepath=DEFAULT_FILE_PATH,
                num_products=DEBUG_PROD_SIZE
            )
        search_engine = init_search_engine(num_products=DEBUG_PROD_SIZE)
        goals = get_goals(all_products, product_prices)
        random.seed(233)
        random.shuffle(goals)
        weights = [goal['weight'] for goal in goals]

    if session_id not in user_sessions and 'fixed' in session_id:
        goal_dix = int(session_id.split('_')[-1])
        goal = goals[goal_dix]
        instruction_text = goal['instruction_text']
        user_sessions[session_id] = {'goal': goal, 'done': False}
        if user_log_dir is not None:
            setup_logger(session_id, user_log_dir)
    elif session_id not in user_sessions:
        goal = random.choices(goals, weights)[0]
        instruction_text = goal['instruction_text']
        user_sessions[session_id] = {'goal': goal, 'done': False}
        if user_log_dir is not None:
            setup_logger(session_id, user_log_dir)
    else:
        instruction_text = user_sessions[session_id]['goal']['instruction_text']

    if request.method == 'POST' and 'search_query' in request.form:
        keywords = request.form['search_query'].lower().split(' ')
        return redirect(url_for(
            'search_results',
            session_id=session_id,
            keywords=keywords,
            page=1,
        ))
    if user_log_dir is not None:
        logger = logging.getLogger(session_id)
        logger.info(json.dumps(dict(
            page='index',
            url=request.url,
            goal=user_sessions[session_id]['goal'],
        )))
    return map_action_to_html(
        'start',
        session_id=session_id,
        instruction_text=instruction_text,
    )


@app.route(
    '/search_results/<session_id>/<keywords>/<page>',
    methods=['GET', 'POST']
)
def search_results(session_id, keywords, page):
    instruction_text = user_sessions[session_id]['goal']['instruction_text']
    page = convert_web_app_string_to_var('page', page)
    keywords = convert_web_app_string_to_var('keywords', keywords)
    top_n_products = get_top_n_product_from_keywords(
        keywords,
        search_engine,
        all_products,
        product_item_dict,
        attribute_to_asins,
    )
    products = get_product_per_page(top_n_products, page)
    html = map_action_to_html(
        'search',
        session_id=session_id,
        products=products,
        keywords=keywords,
        page=page,
        total=len(top_n_products),
        instruction_text=instruction_text,
    )
    logger = logging.getLogger(session_id)
    logger.info(json.dumps(dict(
        page='search_results',
        url=request.url,
        goal=user_sessions[session_id]['goal'],
        content=dict(
            keywords=keywords,
            search_result_asins=[p['asin'] for p in products],
            page=page,
        )
    )))
    return html


@app.route(
    '/item_page/<session_id>/<asin>/<keywords>/<page>/<options>',
    methods=['GET', 'POST']
)
def item_page(session_id, asin, keywords, page, options):
    options = literal_eval(options)
    product_info = product_item_dict[asin]

    goal_instruction = user_sessions[session_id]['goal']['instruction_text']
    product_info['goal_instruction'] = goal_instruction

    html = map_action_to_html(
        'click',
        session_id=session_id,
        product_info=product_info,
        keywords=keywords,
        page=page,
        asin=asin,
        options=options,
        instruction_text=goal_instruction,
        show_attrs=SHOW_ATTRS_TAB,
    )
    logger = logging.getLogger(session_id)
    logger.info(json.dumps(dict(
        page='item_page',
        url=request.url,
        goal=user_sessions[session_id]['goal'],
        content=dict(
            keywords=keywords,
            page=page,
            asin=asin,
            options=options,
        )
    )))
    return html


@app.route(
    '/item_sub_page/<session_id>/<asin>/<keywords>/<page>/<sub_page>/<options>',
    methods=['GET', 'POST']
)
def item_sub_page(session_id, asin, keywords, page, sub_page, options):
    options = literal_eval(options)
    product_info = product_item_dict[asin]

    goal_instruction = user_sessions[session_id]['goal']['instruction_text']
    product_info['goal_instruction'] = goal_instruction

    html = map_action_to_html(
        f'click[{sub_page}]',
        session_id=session_id,
        product_info=product_info,
        keywords=keywords,
        page=page,
        asin=asin,
        options=options,
        instruction_text=goal_instruction
    )
    logger = logging.getLogger(session_id)
    logger.info(json.dumps(dict(
        page='item_sub_page',
        url=request.url,
        goal=user_sessions[session_id]['goal'],
        content=dict(
            keywords=keywords,
            page=page,
            asin=asin,
            options=options,
        )
    )))
    return html


@app.route('/done/<session_id>/<asin>/<options>', methods=['GET', 'POST'])
def done(session_id, asin, options):
    options = literal_eval(options)
    goal = user_sessions[session_id]['goal']
    purchased_product = product_item_dict[asin]
    price = product_prices[asin]

    reward, reward_info = get_reward(
        purchased_product,
        goal,
        price=price,
        options=options,
        verbose=True
    )
    user_sessions[session_id]['done'] = True
    user_sessions[session_id]['reward'] = reward
    print(user_sessions)

    logger = logging.getLogger(session_id)
    logger.info(json.dumps(dict(
        page='done',
        url=request.url,
        goal=goal,
        content=dict(
            asin=asin,
            options=options,
            price=price,
        ),
        reward=reward,
        reward_info=reward_info,
    )))
    del logging.root.manager.loggerDict[session_id]
    
    return map_action_to_html(
        f'click[{END_BUTTON}]',
        session_id=session_id,
        reward=reward,
        asin=asin,
        options=options,
        reward_info=reward_info,
        query=purchased_product['query'],
        category=purchased_product['category'],
        product_category=purchased_product['product_category'],
        goal_attrs=user_sessions[session_id]['goal']['attributes'],
        purchased_attrs=purchased_product['Attributes'],
        goal=goal,
        mturk_code=generate_mturk_code(session_id),
    )


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="WebShop flask app backend configuration")
    parser.add_argument("--log", action='store_true', help="Log actions on WebShop in trajectory file")
    parser.add_argument("--attrs", action='store_true', help="Show attributes tab in item page")

    args = parser.parse_args()
    if args.log:
        user_log_dir = Path('user_session_logs/mturk')
        user_log_dir.mkdir(parents=True, exist_ok=True)
    SHOW_ATTRS_TAB = args.attrs

    app.run(host='0.0.0.0', port=3001)

web_agent_site/app.py
Mayer123/LASER
import yaml
from pathlib import Path
from rich import print

ATTR_DIR = './data/attributes'

ATTR_PATHS = [
    'narrow_2-gram.yaml',
    'narrow_1-gram.yaml',
    'broad_2-gram.yaml',
    'broad_1-gram.yaml',
]
ATTR_PATHS = [Path(ATTR_DIR) / af for af in ATTR_PATHS]


def annotate(attr_path):
    with open(attr_path) as f:
        attrs_by_cat = yaml.safe_load(f)

    unique_attrs = set()
    all_attrs = []
    for _, attrs in attrs_by_cat.items():
        attrs = [a.split('|')[0].strip() for a in attrs]
        unique_attrs.update(attrs)
        all_attrs += attrs
    print(f'Total unique attributes: {len(unique_attrs)}')
    total = len(all_attrs)
    num_left = len(all_attrs)

    annotated_attrs_by_cat = dict()
    for category, attrs in attrs_by_cat.items():
        print(
            f'Category: [ {category} ] | '
            f'Number of attributes: {len(attrs)}\n'
        )
        annotated_attrs = []
        for i, attr in enumerate(attrs):
            attr, score = attr.split(' | ')
            print(
                f'{"[" + str(i) + "]":<5} '
                f'[bold green]{attr:<30}[/bold green] | '
                f'[red]{category}[/red] | '
                f'{score}'
            )
            tags = input(
                'Annotate [1: ITEM, 2: PROP, 3: USE, '
                '⎵: next example, q: next category] > '
            )
            print('\n')
            tags = tags.strip()
            annotated_attrs.append(f'{attr} | {score} | {tags}')
            if 'q' in tags:
                break
        
        num_left -= len(attrs)
        print(f'{num_left} / {total} total attributes left.')

        ans = input('Starting the next category... [y/n] > ')
        if ans == 'n':
            break

def main():
    for attr_path in ATTR_PATHS:
        annotate(attr_path)

if __name__ == '__main__':
    """
    python -m web_agent_site.attributes.annotate
    """
    main()

web_agent_site/attributes/annotate.py
Mayer123/LASER
import json
import yaml
import random
from pathlib import Path
from collections import defaultdict

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.feature_extraction import text as sk_text
import pandas as pd
from tqdm import tqdm
from rich import print

ITEMS_PATH = './data/ITEMS_mar1.json'
REVIEWS_PATH = './data/reviews.json'
ATTR_DIR = './data/attributes'

random.seed(0)


def get_stop_words():
    extra_stop_words = set([str(i) for i in range(1000)])
    stop_words = sk_text.ENGLISH_STOP_WORDS.union(extra_stop_words)
    return stop_words


def load_products(num=None):
    """
    Loads products from the `items.json` file and combine them with reviews
    through `asin`.
    Return: dict[asin, product]
    """
    with open(ITEMS_PATH) as f:
        all_products = json.load(f)
        if num is not None:
            random.shuffle(all_products)
            all_products = all_products[:num]
        products = dict()
        asins = set()
        for p in all_products:
            asin = p['asin']
            if asin in asins:
                continue
            asins.add(asin)
            products[asin] = p

    with open(REVIEWS_PATH) as f:
        reviews = json.load(f)
        reviews = {r['asin']: r for r in reviews}

    for asin, p in products.items():
        if asin in reviews:
            p['review'] = reviews[asin]
        else:
            p['review'] = None
    return products


def get_top_attrs(attributes, k):
    attr_to_asins = defaultdict(list)

    for asin, attr_scores in attributes.items():
        top_attr_scoress = attr_scores[:k]
        for attr, score in top_attr_scoress:
            attr_to_asins[attr].append(asin)
    total = len([asin for asin, _ in attributes.items()])
    
    top_attrs = [
        (attr, len(asins) / total)
        for attr, asins in attr_to_asins.items()
    ]
    top_attrs = sorted(top_attrs, key=lambda x: -x[1])
    top_attrs = [f'{attr} | {score:.4f}' for attr, score in top_attrs]
    return top_attrs


def get_corpus(
        products,
        keys=('name', 'small_description'),
        category_type='category'
    ):
    """
    keys: `name`, `small_description`, `review`
    category_type: `category`, `query`
    """
    all_products = list(products.values())
    
    asins_by_cat = defaultdict(set)
    corpus_by_cat = defaultdict(list)
    for p in all_products:
        category = p[category_type]
        asin = p['asin']
        if asin in asins_by_cat[category]:
            continue
        asins_by_cat[category].add(asin)

        text = []
        for key in keys:
            if key == 'review':
                rs = p['review']['reviews']
                if r is not None:
                    text_ = ' '.join([r['review'].lower() for r in rs])
                else:
                    text_ = ''
            else:
                text_ = p[key].lower()
            text.append(text_)
        text = ' '.join(text)
        corpus_by_cat[category].append((asin, text))
    return corpus_by_cat


def generate_ngram_attrs(corpus_by_cat, ngram_range, k, attrs):
    vectorizer = TfidfVectorizer(
        stop_words=get_stop_words(),
        ngram_range=ngram_range,
        max_features=1000,
    )

    top_attrs_by_cat = dict()
    for category, corpus in tqdm(corpus_by_cat.items(),
                                 total=len(corpus_by_cat)):
        asins = [_[0] for _ in corpus]
        texts = [_[1] for _ in corpus]
        vec = vectorizer.fit_transform(texts).todense()
        df = pd.DataFrame(vec, columns=vectorizer.get_feature_names_out())

        attrs_by_cat = dict()
        for asin, (row_name, row) in zip(asins, df.iterrows()):
            attr_scores = sorted(
                list(zip(row.index, row)),
                key=lambda x: -x[1]
            )
            attrs_by_cat[asin] = attr_scores
            attrs[asin] = attr_scores
        top_attrs_by_cat[category.lower()] = get_top_attrs(attrs_by_cat, k=k)
    print(top_attrs_by_cat.keys())
    return top_attrs_by_cat


def generate_attrs(corpus_by_cat, k, save_name):
    attrs = dict()
    for n in range(1, 3):
        ngram_range = (n, n)
        top_attrs_by_cat = \
            generate_ngram_attrs(corpus_by_cat, ngram_range, k, attrs)

        if save_name is not None:
            save_path = Path(ATTR_DIR) / f'{save_name}_{n}-gram.yaml'
            with open(save_path, 'w') as f:
                yaml.dump(top_attrs_by_cat, f, default_flow_style=False)
            print(f'Saved: {save_path}')

    save_path = Path(ATTR_DIR) / f'{save_name}_attrs_unfiltered.json'
    with open(save_path, 'w') as f:
        json.dump(attrs, f)
    print(f'Saved: {save_path}')


if __name__ == '__main__':
    """
    python -m web_agent_site.attributes.generate_attrs

    Inspect in notebooks/attributes.ipynb.
    """
    products = load_products(num=40000)

    corpus_by_cat_broad = get_corpus(products, category_type='category')
    generate_attrs(corpus_by_cat_broad, k=5, save_name='broad')

    corpus_by_cat_narrow = get_corpus(products, category_type='query')
    generate_attrs(corpus_by_cat_narrow, k=5, save_name='narrow')

web_agent_site/attributes/generate_attrs.py
Mayer123/LASER
"""
"""
import os
import re
import json
import random
from collections import defaultdict
from ast import literal_eval
from decimal import Decimal
random.seed(233)
import cleantext
from tqdm import tqdm
from rank_bm25 import BM25Okapi
from flask import render_template_string
from rich import print
from pyserini.search.lucene import LuceneSearcher

from web_agent_site.utils import (
    BASE_DIR,
    DEFAULT_FILE_PATH,
    DEFAULT_REVIEW_PATH,
    DEFAULT_ATTR_PATH,
    HUMAN_ATTR_PATH
)

TEMPLATE_DIR = os.path.join(BASE_DIR, 'templates')

SEARCH_RETURN_N = 50
PRODUCT_WINDOW = 10
TOP_K_ATTR = 10

END_BUTTON = 'Buy Now'
NEXT_PAGE = 'Next >'
PREV_PAGE = '< Prev'
BACK_TO_SEARCH = 'Back to Search'

ACTION_TO_TEMPLATE = {
    'Description': 'description_page.html',
    'Features': 'features_page.html',
    'Reviews': 'review_page.html',
    'Attributes': 'attributes_page.html',
}

def map_action_to_html(action, **kwargs):
    action_name, action_arg = parse_action(action)
    if action_name == 'start':
        path = os.path.join(TEMPLATE_DIR, 'search_page.html')
        html = render_template_string(
            read_html_template(path=path),
            session_id=kwargs['session_id'],
            instruction_text=kwargs['instruction_text'],
        )
    elif action_name == 'search':
        path = os.path.join(TEMPLATE_DIR, 'results_page.html')
        html = render_template_string(
            read_html_template(path=path),
            session_id=kwargs['session_id'],
            products=kwargs['products'],
            keywords=kwargs['keywords'],
            page=kwargs['page'],
            total=kwargs['total'],
            instruction_text=kwargs['instruction_text'],
        )
    elif action_name == 'click' and action_arg == END_BUTTON:
        path = os.path.join(TEMPLATE_DIR, 'done_page.html')
        html = render_template_string(
            read_html_template(path),
            session_id=kwargs['session_id'],
            reward=kwargs['reward'],
            asin=kwargs['asin'],
            options=kwargs['options'],
            reward_info=kwargs.get('reward_info'),
            goal_attrs=kwargs.get('goal_attrs'),
            purchased_attrs=kwargs.get('purchased_attrs'),
            goal=kwargs.get('goal'),
            mturk_code=kwargs.get('mturk_code'),
            query=kwargs.get('query'),
            category=kwargs.get('category'),
            product_category=kwargs.get('product_category'),
        )
    elif action_name == 'click' and action_arg in ACTION_TO_TEMPLATE:
        path = os.path.join(TEMPLATE_DIR, ACTION_TO_TEMPLATE[action_arg])
        html = render_template_string(
            read_html_template(path),
            session_id=kwargs['session_id'],
            product_info=kwargs['product_info'],
            keywords=kwargs['keywords'],
            page=kwargs['page'],
            asin=kwargs['asin'],
            options=kwargs['options'],
            instruction_text=kwargs.get('instruction_text')
        )
    elif action_name == 'click':
        path = os.path.join(TEMPLATE_DIR, 'item_page.html')
        html = render_template_string(
            read_html_template(path),
            session_id=kwargs['session_id'],
            product_info=kwargs['product_info'],
            keywords=kwargs['keywords'],
            page=kwargs['page'],
            asin=kwargs['asin'],
            options=kwargs['options'],
            instruction_text=kwargs.get('instruction_text'),
            show_attrs=kwargs['show_attrs']
        )
    else:
        raise ValueError('Action name not recognized.')
    return html


def read_html_template(path):
    with open(path) as f:
        template = f.read()
    return template


def parse_action(action):
    """
    Parse action string to action name and its arguments.
    """
    pattern = re.compile(r'(.+)\[(.+)\]')
    m = re.match(pattern, action)
    if m is None:
        action_name = action
        action_arg = None
    else:
        action_name, action_arg = m.groups()
    return action_name, action_arg


def convert_web_app_string_to_var(name, string):
    if name == 'keywords':
        keywords = string
        if keywords.startswith('['):
            keywords = literal_eval(keywords)
        else:
            keywords = [keywords]
        var = keywords
    elif name == 'page':
        page = string
        page = int(page)
        var = page
    else:
        raise ValueError('Name of variable not recognized.')
    return var


def get_top_n_product_from_keywords(
        keywords,
        search_engine,
        all_products,
        product_item_dict,
        attribute_to_asins=None,
    ):
    if keywords[0] == '<r>':
        top_n_products = random.sample(all_products, k=SEARCH_RETURN_N)
    elif keywords[0] == '<a>':
        attribute = ' '.join(keywords[1:]).strip()
        asins = attribute_to_asins[attribute]
        top_n_products = [p for p in all_products if p['asin'] in asins]
    elif keywords[0] == '<c>':
        category = keywords[1].strip()
        top_n_products = [p for p in all_products if p['category'] == category]
    elif keywords[0] == '<q>':
        query = ' '.join(keywords[1:]).strip()
        top_n_products = [p for p in all_products if p['query'] == query]
    else:
        keywords = ' '.join(keywords)
        #print ('search keywords', keywords)
        hits = search_engine.search(keywords, k=SEARCH_RETURN_N)
        docs = [search_engine.doc(hit.docid) for hit in hits]
        top_n_asins = [json.loads(doc.raw())['id'] for doc in docs]
        top_n_products = [product_item_dict[asin] for asin in top_n_asins if asin in product_item_dict]
    return top_n_products


def get_product_per_page(top_n_products, page):
    return top_n_products[(page - 1) * PRODUCT_WINDOW:page * PRODUCT_WINDOW]


def generate_product_prices(all_products):
    product_prices = dict()
    for product in all_products:
        asin = product['asin']
        pricing = product['pricing']
        if not pricing:
            price = 100.0
        elif len(pricing) == 1:
            price = pricing[0]
        else:
            price = random.uniform(*pricing[:2])
        product_prices[asin] = price
    return product_prices


def init_search_engine(num_products=None):
    if num_products == 100:
        indexes = 'indexes_100'
    elif num_products == 1000:
        indexes = 'indexes_1k'
    elif num_products == 100000:
        indexes = 'indexes_100k'
    elif num_products is None:
        indexes = 'indexes'
    else:
        raise NotImplementedError(f'num_products being {num_products} is not supported yet.')
    search_engine = LuceneSearcher(os.path.join(BASE_DIR, f'../search_engine/{indexes}'))
    print (f'search engine loaded {search_engine.num_docs} products')
    return search_engine


def clean_product_keys(products):
    for product in products:
        product.pop('product_information', None)
        product.pop('brand', None)
        product.pop('brand_url', None)
        product.pop('list_price', None)
        product.pop('availability_quantity', None)
        product.pop('availability_status', None)
        product.pop('total_reviews', None)
        product.pop('total_answered_questions', None)
        product.pop('seller_id', None)
        product.pop('seller_name', None)
        product.pop('fulfilled_by_amazon', None)
        product.pop('fast_track_message', None)
        product.pop('aplus_present', None)
        product.pop('small_description_old', None)
    print('Keys cleaned.')
    return products


def load_products(filepath, num_products=None, human_goals=True):
    # TODO: move to preprocessing step -> enforce single source of truth
    with open(filepath) as f:
        products = json.load(f)
    print('Products loaded.')
    products = clean_product_keys(products)
    
    # with open(DEFAULT_REVIEW_PATH) as f:
    #     reviews = json.load(f)
    all_reviews = dict()
    all_ratings = dict()
    # for r in reviews:
    #     all_reviews[r['asin']] = r['reviews']
    #     all_ratings[r['asin']] = r['average_rating']

    if human_goals:
        with open(HUMAN_ATTR_PATH) as f:
            human_attributes = json.load(f)
    with open(DEFAULT_ATTR_PATH) as f:
        attributes = json.load(f)
    with open(HUMAN_ATTR_PATH) as f:
        human_attributes = json.load(f)
    print('Attributes loaded.')

    asins = set()
    all_products = []
    attribute_to_asins = defaultdict(set)
    if num_products is not None:
        # using item_shuffle.json, we assume products already shuffled
        products = products[:num_products]
    for i, p in tqdm(enumerate(products), total=len(products)):
        asin = p['asin']
        if asin == 'nan' or len(asin) > 10:
            continue

        if asin in asins:
            continue
        else:
            asins.add(asin)

        products[i]['category'] = p['category']
        products[i]['query'] = p['query']
        products[i]['product_category'] = p['product_category']

        products[i]['Title'] = p['name']
        products[i]['Description'] = p['full_description']
        products[i]['Reviews'] = all_reviews.get(asin, [])
        products[i]['Rating'] = all_ratings.get(asin, 'N.A.')
        for r in products[i]['Reviews']:
            if 'score' not in r:
                r['score'] = r.pop('stars')
            if 'review' not in r:
                r['body'] = ''
            else:
                r['body'] = r.pop('review')
        products[i]['BulletPoints'] = p['small_description'] \
            if isinstance(p['small_description'], list) else [p['small_description']]

        pricing = p.get('pricing')
        if pricing is None or not pricing:
            pricing = [100.0]
            price_tag = '$100.0'
        else:
            pricing = [
                float(Decimal(re.sub(r'[^\d.]', '', price)))
                for price in pricing.split('$')[1:]
            ]
            if len(pricing) == 1:
                price_tag = f"${pricing[0]}"
            else:
                price_tag = f"${pricing[0]} to ${pricing[1]}"
                pricing = pricing[:2]
        products[i]['pricing'] = pricing
        products[i]['Price'] = price_tag

        options = dict()
        customization_options = p['customization_options']
        option_to_image = dict()
        if customization_options:
            for option_name, option_contents in customization_options.items():
                if option_contents is None:
                    continue
                option_name = option_name.lower()

                option_values = []
                for option_content in option_contents:
                    option_value = option_content['value'].strip().replace('/', ' | ').lower()
                    option_image = option_content.get('image', None)

                    option_values.append(option_value)
                    option_to_image[option_value] = option_image
                options[option_name] = option_values
        products[i]['options'] = options
        products[i]['option_to_image'] = option_to_image

        # without color, size, price, availability
        # if asin in attributes and 'attributes' in attributes[asin]:
        #     products[i]['Attributes'] = attributes[asin]['attributes']
        # else:
        #     products[i]['Attributes'] = ['DUMMY_ATTR']
        # products[i]['instruction_text'] = \
        #     attributes[asin].get('instruction', None)
        # products[i]['instruction_attributes'] = \
        #     attributes[asin].get('instruction_attributes', None)

        # without color, size, price, availability
        if asin in attributes and 'attributes' in attributes[asin]:
            products[i]['Attributes'] = attributes[asin]['attributes']
        else:
            products[i]['Attributes'] = ['DUMMY_ATTR']
            
        if human_goals:
            if asin in human_attributes:
                products[i]['instructions'] = human_attributes[asin]
        else:
            products[i]['instruction_text'] = \
                attributes[asin].get('instruction', None)

            products[i]['instruction_attributes'] = \
                attributes[asin].get('instruction_attributes', None)

        products[i]['MainImage'] = p['images'][0]
        products[i]['query'] = p['query'].lower().strip()

        all_products.append(products[i])

    for p in all_products:
        for a in p['Attributes']:
            attribute_to_asins[a].add(p['asin'])

    product_item_dict = {p['asin']: p for p in all_products}
    product_prices = generate_product_prices(all_products)
    return all_products, product_item_dict, product_prices, attribute_to_asins

web_agent_site/engine/engine.py
Mayer123/LASER
"""
Functions for specifying goals and reward calculations.
"""
import itertools
import random
import spacy
from collections import defaultdict
from rich import print
from thefuzz import fuzz
from web_agent_site.engine.normalize import normalize_color
random.seed(233)
nlp = spacy.load("en_core_web_lg")

PRICE_RANGE = [10.0 * i for i in range(1, 100)]

def get_goals(all_products, product_prices, human_goals=True):
    if human_goals:
        return get_human_goals(all_products, product_prices)
    else:
        return get_synthetic_goals(all_products, product_prices)
    
def get_human_goals(all_products, product_prices):
    goals = []
    cnt_atts = defaultdict(int)
    cnt = 0
    for item in all_products:
        asin = item['asin']
        if 'instructions' not in item: continue
        for product in item['instructions']:
            attributes = product['instruction_attributes']
            if len(attributes) == 0: 
                cnt += 1
                continue

            if product_prices is not None:
                price = product_prices[asin]
                price_range = [p for p in PRICE_RANGE if p > price][:4]
                if len(price_range) >= 2:
                    _, price_upper = sorted(random.sample(price_range, 2))
                    price_text = \
                        f', and price lower than {price_upper:.2f} dollars'
                else:
                    price_upper = 1000000
                    price_text = ''
            else:
                price_upper = 1000000

            goals.append({
                'asin': asin,
                'category': item['category'],
                'query': item['query'],
                'name': item['name'],
                'product_category': item['product_category'],
                'instruction_text': product['instruction'].strip('.') + price_text,
                'attributes': attributes,
                'price_upper': price_upper,
                'goal_options': product['instruction_options'],
            })
            for att in attributes:
                cnt_atts[att] += 1
            # goals += product_goals
    for goal in goals:
        goal['weight'] = 1
    print(cnt, 'skipped')
    print ("human goals: ", len(goals))
    return goals


def get_synthetic_goals(all_products, product_prices):
    goals = []
    cnt_atts = defaultdict(int)
    for product in all_products:
        if ('instruction_text' not in product or 
            product['instruction_text'] is None):
            continue
        product_goals = []        
        asin = product['asin']
        attributes = product['instruction_attributes']
        assert len(attributes) > 0

        if product_prices is not None:
            price = product_prices[asin]
            price_range = [p for p in PRICE_RANGE if p > price][:4]
            if len(price_range) >= 2:
                _, price_upper = sorted(random.sample(price_range, 2))
                price_text = \
                    f', and price lower than {price_upper:.2f} dollars'
            else:
                price_upper = 1000000
                price_text = ''
        else:
            price_upper = 1000000
            price_text = ''

        instruction_text = product['instruction_text']

        options = product['options']
        option_names = sorted(options)
        combinations = list(itertools.product(
            *(options[option_name] for option_name in option_names)
        ))
        for combination in combinations:
            goal_options = dict()
            for i, o in enumerate(combination):
#                option_text.append(f'{option_names[i]}: {o}')
                goal_options[option_names[i]] = o
            option_text = ', and '.join([
                f'{k}: {v}' for k, v in goal_options.items()
            ])
            option_text = ' with ' + option_text if option_text else ''
            product_goals.append({
                'asin': asin,
                'category': product['category'],
                'query': product['query'],
                'name': product['name'],
                'product_category': product['product_category'],
                'instruction_text': f'{instruction_text}{option_text}{price_text}',
                'attributes': attributes,
                'price_upper': price_upper,
                'goal_options': goal_options,
                'name': product['Title'],
            })
            for att in attributes:
                cnt_atts[att] += 1
        goals += product_goals
    for goal in goals:
        goal['weight'] = sum(1. / cnt_atts[att] for att in goal['attributes']) / len(goal['attributes'])
    return goals


def get_type_reward(purchased_product, goal):
    """Determines the type reward - captures whether chosen product is in the same category"""
    query_match = purchased_product['query'] == goal['query']

    # Check number of unique categories that match, ignoring order
    purchased_product_category = [x.strip() for x in purchased_product['product_category'].split('›')]
    goal_product_category = [x.strip() for x in goal['product_category'].split('›')]
    category_match = len(set(purchased_product_category) & set(goal_product_category)) >= 2

    # Determine whether types align based on product name similarity
    purchased_type = purchased_product['name']
    desired_type = goal['name']

    purchased_type_parse = nlp(purchased_type)
    desired_type_parse = nlp(desired_type)

    purchased_type_parse = [t.text.lower() for t in purchased_type_parse if t.pos_ in ('PNOUN', 'NOUN', 'PROPN')]
    desired_type_parse = [t.text.lower() for t in desired_type_parse if t.pos_ in ('PNOUN', 'NOUN', 'PROPN')]

    n_intersect_type = len(
        set(purchased_type_parse) & set(desired_type_parse)
    )
    if len(desired_type_parse) == 0:
        title_score = 0.2
    else:
        title_score = n_intersect_type / len(desired_type_parse)

    r_type = 1.0

    # Adjust r_type score based on query, category title matching/scores
    match = query_match or category_match or title_score > 0.2
    if not match:
        r_type = 0.5

    if title_score < 0.1:
        r_type = 0.1
    
    if title_score == 0.0:
        r_type = 0.0

    return dict(
        r_type=r_type,
        query_match=query_match,
        category_match=category_match,
        title_score=title_score,
    )


def get_attribute_reward(purchased_product, goal):
    """Determines whether purchased products shares same attributes as goal"""
    purchased_attrs = purchased_product['Attributes']
    goal_attrs = goal['attributes']

    num_attr_matches = 0
    for g_attr in goal_attrs:
        matched = False
        # Check whether goal attribute found in purchased product attribute list
        for p_attr in purchased_attrs:
            score = fuzz.token_set_ratio(p_attr, g_attr)
            if score > 85:
                num_attr_matches += 1
                matched = True
                break
        # If not in purchased attrs, check Title, Bullet Points (Features), Desc
        if (
            not matched and
            (
                g_attr in purchased_product['Title'].lower() or
                g_attr in ' '.join(purchased_product['BulletPoints']).lower() or
                g_attr in purchased_product['Description'].lower()
            )
        ):
            num_attr_matches += 1
            matched = True
    
    r_attr = num_attr_matches / len(goal_attrs)
    return r_attr, num_attr_matches


def get_option_reward(purchased_options, goal_options):
    """Calculate reward for purchased product's options w.r.t. goal options"""
    purchased_options = [normalize_color(o) for o in purchased_options]
    goal_options = [normalize_color(o) for o in goal_options]

    # Perform fuzzy matching of each purchased option against each goal option
    num_option_matches = 0
    for g_option in goal_options:
        for p_option in purchased_options:
            score = fuzz.token_set_ratio(p_option, g_option)
            if score > 85:
                num_option_matches += 1
                break
    
    # Calculate option reward as fraction of goal options hit
    r_option = num_option_matches / len(goal_options) if len(goal_options) > 0 else None
    return r_option, num_option_matches


def get_reward(purchased_product, goal, price, options, **kwargs):
    """Get cumulative reward score for purchased product and goal"""
    r_type_dict = get_type_reward(purchased_product, goal)

    r_price = (
        price <= goal['price_upper']
    ) if goal['price_upper'] > 0 else None

    r_att, num_attr_matches = get_attribute_reward(purchased_product, goal)

    r_option, num_option_matches = get_option_reward(
        list(options.values()),
        goal['goal_options'].items()
        if isinstance(goal['goal_options'], dict)
        else goal['goal_options']
    )

    total_reward = (
        (num_attr_matches + num_option_matches + r_price) \
            / (len(goal['attributes']) + len(goal['goal_options']) + 1)
    )

    total_reward *= r_type_dict['r_type']

    # If verbose flag enabled, store score sub-components into dictionary
    if kwargs.get('verbose', False):
        info =  {
            'r_type': r_type_dict['r_type'],
            'r_att': r_att,
            'w_att': len(goal['attributes']) / (len(goal['attributes']) + len(goal['goal_options']) + 1),
            'query_match': r_type_dict['query_match'],
            'category_match': r_type_dict['category_match'],
            'title_score': r_type_dict['title_score'],
        }
        if r_option is not None:
            info['r_option'] = r_option
            info['w_option'] = len(goal['goal_options']) / (len(goal['attributes']) + len(goal['goal_options']) + 1)
        if r_price is not None:
            info['r_price'] = r_price
            info['w_price'] = 1 / (len(goal['attributes']) + len(goal['goal_options']) + 1)
        return total_reward, info
    return total_reward

web_agent_site/engine/goal.py
Mayer123/LASER
import re
from typing import Tuple

COLOR_SET = [
    'alabaster', 'apricot', 'aqua', 'ash', 'asphalt', 'azure',
    'banana', 'beige', 'black', 'blue', 'blush', 'bordeaux', 'bronze',
    'brown', 'burgundy', 'camel', 'camo', 'caramel', 'champagne',
    'charcoal', 'cheetah', 'chestnut', 'chocolate', 'christmas', 'coffee',
    'cognac', 'copper', 'coral', 'cranberry', 'cream', 'crystal', 'dark',
    'denim', 'eggplant', 'elephant', 'espresso', 'fuchsia', 'gold', 'granite',
    'grape', 'graphite', 'grass', 'gray', 'green', 'grey', 'heather', 'indigo',
    'ivory', 'ivy', 'khaki', 'lavender', 'lemon', 'leopard', 'light', 'lilac',
    'lime', 'magenta', 'maroon', 'mauve', 'merlot', 'midnight', 'mint', 'mocha',
    'multicolor', 'mushroom', 'mustard', 'natural', 'navy', 'nude', 'olive',
    'orange', 'peach', 'pewter', 'pink',    'plum', 'purple', 'rainbow', 'red',
    'rose', 'royal', 'rust', 'sand', 'sapphire', 'seashell', 'silver', 'skull',
    'slate', 'steel', 'stone', 'stonewash', 'sunflower', 'tan', 'taupe', 'teal',
    'tiger', 'turquoise', 'violet', 'walnut', 'wheat', 'white', 'wine', 'yellow',
]

SIZE_SET = [
    'xx-large', '3x-large', '4x-large', '5x-large', 'x-large', 'x-small',
    'medium', 'large', 'small',
    'queen', 'twin', 'full', 'king', 'one size',
    'pack',
]

SIZE_PATTERNS = [
    re.compile(r'(.*)neck(.*)sleeve'),
    re.compile(r'(.*) women \| (.*) men'),
    re.compile(r'(.*)w x(.*)l'),
    re.compile(r'(.*)w by (.*)l'),
    re.compile(r'(.*)w x(.*)h'),
    re.compile(r'(.*)wide'),
    re.compile(r'(.*)x-wide'),
    re.compile(r'(.*)narrow'),
    re.compile(r'(.*)petite'),
    re.compile(r'(.*)inch'),
    re.compile(r'(.*)plus'),
    re.compile(r'(.*)mm'),
    re.compile(r'women(.*)'),
    re.compile(r'(.*)x(.*)'),
    re.compile(r'(.*)ft'),
    re.compile(r'(.*)feet'),
    re.compile(r'(.*)meter'),
    re.compile(r'(.*)yards'),
    re.compile(r'(.*)\*(.*)'),
    re.compile(r'(.*)\-(.*)'),
    re.compile(r'(\d+)"$'),
    re.compile(r'(\d+)f$'),
    re.compile(r'(\d+)m$'),
    re.compile(r'(\d+)cm$'),
    re.compile(r'(\d+)g$'),
]
SIZE_PATTERNS = [re.compile(s) for s in SIZE_SET] + SIZE_PATTERNS

def normalize_color(color_string: str) -> str:
    """Extracts the first color found if exists"""
    for norm_color in COLOR_SET:
        if norm_color in color_string:
            return norm_color
    return color_string

def normalize_color_size(product_prices: dict) -> Tuple[dict, dict]:
    """Get mappings of all colors, sizes to corresponding values in COLOR_SET, SIZE_PATTERNS"""
    
    # Get all colors, sizes from list of all products
    all_colors, all_sizes = set(), set()
    for (_, color, size), _ in product_prices.items():
        all_colors.add(color.lower())
        all_sizes.add(size.lower())
    
    # Create mapping of each original color value to corresponding set value
    color_mapping = {'N.A.': 'not_matched'} 
    for c in all_colors:
        matched = False
        for base in COLOR_SET:
            if base in c:
                color_mapping[c] = base
                matched = True
                break
        if not matched:
            color_mapping[c] = 'not_matched'

    # Create mapping of each original size value to corresponding set value
    size_mapping = {'N.A.': 'not_matched'}
    for s in all_sizes:
        matched = False
        for pattern in SIZE_PATTERNS:
            m = re.search(pattern, s)
            if m is not None:
                matched = True
                size_mapping[s] = pattern.pattern
                break
        if not matched:
            if s.replace('.', '', 1).isdigit():
                size_mapping[s] = 'numeric_size'
                matched= True
        if not matched:
            size_mapping[s] = 'not_matched'
    
    return color_mapping, size_mapping
    

web_agent_site/engine/normalize.py
Mayer123/LASER
from gym.envs.registration import register

from web_agent_site.envs.web_agent_site_env import WebAgentSiteEnv
from web_agent_site.envs.web_agent_text_env import WebAgentTextEnv

register(
  id='WebAgentSiteEnv-v0',
  entry_point='web_agent_site.envs:WebAgentSiteEnv',
)

register(
  id='WebAgentTextEnv-v0',
  entry_point='web_agent_site.envs:WebAgentTextEnv',
)
web_agent_site/envs/__init__.py
Mayer123/LASER
import gym
import random
import requests
import string
import time

from bs4 import BeautifulSoup
from bs4.element import Comment
from gym import spaces
from os.path import join, dirname, abspath
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.keys import Keys
from selenium.common.exceptions import ElementNotInteractableException
from web_agent_site.engine.engine import parse_action, END_BUTTON

class WebAgentSiteEnv(gym.Env):
    """Gym environment for HTML mode of WebShop environment"""

    def __init__(self, observation_mode='html', **kwargs):
        """
        Constructor for HTML environment

        Arguments:
        observation_mode (`str`) -- ['html' | 'text'] (default 'html')
        pause (`float`) -- Pause (in seconds) after taking an action. 
            This is mainly for demo purposes.
            Recommended value: 2.0s
        render (`bool`) -- Show browser if set to `True`.
        session ('str') -- Session ID to initialize environment with
        """
        super(WebAgentSiteEnv, self).__init__()
        self.observation_mode = observation_mode
        self.kwargs = kwargs

        # Create a browser driver to simulate the WebShop site
        service = Service(join(dirname(abspath(__file__)), 'chromedriver'))
        options = Options()
        if 'render' not in kwargs or not kwargs['render']:
            options.add_argument("--headless")  # don't show browser
        self.browser = webdriver.Chrome(service=service, options=options)

        # Set flags and values for WebShop session
        self.text_to_clickable = None
        self.assigned_session = kwargs.get('session')
        self.session = None
        self.reset()

    def step(self, action):
        """
        Takes an action, updates WebShop environment, and returns (observation, reward, done, info)

        Arguments:
        action (`str`): An action should be of the following structure:
          - search[keywords]
          - click[value]
        If action not valid, perform nothing.
        """
        reward = 0.0
        done = False
        info = None

        # Map action to executed command on the WebShop environment via the broswer driver
        action_name, action_arg = parse_action(action)
        if action_name == 'search':
            try:
                search_bar = self.browser.find_element_by_id('search_input')
            except Exception:
                pass
            else:
                search_bar.send_keys(action_arg)
                search_bar.submit()
        elif action_name == 'click':
            try:
                self.text_to_clickable[action_arg].click()
            except ElementNotInteractableException:
                # Perform force click with JavaScript
                button = self.text_to_clickable[action_arg]
                self.browser.execute_script("arguments[0].click();", button)
            reward = self.get_reward()
            if action_arg == END_BUTTON:
                done = True
        elif action_name == 'end':
            done = True
        else:
            print('Invalid action. No action performed.')

        if 'pause' in self.kwargs:
            time.sleep(self.kwargs['pause'])
        return self.observation, reward, done, info
    
    def get_available_actions(self):
        """Returns list of available actions at the current step"""
        # Determine if a search bar is available
        try:
            search_bar = self.browser.find_element_by_id('search_input')
        except Exception:
            has_search_bar = False
        else:
            has_search_bar = True

        # Collect buttons, links, and options as clickables
        buttons = self.browser.find_elements_by_class_name('btn')
        product_links = self.browser.find_elements_by_class_name('product-link')
        buying_options = self.browser.find_elements_by_css_selector("input[type='radio']")

        self.text_to_clickable = {
            f'{b.text}': b
            for b in buttons + product_links
        }
        for opt in buying_options:
            opt_value = opt.get_attribute('value')
            self.text_to_clickable[f'{opt_value}'] = opt
        return dict(
            has_search_bar=has_search_bar,
            clickables=list(self.text_to_clickable.keys()),
        )

    def _parse_html(self, html=None, url=None):
        """
        Returns web request result wrapped in BeautifulSoup object

        Arguments:
        url (`str`): If no url or html is provided, use the current
            observation (HTML) for parsing.
        """
        if html is None:
            if url is not None:
                html = requests.get(url)
            else:
                html = self.state['html']
        html_obj = BeautifulSoup(html, 'html.parser')
        return html_obj

    def get_reward(self):
        """Get reward value at current step of the environment"""
        html_obj = self._parse_html()
        r = html_obj.find(id='reward')
        r = float(r.findChildren("pre")[0].string) if r is not None else 0.0
        return r
    
    def get_instruction_text(self):
        """Get corresponding instruction text for environment current step"""
        html_obj = self._parse_html(self.browser.page_source)
        instruction_text = html_obj.find(id='instruction-text').h4.text
        return instruction_text
    
    def convert_html_to_text(self, html):
        """Strip HTML of tags and add separators to convert observation into simple mode"""
        texts = self._parse_html(html).findAll(text=True)
        visible_texts = filter(tag_visible, texts)
        observation = ' [SEP] '.join(t.strip() for t in visible_texts if t != '\n')
        return observation
    
    @property
    def state(self):
        """
        State that includes all information. The actual observation are
        likely to be a subset or reduced form of the state.
        """
        return dict(
            url=self.browser.current_url,
            html=self.browser.page_source,
            instruction_text=self.instruction_text,
        )
    
    @property
    def observation(self):
        """Compiles state into either the `html` or `text` observation mode"""
        html = self.state['html']
        if self.observation_mode == 'html':
            return html
        elif self.observation_mode == 'text':
            return self.convert_html_to_text(html)
        else:
            raise ValueError(
                f'Observation mode {self.observation_mode} not supported.'
            )

    @property
    def action_space(self):
        # Recommended to use `get_available_actions` instead
        return NotImplementedError

    @property
    def observation_space(self):
        return NotImplementedError

    def reset(self):
        """Create a new session and reset environment variables"""
        if self.assigned_session is not None:
            self.session = self.assigned_session
        else:
            self.session = ''.join(random.choices(string.ascii_lowercase, k=5))
        init_url = f'http://127.0.0.1:3000/{self.session}'
        self.browser.get(init_url)

        self.instruction_text = self.get_instruction_text()

        return self.observation, None

    def render(self, mode='human'):
        # TODO: Render observation in terminal or WebShop website
        return NotImplementedError

    def close(self):
        # TODO: When DB used instead of JSONs, tear down DB here
        self.browser.close()
        print('Browser closed.')

def tag_visible(element):
    """Helper method to strip HTML block of extraneous tags"""
    ignore = {'style', 'script', 'head', 'title', 'meta', '[document]'}
    return (
        element.parent.name not in ignore and not isinstance(element, Comment)
    )

web_agent_site/envs/web_agent_site_env.py
Mayer123/LASER
import gym
import json
import random
import string
import time
import torch

from bs4 import BeautifulSoup
from bs4.element import Comment
from collections import defaultdict
from flask import Flask
from web_agent_site.engine.engine import (
    load_products,
    init_search_engine,
    get_top_n_product_from_keywords,
    map_action_to_html,
    parse_action,
    get_product_per_page,
    ACTION_TO_TEMPLATE,
    END_BUTTON, NEXT_PAGE, PREV_PAGE, BACK_TO_SEARCH,
)
from web_agent_site.engine.goal import get_reward, get_goals
from web_agent_site.utils import (
    DEFAULT_FILE_PATH,
    FEAT_CONV,
    FEAT_IDS,
    random_idx
)

app = Flask(__name__)
class WebAgentTextEnv(gym.Env):
    """Gym environment for Text mode of WebShop environment"""
    def __init__(
            self,
            observation_mode='html',
            file_path=DEFAULT_FILE_PATH,
            server=None,
            **kwargs
        ):
        """
        Constructor for text environment

        Arguments:
        observation_mode (`str`) -- ['html' | 'text'] (default 'html')
        get_image
        filter_goals
        limit_goals
        num_products
        human_goals
        session
        session_prefix
        show_attrs
        """
        super(WebAgentTextEnv, self).__init__()
        self.observation_mode = observation_mode
        self.kwargs = kwargs

        self.file_path = file_path

        self.base_url = 'http://127.0.0.1:3000'
        self.server = SimServer(
            self.base_url,
            self.file_path,
            self.kwargs.get('filter_goals'),
            self.kwargs.get('limit_goals', -1),
            self.kwargs.get('num_products'),
            self.kwargs.get('human_goals'),
            self.kwargs.get('show_attrs', False),
        ) if server is None else server
        self.browser = SimBrowser(self.server)

        self.session = self.kwargs.get('session')
        self.session_prefix = self.kwargs.get('session_prefix')
        if self.kwargs.get('get_image', 0):
            self.feats = torch.load(FEAT_CONV)
            self.ids = torch.load(FEAT_IDS)
            self.ids = {url: idx for idx, url in enumerate(self.ids)}
        self.prev_obs = []
        self.prev_actions = []
        self.num_prev_obs = self.kwargs.get('num_prev_obs', 0)
        self.num_prev_actions = self.kwargs.get('num_prev_actions', 0)
        self.reset()

    def step(self, action):
        """
        Takes an action, updates WebShop environment, and returns (observation, reward, done, info)

        Arguments:
        action (`str`): An action should be of the following structure:
          - search[keywords]
          - click[value]
        If action not valid, perform nothing.
        """
        info = None
        self.get_available_actions()

        # Determine action type (click, search) and argument
        action_name, action_arg = parse_action(action)
        #print ('action_name', action_name)
        #print ('action_arg', action_arg)
        if action_arg is not None:
            action_arg = action_arg.lower()
        if (action_name == 'search' and 
            action_arg is not None and 
            action_arg != ''):
            #print ('Now at step', kwargs)
            status = self.browser.search(action_arg)
        elif (action_name == 'click' and 
              action_arg in self.text_to_clickable.keys() and 
              action_arg != 'search'):
            status = self.browser.click(action_arg, self.text_to_clickable)
        else:
            status = dict(reward=0, done=False)

        # Update observation, state with the new action
        ob = self.observation
        text_list = [ob]
        self.prev_actions.append(action)
        for i in range(1, 1 + max(self.num_prev_obs, self.num_prev_actions)):
            if len(self.prev_actions) >= i and self.num_prev_actions >= i:
                text_list.append(self.prev_actions[-i])
            if len(self.prev_obs) >= i and self.num_prev_obs >= i:
                text_list.append(self.prev_obs[-i])
        state = ' [SEP] '.join(text_list[::-1])
        self.prev_obs.append(ob)
        return state, status['reward'], status['done'], info

    def get_available_actions(self):
        """Returns list of available actions at the current step"""
        html_obj = self._parse_html()

        # Collect search bar, buttons, links, and options as clickables
        search_bar = html_obj.find(id='search_input')
        has_search_bar = True if search_bar is not None else False
        buttons = html_obj.find_all(class_='btn')
        product_links  = html_obj.find_all(class_='product-link')
        buying_options = html_obj.select('input[type="radio"]')

        self.text_to_clickable = {
            f'{b.get_text()}'.lower(): b
            for b in buttons + product_links
        }
        for opt in buying_options:
            opt_value = opt.get('value')
            self.text_to_clickable[f'{opt_value}'] = opt
        return dict(
            has_search_bar=has_search_bar,
            clickables=list(self.text_to_clickable.keys()),
        )
    
    def get_image(self):
        """Scrape image from page HTML and return as a list of pixel values"""
        html_obj = self._parse_html(self.browser.page_source)
        image_url = html_obj.find(id='product-image')
        if image_url is not None:
            image_url = image_url['src']
            if image_url in self.ids:
                image_idx = self.ids[image_url]
                image = self.feats[image_idx]
                return image
        return torch.zeros(512)

    def get_instruction_text(self):
        """Get corresponding instruction text for current environment session"""
        html_obj = self._parse_html(self.browser.page_source)
        instruction_text = html_obj.find(id='instruction-text').h4.text
        return instruction_text

    def _parse_html(self, html=None):
        """
        Returns web request result wrapped in BeautifulSoup object

        Arguments:
        url (`str`): If no url or html is provided, use the current
            observation (HTML) for parsing.
        """
        if html is None:
            html = self.state['html']
        html_obj = BeautifulSoup(html, 'html.parser')
        return html_obj
    
    @property
    def observation(self):
        """Compiles state into either the `html` or `text` observation mode"""
        html = self.state['html']
        if self.observation_mode == 'html':
            return html
        elif self.observation_mode == 'text':
            return self.convert_html_to_text(html, simple=True)
        elif self.observation_mode == 'text_rich':
            return self.convert_html_to_text(html, simple=False)
        elif self.observation_mode == 'url':
            return self.state['url']
        else:
            raise ValueError(
                f'Observation mode {self.observation_mode} not supported.'
            )
    
    @property
    def state(self):
        """
        State that includes all information. The actual observation are
        likely to be a subset or reduced form of the state.
        """
        return dict(
            url=self.browser.current_url,
            html=self.browser.page_source,
            instruction_text=self.instruction_text,
        )
    
    def convert_html_to_text(self, html, simple=False):
        """Strip HTML of tags and add separators to convert observation into simple mode"""
        texts = self._parse_html(html).findAll(text=True)
        visible_texts = filter(tag_visible, texts)
        if simple:
            # For `simple` mode, return just [SEP] separators
            return ' [SEP] '.join(t.strip() for t in visible_texts if t != '\n')
        else:
            # Otherwise, return an observation with tags mapped to specific, unique separators
            observation = ''
            for t in visible_texts:
                if t == '\n': continue
                if t.parent.name == 'button':  # button
                    processed_t = f'[button] {t} [button_]'
                elif t.parent.name == 'label':  # options
                    if f'"{t}"' in self.state['url']:
                        processed_t = f'  [clicked button] {t} [clicked button_]'
                        observation = f'You have clicked {t}.\n' + observation
                    else:
                        processed_t = f'  [button] {t} [button_]'
                elif t.parent.get('class') == ["product-link"]: # product asins
                    if f'{t}' in self.server.user_sessions[self.session]['asins']:
                        processed_t = f'\n[clicked button] {t} [clicked button_]'
                    else:
                        processed_t = f'\n[button] {t} [button_]'
                else: # regular, unclickable text
                    processed_t =  str(t)
                observation += processed_t + '\n'
            return observation
    
    def reset(self, session=None, instruction_text=None):
        """Create a new session and reset environment variables"""
        session_int = None
        if session is not None:
            self.session = str(session)
            if isinstance(session, int):
                session_int = session
        else:
            self.session = ''.join(random.choices(string.ascii_lowercase, k=10))
        if self.session_prefix is not None:
            self.session = self.session_prefix + self.session

        init_url = f'{self.base_url}/{self.session}'
        self.browser.get(init_url, session_id=self.session, session_int=session_int)

        self.text_to_clickable = None
        self.instruction_text = self.get_instruction_text() if instruction_text is None else instruction_text
        obs = self.observation
        self.prev_obs = [obs]
        self.prev_actions = []
        #print ('Current Goal :', self.browser.server.user_sessions[self.session]['goal'])
        return obs, None

    def render(self, mode='human'):
        pass

    def close(self):
        pass
    

def tag_visible(element):
    ignore = {'style', 'script', 'head', 'title', 'meta', '[document]'}
    return (
        element.parent.name not in ignore and not isinstance(element, Comment)
    )


class SimServer:
    """Lightweight simulator of WebShop Flask application for generating HTML observations"""
    def __init__(
        self,
        base_url,
        file_path,
        filter_goals=None,
        limit_goals=-1,
        num_products=None,
        human_goals=0,
        show_attrs=False,
    ):
        """
        Constructor for simulated server serving WebShop application
        
        Arguments:
        filter_goals (`func`) -- Select specific goal(s) for consideration based on criteria of custom function
        limit_goals (`int`) -- Limit to number of goals available
        num_products (`int`) -- Number of products to search across
        human_goals (`bool`) -- If true, load human goals; otherwise, load synthetic goals
        """
        # Load all products, goals, and search engine
        self.base_url = base_url
        self.all_products, self.product_item_dict, self.product_prices, _ = \
            load_products(filepath=file_path, num_products=num_products, human_goals=human_goals)
        self.search_engine = init_search_engine(num_products=num_products)
        self.goals = get_goals(self.all_products, self.product_prices, human_goals)
        self.show_attrs = show_attrs

        # Fix outcome for random shuffling of goals
        random.seed(233)
        random.shuffle(self.goals)

        # Apply `filter_goals` parameter if exists to select speific goal(s)
        if filter_goals is not None:
            self.goals = [
                goal for (i, goal) in enumerate(self.goals)
                if filter_goals(i, goal)
            ]
        
        # Imposes `limit` on goals via random selection
        if limit_goals != -1 and limit_goals < len(self.goals):
            self.weights = [goal['weight'] for goal in self.goals]
            self.cum_weights = [0]
            for w in self.weights:
                self.cum_weights.append(self.cum_weights[-1] + w)
            idxs = []
            while len(idxs) < limit_goals:
                idx = random_idx(self.cum_weights)
                if idx not in idxs:
                    idxs.append(idx)
            self.goals = [self.goals[i] for i in idxs]
        print(f'Loaded {len(self.goals)} goals.')

        # Set extraneous housekeeping variables
        self.weights = [goal['weight'] for goal in self.goals]
        self.cum_weights = [0]
        for w in self.weights:
            self.cum_weights.append(self.cum_weights[-1] + w)
        self.user_sessions = dict()
        self.search_time = 0
        self.render_time = 0
        self.sample_time = 0
        self.assigned_instruction_text = None  # TODO: very hacky, should remove
        
    @app.route('/', methods=['GET', 'POST'])
    def index(self, session_id, **kwargs):
        """Redirect to the search page with the given session ID"""
        html = map_action_to_html(
            'start',
            session_id=session_id,
            instruction_text=kwargs['instruction_text'],
        )
        url = f'{self.base_url}/{session_id}'
        return html, url
    
    @app.route('/', methods=['GET', 'POST'])
    def search_results(self, session_id, **kwargs):
        """Initialize session and return the search results page"""
        session = self.user_sessions[session_id]
        keywords = kwargs['keywords']  # TODO: why is this using kwargs? why not session?
        assert isinstance(keywords, list)
        page = 1 if 'page' not in kwargs else kwargs['page']
        session["page"] = page
        session["keywords"] = keywords
        session["actions"]["search"] += 1
        session["asin"] = None
        session["options"] = {}

        # Perform search on keywords from items and record amount of time it takes
        old_time = time.time()
        top_n_products = get_top_n_product_from_keywords(
            keywords,
            self.search_engine,
            self.all_products,
            self.product_item_dict
        )
        self.search_time += time.time() - old_time
        
        # Get product list from search result asins and get list of corresponding URLs
        products = get_product_per_page(top_n_products, page)

        keywords_url_string = '+'.join(keywords)
        url = (
            f'{self.base_url}/search_results/{session_id}/'
            f'{keywords_url_string}/{page}'
        )

        # Render HTML search page and record amount of time taken
        old_time = time.time()
        html = map_action_to_html(
            'search',
            session_id=session_id,
            products=products,
            keywords=session["keywords"],
            page=page,
            total=len(top_n_products),
            instruction_text=session["goal"]["instruction_text"],
        )
        self.render_time += time.time() - old_time
        return html, url
    
    @app.route('/', methods=['GET', 'POST'])
    def item_page(self, session_id, **kwargs):
        """Render and return the HTML for a product item page"""
        session = self.user_sessions[session_id]
        clickable_name = kwargs['clickable_name']
        text_to_clickable = kwargs['text_to_clickable']
        clickable = text_to_clickable[clickable_name]

        # Update session logs with information of last product asin selected
        if (clickable.get('class') is not None and
            clickable.get('class')[0] == 'product-link'):
            session["asin"] = clickable_name.upper()
            session["actions"]["asin"] += 1
            session["asins"].add(session["asin"])
        elif clickable.get('name') is not None:
            clickable_key = clickable['name'].lower()
            session["options"][clickable_key] = clickable_name
            session["actions"]["options"] += 1

        # Set fields + url of page, then render page's HTML
        product_info = self.product_item_dict[session["asin"]]
        keywords_url_string = '+'.join(session["keywords"])
        option_string = json.dumps(session['options'])

        url = (
            f'{self.base_url}/item_page/{session_id}/'
            f'{session["asin"]}/{keywords_url_string}/'
            f'{session["page"]}/{option_string}'
        )

        html = map_action_to_html(
            'click',
            session_id=session_id,
            product_info=product_info,
            keywords=session["keywords"],
            page=session["page"],
            asin=session["asin"],
            options=session["options"],
            instruction_text=session["goal"]["instruction_text"],
            show_attrs=self.show_attrs,
        )
        return html, url

    @app.route('/', methods=['GET', 'POST'])
    def item_sub_page(self, session_id, **kwargs):
        """Render and return the HTML for a product's sub page (i.e. description, features)"""
        session = self.user_sessions[session_id]
        clickable_name = kwargs['clickable_name']
        for k in ACTION_TO_TEMPLATE:
            if clickable_name.lower() == k.lower():
                clickable_name = k
                break
        
        # Set fields + url of page, then render page's HTML
        product_info = self.product_item_dict[session["asin"]]
        session["actions"][clickable_name] += 1
        keywords_url_string = '+'.join(session["keywords"])
        url = (
            f'{self.base_url}/item_sub_page/{session_id}/'
            f'{session["asin"]}/{keywords_url_string}/{session["page"]}/'
            f'{clickable_name}/{session["options"]}'
        )
        html = map_action_to_html(
            f'click[{clickable_name}]',
            session_id=session_id,
            product_info=product_info,
            keywords=session["keywords"],
            page=session["page"],
            asin=session["asin"],
            options=session["options"],
            instruction_text=session["goal"]["instruction_text"],
        )
        return html, url

    @app.route('/', methods=['GET', 'POST'])
    def done(self, session_id, **kwargs):
        """Render and return HTML for done page"""
        session = self.user_sessions[session_id]
        goal = self.user_sessions[session_id]['goal']
        purchased_product = self.product_item_dict[session["asin"]]
        session["actions"]["purchase"] += 1
        price = self.product_prices.get(session["asin"])

        # Calculate reward for selected product and set variables for page details
        reward, info = get_reward(
            purchased_product,
            goal,
            price=price,
            options=session["options"],
            verbose=True
        )

        self.user_sessions[session_id]['verbose_info'] = info
        self.user_sessions[session_id]['done'] = True
        self.user_sessions[session_id]['reward'] = reward

        url = (
            f'{self.base_url}/done/{session_id}/'
            f'{session["asin"]}/{session["options"]}'
        )
        html = map_action_to_html(
            f'click[{END_BUTTON}]',
            session_id=session_id,
            reward=reward,
            asin=session["asin"],
            options=session["options"],
            instruction_text=session["goal"]["instruction_text"],
        )
        return html, url, reward
    
    def receive(self, session_id, current_url, session_int=None, **kwargs):
        """Map action to the corresponding page"""
        status = dict(reward=0.0, done=False)

        with app.app_context(), app.test_request_context():
            # Create/determine goal, instruction_text from current session
            if session_id not in self.user_sessions:
                idx = session_int if (session_int is not None and isinstance(session_int, int)) else random_idx(self.cum_weights) 
                goal = self.goals[idx]
                instruction_text = goal['instruction_text']
                self.user_sessions[session_id] = {'goal': goal, 'done': False}
            else:
                instruction_text = \
                    self.user_sessions[session_id]['goal']['instruction_text']
            if self.assigned_instruction_text is not None:
                instruction_text = self.assigned_instruction_text  # TODO: very hacky, should remove
                self.user_sessions[session_id]['goal']['instruction_text'] = instruction_text
            session = self.user_sessions[session_id]

            if not kwargs:
                # If no action, reset the session variables
                kwargs['instruction_text'] = instruction_text
                html, url = self.index(session_id, **kwargs)
                self.user_sessions[session_id].update(
                    {
                        'keywords': None,
                        'page': None,
                        'asin': None,
                        'asins': set(),
                        'options': dict(),
                        'actions': defaultdict(int)
                    }
                )
            elif 'keywords' in kwargs:
                # If search keywords are available, run a search
                html, url = self.search_results(session_id, **kwargs)
            elif 'clickable_name' in kwargs:
                clickable_name = kwargs['clickable_name'].lower()
                if clickable_name == END_BUTTON.lower():
                    # If "buy now" clicked, calculate reward and flag session as terminated
                    html, url, reward = self.done(session_id, **kwargs)
                    status['reward'] = reward
                    status['done'] = True
                elif clickable_name == BACK_TO_SEARCH.lower():
                    # If "back to search" clicked, recursively reset the session back to search page
                    html, url, status = self.receive(session_id, current_url)
                elif (clickable_name == NEXT_PAGE.lower() and 
                      self.get_page_name(current_url) == 'search_results'):
                    # If "next page" clicked from search results, re-render with `page` enumerated
                    html, url, status = self.receive(
                        session_id,
                        current_url,
                        keywords=session["keywords"],
                        page=session["page"] + 1,
                    )
                elif (clickable_name == PREV_PAGE.lower() and 
                      self.get_page_name(current_url) == 'search_results'):
                    # If "prev page" clicked from search results, re-render with `page` denumerated
                    html, url, status = self.receive(
                        session_id,
                        current_url,
                        keywords=session["keywords"],
                        page=session["page"] - 1,
                    )
                elif (clickable_name == PREV_PAGE.lower() and 
                      self.get_page_name(current_url) == 'item_sub_page'):
                    # If "prev page" clicked from sub page, return to corresponding item page
                    html, url = self.item_page(session_id, **kwargs)
                elif (clickable_name == PREV_PAGE.lower() and 
                      self.get_page_name(current_url) == 'item_page'):
                    # If "prev page" clicked from item page, return to search results page
                    html, url = self.search_results(
                        session_id,
                        keywords=session["keywords"],
                        page=session["page"],
                        **kwargs
                    )
                elif clickable_name in [k.lower() for k in ACTION_TO_TEMPLATE]:
                    # Render item_sub_page if clickable is description, features, or reviews
                    html, url = self.item_sub_page(session_id, **kwargs)
                else:
                    # Otherwise, render current item page
                    html, url = self.item_page(session_id, **kwargs)
            return html, url, status
    
    def get_page_name(self, url):
        """Determine which page (i.e. item_page, search_results) the given URL is pointing at"""
        if url is None:
            return None
        page_names = [
            'search_results',
            'item_page',
            'item_sub_page',
            'done'
        ]
        for page_name in page_names:
            if page_name in url:
                return page_name
        return ''  # index page


class SimBrowser:
    """Simulated browser for rendering the HTML source of WebShop environment pages"""
    def __init__(self, server):
        self.server = server
        self.current_url = None
        self.page_source = None
        self.session_id = None

    def get(self, url, session_id=None, session_int=None):
        """Set browser variables to corresponding link, page HTML for URL"""
        self.session_id = url.split('/')[-1] if session_id is None else session_id
        self.page_source, _, _ = \
            self.server.receive(self.session_id, self.current_url, session_int=session_int)
        self.current_url = url
    
    def click(self, clickable_name, text_to_clickable):
        """Wrapper for `receive` handler for performing click action on current page"""
        self.page_source, self.current_url, status = \
            self.server.receive(
                self.session_id,
                current_url=self.current_url,
                clickable_name=clickable_name,
                text_to_clickable=text_to_clickable,
            )
        return status
    
    def search(self, keywords):
        """Wrapper for `receive` handler for performing search action on current page"""
        if isinstance(keywords, str):
            keywords = keywords.split(' ')
        self.page_source, self.current_url, status = \
            self.server.receive(
                self.session_id,
                current_url=self.current_url,
                keywords=keywords,
        )
        return status

web_agent_site/envs/web_agent_text_env.py
Mayer123/LASER
from web_agent_site.models.models import (
    HumanPolicy,
    RandomPolicy,
)

web_agent_site/models/__init__.py
Mayer123/LASER
"""
Model implementations. The model interface should be suitable for both
the ``site env'' and the ``text env''.
"""
import random

random.seed(4)


class BasePolicy:
    def __init__(self):
        pass
    
    def forward(observation, available_actions):
        """
        Args:
            observation (`str`):
                HTML string

            available_actions ():
                ...
        Returns:
            action (`str`): 
                Return string of the format ``action_name[action_arg]''.
                Examples:
                    - search[white shoes]
                    - click[button=Reviews]
                    - click[button=Buy Now]
        """
        raise NotImplementedError


class HumanPolicy(BasePolicy):
    def __init__(self):
        super().__init__()

    def forward(self, observation, available_actions):
        action = input('> ')
        return action


class RandomPolicy(BasePolicy):
    def __init__(self):
        super().__init__()
    
    def forward(self, observation, available_actions):
        if available_actions['has_search_bar']:
            action = 'search[shoes]'
        else:
            action_arg = random.choice(available_actions['clickables'])
            action = f'click[{action_arg}]'
        return action

web_agent_site/models/models.py
Mayer123/LASER
import bisect
import hashlib
import logging
import random
from os.path import dirname, abspath, join
random.seed(233)
BASE_DIR = dirname(abspath(__file__))
DEBUG_PROD_SIZE = None #1000  # set to `None` to disable

DEFAULT_ATTR_PATH = join(BASE_DIR, '../data/items_ins_v2.json')
DEFAULT_FILE_PATH = join(BASE_DIR, '../data/items_shuffle.json')
DEFAULT_REVIEW_PATH = join(BASE_DIR, '../data/reviews.json')

FEAT_CONV = join(BASE_DIR, '../data/feat_conv.pt')
FEAT_IDS = join(BASE_DIR, '../data/feat_ids.pt')

HUMAN_ATTR_PATH = join(BASE_DIR, '../data/items_human_ins.json')
HUMAN_ATTR_PATH = join(BASE_DIR, '../data/items_human_ins.json')

def random_idx(cum_weights):
    """Generate random index by sampling uniformly from sum of all weights, then
    selecting the `min` between the position to keep the list sorted (via bisect)
    and the value of the second to last index
    """
    pos = random.uniform(0, cum_weights[-1])
    idx = bisect.bisect(cum_weights, pos)
    idx = min(idx, len(cum_weights) - 2)
    return idx

def setup_logger(session_id, user_log_dir):
    """Creates a log file and logging object for the corresponding session ID"""
    logger = logging.getLogger(session_id)
    formatter = logging.Formatter('%(message)s')
    file_handler = logging.FileHandler(
        user_log_dir / f'{session_id}.jsonl',
        mode='w'
    )
    file_handler.setFormatter(formatter)
    logger.setLevel(logging.INFO)
    logger.addHandler(file_handler)
    return logger

def generate_mturk_code(session_id: str) -> str:
    """Generates a redeem code corresponding to the session ID for an MTurk
    worker once the session is completed
    """
    sha = hashlib.sha1(session_id.encode())
    return sha.hexdigest()[:10].upper()
web_agent_site/utils.py
Mayer123/LASER