main_two_column.py

import json
import os
import time
from functools import wraps

import gradio as gr
from openai import OpenAI
from opensearchpy import OpenSearch


def debounce(wait):
    def decorator(fn):
        last_call = 0

        @wraps(fn)
        def debounced(*args, **kwargs):
            nonlocal last_call
            now = time.time()
            if now - last_call >= wait:
                last_call = now
                return fn(*args, **kwargs)

        return debounced

    return decorator


USE_OPENSEARCH = os.getenv("USE_OPENSEARCH", False)
MODEL = os.getenv("MODEL", "gpt-4o-mini")

host = "localhost"
port = 9200
OPENSEARCH_ADMIN_PASSWORD = os.getenv("OPENSEARCH_ADMIN_PASSWORD", "yw7L5u9nLs3a")
auth = (
    "admin",
    OPENSEARCH_ADMIN_PASSWORD,
)  # For testing only. Don't store credentials in code.

# Create the client with SSL/TLS enabled, but hostname verification disabled.
os_client = OpenSearch(
    hosts=[{"host": host, "port": port}],
    http_compress=True,  # enables gzip compression for request bodies
    http_auth=auth,
    use_ssl=True,
    verify_certs=False,
    ssl_assert_hostname=False,
    ssl_show_warn=False,
)

all_props = json.load(open("all_properties.json"))
props_core = [
    {
        "property": prop["property"],
        "address": prop["address"],
        "school": prop["school"],
        "listPrice": prop["listPrice"],
    }
    for prop in all_props
]

client = OpenAI(
    # This is the default and can be omitted
    api_key=os.environ.get("OPENAI_API_KEY"),
)


chat_history = []
current_properties = {}

scoring_prompt = """
# System Prompt: Real Estate Assistant with JSON Output

You are an AI assistant specializing in real estate. Your role is to handle three types of inputs:
1. General questions about properties or real estate from users.
2. Comments about the user's criteria for properties.
3. A JSON list of properties from another system.

For all types of interactions, your output should always be in JSON format.

## Input Types:
1. General question: A string containing a question about real estate or properties.
2. Criteria update: A string containing the user's comments about their property preferences.
3. Property list: A JSON array of property objects.

## Property Format:
Properties will be in the following JSON format:

{
  "address": "Property Address",
  "description": "Property Description with Highlights"
}

## Your Tasks:
1. Determine the type of input received.
2. For general questions:
   - Provide a clear, concise answer to the question.
3. For criteria updates:
   - Update your understanding of the user's preferences.
   - Re-score all known properties based on the new criteria.
4. For property lists:
   - Update your list of known properties.
   - Score each property based on the current understanding of user preferences.
   - If no user preferences have been specified yet, use general desirability factors.

## Output Format:
Your output should always be a JSON object with a "type" field indicating the type of response, and additional fields based on the type:

For general questions:
{
  "type": "answer",
  "content": "Your answer to the question"
}

For criteria updates and property lists:
{
  "type": "scoring",
  "updatedCriteria": "Brief summary of the current criteria (include this field for criteria updates only)",
  "properties": [
    {
      "address": "Property Address",
      "score": 0-100,
      "explanation": "Brief explanation for the score with *highlighted* matching criteria"
    },
    // ... (more properties)
  ]
}

## Important Considerations:
- Ensure your entire output is valid JSON.
- For general questions, provide accurate and helpful information.
- For criteria updates and property lists:
  - Always re-score all known properties.
  - Focus on the unique features and highlights mentioned in the property descriptions.
  - Be objective and consistent in your scoring across all properties.
  - Consider both explicitly stated criteria and implied preferences.
  - If no user preferences are known, use general desirability factors.
  - In the explanation field, use double asterisks (**) to highlight words or phrases that directly match user criteria.
- Look through all user preferences across interactions.

## Example Inputs and Outputs:

1. General Question:
Input: "What factors should I consider when buying a vacation home?"

Output:
{
  "type": "answer",
  "content": "When buying a vacation home, consider: 1) Location (proximity to attractions and ease of access), 2) Rental potential if you plan to rent it out, 3) Maintenance costs and property management, 4) Local market conditions and potential for appreciation, 5) Climate and weather patterns, 6) Amenities and nearby services, 7) Security, especially if the home will be vacant for long periods, 8) Insurance costs, which may be higher for second homes, 9) Tax implications, and 10) Your long-term goals for the property."
}

2. Criteria Update:
Input: "I'm interested in properties with a modern kitchen and energy-efficient features."

Output:
{
  "type": "scoring",
  "updatedCriteria": "Preference for modern kitchens and energy-efficient features",
  "properties": [
    {
      "address": "123 Main St, Anytown, USA",
      "score": 85,
      "explanation": "This property has a recently updated kitchen, which likely includes **modern** features. While **energy efficiency** isn't explicitly mentioned, the recent updates suggest some improvements in this area."
    },
    {
      "address": "456 Elm St, Somewhere, USA",
      "score": 70,
      "explanation": "The description mentions *modern* appliances, which may include an **updated kitchen**, but doesn't specifically highlight **energy efficiency** features."
    }
  ]
}

3. Property List:
Input: 
[
  {
    "address": "789 Oak Rd, Newtown, USA",
    "description": "Eco-friendly home with solar panels and energy-efficient appliances. Gourmet kitchen with state-of-the-art equipment. Smart home features throughout."
  },
  {
    "address": "101 Pine Ave, Greenville, USA",
    "description": "Charming cottage with original features. Cozy kitchen with vintage appeal. Beautiful garden with mature trees."
  }
]

Output:
{
  "type": "scoring",
  "properties": [
    {
      "address": "789 Oak Rd, Newtown, USA",
      "score": 95,
      "explanation": "This property strongly matches the criteria with its **energy-efficient** features and **modern**, **gourmet kitchen**. The **state-of-the-art equipment** and smart home features are an added bonus."
    },
    {
      "address": "101 Pine Ave, Greenville, USA",
      "score": 60,
      "explanation": "While charming, this property doesn't align well with the preferences for **modern kitchens** and **energy efficiency**. The vintage kitchen and lack of mentioned energy features lower its score."
    }
  ]
}

Remember, your goal is to provide helpful information or accurate property scoring based on the input received and all known information, always in a valid JSON format. Track and apply all user preferences across the interactions, and highlight matching criteria in explanations using double asterisks.
"""


