swarms/new_features_examples/real_estate_agent.py

"""
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()