File size: 4,718 Bytes
6b62834 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | """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()
|