Spaces:
Running
Running
| """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] <question> - Ask a question | |
| [cyan]search[/cyan] <query> - 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() | |