Coding implementation for constructing superior internet intelligence brokers utilizing 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 observe
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, guaranteeing seamless integration and customization of internet content material extraction duties inside 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...", model="daring blue")
       
        strive:
            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}", model="inexperienced")
           
            self.agent = create_react_agent(self.llm, [self.extractor])
           
            self.console.print("✅ AI Agent Prepared!", model="daring inexperienced")
           
        besides Exception as e:
            self.console.print(f"❌ Initialization Error: {e}", model="daring crimson")
            self.console.print("💡 Verify your API keys and web connection", model="yellow")
            increase
   
    def extract_content(self, urls: Record[str]) -> Dict[str, Any]:
        """Extract and construction content material from URLs"""
        outcomes = {}
       
        for url in observe(urls, description="🌐 Extracting content material..."):
            strive:
                response = self.extractor.invoke({"urls": [url]})
                content material = json.hundreds(response.content material) if isinstance(response.content material, str) else response.content material
                outcomes[url] = {
                    "standing": "success",
                    "information": 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"""
        strive:
            if urls:
                message = f"Use the tavily_extract device to investigate these URLs and reply: {question}nURLs: {urls}"
            else:
                message = question
               
            self.console.print(f"🤖 AI Evaluation: {question}", model="daring magenta")
           
            messages = [{"role": "user", "content": message}]
           
            all_content = []
            with self.console.standing("🔄 AI pondering..."):
                strive:
                    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}", model="yellow")
           
            if not all_content:
                self.console.print("🔄 Attempting direct AI invocation...", model="yellow")
                strive:
                    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}", model="yellow")
                   
                    if urls:
                        self.console.print("🔄 Extracting content material first...", model="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 take a look at once more."
           
        besides Exception as e:
            return f"❌ Evaluation failed: {str(e)}nnTip: Be sure your API keys are legitimate and you've got web connectivity."
   
    def display_results(self, outcomes: Dict[str, Any]):
        """Stunning end result show"""
        for url, end in outcomes.objects():
            if end result["status"] == "success":
                panel = Panel(
                    f"🔗 [bold blue]{url}[/bold blue]nn{end result['summary']}",
                    title="✅ Extracted Content material",
                    border_style="inexperienced"
                )
            else:
                panel = Panel(
                    f"🔗 [bold red]{url}[/bold red]nn❌ Error: {end result['error']}",
                    title="❌ Extraction Failed",
                    border_style="crimson"
                )
            self.console.print(panel)

def run_async_safely(coro):
    """Run async operate safely in any atmosphere"""
    strive:
        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 foremost():
    """Interactive Internet Intelligence Demo"""
    console = Console()
    console.print(Panel("🚀 Internet Intelligence Agent", model="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 possibility:", model="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")
       
        selection = enter("nEnter selection (1-4): ").strip()
       
        if selection == "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 selection == "2":
            question = enter("Enter your evaluation question: ")
            urls_input = enter("Enter URLs to investigate (elective, comma-separated): ")
            urls = [url.strip() for url in urls_input.split(",") if url.strip()] if urls_input.strip() else None
           
            strive:
                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}", model="daring crimson")
           
        elif selection == "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(
                "Examine 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 selection == "4":
            console.print("👋 Goodbye!", model="daring inexperienced")
            break
        else:
            console.print("❌ Invalid selection!", model="daring crimson")


if __name__ == "__main__":
    foremost()