from __future__ import annotations import argparse import json import sys from pathlib import Path from dotenv import load_dotenv load_dotenv() if sys.stdout and hasattr(sys.stdout, "reconfigure"): try: sys.stdout.reconfigure(encoding="utf-8") except Exception: pass from .components.chunking import ChunkingPipeline from .components.indexing import ( BM25Indexer, DocumentStore, IndexingPipeline, QdrantIndexer, SentenceTransformerEmbedder, ) from .components.ingestion.config import IngestionConfig from .components.ingestion.pipeline import IngestionPipeline as DocIngestionPipeline from .components.orchestration import RAGOrchestrator from .components.registry import SystemRegistry def get_system(): embedder = SystemRegistry.get_embedder() qdrant = SystemRegistry.get_qdrant() bm25 = SystemRegistry.get_bm25() doc_store = SystemRegistry.get_doc_store() chunker = ChunkingPipeline() indexing_pipe = IndexingPipeline( embedder=embedder, qdrant_indexer=qdrant, bm25_indexer=bm25, document_store=doc_store, chunking_pipeline=chunker, ) ingestion_pipe = DocIngestionPipeline(IngestionConfig()) orchestrator = RAGOrchestrator( embedder=embedder, qdrant_indexer=qdrant, bm25_indexer=bm25, document_store=doc_store, ) return ingestion_pipe, indexing_pipe, orchestrator, qdrant, bm25, doc_store def cmd_health(args): _, _, _, qdrant, bm25, doc_store = get_system() q_ok = qdrant.health_check() docs = doc_store.list_documents() status = { "status": "healthy" if q_ok else "degraded", "qdrant_connected": q_ok, "indexed_documents": len(docs), "qdrant_points": qdrant.count(), "bm25_chunks": bm25.count(), } print(json.dumps(status, indent=2)) def cmd_ingest(args): path = Path(args.path) if not path.exists(): print(f"Error: Path does not exist: {path}", file=sys.stderr) sys.exit(1) ingestion_pipe, indexing_pipe, _, qdrant, bm25, doc_store = get_system() if getattr(args, "rebuild_index", False): print("Rebuilding indexes (clearing Qdrant collection, BM25, and DocumentStore)...") qdrant.recreate_collection() bm25.clear() doc_store.clear() if path.is_file(): files = [path] else: files = [p for p in path.rglob("*") if p.is_file()] print(f"Ingesting {len(files)} files from {path}...") ingest_result = ingestion_pipe.ingest_many(files, strict=False) print(f"Parsed {len(ingest_result.documents)} documents. Indexing...") stats = indexing_pipe.index_documents( ingest_result.documents, force_reindex=args.force or getattr(args, "rebuild_index", False), ) print( f"Indexed {stats.documents_indexed} documents ({stats.chunks_indexed} chunks) " f"in {stats.duration_ms:.1f}ms. Skipped: {stats.documents_skipped}." ) if stats.errors: print(f"Errors encountered: {len(stats.errors)}") for err in stats.errors: print(f" - {err}") def cmd_search(args): _, _, orchestrator, _, bm25, _ = get_system() print(f"Searching for: '{args.query}' (strategy: {args.strategy})...") if args.strategy == "keyword": hits = bm25.search(args.query, top_k=args.top_k) for h in hits: print(f"[{h.rank}] score={h.score:.3f} | {h.source_locator} | {h.text[:120]}...") elif args.strategy == "dense": resp = orchestrator.dense_retriever.retrieve(args.query, top_k=args.top_k) for h in resp.results: print(f"[{h.rank}] score={h.score:.3f} | {h.source_locator} | {h.text[:120]}...") else: # Hybrid hits = bm25.search(args.query, top_k=args.top_k) for h in hits: print(f"[{h.rank}] score={h.score:.3f} | {h.source_locator} | {h.text[:120]}...") def cmd_query(args): _, _, orchestrator, _, _, _ = get_system() mode_str = f" [mode: {args.mode}]" if getattr(args, "mode", None) else "" print(f"Executing RAG query: '{args.query}'{mode_str}...") res = orchestrator.query(args.query, mode=getattr(args, "mode", None)) print("-" * 60) print(f"QUERY TYPE : {res.query_type}") print(f"EXEC MODE : {res.execution_mode}") print(f"CONFIDENCE : {res.confidence:.2f}") print(f"ABSTAINED : {res.abstained}") if res.abstained: print(f"REASON : {res.abstention_reason}") print(f"LATENCY : {res.latency_ms:.1f}ms") print("-" * 60) print("ANSWER:") print(res.answer) print("-" * 60) if res.sources: print(f"SOURCES ({len(res.sources)}):") for s in res.sources: print(f" [{s.get('citation_id')}] {s.get('filename')} (locator: {s.get('source_locator')})") print(f" \"{s.get('snippet', '')}...\"") def main(): parser = argparse.ArgumentParser( description="Nexus - Production-Grade RAG CLI" ) subparsers = parser.add_subparsers(dest="command", required=True) # health subparsers.add_parser("health", help="Check system and index health") # ingest p_ingest = subparsers.add_parser("ingest", help="Ingest and index documents") p_ingest.add_argument("path", help="Path to document file or directory") p_ingest.add_argument( "--force", action="store_true", help="Force reindexing existing documents" ) p_ingest.add_argument( "--rebuild-index", action="store_true", help="Rebuild Qdrant collection, BM25, and metadata matching active embedder dimension", ) # search p_search = subparsers.add_parser("search", help="Execute vector/keyword search") p_search.add_argument("query", help="Search query") p_search.add_argument( "--strategy", choices=["hybrid", "dense", "keyword"], default="hybrid" ) p_search.add_argument("--top-k", type=int, default=5) # query p_query = subparsers.add_parser("query", help="Execute grounded RAG query") p_query.add_argument("query", help="Natural language question") p_query.add_argument( "--mode", choices=["fast", "balanced", "deep"], default=None, help="Execution mode (fast, balanced, deep)" ) args = parser.parse_args() if args.command == "health": cmd_health(args) elif args.command == "ingest": cmd_ingest(args) elif args.command == "search": cmd_search(args) elif args.command == "query": cmd_query(args) if __name__ == "__main__": main()