def format_house_markdown(house):
    return f"""
Price: {"${:,.0f}".format(house["listingPrice"])}

Size: {house["squareFootage"]} sqft

Bedrooms: {house["bedrooms"]}

Bathrooms: {house["bathrooms"]}

Score: {house["score"]}

Explanation: {house["explanation"]}

Description: {house["publicDescription"]}
"""


def get_search_results(beds, baths, min_size, max_size, min_price, max_price):
    if USE_OPENSEARCH:
        conditions = [
            {"range": {"listingPrice": {"gte": min_price, "lte": max_price}}},
            {"range": {"squareFootage": {"gte": min_size, "lte": max_size}}},
        ]
        if beds:
            conditions.append({"terms": {"bedrooms": beds}})
        if baths:
            conditions.append({"terms": {"bathrooms": baths}})

        result = os_client.search(
            index="sunnyvale_props",
            body={"query": {"bool": {"must": conditions}}},
        )
        houses = [hit["_source"] for hit in result["hits"]["hits"]]
        return houses
    else:
        with open("sunnyvale.json", "r") as f:
            data = json.load(f)

            def condition(house):
                if beds:
                    if house["bedrooms"] not in beds:
                        return False
                if baths:
                    if house["bathrooms"] not in baths:
                        return False
                if (
                    house["listingPrice"] < min_price
                    or house["listingPrice"] > max_price
                ):
                    return False
                if (
                    house["squareFootage"] < min_size
                    or house["squareFootage"] > max_size
                ):
                    return False
                return True

            data = list(filter(condition, data))
            return data


@debounce(1)
def update_filters(
    beds,
    baths,
    min_size,
    max_size,
    min_price,
    max_price,
    history,
):
    houses = get_search_results(
        beds=beds,
        baths=baths,
        min_size=min_size,
        max_size=max_size,
        min_price=min_price,
        max_price=max_price,
    )
    print(f"collected houses {houses}")

    global current_properties
    current_properties = dict([(house["address"], house) for house in houses])
    print(f"updating current properties to {current_properties} due to filters change")

    model_input = json.dumps(
        [
            {"address": house["address"], "description": house["publicDescription"]}
            for house in houses
        ]
    )

    output = get_model_output(
        model_input,
    )
    print("updating properties details due to filter change")
    update_property_details(output)
    output_text = get_response_to_user(output)
    history.append(("Updated filters, recalculating recommendations...", output_text))
    return history, str(time.time())


def update_property_details(output_str):
    global current_properties

    output = json.loads(output_str)
    for property in output["properties"]:
        address = property["address"]
        current_properties[address]["score"] = property["score"]
        current_properties[address]["explanation"] = property["explanation"]


