Spaces:
Running on Zero
Running on Zero
| 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") | |