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

import os
import json
import asyncio
from typing import Listing, 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:
    """Net 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:
    """Net 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 throughout the intelligence agent.

class SmartWebAgent:
    """Clever Net 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 Take a look at: {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 purple")
            self.console.print("💡 Test your API keys and web connection", model="yellow")
            elevate
   
    def extract_content(self, urls: Listing[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 accessible")[:200] + "..."
                }
            besides Exception as e:
                outcomes[url] = {"standing": "error", "error": str(e)}
       
        return outcomes
   
    def analyze_with_ai(self, question: str, urls: Listing[str] = None) -> str:
        """Clever evaluation utilizing AI agent"""
        strive:
            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}", 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("🔄 Making an attempt 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 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"Primarily based 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 part of(all_content) if all_content else "❌ Unable to generate response. Please test your API keys and check out once more."
           
        besides Exception as e:
            return f"❌ Evaluation failed: {str(e)}nnTip: Be certain your API keys are legitimate and you've got web connectivity."
   
    def display_results(self, outcomes: Dict[str, Any]):
        """Lovely 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 surroundings"""
    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 principal():
    """Interactive Net Intelligence Demo"""
    console = Console()
    console.print(Panel("🚀 Net 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 choice:", 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")
       
        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
           
            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 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!", model="daring inexperienced")
            break
        else:
            console.print("❌ Invalid alternative!", model="daring purple")


if __name__ == "__main__":
    principal()