Create app.py

app.py

@@ -0,0 +1,332 @@
+# -*- coding: utf-8 -*-
+"""
+Created on Fri Feb  7 13:26:43 2025
+
+@author: Jacob Dearmon
+"""
+import os
+import time
+import csv
+import datetime
+import base64
+import gradio as gr
+import openai
+import io
+from PIL import Image
+from pinecone import Pinecone
+
+# ---------------------------------------------------
+# 1. Convert local SERMONS logo (JFIF) to PIL Image
+# ---------------------------------------------------
+def to_base64(path_to_img):
+    """Convert an image file to Base64 string."""
+    with open(path_to_img, "rb") as f:
+        encoded = base64.b64encode(f.read()).decode("utf-8")
+    return encoded
+
+def base64_to_image(base64_string):
+    """Convert Base64 string back to PIL Image."""
+    image_data = base64.b64decode(base64_string)
+    # Pillow can handle JFIF as it’s effectively a JPEG
+    return Image.open(io.BytesIO(image_data))
+
+# Update the path to your JFIF logo file here
+SERMONS_LOGO_B64 = to_base64(r"D:\Dearmon\Legacy\DP\Sermons\DP_logo.jfif")  
+SERMONS_LOGO_IMG = base64_to_image(SERMONS_LOGO_B64)
+
+# ---------------------------------------------------
+# 2. Configuration
+# ---------------------------------------------------
+
+openai.api_key = os.getenv("OPENAI_API_KEY")
+PINECONE_API_KEY = os.getenv("PINECONE_API_KEY")
+
+# From your screenshot: "Cloud: AWS | Region: us-east-1 | Dimension: 1536"
+PINECONE_ENV = "us-east-1"
+INDEX_NAME = "idx-sermons-1536"  # name from Pinecone console
+EMBED_DIMENSION = 1536          # matches your screenshot
+EMBED_MODEL = "text-embedding-ada-002"
+CHAT_MODEL = "gpt-4o"
+TOP_K = 20
+SIMILARITY_THRESHOLD = 0.4
+
+NEGATIVE_FEEDBACK_CSV = "negative_feedback.csv"
+NEUTRAL_FEEDBACK_CSV = "neutral_feedback.csv"
+SESSION_HISTORY_CSV = "session_history.csv"
+
+# ---------------------------------------------------
+# 2.5. Automatically Initialize Pinecone Index
+# ---------------------------------------------------
+def init_pinecone_index(index_name=INDEX_NAME, dimension=EMBED_DIMENSION):
+    """
+    Creates (or reuses) the Pinecone index with the given name and dimension.
+    Returns a Pinecone index object.
+    """
+    pc = Pinecone(api_key=PINECONE_API_KEY, environment=PINECONE_ENV)
+    existing_indexes = pc.list_indexes().names()  # get list of index names
+    if index_name not in existing_indexes:
+        print(f"[Info] Creating Pinecone index '{index_name}' in env '{PINECONE_ENV}'...")
+        pc.create_index(name=index_name, dimension=dimension)
+        time.sleep(5)  # short pause
+    else:
+        print(f"[Info] Reusing existing Pinecone index '{index_name}' in env '{PINECONE_ENV}'.")
+    return pc.Index(index_name)
+
+# Initialize Pinecone Index
+pc_index = init_pinecone_index()
+
+# ---------------------------------------------------
+# 3. Session Memory
+# ---------------------------------------------------
+session_history = [
+    {
+        "role": "system",
+        "content": "You are a helpful AI assistant specialized in sermons and biblical questions. Answer in a compassionate and loving tone, while recognizing the emotive content of the question - if any."
+    }
+]
+
+# ---------------------------------------------------
+# 4. Helper Functions
+# ---------------------------------------------------
+def embed_text(text: str):
+    """Get embeddings from OpenAI."""
+    try:
+        resp = openai.Embedding.create(model=EMBED_MODEL, input=[text])
+        return resp["data"][0]["embedding"]
+    except Exception as e:
+        print(f"[Error] Embedding failed: {e}")
+        return None
+
+def query_index(user_query: str, top_k=TOP_K):
+    """Query Pinecone for relevant matches based on 'user_query' embeddings."""
+    vector = embed_text(user_query)
+    if vector is None:
+        return []
+    try:
+        response = pc_index.query(vector=vector, top_k=top_k, include_metadata=True)
+        return response.matches
+    except Exception as e:
+        print(f"[Error] Pinecone query failed: {e}")
+        return []
+
+def build_rag_answer(user_query, matches):
+    """
+    Build a RAG-based answer using retrieved chunks as context for the LLM.
+    """
+    # Combine top matches into a context string
+    combined_context = "\n\n".join(
+        f"Chunk ID: {m.id}\n{m.metadata.get('text', '')}"
+        for m in matches
+    )
+
+    # Create a system message with retrieved context
+    context_system_message = {
+        "role": "system",
+        "content": (
+            "Relevant reference text from Pinecone:\n"
+            f"CONTEXT:\n{combined_context}\n\n"
+            "Answer the user's question using this context where helpful."
+        )
+    }
+
+    # Full conversation: existing history + new system context + user query
+    conversation = session_history + [
+        context_system_message,
+        {"role": "user", "content": user_query}
+    ]
+
+    try:
+        response = openai.ChatCompletion.create(
+            model=CHAT_MODEL,
+            messages=conversation,
+            temperature=0.2,
+            max_tokens=1750
+        )
+        final_answer = response["choices"][0]["message"]["content"].strip()
+    except Exception as e:
+        print(f"[Error] ChatCompletion failed: {e}")
+        final_answer = "Error generating RAG answer."
+
+    # Append the new assistant message to session history
+    session_history.append({"role": "assistant", "content": final_answer})
+    return final_answer
+
+def direct_llm_call(user_query):
+    """
+    If no relevant results or below threshold, do a direct LLM call with session history only.
+    """
+    conversation = session_history + [
+        {"role": "user", "content": user_query}
+    ]
+
+    try:
+        response = openai.ChatCompletion.create(
+            model=CHAT_MODEL,
+            messages=conversation,
+            temperature=0.2
+        )
+        final_answer = response["choices"][0]["message"]["content"].strip()
+    except Exception as e:
+        print(f"[Error] Direct LLM call failed: {e}")
+        final_answer = "Error generating direct LLM answer."
+
+    session_history.append({"role": "assistant", "content": final_answer})
+    return final_answer
+
+def query_rag(user_query: str) -> str:
+    """
+    Main pipeline:
+      1) Add user query to session history
+      2) Query Pinecone
+      3) If top match above threshold -> build RAG answer
+         else do direct call
+    """
+    user_query = user_query.strip()
+    if not user_query:
+        return "Please enter a valid query."
+
+    # Add user query to session memory
+    session_history.append({"role": "user", "content": user_query})
+
+    # Retrieve relevant context from Pinecone
+    matches = query_index(user_query, top_k=TOP_K)
+    if not matches:
+        # If no matches, do direct LLM call
+        return direct_llm_call(user_query)
+
+    top_score = matches[0].score or 0.0
+    if top_score >= SIMILARITY_THRESHOLD:
+        return build_rag_answer(user_query, matches)
+    else:
+        return direct_llm_call(user_query)
+
+# ---------------------------------------------------
+# 5. Feedback + Logging
+# ---------------------------------------------------
+def incorporate_feedback_into_pinecone(user_query, answer):
+    """
+    If thumbs-up, store Q&A as a new chunk in Pinecone.
+    """
+    text_chunk = f"User Query: {user_query}\nAI Answer: {answer}"
+    vector = embed_text(text_chunk)
+    if vector is None:
+        return
+    feedback_id = f"feedback_{int(time.time())}"
+    metadata = {"source": "feedback", "text": text_chunk}
+    try:
+        pc_index.upsert([
+            {"id": feedback_id, "values": vector, "metadata": metadata}
+        ])
+        print("[Info] User feedback upserted to Pinecone.")
+    except Exception as e:
+        print(f"[Error] Could not upsert feedback: {e}")
+
+def store_feedback_to_csv(user_query, answer, csv_path):
+    """
+    Log negative/neutral feedback in separate CSV.
+    """
+    file_exists = os.path.exists(csv_path)
+    with open(csv_path, mode="a", newline="", encoding="utf-8") as f:
+        fieldnames = ["timestamp", "query", "answer"]
+        writer = csv.DictWriter(f, fieldnames=fieldnames)
+        if not file_exists:
+            writer.writeheader()
+        writer.writerow({
+            "timestamp": datetime.datetime.now().isoformat(),
+            "query": user_query,
+            "answer": answer
+        })
+    print(f"[Info] Feedback logged to {csv_path}.")
+
+def store_session_history(user_query, answer, feedback):
+    """
+    Log (Q, A, feedback) to a single CSV: session_history.csv
+    """
+    file_exists = os.path.exists(SESSION_HISTORY_CSV)
+    with open(SESSION_HISTORY_CSV, mode="a", newline="", encoding="utf-8") as f:
+        fieldnames = ["timestamp", "user_query", "ai_answer", "feedback"]
+        writer = csv.DictWriter(f, fieldnames=fieldnames)
+        if not file_exists:
+            writer.writeheader()
+        writer.writerow({
+            "timestamp": datetime.datetime.now().isoformat(),
+            "user_query": user_query,
+            "ai_answer": answer,
+            "feedback": feedback
+        })
+    print(f"[Info] Session Q&A stored in {SESSION_HISTORY_CSV}.")
+
+def handle_feedback(user_query, answer, feedback_option):
+    """
+    Called when user selects feedback in Gradio UI.
+    """
+    if not user_query.strip() or not answer.strip():
+        return "No valid Q&A to provide feedback on."
+
+    if feedback_option == "👍":
+        incorporate_feedback_into_pinecone(user_query, answer)
+        store_session_history(user_query, answer, "positive")
+        return "👍 Your Q&A has been stored in Pinecone (and logged)."
+    elif feedback_option == "⚖️":
+        store_feedback_to_csv(user_query, answer, NEUTRAL_FEEDBACK_CSV)
+        store_session_history(user_query, answer, "neutral")
+        return "⚖️ Q&A logged to neutral_feedback.csv and session_history.csv."
+    else:  # "👎"
+        store_feedback_to_csv(user_query, answer, NEGATIVE_FEEDBACK_CSV)
+        store_session_history(user_query, answer, "negative")
+        return "👎 Q&A logged to negative_feedback.csv and session_history.csv."
+
+# ---------------------------------------------------
+# 6. Gradio Interface
+# ---------------------------------------------------
+def run_query(user_query):
+    return query_rag(user_query)
+
+with gr.Blocks() as demo:
+    # Row with two columns: (1) SERMONS jfif logo, (2) headings
+    with gr.Row():
+        with gr.Column(scale=1, min_width=100):
+            gr.Image(
+                value=SERMONS_LOGO_IMG,
+                label=None,
+                show_label=False,
+                width=80,
+                height=80
+            )
+        with gr.Column(scale=6):
+            gr.Markdown("## Derek Prince RAG Demo")
+            gr.Markdown("Ask questions about DP's sermons data, stored in Pinecone.\n"
+                        "Now with session memory!")
+
+    with gr.Column():
+        user_query = gr.Textbox(
+            label="Your Query",
+            lines=1,
+            placeholder="Ask about a sermon..."
+        )
+        get_answer_btn = gr.Button("Get Answer")
+
+    answer_output = gr.Textbox(label="AI Answer", lines=4)
+
+    feedback_radio = gr.Radio(
+        choices=["👍", "⚖️", "👎"],
+        value="⚖️",
+        label="Feedback"
+    )
+    feedback_btn = gr.Button("Submit Feedback")
+    feedback_result = gr.Label()
+
+    get_answer_btn.click(fn=run_query, inputs=[user_query], outputs=[answer_output])
+    feedback_btn.click(
+        fn=handle_feedback,
+        inputs=[user_query, answer_output, feedback_radio],
+        outputs=[feedback_result]
+    )
+
+if __name__ == "__main__":
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=True,
+        auth=("DP", "DP#1234")
+    )