agentic_rag/entrypoints/cli/main.py

"""CLI entry point for Agentic RAG."""

import asyncio
import sys
from pathlib import Path

import typer
from rich.console import Console
from rich.markdown import Markdown
from rich.panel import Panel

app = typer.Typer(help="Agentic RAG — Multi-modal ReAct-powered RAG System")
console = Console()


@app.command()
def chat(
    message: str = typer.Argument(..., help="Your message/question"),
    mode: str = typer.Option("auto", help="Agent mode: auto, chat, rag, research, media"),
    stream: bool = typer.Option(False, "--stream", "-s", help="Stream the response"),
    provider: str = typer.Option("", help="LLM provider to use"),
):
    """Send a message to the agent."""
    from agentic_rag.services.llm.factory import get_llm
    from agentic_rag.orchestration.l1_tools.registry import get_tool_registry
    from agentic_rag.agent.router import AgentRouter
    from agentic_rag.data.models import AgentInput

    async def _run():
        llm = get_llm(provider) if provider else get_llm()
        tool_registry = get_tool_registry()
        _register_tools(tool_registry)

        router = AgentRouter(llm, tool_registry)
        engine = await router.route(query=message, preferred_mode=mode if mode != "auto" else None)

        if stream:
            input_data = AgentInput(query=message)
            async for event in engine.stream(input_data):
                if event.event_type.value == "text_delta":
                    console.print(event.data.get("content", ""), end="")
                elif event.event_type.value == "tool_call_start":
                    console.print(f"\n[dim]🔧 {event.data['tool']}...[/dim]")
                elif event.event_type.value == "tool_call_result":
                    status = "✓" if event.data.get("success") else "✗"
                    console.print(f"[dim]   {status} Done[/dim]")
            console.print()
        else:
            with console.status("[bold green]Thinking..."):
                input_data = AgentInput(query=message)
                output = await engine.run(input_data)

            console.print(Panel(Markdown(output.final_answer), title="Answer"))
            if output.tool_calls_made:
                console.print(f"[dim]Tools used: {len(output.tool_calls_made)}, Iterations: {output.iterations}[/dim]")

    asyncio.run(_run())


@app.command()
def serve(
    host: str = typer.Option("0.0.0.0", help="Host to bind"),
    port: int = typer.Option(8000, help="Port to bind"),
    reload: bool = typer.Option(False, help="Enable auto-reload"),
):
    """Start the API server."""
    import uvicorn
    console.print(f"[bold green]Starting Agentic RAG server on {host}:{port}[/bold green]")
    uvicorn.run(
        "agentic_rag.entrypoints.rest.app:app",
        host=host,
        port=port,
        reload=reload,
        log_level="info",
    )


@app.command()
def ingest(
    file: str = typer.Option(..., "--file", "-f", help="File to ingest"),
    source: str = typer.Option("cli", help="Source identifier"),
):
    """Ingest a document into the knowledge base."""
    from agentic_rag.orchestration.l1_tools.rag_tools import RAGIngestTool

    async def _run():
        path = Path(file)
        if not path.exists():
            console.print(f"[red]File not found: {file}[/red]")
            sys.exit(1)

        content = path.read_text()
        tool = RAGIngestTool()
        result = await tool.execute(content=content, source=source)
        console.print(f"[green]{result}[/green]")

    asyncio.run(_run())


@app.command()
def info():
    """Show system information."""
    from agentic_rag import __version__
    from agentic_rag.config.settings import get_settings

    settings = get_settings()

    console.print(Panel(f"Agentic RAG v{__version__}", title="System Info"))
    console.print(f"Default LLM Provider: {settings.default_provider}")
    for name, cfg in settings.llm_providers.items():
        console.print(f"  {name}: {cfg.model} @ {cfg.api_base}")
    console.print(f"Milvus: {settings.milvus.host}:{settings.milvus.port}")
    console.print(f"Embedding: {settings.embedding.model} (dim={settings.embedding.dim})")


def _register_tools(registry):
    """Register built-in tools."""
    if registry.tool_count == 0:
        from agentic_rag.orchestration.l1_tools.rag_tools import RAGSearchTool
        from agentic_rag.orchestration.l1_tools.web_tools import WebFetchTool, WebSearchTool
        from agentic_rag.orchestration.l1_tools.code_tools import CodeExecuteTool

        registry.register(RAGSearchTool())
        registry.register(WebSearchTool())
        registry.register(WebFetchTool())
        registry.register(CodeExecuteTool())


if __name__ == "__main__":
    app()