A Coding Implementation to Construct an Superior Internet Intelligence Agent with Tavily and Gemini AI

import os
import json
import asyncio
from typing import Record, Dict, Any
from dataclasses import dataclass
from wealthy.console import Console
from wealthy.progress import monitor
from wealthy.panel import Panel
from wealthy.markdown import Markdown

from langchain_tavily import TavilyExtract
from langchain.chat_models import init_chat_model
from langgraph.prebuilt import create_react_agent

@dataclass
class WebIntelligence:
    """Internet Intelligence Configuration"""
    tavily_key: str = os.getenv("TAVILY_API_KEY", "")
    google_key: str = os.getenv("GOOGLE_API_KEY", "")
    extract_depth: str = "superior"
    max_urls: int = 10

@dataclass
class WebIntelligence:
    """Internet Intelligence Configuration"""
    tavily_key: str = os.getenv("TAVILY_API_KEY", "")
    google_key: str = os.getenv("GOOGLE_API_KEY", "")
    extract_depth: str = "superior"
    max_urls: int = 10
The WebIntelligence dataclass serves as a structured configuration container, holding API keys for Tavily and Google Gemini, and setting extraction parameters like extract_depth and the utmost variety of URLs (max_urls). It simplifies the administration and entry of essential settings, making certain seamless integration and customization of net content material extraction duties throughout the intelligence agent.

class SmartWebAgent:
    """Clever Internet Content material Extraction & Evaluation Agent"""
   
    def __init__(self, config: WebIntelligence):
        self.config = config
        self.console = Console()
        self._setup_environment()
        self._initialize_tools()
   
    def _setup_environment(self):
        """Setup API keys with interactive prompts"""
        if not self.config.tavily_key:
            self.config.tavily_key = enter("🔑 Enter Tavily API Key: ")
            os.environ["TAVILY_API_KEY"] = self.config.tavily_key
           
        if not self.config.google_key:
            self.config.google_key = enter("🔑 Enter Google Gemini API Key: ")
            os.environ["GOOGLE_API_KEY"] = self.config.google_key
   
    def _initialize_tools(self):
        """Initialize AI instruments and brokers"""
        self.console.print("🛠️  Initializing AI Instruments...", fashion="daring blue")
       
        attempt:
            self.extractor = TavilyExtract(
                extract_depth=self.config.extract_depth,
                include_images=False,  
                include_raw_content=False,
                max_results=3
            )
           
            self.llm = init_chat_model(
                "gemini-2.0-flash",
                model_provider="google_genai",
                temperature=0.3,
                max_tokens=1024
            )
           
            test_response = self.llm.invoke("Say 'AI instruments initialized efficiently!'")
            self.console.print(f"✅ LLM Check: {test_response.content material}", fashion="inexperienced")
           
            self.agent = create_react_agent(self.llm, [self.extractor])
           
            self.console.print("✅ AI Agent Prepared!", fashion="daring inexperienced")
           
        besides Exception as e:
            self.console.print(f"❌ Initialization Error: {e}", fashion="daring purple")
            self.console.print("💡 Verify your API keys and web connection", fashion="yellow")
            increase
   
    def extract_content(self, urls: Record[str]) -> Dict[str, Any]:
        """Extract and construction content material from URLs"""
        outcomes = {}
       
        for url in monitor(urls, description="🌐 Extracting content material..."):
            attempt:
                response = self.extractor.invoke({"urls": [url]})
                content material = json.masses(response.content material) if isinstance(response.content material, str) else response.content material
                outcomes[url] = {
                    "standing": "success",
                    "knowledge": content material,
                    "abstract": content material.get("abstract", "No abstract out there")[:200] + "..."
                }
            besides Exception as e:
                outcomes[url] = {"standing": "error", "error": str(e)}
       
        return outcomes
   
    def analyze_with_ai(self, question: str, urls: Record[str] = None) -> str:
        """Clever evaluation utilizing AI agent"""
        attempt:
            if urls:
                message = f"Use the tavily_extract instrument to research these URLs and reply: {question}nURLs: {urls}"
            else:
                message = question
               
            self.console.print(f"🤖 AI Evaluation: {question}", fashion="daring magenta")
           
            messages = [{"role": "user", "content": message}]
           
            all_content = []
            with self.console.standing("🔄 AI considering..."):
                attempt:
                    for step in self.agent.stream({"messages": messages}, stream_mode="values"):
                        if "messages" in step and step["messages"]:
                            for msg in step["messages"]:
                                if hasattr(msg, 'content material') and msg.content material and msg.content material not in all_content:
                                    all_content.append(str(msg.content material))
                besides Exception as stream_error:
                    self.console.print(f"⚠️ Stream error: {stream_error}", fashion="yellow")
           
            if not all_content:
                self.console.print("🔄 Attempting direct AI invocation...", fashion="yellow")
                attempt:
                    response = self.llm.invoke(message)
                    return str(response.content material) if hasattr(response, 'content material') else str(response)
                besides Exception as direct_error:
                    self.console.print(f"⚠️ Direct error: {direct_error}", fashion="yellow")
                   
                    if urls:
                        self.console.print("🔄 Extracting content material first...", fashion="blue")
                        extracted = self.extract_content(urls)
                        content_summary = "n".be a part of([
                            f"URL: {url}nContent: {result.get('summary', 'No content')}n"
                            for url, result in extracted.items() if result.get('status') == 'success'
                        ])
                       
                        fallback_query = f"Based mostly on this content material, {question}:nn{content_summary}"
                        response = self.llm.invoke(fallback_query)
                        return str(response.content material) if hasattr(response, 'content material') else str(response)
           
            return "n".be a part of(all_content) if all_content else "❌ Unable to generate response. Please verify your API keys and check out once more."
           
        besides Exception as e:
            return f"❌ Evaluation failed: {str(e)}nnTip: Be certain that your API keys are legitimate and you've got web connectivity."
   
    def display_results(self, outcomes: Dict[str, Any]):
        """Stunning consequence show"""
        for url, lead to outcomes.objects():
            if consequence["status"] == "success":
                panel = Panel(
                    f"🔗 [bold blue]{url}[/bold blue]nn{consequence['summary']}",
                    title="✅ Extracted Content material",
                    border_style="inexperienced"
                )
            else:
                panel = Panel(
                    f"🔗 [bold red]{url}[/bold red]nn❌ Error: {consequence['error']}",
                    title="❌ Extraction Failed",
                    border_style="purple"
                )
            self.console.print(panel)

