From the uDemy course on LLM engineering.
https://www.udemy.com/course/llm-engineering-master-ai-and-large-language-models
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
57 lines
2.4 KiB
57 lines
2.4 KiB
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 |