main_datafiniti.py

import json
import os
import time
from functools import wraps

import chromadb
import gradio as gr
from openai import OpenAI
from opensearchpy import OpenSearch
from sentence_transformers import SentenceTransformer

embedding_model = SentenceTransformer("all-MiniLM-L6-v2")


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


VECTOR_DB = os.getenv("VECTOR_DB", "chroma")
MODEL = os.getenv("MODEL", "gpt-4o-mini")

TOP_K = 10
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.

chroma_client = chromadb.Client()

# 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 = {}
user_criteria = []

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.
"""

categorize_prompt = """
You are an AI assistant specializing in real estate. Your role is to help users find properties that match their criteria and answer general questions about real estate. You must always respond in JSON format with two fields: "type" and "content". Follow these guidelines:

1. Determine if the user is asking a general question about real estate or specifying/updating their property search criteria.

2. Always structure your response as a JSON object with two fields:
   - "type": Indicate the type of user message. Use one of the following values:
     * "general_question": For general real estate queries or if clarification is needed
     * "search_criteria": For specifying or updating property search criteria
   - "content": Provide your response here based on the type of message

3. For general questions (type: "general_question"):
   - Provide clear, concise answers based on your real estate knowledge in the "content" field.
   - Offer additional relevant information if appropriate.
   - If the question is outside your expertise, politely say so and suggest consulting a real estate professional.
   - If the user's intent is unclear, use this type and ask for clarification in the "content" field.

4. For property search criteria (type: "search_criteria"):
   - Identify and summarize the specific criteria mentioned by the user (e.g., location, price range, number of bedrooms, etc.).
   - Include only the criteria explicitly stated by the user.
   - Do not ask questions or request additional information.
   - Present the summary in a clear, concise manner in the "content" field.

5. Always maintain a professional, helpful tone in your "content".

6. Remember previous interactions within the conversation to provide context-aware responses.

7. Be prepared to explain real estate terms or processes if the user seems unfamiliar with them.

8. If the user mentions specific financial concerns, suggest they consult with a financial advisor for personalized advice.

9. Respect privacy and do not ask for or store any personal identifying information.

Remember, your goal is to assist the user in their property search or answer their real estate questions to the best of your ability, while knowing when to defer to human experts for complex or specialized inquiries. Always output your response as a JSON object with "type" and "content" fields.

Examples:

Input: "What's a mortgage?"
Output:
{
  "type": "general_question",
  "content": "A mortgage is a loan used to purchase real estate, where the property serves as collateral. The borrower repays the loan with interest over a set period, typically 15 or 30 years."
}