test = """
[{
      "address": "1394 Lillian Avenue, Sunnyvale, California 94087",
      "description": "A must-see!! This Mediterranean-style home, masterfully built in 2019, offers a lifestyle centered around well-being and elegance. The Grand Room boasts high ceilings with wood beams, natural stone finishes, and extra-large windows that flood the space with natural light. This room seamlessly transitions to an expansive covered deck, adorned with a beautiful wood pergola and a cozy fire pit, perfect for enjoying California living at its finest."
    },
    {
      "address": "1487 Revelstoke Way, Sunnyvale, California 94087",
      "description": "Sunnyvale's Prime, Commuters Dream location in a QUIET neighborhood! Upgraded single story home w/10,000+ sq.ft. large lot w/ lots of potential to expand the house or make an ADU. Great CU schools-Nimitz Elem./ Cupertino Middle/ HOMESTEAD High. Upgraded kitchen w/refinished cabinets, granite counters, SS appliances & breakfast nook. Formal dining area or use as separate family room."
    }]
"""


def get_model_output(user_message):
    """
    Takes in message (could come from user, or ES search result), outputs LLM response
    """
    print(f"taking in message {user_message}")
    messages = (
        [
            {
                "role": "system",
                "content": scoring_prompt,
            },
        ]
        + chat_history
        + [
            {
                "role": "user",
                "content": user_message,
            }
        ]
    )
    print(f"sending {messages=}")
    chat_completion = client.chat.completions.create(
        messages=messages,
        model=MODEL,
    )
    content = chat_completion.choices[0].message.content
    print(f"model output: {content}")
    chat_history.append(
        {
            "role": "user",
            "content": user_message,
        }
    )
    chat_history.append({"role": "assistant", "content": content})
    return content


def get_response_to_user(message_str):
    """
    Parse model's message in JSON and return text to user
    """
    message = json.loads(message_str)
    if message["type"] == "scoring":
        if not message["properties"]:
            return "Thank for providing your criteria, please update filters to see recommendations"

        recommended_property = sorted(
            message["properties"], key=lambda item: item["score"], reverse=True
        )[0]
        return f"""
Based on the criteria, I recommended {recommended_property["address"]}. {recommended_property["explanation"]}
"""
    else:
        return message["content"]


def process_user_input(message, history):
    output_str = get_model_output(message)
    output = json.loads(output_str)
    if output["type"] == "scoring":
        print("updating properties details due to filter change")
        update_property_details(output_str)

    response_text = get_response_to_user(output_str)
    history.append((message, response_text))
    return "", history, str(time.time())


# Create Gradio Interface
with gr.Blocks() as demo:
    label_trigger = gr.Label("", visible=False)
    # Top Section with a Textbox and a Slider
    with gr.Column():
        beds = gr.CheckboxGroup([1, 2, 3, 4, 5], label="Beds")
        baths = gr.CheckboxGroup([1, 2, 3, 4, 5], label="Baths")
        min_size_slider = gr.Slider(
            label="Minimal Size Sqft",
            value=0,
            minimum=0,
            maximum=5000,
            step=100,
        )
        max_size_slider = gr.Slider(
            label="Maximal Size Sqft",
            value=10000,
            minimum=0,
            maximum=5000,
            step=100,
        )
        min_price_slider = gr.Slider(
            label="Minimal Price $",
            value=0,
            minimum=0,
            maximum=10000000,
            step=10000,
        )
        max_price_slider = gr.Slider(
            label="Maximal Price $",
            value=10000000,
            minimum=0,
            maximum=10000000,
            step=10000,
        )
        all_filters = [
            beds,
            baths,
            min_size_slider,
            max_size_slider,
            min_price_slider,
            max_price_slider,
        ]

    # Bottom Section with ChatInterface and Canvas
    with gr.Row():
        with gr.Column():
            chat_interface = gr.Chatbot(label="Chat Interface", elem_id="chat")
            msg = gr.Textbox()

        @gr.render(triggers=[label_trigger.change])
        def show_accordions():
            if not current_properties:
                return
            sorted_properties = sorted(
                list(current_properties.values()),
                key=lambda item: item["score"],
                reverse=True,
            )
            highest_score = sorted_properties[0]["score"]
            print(f"formatting current properties {sorted_properties=}")
            with gr.Column():
                for property in sorted_properties:
                    with gr.Accordion(
                        f'{property["address"]} {"${:,.0f}".format(property["listingPrice"])} {property["squareFootage"]} sqft {property["bedrooms"]}bed{property["bathrooms"]}bath',
                        open=True if property["score"] == highest_score else False,
                    ):
                        gr.Markdown(value=format_house_markdown(property))

    msg.submit(
        process_user_input, [msg, chat_interface], [msg, chat_interface, label_trigger]
    )

    for item in all_filters:
        if isinstance(item, gr.Slider):
            item.release(
                update_filters,
                all_filters + [chat_interface],
                [chat_interface, label_trigger],
            )
        else:
            item.change(
                update_filters,
                all_filters + [chat_interface],
                [chat_interface, label_trigger],
            )


# Launch the Gradio Interface
demo.launch()