import os import json import time import gradio as gr from datetime import datetime from typing import List, Dict, Any, Optional, Union import threading import re import aiohttp import asyncio # Import Groq from groq import Groq class CreativeAgenticAI: """ Creative Agentic AI Chat Tool using multiple providers (Groq and Chutes) """ def __init__(self, groq_api_key: str = None, chutes_api_key: str = None, provider: str = "groq", model: str = None): """ Initialize the Creative Agentic AI system. Args: groq_api_key: Groq API key chutes_api_key: Chutes API key provider: Which provider to use ('groq' or 'chutes') model: Which model to use """ self.groq_api_key = groq_api_key self.chutes_api_key = chutes_api_key self.provider = provider self.conversation_history = [] # Initialize clients based on provider if provider == "groq" and groq_api_key: if not groq_api_key: raise ValueError("No Groq API key provided") self.groq_client = Groq(api_key=groq_api_key) self.model = model or "compound-beta" elif provider == "chutes" and chutes_api_key: if not chutes_api_key: raise ValueError("No Chutes API key provided") self.model = model or "openai/gpt-oss-20b" else: raise ValueError(f"Invalid provider or missing API key for {provider}") async def _chutes_chat_async(self, messages: List[Dict], temperature: float = 0.7, max_tokens: int = 1024) -> str: """ Async method for Chutes API chat """ headers = { "Authorization": f"Bearer {self.chutes_api_key}", "Content-Type": "application/json" } body = { "model": self.model, "messages": messages, "stream": False, # Set to False for simpler handling "max_tokens": max_tokens, "temperature": temperature } async with aiohttp.ClientSession() as session: async with session.post( "https://llm.chutes.ai/v1/chat/completions", headers=headers, json=body ) as response: if response.status == 200: result = await response.json() return result['choices'][0]['message']['content'] else: error_text = await response.text() raise Exception(f"Chutes API error: {response.status} - {error_text}") def _chutes_chat_sync(self, messages: List[Dict], temperature: float = 0.7, max_tokens: int = 1024) -> str: """ Synchronous wrapper for Chutes API chat """ try: loop = asyncio.get_event_loop() except RuntimeError: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) return loop.run_until_complete( self._chutes_chat_async(messages, temperature, max_tokens) ) def chat(self, message: str, include_domains: List[str] = None, exclude_domains: List[str] = None, system_prompt: str = None, temperature: float = 0.7, max_tokens: int = 1024) -> Dict: """ Send a message to the AI and get a response Args: message: User's message include_domains: List of domains to include for web search (Groq only) exclude_domains: List of domains to exclude from web search (Groq only) system_prompt: Custom system prompt temperature: Model temperature (0.0-2.0) max_tokens: Maximum tokens in response Returns: AI response with metadata """ # Enhanced system prompt for better citation behavior if not system_prompt: if self.provider == "groq": citation_instruction = """ IMPORTANT: When you search the web and find information, you MUST: 1. Always cite your sources with clickable links in this format: [Source Title](URL) 2. Include multiple diverse sources when possible 3. Show which specific websites you used for each claim 4. At the end of your response, provide a "Sources Used" section with all the links 5. Be transparent about which information comes from which source """ domain_context = "" if include_domains: domain_context = f"\nYou are restricted to searching ONLY these domains: {', '.join(include_domains)}. Make sure to find and cite sources specifically from these domains." elif exclude_domains: domain_context = f"\nAvoid searching these domains: {', '.join(exclude_domains)}. Search everywhere else on the web." system_prompt = f"""You are a creative and intelligent AI assistant with agentic capabilities. You can search the web, analyze information, and provide comprehensive responses. Be helpful, creative, and engaging while maintaining accuracy. {citation_instruction} {domain_context} Your responses should be well-structured, informative, and properly cited with working links.""" else: # Simpler system prompt for Chutes (no web search capabilities) system_prompt = """You are a creative and intelligent AI assistant. Be helpful, creative, and engaging while maintaining accuracy. Your responses should be well-structured, informative, and comprehensive.""" # Build messages messages = [{"role": "system", "content": system_prompt}] # Add conversation history (last 10 exchanges) messages.extend(self.conversation_history[-20:]) # Last 10 user-assistant pairs # Add current message with domain filtering context (Groq only) enhanced_message = message if self.provider == "groq" and (include_domains or exclude_domains): filter_context = [] if include_domains: filter_context.append(f"ONLY search these domains: {', '.join(include_domains)}") if exclude_domains: filter_context.append(f"EXCLUDE these domains: {', '.join(exclude_domains)}") enhanced_message += f"\n\n[Domain Filtering: {' | '.join(filter_context)}]" messages.append({"role": "user", "content": enhanced_message}) try: if self.provider == "groq": return self._handle_groq_chat(messages, include_domains, exclude_domains, temperature, max_tokens, message) elif self.provider == "chutes": return self._handle_chutes_chat(messages, temperature, max_tokens, message) else: raise ValueError(f"Unknown provider: {self.provider}") except Exception as e: error_msg = f"Error: {str(e)}" self.conversation_history.append({"role": "user", "content": message}) self.conversation_history.append({"role": "assistant", "content": error_msg}) return { "content": error_msg, "timestamp": datetime.now().isoformat(), "model": self.model, "provider": self.provider, "tool_usage": None, "error": str(e) } def _handle_groq_chat(self, messages: List[Dict], include_domains: List[str], exclude_domains: List[str], temperature: float, max_tokens: int, original_message: str) -> Dict: """Handle Groq API chat""" # Set up API parameters params = { "messages": messages, "model": self.model, "temperature": temperature, "max_tokens": max_tokens } # Add domain filtering if specified if include_domains and include_domains[0].strip(): params["include_domains"] = [domain.strip() for domain in include_domains if domain.strip()] if exclude_domains and exclude_domains[0].strip(): params["exclude_domains"] = [domain.strip() for domain in exclude_domains if domain.strip()] # Make the API call response = self.groq_client.chat.completions.create(**params) content = response.choices[0].message.content # Extract tool usage information and enhance it tool_info = self._extract_tool_info(response) # Process content to enhance citations processed_content = self._enhance_citations(content, tool_info) # Add to conversation history self.conversation_history.append({"role": "user", "content": original_message}) self.conversation_history.append({"role": "assistant", "content": processed_content}) # Create response object return { "content": processed_content, "timestamp": datetime.now().isoformat(), "model": self.model, "provider": "groq", "tool_usage": tool_info, "parameters": { "temperature": temperature, "max_tokens": max_tokens, "include_domains": include_domains, "exclude_domains": exclude_domains } } def _handle_chutes_chat(self, messages: List[Dict], temperature: float, max_tokens: int, original_message: str) -> Dict: """Handle Chutes API chat""" content = self._chutes_chat_sync(messages, temperature, max_tokens) # Add to conversation history self.conversation_history.append({"role": "user", "content": original_message}) self.conversation_history.append({"role": "assistant", "content": content}) # Create response object return { "content": content, "timestamp": datetime.now().isoformat(), "model": self.model, "provider": "chutes", "tool_usage": None, # Chutes doesn't have tool usage info "parameters": { "temperature": temperature, "max_tokens": max_tokens } } def _extract_tool_info(self, response) -> Dict: """Extract tool usage information in a JSON serializable format (Groq only)""" tool_info = { "tools_used": [], "search_queries": [], "sources_found": [] } if hasattr(response.choices[0].message, 'executed_tools'): tools = response.choices[0].message.executed_tools if tools: for tool in tools: tool_dict = { "tool_type": getattr(tool, "type", "unknown"), "tool_name": getattr(tool, "name", "unknown"), } # Extract search queries and results if hasattr(tool, "input"): tool_input = str(tool.input) tool_dict["input"] = tool_input # Try to extract search query if "search" in tool_dict["tool_name"].lower(): tool_info["search_queries"].append(tool_input) if hasattr(tool, "output"): tool_output = str(tool.output) tool_dict["output"] = tool_output # Try to extract URLs from output urls = self._extract_urls(tool_output) tool_info["sources_found"].extend(urls) tool_info["tools_used"].append(tool_dict) return tool_info def _extract_urls(self, text: str) -> List[str]: """Extract URLs from text""" url_pattern = r'https?://[^\s<>"]{2,}' urls = re.findall(url_pattern, text) return list(set(urls)) # Remove duplicates def _enhance_citations(self, content: str, tool_info: Dict) -> str: """Enhance content with better citation formatting (Groq only)""" if not tool_info or not tool_info.get("sources_found"): return content # Add sources section if not already present if "Sources Used:" not in content and "sources:" not in content.lower(): sources_section = "\n\n---\n\n### 📚 Sources Used:\n" for i, url in enumerate(tool_info["sources_found"][:10], 1): # Limit to 10 sources # Try to extract domain name for better formatting domain = self._extract_domain(url) sources_section += f"{i}. [{domain}]({url})\n" content += sources_section return content def _extract_domain(self, url: str) -> str: """Extract domain name from URL for display""" try: if url.startswith(('http://', 'https://')): domain = url.split('/')[2] # Remove www. prefix if present if domain.startswith('www.'): domain = domain[4:] return domain return url except: return url def clear_history(self): """Clear conversation history""" self.conversation_history = [] def get_history_summary(self) -> str: """Get a summary of conversation history""" if not self.conversation_history: return "No conversation history" user_messages = [msg for msg in self.conversation_history if msg["role"] == "user"] assistant_messages = [msg for msg in self.conversation_history if msg["role"] == "assistant"] return f"Conversation: {len(user_messages)} user messages, {len(assistant_messages)} assistant responses" # Global variables ai_instance = None current_provider = "groq" api_key_status = {"groq": "Not Set", "chutes": "Not Set"} def validate_api_keys(groq_api_key: str, chutes_api_key: str, provider: str, model: str) -> str: """Validate API keys and initialize AI instance""" global ai_instance, current_provider, api_key_status current_provider = provider if provider == "groq": if not groq_api_key or len(groq_api_key.strip()) < 10: api_key_status["groq"] = "Invalid ❌" return "❌ Please enter a valid Groq API key (should be longer than 10 characters)" try: # Test the Groq API key client = Groq(api_key=groq_api_key) test_response = client.chat.completions.create( messages=[{"role": "user", "content": "Hello"}], model=model, max_tokens=10 ) # Create AI instance ai_instance = CreativeAgenticAI(groq_api_key=groq_api_key, provider="groq", model=model) api_key_status["groq"] = "Valid ✅" return f"✅ Groq API Key Valid! Creative Agentic AI is ready.\n\n**Provider:** Groq\n**Model:** {model}\n**Status:** Connected with web search capabilities!" except Exception as e: api_key_status["groq"] = "Invalid ❌" ai_instance = None return f"❌ Error validating Groq API key: {str(e)}\n\nPlease check your API key and try again." elif provider == "chutes": if not chutes_api_key or len(chutes_api_key.strip()) < 10: api_key_status["chutes"] = "Invalid ❌" return "❌ Please enter a valid Chutes API key (should be longer than 10 characters)" try: # Test the Chutes API key with a simple request test_ai = CreativeAgenticAI(chutes_api_key=chutes_api_key, provider="chutes", model=model) test_response = test_ai._chutes_chat_sync( [{"role": "user", "content": "Hello"}], temperature=0.7, max_tokens=10 ) # Create AI instance ai_instance = CreativeAgenticAI(chutes_api_key=chutes_api_key, provider="chutes", model=model) api_key_status["chutes"] = "Valid ✅" return f"✅ Chutes API Key Valid! Creative AI is ready.\n\n**Provider:** Chutes\n**Model:** {model}\n**Status:** Connected (text generation focused)!" except Exception as e: api_key_status["chutes"] = "Invalid ❌" ai_instance = None return f"❌ Error validating Chutes API key: {str(e)}\n\nPlease check your API key and try again." def get_available_models(provider: str) -> List[str]: """Get available models for the selected provider""" if provider == "groq": return ["compound-beta", "compound-beta-mini"] elif provider == "chutes": return ["openai/gpt-oss-20b", "meta-llama/llama-3.1-8b-instruct", "anthropic/claude-3-sonnet"] return [] def update_model_choices(provider: str): """Update model choices based on provider selection""" models = get_available_models(provider) return gr.Radio(choices=models, value=models[0] if models else None, label=f"🧠 {provider.title()} Models") def chat_with_ai(message: str, include_domains: str, exclude_domains: str, system_prompt: str, temperature: float, max_tokens: int, history: List) -> tuple: """Main chat function""" global ai_instance, current_provider if not ai_instance: error_msg = f"⚠️ Please set your {current_provider.title()} API key first!" history.append([message, error_msg]) return history, "" if not message.strip(): return history, "" # Process domain lists (only for Groq) include_list = None exclude_list = None if current_provider == "groq": include_list = [d.strip() for d in include_domains.split(",")] if include_domains.strip() else [] exclude_list = [d.strip() for d in exclude_domains.split(",")] if exclude_domains.strip() else [] try: # Get AI response response = ai_instance.chat( message=message, include_domains=include_list if include_list else None, exclude_domains=exclude_list if exclude_list else None, system_prompt=system_prompt if system_prompt.strip() else None, temperature=temperature, max_tokens=int(max_tokens) ) # Format response ai_response = response["content"] # Add enhanced tool usage info (Groq only) if response.get("tool_usage") and current_provider == "groq": tool_info = response["tool_usage"] tool_summary = [] if tool_info.get("search_queries"): tool_summary.append(f"🔍 Search queries: {len(tool_info['search_queries'])}") if tool_info.get("sources_found"): tool_summary.append(f"📄 Sources found: {len(tool_info['sources_found'])}") if tool_info.get("tools_used"): tool_summary.append(f"🔧 Tools used: {len(tool_info['tools_used'])}") if tool_summary: ai_response += f"\n\n*{' | '.join(tool_summary)}*" # Add domain filtering info (Groq only) if current_provider == "groq" and (include_list or exclude_list): filter_info = [] if include_list: filter_info.append(f"✅ Included domains: {', '.join(include_list)}") if exclude_list: filter_info.append(f"❌ Excluded domains: {', '.join(exclude_list)}") ai_response += f"\n\n*🌐 Domain filtering applied: {' | '.join(filter_info)}*" # Add provider info ai_response += f"\n\n*🤖 Powered by: {current_provider.title()} ({response.get('model', 'unknown')})*" # Add to history history.append([message, ai_response]) return history, "" except Exception as e: error_msg = f"❌ Error: {str(e)}" history.append([message, error_msg]) return history, "" def clear_chat_history(): """Clear the chat history""" global ai_instance if ai_instance: ai_instance.clear_history() return [] def create_gradio_app(): """Create the main Gradio application""" # Custom CSS for better styling css = """ .container { max-width: 1200px; margin: 0 auto; } .header { text-align: center; background: linear-gradient(to right, #00ff94, #00b4db); color: white; padding: 20px; border-radius: 10px; margin-bottom: 20px; } .status-box { background-color: #f8f9fa; border: 1px solid #dee2e6; border-radius: 8px; padding: 15px; margin: 10px 0; } .example-box { background-color: #e8f4fd; border-left: 4px solid #007bff; padding: 15px; margin: 10px 0; border-radius: 0 8px 8px 0; } .domain-info { background-color: #fff3cd; border: 1px solid #ffeaa7; border-radius: 8px; padding: 15px; margin: 10px 0; } .citation-info { background-color: #d1ecf1; border: 1px solid #bee5eb; border-radius: 8px; padding: 15px; margin: 10px 0; } .provider-info { background-color: #f8d7da; border: 1px solid #f5c6cb; border-radius: 8px; padding: 15px; margin: 10px 0; } #neuroscope-accordion { background: linear-gradient(to right, #00ff94, #00b4db); border-radius: 8px; } """ with gr.Blocks(css=css, title="🤖 Multi-Provider Creative Agentic AI Chat", theme=gr.themes.Ocean()) as app: # Header gr.HTML("""
Multi-Provider AI Chat Tool - Powered by Groq's Compound Models & Chutes API
Note: Domain filtering only works with Groq's compound models that have web search capabilities.
Control which websites the AI can search when answering your questions:
New: Domain filtering status will be shown in responses!
This enhanced version now supports both Groq and Chutes AI providers:
When using Groq, you get:
When using Chutes, you get access to:
Try the same prompts with both providers to see the difference: