import gradio as gr import os import time import requests from bs4 import BeautifulSoup from urllib.parse import urljoin, urlparse from groq import Groq import re import json # --- Constants and API Setup --- # **Environment Variable Validation** def validate_env_var(var_name, env_var): if not env_var: raise ValueError(f"{var_name} environment variable is not set.") return env_var CEREBRAS_API_KEY = validate_env_var("CEREBRAS_API_KEY", os.getenv("CEREBRAS_API_KEY")) GROQ_API_KEY = validate_env_var("GROQ_API_KEY", os.getenv("GROQ_API_KEY")) client_cerebras = Cerebras(api_key=CEREBRAS_API_KEY) client_groq = Groq(api_key=GROQ_API_KEY) # --- Model Rate Limit Info --- # **Formatted as a Dictionary for Easy Access** MODEL_INFO = { "Chat Completion": { "gemma-7b-it": {"requests_per_minute": 30, "tokens_per_minute": 15000}, # Add more models here... }, "Speech to Text": { "distil-whisper-large-v3-en": {"requests_per_minute": 20, "audio_seconds_per_hour": 7200}, # Add more models here... } } def get_model_info(): """Returns formatted model info as a string""" output = "" for category, models in MODEL_INFO.items(): output += f"**{category}**\n" for model, limits in models.items(): output += f"* {model}: {limits}\n" return output # --- Helper Functions --- def is_valid_url(url): """Checks if a URL is valid""" try: result = urlparse(url) return all([result.scheme, result.netloc]) except ValueError: return False def fetch_webpage(url): """Fetches a webpage with a 10-second timeout""" try: response = requests.get(url, timeout=10) response.raise_for_status() return response.text except requests.exceptions.RequestException as e: return f"Error fetching URL: {e}" def extract_text_from_html(html): """Extracts text from HTML using BeautifulSoup""" soup = BeautifulSoup(html, 'html.parser') text = soup.get_text(separator=' ', strip=True) return text # --- Chat Logic with Groq --- async def chat_with_groq(user_input, chat_history): """Handles user input and returns AI response, chain of thought, compute time, and token usage""" start_time = time.time() try: # **Simplified History Formatting** formatted_history = "\n\n".join([f"User: {msg[0]}\nAI: {msg[1]}" for msg in chat_history[-10:]]) messages = [ {"role": "system", "content": f""" You are IntellijMind, a highly advanced and proactive AI agent. Available tools: scrape with a URL, and search_internet with a query. Current conversation: {formatted_history} """}, {"role": "user", "content": user_input} ] if user_input.lower() == "model info": response = get_model_info() return response, "", f"Compute Time: {time.time() - start_time:.2f} seconds", f"Tokens used: {len(user_input.split()) + len(response.split())}" completion = client_groq.chat.completions.create( model="llama3-groq-70b-8192-tool-use-preview", messages=messages, temperature=1, max_tokens=2048, top_p=1, stream=True, stop=None, ) response = "" chain_of_thought = "" tool_execution_count = 0 for chunk in completion: if chunk.choices[0].delta and chunk.choices[0].delta.content: content = chunk.choices[0].delta.content response += content # **Simplified Chain of Thought Extraction** if "Chain of Thought:" in content: chain_of_thought += content.split("Chain of Thought:", 1)[-1].strip() # **Simplified Tool Execution** if "Action:" in content: action_match = re.search(r"Action: (\w+), Parameters: (\{.*\})", content) if action_match and tool_execution_count < 3: tool_execution_count += 1 action, parameters = action_match.groups() parameters = json.loads(parameters) if action == "take_action": if parameters.get("action") == "scrape": # **Simplified Scrape Action** url = parameters.get("url") if is_valid_url(url): html_content = fetch_webpage(url) if not html_content.startswith("Error"): webpage_text = extract_text_from_html(html_content) response += f"\nWebpage Content: {webpage_text}\n" else: response += f"\nError scraping webpage: {html_content}\n" else: response += "\nInvalid URL provided.\n" elif parameters.get("action") == "search_internet": # **Simplified Search Action** query = parameters.get("query") response += f"\nSearch query: {query}. Note: Search is simulated in this environment. Results may vary.\n" response += f"\nSearch Results: Mock Results for query: {query}\n" compute_time = time.time() - start_time token_usage = len(user_input.split()) + len(response.split()) return response, chain_of_thought, f"Compute Time: {compute_time:.2f} seconds", f"Tokens used: {token_usage}" except Exception as e: return "Error: Unable to process your request.", "", str(e), "" # --- Gradio Interface --- def gradio_ui(): with gr.Blocks() as demo: gr.Markdown("""# 🚀 IntellijMind: The Autonomous AI Agent\nExperience the forefront of AI capabilities, where the agent proactively achieves your goals!""") with gr.Row(): with gr.Column(scale=6): chat_history = gr.Chatbot(label="Chat History") with gr.Column(scale=2): compute_time = gr.Textbox(label="Compute Time", interactive=False) chain_of_thought_display = gr.Textbox(label="Chain of Thought", interactive=False, lines=10) token_usage_display = gr.Textbox(label="Token Usage", interactive=False) user_input = gr.Textbox(label="Type your message", placeholder="Ask me anything...", lines=2) with gr.Row(): send_button = gr.Button("Send", variant="primary") clear_button = gr.Button("Clear Chat") export_button = gr.Button("Export Chat History") async def handle_chat(chat_history, user_input): if not user_input.strip(): return chat_history, "", "", "", "Please enter a valid message." ai_response, chain_of_thought, compute_info, token_usage = await chat_with_groq(user_input, chat_history) chat_history.append((user_input, ai_response)) return chat_history, chain_of_thought, compute_info, token_usage def clear_chat(): return [], "", "", "" def export_chat(chat_history): if not chat_history: return "", "No chat history to export." chat_text = "\n".join([f"User: {item[0]}\nAI: {item[1]}" for item in chat_history]) filename = f"chat_history_{int(time.time())}.txt" with open(filename, "w") as file: file.write(chat_text) return f"Chat history exported to {filename}.", "" send_button.click(handle_chat, inputs=[chat_history, user_input], outputs=[chat_history, chain_of_thought_display, compute_time, token_usage_display]) clear_button.click(clear_chat, outputs=[chat_history, chain_of_thought_display, compute_time, token_usage_display]) export_button.click(export_chat, inputs=[chat_history], outputs=[compute_time, chain_of_thought_display]) user_input.submit(handle_chat, inputs=[chat_history, user_input], outputs=[chat_history, chain_of_thought_display, compute_time, token_usage_display]) gr.Markdown("""---\n### 🌟 Features:\n- **Autonomous Agent**: Proactively pursues your goals.\n- **Advanced Tool Use**: Utilizes multiple tools like web scraping and search.\n- **Dynamic and Creative**: Engages with humor and creative responses.\n- **Enhanced Chat History**: Maintains better context of the conversation.\n- **Real-Time Performance Metrics**: Measure response compute time instantly.\n- **Token Usage Tracking**: Monitor token usage per response for transparency.\n- **Export Chat History**: Save your conversation as a text file for future reference.\n- **User-Friendly Design**: Intuitive chatbot interface with powerful features.\n- **Insightful Chain of Thought**: See the reasoning process behind AI decisions.\n- **Submit on Enter**: Seamless interaction with keyboard support.\n""") return demo # Run the Gradio app if __name__ == "__main__": demo = gradio_ui() demo.launch()