Spaces:
Running on Zero
Running on Zero
File size: 3,193 Bytes
58bd26a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | import os
import requests
from litellm import completion
from agents.deals import Opportunity
from agents.agent import Agent
from free_config import MESSAGING_MODEL, PUSHOVER_CONFIGURED
pushover_url = "https://api.pushover.net/1/messages.json"
class MessagingAgent(Agent):
name = "Messaging Agent"
color = Agent.WHITE
def __init__(self):
"""
Set up Pushover notifications (optional) and Groq for message crafting.
"""
self.log("Messaging Agent is initializing")
self.pushover_user = os.getenv("PUSHOVER_USER", "")
self.pushover_token = os.getenv("PUSHOVER_TOKEN", "")
self.model = MESSAGING_MODEL
if PUSHOVER_CONFIGURED:
self.log("Messaging Agent has Pushover configured")
else:
self.log("Messaging Agent: Pushover not configured — alerts will be logged only")
def push(self, text):
"""
Send a Push Notification using the Pushover API, or log if not configured.
"""
if not PUSHOVER_CONFIGURED:
self.log(f"Deal alert (Pushover skipped): {text[:120]}...")
return
self.log("Messaging Agent is sending a push notification")
payload = {
"user": self.pushover_user,
"token": self.pushover_token,
"message": text,
"sound": "cashregister",
}
requests.post(pushover_url, data=payload)
def alert(self, opportunity: Opportunity):
"""
Make an alert about the specified Opportunity
"""
text = f"Deal Alert! Price=${opportunity.deal.price:.2f}, "
text += f"Estimate=${opportunity.estimate:.2f}, "
text += f"Discount=${opportunity.discount:.2f} :"
text += opportunity.deal.product_description[:10] + "... "
text += opportunity.deal.url
self.push(text)
self.log("Messaging Agent has completed")
def craft_message(
self, description: str, deal_price: float, estimated_true_value: float
) -> str:
user_prompt = "Please summarize this great deal in 2-3 sentences to be sent as an exciting push notification alerting the user about this deal.\n"
user_prompt += f"Item Description: {description}\nOffered Price: {deal_price}\nEstimated true value: {estimated_true_value}"
user_prompt += "\n\nRespond only with the 2-3 sentence message which will be used to alert & excite the user about this deal"
response = completion(
model=self.model,
messages=[
{"role": "user", "content": user_prompt},
],
)
return response.choices[0].message.content
def notify(self, description: str, deal_price: float, estimated_true_value: float, url: str):
"""
Make an alert about the specified details
"""
self.log(f"Messaging Agent is using Groq ({self.model}) to craft the message")
text = self.craft_message(description, deal_price, estimated_true_value)
self.push(text[:200] + "... " + url)
self.log("Messaging Agent has completed")
|