from typing import List from langchain.schema import ( AIMessage, HumanMessage, SystemMessage ) from .bidder_base import Bidder, draw_plot from .item_base import Item from langchain.input import get_colored_text import time class HumanBidder(Bidder): name: str human_name: str = "Adam" budget: int auction_hash: str cur_item_id = 0 items: list = [] withdraw: bool = False engagement_count: int = 0 original_budget: int = 0 profit: int = 0 items_won = [] all_bidders_status = {} # track others' profit # essential for demo need_input: bool = False semaphore: int = 0 # if needs input, then semaphore is set as 1, else waits. input_box: str = None # global variable for accepting user input # not used model_name: str = 'human' openai_cost = 0 desire = '' plan_strategy = '' correct_belief = True class Config: arbitrary_types_allowed = True def get_plan_instruct(self, items: List[Item]): self.items = items plan_instruct = "As {bidder_name}, you have a total budget of ${budget}. This auction has a total of {item_num} items to be sequentially presented, they are:\n{items_info}".format( bidder_name=self.name, budget=self.budget, item_num=len(items), items_info=self._get_items_value_str(items) ) return plan_instruct def init_plan(self, plan_instruct: str): # Human = auctioneer, AI = bidder self.dialogue_history += [ HumanMessage(content=plan_instruct), AIMessage(content='Got it!') ] return '' def get_bid_instruct(self, auctioneer_msg, bid_round): self.dialogue_history += [ HumanMessage(content=auctioneer_msg), AIMessage(content='') ] return auctioneer_msg def bid(self, bid_instruct): # wait for the cue to handle user input while self.semaphore <= 0: time.sleep(1) self.dialogue_history += [ HumanMessage(content=''), AIMessage(content=self.input_box) ] self.semaphore -= 1 self.need_input = False return self.input_box def get_summarize_instruct(self, bidding_history: str, hammer_msg: str, win_lose_msg: str): instruct_summarize = f"{bidding_history}\n\n{hammer_msg}\n{win_lose_msg}" return instruct_summarize def summarize(self, instruct_summarize: str): self.dialogue_history += [ HumanMessage(content=instruct_summarize), AIMessage(content='Noted.') ] self.budget_history.append(self.budget) self.profit_history.append(self.profit) return '' def get_replan_instruct(self): return '' def replan(self, instruct_replan): self.withdraw = False self.cur_item_id += 1 return '' def to_monitors(self, as_json=False): items_won = [] for item, bid in self.items_won: items_won.append([str(item), bid, item.true_value]) if as_json: return { 'auction_hash': self.auction_hash, 'bidder_name': self.name, 'human_name': self.human_name, 'model_name': self.model_name, 'budget': self.original_budget, 'money_left': self.budget, 'profit': self.profit, 'items_won': items_won, 'engagement_count': self.engagement_count, } else: return [ self.budget, self.profit, items_won, 0, 0, round(self.failed_bid_cnt / (self.total_bid_cnt+1e-8), 2), 0, 0, self.engagement_count, draw_plot(f"{self.name} ({self.model_name})", self.budget_history, self.profit_history), [], [], [], [] ]