smartwebagent_tavily_gemini_webintelligence_marktechpost2.py

# -*- coding: utf-8 -*-
"""SmartWebAgent_Tavily_Gemini_WebIntelligence_Marktechpost2.ipynb

Automatically generated by Colab.

Original file is located at
    https://colab.research.google.com/drive/1CrcBSqFMStW8yTr27dcKjOhK1eypqOL_
"""

import os
import json
import asyncio
from typing import List, Dict, Any
from dataclasses import dataclass
from rich.console import Console
from rich.progress import track
from rich.panel import Panel
from rich.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:
    """Web Intelligence Configuration"""
    tavily_key: str = os.getenv("TAVILY_API_KEY", "")
    google_key: str = os.getenv("GOOGLE_API_KEY", "")
    extract_depth: str = "advanced"
    max_urls: int = 10

class SmartWebAgent:
    """Intelligent Web Content Extraction & Analysis 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 = input("🔑 Enter Tavily API Key: ")
            os.environ["TAVILY_API_KEY"] = self.config.tavily_key

        if not self.config.google_key:
            self.config.google_key = input("🔑 Enter Google Gemini API Key: ")
            os.environ["GOOGLE_API_KEY"] = self.config.google_key

    def _initialize_tools(self):
        """Initialize AI tools and agents"""
        self.console.print("🛠️  Initializing AI Tools...", style="bold blue")

        try:
            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 tools initialized successfully!'")
            self.console.print(f"✅ LLM Test: {test_response.content}", style="green")

            self.agent = create_react_agent(self.llm, [self.extractor])

            self.console.print("✅ AI Agent Ready!", style="bold green")

        except Exception as e:
            self.console.print(f"❌ Initialization Error: {e}", style="bold red")
            self.console.print("💡 Check your API keys and internet connection", style="yellow")
            raise

    def extract_content(self, urls: List[str]) -> Dict[str, Any]:
        """Extract and structure content from URLs"""
        results = {}

        for url in track(urls, description="🌐 Extracting content..."):
            try:
                response = self.extractor.invoke({"urls": [url]})
                content = json.loads(response.content) if isinstance(response.content, str) else response.content
                results[url] = {
                    "status": "success",
                    "data": content,
                    "summary": content.get("summary", "No summary available")[:200] + "..."
                }
            except Exception as e:
                results[url] = {"status": "error", "error": str(e)}

        return results

    def analyze_with_ai(self, query: str, urls: List[str] = None) -> str:
        """Intelligent analysis using AI agent"""
        try:
            if urls:
                message = f"Use the tavily_extract tool to analyze these URLs and answer: {query}\nURLs: {urls}"
            else:
                message = query

            self.console.print(f"🤖 AI Analysis: {query}", style="bold magenta")

            messages = [{"role": "user", "content": message}]

            all_content = []
            with self.console.status("🔄 AI thinking..."):
                try:
                    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') and msg.content and msg.content not in all_content:
                                    all_content.append(str(msg.content))
                except Exception as stream_error:
                    self.console.print(f"⚠️ Stream error: {stream_error}", style="yellow")

            if not all_content:
                self.console.print("🔄 Trying direct AI invocation...", style="yellow")
                try:
                    response = self.llm.invoke(message)
                    return str(response.content) if hasattr(response, 'content') else str(response)
                except Exception as direct_error:
                    self.console.print(f"⚠️ Direct error: {direct_error}", style="yellow")

                    if urls:
                        self.console.print("🔄 Extracting content first...", style="blue")
                        extracted = self.extract_content(urls)
                        content_summary = "\n".join([
                            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 on this content, {query}:\n\n{content_summary}"
                        response = self.llm.invoke(fallback_query)
                        return str(response.content) if hasattr(response, 'content') else str(response)

            return "\n".join(all_content) if all_content else "❌ Unable to generate response. Please check your API keys and try again."

        except Exception as e:
            return f"❌ Analysis failed: {str(e)}\n\nTip: Make sure your API keys are valid and you have internet connectivity."

    def display_results(self, results: Dict[str, Any]):
        """Beautiful result display"""
        for url, result in results.items():
            if result["status"] == "success":
                panel = Panel(
                    f"🔗 [bold blue]{url}[/bold blue]\n\n{result['summary']}",
                    title="✅ Extracted Content",
                    border_style="green"
                )
            else:
                panel = Panel(
                    f"🔗 [bold red]{url}[/bold red]\n\n❌ Error: {result['error']}",
                    title="❌ Extraction Failed",
                    border_style="red"
                )
            self.console.print(panel)

def run_async_safely(coro):
    """Run async function safely in any environment"""
    try:
        loop = asyncio.get_running_loop()
        import nest_asyncio
        nest_asyncio.apply()
        return asyncio.run(coro)
    except RuntimeError:
        return asyncio.run(coro)
    except ImportError:
        print("⚠️  Running in sync mode. Install nest_asyncio for better performance.")
        return None

def main():
    """Interactive Web Intelligence Demo"""
    console = Console()
    console.print(Panel("🚀 Web Intelligence Agent", style="bold 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"
    ]

    while True:
        console.print("\n" + "="*60)
        console.print("🎯 Choose an option:", style="bold yellow")
        console.print("1. Extract content from URLs")
        console.print("2. AI-powered analysis")
        console.print("3. Demo with sample URLs")
        console.print("4. Exit")

        choice = input("\nEnter choice (1-4): ").strip()

        if choice == "1":
            urls_input = input("Enter URLs (comma-separated): ")
            urls = [url.strip() for url in urls_input.split(",")]
            results = agent.extract_content(urls)
            agent.display_results(results)

        elif choice == "2":
            query = input("Enter your analysis query: ")
            urls_input = input("Enter URLs to analyze (optional, comma-separated): ")
            urls = [url.strip() for url in urls_input.split(",") if url.strip()] if urls_input.strip() else None

            try:
                response = agent.analyze_with_ai(query, urls)
                console.print(Panel(Markdown(response), title="🤖 AI Analysis", border_style="blue"))
            except Exception as e:
                console.print(f"❌ Analysis failed: {e}", style="bold red")

        elif choice == "3":
            console.print("🎬 Running demo with AI & Quantum Computing URLs...")
            results = agent.extract_content(demo_urls)
            agent.display_results(results)

            response = agent.analyze_with_ai(
                "Compare AI, ML, and Quantum Computing. What are the key relationships?",
                demo_urls
            )
            console.print(Panel(Markdown(response), title="🧠 Comparative Analysis", border_style="magenta"))

        elif choice == "4":
            console.print("👋 Goodbye!", style="bold green")
            break
        else:
            console.print("❌ Invalid choice!", style="bold red")

if __name__ == "__main__":
    main()