# FILE: app.py
# FINAL REFACTORED VERSION WITH ADVANCED UI/UX
import gradio as gr
import os
import logging
from typing import Dict, Any, List, Optional
from dotenv import load_dotenv
# --- Backend & Configuration Imports ---
load_dotenv()
import dspy
from config import (
initialize_dspy, RateLimiter,
STATE_STAGE, STATE_HISTORY, STATE_CURRENT_TITLE, STATE_GENERATED_TITLE,
STATE_EXPLAINER_PROMPT, STATE_TRANSITION_EXPLAINER_FLAG, STAGE_START,
STATE_IS_FIRST_TURN, DEFAULT_CHAT_TITLE,
)
from orchestrator import process_chat_message, initialize_orchestrator_modules
from resource_processor import process_uploaded_files
# --- Setup Logging ---
logging.basicConfig(level=logging.INFO, format='{levelname} {asctime} {name}: {message}', style='{')
logger = logging.getLogger(__name__)
# --- Application Mode & UI Configuration ---
APP_MODE = os.getenv("APP_MODE", "visitor").lower()
if APP_MODE not in ["admin", "visitor"]:
raise ValueError("APP_MODE environment variable must be 'admin' or 'visitor'")
UI_FEEDBACK_MAP = {
"PROCESSING_INPUT": "🧠 Thinking...",
"ANALYZING_RESOURCES_INITIAL": "📄 Analyzing your documents...",
"GENERATING_SYLLABUS": "✍️ Crafting syllabus...",
"EXPLAINER_RESPONSE": "💬 Tutor is responding...",
# NOTE: The UI will use these for more granular feedback in the future if needed.
# For now, we use a general "Thinking..." message for stability during backend processing.
}
# --- Helper & Core Functions ---
def initialize_session_state() -> Dict[str, Any]:
"""Creates a fresh session state dictionary, including the initial greeting."""
initial_history = [{'role': 'model', 'parts': [{'text': "Hello! I'm ready to help build a personalized syllabus. What topic are you interested in learning about?"}]}]
return {
STATE_STAGE: STAGE_START,
STATE_HISTORY: initial_history,
STATE_CURRENT_TITLE: DEFAULT_CHAT_TITLE,
STATE_IS_FIRST_TURN: True,
}
def start_session(api_key: str, api_tier: str):
"""
(Visitor Mode) Validates API, initializes backends, and transitions UI.
ADDED: Debugging print statements to trace the authentication flow.
"""
# --- START OF DEBUG BLOCK ---
print("\n--- [start_session] Function Entered ---")
# Use slicing to avoid printing the full secret key in logs
print(f"[start_session] Received API Key ending in: '...{api_key[-4:] if len(api_key) > 4 else api_key}'")
print(f"[start_session] Received API Tier: {api_tier}")
# --- END OF DEBUG BLOCK ---
# --- Case 1: Empty API Key ---
if not api_key.strip():
print("[start_session] DEBUG: Condition MET - API Key is empty or whitespace.")
error_msg = "API Key cannot be empty. Please provide a valid key.
"
yield (gr.update(visible=True), gr.update(visible=False), None, gr.update(value=error_msg, visible=True))
print("--- [start_session] END ---\n")
return
# --- Case 2: Authentication with DSPy Fails ---
print("[start_session] DEBUG: Attempting to call initialize_dspy...")
# Store the result in a variable so we can print it
auth_result = initialize_dspy(api_key=api_key)
print(f"[start_session] DEBUG: Result from initialize_dspy is: '{auth_result}' (Type: {type(auth_result)})")
if not isinstance(auth_result, list) :
#
print("[start_session] DEBUG: Condition MET - `if not auth_result` is TRUE. Authentication failed.")
error_msg = "Authentication failed. The provided API key is invalid or has expired. Please check the key and try again.
"
yield (gr.update(visible=True), gr.update(visible=False), None, gr.update(value=error_msg, visible=True))
print("--- [start_session] END ---\n")
return
print("[start_session] DEBUG: Condition SKIPPED - `if not auth_result` is FALSE. Assuming auth succeeded.")
# --- Case 3: Backend Module Initialization Fails ---
print("[start_session] DEBUG: Attempting to call initialize_orchestrator_modules...")
modules_result = initialize_orchestrator_modules()
print(f"[start_session] DEBUG: Result from initialize_orchestrator_modules is: {modules_result}")
if not modules_result:
print("[start_session] DEBUG: Condition MET - Backend module initialization failed.")
error_msg = "Failed to initialize backend AI modules after authentication. Please contact support.
"
yield (gr.update(visible=True), gr.update(visible=False), None, gr.update(value=error_msg, visible=True))
print("--- [start_session] END ---\n")
return
# --- Case 4: Success ---
print("[start_session] DEBUG: All checks passed. Proceeding to success case.")
max_calls = 7 if "Free" in api_tier else 1000
limiter = RateLimiter(max_calls=max_calls, time_period=60)
yield (
gr.update(visible=False), # Hide API view
gr.update(visible=True), # Show chat view
limiter, # Set the rate limiter state
gr.update(visible=False), # Hide the error message
)
print("--- [start_session] END: UI transitioned successfully. ---\n")
# FILE: app.py
# REPLACE the existing function with this new "consumer" version.
def handle_user_interaction(
user_message: str,
uploaded_files: Optional[List[Any]],
current_state: Dict[str, Any],
file_names_state: List[str],
limiter: Optional[RateLimiter],
modified_explainer_prompt: str
):
"""
Main generator function, refactored for cleaner history management and clarity
while consuming the stream of updates from the orchestrator.
"""
# --- 1. PREPARATION AND UI UPDATE (YIELD 1) ---
# Get the *current* UI-compatible history from the state for a clean start
# We will build on top of this for all UI updates
ui_history = []
is_first_turn = current_state.get(STATE_IS_FIRST_TURN, False)
for msg in current_state.get(STATE_HISTORY, []):
if msg.get('message_type') != "internal_resource_summary":
role = "assistant" if msg.get("role") == "model" else "user"
content = msg.get("content", "") or (msg.get("parts")[0].get("text", "") if msg.get("parts") else "")
ui_history.append({"role": role, "content": content})
# Add the new user message in the clean UI format
ui_history.append({"role": "user", "content": user_message})
# Add the initial placeholder for the bot's response/status
ui_history.append({"role": "assistant", "content": f"*{UI_FEEDBACK_MAP['PROCESSING_INPUT']}*"})
# Update the backend-formatted history in the state for the orchestrator
backend_history = current_state.get(STATE_HISTORY, [])
backend_history.append({'role': 'user', 'parts': [{'text': user_message}]})
current_state[STATE_HISTORY] = backend_history
# Handle file display logic (this part remains the same)
all_file_names = sorted(list(set(file_names_state + [f.name for f in uploaded_files or []])))
file_display_md = "#### Uploaded Resources:\n" + "\n".join([f"- `{os.path.basename(name)}`" for name in all_file_names]) if all_file_names else ""
file_uploader_visible = is_first_turn and not bool(uploaded_files)
file_display_visible = bool(all_file_names)
print("YIELD 1: Displaying user message and initial status.")
yield (
gr.update(value=ui_history), # Use the cleanly built UI history
gr.update(value=""),
current_state,
gr.update(visible=file_uploader_visible),
gr.update(interactive=False),
gr.update(), gr.update(),
gr.update(value=file_display_md, visible=file_display_visible),
all_file_names,
gr.update(visible=False),
gr.update(), gr.update()
)
# --- 2. BACKEND PROCESSING (CONSUMER LOOP) ---
orchestrator_kwargs = {
"user_message_text": user_message,
"current_session_state": current_state,
"modified_explainer_prompt": modified_explainer_prompt or None,
"uploaded_resource_data": process_uploaded_files(uploaded_files) if uploaded_files else None
}
current_state[STATE_IS_FIRST_TURN] = False
if limiter: limiter.wait_if_needed()
final_state = None
try:
# This loop consumes the `yield`ed updates from the backend orchestrator
for update_type, payload in process_chat_message(**orchestrator_kwargs):
if update_type == "status":
status_message = UI_FEEDBACK_MAP.get(payload, "Processing...")
# Update the last message in the chat (our status placeholder)
ui_history[-1]['content'] = f"*{status_message}*"
print(f"YIELD (Status Update): {status_message}")
# Yield a UI update with the new status message
yield (
gr.update(value=ui_history),
gr.update(), gr.update(), gr.update(), gr.update(),
gr.update(), gr.update(), gr.update(), gr.update(),
gr.update(), gr.update(), gr.update()
)
elif update_type == "final_result":
final_state = payload
break
except Exception as e:
logger.error(f"Orchestrator stream error: {e}", exc_info=True)
current_state[STATE_HISTORY].append({'role': 'model', 'parts': [{'text': "An error occurred during processing."}]})
final_state = current_state
# --- 3. FINAL UI UPDATE (YIELD 2) ---
if not final_state:
logger.error("Processing finished without a final state.")
current_state[STATE_HISTORY].append({'role': 'model', 'parts': [{'text': "A critical error occurred."}]})
final_state = current_state
# --- Final Robust Translator ---
final_ui_history = []
for message in final_state.get(STATE_HISTORY, []):
if message.get('message_type') == "internal_resource_summary": continue
role = "assistant" if message.get("role") == "model" else "user"
content = message.get('content', '') or (message.get("parts")[0].get("text", "") if message.get("parts") else "")
final_ui_history.append({"role": role, "content": content})
# Logic for Suggested Replies and Tab Switching
looks_good_btn_update = gr.update(visible=False)
if final_ui_history and "what are your thoughts?" in final_ui_history[-1]['content'].lower():
looks_good_btn_update = gr.update(visible=True)
explainer_accordion_update, explainer_box_update, app_tabs_update, documentation_update = (gr.update(),)*4
if final_state.get(STATE_TRANSITION_EXPLAINER_FLAG):
explainer_accordion_update = gr.update(visible=True, open=True)
explainer_prompt_value = final_state.get(STATE_EXPLAINER_PROMPT, "")
explainer_box_update = gr.update(value=explainer_prompt_value)
app_tabs_update = gr.update(selected="doc_tab")
full_documentation_text = f"""
# Generated Tutor Persona
*This document contains the complete system prompt, including the finalized syllabus, that defines the AI tutor's behavior, personality, and knowledge base. You can copy this for reference or use in other applications.*
***
{explainer_prompt_value}
"""
documentation_update = gr.update(value=full_documentation_text)
print("YIELD 2: Displaying final AI response.")
yield (
gr.update(value=final_ui_history),
gr.update(value=""),
final_state,
gr.update(visible=False, interactive=False),
gr.update(interactive=True),
explainer_accordion_update,
explainer_box_update,
gr.update(value=file_display_md, visible=file_display_visible),
all_file_names,
looks_good_btn_update,
app_tabs_update,
documentation_update,
)
# --- NEW AESTHETIC CSS ---
# --- NEW, LAYOUT-FOCUSED CSS ---
# --- FINAL, CORRECTED CSS ---
custom_css = """
/* --- Overall Page & Theme --- */
.gradio-container { background-color: #F9FAFB; }
.dark .gradio-container { background-color: #111827; }
/* --- THE KEY FIX: Make the chatbot tall and scrollable --- */
#chatbot {
/* Use vh (viewport height) to make the chatbot take up most of the screen */
min-height: 60vh !important;
background-color: white !important; /* Distinct white background */
border: 1px solid #E5E7EB !important;
border-radius: 12px !important;
}
.dark #chatbot {
background-color: #1F2937 !important; /* Distinct dark background for chatbot */
border-color: #374151 !important;
}
/* --- Input Bar Styling --- */
/* Add a top margin to the input row to create visual space from the chatbot */
.chat-input-row { margin-top: 1rem !important; }
/* Style the textbox itself for a clean look */
#user_input_textbox textarea {
background-color: #FFFFFF !important;
border: 1px solid #D1D5DB !important;
border-radius: 8px !important;
}
.dark #user_input_textbox textarea {
background-color: #374151 !important;
color: #F3F4F6 !important;
border-color: #4B5563 !important;
}
/* --- General styling from before (tabs, small buttons) --- */
/* (This part remains the same as the version you liked) */
#app-header { text-align: center; color: #4A5568; margin-bottom: 0.5rem !important; }
.dark #app-header { color: #D1D5DB; }
.sm-button { min-width: 0 !important; padding: 0.2rem 0.6rem !important; /* ...etc... */ }
#app-tabs .tab-nav button.selected { border-bottom-color: #4F46E5 !important; color: #4F46E5 !important; }
.dark #app-tabs .tab-nav button.selected { border-bottom-color: #A5B4FC !important; color: #A5B4FC !important; }
"""
# --- Gradio UI Definition ---
with gr.Blocks(theme=gr.themes.Soft(), css=custom_css, title="AI Syllabus Architect") as demo:
# --- State Management ---
session_state = gr.State(value=initialize_session_state)
rate_limiter_state = gr.State(value=None)
uploaded_file_names_state = gr.State([])
header_visibility_state = gr.State(value=True)
# --- TOP LEVEL HEADER (Collapsible) ---
with gr.Column(visible=True) as header_view:
gr.Markdown(
"""
# AI Tutor & Syllabus Planner
* Upload a resource or ask what you want to learn to generate a syllabus. Finalize the plan,
then tell the AI how you want to be taught. The tutor will then explain the topics based on your persona.
""",
elem_id="app-header"
)
toggle_header_btn = gr.Button("⬆️", elem_classes="sm-button")
# --- UI for Visitor Mode (API Key Entry) ---
with gr.Column(visible=(APP_MODE == 'visitor'), elem_id="api_setup_view") as api_setup_view:
gr.Markdown("### 🔑 Please enter your Google AI API Key below to begin. You can get your free key from [Google AI Studio](https://aistudio.google.com/app/apikey).")
api_key_input = gr.Textbox(label="Google API Key", placeholder="Enter your API key here...", type="password")
api_tier_radio = gr.Radio(["Free Tier (7 calls/min)", "Paid Tier (1000 calls/min)"], label="Select API Tier", value="Free Tier (7 calls/min)")
api_key_submit_button = gr.Button("Start Session", variant="primary")
api_error_output = gr.Markdown(visible=False)
# ===================================================================
# CORRECTED MAIN UI: Top-level tabs and proper nesting
# ===================================================================
with gr.Tabs(elem_id="app-tabs", visible=(APP_MODE == 'admin')) as app_tabs:
with gr.Tab("Chat Interface", id="chat_tab"):
with gr.Row(equal_height=False):
# --- MAIN CHAT COLUMN ---
with gr.Column(scale=7) as chat_column:
open_sidebar_btn = gr.Button("Show Resources & Persona ➡️", elem_classes="sm-button", visible=False)
initial_chat_ui = [{"role": "assistant", "content": "Hello! I'm ready to help build a personalized syllabus. What topic are you interested in learning about?"}]
chatbot = gr.Chatbot(
initial_chat_ui, elem_id="chatbot", type="messages",
show_label=False, render_markdown=True,
avatar_images=(None, "https://i.imgur.com/3pyR0Vf.png")
)
# This row contains the suggested replies
with gr.Row():
looks_good_btn = gr.Button("looks good", size="sm", visible=False)
# This row contains the main input textbox and send button
with gr.Row(elem_classes="chat-input-row"):
user_input_textbox = gr.Textbox(
elem_id="user_input_textbox", # ID for specific styling
scale=4, show_label=False,
placeholder="Type your message here...", container=False
)
send_button = gr.Button("Send", variant="primary", scale=1, min_width=120)
# --- TOOLS SIDEBAR COLUMN (Correctly nested inside the Chat Tab's Row) ---
with gr.Column(scale=3, visible=True) as tools_sidebar:
close_sidebar_btn = gr.Button("Close Sidebar ➡️", elem_classes="sm-button")
with gr.Accordion("🛠️ Session Content", open=True):
gr.Markdown("### 📄 Resources")
file_uploader = gr.File(label="Upload Learning Materials (First Turn Only)", file_count="multiple", file_types=[".pdf", ".docx", ".txt"], interactive=True)
file_display = gr.Markdown(visible=False)
with gr.Accordion("View/Edit AI Tutor Persona", open=False, visible=False) as explainer_prompt_accordion:
explainer_prompt_box = gr.Textbox(label="Tutor Persona System Prompt", lines=15, interactive=True, show_copy_button=True, info="You can view and modify the AI Tutor's persona here.")
# --- TAB 2: How to Use Guide ---
with gr.Tab("How to Use"):
gr.Markdown(
"""
### Video
How to Use AI Tutor - https://www.youtube.com/embed/p8uxJPLlQg4
Inspiration & Limitations- https://www.youtube.com/embed/gBsjCHJn1BA
---
**1. Start a Conversation:**
Begin by telling the AI what you want to learn. Be specific!
*Example: "I want to learn about building a RAG pipeline from scratch using Python."*
**2. Upload Resources (Optional):**
For a more tailored syllabus, upload relevant documents (`.pdf`, `.txt`, `.docx`) in your first message. The AI will use these as context.
**3. Negotiate the Syllabus:**
The AI will propose a syllabus. You can ask for changes, additions, or removals.
*Example: "Can you add a section on vector databases?"*
**4. Finalize and Learn:**
Once you're happy with the syllabus, tell the AI to finalize it (e.g., "This looks good, let's finalize it."). This will generate the tutor's persona and move to the learning phase.
**5. Interact with the Tutor:**
You can edit the tutor's persona at any time in the "Session Content" section to change its teaching style.
---
### Pro-User Guide: For a More Advanced Session
This is for users who want to architect a superior learning experience. Your prompts are the blueprints.
**1. On Architecting the Syllabus**
To get a truly calibrated result, you need to be precise.
* **Pinpoint Everything:** The key is to **pinpoint to every thing**. The more extra information you give, the better the AI can calibrate the syllabus to your exact needs.
* **Request Specific comparison :** Don't be afraid to demand more. For instance, you can **ask it for "the advanced version that contrasts one concept with another."**
* *Example:* `"For the data analysis module, I want the advanced version that contrasts the performance of Pandas with Polars for datasets over 10GB."*
**2. On Commanding the Tutor Persona**
This is the most important part. A well-defined persona yields a far more convincing and effective tutor.
* **The Core Principle:** When handling the Persona, **just don't ask the AI what persona you need.** A power user doesn't ask for a persona, they command one. You must *instruct* the AI on who it needs to be.
* **Advanced Techniques for Persona Prompts:**
* **First-Principles Thinking:** To force deep, foundational understanding.
* *Prompt Example:* `"Adopt the persona of a modern-day Feynman who thinks from first principles. When I ask about 'RAG pipelines,' start from the fundamental problem: 'How does a machine retrieve relevant information from a vast text library?' and build up from there."`
* **The Socratic Method:** To challenge your own thinking.
* *Prompt Example:* `"Become a Socratic tutor. Never give me a direct answer. Instead, relentlessly question my assumptions to lead me to the answer myself. If I say 'A vector database is faster,' your first response should be 'What do you mean by 'faster,' and compared to what?'"`
* **Key Figures from History:** To get a unique and powerful perspective.
* *Prompt Example:* `"Teach me chess as if you are Bobby Fischer preparing me for a world championship. Your tone should be intense, obsessive, and focused on total dominance. We will analyze unorthodox openings, drill methods to exploit opponent weaknesses, and master endgame precision. Every lesson is a step toward crushing the competition."`
**"After finalizing your perfected syllabus, you will be asked a persona question by the AI. Answer it based on that persona to begin your advanced learning session."**.
"""
)
# --- TAB 3: Generated Persona Documentation ---
with gr.Tab("Explainer Persona Prompt", id="doc_tab"):
# This is the new, more descriptive placeholder.
brief_documentation_display = gr.Markdown(
"""
## Tutor Persona Prompt
This Prompt will be generated after you finalize the syllabus in the chat.
It will contain the complete persona and system prompt that guides the AI tutor's teaching style, personality, and knowledge base.
---
*Waiting for syllabus to be finalized...*
"""
)
api_key_submit_button.click(fn=start_session, inputs=[api_key_input, api_tier_radio], outputs=[api_setup_view, app_tabs, rate_limiter_state, api_error_output])
def toggle_header(current_visibility_state):
new_visibility = not current_visibility_state
new_button_icon = "⬇️ Show Header" if not new_visibility else "⬆️ Hide Header"
return new_visibility, gr.update(visible=new_visibility), gr.update(value=new_button_icon)
toggle_header_btn.click(toggle_header, inputs=[header_visibility_state], outputs=[header_visibility_state, header_view, toggle_header_btn])
def close_sidebar():
return {tools_sidebar: gr.update(visible=False), open_sidebar_btn: gr.update(visible=True), chat_column: gr.update(scale=10)}
def open_sidebar():
return {tools_sidebar: gr.update(visible=True), open_sidebar_btn: gr.update(visible=False), chat_column: gr.update(scale=7)}
close_sidebar_btn.click(close_sidebar, outputs=[tools_sidebar, open_sidebar_btn, chat_column])
open_sidebar_btn.click(open_sidebar, outputs=[tools_sidebar, open_sidebar_btn, chat_column])
chat_inputs = [user_input_textbox, file_uploader, session_state, uploaded_file_names_state, rate_limiter_state, explainer_prompt_box]
chat_outputs = [chatbot, user_input_textbox, session_state, file_uploader, send_button, explainer_prompt_accordion, explainer_prompt_box, file_display, uploaded_file_names_state, looks_good_btn, app_tabs, brief_documentation_display]
user_input_textbox.submit(fn=handle_user_interaction, inputs=chat_inputs, outputs=chat_outputs)
send_button.click(fn=handle_user_interaction, inputs=chat_inputs, outputs=chat_outputs)
looks_good_btn.click(lambda: "looks good", inputs=[], outputs=[user_input_textbox]).then(
fn=handle_user_interaction,
inputs=chat_inputs,
outputs=chat_outputs
)
if __name__ == "__main__":
# This section remains unchanged
if APP_MODE == 'admin':
logger.info("Starting in ADMIN mode.")
if initialize_dspy() and initialize_orchestrator_modules():
logger.info("Admin mode ready.")
else:
logger.critical("FATAL: Could not initialize DSPy or orchestrator modules in Admin mode.")
else:
logger.info("Starting in VISITOR mode. Waiting for user to provide API key.")
demo.queue().launch(debug=True)