""" Project Echo - AI-Powered Qualitative Research Assistant Production-grade survey generation, translation, and analysis platform """ import gradio as gr import json import os import traceback from typing import Dict, List, Optional from llm_backend import LLMBackend, LLMProvider from survey_generator import SurveyGenerator from survey_translator import SurveyTranslator from data_analyzer import DataAnalyzer from export_utils import (save_json_file, survey_to_csv, analysis_to_markdown_file, conversation_to_transcript, conversation_to_json, conversation_to_csv, flow_to_markdown) from conversation_flow import ConversationFlow, ConversationNode, create_example_flow from conversation_session import ConversationSession, SessionManager from conversation_moderator import ConversationModerator from conversation_analytics import ConversationAnalytics # Global state for current survey current_survey = None current_responses = [] # Global state for conversational research current_flow = None session_manager = SessionManager() current_session = None saved_flows = {} def initialize_backend(): """Initialize LLM backend based on environment""" try: # Debug: Print all environment variables related to LLM print("=== LLM Backend Initialization ===") print(f"HF_TOKEN: {'SET' if os.getenv('HF_TOKEN') else 'NOT SET'}") print(f"HUGGINGFACE_API_KEY: {'SET' if os.getenv('HUGGINGFACE_API_KEY') else 'NOT SET'}") print(f"OPENAI_API_KEY: {'SET' if os.getenv('OPENAI_API_KEY') else 'NOT SET'}") print(f"ANTHROPIC_API_KEY: {'SET' if os.getenv('ANTHROPIC_API_KEY') else 'NOT SET'}") print(f"LLM_PROVIDER: {os.getenv('LLM_PROVIDER', 'NOT SET')}") # Check for explicit provider setting provider_env = os.getenv("LLM_PROVIDER", "").lower() # Priority 1: Explicitly set provider if provider_env == "openai" and os.getenv("OPENAI_API_KEY"): print("Using OpenAI (explicit)") return LLMBackend(provider=LLMProvider.OPENAI) elif provider_env == "anthropic" and os.getenv("ANTHROPIC_API_KEY"): print("Using Anthropic (explicit)") return LLMBackend(provider=LLMProvider.ANTHROPIC) elif provider_env == "huggingface" and (os.getenv("HUGGINGFACE_API_KEY") or os.getenv("HF_TOKEN")): api_key = os.getenv("HUGGINGFACE_API_KEY") or os.getenv("HF_TOKEN") print("Using HuggingFace (explicit)") return LLMBackend(provider=LLMProvider.HUGGINGFACE, api_key=api_key) elif provider_env == "lm_studio": print("Using LM Studio (explicit)") return LLMBackend(provider=LLMProvider.LM_STUDIO) # Priority 2: Auto-detect based on available credentials # HF_TOKEN is automatically available in HF Spaces, so check it first hf_token = os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACE_API_KEY") if hf_token: print(f"Auto-detected HuggingFace credentials, using HF Inference API") print(f"Token preview: {hf_token[:10]}...") return LLMBackend(provider=LLMProvider.HUGGINGFACE, api_key=hf_token) elif os.getenv("OPENAI_API_KEY"): print(f"Auto-detected OpenAI credentials") return LLMBackend(provider=LLMProvider.OPENAI) elif os.getenv("ANTHROPIC_API_KEY"): print(f"Auto-detected Anthropic credentials") return LLMBackend(provider=LLMProvider.ANTHROPIC) else: # No credentials found - return None to show error in UI print("="*60) print("WARNING: No LLM provider credentials found!") print("="*60) print("For HuggingFace Spaces:") print(" - HF_TOKEN should be automatically available") print(" - Make sure your Space is PUBLIC") print(" - Or add HUGGINGFACE_API_KEY in Settings") print("") print("For other providers, set one of:") print(" - OPENAI_API_KEY") print(" - ANTHROPIC_API_KEY") print(" - HUGGINGFACE_API_KEY") print("="*60) return None except Exception as e: print(f"Error during backend initialization: {e}") import traceback traceback.print_exc() return None # Initialize components llm_backend = initialize_backend() # Only initialize if backend is available if llm_backend: survey_gen = SurveyGenerator(llm_backend) survey_trans = SurveyTranslator(llm_backend) data_analyzer = DataAnalyzer(llm_backend) print(f"✓ Project Echo initialized with {llm_backend.provider.value} provider") else: survey_gen = None survey_trans = None data_analyzer = None print("✗ Project Echo initialization incomplete - no LLM credentials found") # =========================== # Survey Generation Functions # =========================== def generate_survey_from_outline(outline: str, survey_type: str, num_questions: int, audience: str): """Generate survey from user outline""" global current_survey # Check if backend is initialized if not survey_gen: return ( "❌ LLM backend not configured. Please set up API credentials:\n" "- For HuggingFace Spaces: HF_TOKEN is auto-available\n" "- For OpenAI: Set OPENAI_API_KEY\n" "- For Anthropic: Set ANTHROPIC_API_KEY\n" "- For HuggingFace: Set HUGGINGFACE_API_KEY", "", None ) if not outline or not outline.strip(): return "❌ Please provide an outline or topic description.", "", None # Validate inputs if num_questions < 1 or num_questions > 50: return "❌ Number of questions must be between 1 and 50.", "", None try: # Generate survey survey_data = survey_gen.generate_survey( outline=outline, survey_type=survey_type.lower(), num_questions=num_questions, target_audience=audience ) current_survey = survey_data # Format for display display_text = format_survey_display(survey_data) # Save to file for download filepath = save_json_file(survey_data, "survey") return ( f"✅ Survey generated successfully! Contains {len(survey_data.get('questions', []))} questions.", display_text, filepath ) except Exception as e: error_msg = f"❌ Error generating survey: {str(e)}" print(f"Survey generation error: {traceback.format_exc()}") return error_msg, "", None def format_survey_display(survey_data: Dict) -> str: """Format survey data for readable display""" output = f"# {survey_data.get('title', 'Survey')}\n\n" output += f"## Introduction\n{survey_data.get('introduction', '')}\n\n" output += "## Questions\n\n" for i, q in enumerate(survey_data.get('questions', []), 1): output += f"**{i}. {q.get('question_text', '')}**\n" output += f" - Type: {q.get('question_type', 'N/A')}\n" if q.get('options'): output += " - Options:\n" for opt in q['options']: output += f" - {opt}\n" if q.get('help_text'): output += f" - Help: {q['help_text']}\n" output += f" - Required: {'Yes' if q.get('required', False) else 'No'}\n\n" output += f"## Closing\n{survey_data.get('closing', '')}\n" return output # =========================== # Translation Functions # =========================== def translate_current_survey(target_languages: List[str]): """Translate the current survey to selected languages""" global current_survey # Check if backend is initialized if not survey_trans: return ( "❌ LLM backend not configured. Please set up API credentials in Settings.", "", None ) if not current_survey: return "❌ Please generate or upload a survey first.", "", None if not target_languages: return "❌ Please select at least one target language.", "", None try: # Translate to all selected languages translations = {} status_messages = [] success_count = 0 for lang_code in target_languages: try: translated = survey_trans.translate_survey(current_survey, lang_code) translations[lang_code] = translated lang_name = survey_trans._resolve_language(lang_code) status_messages.append(f"✅ Translated to {lang_name}") success_count += 1 except Exception as e: lang_name = survey_trans._resolve_language(lang_code) status_messages.append(f"❌ Failed to translate to {lang_name}: {str(e)}") print(f"Translation error for {lang_code}: {traceback.format_exc()}") if success_count == 0: return "❌ All translations failed. Please check your LLM configuration.", "", None # Format translations for display display_text = "" for lang_code, trans_survey in translations.items(): if "error" not in trans_survey: lang_name = survey_trans._resolve_language(lang_code) display_text += f"\n{'='*50}\n" display_text += f"TRANSLATION: {lang_name.upper()}\n" display_text += f"{'='*50}\n\n" display_text += format_survey_display(trans_survey) # Save to file for download filepath = save_json_file(translations, "translations") status = "\n".join(status_messages) return status, display_text, filepath except Exception as e: error_msg = f"❌ Error during translation: {str(e)}" print(f"Translation error: {traceback.format_exc()}") return error_msg, "", None def get_language_choices(): """Get language choices for dropdown""" # Get languages directly from SurveyTranslator class (static list) from survey_translator import SurveyTranslator langs = SurveyTranslator.SUPPORTED_LANGUAGES return [f"{code} - {name}" for code, name in langs.items()] # =========================== # Data Analysis Functions # =========================== def analyze_survey_data(responses_json: str, questions_json: str = None): """Analyze survey responses""" # Check if backend is initialized if not data_analyzer: return ( "❌ LLM backend not configured. Please set up API credentials in Settings.", "", None ) if not responses_json or not responses_json.strip(): return "❌ Please provide survey responses in JSON format.", "", None try: # Parse responses responses = json.loads(responses_json) questions = json.loads(questions_json) if questions_json and questions_json.strip() else None if not isinstance(responses, list): return "❌ Responses must be a JSON array.", "", None if len(responses) == 0: return "❌ No responses to analyze.", "", None # Validate questions if provided if questions and not isinstance(questions, list): return "❌ Questions must be a JSON array.", "", None # Run analysis analysis_results = data_analyzer.analyze_responses(responses, questions) if "error" in analysis_results: return f"❌ Analysis error: {analysis_results['error']}", "", None # Generate report report_md = data_analyzer.generate_report(analysis_results, format="markdown") # Save both JSON and Markdown json_filepath = save_json_file(analysis_results, "analysis_results") md_filepath = analysis_to_markdown_file(report_md, "analysis_report") status_msg = f"✅ Analysis complete! Analyzed {len(responses)} responses." if questions: status_msg += f" Considered {len(questions)} questions." return status_msg, report_md, json_filepath except json.JSONDecodeError as e: return f"❌ Invalid JSON format: {str(e)}", "", None except Exception as e: error_msg = f"❌ Error during analysis: {str(e)}" print(f"Analysis error: {traceback.format_exc()}") return error_msg, "", None def load_example_responses(): """Load example responses for demonstration""" example = [ { "q1": "The medication helped reduce my symptoms significantly within the first week.", "q2": "I experienced some mild side effects like drowsiness in the beginning.", "q3": "Overall, I'm satisfied with the treatment and would recommend it to others." }, { "q1": "I didn't notice much improvement in my condition after taking the medication.", "q2": "The side effects were quite severe and made it difficult to continue.", "q3": "I had to stop taking it after two weeks due to adverse reactions." }, { "q1": "The medication worked well but took about 3-4 weeks to show results.", "q2": "No major side effects, just some occasional nausea.", "q3": "It's been effective for managing my symptoms on a daily basis." } ] return json.dumps(example, indent=2) # =========================== # Conversational Research Handlers # =========================== def create_new_flow(flow_name: str, flow_description: str): """Create a new conversation flow with AI-generated initial structure""" global current_flow, saved_flows, llm_backend if not flow_name or not flow_name.strip(): return "❌ Please provide a flow name.", "", None if not flow_description or not flow_description.strip(): return "❌ Please provide a description of what you want to discuss in this flow.", "", None if not llm_backend: return "❌ LLM backend not configured. Cannot generate flow.", "", None try: # Create empty flow flow = ConversationFlow(name=flow_name, description=flow_description) # Generate initial conversation structure using AI success, message = flow.generate_flow_with_ai(llm_backend, num_questions=5) if not success: return f"⚠️ Flow created but generation failed: {message}", display_flow(flow), None current_flow = flow saved_flows[flow.id] = flow status_msg = f"✅ Flow '{flow_name}' created with {len(flow.nodes)} conversation steps!" return ( status_msg, display_flow(flow), flow.id ) except Exception as e: error_msg = f"❌ Error creating flow: {str(e)}" print(f"Flow creation error: {traceback.format_exc()}") return error_msg, "", None def regenerate_flow_content(flow_id: str): """Regenerate the conversation flow nodes using AI""" global saved_flows, current_flow, llm_backend if not flow_id: return "❌ No flow selected.", "" flow = saved_flows.get(flow_id) if not flow: return "❌ Flow not found.", "" if not llm_backend: return "❌ LLM backend not configured.", "" try: # Clear existing nodes flow.nodes = [] # Regenerate with AI success, message = flow.generate_flow_with_ai(llm_backend, num_questions=5) if not success: return f"⚠️ Regeneration failed: {message}", "" current_flow = flow return ( f"✅ Flow regenerated with {len(flow.nodes)} new steps!", display_flow(flow) ) except Exception as e: return f"❌ Error regenerating flow: {str(e)}", "" def load_example_flow(): """Load an example conversation flow""" global current_flow, saved_flows flow = create_example_flow() current_flow = flow saved_flows[flow.id] = flow return ( f"✅ Example flow loaded: {flow.name}", display_flow(flow), flow.id ) def add_flow_node(flow_id: str, node_content: str, node_type: str): """Add a node to the current flow""" global current_flow, saved_flows if not flow_id: return "❌ No flow selected.", "" flow = saved_flows.get(flow_id) if not flow: return "❌ Flow not found.", "" if not node_content or not node_content.strip(): return "❌ Please provide content for the node.", "" try: node = ConversationNode(content=node_content, node_type=node_type.lower()) # Link to previous node if exists if flow.nodes: last_node = flow.nodes[-1] last_node.next = node.id flow.add_node(node) current_flow = flow return ( f"✅ Node added successfully! Total nodes: {len(flow.nodes)}", display_flow(flow) ) except Exception as e: return f"❌ Error adding node: {str(e)}", "" def display_flow(flow: ConversationFlow) -> str: """Display flow as markdown""" if not flow or not flow.nodes: return "No flow to display" output = f"# {flow.name}\n\n" output += f"**Description:** {flow.description}\n\n" output += f"**Total Steps:** {len(flow.nodes)}\n\n" output += "---\n\n" for i, node in enumerate(flow.nodes, 1): output += f"### Step {i}: {node.type.capitalize()}\n\n" output += f"{node.content}\n\n" return output def save_current_flow(flow_id: str): """Save the current flow to file""" if not flow_id: return "❌ No flow selected.", None flow = saved_flows.get(flow_id) if not flow: return "❌ Flow not found.", None try: filepath = save_json_file(flow.to_dict(), "conversation_flow") return f"✅ Flow saved to {filepath}", filepath except Exception as e: return f"❌ Error saving flow: {str(e)}", None def start_conversation_session(flow_id: str): """Start a new conversation session""" global current_session, session_manager if not flow_id: return [], "❌ Please select a flow first." flow = saved_flows.get(flow_id) if not flow: return [], "❌ Flow not found." if not llm_backend: return [], "❌ LLM backend not initialized." try: # Create session session = session_manager.create_session(flow_id=flow.id, flow_name=flow.name) current_session = session # Create moderator moderator = ConversationModerator(llm_backend, flow) # Start conversation opening_message = moderator.start_conversation(session) # Return chat history in Gradio format return [[None, opening_message]], f"✅ Conversation started! Session ID: {session.id}" except Exception as e: return [], f"❌ Error starting conversation: {str(e)}" def chat_with_moderator(user_message: str, history: List): """Handle chat messages with the AI moderator""" global current_session if not current_session: return history, "❌ No active session. Please start a conversation first." if not llm_backend: return history, "❌ LLM backend not initialized." if not user_message or not user_message.strip(): return history, "❌ Please enter a message." try: # Get the flow flow = saved_flows.get(current_session.flow_id) if not flow: return history, "❌ Flow not found." # Create moderator moderator = ConversationModerator(llm_backend, flow) # Process user response ai_response = moderator.process_user_response(current_session, user_message) # Update history history.append([user_message, ai_response]) status = f"Session: {current_session.id} | Turns: {current_session.get_turn_count()}" if current_session.status == "completed": status += " | ✅ Conversation completed" return history, status except Exception as e: return history, f"❌ Error: {str(e)}" def export_conversation(): """Export the current conversation""" global current_session if not current_session: return "❌ No active session to export.", None try: filepath = conversation_to_transcript(current_session) return f"✅ Conversation exported to {filepath}", filepath except Exception as e: return f"❌ Error exporting conversation: {str(e)}", None def generate_conversation_summary(): """Generate AI summary of the current conversation""" global current_session if not current_session: return "❌ No active session. Start a conversation first.", "" if not llm_backend: return "❌ LLM backend not initialized.", "" if current_session.get_turn_count() < 3: return "❌ Not enough conversation data. Have at least 2-3 exchanges first.", "" try: # Get the flow flow = saved_flows.get(current_session.flow_id) if not flow: return "❌ Flow not found.", "" # Create moderator and generate summary moderator = ConversationModerator(llm_backend, flow) summary = moderator.generate_summary(current_session) # Format summary with stats stats = current_session.get_summary_stats() formatted_summary = f"""## Conversation Summary **Session Details:** - Session ID: {current_session.id} - Flow: {current_session.flow_name} - Total Turns: {stats['total_turns']} ({stats['user_turns']} user, {stats['ai_turns']} AI) - Duration: {stats['duration_minutes']} minutes - Status: {stats['status']} --- {summary} --- *Summary generated by AI. Review for accuracy.* """ return "✅ Summary generated successfully!", formatted_summary except Exception as e: return f"❌ Error generating summary: {str(e)}", "" def update_probing_threshold(threshold: int): """Update the probing threshold for follow-up questions""" # This will be used when creating new moderators return f"✅ Probing threshold set to every {threshold} responses" def get_conversation_metrics(): """Get real-time conversation metrics""" global current_session if not current_session: return """**No Active Session** Start a conversation to see metrics.""" stats = current_session.get_summary_stats() user_turns = [t for t in current_session.conversation_history if t.role == "user"] # Calculate follow-up count (AI turns that aren't linked to nodes) follow_ups = len([t for t in current_session.conversation_history if t.role == "ai" and not t.node_id]) scripted = stats['ai_turns'] - follow_ups metrics_md = f"""## 📊 Live Conversation Metrics **Engagement:** - Total Exchanges: {stats['user_turns']} - User Responses: {stats['user_turns']} - AI Questions: {stats['ai_turns']} **Question Mix:** - Scripted Questions: {scripted} - Dynamic Follow-ups: {follow_ups} - Follow-up Rate: {(follow_ups / max(stats['ai_turns'], 1) * 100):.1f}% **Quality Indicators:** - Avg Response Length: {stats['avg_user_response_length']:.0f} characters - Duration: {stats['duration_minutes']} min - Status: {stats['status'].upper()} **Session Info:** - Session ID: `{current_session.id[:8]}...` - Flow: {current_session.flow_name} """ return metrics_md def analyze_multiple_sessions(uploaded_files): """Analyze multiple conversation sessions""" if not uploaded_files: return "❌ Please upload at least one conversation JSON file.", "", None if not llm_backend: return "⚠️ LLM backend not configured. Basic analysis only (no AI insights).", "", None try: # Load session data from uploaded files session_data_list = [] for file in uploaded_files: with open(file.name, 'r') as f: data = json.load(f) session_data_list.append(data) # Create analytics instance analytics = ConversationAnalytics(llm_backend) loaded_count = analytics.load_sessions(session_data_list) if loaded_count == 0: return "❌ No valid sessions found in uploaded files.", "", None # Generate comprehensive report report = analytics.generate_comprehensive_report() # Export aggregated data export_data = analytics.export_aggregated_data() export_file = save_json_file(export_data, "multi_session_analysis") status = f"✅ Successfully analyzed {loaded_count} sessions from {len(uploaded_files)} files" return status, report, export_file except Exception as e: return f"❌ Error analyzing sessions: {str(e)}", "", None # =========================== # Gradio Interface # =========================== def create_interface(): """Create the main Gradio interface""" with gr.Blocks( title="Project Echo - Qualitative Research Assistant", theme=gr.themes.Soft(primary_hue="blue", secondary_hue="slate") ) as app: gr.Markdown(""" # Project Echo - Your AI-Powered Qualitative Research Assistant Battle the blank page, reach global audiences, and uncover insights with AI assistance. """) # Show backend status if llm_backend: status_msg = f"✅ **Active LLM Provider:** {llm_backend.provider.value.upper()} | Model: {llm_backend.model}" bg_color = "rgba(0, 255, 0, 0.1)" else: status_msg = """⚠️ **LLM Provider Not Configured** **To use this app, you need to configure an LLM provider:** 1. **Easiest (HuggingFace Spaces):** Make sure your Space is PUBLIC and HF_TOKEN will be auto-available 2. **Best Quality:** Add `OPENAI_API_KEY` in Space Settings → Variables 3. **Alternative:** Add `ANTHROPIC_API_KEY` or `HUGGINGFACE_API_KEY` See the **About** tab for detailed instructions.""" bg_color = "rgba(255, 165, 0, 0.2)" gr.Markdown(f'
{status_msg}
') with gr.Tabs() as tabs: # ========== SURVEY GENERATION TAB ========== with gr.Tab("📝 Generate Survey"): gr.Markdown(""" ## Battle the Blank Page Share an outline and get AI-powered surveys drafted in minutes, complete with industry best practices. """) with gr.Row(): with gr.Column(scale=1): outline_input = gr.Textbox( label="Your Survey Outline or Topic", placeholder="Example: I want to understand patient experiences with a new diabetes medication, focusing on effectiveness, side effects, and quality of life impacts.", lines=6 ) survey_type_input = gr.Radio( label="Survey Type", choices=["Qualitative", "Quantitative", "Mixed"], value="Qualitative" ) num_questions_input = gr.Slider( label="Number of Questions", minimum=5, maximum=25, value=10, step=1 ) audience_input = gr.Textbox( label="Target Audience", placeholder="Example: Adults aged 30-65 with Type 2 diabetes", value="General audience" ) generate_btn = gr.Button("🚀 Generate Survey", variant="primary", size="lg") with gr.Column(scale=1): gen_status = gr.Textbox(label="Status", interactive=False) gen_output = gr.Markdown(label="Generated Survey") gen_download = gr.File(label="Download Survey JSON", visible=False) # Event handlers generate_btn.click( fn=generate_survey_from_outline, inputs=[outline_input, survey_type_input, num_questions_input, audience_input], outputs=[gen_status, gen_output, gen_download] ).then( fn=lambda x: gr.File(value=x, visible=True) if x else gr.File(visible=False), inputs=[gen_download], outputs=[gen_download] ) # ========== TRANSLATION TAB ========== with gr.Tab("🌍 Translate Survey"): gr.Markdown(""" ## Reach Global Audiences Translate your surveys automatically to streamline efforts and reach wider audiences. """) with gr.Row(): with gr.Column(scale=1): gr.Markdown("### Select Target Languages") # Create checkboxes for popular languages lang_checkboxes = gr.CheckboxGroup( label="Languages", choices=get_language_choices(), value=[] ) translate_btn = gr.Button("🌐 Translate Survey", variant="primary", size="lg") gr.Markdown(""" **Note:** Make sure you've generated a survey first, or upload one using the JSON format. """) with gr.Column(scale=1): trans_status = gr.Textbox(label="Translation Status", interactive=False) trans_output = gr.Markdown(label="Translations") trans_download = gr.File(label="Download Translations JSON", visible=False) # Event handlers def extract_lang_codes(selected_items): """Extract language codes from checkbox selections""" return [item.split(" - ")[0] for item in selected_items] translate_btn.click( fn=lambda x: translate_current_survey(extract_lang_codes(x)), inputs=[lang_checkboxes], outputs=[trans_status, trans_output, trans_download] ).then( fn=lambda x: gr.File(value=x, visible=True) if x else gr.File(visible=False), inputs=[trans_download], outputs=[trans_download] ) # ========== ANALYSIS TAB ========== with gr.Tab("📊 Analyze Data"): gr.Markdown(""" ## Uncover Key Insights Upload your survey responses and get AI-assisted summaries of key findings, themes, and trends. """) with gr.Row(): with gr.Column(scale=1): responses_input = gr.Textbox( label="Survey Responses (JSON)", placeholder='[{"q1": "response 1", "q2": "response 2"}, ...]', lines=10 ) questions_input = gr.Textbox( label="Questions (JSON, Optional)", placeholder='[{"question_text": "What is your experience?", ...}]', lines=5 ) with gr.Row(): analyze_btn = gr.Button("🔍 Analyze Data", variant="primary", size="lg") example_btn = gr.Button("Load Example", variant="secondary") with gr.Column(scale=1): analysis_status = gr.Textbox(label="Status", interactive=False) analysis_output = gr.Markdown(label="Analysis Report") analysis_download = gr.File(label="Download Analysis JSON", visible=False) # Event handlers analyze_btn.click( fn=analyze_survey_data, inputs=[responses_input, questions_input], outputs=[analysis_status, analysis_output, analysis_download] ).then( fn=lambda x: gr.File(value=x, visible=True) if x else gr.File(visible=False), inputs=[analysis_download], outputs=[analysis_download] ) example_btn.click( fn=load_example_responses, outputs=[responses_input] ) # ========== CONVERSATIONAL RESEARCH TAB ========== with gr.Tab("💬 Conversational Research"): gr.Markdown(""" ## AI-Moderated Conversations Design conversation flows and conduct AI-powered qualitative interviews with respondents. """) with gr.Tabs(): # Design Flow Sub-Tab with gr.Tab("🎨 Design Flow"): gr.Markdown(""" ### Create Conversation Flows Design custom conversation paths for AI-moderated interviews. """) with gr.Row(): with gr.Column(scale=1): gr.Markdown("#### Flow Setup") flow_name_input = gr.Textbox( label="Flow Name", placeholder="e.g., HCP Interview for New Dermatology Product", value="" ) flow_desc_input = gr.Textbox( label="Flow Description", placeholder="Describe the purpose of this conversation flow...", lines=3 ) with gr.Row(): create_flow_btn = gr.Button("✨ Create New Flow", variant="primary") load_example_flow_btn = gr.Button("📋 Load Example", variant="secondary") with gr.Row(): regenerate_flow_btn = gr.Button("🔄 Regenerate Flow", variant="secondary") clear_flow_btn = gr.Button("🗑️ Clear All Steps", variant="stop") flow_id_state = gr.State(value="") gr.Markdown("#### Add Steps to Flow") node_content_input = gr.Textbox( label="Question/Message", placeholder="Enter the question or message for this step...", lines=4 ) node_type_input = gr.Radio( label="Step Type", choices=["Question", "End"], value="Question" ) add_node_btn = gr.Button("➕ Add Step", variant="secondary") save_flow_btn = gr.Button("💾 Save Flow", variant="primary") with gr.Column(scale=1): flow_status = gr.Textbox(label="Status", interactive=False) flow_display = gr.Markdown(label="Flow Preview", value="No flow created yet") flow_download = gr.File(label="Download Flow JSON", visible=False) # Event handlers for flow design create_flow_btn.click( fn=create_new_flow, inputs=[flow_name_input, flow_desc_input], outputs=[flow_status, flow_display, flow_id_state] ) load_example_flow_btn.click( fn=load_example_flow, outputs=[flow_status, flow_display, flow_id_state] ) regenerate_flow_btn.click( fn=regenerate_flow_content, inputs=[flow_id_state], outputs=[flow_status, flow_display] ) def clear_flow(flow_id): """Clear all steps from the current flow""" if not flow_id: return "❌ No flow selected.", "" flow = saved_flows.get(flow_id) if flow: flow.nodes = [] return "✅ All steps cleared. You can now add new ones.", display_flow(flow) return "❌ Flow not found.", "" clear_flow_btn.click( fn=clear_flow, inputs=[flow_id_state], outputs=[flow_status, flow_display] ) add_node_btn.click( fn=add_flow_node, inputs=[flow_id_state, node_content_input, node_type_input], outputs=[flow_status, flow_display] ).then( fn=lambda: "", outputs=[node_content_input] ) save_flow_btn.click( fn=save_current_flow, inputs=[flow_id_state], outputs=[flow_status, flow_download] ).then( fn=lambda x: gr.File(value=x, visible=True) if x else gr.File(visible=False), inputs=[flow_download], outputs=[flow_download] ) # Conduct Interview Sub-Tab with gr.Tab("🎙️ Conduct Interview"): gr.Markdown(""" ### AI-Moderated Interview Start a conversation session with the AI moderator using your designed flow. """) with gr.Row(): with gr.Column(scale=2): conversation_flow_selector = gr.State(value="") gr.Markdown(""" **Instructions:** 1. Design a flow in the 'Design Flow' tab first (or load the example) 2. Configure AI moderator settings below (optional) 3. Click 'Start Conversation' to begin 4. The AI moderator will ask questions and adapt with follow-ups 5. Generate summary and export when finished """) # Moderator Configuration with gr.Accordion("⚙️ AI Moderator Settings", open=False): gr.Markdown("**Follow-up Question Configuration**") probing_threshold_slider = gr.Slider( label="Follow-up Frequency", info="Ask dynamic follow-ups every N user responses", minimum=2, maximum=10, value=3, step=1 ) probing_status = gr.Textbox(label="Settings Status", interactive=False, value="Default: Every 3 responses") with gr.Row(): start_conversation_btn = gr.Button("🚀 Start Conversation", variant="primary", scale=2) export_conversation_btn = gr.Button("📥 Export", variant="secondary", scale=1) summary_btn = gr.Button("✨ Generate Summary", variant="secondary", scale=2) conversation_status = gr.Textbox(label="Session Status", interactive=False) conversation_download = gr.File(label="Download Transcript", visible=False) # Summary Display with gr.Accordion("📝 Conversation Summary", open=False): summary_display = gr.Markdown(label="AI-Generated Summary", value="No summary yet. Complete conversation and click 'Generate Summary'.") with gr.Column(scale=3): chatbot = gr.Chatbot( label="AI-Moderated Interview", height=400 ) msg_input = gr.Textbox( label="Your Response", placeholder="Type your response here...", lines=2 ) with gr.Row(): submit_btn = gr.Button("Send", variant="primary") clear_btn = gr.Button("Clear") # Live Metrics Panel with gr.Accordion("📊 Live Metrics", open=True): metrics_display = gr.Markdown(value="**No Active Session**\n\nStart a conversation to see metrics.") # Chat event handlers def user_submit(user_message, history): """Handle user message submission""" if not user_message: return history, history, "" return history, history + [[user_message, None]], "" def bot_respond(history): """Get bot response and update metrics""" if not history or history[-1][1] is not None: return history, "", get_conversation_metrics() user_msg = history[-1][0] updated_history, status = chat_with_moderator(user_msg, history[:-1]) metrics = get_conversation_metrics() return updated_history, status, metrics # Probing threshold configuration probing_threshold_slider.change( fn=update_probing_threshold, inputs=[probing_threshold_slider], outputs=[probing_status] ) # Start conversation start_conversation_btn.click( fn=lambda: saved_flows[list(saved_flows.keys())[-1]].id if saved_flows else "", outputs=[conversation_flow_selector] ).then( fn=start_conversation_session, inputs=[conversation_flow_selector], outputs=[chatbot, conversation_status] ).then( fn=get_conversation_metrics, outputs=[metrics_display] ) # Message submission msg_input.submit( fn=user_submit, inputs=[msg_input, chatbot], outputs=[chatbot, chatbot, msg_input], queue=False ).then( fn=bot_respond, inputs=[chatbot], outputs=[chatbot, conversation_status, metrics_display] ) submit_btn.click( fn=user_submit, inputs=[msg_input, chatbot], outputs=[chatbot, chatbot, msg_input], queue=False ).then( fn=bot_respond, inputs=[chatbot], outputs=[chatbot, conversation_status, metrics_display] ) clear_btn.click(lambda: None, None, chatbot, queue=False) # Generate summary summary_btn.click( fn=generate_conversation_summary, outputs=[conversation_status, summary_display] ) # Export conversation export_conversation_btn.click( fn=export_conversation, outputs=[conversation_status, conversation_download] ).then( fn=lambda x: gr.File(value=x, visible=True) if x else gr.File(visible=False), inputs=[conversation_download], outputs=[conversation_download] ) # Analyze Conversations Sub-Tab with gr.Tab("📊 Analyze Conversations"): gr.Markdown(""" ### Multi-Session Analysis Analyze patterns and insights across multiple conversation sessions. Upload conversation JSON files (exported from the 'Conduct Interview' tab). """) with gr.Row(): with gr.Column(scale=1): gr.Markdown(""" **How it works:** 1. Conduct multiple interviews in the 'Conduct Interview' tab 2. Export each conversation as JSON 3. Upload all JSON files here 4. Click 'Analyze Sessions' to generate comprehensive report 5. Get AI-powered insights across all conversations **Minimum Requirements:** - At least 3-5 sessions recommended - 10+ total user responses across all sessions """) session_files_upload = gr.File( label="Upload Conversation Sessions (JSON)", file_count="multiple", file_types=[".json"], type="filepath" ) analyze_sessions_btn = gr.Button("🔍 Analyze Sessions", variant="primary", size="lg") analytics_status = gr.Textbox(label="Analysis Status", interactive=False) analytics_download = gr.File(label="Download Analysis JSON", visible=False) with gr.Column(scale=1): analytics_report = gr.Markdown( label="Multi-Session Analysis Report", value="""# Multi-Session Analysis **Upload session files to begin analysis.** The report will include: - 📊 Aggregate statistics across all sessions - 🔑 Common keywords and topics - 💡 AI-powered cross-session insights - 📋 Individual session summaries - 🎯 Research recommendations """ ) # Analytics event handlers analyze_sessions_btn.click( fn=analyze_multiple_sessions, inputs=[session_files_upload], outputs=[analytics_status, analytics_report, analytics_download] ).then( fn=lambda x: gr.File(value=x, visible=True) if x else gr.File(visible=False), inputs=[analytics_download], outputs=[analytics_download] ) # ========== ABOUT TAB ========== with gr.Tab("ℹ️ About"): gr.Markdown(""" ## About Project Echo Project Echo is a comprehensive qualitative research assistant that helps you: ### 🎯 Generate Surveys - Create professional surveys from simple outlines - Follow industry best practices automatically - Save hours of questionnaire design time ### 🌍 Translate Globally - Reach audiences in 18+ languages - Maintain cultural appropriateness - Expand your research scope effortlessly ### 📊 Analyze Results - Extract key themes automatically - Identify patterns and trends - Generate actionable insights ### 🔧 Configuration Guide **For HuggingFace Spaces (Recommended):** No configuration needed! The app automatically uses the HF Inference API with the built-in `HF_TOKEN`. **Supported Models:** - Default: `mistralai/Mixtral-8x7B-Instruct-v0.1` - You can change by setting `LLM_MODEL` environment variable **For Other LLM Providers:** Add these environment variables in your Space Settings: 1. **OpenAI** (Best quality, paid): - `LLM_PROVIDER=openai` - `OPENAI_API_KEY=sk-your-key` 2. **Anthropic Claude** (Best reasoning, paid): - `LLM_PROVIDER=anthropic` - `ANTHROPIC_API_KEY=your-key` 3. **Custom HuggingFace Model**: - `LLM_PROVIDER=huggingface` - `LLM_MODEL=your-model-name` **💡 Pro Tip:** For production use, we recommend OpenAI or Anthropic for faster, more reliable results. **Supported LLM Providers:** - HuggingFace Inference API (Free tier available) - OpenAI (GPT-4, GPT-4o-mini, GPT-3.5) - Anthropic (Claude 3.5 Sonnet, Claude 3 Opus) - LM Studio (local development only) ### 📄 Data Privacy - All processing is done through your configured LLM provider - No data is stored permanently by this application - Survey data and responses remain in your control ### 🚀 Getting Started 1. **Generate** a survey from your research outline 2. **Translate** it to reach global audiences 3. Collect responses from participants 4. **Analyze** the data to uncover insights --- Built with ❤️ using Gradio and state-of-the-art LLMs """) return app # =========================== # Main Entry Point # =========================== if __name__ == "__main__": demo = create_interface() # Launch with appropriate settings demo.launch( server_name="0.0.0.0", # Allow external access server_port=7860, # Standard HF Spaces port share=False, # Don't create a public link (HF Spaces handles this) show_error=True )