Input: "I want a 2-bedroom condo in Miami under $300k"
Output:
{
  "type": "search_criteria",
  "content": "Updated search criteria: 1) Type: Condo, 2) 2-Bedrooms, 3) Location: Miami, 4) Max price: $300,000"
}
"""


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 VECTOR_DB == "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}})
        if user_criteria:
            embedding = embedding_model.encode(" ".join(user_criteria)).tolist()
            conditions.append(
                {"knn": {"publicDescriptionKnn": {"vector": embedding, "k": TOP_K}}}
            )

        result = os_client.search(
            index="datafiniti_props",
            body={"size": TOP_K, "query": {"bool": {"must": conditions}}},
        )
        houses = [hit["_source"] for hit in result["hits"]["hits"]]
        return houses
    elif VECTOR_DB == "chroma":
        collection = chroma_client.get_collection(name="datafiniti_properties")
        query_text = ""
        if user_criteria:
            query_text = " ".join(user_criteria)

        where_conditions = [
            {
                "listingPrice": {"$gte": min_price},
            },
            {
                "listingPrice": {"$lte": max_price},
            },
            {
                "squareFootage": {"$gte": min_size},
            },
            {
                "squareFootage": {"$lte": max_size},
            },
        ]
        if beds:
            where_conditions.append({"bedrooms": {"$in": beds}})
        if baths:
            where_conditions.append({"bathrooms": {"$in": baths}})

        results = collection.query(
            query_texts=[query_text],
            where={"$and": where_conditions},
            n_results=TOP_K,
        )
        all_values = []
        for metadata, document in zip(results["metadatas"][0], results["documents"][0]):
            metadata["publicDescription"] = document
            all_values.append(metadata)
        return all_values
    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


def update_property_list_and_get_response(
    beds, baths, min_size, max_size, min_price, max_price
):
    houses = get_search_results(
        beds=beds,
        baths=baths,
        min_size=min_size,
        max_size=max_size,
        min_price=min_price,
        max_price=max_price,
    )
    global current_properties
    current_properties = dict([(house["address"], house) for house in houses])
    print(
        f"updating current properties to {[property['address'] for property in current_properties.values()]} due to filters change"
    )

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

    output = get_scoring_output(
        model_input,
    )
    print("updating properties details due to filter change")
    update_property_details(output)
    output_text = get_scoring_response_to_user(output)
    return output_text


# @debounce(1)
def update_filters(
    beds,
    baths,
    min_size,
    max_size,
    min_price,
    max_price,
    history,
):
    """
    Triggered by filters and criteria change, update the list of properties, and re-score
    """
    output_text = update_property_list_and_get_response(
        beds, baths, min_size, max_size, min_price, max_price
    )
    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"]


def get_scoring_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,
            },
        ]
        + [
            {
                "role": "user",
                "content": criteria_update,
            }
            for criteria_update in user_criteria
        ]
        + [
            {
                "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_scoring_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 categorize_user_input(message):
    """
    Categorize user input as general question or search criteria
    """
    messages = (
        [
            {
                "role": "system",
                "content": categorize_prompt,
            }
        ]
        + chat_history
        + [
            {
                "role": "user",
                "content": message,
            },
        ]
    )
    chat_completion = client.chat.completions.create(
        messages=messages,
        model=MODEL,
    )
    content = chat_completion.choices[0].message.content
    # no need to store this in chat history
    return content


def process_user_input(
    message,
    history,
    beds,
    baths,
    min_size,
    max_size,
    min_price,
    max_price,
):
    categorization_str = categorize_user_input(message)
    categorization = json.loads(categorization_str)
    if categorization["type"] == "general_question":
        response_text = categorization["content"]
    else:
        global user_criteria
        user_criteria.append(message)
        print(f"appending user criteria {user_criteria}")
        response_text = update_property_list_and_get_response(
            beds, baths, min_size, max_size, min_price, max_price
        )
    history.append((message, response_text))
    return "", history, str(time.time())


# Create Gradio Interface
with gr.Blocks() as demo:
    label_trigger = gr.Label("", visible=False)
    version = gr.Label("Build 09032024")
    # 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"]
            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] + all_filters,
        [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],
            )


def load_data_to_chroma():
    import chromadb

    chroma_client = chromadb.Client()
    try:
        collection = chroma_client.create_collection(name="datafiniti_properties")
    except:
        print("data is already added, exit")
        return

    ids = []
    documents = []
    metadatas = []
    with open("datafiniti_properties_sunnyvale_400.json", "r") as f:
        for line in f:
            property = json.loads(line)
            try:
                print(f'indexing {property["address"]}')
                bathrooms = int(property["numBathroom"])
                beds = property["numBedroom"]
                price = property["mostRecentPriceAmount"]
                size = property["floorSizeValue"]
                address = ", ".join(
                    [
                        property["address"],
                        property["city"],
                        property["province"],
                        property["postalCode"][:5],
                    ]
                )

                descriptions = property["descriptions"]
                descriptions = sorted(
                    descriptions, key=lambda x: x["dateSeen"], reverse=True
                )
                description = descriptions[0]["value"]

                metadata = {
                    "bathrooms": bathrooms,
                    "bedrooms": beds,
                    "listingPrice": price,
                    "squareFootage": size,
                    "address": address,
                }

                ids.append(address)
                documents.append(description)
                metadatas.append(metadata)
            except Exception as e:
                print(e)
                pass
    collection.add(ids=ids, documents=documents, metadatas=metadatas)


load_data_to_chroma()
# Launch the Gradio Interface
demo.launch()