App.py

App.py

@@ -0,0 +1,415 @@
+import gradio as gr
+from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline, TextIteratorStreamer
+import random
+import threading
+import torch
+import os
+import time
+from typing import List, Dict, Generator, Tuple, Optional
+import logging
+
+# Set up logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+# --- Best Free Models for Council ---
+MODELS = [
+    ("meta-llama/Meta-Llama-3-8B-Instruct", "Llama 3 8B Instruct"),
+    ("Qwen/Qwen1.5-7B-Chat", "Qwen1.5 7B Chat"),
+    ("HuggingFaceH4/zephyr-7b-beta", "Zephyr 7B Beta"),
+    ("mistralai/Mistral-7B-Instruct-v0.2", "Mistral 7B Instruct"),
+]
+
+# Define council member personas with enhanced characteristics
+PERSONAS = [
+    {
+        "name": "Dr. Ana Rodriguez",
+        "description": "An analytical scientist who values empirical evidence and logical reasoning. Often plays devil's advocate and questions assumptions.",
+        "traits": "analytical, skeptical, evidence-focused",
+        "style": "formal, precise, methodical",
+        "emoji": "🔬"
+    },
+    {
+        "name": "Professor Marcus Chen",
+        "description": "A creative philosopher with an interest in ethics and societal implications. Considers the bigger picture and long-term consequences.",
+        "traits": "philosophical, visionary, empathetic",
+        "style": "eloquent, metaphorical, conceptual",
+        "emoji": "🧠"
+    },
+    {
+        "name": "Sarah Johnson",
+        "description": "A pragmatic problem-solver with real-world experience. Focuses on practicality and implementation details.",
+        "traits": "practical, solution-oriented, experienced",
+        "style": "direct, concise, example-driven",
+        "emoji": "🛠️"
+    },
+    {
+        "name": "Dr. Emeka Okafor",
+        "description": "A social scientist specializing in cultural perspectives and community impacts. Brings diverse viewpoints and contextual understanding.",
+        "traits": "culturally aware, nuanced, community-focused",
+        "style": "inclusive, storytelling, perspective-oriented",
+        "emoji": "🌍"
+    }
+]
+
+# Cache for models to avoid reloading
+model_cache = {}
+
+def load_model(model_id: str) -> Tuple[pipeline, AutoTokenizer]:
+    """Load model and tokenizer with caching to improve performance"""
+    global model_cache
+    
+    if model_id in model_cache:
+        logger.info(f"Using cached model: {model_id}")
+        return model_cache[model_id]
+    
+    logger.info(f"Loading model: {model_id}")
+    try:
+        # Set environmental variables for optimizations
+        os.environ["TOKENIZERS_PARALLELISM"] = "true"
+        
+        # Load tokenizer and model
+        tokenizer = AutoTokenizer.from_pretrained(model_id)
+        
+        # Determine if CUDA is available and set appropriate device
+        device = "cuda" if torch.cuda.is_available() else "cpu"
+        
+        # Configure model loading for memory efficiency
+        model_kwargs = {
+            "trust_remote_code": True,
+            "device_map": "auto",
+            "torch_dtype": torch.float16 if device == "cuda" else torch.float32
+        }
+        
+        model = AutoModelForCausalLM.from_pretrained(model_id, **model_kwargs)
+        
+        # Create pipeline with appropriate settings
+        pipe = pipeline("text-generation", 
+                       model=model, 
+                       tokenizer=tokenizer, 
+                       max_new_tokens=512, 
+                       device=model.device)
+        
+        # Cache the model and tokenizer
+        model_cache[model_id] = (pipe, tokenizer)
+        logger.info(f"Model loaded successfully: {model_id} on {device}")
+        return pipe, tokenizer
+    
+    except Exception as e:
+        logger.error(f"Failed to load model {model_id}: {str(e)}")
+        raise
+
+def create_debate_prompt(user_prompt: str, 
+                        persona: Dict, 
+                        debate_style: str = "Balanced", 
+                        previous_responses: Optional[List[str]] = None) -> str:
+    """Create a prompt that encourages a natural debate-like response with adjustable style"""
+    persona_desc = f"You are {persona['name']}, {persona['description']} Your communication style is {persona['style']}."
+    
+    # Adjust prompt based on debate style
+    style_guidance = ""
+    if debate_style == "Collaborative":
+        style_guidance = "Focus on building upon and synthesizing the ideas of others. Look for common ground and areas of agreement."
+    elif debate_style == "Adversarial":
+        style_guidance = "Challenge assumptions and present contrasting viewpoints. Don't be afraid to disagree strongly with others."
+    else:  # Balanced
+        style_guidance = "Present your authentic perspective while being respectful of other viewpoints. Balance critique with constructive ideas."
+    
+    if not previous_responses:
+        prompt = f"""{persona_desc}
+
+You are part of a council debating the following topic:
+"{user_prompt}"
+
+{style_guidance}
+
+Give your authentic perspective on this topic based on your persona. Be natural and conversational.
+Directly address the topic without hedging or being overly formal. Make specific points that others can respond to.
+Keep your response to 3-4 paragraphs maximum.
+
+{persona['name']}:"""
+    else:
+        debate_history = "\n\n".join(previous_responses)
+        prompt = f"""{persona_desc}
+
+You are part of a council debating the following topic:
+"{user_prompt}"
+
+{style_guidance}
+
+The debate so far:
+{debate_history}
+
+Now it's your turn to speak. Based on your persona and the previous speakers:
+- You may agree or disagree with previous points
+- Add new perspectives they missed
+- Point out flaws in reasoning or suggest compromises
+- Address someone directly if appropriate
+- Be authentic to your character - don't just summarize
+
+Give your natural, conversational response as if in a real discussion.
+Keep your response to 3-4 paragraphs maximum.
+
+{persona['name']}:"""
+    
+    return prompt
+
+def create_synthesis_prompt(user_prompt: str, all_responses: List[str]) -> str:
+    """Create a prompt for the facilitator to synthesize the debate"""
+    debate_history = "\n\n".join(all_responses)
+    prompt = f"""You are the Facilitator, responsible for synthesizing the council's discussion on:
+"{user_prompt}"
+
+The full debate:
+{debate_history}
+
+Provide a thoughtful synthesis that:
+1. Identifies the key points of agreement and disagreement
+2. Highlights the most compelling insights from each perspective
+3. Draws a balanced conclusion that respects the nuance of the discussion
+4. Offers a path forward or recommendation when appropriate
+
+Be concise but comprehensive. Focus on substance over style.
+Keep your synthesis to 3-5 paragraphs maximum.
+
+Facilitator:"""
+    return prompt
+
+def stream_model_response(pipe: pipeline, 
+                         tokenizer: AutoTokenizer,
+                         prompt: str, 
+                         speaker_name: str,
+                         temperature: float = 0.7) -> Generator[str, None, None]:
+    """Stream model responses with better error handling"""
+    try:
+        # Set up the streamer
+        streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)
+        input_ids = tokenizer(prompt, return_tensors="pt").input_ids.to(pipe.model.device)
+        
+        # Run model generation in a separate thread
+        generation_kwargs = dict(
+            input_ids=input_ids,
+            streamer=streamer,
+            max_new_tokens=512,
+            do_sample=True,
+            temperature=temperature,
+            top_p=0.95,
+            repetition_penalty=1.1,
+            eos_token_id=tokenizer.eos_token_id,
+        )
+        
+        thread = threading.Thread(
+            target=pipe.model.generate,
+            kwargs=generation_kwargs
+        )
+        thread.start()
+        
+        # Stream the response as it's generated
+        response = ""
+        for new_text in streamer:
+            response += new_text
+            # Add the emoji to the speaker name
+            yield f"**{speaker_name}:** {response.strip()}"
+        
+        thread.join()
+        return response.strip()
+    
+    except Exception as e:
+        logger.error(f"Error streaming response: {str(e)}")
+        yield f"**{speaker_name}:** [Error generating response. Please try again.]"
+
+def council_chat_stream(user_prompt: str, 
+                       num_members: int = 3, 
+                       debate_style: str = "Balanced",
+                       temperature: float = 0.7) -> Generator[str, None, None]:
+    """Generate a council debate with configurable number of members and style"""
+    # Validate inputs
+    if not user_prompt.strip():
+        yield "Please enter a topic for the council to debate."
+        return
+    
+    start_time = time.time()
+    
+    # Determine which personas and models to use
+    selected_personas = random.sample(PERSONAS, min(num_members, len(PERSONAS)))
+    selected_models = random.sample(MODELS, min(num_members, len(MODELS)))
+    
+    # Load models
+    loaded_models = []
+    for model_id, _ in selected_models:
+        try:
+            pipe, tokenizer = load_model(model_id)
+            loaded_models.append((pipe, tokenizer))
+        except Exception as e:
+            logger.error(f"Failed to load model {model_id}: {str(e)}")
+            yield f"Error loading model {model_id}. Please try again."
+            return
+    
+    responses = []
+    formatted_responses = []
+    persona_responses = []
+
+    # Generate responses from each council member
+    for i, (persona, (pipe, tokenizer), (model_id, model_name)) in enumerate(zip(selected_personas, loaded_models, selected_models)):
+        display_name = f"{persona['emoji']} {persona['name']} ({model_name})"
+        
+        if i == 0:
+            prompt = create_debate_prompt(user_prompt, persona, debate_style)
+        else:
+            prompt = create_debate_prompt(user_prompt, persona, debate_style, persona_responses)
+        
+        # Stream and collect response
+        response_text = ""
+        for partial in stream_model_response(pipe, tokenizer, prompt, display_name, temperature):
+            # Format the full output
+            current_output = f"**User:** {user_prompt}\n\n" + "\n\n".join(formatted_responses + [partial])
+            yield current_output
+            response_text = partial.split("**:")[-1].strip()
+        
+        # Add this response to the collected responses
+        persona_responses.append(f"{persona['name']}: {response_text}")
+        formatted_responses.append(partial)
+    
+    # Facilitator synthesis (use a random model)
+    rand_model_idx = random.randint(0, len(loaded_models) - 1)
+    pipe, tokenizer = loaded_models[rand_model_idx]
+    
+    synthesis_prompt = create_synthesis_prompt(user_prompt, persona_responses)
+    synthesis = ""
+    
+    for partial in stream_model_response(pipe, tokenizer, synthesis_prompt, "✨ Facilitator's Synthesis", temperature):
+        current_output = f"**User:** {user_prompt}\n\n" + "\n\n".join(formatted_responses + [partial])
+        yield current_output
+        synthesis = partial
+    
+    # Final output with timing
+    elapsed_time = time.time() - start_time
+    transcript = f"**User:** {user_prompt}\n\n" + "\n\n".join(formatted_responses) + f"\n\n{synthesis}\n\n---\n*Debate completed in {elapsed_time:.1f} seconds*"
+    yield transcript
+
+# Gradio interface with improved UI
+def build_gradio_interface():
+    """Build a more structured and visually appealing Gradio interface"""
+    
+    # Custom CSS for better appearance
+    custom_css = """
+    .gradio-container {
+        font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
+    }
+    .council-header {
+        text-align: center;
+        margin-bottom: 1em;
+    }
+    .council-member {
+        margin: 0.5em 0;
+        padding: 0.5em;
+        border-radius: 8px;
+        background-color: #f5f5f5;
+    }
+    """
+    
+    with gr.Blocks(theme=gr.themes.Soft(), css=custom_css) as demo:
+        gr.Markdown("# 🤖🏛️ AI Council Debate", elem_classes=["council-header"])
+        gr.Markdown("Ask a question and watch as AI personas debate and deliberate on your topic with different perspectives.")
+        
+        with gr.Row():
+            with gr.Column():
+                inp = gr.Textbox(
+                    label="Your Topic or Question", 
+                    lines=4, 
+                    placeholder="Enter a topic, question, or issue for the council to debate..."
+                )
+                
+                # Advanced options
+                with gr.Accordion("Advanced Options", open=False):
+                    with gr.Row():
+                        num_members = gr.Slider(
+                            minimum=2, 
+                            maximum=len(PERSONAS), 
+                            value=3, 
+                            step=1, 
+                            label="Number of Council Members"
+                        )
+                        
+                    with gr.Row():
+                        debate_style = gr.Radio(
+                            ["Collaborative", "Adversarial", "Balanced"], 
+                            label="Debate Style", 
+                            value="Balanced"
+                        )
+                        
+                    with gr.Row():
+                        temperature = gr.Slider(
+                            minimum=0.1, 
+                            maximum=1.0, 
+                            value=0.7, 
+                            step=0.1, 
+                            label="Temperature (Creativity)"
+                        )
+                
+                btn = gr.Button("Start Council Debate", variant="primary")
+                
+            with gr.Column():
+                out = gr.Markdown(label="Council Debate Transcript")
+        
+        # Display council members information
+        with gr.Accordion("Meet the Council Members", open=False):
+            member_info = ""
+            for persona in PERSONAS:
+                member_info += f"""
+                <div class="council-member">
+                    <h3>{persona['emoji']} {persona['name']}</h3>
+                    <p><strong>Description:</strong> {persona['description']}</p>
+                    <p><strong>Traits:</strong> {persona['traits']}</p>
+                    <p><strong>Communication Style:</strong> {persona['style']}</p>
+                </div>
+                """
+            gr.HTML(member_info)
+        
+        # Example prompts for users to try
+        with gr.Accordion("Example Topics", open=False):
+            examples = [
+                "What role should AI play in education?",
+                "Is universal basic income a good idea?",
+                "How should society balance privacy concerns with security needs?",
+                "What are the ethical implications of genetic engineering?",
+                "How can we address climate change effectively?"
+            ]
+            gr.Examples(examples=examples, inputs=inp)
+        
+        # Event handlers
+        btn.click(
+            fn=council_chat_stream, 
+            inputs=[inp, num_members, debate_style, temperature], 
+            outputs=out
+        )
+        
+        # Footer with additional information
+        gr.Markdown("""
+        ### About This App
+        
+        This application demonstrates how multiple AI models can collaborate in a structured debate.
+        Each AI persona has distinctive traits and perspectives that influence how they approach topics.
+        
+        The models used are open-source LLMs hosted on Hugging Face:
+        - Meta's Llama 3 8B Instruct
+        - Qwen 1.5 7B Chat
+        - Zephyr 7B Beta
+        - Mistral 7B Instruct v0.2
+        
+        ⚠️ Note: First-time loading may take a minute as models are downloaded and initialized.
+        """)
+    
+    return demo
+
+# Main application
+if __name__ == "__main__":
+    # Check GPU availability
+    if torch.cuda.is_available():
+        logger.info(f"GPU available: {torch.cuda.get_device_name(0)}")
+    else:
+        logger.info("No GPU available, using CPU. Performance may be slower.")
+    
+    # Create and launch the Gradio interface
+    demo = build_gradio_interface()
+    demo.launch()