-
Notifications
You must be signed in to change notification settings - Fork 1
/
real-estae-swar.py
166 lines (141 loc) · 6.52 KB
/
real-estae-swar.py
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
import os
from dotenv import load_dotenv
from typing import List, Dict, Any
from pydantic import BaseModel, Field
import logging
from transformers import AutoModelForCausalLM, AutoTokenizer
from swarms import Agent, OpenAIChat, Anthropic, GPT4VisionAPI, ChromaDB, ToolAgent
from swarms.models import LlamaModel
from swarms.structs import SequentialWorkflow, ConcurrentWorkflow, MixtureOfAgents
from swarms.utils.json_utils import base_model_to_json
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(filename='real_estate_ai_agent.log', level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# Initialize different models
claude_model = Anthropic(
temperature=0.5,
anthropic_api_key=os.environ.get("ANTHROPIC_API_KEY"),
model="claude-3.5"
)
llama_model = LlamaModel(model_path="path/to/llama/model.bin")
# Load Dolly model
dolly_model = AutoModelForCausalLM.from_pretrained(
"databricks/dolly-v2-12b",
load_in_4bit=True,
device_map="auto",
)
dolly_tokenizer = AutoTokenizer.from_pretrained("databricks/dolly-v2-12b")
# Initialize long-term memory
memory = ChromaDB(
metric="cosine",
n_results=3,
output_dir="real_estate_data",
docs_folder="property_docs",
)
# Define data schemas
class PropertySchema(BaseModel):
address: str = Field(..., title="Property address")
total_units: int = Field(..., title="Total number of units", ge=400)
price: float = Field(..., title="Total property price")
price_per_unit: float = Field(..., title="Price per unit")
occupancy_rate: float = Field(..., title="Current occupancy rate", ge=0, le=1)
average_rent: float = Field(..., title="Average monthly rent per unit")
estimated_roi: float = Field(..., title="Estimated annual ROI", ge=0)
neighborhood_score: int = Field(..., title="Neighborhood quality score (1-10)", ge=1, le=10)
proximity_to_services: int = Field(..., title="Proximity to essential services score (1-10)", ge=1, le=10)
condition_score: int = Field(..., title="Property condition score (1-10)", ge=1, le=10)
class MarketAnalysisSchema(BaseModel):
location: str = Field(..., title="Location analyzed")
population_growth: float = Field(..., title="Annual population growth rate")
job_market_score: int = Field(..., title="Job market strength (1-10)", ge=1, le=10)
average_income: float = Field(..., title="Average annual income in the area")
affordability_index: float = Field(..., title="Housing affordability index")
demand_forecast: str = Field(..., title="Forecasted demand for affordable housing")
# Define specialized agents
class RealEstateAgent(Agent):
def __init__(self, agent_name, system_prompt, model, *args, **kwargs):
super().__init__(
llm=model,
agent_name=agent_name,
system_prompt=system_prompt,
max_loops=5,
autosave=True,
dashboard=True,
long_term_memory=memory,
*args,
**kwargs
)
# Create specialized agents with different models
market_analyst = RealEstateAgent(
"Market Analyst",
"Analyze real estate markets focusing on multi-family properties (400+ units) for low-income housing. Provide detailed insights on market trends, growth potential, and affordability factors.",
claude_model
)
financial_analyst = RealEstateAgent(
"Financial Analyst",
"Analyze the financial aspects of multi-family properties (400+ units), focusing on affordability for low-income families. Provide ROI projections, cash flow analysis, and investment recommendations.",
llama_model
)
property_inspector = RealEstateAgent(
"Property Inspector",
"Evaluate property conditions and potential renovation needs for multi-family units (400+ units). Focus on cost-effective improvements that can enhance living conditions for low-income residents.",
claude_model
)
community_impact_analyst = ToolAgent(
name="Community Impact Analyst",
description="Assess the impact of affordable housing projects (400+ units) on local communities. Provide recommendations for maximizing positive community impact and addressing potential challenges.",
model=dolly_model,
tokenizer=dolly_tokenizer,
json_schema=base_model_to_json(PropertySchema)
)
# Define tool functions (same as before)
def search_properties(location: str, min_units: int, max_price: float) -> List[PropertySchema]:
# Implementation (same as before)
pass
def analyze_financials(property_data: PropertySchema) -> Dict[str, float]:
# Implementation (same as before)
pass
def assess_property_condition(property_data: PropertySchema) -> Dict[str, Any]:
# Implementation (same as before)
pass
def evaluate_community_impact(property_data: PropertySchema, location: str) -> Dict[str, Any]:
# Implementation (same as before)
pass
# Add tools to agents
market_analyst.add_tool(search_properties)
financial_analyst.add_tool(analyze_financials)
property_inspector.add_tool(assess_property_condition)
community_impact_analyst.add_tool(evaluate_community_impact)
# Create workflows (same as before)
sequential_workflow = SequentialWorkflow(
agents=[market_analyst, financial_analyst, property_inspector, community_impact_analyst],
max_loops=1
)
concurrent_workflow = ConcurrentWorkflow(max_workers=4)
concurrent_workflow.add(tasks=[
market_analyst.run("Analyze the multi-family property market in New York City for low-income housing"),
financial_analyst.run("Evaluate the financial viability of a 400-unit property priced at $50 million for low-income housing"),
property_inspector.run("Assess the condition and renovation needs of a 400-unit property at 123 Main St"),
community_impact_analyst.run("Evaluate the impact of a new 400-unit affordable housing project in Brooklyn")
])
mixture_of_agents = MixtureOfAgents(
name="Real Estate Investment Team",
agents=[market_analyst, financial_analyst, property_inspector, community_impact_analyst],
layers=3,
final_agent=market_analyst
)
# Main execution function (same as before)
def optimize_real_estate_investment(location: str, budget: float, min_units: int = 400):
# Implementation (same as before)
pass
if __name__ == "__main__":
location = "New York City"
budget = 100000000 # $100 million
recommendation = optimize_real_estate_investment(location, budget)
print("Final Investment Recommendation:")
print(recommendation)
# Save recommendation to file
with open("investment_recommendation.txt", "w") as f:
f.write(recommendation)