Spaces:
Sleeping
Sleeping
File size: 10,248 Bytes
d8d14f1 |
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 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 |
"""
Zoe - Real Estate Agent
"""
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from datetime import datetime
import os
import json
import requests
from loguru import logger
from swarms import Agent
from swarm_models import OpenAIChat
from dotenv import load_dotenv
from enum import Enum
# Configure loguru logger
logger.add(
"logs/real_estate_agent_{time}.log",
rotation="500 MB",
retention="10 days",
level="INFO",
format="{time:YYYY-MM-DD at HH:mm:ss} | {level} | {message}",
)
class PropertyType(str, Enum):
"""Enum for property types"""
OFFICE = "office"
RETAIL = "retail"
INDUSTRIAL = "industrial"
MIXED_USE = "mixed-use"
LAND = "land"
@dataclass
class PropertyListing:
"""Data class for commercial property listings"""
property_id: str
address: str
city: str
state: str
zip_code: str
price: float
square_footage: float
property_type: PropertyType
zoning: str
listing_date: datetime
lat: float
lng: float
description: Optional[str] = None
features: Optional[List[str]] = None
images: Optional[List[str]] = None
class PropertyRadarAPI:
"""Client for PropertyRadar API integration"""
def __init__(self, api_key: str):
"""Initialize PropertyRadar API client
Args:
api_key (str): PropertyRadar API key
"""
self.api_key = api_key
self.base_url = "https://api.propertyradar.com/v1"
self.session = requests.Session()
self.session.headers.update(
{
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
)
def search_properties(
self,
max_price: float = 10_000_000,
property_types: List[PropertyType] = None,
location: Dict[str, Any] = None,
min_sqft: Optional[float] = None,
max_sqft: Optional[float] = None,
page: int = 1,
limit: int = 20,
) -> List[PropertyListing]:
"""
Search for commercial properties using PropertyRadar API
Args:
max_price (float): Maximum property price
property_types (List[PropertyType]): Types of properties to search for
location (Dict[str, Any]): Location criteria (city, county, or coordinates)
min_sqft (Optional[float]): Minimum square footage
max_sqft (Optional[float]): Maximum square footage
page (int): Page number for pagination
limit (int): Number of results per page
Returns:
List[PropertyListing]: List of matching properties
"""
try:
# Build the query parameters
params = {
"price_max": max_price,
"property_types": (
[pt.value for pt in property_types]
if property_types
else None
),
"page": page,
"limit": limit,
"for_sale": True,
"state": "FL", # Florida only
"commercial_property": True,
}
# Add location parameters
if location:
params.update(location)
# Add square footage filters
if min_sqft:
params["square_feet_min"] = min_sqft
if max_sqft:
params["square_feet_max"] = max_sqft
# Make the API request
response = self.session.get(
f"{self.base_url}/properties",
params={
k: v for k, v in params.items() if v is not None
},
)
response.raise_for_status()
# Parse the response
properties_data = response.json()
# Convert to PropertyListing objects
return [
PropertyListing(
property_id=prop["id"],
address=prop["address"],
city=prop["city"],
state=prop["state"],
zip_code=prop["zip_code"],
price=float(prop["price"]),
square_footage=float(prop["square_feet"]),
property_type=PropertyType(prop["property_type"]),
zoning=prop["zoning"],
listing_date=datetime.fromisoformat(
prop["list_date"]
),
lat=float(prop["latitude"]),
lng=float(prop["longitude"]),
description=prop.get("description"),
features=prop.get("features", []),
images=prop.get("images", []),
)
for prop in properties_data["results"]
]
except requests.RequestException as e:
logger.error(f"Error fetching properties: {str(e)}")
raise
class CommercialRealEstateAgent:
"""Agent for searching and analyzing commercial real estate properties"""
def __init__(
self,
openai_api_key: str,
propertyradar_api_key: str,
model_name: str = "gpt-4",
temperature: float = 0.1,
saved_state_path: Optional[str] = None,
):
"""Initialize the real estate agent
Args:
openai_api_key (str): OpenAI API key
propertyradar_api_key (str): PropertyRadar API key
model_name (str): Name of the LLM model to use
temperature (float): Temperature setting for the LLM
saved_state_path (Optional[str]): Path to save agent state
"""
self.property_api = PropertyRadarAPI(propertyradar_api_key)
# Initialize OpenAI model
self.model = OpenAIChat(
openai_api_key=openai_api_key,
model_name=model_name,
temperature=temperature,
)
# Initialize the agent
self.agent = Agent(
agent_name="Commercial-Real-Estate-Agent",
system_prompt=self._get_system_prompt(),
llm=self.model,
max_loops=1,
autosave=True,
dashboard=False,
verbose=True,
saved_state_path=saved_state_path,
context_length=200000,
streaming_on=False,
)
logger.info(
"Commercial Real Estate Agent initialized successfully"
)
def _get_system_prompt(self) -> str:
"""Get the system prompt for the agent"""
return """You are a specialized commercial real estate agent assistant focused on Central Florida properties.
Your primary responsibilities are:
1. Search for commercial properties under $10 million
2. Focus on properties zoned for commercial use
3. Provide detailed analysis of property features, location benefits, and potential ROI
4. Consider local market conditions and growth potential
5. Verify zoning compliance and restrictions
When analyzing properties, consider:
- Current market valuations
- Local business development plans
- Traffic patterns and accessibility
- Nearby amenities and businesses
- Future development potential"""
def search_properties(
self,
max_price: float = 10_000_000,
property_types: List[PropertyType] = None,
location: Dict[str, Any] = None,
min_sqft: Optional[float] = None,
max_sqft: Optional[float] = None,
) -> List[Dict[str, Any]]:
"""
Search for properties and provide analysis
Args:
max_price (float): Maximum property price
property_types (List[PropertyType]): Types of properties to search
location (Dict[str, Any]): Location criteria
min_sqft (Optional[float]): Minimum square footage
max_sqft (Optional[float]): Maximum square footage
Returns:
List[Dict[str, Any]]: List of properties with analysis
"""
try:
# Search for properties
properties = self.property_api.search_properties(
max_price=max_price,
property_types=property_types,
location=location,
min_sqft=min_sqft,
max_sqft=max_sqft,
)
# Analyze each property
analyzed_properties = []
for prop in properties:
analysis = self.agent.run(
f"Analyze this commercial property:\n"
f"Address: {prop.address}, {prop.city}, FL {prop.zip_code}\n"
f"Price: ${prop.price:,.2f}\n"
f"Square Footage: {prop.square_footage:,.0f}\n"
f"Property Type: {prop.property_type.value}\n"
f"Zoning: {prop.zoning}\n"
f"Description: {prop.description or 'Not provided'}"
)
analyzed_properties.append(
{"property": prop.__dict__, "analysis": analysis}
)
logger.info(
f"Successfully analyzed {len(analyzed_properties)} properties"
)
return analyzed_properties
except Exception as e:
logger.error(
f"Error in property search and analysis: {str(e)}"
)
raise
def main():
"""Main function to demonstrate usage"""
load_dotenv()
# Initialize the agent
agent = CommercialRealEstateAgent(
openai_api_key=os.getenv("OPENAI_API_KEY"),
propertyradar_api_key=os.getenv("PROPERTYRADAR_API_KEY"),
saved_state_path="real_estate_agent_state.json",
)
# Example search
results = agent.search_properties(
max_price=5_000_000,
property_types=[PropertyType.RETAIL, PropertyType.OFFICE],
location={"city": "Orlando", "radius_miles": 25},
min_sqft=2000,
)
# Save results
with open("search_results.json", "w") as f:
json.dump(results, f, default=str, indent=2)
if __name__ == "__main__":
main()
|