"""Command-line interface for Ask-the-Web Agent.""" from __future__ import annotations import asyncio import sys from typing import Optional from rich.console import Console from rich.markdown import Markdown from rich.panel import Panel from rich.progress import Progress, SpinnerColumn, TextColumn from rich.prompt import Prompt from rich.table import Table console = Console() def print_banner() -> None: """Print the application banner.""" banner = """ ╔═══════════════════════════════════════════════════════════╗ ║ 🌐 Ask-the-Web Agent 🌐 ║ ║ AI-powered answers with real-time web search ║ ╚═══════════════════════════════════════════════════════════╝ """ console.print(banner, style="bold blue") def print_help() -> None: """Print help information.""" help_text = """ [bold]Available Commands:[/bold] [cyan]ask[/cyan] - Ask a question [cyan]search[/cyan] - Perform a web search only [cyan]history[/cyan] - Show conversation history [cyan]clear[/cyan] - Clear conversation history [cyan]config[/cyan] - Show current configuration [cyan]help[/cyan] - Show this help message [cyan]exit[/cyan] / [cyan]quit[/cyan] - Exit the application [bold]Tips:[/bold] - Ask specific questions for better results - Use follow-up questions to dive deeper - Sources are numbered for reference """ console.print(help_text) def print_sources(sources: list[dict]) -> None: """Print sources in a formatted table.""" if not sources: return table = Table(title="Sources", show_header=True, header_style="bold magenta") table.add_column("#", style="dim", width=3) table.add_column("Title", style="cyan", width=40) table.add_column("URL", style="blue", width=50) for i, source in enumerate(sources, 1): title = source.get("title", "Unknown")[:40] url = source.get("url", "")[:50] table.add_row(str(i), title, url) console.print(table) def print_follow_ups(follow_ups: list[str]) -> None: """Print follow-up question suggestions.""" if not follow_ups: return console.print("\n[bold yellow]Follow-up questions:[/bold yellow]") for i, question in enumerate(follow_ups, 1): console.print(f" {i}. {question}", style="italic") async def process_query(agent, question: str, history: list) -> None: """Process a user query and display results.""" with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), console=console, ) as progress: task = progress.add_task("Searching and analyzing...", total=None) try: response = await agent.query( question=question, history=history, enable_search=True, max_sources=5, ) progress.update(task, completed=True) except Exception as e: progress.update(task, completed=True) console.print(f"\n[red]Error: {e}[/red]") return # Display answer console.print("\n") console.print(Panel( Markdown(response.answer), title="Answer", border_style="green", )) # Display confidence confidence = response.confidence confidence_color = "green" if confidence > 0.7 else "yellow" if confidence > 0.4 else "red" console.print(f"\n[{confidence_color}]Confidence: {confidence:.0%}[/{confidence_color}]") # Display sources print_sources(response.sources) # Display follow-up questions print_follow_ups(response.follow_up_questions) # Update history history.extend([ {"role": "user", "content": question}, {"role": "assistant", "content": response.answer}, ]) async def async_main() -> None: """Async main function for the CLI.""" from src.agent.agent import AskTheWebAgent from src.utils.config import get_settings settings = get_settings() # Check configuration if not settings.openai_api_key and not settings.anthropic_api_key: console.print("[red]Error: No LLM API key configured.[/red]") console.print("Please set OPENAI_API_KEY or ANTHROPIC_API_KEY in your .env file.") sys.exit(1) print_banner() # Initialize agent try: agent = AskTheWebAgent() console.print("[green]✓ Agent initialized successfully[/green]\n") except Exception as e: console.print(f"[red]Failed to initialize agent: {e}[/red]") sys.exit(1) # Conversation history history: list[dict] = [] console.print("Type your question or 'help' for commands. Press Ctrl+C to exit.\n") while True: try: # Get user input user_input = Prompt.ask("[bold cyan]You[/bold cyan]") if not user_input.strip(): continue command = user_input.strip().lower() # Handle commands if command in ("exit", "quit", "q"): console.print("\n[yellow]Goodbye! 👋[/yellow]") break elif command == "help": print_help() elif command == "history": if not history: console.print("[dim]No conversation history yet.[/dim]") else: for msg in history: role = msg["role"] content = msg["content"][:100] + "..." if len(msg["content"]) > 100 else msg["content"] style = "cyan" if role == "user" else "green" console.print(f"[{style}]{role.title()}:[/{style}] {content}") elif command == "clear": history.clear() console.print("[green]Conversation history cleared.[/green]") elif command == "config": table = Table(title="Configuration", show_header=False) table.add_column("Setting", style="cyan") table.add_column("Value", style="white") table.add_row("LLM Provider", settings.llm_provider) table.add_row("LLM Model", settings.llm_model) table.add_row("Max Tokens", str(settings.max_tokens)) table.add_row("Search API", "Tavily" if settings.tavily_api_key else "Not configured") console.print(table) elif command.startswith("search "): query = user_input[7:].strip() if query: with console.status("Searching..."): from src.tools.web_search import WebSearchTool search = WebSearchTool() results = await search.execute(query=query) if results.success: console.print(f"\n[green]Found {len(results.data)} results:[/green]\n") for i, result in enumerate(results.data[:5], 1): console.print(f"[cyan]{i}. {result.get('title', 'No title')}[/cyan]") console.print(f" {result.get('url', '')}") console.print(f" [dim]{result.get('snippet', '')[:150]}...[/dim]\n") else: console.print(f"[red]Search failed: {results.error}[/red]") else: # Treat as a question await process_query(agent, user_input, history) console.print() except KeyboardInterrupt: console.print("\n\n[yellow]Goodbye! 👋[/yellow]") break except Exception as e: console.print(f"\n[red]Error: {e}[/red]") def main() -> None: """Main entry point for the CLI.""" asyncio.run(async_main()) def app() -> None: """Alias for main, used by pyproject.toml entry point.""" main() if __name__ == "__main__": main()