Spaces:
Running on Zero
Running on Zero
File size: 7,426 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 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 | from typing import Optional, List, Dict
from agents.agent import Agent
from agents.deals import Deal, Opportunity
from agents.scanner_agent import ScannerAgent
from agents.ensemble_agent import EnsembleAgent
from agents.messaging_agent import MessagingAgent
from openai import OpenAI
import json
class AutonomousPlanningAgent(Agent):
name = "Autonomous Planning Agent"
color = Agent.GREEN
MODEL = "gpt-5.1"
def __init__(self, collection):
"""
Create instances of the 3 Agents that this planner coordinates across
"""
self.log("Autonomous Planning Agent is initializing")
self.scanner = ScannerAgent()
self.ensemble = EnsembleAgent(collection)
self.messenger = MessagingAgent()
self.openai = OpenAI()
self.memory = None
self.opportunity = None
self.log("Autonomous Planning Agent is ready")
def scan_the_internet_for_bargains(self) -> str:
"""
Run the tool to scan
"""
self.log("Autonomous Planning agent is calling scanner")
results = self.scanner.scan(memory=self.memory)
return results.model_dump_json() if results else "No deals found"
def estimate_true_value(self, description: str) -> str:
"""
Run the tool to estimate true value
"""
self.log("Autonomous Planning agent is estimating value via Ensemble Agent")
estimate = self.ensemble.price(description)
return f"The estimated true value of {description} is {estimate}"
def notify_user_of_deal(
self, description: str, deal_price: float, estimated_true_value: float, url: str
) -> Dict:
"""
Run the tool to notify the user
"""
if self.opportunity:
self.log("Autonomous Planning agent is trying to notify the user a 2nd time; ignoring")
else:
self.log("Autonomous Planning agent is notifying user")
self.messenger.notify(description, deal_price, estimated_true_value, url)
deal = Deal(product_description=description, price=deal_price, url=url)
discount = estimated_true_value - deal_price
self.opportunity = Opportunity(
deal=deal, estimate=estimated_true_value, discount=discount
)
return "Notification sent ok"
scan_function = {
"name": "scan_the_internet_for_bargains",
"description": "Returns top bargains scraped from the internet along with the price each item is being offered for",
"parameters": {
"type": "object",
"properties": {},
"required": [],
"additionalProperties": False,
},
}
estimate_function = {
"name": "estimate_true_value",
"description": "Given the description of an item, estimate how much it is actually worth",
"parameters": {
"type": "object",
"properties": {
"description": {
"type": "string",
"description": "The description of the item to be estimated",
},
},
"required": ["description"],
"additionalProperties": False,
},
}
notify_function = {
"name": "notify_user_of_deal",
"description": "Send the user a push notification about the single most compelling deal; only call this one time",
"parameters": {
"type": "object",
"properties": {
"description": {
"type": "string",
"description": "The description of the item itself scraped from the internet",
},
"deal_price": {
"type": "number",
"description": "The price offered by this deal scraped from the internet",
},
"estimated_true_value": {
"type": "number",
"description": "The estimated actual value that this is worth",
},
"url": {
"type": "string",
"description": "The URL of this deal as scraped from the internet",
},
},
"required": ["description", "deal_price", "estimated_true_value", "url"],
"additionalProperties": False,
},
}
def get_tools(self):
"""
Return the json for the tools to be used
"""
return [
{"type": "function", "function": self.scan_function},
{"type": "function", "function": self.estimate_function},
{"type": "function", "function": self.notify_function},
]
def handle_tool_call(self, message):
"""
Actually call the tools associated with this message
"""
mapping = {
"scan_the_internet_for_bargains": self.scan_the_internet_for_bargains,
"estimate_true_value": self.estimate_true_value,
"notify_user_of_deal": self.notify_user_of_deal,
}
results = []
for tool_call in message.tool_calls:
tool_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
tool = mapping.get(tool_name)
result = tool(**arguments) if tool else ""
results.append({"role": "tool", "content": result, "tool_call_id": tool_call.id})
return results
system_message = "You find great deals on bargain products using your tools, and notify the user of the best bargain."
user_message = """
First, use your tool to scan the internet for bargain deals. Then for each deal, use your tool to estimate its true value.
Then pick the single most compelling deal where the price is much lower than the estimated true value, and use your tool to notify the user.
Then just reply OK to indicate success.
"""
messages = [
{"role": "system", "content": system_message},
{"role": "user", "content": user_message},
]
def plan(self, memory: List[str] = []) -> Optional[Opportunity]:
"""
Run the full workflow, providing the LLM with tools to surface scraped deals to the user
:param memory: a list of URLs that have been surfaced in the past
:return: an Opportunity if one was surfaced, otherwise None
"""
self.log("Autonomous Planning Agent is kicking off a run")
self.memory = memory
self.opportunity = None
messages = self.messages[:]
done = False
while not done:
response = self.openai.chat.completions.create(
model=self.MODEL, messages=messages, tools=self.get_tools()
)
if response.choices[0].finish_reason == "tool_calls":
message = response.choices[0].message
results = self.handle_tool_call(message)
messages.append(message)
messages.extend(results)
else:
done = True
reply = response.choices[0].message.content
self.log(f"Autonomous Planning Agent completed with: {reply}")
return self.opportunity
|