Building an AI-Powered Insurance Q and A Assistant With RAG and Snowflake Cortex

pip install streamlit snowflake-connector-python python-dotenv

SNOWFLAKE_ACCOUNT=your_snowflake_account
SNOWFLAKE_USER=your_snowflake_user
SNOWFLAKE_PASSWORD=your_snowflake_password
SNOWFLAKE_WAREHOUSE=your_snowflake_warehouse
SNOWFLAKE_DATABASE=your_snowflake_database
SNOWFLAKE_SCHEMA=your_snowflake_schema

CREATE TABLE INSURANCE_DOCUMENTS (
    DOCUMENT_CONTENT STRING
);

INSERT INTO INSURANCE_DOCUMENTS (DOCUMENT_CONTENT) VALUES
('Auto insurance policy coverage: This policy covers damages caused by collisions, theft, and natural disasters. Liability coverage is also included.'),
('Claim filing guide for water damage: To file a water damage claim, provide photos, a description of the damage, and any repair estimates.'),
('Health insurance plan limitations: This plan does not cover cosmetic surgery or experimental treatments. Pre-existing conditions may have waiting periods.'),
('Life insurance beneficiary change process: To change your beneficiary, complete a beneficiary change form and submit it to the insurance company.'),
('Documents needed for car accident claims: You will need a police report, photos of the accident scene, and contact information for all parties involved.');

import streamlit as st
import snowflake.connector
import os
from dotenv import load_dotenv

load_dotenv()

# Snowflake Cortex Search Configuration
SNOWFLAKE_ACCOUNT = os.environ.get("SNOWFLAKE_ACCOUNT")
SNOWFLAKE_USER = os.environ.get("SNOWFLAKE_USER")
SNOWFLAKE_PASSWORD = os.environ.get("SNOWFLAKE_PASSWORD")
SNOWFLAKE_WAREHOUSE = os.environ.get("SNOWFLAKE_WAREHOUSE")
SNOWFLAKE_DATABASE = os.environ.get("SNOWFLAKE_DATABASE")
SNOWFLAKE_SCHEMA = os.environ.get("SNOWFLAKE_SCHEMA")
SNOWFLAKE_TABLE = "INSURANCE_DOCUMENTS"

def connect_to_snowflake():
    try:
        conn = snowflake.connector.connect(
            account=SNOWFLAKE_ACCOUNT,
            user=SNOWFLAKE_USER,
            password=SNOWFLAKE_PASSWORD,
            warehouse=SNOWFLAKE_WAREHOUSE,
            database=SNOWFLAKE_DATABASE,
            schema=SNOWFLAKE_SCHEMA,
        )
        return conn
    except Exception as e:
        st.error(f"Error connecting to Snowflake: {e}")
        return None

def snowflake_cortex_search(conn, query):
    try:
        cursor = conn.cursor()
        sql = f"""
            SELECT DOCUMENT_CONTENT
            FROM {SNOWFLAKE_TABLE}
            WHERE SEMANTIC_SIMILARITY(DOCUMENT_CONTENT, '{query}') > 0.7  # Adjust similarity threshold
            ORDER BY SEMANTIC_SIMILARITY(DOCUMENT_CONTENT, '{query}') DESC
            LIMIT 3;  # Limit the number of retrieved documents
        """
        cursor.execute(sql)
        results = cursor.fetchall()
        return [row[0] for row in results]
    except Exception as e:
        st.error(f"Error performing semantic search: {e}")
        return []

def generate_response(query, context):
    if not context:
        return "No relevant information found in the documents."

    prompt = f"""
    You are an insurance assistant. Use the following context to answer the user's question.

    Context:
    {context}

    Question: {query}

    Answer:
    """
    # Replace with your actual LLM API call (e.g., OpenAI, Vertex AI)
    # Example using a placeholder:
    response = f"Based on the provided information: [Simulated LLM Response] {prompt}"
    return response

# Streamlit App
st.title("Insurance Q&A Assistant")

query = st.text_input("Enter your insurance question:")

if st.button("Get Answer"):
    if query:
        conn = connect_to_snowflake()
        if conn:
            context_docs = snowflake_cortex_search(conn, query)
            conn.close()
            context = "\n\n".join(context_docs)
            response = generate_response(query, context)
            st.write(response)
        else:
            st.error("Could not connect to Snowflake.")