import streamlit as st import os import json import anthropic import requests from typing import Dict, Any, List, Optional # Initialize the environment variables for different providers ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY") OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY") PERPLEXITY_API_KEY = os.getenv("PERPLEXITY_API_KEY") class AIProviderManager: """Manager for multiple AI providers""" def __init__(self): self.providers = {} self.initialize_providers() def initialize_providers(self): """Initialize available AI providers""" # Initialize Anthropic (Claude) if ANTHROPIC_API_KEY: try: self.providers["claude"] = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY) # Test connection by listing models try: models = self.providers["claude"].models.list() model_list = [model.id for model in models.data] st.sidebar.success(f"🟢 Claude AI connected - Available models: {', '.join(model_list[:3])}") except Exception as e: st.sidebar.warning(f"🟡 Claude AI connected but couldn't list models: {str(e)}") except Exception as e: st.sidebar.error(f"🔴 Error initializing Claude: {str(e)}") # Initialize OpenAI if available if OPENAI_API_KEY: try: import openai openai.api_key = OPENAI_API_KEY self.providers["openai"] = True # Test connection by listing models try: client = openai.OpenAI(api_key=OPENAI_API_KEY) models = client.models.list() st.sidebar.success("🟢 OpenAI connected") except Exception as e: st.sidebar.warning(f"🟡 OpenAI connected but couldn't list models: {str(e)}") except ImportError: st.sidebar.warning("⚠️ OpenAI SDK not installed. Run 'pip install openai'") except Exception as e: st.sidebar.error(f"🔴 Error initializing OpenAI: {str(e)}") # Initialize DeepSeek if available if DEEPSEEK_API_KEY: try: import openai as deepseek_client self.providers["deepseek"] = { "client": deepseek_client, "base_url": "https://api.deepseek.com" } st.sidebar.success("🟢 DeepSeek AI connected") except ImportError: st.sidebar.warning("⚠️ OpenAI SDK needed for DeepSeek. Run 'pip install openai'") except Exception as e: st.sidebar.error(f"🔴 Error initializing DeepSeek: {str(e)}") # Initialize Perplexity if available if PERPLEXITY_API_KEY: self.providers["perplexity"] = True st.sidebar.success("🟢 Perplexity API connected") # In multi_llm_provider.py, update the get_available_models method to provide cleaner model names def get_available_models(self): """Get all available models across providers with improved naming""" models = {} # Claude models - dynamically get if possible if "claude" in self.providers: try: claude_models = self.providers["claude"].models.list() for model in claude_models.data: # Create more readable model names model_id = model.id # Extract version and type for more readable names if "claude-3" in model_id: parts = model_id.split('-') if len(parts) >= 4: version = parts[1] variant = parts[2].capitalize() date = parts[3][:8] # Get just the date part display_name = f"Claude {version} {variant} ({date})" models[model_id] = display_name else: models[model_id] = f"Claude ({model_id})" else: models[model_id] = f"Claude ({model_id})" except Exception: # Fallback to hardcoded models if API call fails models.update({ "claude-3-sonnet-20250219": "Claude 3 Sonnet (Feb 2025)", "claude-3-haiku-20250319": "Claude 3 Haiku (Mar 2025)", "claude-3-opus-20250229": "Claude 3 Opus (Feb 2025)", "claude-3-7-sonnet-20250219": "Claude 3.7 Sonnet (Feb 2025)" }) # OpenAI models - with better naming if "openai" in self.providers: models.update({ "gpt-4": "GPT-4", "gpt-4-turbo": "GPT-4 Turbo", "gpt-3.5-turbo": "GPT-3.5 Turbo" }) # DeepSeek models - with better naming if "deepseek" in self.providers: models.update({ "deepseek-chat": "DeepSeek Chat", "deepseek-coder": "DeepSeek Coder" }) return models def generate_text(self, prompt: str, model: str, system_prompt: str = None, temperature: float = 0.7, max_tokens: int = 1000): """Generate text using the specified model""" # Check if the specified model is available if model.startswith("claude") and "claude" in self.providers: return self._generate_with_claude(prompt, model, system_prompt, temperature, max_tokens) elif model.startswith("gpt") and "openai" in self.providers: return self._generate_with_openai(prompt, model, system_prompt, temperature, max_tokens) elif model.startswith("deepseek") and "deepseek" in self.providers: return self._generate_with_deepseek(prompt, model, system_prompt, temperature, max_tokens) else: # Try to find any available provider available_providers = [] if "claude" in self.providers: available_providers.append("claude") if "openai" in self.providers: available_providers.append("openai") if "deepseek" in self.providers: available_providers.append("deepseek") if available_providers: provider = available_providers[0] if provider == "claude": # Get available Claude models try: models = self.providers["claude"].models.list() if models.data: fallback_model = models.data[0].id st.warning(f"Model {model} not available. Falling back to {fallback_model}.") return self._generate_with_claude(prompt, fallback_model, system_prompt, temperature, max_tokens) except: pass # If model list fails, use hardcoded fallback st.warning(f"Model {model} not available. Falling back to Claude 3 Sonnet.") return self._generate_with_claude(prompt, "claude-3-sonnet-20250219", system_prompt, temperature, max_tokens) elif provider == "openai": st.warning(f"Model {model} not available. Falling back to GPT-3.5 Turbo.") return self._generate_with_openai(prompt, "gpt-3.5-turbo", system_prompt, temperature, max_tokens) elif provider == "deepseek": st.warning(f"Model {model} not available. Falling back to DeepSeek Chat.") return self._generate_with_deepseek(prompt, "deepseek-chat", system_prompt, temperature, max_tokens) else: raise ValueError(f"No AI provider available for model: {model}") def _generate_with_claude(self, prompt: str, model: str, system_prompt: str = None, temperature: float = 0.7, max_tokens: int = 1000): """Generate text using Claude with enhanced error handling""" client = self.providers["claude"] messages = [{"role": "user", "content": prompt}] try: response = client.messages.create( model=model, max_tokens=max_tokens, temperature=temperature, system=system_prompt if system_prompt else "You are a helpful assistant.", messages=messages ) return response.content[0].text except Exception as e: error_msg = str(e) st.error(f"Claude API Error ({model}): {error_msg}") # Check for model availability errors if ("model" in error_msg.lower() and "not" in error_msg.lower()) or "not_found" in error_msg.lower(): try: # Try to get available models available_models = [] try: models_list = client.models.list() available_models = [m.id for m in models_list.data] st.info(f"Available Claude models: {', '.join(available_models)}") except: # If listing fails, use fallback list available_models = ["claude-3-7-sonnet-20250219", "claude-3-sonnet-20250219", "claude-3-haiku-20250319", "claude-3-opus-20250229"] # Try available models for fallback_model in available_models: if fallback_model != model: try: st.warning(f"Trying fallback model: {fallback_model}") response = client.messages.create( model=fallback_model, max_tokens=max_tokens, temperature=temperature, system=system_prompt if system_prompt else "You are a helpful assistant.", messages=messages ) return response.content[0].text except Exception as fallback_error: st.warning(f"Fallback to {fallback_model} failed: {str(fallback_error)}") continue except Exception as list_error: st.error(f"Error while attempting fallbacks: {str(list_error)}") # If we reach here, all fallbacks failed or it's another type of error raise ValueError(f"Claude API failed with error: {error_msg}") def _generate_with_openai(self, prompt: str, model: str, system_prompt: str = None, temperature: float = 0.7, max_tokens: int = 1000): """Generate text using OpenAI with enhanced error handling""" import openai messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) messages.append({"role": "user", "content": prompt}) try: response = openai.chat.completions.create( model=model, messages=messages, temperature=temperature, max_tokens=max_tokens ) return response.choices[0].message.content except Exception as e: error_msg = str(e) st.error(f"OpenAI API Error ({model}): {error_msg}") # Check for model availability errors if "model" in error_msg.lower() and ("not" in error_msg.lower() or "find" in error_msg.lower()): # Try fallback models fallback_models = ["gpt-3.5-turbo", "gpt-4"] for fallback_model in fallback_models: if fallback_model != model: try: st.warning(f"Trying fallback model: {fallback_model}") response = openai.chat.completions.create( model=fallback_model, messages=messages, temperature=temperature, max_tokens=max_tokens ) return response.choices[0].message.content except Exception as fallback_error: st.warning(f"Fallback to {fallback_model} failed: {str(fallback_error)}") continue # If all fallbacks fail or it's another type of error raise ValueError(f"OpenAI API failed with error: {error_msg}") def _generate_with_deepseek(self, prompt: str, model: str, system_prompt: str = None, temperature: float = 0.7, max_tokens: int = 1000): """Generate text using DeepSeek with enhanced error handling""" import openai as deepseek deepseek.api_key = DEEPSEEK_API_KEY deepseek.base_url = "https://api.deepseek.com" messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) messages.append({"role": "user", "content": prompt}) try: response = deepseek.chat.completions.create( model="deepseek-chat" if model == "deepseek-chat" else "deepseek-coder", messages=messages, temperature=temperature, max_tokens=max_tokens ) return response.choices[0].message.content except Exception as e: error_msg = str(e) st.error(f"DeepSeek API Error ({model}): {error_msg}") # Check for model availability errors and try fallback if model == "deepseek-chat": try: st.warning("Trying fallback model: deepseek-coder") response = deepseek.chat.completions.create( model="deepseek-coder", messages=messages, temperature=temperature, max_tokens=max_tokens ) return response.choices[0].message.content except Exception as fallback_error: st.warning(f"Fallback failed: {str(fallback_error)}") elif model == "deepseek-coder": try: st.warning("Trying fallback model: deepseek-chat") response = deepseek.chat.completions.create( model="deepseek-chat", messages=messages, temperature=temperature, max_tokens=max_tokens ) return response.choices[0].message.content except Exception as fallback_error: st.warning(f"Fallback failed: {str(fallback_error)}") # If fallbacks fail or it's another type of error raise ValueError(f"DeepSeek API failed with error: {error_msg}") def web_search(self, query: str) -> List[Dict[str, Any]]: """Perform a web search using Perplexity API""" if "perplexity" not in self.providers or not PERPLEXITY_API_KEY: raise ValueError("Perplexity API not available") headers = { "Authorization": f"Bearer {PERPLEXITY_API_KEY}", "Content-Type": "application/json" } data = { "query": query, "max_results": 5 } try: response = requests.post( "https://api.perplexity.ai/search", headers=headers, json=data ) if response.status_code == 200: results = response.json() return results.get("results", []) else: st.error(f"Error in web search: {response.status_code} - {response.text}") return [] except Exception as e: st.error(f"Error performing web search: {str(e)}") return [] def enhance_with_web_search(self, slide_content: Dict[str, Any], search_query: str = None) -> Dict[str, Any]: """Enhance slide content with web search results""" if "perplexity" not in self.providers or not PERPLEXITY_API_KEY: st.warning("Web search enhancement not available (Perplexity API key required)") return slide_content # Use slide title if no search query provided if not search_query: title = slide_content.get("title", "") search_query = f"latest information about {title}" try: with st.spinner(f"Searching the web for: {search_query}"): results = self.web_search(search_query) if not results: st.warning("No search results found") return slide_content # Compile search results search_content = "Web search results:\n\n" for i, result in enumerate(results, 1): title = result.get("title", "No title") snippet = result.get("snippet", "No snippet") url = result.get("url", "No URL") search_content += f"{i}. {title}\n" search_content += f" {snippet}\n" search_content += f" Source: {url}\n\n" # Create an enrichment prompt prompt = f""" You are an expert presentation content creator. Current slide: Title: {slide_content.get('title', 'No title')} Content: {slide_content.get('content', [])} I have gathered recent information about this topic. Please use this information to enhance the slide content: {search_content} Please provide: 1. An updated slide title (if needed) 2. Enhanced content points that incorporate the latest information 3. A note about the sources of this information Return your response as JSON with the following structure: {{ "title": "Enhanced title", "content": ["Point 1", "Point 2", "Point 3"], "notes": "Notes including source information" }} """ # Get models from available providers available_models = [] if "claude" in self.providers: available_models.append("claude-3-7-sonnet-20250219") # Use newest model first available_models.append("claude-3-sonnet-20250219") if "openai" in self.providers: available_models.append("gpt-4") available_models.append("gpt-3.5-turbo") if "deepseek" in self.providers: available_models.append("deepseek-chat") # Try models until one works response = None for model in available_models: try: response = self.generate_text( prompt=prompt, model=model, system_prompt="You are an expert at enhancing presentation content with the latest information. Always respond with valid JSON.", temperature=0.5, max_tokens=2000 ) if response: break except Exception as e: st.warning(f"Error using {model}: {str(e)}. Trying next model...") continue if not response: st.error("All models failed. Could not enhance content.") return slide_content # Extract the JSON from the response try: # Find JSON in the text json_start = response.find("{") json_end = response.rfind("}") + 1 if json_start >= 0 and json_end > 0: json_str = response[json_start:json_end] enhanced_content = json.loads(json_str) # Update the slide content slide_content["title"] = enhanced_content.get("title", slide_content.get("title", "")) slide_content["content"] = enhanced_content.get("content", slide_content.get("content", [])) # Append to existing notes existing_notes = slide_content.get("notes", "") new_notes = enhanced_content.get("notes", "") slide_content["notes"] = f"{existing_notes}\n\n{new_notes}".strip() # Add source information slide_content["source_info"] = { "search_query": search_query, "results_count": len(results), "timestamp": "March 2025" } st.success("Slide enhanced with latest web information!") else: st.error("Could not extract JSON from AI response") st.info("Raw response: " + response[:500] + "...") # Show part of the response for debugging except Exception as e: st.error(f"Error processing AI response: {str(e)}") st.info("Raw response: " + response[:500] + "...") # Show part of the response for debugging except Exception as e: st.error(f"Error enhancing content with web search: {str(e)}") return slide_content def generate_image_description(self, slide_content: Dict[str, Any]) -> str: """Generate an optimized description for image generation""" if not "claude" in self.providers and not "openai" in self.providers: return "" title = slide_content.get('title', '') if isinstance(slide_content.get('content', []), list): content_text = " ".join(slide_content.get('content', []))[:500] # Limit length else: content_text = str(slide_content.get('content', ''))[:500] prompt = f""" You are an expert at creating image generation prompts. I need a detailed visual description for a presentation slide with the following content: Title: {title} Content: {content_text} Please create a detailed description (under a single paragraph of 50 words or less) that would help an AI image generator create an appropriate, professional, metaphorical image for this slide. Focus on imagery that would work well in a business presentation context - prefer abstract, conceptual, or metaphorical imagery over literal representations. Specify style (photorealistic, illustration, 3D render), color scheme, and composition. ONLY provide the description itself with no explanations, introductions, or extra text. """ try: # Choose the model based on availability available_models = [] if "claude" in self.providers: available_models.append("claude-3-haiku-20250319") if "openai" in self.providers: available_models.append("gpt-3.5-turbo") if "deepseek" in self.providers: available_models.append("deepseek-chat") # Try models until one works description = None for model in available_models: try: description = self.generate_text( prompt=prompt, model=model, system_prompt="You are an expert at creating image generation prompts for business presentations.", temperature=0.7, max_tokens=200 ) if description: break except Exception as e: st.warning(f"Error using {model} for image description: {str(e)}. Trying next model...") continue if not description: return f"An image representing {title}" # Clean up the response return description.strip() except Exception as e: st.error(f"Error generating image description: {str(e)}") return f"An image representing {title}" # Initialize the AI provider manager def get_ai_manager(): """Get or create the AI provider manager""" if 'ai_manager' not in st.session_state: st.session_state.ai_manager = AIProviderManager() return st.session_state.ai_manager # Add a diagnostic function to test model availability def test_models(): """Test model availability across providers""" ai_manager = get_ai_manager() st.write("### LLM Provider Diagnostics") # Test Claude if "claude" in ai_manager.providers: try: st.write("#### Testing Claude API:") claude = ai_manager.providers["claude"] models = claude.models.list() st.success(f"Available Claude models:") for model in models.data: st.write(f"- {model.id}") # Test a quick generation test_prompt = "Say hello in one word." with st.spinner(f"Testing with {models.data[0].id}..."): response = claude.messages.create( model=models.data[0].id, max_tokens=10, messages=[{"role": "user", "content": test_prompt}] ) st.success(f"Test response: {response.content[0].text}") except Exception as e: st.error(f"Claude API test failed: {str(e)}") else: st.warning("Claude API not configured") # Test OpenAI if "openai" in ai_manager.providers: try: st.write("#### Testing OpenAI API:") import openai client = openai.OpenAI(api_key=OPENAI_API_KEY) models = client.models.list() st.success(f"Available OpenAI models:") for model in models.data[:5]: # Show first 5 to avoid cluttering st.write(f"- {model.id}") # Test a quick generation test_prompt = "Say hello in one word." with st.spinner("Testing with gpt-3.5-turbo..."): response = client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": test_prompt}] ) st.success(f"Test response: {response.choices[0].message.content}") except Exception as e: st.error(f"OpenAI API test failed: {str(e)}") else: st.warning("OpenAI API not configured")