# -*- coding: utf-8 -*-
"""
NYC Building Chatbot โ Streamlit + LangChain
Converted from Full_model_updated_functions.py (Gradio + OpenAI + LlamaIndex)
"""
import os
import sys
import subprocess
import time
import jwt
import base64
import secrets
import requests
import re
import io
import json
import difflib
import warnings
import logging
import contextlib
from typing import Annotated, Any
from typing_extensions import TypedDict
import pandas as pd
import streamlit as st
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.graph.message import add_messages
from huggingface_hub import hf_hub_download, HfApi
from huggingface_hub.utils import EntryNotFoundError
USAGE_DATASET_REPO = os.environ.get("USAGE_DATASET_REPO", "NYSERDA-CRE-Working-Group/nyserda_demo_useage_store")
USAGE_FILENAME = os.environ.get("USAGE_FILENAME", "usage.csv")
MAX_RUNS_PER_USER = int(os.environ.get("MAX_RUNS_PER_USER", "10"))
HF_TOKEN = os.environ.get("HF_TOKEN")
api = HfApi(token=HF_TOKEN)
# --- HF OAuth Configuration ---
# These are automatically set by HF Spaces when hf_oauth: true is in README.md
OAUTH_CLIENT_ID = os.environ.get("OAUTH_CLIENT_ID")
OAUTH_CLIENT_SECRET = os.environ.get("OAUTH_CLIENT_SECRET")
OPENID_PROVIDER_URL = os.environ.get("OPENID_PROVIDER_URL", "https://huggingface.co")
SPACE_HOST = os.environ.get("SPACE_HOST", "") # e.g. "user-space.hf.space"
def _get_oauth_state() -> str:
"""Generate a deterministic OAuth state string.
We can't use a random per-request state because Streamlit's session_state
is lost when the browser redirects to HF and back (new HTTP request = new
session). Instead we derive a stable HMAC from the client secret, which is
only known server-side, so it still protects against CSRF.
"""
import hashlib, hmac
secret = (OAUTH_CLIENT_SECRET or "fallback-secret").encode()
return hmac.new(secret, b"hf-oauth-state", hashlib.sha256).hexdigest()[:32]
def get_hf_user():
"""Check if user is logged in via HF OAuth (authorization code flow)."""
# Already authenticated this session
if "hf_user" in st.session_state:
return st.session_state["hf_user"]
params = st.query_params
# --- Handle OAuth callback: exchange code for token ---
code = params.get("code")
returned_state = params.get("state")
if code:
expected_state = _get_oauth_state()
if returned_state == expected_state:
user = _exchange_code_for_user(code)
if user:
st.session_state["hf_user"] = user
# Clean up URL params
st.query_params.clear()
st.rerun()
else:
st.error("OAuth login failed โ could not retrieve user info.")
st.query_params.clear()
else:
st.error("OAuth state mismatch โ please try signing in again.")
st.query_params.clear()
return None
def _exchange_code_for_user(code: str) -> str | None:
"""Exchange the OAuth authorization code for an access token, then fetch username."""
if not OAUTH_CLIENT_ID or not OAUTH_CLIENT_SECRET:
st.error("OAuth is not configured. Set hf_oauth: true in your Space's README.md")
return None
redirect_uri = _get_redirect_uri()
token_url = f"{OPENID_PROVIDER_URL}/oauth/token"
# Build Basic auth header
credentials = base64.b64encode(f"{OAUTH_CLIENT_ID}:{OAUTH_CLIENT_SECRET}".encode()).decode()
try:
resp = requests.post(
token_url,
headers={"Authorization": f"Basic {credentials}"},
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": redirect_uri,
"client_id": OAUTH_CLIENT_ID,
},
timeout=10,
)
resp.raise_for_status()
token_data = resp.json()
except Exception as e:
st.error(f"Token exchange failed: {e}")
return None
access_token = token_data.get("access_token")
if not access_token:
st.error("No access token in OAuth response.")
return None
# Fetch user info
try:
userinfo_resp = requests.get(
f"{OPENID_PROVIDER_URL}/oauth/userinfo",
headers={"Authorization": f"Bearer {access_token}"},
timeout=10,
)
userinfo_resp.raise_for_status()
userinfo = userinfo_resp.json()
except Exception as e:
st.error(f"Failed to fetch user info: {e}")
return None
username = userinfo.get("preferred_username") or userinfo.get("sub")
if username:
return username.strip().lower()
return None
def _get_redirect_uri() -> str:
"""Build the OAuth redirect URI pointing back to this Space."""
if SPACE_HOST:
return f"https://{SPACE_HOST}"
# Fallback: try to reconstruct from the current request
return "http://localhost:8501"
def login_button():
"""Show a 'Sign in with Hugging Face' button that starts the OAuth flow."""
if not OAUTH_CLIENT_ID:
st.error(
"OAuth is not configured for this Space. "
"Add `hf_oauth: true` to your README.md metadata."
)
return
if st.button("๐ค Sign in with Hugging Face", use_container_width=True, type="primary"):
state = _get_oauth_state()
redirect_uri = _get_redirect_uri()
auth_url = (
f"{OPENID_PROVIDER_URL}/oauth/authorize"
f"?client_id={OAUTH_CLIENT_ID}"
f"&redirect_uri={redirect_uri}"
f"&scope=openid%20profile"
f"&response_type=code"
f"&state={state}"
)
# Use JS redirect since Streamlit can't do HTTP redirects natively
st.markdown(
f'',
unsafe_allow_html=True,
)
st.stop()
def _load_usage_df() -> pd.DataFrame:
try:
local_path = hf_hub_download(
repo_id=USAGE_DATASET_REPO,
repo_type="dataset",
filename=USAGE_FILENAME,
token=HF_TOKEN,
)
return pd.read_csv(local_path)
except EntryNotFoundError:
# First run: create empty table
return pd.DataFrame(columns=["user_id", "runs", "first_seen", "last_seen"])
def _save_usage_df(df: pd.DataFrame, commit_message: str) -> None:
tmp_path = "/tmp/usage.csv"
df.to_csv(tmp_path, index=False)
api.upload_file(
path_or_fileobj=tmp_path,
path_in_repo=USAGE_FILENAME,
repo_id=USAGE_DATASET_REPO,
repo_type="dataset",
commit_message=commit_message,
)
def check_and_increment_quota(user_id: str) -> tuple[bool, int]:
now = int(time.time())
df = _load_usage_df()
if df.empty or (df["user_id"] == user_id).sum() == 0:
runs = 0
if runs >= MAX_RUNS_PER_USER:
return False, 0
new_row = {
"user_id": user_id,
"runs": 1,
"first_seen": now,
"last_seen": now,
}
df = pd.concat([df, pd.DataFrame([new_row])], ignore_index=True)
_save_usage_df(df, commit_message=f"usage: increment {user_id} to 1")
return True, MAX_RUNS_PER_USER - 1
idx = df.index[df["user_id"] == user_id][0]
runs = int(df.loc[idx, "runs"])
if runs >= MAX_RUNS_PER_USER:
return False, 0
runs += 1
df.loc[idx, "runs"] = runs
df.loc[idx, "last_seen"] = now
_save_usage_df(df, commit_message=f"usage: increment {user_id} to {runs}")
return True, MAX_RUNS_PER_USER - runs
load_dotenv()
# Suppress noisy pandas regex warnings from LLM-generated code
warnings.filterwarnings("ignore", message=".*interpreted as a regular expression.*")
# Set up logging so all agent activity shows in the terminal
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("nyc_chatbot")
# =============================================================================
# Constants โ LL97 domain data
# =============================================================================
LL97_DESCRIPTION = """
Local Law 97 (LL97) of New York City limits greenhouse gas emissions for buildings
over 25,000 square feet. Starting in 2024, each building type has a maximum allowed
emissions intensity (metric tons CO2e per square foot per year). Exceeding this limit
leads to a fine of $268 per metric ton over the limit, per year.
Key points:
- Applies to buildings >25,000 sqft or two or more buildings on the same tax lot that together exceed 50,000 gross square feet.
- Limits depend on occupancy group (e.g., residential, office, university).
- Calculate: building_emissions = Site EUI x emissions_factor x floor_area.
- Compare to LL97 threshold for that occupancy group.
- Fine = (building_emissions - threshold x floor_area) x $268 if positive.
- Retrofits (insulation, HVAC upgrades, electrification) can lower site EUI or emissions factors to avoid fines.
- Emission Factors are:
Electricity: 0.000288962 tCO2e/kWh
Natural Gas: 0.00005311 tCO2e/kBtu
#2 Fuel Oil: 0.00007421 tCO2e/kBtu
#4 Fuel Oil: 0.00007529 tCO2e/kBtu
District Steam: 0.00004493 tCO2e/kBtu
"""
LL97_LIMITS = {
"Adult Education": {"2024": 0.00846},
"Ambulatory Surgical Center": {"2024": 0.02381},
"Automobile Dealership": {"2024": 0.00846},
"Bank Branch": {"2024": 0.00846},
"Bowling Alley": {"2024": 0.01074},
"College/University": {"2024": 0.00846},
"Convenience Store without Gas Station": {"2024": 0.01181},
"Courthouse": {"2024": 0.01074},
"Data Center": {"2024": 0.00846},
"Distribution Center": {"2024": 0.00426},
"Enclosed Mall": {"2024": 0.01181},
"Financial Office": {"2024": 0.00846},
"Fitness Center/Health Club/Gym": {"2024": 0.01074},
"Food Sales": {"2024": 0.01181},
"Food Service": {"2024": 0.01181},
"Hospital (General Medical & Surgical)": {"2024": 0.02381},
"Hotel": {"2024": 0.00987},
"K-12 School": {"2024": 0.00758},
"Laboratory": {"2024": 0.02381},
"Library": {"2024": 0.00846},
"Lifestyle Center": {"2024": 0.01181},
"Mailing Center/Post Office": {"2024": 0.00846},
"Manufacturing/Industrial Plant": {"2024": 0.00574},
"Medical Office": {"2024": 0.00846},
"Movie Theater": {"2024": 0.01074},
"Multifamily Housing": {"2024": 0.00675},
"Museum": {"2024": 0.01074},
"Non-Refrigerated Warehouse": {"2024": 0.00426},
"Office": {"2024": 0.00846},
"Other - Education": {"2024": 0.00846},
"Other - Entertainment/Public Assembly": {"2024": 0.01074},
"Other - Lodging/Residential": {"2024": 0.00987},
"Other - Mall": {"2024": 0.01181},
"Other - Public Services": {"2024": 0.00846},
"Other - Recreation": {"2024": 0.01074},
"Other - Restaurant/Bar": {"2024": 0.01074},
"Other - Services": {"2024": 0.00846},
"Other - Specialty Hospital": {"2024": 0.02381},
"Other - Technology/Science": {"2024": 0.02381},
"Outpatient Rehabilitation/Physical Therapy": {"2024": 0.00846},
"Parking": {"2024": 0.00426},
"Performing Arts": {"2024": 0.01074},
"Personal Services (Health/Beauty, Dry Cleaning, etc.)": {"2024": 0.00846},
"Pre-school/Daycare": {"2024": 0.00758},
"Refrigerated Warehouse": {"2024": 0.00426},
"Repair Services (Vehicle, Shoe, Locksmith, etc.)": {"2024": 0.00574},
"Residence Hall/Dormitory": {"2024": 0.00987},
"Residential Care Facility": {"2024": 0.01138},
"Restaurant": {"2024": 0.01074},
"Retail Store": {"2024": 0.01181},
"Self-Storage Facility": {"2024": 0.00426},
"Senior Care Community": {"2024": 0.02381},
"Social/Meeting Hall": {"2024": 0.01074},
"Strip Mall": {"2024": 0.01181},
"Supermarket/Grocery Store": {"2024": 0.01181},
"Transportation Terminal/Station": {"2024": 0.01074},
"Urgent Care/Clinic/Other Outpatient": {"2024": 0.00846},
"Vocational School": {"2024": 0.00758},
"Wholesale Club/Supercenter": {"2024": 0.01181},
"Worship Facility": {"2024": 0.01074},
}
BOROUGH_MAP = {
"manhattan": "MN", "mn": "MN",
"brooklyn": "BK", "bk": "BK",
"bronx": "BX", "bx": "BX",
"queens": "QN", "qn": "QN",
"staten island": "SI", "si": "SI",
}
# Emissions factors and unit conversions for computing CO2e reductions from fuel savings.
# Used by the compute_emissions_reduction tool and available in the exec namespace.
EMISSIONS_FACTORS = {
"electricity": {"tCO2e_per_kWh": 0.000288962, "kBtu_per_unit": 3.412, "unit": "kWh"},
"natural_gas": {"tCO2e_per_kBtu": 0.00005311, "kBtu_per_unit": 100.0, "unit": "therms"},
"fuel_oil_2": {"tCO2e_per_kBtu": 0.00007421, "kBtu_per_unit": 138.0, "unit": "gallons"},
"fuel_oil_4": {"tCO2e_per_kBtu": 0.00007529, "kBtu_per_unit": 145.0, "unit": "gallons"},
"district_steam": {"tCO2e_per_kBtu": 0.00004493, "kBtu_per_unit": 1.0, "unit": "mlbs"},
}
# GFA column candidates in order of preference (used for LL97 compliance calculations)
GFA_COLUMN_CANDIDATES = [
"Property GFA - Self-Reported (ftยฒ)",
"Property GFA - Calculated (Buildings) (ftยฒ)",
"Property GFA - Calculated (Buildings and Parking) (ftยฒ)",
"Largest Property Use Type - Gross Floor Area (ftยฒ)",
]
# =============================================================================
# LangGraph State Schema โ carries intermediate results across tool calls
# =============================================================================
class ChatbotState(TypedDict):
messages: Annotated[list, add_messages] # Core message channel (auto-accumulates)
filtered_bbls: list[str] # BBLs discovered in previous steps
ll97_limits_cache: dict[str, Any] # Looked-up LL97 limits by building type
emissions_reductions: list[dict[str, Any]] # CO2e reduction results from compute_emissions_reduction
query_results: list[dict[str, Any]] # General query outputs for cross-reference
analysis_steps: list[str] # Tracks completed analysis steps in current turn
execution_plan: list[str] # Steps from the planning node
plan_step_index: int # Current step being executed
SYSTEM_PROMPT = f"""You are a data analyst answering questions about NYC building datasets.
CRITICAL: All three datasets are ALREADY LOADED and accessible through your tools. You have FULL ACCESS to the data RIGHT NOW. NEVER tell the user you cannot access the data or ask them to load it. Just USE YOUR TOOLS to answer questions.
DATASETS AVAILABLE (already loaded as DataFrames in the code interpreter):
- map_pluto (or df1): Parcel-level tax lot data (BBL, addresses, zoning, building attributes).
- LL84 (or df3): Benchmarking data (annual energy use, emissions, EUI, GFA, property IDs, years).
Key columns: 'BBL', 'Calendar Year', 'OwnerName', 'Borough', 'Largest Property Use Type',
'Total (Location-Based) GHG Emissions (Metric Tons CO2e)', 'Site EUI (kBtu/ftยฒ)',
GFA columns (try GFA_COLUMN_CANDIDATES list for the right one).
Always use LL84/df3 for energy/emissions related queries and compliance checks, not map_pluto.
- LL87 (or df2): Audit/retrofit data (Energy Conservation Measures, audit year, floor area, systems).
Has 2,080 columns including 48 "Measure Name"/"Measure Description" columns for ECM packages
(e.g. "Measure Name: Package 1-ECM 1", "Measure Description: Package 1-ECM 1", up to Package 4-ECM 12).
Fuel savings columns: "Annual Fuel {{i}} Savings: Package {{n}}" (i=1-5, n=1-4),
fuel type: "Annual Fuel {{i}} Annual Energy and Cost Savings: Energy Type: Package {{n}}",
units: "Annual Fuel {{i}} Annual Energy and Cost Savings: Energy Units: Package {{n}}".
These datasets refer to many of the same buildings.
Columns ending in '_pluto' in LL87/LL84 were merged from map_pluto.
Always try to search both the normal column and the _pluto column.
Borough codes: MN=Manhattan, BK=Brooklyn, BX=Bronx, QN=Queens, SI=Staten Island.
CONSTANTS AVAILABLE IN INTERPRETER:
- LL97_LIMITS: dict mapping building types to {{"2024": limit_tCO2e_per_sqft}} (e.g. LL97_LIMITS["Office"]["2024"] = 0.00846)
- EMISSIONS_FACTORS: dict with conversion factors for electricity, natural_gas, fuel_oil_2, fuel_oil_4, district_steam
- GFA_COLUMN_CANDIDATES: list of GFA column names to try in order of preference
- BOROUGH_MAP: dict for normalizing borough names to codes
TOOL ROUTING โ you have 4 tools:
1. **execute_code** โ YOUR PRIMARY TOOL. Use for ALL data queries, filtering, searching, aggregation,
statistics, compliance calculations, measure lookups, cross-dataset joins, and any pandas operations.
Write pandas code directly. Variables persist across calls โ build incrementally.
Assign results to 'result' to see a summary. Use print() for intermediate outputs.
LL97_LIMITS, EMISSIONS_FACTORS, GFA_COLUMN_CANDIDATES, BOROUGH_MAP are all pre-loaded.
2. **get_emissions_limit** โ Quick lookup of the LL97 emissions limit for a building type and year.
Returns the limit in tCO2e/sqft/year. Use for simple limit lookups.
For anything beyond a simple lookup, use execute_code with LL97_LIMITS dict directly.
3. **compute_emissions_reduction** โ Convert a single fuel saving (from LL87 data) into CO2e reduction.
Pass fuel type, savings amount, and units. Returns emissions_mtco2e.
For bulk conversions of many fuel entries, prefer execute_code with EMISSIONS_FACTORS dict.
4. **reset_interpreter** โ Clear all user-created variables. DataFrames and constants are preserved.
CODE PATTERNS FOR COMMON TASKS:
PATTERN A โ LL97 Compliance Check (find buildings near/over threshold):
```python
# Step 1: Filter LL84 for the target buildings
# For owner search:
mask = LL84['OwnerName'].fillna('').str.contains('owner_name', case=False, regex=True)
# Also try: LL84['OwnerName_pluto'].fillna('').str.contains(...)
# For borough: LL84['Borough_pluto'] == 'MN'
# For building type: LL84['Largest Property Use Type'].str.contains('Office', case=False)
subset = LL84[mask].copy()
# Step 2: Deduplicate by BBL keeping most recent year
subset = subset.sort_values('Calendar Year', ascending=False).drop_duplicates(subset='BBL', keep='first')
# Step 3: Compute emissions intensity and classify
ghg_col = 'Total (Location-Based) GHG Emissions (Metric Tons CO2e)'
# Find the GFA column
gfa_col = next((c for c in GFA_COLUMN_CANDIDATES if c in subset.columns), None)
subset['ghg'] = pd.to_numeric(subset[ghg_col], errors='coerce')
subset['gfa'] = pd.to_numeric(subset[gfa_col], errors='coerce')
subset = subset[(subset['gfa'] > 0) & subset['ghg'].notna()]
# Match building type to LL97 limits (use difflib for fuzzy matching)
def match_type(use_type):
if pd.isna(use_type): return None
s = str(use_type).strip()
if s in LL97_LIMITS: return s
matches = difflib.get_close_matches(s, LL97_LIMITS.keys(), n=1, cutoff=0.6)
return matches[0] if matches else None
subset['matched_type'] = subset['Largest Property Use Type'].apply(match_type)
subset = subset[subset['matched_type'].notna()]
subset['limit'] = subset['matched_type'].map(lambda t: LL97_LIMITS[t]['2024'])
subset['intensity'] = subset['ghg'] / subset['gfa']
subset['pct_of_limit'] = subset['intensity'] / subset['limit']
subset['status'] = subset['pct_of_limit'].apply(lambda p: 'over' if p >= 1.0 else ('near' if p >= 0.8 else 'under'))
result = subset[['BBL', 'matched_type', 'ghg', 'gfa', 'intensity', 'limit', 'pct_of_limit', 'status']].sort_values('pct_of_limit', ascending=False)
```
PATTERN B โ Extract LL87 ECM Savings for BBLs and Convert to CO2e:
```python
# Get BBLs from previous compliance check
target_bbls = subset[subset['status'].isin(['near','over'])]['BBL'].astype(str).tolist()
# Filter LL87
ll87_match = LL87[LL87['BBL'].astype(str).isin(target_bbls)]
# Extract fuel savings across all 4 packages, up to 5 fuels each
savings_rows = []
for _, row in ll87_match.iterrows():
bbl = str(row['BBL'])
for pkg in range(1, 5):
for fuel_i in range(1, 6):
sav_col = f'Annual Fuel {{fuel_i}} Savings: Package {{pkg}}'
type_col = f'Annual Fuel {{fuel_i}} Annual Energy and Cost Savings: Energy Type: Package {{pkg}}'
unit_col = f'Annual Fuel {{fuel_i}} Annual Energy and Cost Savings: Energy Units: Package {{pkg}}'
if sav_col in LL87.columns:
sav = pd.to_numeric(row.get(sav_col), errors='coerce')
ftype = row.get(type_col)
funit = row.get(unit_col)
if pd.notna(sav) and sav != 0 and pd.notna(ftype):
savings_rows.append({{'BBL': bbl, 'package': pkg, 'fuel': str(ftype), 'savings': sav, 'units': str(funit) if pd.notna(funit) else 'unknown'}})
savings_df = pd.DataFrame(savings_rows)
result = savings_df
```
Then call compute_emissions_reduction for each unique fuel/savings/units combo, or compute in code using EMISSIONS_FACTORS.
PATTERN C โ Find ECM Measure Names and Descriptions:
```python
# Get measure name/description columns
measure_cols = [c for c in LL87.columns if 'Measure Name' in c or 'Measure Description' in c]
# Filter by BBL, borough, or other criteria
filtered = LL87[LL87['Borough_pluto'] == 'MN'][['BBL'] + measure_cols]
# Stack to find most common measures
measures = filtered[measure_cols].stack().dropna()
result = measures.value_counts().head(20)
```
PATTERN D โ Search Measure Descriptions for Specific Equipment:
```python
desc_cols = [c for c in LL87.columns if 'Measure Description' in c]
all_descs = LL87[desc_cols].stack().dropna()
# Search for specific terms
for term in ['T12', 'T8', 'LED', 'incandescent', 'boiler']:
count = all_descs.str.contains(term, case=False).sum()
print(f'{{term}}: {{count}} matches')
```
IMPORTANT RULES:
- Use execute_code as your PRIMARY tool for ALL data analysis. The code interpreter has direct
access to the DataFrames, LL97_LIMITS, EMISSIONS_FACTORS, and all pandas/numpy functionality.
- Only use get_emissions_limit for quick single-type limit lookups.
- Only use compute_emissions_reduction when you have a small number of individual fuel entries.
For bulk conversion (many buildings), compute directly in code using EMISSIONS_FACTORS.
- When searching for owners, try MULTIPLE patterns โ e.g. for "Department of Education", also
search for "DOE", "NYC DOE", "BOARD OF EDUCATION", etc. Build regex patterns with | (OR).
- When querying LL84, ALWAYS deduplicate by BBL keeping the most recent Calendar Year.
- For building type matching, use difflib.get_close_matches() against LL97_LIMITS keys.
- LL87 has 2,080 columns. Use [c for c in LL87.columns if 'keyword' in c.lower()] to explore.
- Before writing execute_code, check if useful variables already exist in the interpreter state.
- When answering give context โ what analysis did you do and what does it mean.
- If available give specific supporting data โ distributions, counts, or examples from the dataset.
- If a dataset has no matching rows, say so explicitly.
- Do NOT say "measures exist so they'll be under the threshold" โ CALCULATE the actual reduction.
- LL97 and LL84 apply to buildings > 25,000 sqft; LL87 applies to buildings > 50,000 sqft.
CRITICAL โ LL97 BUILDING TYPE MATCHING:
- NEVER assume a single LL97 building type for an entire owner portfolio. An owner like NYC DOE
has K-12 Schools, Offices, Pre-schools, and other use types โ each has a DIFFERENT LL97 limit.
- ALWAYS match EACH building's "Largest Property Use Type" from LL84 individually to LL97_LIMITS
using difflib.get_close_matches(). See Pattern A above for the per-row approach.
- Key LL97 types for education: "K-12 School" (0.00758), "Pre-school/Daycare" (0.00758),
"College/University" (0.00846), "Adult Education" (0.00846). Do NOT default to "Adult Education"
for K-12 schools โ use the actual use type from the data.
Here is the context for LL97:
{LL97_DESCRIPTION}
PERSISTENT CODE INTERPRETER:
Your execute_code tool runs in a persistent Python environment. Variables survive across calls
within the same conversation. When the "CURRENT ANALYSIS STATE" shows interpreter variables,
reference them directly in subsequent calls โ build on prior work rather than starting from scratch.
"""
PLANNER_PROMPT = """You are a planning assistant for an NYC building data analyst.
Given the user's question, decompose it into an ordered list of concrete analysis steps.
Each step should describe ONE execute_code call or tool call.
Return ONLY a JSON object: {"steps": ["step 1 description", "step 2 description", ...]}
If the question is simple and needs only one tool call, return {"steps": ["single step description"]}.
Available tools:
- execute_code: Run pandas code on map_pluto, LL87, LL84 DataFrames (PRIMARY tool for all data queries)
Also has access to: LL97_LIMITS dict, EMISSIONS_FACTORS dict, GFA_COLUMN_CANDIDATES, BOROUGH_MAP, difflib
Always filter with the LL84 benchmarking dataset for energy/emissions related questions, not map_pluto.
- get_emissions_limit: Quick lookup of LL97 limit by building type and year
- compute_emissions_reduction: Convert one fuel saving entry to CO2e reduction
- reset_interpreter: Clear interpreter variables
IMPORTANT: Use execute_code for ALL data searching, filtering, aggregation, compliance calculations,
and measure lookups. get_emissions_limit and compute_emissions_reduction are convenience helpers only.
Example for "Find Manhattan hotels near LL97 limit that have ECMs":
{"steps": [
"execute_code: filter LL84 for Manhattan hotels, deduplicate by BBL, compute emissions intensity, classify as over/near/under using LL97_LIMITS",
"execute_code: extract fuel savings from LL87 for near/over BBLs across all packages",
"compute_emissions_reduction for each fuel entry, or compute in code using EMISSIONS_FACTORS",
"execute_code: compare original vs post-ECM emissions, summarize which hotels would fall under threshold"
]}
Example for "Find buildings owned by Vornado near the LL97 limit":
{"steps": [
"execute_code: search LL84 OwnerName for 'Vornado' (try multiple patterns), get full BBL list, deduplicate by year",
"execute_code: compute emissions intensity for all Vornado BBLs, look up LL97 limits, classify status",
"execute_code: for near/over buildings, extract LL87 fuel savings across packages 1-4",
"execute_code: convert fuel savings to CO2e using EMISSIONS_FACTORS, compute post-ECM intensity",
"Summarize which Vornado buildings would fall under threshold after ECMs"
]}
Example for "What are the most common retrofit measures in Manhattan?":
{"steps": [
"execute_code: filter LL87 by Borough_pluto=='MN', get measure name/description columns, stack and count most common values"
]}"""
# Simple-query heuristic patterns for skipping the planner
_SIMPLE_QUERY_PATTERNS = [
(re.compile(r'\b(ll97|emissions?)\s+limit', re.I), "get_emissions_limit"),
(re.compile(r'\bhow\s+much\s+co2', re.I), "compute_emissions_reduction"),
(re.compile(r'\bconvert\b.*\b(kwh|therms?|gallons?|mlbs)\b', re.I), "compute_emissions_reduction"),
]
_MAX_SIMPLE_QUERY_LEN = 80
def _is_simple_query(question: str) -> bool:
"""Check if a question is simple enough to skip the planner."""
if len(question) > _MAX_SIMPLE_QUERY_LEN:
return False
return any(pattern.search(question) for pattern, _ in _SIMPLE_QUERY_PATTERNS)
# =============================================================================
# Data Loading
# =============================================================================
@st.cache_resource(show_spinner="Loading NYC building datasets...")
def load_dataframes():
"""Load and preprocess CSV datasets. Returns dict of DataFrames."""
base = os.path.dirname(os.path.abspath(__file__))
dataframes = {}
# map_pluto
pluto_path = os.path.join(base, "map_pluto.csv")
dataframes["map_pluto"] = pd.read_csv(pluto_path, low_memory=False)
# LL87 โ drop the original Borough column (Borough_pluto is kept from the merge)
ll87_path = os.path.join(base, "LL87_with_pluto.csv")
ll87_df = pd.read_csv(ll87_path, low_memory=False)
if "Borough" in ll87_df.columns:
ll87_df = ll87_df.drop("Borough", axis=1)
dataframes["LL87"] = ll87_df
# LL84
ll84_path = os.path.join(base, "LL84_with_pluto.csv")
dataframes["LL84"] = pd.read_csv(ll84_path, low_memory=False)
return dataframes
# =============================================================================
# Custom Tools โ domain logic that is NOT simple DataFrame querying
# =============================================================================
@tool
def get_emissions_limit(building_type: str, year: str) -> str:
"""Look up the LL97 emissions limit (tCO2e/sqft/year) for a building type and compliance year.
OUTPUT: Returns {"building_type": str, "year": str, "limit_tCO2e_per_sqft": float}.
To get total allowed emissions for a building, multiply limit_tCO2e_per_sqft by gross floor area (sqft).
A building is in violation if its actual emissions intensity exceeds this limit.
Args:
building_type: The building occupancy type, e.g. "Office", "Multifamily Housing", "Hotel".
Must match an LL97 category exactly.
year: The compliance period year. Currently only "2024" is available.
"""
logger.info(f"[get_emissions_limit] building_type={building_type}, year={year}")
if building_type not in LL97_LIMITS:
available = ", ".join(LL97_LIMITS.keys())
result = json.dumps({"error": f"Building type '{building_type}' not found. Valid types: {available}"})
logger.warning(f"[get_emissions_limit] {result}")
return result
if year not in LL97_LIMITS[building_type]:
result = json.dumps({"error": f"Year '{year}' not available for '{building_type}'."})
logger.warning(f"[get_emissions_limit] {result}")
return result
result = json.dumps({"building_type": building_type, "year": year, "limit_tCO2e_per_sqft": LL97_LIMITS[building_type][year]})
logger.info(f"[get_emissions_limit] Result: {result}")
return result
@tool
def compute_emissions_reduction(fuel: str, savings: float, units: str) -> str:
"""Convert a single LL87 ECM fuel saving into CO2e emissions reduction.
OUTPUT: Returns {"fuel": str, "savings_input": str, "savings_kbtu": float, "emissions_mtco2e": float}.
The emissions_mtco2e value is the CO2e reduction from this single fuel saving.
SUM all emissions_mtco2e values across fuels and packages to get total building CO2e reduction.
For bulk conversions of many entries, prefer using execute_code with EMISSIONS_FACTORS dict directly.
Args:
fuel: Fuel type, e.g. "Electric", "Natural Gas", "Fuel Oil #2", "Fuel Oil #4", "District Steam".
savings: Reported energy savings value (numeric).
units: Units of the savings, e.g. "kWh", "therms", "gallons", "mlbs".
"""
logger.info(f"[compute_emissions_reduction] fuel={fuel}, savings={savings}, units={units}")
fuel_lower = fuel.lower().strip()
units_lower = units.lower().strip()
kbtu = None
factor = None
if units_lower == "kwh":
kbtu = savings * 3.412
factor = 0.000288962 / 3.412 # tCO2e per kBtu (electricity)
elif units_lower in ("therms", "therm"):
kbtu = savings * 100.0
factor = 0.00005311
elif "gallon" in units_lower and "fuel" in fuel_lower and "2" in fuel_lower:
kbtu = savings * 138
factor = 0.00007421
elif "gallon" in units_lower and "fuel" in fuel_lower and "4" in fuel_lower:
kbtu = savings * 145
factor = 0.00007529
elif units_lower == "mlbs":
kbtu = savings
factor = 0.00004493
else:
return json.dumps({"error": f"Unsupported units: {units}"})
emissions = kbtu * factor
result = json.dumps({
"fuel": fuel,
"savings_input": f"{savings} {units}",
"savings_kbtu": round(kbtu, 2),
"emissions_mtco2e": round(emissions, 6),
})
logger.info(f"[compute_emissions_reduction] Result: {result}")
return result
def create_exec_tools(namespace: dict):
"""Factory that creates execute_code and reset_interpreter tools with the
namespace dict bound via closure. This avoids accessing st.session_state
from inside LangGraph's ToolNode thread pool (where Streamlit's thread-local
session state is unavailable).
"""
@tool
def execute_code(code: str) -> str:
"""Execute Python/pandas code in a persistent interpreter environment.
The interpreter maintains state across calls โ variables you create (e.g.
df_filtered, my_subset, counts) are available in subsequent calls.
IMPORTANT BEHAVIOR:
- The variable 'result' is CLEARED before each call. Do NOT reference 'result'
from a previous call. Instead, store intermediate data in named variables
(e.g. my_df, near_buildings, savings_df) and only assign 'result' at the END
of a call to display its summary.
- Use print() for intermediate outputs within the same call.
- Variables OTHER than 'result' persist across calls.
- For wide DataFrames (many columns), use print() with specific columns rather
than assigning to 'result', because the auto-summary truncates wide tables.
Example: print(df[['BBL','address','status','intensity']].to_string())
- ALWAYS print actual numeric values in your final output. Never write
placeholder text like "(see dataset)" โ the user cannot see the dataset.
PRE-LOADED DATAFRAMES:
- map_pluto (or df1): MapPLUTO parcel data (~857K rows, 94 cols)
Key cols: 'BBL', 'Address', 'Borough', 'OwnerName', 'BldgClass', 'LandUse',
'NumFloors', 'UnitsRes', 'UnitsTotal', 'LotArea', 'BldgArea', 'ZoneDist1'
- LL84 (or df3): LL84 benchmarking data
Key cols: 'BBL', 'Calendar Year', 'OwnerName', 'Borough', 'Borough_pluto',
'Largest Property Use Type', 'Largest Property Use Type - Gross Floor Area (ftยฒ)',
'Property GFA - Self-Reported (ftยฒ)', 'Site EUI (kBtu/ftยฒ)',
'Total (Location-Based) GHG Emissions (Metric Tons CO2e)',
'Address_pluto', 'OwnerName_pluto'
Always use this dataframe when doing a broad energy related search
- LL87 (or df2): LL87 audit/ECM data (~2,080 cols)
Key cols: 'BBL', 'Borough_pluto', 'Address_pluto', 'OwnerName_pluto'
Measure cols: 'Measure Name: Package {n}-ECM {m}', 'Measure Description: Package {n}-ECM {m}'
(n=1-4, m=1-12, so up to 48 measure name + 48 description columns)
Savings cols: 'Annual Fuel {i} Savings: Package {n}' (i=1-5, n=1-4)
Fuel type: 'Annual Fuel {i} Annual Energy and Cost Savings: Energy Type: Package {n}'
Units: 'Annual Fuel {i} Annual Energy and Cost Savings: Energy Units: Package {n}'
To explore columns: [c for c in LL87.columns if 'keyword' in c.lower()]
PRE-LOADED CONSTANTS:
- LL97_LIMITS: dict โ LL97_LIMITS["Office"]["2024"] โ 0.00846 (tCO2e/sqft)
- EMISSIONS_FACTORS: dict of dicts. Structure:
EMISSIONS_FACTORS["electricity"] = {"tCO2e_per_kbtu": 0.000288962/3.412, "kBtu_per_unit": 3.412, "unit": "kWh"}
EMISSIONS_FACTORS["natural_gas"] = {"tCO2e_per_kBtu": 0.00005311, "kBtu_per_unit": 100.0, "unit": "therms"}
EMISSIONS_FACTORS["fuel_oil_2"] = {"tCO2e_per_kBtu": 0.00007421, "kBtu_per_unit": 138.0, "unit": "gallons"}
EMISSIONS_FACTORS["fuel_oil_4"] = {"tCO2e_per_kBtu": 0.00007529, "kBtu_per_unit": 145.0, "unit": "gallons"}
EMISSIONS_FACTORS["district_steam"] = {"tCO2e_per_kBtu": 0.00004493, "kBtu_per_unit": 1.0, "unit": "mlbs"}
To convert kWh electricity: mtco2e = kwh_saved * EMISSIONS_FACTORS["electricity"]["tCO2e_per_kWh"]
To convert therms gas: kbtu = therms * 100; mtco2e = kbtu * EMISSIONS_FACTORS["natural_gas"]["tCO2e_per_kBtu"]
- GFA_COLUMN_CANDIDATES: list of GFA column names in preference order. Usage:
gfa_col = next((c for c in GFA_COLUMN_CANDIDATES if c in LL84.columns), None)
- BOROUGH_MAP: {"manhattan": "MN", "brooklyn": "BK", ...}
- difflib: for fuzzy matching (difflib.get_close_matches)
PRE-LOADED LIBRARIES: pd (pandas), json, re, np (numpy), difflib
CONVENTIONS:
- Borough codes: 'MN', 'BK', 'BX', 'QN', 'SI'. Use Borough_pluto in LL87.
- Columns ending in '_pluto' in LL87/LL84 were merged from map_pluto.
- BBLs are 10-digit strings. Always normalize: .astype(str).str.replace(r'\\.0$','',regex=True)
- When merging DataFrames, check for existing columns to avoid suffix conflicts.
Args:
code: Python code to execute. May span multiple lines.
"""
if not namespace:
return "ERROR: Interpreter namespace not initialized. Reload the app."
logger.info(f"[execute_code] Executing code:\n{code}")
output, error = _execute_code(code, namespace)
# Describe namespace AFTER execution so the model sees what it just created
ns_description = _describe_namespace(namespace)
if error:
logger.error(f"[execute_code] Error: {error}")
partial = f"Partial output:\n{output}\n\n" if output.strip() else ""
return f"{partial}ERROR: {error}\n\nAvailable variables:\n{ns_description}"
if not output.strip():
output = "(code executed, no output โ assign to 'result' or use print())"
result_text = f"{output}\n\nCurrent interpreter state:\n{ns_description}"
# Truncate total return to ~4000 chars to protect context window
if len(result_text) > 4000:
result_text = result_text[:3900] + "\n...[output truncated]"
logger.info(f"[execute_code] Output (first 500): {result_text[:500]}")
return result_text
@tool
def reset_interpreter() -> str:
"""Clear all user-created variables from the persistent interpreter.
Keeps the pre-loaded DataFrames (map_pluto, LL87, LL84, df1, df2, df3) and
library imports (pd, json, re, np). Only removes variables you created during
analysis. Use this when starting a fresh analysis or when variables are stale.
"""
keys_to_remove = [
k for k in namespace
if k not in _PROTECTED_NAMESPACE_KEYS and not k.startswith("__")
]
for k in keys_to_remove:
del namespace[k]
ns_description = _describe_namespace(namespace)
logger.info("[reset_interpreter] Namespace cleared")
return f"Interpreter reset. Remaining variables:\n{ns_description}"
return [execute_code, reset_interpreter]
def _normalize_borough(val):
"""Normalize borough name to two-letter code (MN, BK, BX, QN, SI)."""
return BOROUGH_MAP.get(str(val).strip().lower(), val)
# =============================================================================
# LangGraph Helpers โ state extraction and context injection
# =============================================================================
def _extract_bbls_from_result(content: str) -> list[str]:
"""Try to extract BBL identifiers (10-digit numbers) from pandas agent output."""
bbl_pattern = re.compile(r'\b(\d{10})\b')
matches = bbl_pattern.findall(content)
seen = set()
result = []
for m in matches:
if m not in seen:
seen.add(m)
result.append(m)
return result
# --- Persistent Execution Environment Helpers ---
_PROTECTED_NAMESPACE_KEYS = frozenset({
"pd", "json", "re", "np", "difflib",
"map_pluto", "LL87", "LL84",
"df1", "df2", "df3",
"LL97_LIMITS", "EMISSIONS_FACTORS", "GFA_COLUMN_CANDIDATES", "BOROUGH_MAP",
})
def _summarize_value(val) -> str:
"""Convert a Python object to a compact string summary for LLM context.
For DataFrames, limits both rows and columns to keep output under ~2500 chars
so it fits within the 4000-char tool output budget alongside namespace info.
"""
if isinstance(val, pd.DataFrame):
nrows, ncols = val.shape
header = f"DataFrame: {nrows} rows x {ncols} cols"
# If very wide, select a subset of columns to avoid truncation
if ncols > 8:
# Prefer columns with short names and numeric data; always include first and last
display_df = val.head(10)
# Show first 8 columns + note about remaining
display_str = display_df.iloc[:, :8].to_string()
omitted = ncols - 8
return f"{header}\nColumns: {list(val.columns)}\nShowing first 8 of {ncols} columns ({omitted} more):\n{display_str}"
else:
return f"{header}\n{val.head(10).to_string()}"
elif isinstance(val, pd.Series):
return f"Series ({len(val)} values):\n{val.head(10).to_string()}"
elif isinstance(val, (list, dict)):
if isinstance(val, list):
if len(val) > 200:
return f"list with {len(val)} items. First 20:\n{val[:20]}"
return repr(val)
else: # dict
if len(val) > 200:
return f"dict with {len(val)} keys. First 10:\n{dict(list(val.items())[:10])}"
return repr(val)
elif isinstance(val, str) and len(val) > 3000:
return val[:3000] + f"\n...[truncated, {len(val)} total chars]"
else:
return repr(val)[:2000]
def _execute_code(code: str, namespace: dict) -> tuple[str, str]:
"""Run code in the persistent namespace, return (stdout_output, error_or_empty)."""
# Clear 'result' before each execution so stale values don't leak
namespace.pop("result", None)
buf = io.StringIO()
try:
with contextlib.redirect_stdout(buf):
exec(code, namespace)
output = buf.getvalue()
# If code assigned to 'result' and nothing was printed, show the result summary
if not output.strip() and "result" in namespace:
raw = namespace["result"]
output = _summarize_value(raw)
return output, ""
except Exception as e:
return buf.getvalue(), f"{type(e).__name__}: {e}"
def _describe_namespace(namespace: dict) -> str:
"""Return a compact description of user-created variables in the namespace."""
user_vars = {
k: v for k, v in namespace.items()
if k not in _PROTECTED_NAMESPACE_KEYS and not k.startswith("_")
}
lines = []
for name, val in user_vars.items():
if isinstance(val, pd.DataFrame):
lines.append(f" {name}: DataFrame({val.shape[0]}r x {val.shape[1]}c)")
elif isinstance(val, pd.Series):
lines.append(f" {name}: Series(len={len(val)})")
elif isinstance(val, (list, dict)):
lines.append(f" {name}: {type(val).__name__}(len={len(val)})")
else:
lines.append(f" {name}: {type(val).__name__} = {repr(val)[:80]}")
return "\n".join(lines) if lines else " (no user variables yet)"
def _reset_namespace(namespace: dict = None):
"""Clear user-created variables from the namespace, keeping DataFrames and imports."""
if namespace is None:
namespace = st.session_state.get("exec_namespace", {})
keys_to_remove = [
k for k in namespace
if k not in _PROTECTED_NAMESPACE_KEYS and not k.startswith("__")
]
for k in keys_to_remove:
del namespace[k]
# --- Live Tool Call Log Helpers ---
def _format_tool_call_label(tool_call: dict) -> str:
"""Convert a tool call dict to a human-readable label for the live log."""
name = tool_call.get("name", "unknown")
args = tool_call.get("args", {})
if name == "execute_code":
code = args.get("code", "")
first_line = next((ln.strip() for ln in code.splitlines() if ln.strip()), code[:60])
return f"execute_code โ `{first_line[:80]}`"
elif name == "compute_emissions_reduction":
fuel = args.get("fuel", "")
savings = args.get("savings", "")
units = args.get("units", "")
return f"compute_emissions_reduction โ {savings} {units} of {fuel}"
elif name == "get_emissions_limit":
return f"get_emissions_limit โ {args.get('building_type', '')} ({args.get('year', '')})"
elif name == "reset_interpreter":
return "reset_interpreter"
else:
return f"{name} โ {str(args)[:60]}"
def _summarize_tool_result(tool_name: str, content: str) -> str:
"""Extract a brief one-line summary of a tool result for the live log."""
try:
data = json.loads(content)
except (json.JSONDecodeError, TypeError):
# Plain text result (e.g., execute_code)
lines = str(content).strip().splitlines()
first = lines[0][:120] if lines else "(no output)"
return first
if tool_name == "compute_emissions_reduction":
mtco2e = data.get("emissions_mtco2e", 0)
fuel = data.get("fuel", "")
return f"{mtco2e:.4f} mtCO2e saved ({fuel})"
elif tool_name == "get_emissions_limit":
limit = data.get("limit_tCO2e_per_sqft", "")
bt = data.get("building_type", "")
return f"{bt}: {limit} tCO2e/sqft/yr"
elif "error" in data:
return f"Error: {str(data['error'])[:100]}"
else:
return str(content)[:120]
def _build_state_context(state: ChatbotState, namespace: dict = None) -> str:
"""Build a tiered summary of accumulated analysis state for the LLM.
Tier 1 (always injected): interpreter variables, plan, BBLs.
Tier 2 (when non-empty): limits, reduction totals, analysis steps.
"""
parts = []
# --- Tier 1: Persistent interpreter variables ---
if namespace is None:
namespace = {}
if namespace:
ns_desc = _describe_namespace(namespace)
if ns_desc.strip() and ns_desc.strip() != "(no user variables yet)":
parts.append(f"Persistent interpreter variables:\n{ns_desc}")
# --- Tier 1: Execution plan ---
plan = state.get("execution_plan", [])
idx = state.get("plan_step_index", 0)
if plan:
completed = plan[:idx]
remaining = plan[idx:]
if completed:
parts.append(f"Plan steps completed: {', '.join(completed)}")
if remaining:
parts.append(f"Next plan step: {remaining[0]}")
if len(remaining) > 1:
parts.append(f" (then: {', '.join(remaining[1:])})")
# --- Tier 1: BBLs (compact) ---
bbls = state.get("filtered_bbls", [])
if bbls:
display = ", ".join(bbls[:50])
suffix = f"... ({len(bbls)} total)" if len(bbls) > 50 else f" ({len(bbls)} total)"
parts.append(f"Buildings identified: [{display}{suffix}]")
# --- Tier 2: CO2e reductions from compute_emissions_reduction tool ---
reductions = state.get("emissions_reductions", [])
if reductions:
total_co2e = sum(r.get("emissions_mtco2e", 0) for r in reductions)
parts.append(f"CO2e reductions computed: {total_co2e:.6f} mtCO2e from {len(reductions)} entries")
# --- Tier 2: LL97 limits cache ---
limits = state.get("ll97_limits_cache", {})
if limits:
parts.append(f"LL97 limits looked up: {json.dumps(limits)}")
# --- Tier 2: Analysis steps ---
steps = state.get("analysis_steps", [])
if steps:
parts.append(f"Completed analysis steps: {', '.join(steps)}")
return "\n".join(parts)
def _parse_tool_results_for_state(current_state: ChatbotState, tool_messages: list) -> dict:
"""Inspect ToolMessage outputs and extract state updates.
Called after each tool execution round to populate intermediate state
fields that subsequent tool calls can leverage.
"""
updates: dict[str, Any] = {}
for msg in tool_messages:
if not isinstance(msg, ToolMessage):
continue
tool_name = msg.name
content = msg.content
# Try to parse as JSON
try:
data = json.loads(content) if isinstance(content, str) else content
except (json.JSONDecodeError, TypeError):
data = None
if tool_name == "execute_code":
updates.setdefault("query_results", list(current_state.get("query_results", []))).append({
"tool": "execute_code",
"result": str(content)[:1000],
})
bbls = _extract_bbls_from_result(str(content))
if bbls:
updates["filtered_bbls"] = bbls
updates.setdefault("analysis_steps", list(current_state.get("analysis_steps", []))).append(
"executed_code"
)
elif tool_name == "reset_interpreter":
updates.setdefault("analysis_steps", list(current_state.get("analysis_steps", []))).append(
"reset_interpreter"
)
elif tool_name == "get_emissions_limit" and isinstance(data, dict) and "error" not in data:
building_type = data.get("building_type")
if building_type:
cache = dict(current_state.get("ll97_limits_cache", {}))
cache.update(updates.get("ll97_limits_cache", {}))
cache[building_type] = data
updates["ll97_limits_cache"] = cache
updates.setdefault("analysis_steps", list(current_state.get("analysis_steps", []))).append(
f"looked_up_ll97_limit_{building_type}"
)
elif tool_name == "compute_emissions_reduction" and isinstance(data, dict) and "error" not in data:
reductions = list(current_state.get("emissions_reductions", []))
reductions.extend(updates.get("emissions_reductions", []))
reductions.append(data)
updates["emissions_reductions"] = reductions
updates.setdefault("analysis_steps", list(current_state.get("analysis_steps", []))).append(
"computed_emissions_reduction"
)
return updates
# =============================================================================
# Graph Builder
# =============================================================================
def build_graph(_dataframes, api_key, namespace):
"""Construct the LangGraph-based agent with persistent code execution.
Uses a persistent Python namespace (passed via closure) instead of
a stateless pandas agent. Includes a planning node for complex queries.
The namespace dict is bound into the exec tools via closure, avoiding
st.session_state access from LangGraph's ToolNode thread pool.
"""
llm = ChatOpenAI(model="gpt-5.2", temperature=0, api_key=api_key)
# --- Assemble tools (hybrid: code interpreter + stateless calculators) ---
exec_tools = create_exec_tools(namespace)
all_tools = exec_tools + [get_emissions_limit, compute_emissions_reduction]
# Bind tools to the LLM once (reused in every agent_node call)
llm_with_tools = llm.bind_tools(all_tools)
# Pre-build the ToolNode executor
tool_executor = ToolNode(all_tools)
# --- Planner node ---
def planner_node(state: ChatbotState) -> dict:
"""Generate an execution plan for the current user turn.
Skips planning for simple queries using a lightweight heuristic.
"""
human_msgs = [m for m in state["messages"] if isinstance(m, HumanMessage)]
if not human_msgs:
return {"execution_plan": [], "plan_step_index": 0}
last_question = human_msgs[-1].content
# Fast-path: skip planner for simple queries
if _is_simple_query(last_question):
logger.info(f"[planner] Skipping plan for simple query: {last_question[:80]}")
return {"execution_plan": [], "plan_step_index": 0}
try:
response = llm.invoke([
SystemMessage(content=PLANNER_PROMPT),
HumanMessage(content=last_question),
])
plan_data = json.loads(response.content)
steps = plan_data.get("steps", [])
except (json.JSONDecodeError, AttributeError, Exception) as e:
logger.warning(f"[planner] Failed to generate plan: {e}")
steps = []
logger.info(f"[planner] Plan ({len(steps)} steps): {steps}")
return {"execution_plan": steps, "plan_step_index": 0}
# --- Agent node ---
def agent_node(state: ChatbotState) -> dict:
"""Call the LLM with system prompt, accumulated state context, and messages."""
state_context = _build_state_context(state, namespace=namespace)
system_messages = [SystemMessage(content=SYSTEM_PROMPT)]
if state_context:
system_messages.append(
SystemMessage(content=f"CURRENT ANALYSIS STATE:\n{state_context}")
)
all_messages = system_messages + list(state["messages"])
response = llm_with_tools.invoke(all_messages)
return {"messages": [response]}
# --- Tool node ---
def tool_node_fn(state: ChatbotState) -> dict:
"""Execute tool calls, then parse outputs to update intermediate state."""
result = tool_executor.invoke(state)
tool_messages = result.get("messages", [])
# Extract state updates from tool outputs
updates = _parse_tool_results_for_state(state, tool_messages)
updates["messages"] = tool_messages
# Advance plan step index
current_idx = state.get("plan_step_index", 0)
plan_len = len(state.get("execution_plan", []))
if current_idx < plan_len:
updates["plan_step_index"] = current_idx + 1
return updates
# Build the graph: planner -> agent <-> tools
graph = StateGraph(ChatbotState)
graph.add_node("planner", planner_node)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node_fn)
graph.add_edge(START, "planner")
graph.add_edge("planner", "agent")
graph.add_conditional_edges(
"agent",
tools_condition,
{"tools": "tools", "__end__": END},
)
graph.add_edge("tools", "agent")
compiled = graph.compile()
logger.info("Graph built successfully with tools: " + ", ".join(t.name for t in all_tools))
return compiled
# =============================================================================
# Streamlit UI
# =============================================================================
st.set_page_config(page_title="NYC Building Chatbot", page_icon="\U0001f3d7", layout="wide")
# --- Sidebar ---
with st.sidebar:
st.title("NYC Building Chatbot")
st.markdown("""
**Available Datasets:**
- **LL84** โ Benchmarking (energy use, emissions, EUI)
- **LL87** โ Audit/retrofit (Energy Conservation Measures)
- **MapPLUTO** โ Parcel-level tax lot data
**Capabilities:**
- Query building data by address, BBL, borough
- Look up LL97 emissions limits by building type
- Calculate emissions reductions from ECMs
- Find most common retrofit measures
""")
st.divider()
st.subheader("Example Queries")
examples = [
"What is the LL97 emissions limit for Office buildings?",
"What are the most common retrofit measures in Manhattan?",
"Show the average Site EUI for hotels in LL84",
"If a building saves 100,000 kWh of electricity, how much CO2e is reduced?",
]
for ex in examples:
if st.button(ex, use_container_width=True):
st.session_state.pending_query = ex
st.divider()
if st.button("Reset Interpreter", use_container_width=True,
help="Clear all user-created variables from the code interpreter"):
_reset_namespace()
st.toast("Interpreter reset โ DataFrames preserved, user variables cleared.")
st.caption("Powered by LangGraph + OpenAI")
# --- API Key ---
api_key = None
if "OPENAI_API_KEY" in os.environ:
api_key = os.environ["OPENAI_API_KEY"]
elif hasattr(st, "secrets") and "OPENAI_API_KEY" in st.secrets:
api_key = st.secrets["OPENAI_API_KEY"]
if not api_key:
with st.sidebar:
api_key = st.text_input("OpenAI API Key", type="password")
if not api_key:
st.info("Please enter your OpenAI API key in the sidebar to get started.")
st.stop()
# --- Load Data ---
dataframes = load_dataframes()
# --- Persistent Execution Namespace ---
if "exec_namespace" not in st.session_state:
import numpy as _np
import difflib as _difflib
st.session_state.exec_namespace = {
"pd": pd,
"json": json,
"re": re,
"np": _np,
"difflib": _difflib,
"map_pluto": dataframes["map_pluto"],
"LL87": dataframes["LL87"],
"LL84": dataframes["LL84"],
"df1": dataframes["map_pluto"],
"df2": dataframes["LL87"],
"df3": dataframes["LL84"],
"LL97_LIMITS": LL97_LIMITS,
"EMISSIONS_FACTORS": EMISSIONS_FACTORS,
"GFA_COLUMN_CANDIDATES": GFA_COLUMN_CANDIDATES,
"BOROUGH_MAP": BOROUGH_MAP,
}
# --- Build Graph (per-session, because exec tools close over the namespace) ---
if "graph" not in st.session_state:
with st.spinner("Initializing AI agent..."):
st.session_state.graph = build_graph(dataframes, api_key, st.session_state.exec_namespace)
# --- Authentication Gate ---
uid = get_hf_user()
if not uid:
st.title("NYC Building Chatbot")
st.info(
"This app requires a Hugging Face account. "
"Click the button below to sign in."
)
login_button()
st.stop()
# --- Show logged-in user in sidebar ---
with st.sidebar:
st.divider()
st.caption(f"Signed in as: **{uid}**")
if st.button("Sign out", use_container_width=True):
st.session_state.pop("hf_user", None)
st.rerun()
# --- Chat History ---
if "messages" not in st.session_state:
st.session_state.messages = []
# Display existing messages
for msg in st.session_state.messages:
with st.chat_message(msg["role"]):
st.markdown(msg["content"])
# --- Welcome screen (shown only before first message) ---
if not st.session_state.messages:
_logo_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logo.png")
_top_spacer, _logo_col, _bottom_spacer = st.columns([1, 1, 1])
with _logo_col:
st.image(_logo_path, use_container_width=True)
st.markdown(
""
"Ask me anything about NYC Building Databases"
"
"
""
"LL84 ยท LL87 ยท LL97 ยท MapPLUTO"
"
",
unsafe_allow_html=True,
)
# --- Handle input (either from chat box or sidebar button) ---
prompt = st.chat_input("Ask about NYC buildings, LL84, LL87, LL97, or MapPLUTO...")
if "pending_query" in st.session_state:
prompt = st.session_state.pending_query
del st.session_state.pending_query
if prompt:
# === QUOTA CHECK โ insert here ===
uid = get_hf_user()
allowed, remaining = check_and_increment_quota(uid)
if not allowed:
st.error(f"Usage limit reached: {MAX_RUNS_PER_USER} runs per user.")
st.stop()
if remaining <= 2:
st.warning(f"โ ๏ธ Only {remaining} run(s) left!")
else:
st.info(f"โ Runs remaining: {remaining}")
# === END QUOTA CHECK ===
# Show user message
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.markdown(prompt)
# Build LangChain messages list from session history
lc_messages = []
for msg in st.session_state.messages:
if msg["role"] == "user":
lc_messages.append(HumanMessage(content=msg["content"]))
else:
lc_messages.append(AIMessage(content=msg["content"]))
# Get graph response with live streaming log
with st.chat_message("assistant"):
answer = ""
initial_state = {
"messages": lc_messages,
"filtered_bbls": [],
"ll97_limits_cache": {},
"emissions_reductions": [],
"query_results": [],
"analysis_steps": [],
"execution_plan": [],
"plan_step_index": 0,
}
final_state = {}
all_messages = []
try:
with st.status("Analyzing building data...", expanded=True) as status:
for chunk in st.session_state.graph.stream(
initial_state,
config={"recursion_limit": 50},
stream_mode="updates",
):
for node_name, node_delta in chunk.items():
# Accumulate messages from each node
delta_msgs = node_delta.get("messages", [])
all_messages.extend(delta_msgs)
# Accumulate non-message state (last-write-wins per key)
for k, v in node_delta.items():
if k != "messages":
final_state[k] = v
# --- Live log updates ---
if node_name == "planner":
plan = node_delta.get("execution_plan", [])
if plan:
st.write("๐ **Plan:**")
for i, step in enumerate(plan, 1):
st.write(f" {i}. {step}")
elif node_name == "agent":
for msg in delta_msgs:
if isinstance(msg, AIMessage) and hasattr(msg, "tool_calls") and msg.tool_calls:
for tc in msg.tool_calls:
st.write(f"๐ง {_format_tool_call_label(tc)}")
elif node_name == "tools":
for msg in delta_msgs:
if isinstance(msg, ToolMessage):
summary = _summarize_tool_result(msg.name, msg.content)
st.write(f" โ {msg.name}: {summary}")
# Extract final answer from accumulated messages
ai_msgs = [m for m in all_messages if isinstance(m, AIMessage)]
answer = ai_msgs[-1].content if ai_msgs else "(no response)"
status.update(label="Analysis complete", state="complete", expanded=False)
# Store final analysis state for debugging visibility
st.session_state.last_analysis_state = {
"filtered_bbls": final_state.get("filtered_bbls", []),
"analysis_steps": final_state.get("analysis_steps", []),
"execution_plan": final_state.get("execution_plan", []),
}
except Exception as e:
answer = f"Error: {e}"
st.markdown(answer)
st.session_state.messages.append({"role": "assistant", "content": answer})