def run_async_safely(coro):
    """Run async perform safely in any setting"""
    attempt:
        loop = asyncio.get_running_loop()
        import nest_asyncio
        nest_asyncio.apply()
        return asyncio.run(coro)
    besides RuntimeError:
        return asyncio.run(coro)
    besides ImportError:
        print("⚠️  Working in sync mode. Set up nest_asyncio for higher efficiency.")
        return None

def primary():
    """Interactive Internet Intelligence Demo"""
    console = Console()
    console.print(Panel("🚀 Internet Intelligence Agent", fashion="daring cyan", subtitle="Powered by Tavily & Gemini"))
   
    config = WebIntelligence()
    agent = SmartWebAgent(config)
   
    demo_urls = [
        "https://en.wikipedia.org/wiki/Artificial_intelligence",
        "https://en.wikipedia.org/wiki/Machine_learning",
        "https://en.wikipedia.org/wiki/Quantum_computing"
    ]
   
    whereas True:
        console.print("n" + "="*60)
        console.print("🎯 Select an choice:", fashion="daring yellow")
        console.print("1. Extract content material from URLs")
        console.print("2. AI-powered evaluation")
        console.print("3. Demo with pattern URLs")
        console.print("4. Exit")
       
        alternative = enter("nEnter alternative (1-4): ").strip()
       
        if alternative == "1":
            urls_input = enter("Enter URLs (comma-separated): ")
            urls = [url.strip() for url in urls_input.split(",")]
            outcomes = agent.extract_content(urls)
            agent.display_results(outcomes)
           
        elif alternative == "2":
            question = enter("Enter your evaluation question: ")
            urls_input = enter("Enter URLs to research (elective, comma-separated): ")
            urls = [url.strip() for url in urls_input.split(",") if url.strip()] if urls_input.strip() else None
           
            attempt:
                response = agent.analyze_with_ai(question, urls)
                console.print(Panel(Markdown(response), title="🤖 AI Evaluation", border_style="blue"))
            besides Exception as e:
                console.print(f"❌ Evaluation failed: {e}", fashion="daring purple")
           
        elif alternative == "3":
            console.print("🎬 Working demo with AI & Quantum Computing URLs...")
            outcomes = agent.extract_content(demo_urls)
            agent.display_results(outcomes)
           
            response = agent.analyze_with_ai(
                "Evaluate AI, ML, and Quantum Computing. What are the important thing relationships?",
                demo_urls
            )
            console.print(Panel(Markdown(response), title="🧠 Comparative Evaluation", border_style="magenta"))
           
        elif alternative == "4":
            console.print("👋 Goodbye!", fashion="daring inexperienced")
            break
        else:
            console.print("❌ Invalid alternative!", fashion="daring purple")


if __name__ == "__main__":
    primary()