57 lines
2.4 KiB
Python
57 lines
2.4 KiB
Python
from typing import Optional, List
|
|
from agents.agent import Agent
|
|
from agents.deals import ScrapedDeal, DealSelection, Deal, Opportunity
|
|
from agents.scanner_agent import ScannerAgent
|
|
from agents.ensemble_agent import EnsembleAgent
|
|
from agents.messaging_agent import MessagingAgent
|
|
|
|
|
|
class PlanningAgent(Agent):
|
|
|
|
name = "Planning Agent"
|
|
color = Agent.GREEN
|
|
DEAL_THRESHOLD = 50
|
|
|
|
def __init__(self, collection):
|
|
"""
|
|
Create instances of the 3 Agents that this planner coordinates across
|
|
"""
|
|
self.log("Planning Agent is initializing")
|
|
self.scanner = ScannerAgent()
|
|
self.ensemble = EnsembleAgent(collection)
|
|
self.messenger = MessagingAgent()
|
|
self.log("Planning Agent is ready")
|
|
|
|
def run(self, deal: Deal) -> Opportunity:
|
|
"""
|
|
Run the workflow for a particular deal
|
|
:param deal: the deal, summarized from an RSS scrape
|
|
:returns: an opportunity including the discount
|
|
"""
|
|
self.log("Planning Agent is pricing up a potential deal")
|
|
estimate = self.ensemble.price(deal.product_description)
|
|
discount = estimate - deal.price
|
|
self.log(f"Planning Agent has processed a deal with discount ${discount:.2f}")
|
|
return Opportunity(deal=deal, estimate=estimate, discount=discount)
|
|
|
|
def plan(self, memory: List[str] = []) -> Optional[Opportunity]:
|
|
"""
|
|
Run the full workflow:
|
|
1. Use the ScannerAgent to find deals from RSS feeds
|
|
2. Use the EnsembleAgent to estimate them
|
|
3. Use the MessagingAgent to send a notification of deals
|
|
:param memory: a list of URLs that have been surfaced in the past
|
|
:return: an Opportunity if one was surfaced, otherwise None
|
|
"""
|
|
self.log("Planning Agent is kicking off a run")
|
|
selection = self.scanner.scan(memory=memory)
|
|
if selection:
|
|
opportunities = [self.run(deal) for deal in selection.deals[:5]]
|
|
opportunities.sort(key=lambda opp: opp.discount, reverse=True)
|
|
best = opportunities[0]
|
|
self.log(f"Planning Agent has identified the best deal has discount ${best.discount:.2f}")
|
|
if best.discount > self.DEAL_THRESHOLD:
|
|
self.messenger.alert(best)
|
|
self.log("Planning Agent has completed a run")
|
|
return best if best.discount > self.DEAL_THRESHOLD else None
|
|
return None |