Skip to content

feat(cli): add ingest progress hint and --quiet/-q flag (with test coverage) #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
als3453 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

feat(cli): add ingest progress hint and --quiet/-q flag (with test coverage) #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f65dff3
feat(cli): add ingest progress output and --quiet/-q flag
als3453 Aug 19, 2026
f675ba3
test(cli): add tests for --quiet flag and ingest progress output
als3453 Aug 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
208 changes: 32 additions & 176 deletions src/rag_python/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,16 +1,12 @@
"""rag-python command-line interface."""
from __future__ import annotations

import argparse
import json
import sys
from dataclasses import replace

from . import __version__
from .client import RAG
from .help_text import CLI_EPILOG, list_topics, print_topic, print_topic_list


def _build_rag(args: argparse.Namespace) -> RAG:
kwargs: dict = dict(
llm_provider=args.llm_provider,
Expand All @@ -30,214 +26,77 @@ def _build_rag(args: argparse.Namespace) -> RAG:
if getattr(args, "metadata_filter", None):
kwargs["metadata_filter"] = args.metadata_filter
return RAG(**kwargs)


def _parse_metadata_filter(raw: str | None) -> dict | None:
if not raw:
return None
try:
return json.loads(raw)
except json.JSONDecodeError as e:
raise argparse.ArgumentTypeError(f"Invalid JSON for metadata filter: {e}") from e


def _add_provider_args(parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"--llm-provider",
default="openai",
choices=["openai", "azure_openai", "anthropic", "gemini", "ollama"],
metavar="PROVIDER",
help="LLM backend (default: openai). See: rag-python docs providers",
)
parser.add_argument(
"--llm-model",
default=None,
metavar="MODEL",
help="LLM model or Azure deployment name (default: from env LLM_MODEL)",
)
parser.add_argument(
"--embedding-provider",
default="openai",
choices=["openai", "azure_openai", "ollama", "local"],
metavar="PROVIDER",
help="Embedding backend (default: openai). Use local for offline embeddings",
)
parser.add_argument(
"--embedding-model",
default=None,
metavar="MODEL",
help="Embedding model name (default: from env EMBEDDING_MODEL)",
)
parser.add_argument(
"--ollama-base-url",
default=None,
metavar="URL",
help="Ollama server URL (default: http://localhost:11434 or OLLAMA_BASE_URL)",
)
parser.add_argument("--llm-provider", default="openai", choices=["openai", "azure_openai", "anthropic", "gemini", "ollama"], metavar="PROVIDER", help="LLM backend (default: openai). See: rag-python docs providers")
parser.add_argument("--llm-model", default=None, metavar="MODEL", help="LLM model or Azure deployment name (default: from env LLM_MODEL)")
parser.add_argument("--embedding-provider", default="openai", choices=["openai", "azure_openai", "ollama", "local"], metavar="PROVIDER", help="Embedding backend (default: openai). Use local for offline embeddings")
parser.add_argument("--embedding-model", default=None, metavar="MODEL", help="Embedding model name (default: from env EMBEDDING_MODEL)")
parser.add_argument("--ollama-base-url", default=None, metavar="URL", help="Ollama server URL (default: http://localhost:11434 or OLLAMA_BASE_URL)")
parser.add_argument("--azure-endpoint", default=None, help="Azure OpenAI endpoint URL")
parser.add_argument("--azure-api-key", default=None, help="Azure OpenAI API key")
parser.add_argument(
"--azure-api-version",
default=None,
help="Azure API version (default: 2023-09-01-preview)",
)
parser.add_argument("--azure-api-version", default=None, help="Azure API version (default: 2023-09-01-preview)")
parser.add_argument("--openai-api-key", default=None, help="OpenAI API key (overrides env)")
parser.add_argument("--anthropic-api-key", default=None, help="Anthropic API key")
parser.add_argument("--gemini-api-key", default=None, help="Gemini API key")


def _add_search_args(parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"--retriever",
choices=["vector", "multi_query", "hybrid"],
default=None,
metavar="MODE",
help=(
"Retrieval mode: vector (single query), multi_query (default, with rewriting), "
"or hybrid (BM25+vector; requires pip install rag-python[hybrid])"
),
)
parser.add_argument(
"--metadata-filter",
type=_parse_metadata_filter,
default=None,
metavar="JSON",
help='Filter chunks by metadata, e.g. \'{"filename": "policy.pdf"}\'',
)


parser.add_argument("--retriever", choices=["vector", "multi_query", "hybrid"], default=None, metavar="MODE", help="Retrieval mode: vector (single query), multi_query (default, with rewriting), or hybrid (BM25+vector; requires pip install rag-python[hybrid])")
parser.add_argument("--metadata-filter", type=_parse_metadata_filter, default=None, metavar="JSON", help='Filter chunks by metadata, e.g. {"filename": "policy.pdf"}')
def _make_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="rag-python",
description=(
"Production-grade RAG for Python — ingest documents, ask questions, "
"get grounded answers with multi-LLM support."
),
epilog=CLI_EPILOG,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser = argparse.ArgumentParser(prog="rag-python", description="Production-grade RAG for Python \u2014 ingest documents, ask questions, get grounded answers with multi-LLM support.", epilog=CLI_EPILOG, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
sub = parser.add_subparsers(dest="command", required=True, metavar="COMMAND")

ing = sub.add_parser(
"ingest",
help="Load files into the vector store (chunk + embed)",
description=(
"Ingest one or more files or directories into the ChromaDB vector store.\n"
"Supported formats: .txt .md .pdf .docx .csv .json .html"
),
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"examples:\n"
" rag-python ingest ./data --reindex\n"
" rag-python ingest policy.pdf handbook/ --embedding-provider local"
),
)
ing.add_argument(
"paths",
nargs="+",
metavar="PATH",
help="File or directory paths to ingest",
)
ing.add_argument(
"--reindex",
action="store_true",
help="Delete existing vectors before ingesting (fresh index)",
)
ing = sub.add_parser("ingest", help="Load files into the vector store (chunk + embed)", description="Ingest one or more files or directories into the ChromaDB vector store.\nSupported formats: .txt .md .pdf .docx .csv .json .html", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="examples:\n rag-python ingest ./data --reindex\n rag-python ingest policy.pdf handbook/ --embedding-provider local")
ing.add_argument("paths", nargs="+", metavar="PATH", help="File or directory paths to ingest")
ing.add_argument("--reindex", action="store_true", help="Delete existing vectors before ingesting (fresh index)")
ing.add_argument("-q", "--quiet", action="store_true", help="Suppress progress and summary output (silent mode)")
_add_provider_args(ing)

q = sub.add_parser(
"query",
help="Ask a question against ingested documents",
description=(
"Run the full RAG pipeline: retrieve relevant chunks, generate an answer, "
"optionally stream tokens and show sources."
),
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"examples:\n"
' rag-python query "How many days of annual leave?"\n'
" rag-python query \"PTO policy\" --stream -v\n"
' rag-python query "benefits" --retriever hybrid --metadata-filter \'{"filename": "hr.pdf"}\''
),
)
q.add_argument(
"question",
nargs="+",
metavar="QUESTION",
help="Question text (multiple words are joined)",
)
q.add_argument(
"--no-multi-query",
action="store_true",
help="Use single-query vector retrieval (same as --retriever vector)",
)
q.add_argument(
"--stream",
action="store_true",
help="Stream answer tokens to stdout as they are generated",
)
q.add_argument(
"-v",
"--verbose",
action="store_true",
help="After the answer, print evaluation scores and top source paths",
)
q = sub.add_parser("query", help="Ask a question against ingested documents", description="Run the full RAG pipeline: retrieve relevant chunks, generate an answer, optionally stream tokens and show sources.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog='examples:\n rag-python query "How many days of annual leave?"\n rag-python query "PTO policy" --stream -v\n rag-python query "benefits" --retriever hybrid --metadata-filter {"filename": "hr.pdf"}')
q.add_argument("question", nargs="+", metavar="QUESTION", help="Question text (multiple words are joined)")
q.add_argument("--no-multi-query", action="store_true", help="Use single-query vector retrieval (same as --retriever vector)")
q.add_argument("--stream", action="store_true", help="Stream answer tokens to stdout as they are generated")
q.add_argument("-v", "--verbose", action="store_true", help="After the answer, print evaluation scores and top source paths")
_add_provider_args(q)
_add_search_args(q)

docs = sub.add_parser(
"docs",
help="Show user documentation in the terminal",
description="Print built-in help topics. Full docs: https://github.com/RaghavOG/rag-python/tree/main/docs",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="topics: " + ", ".join(list_topics()),
)
docs.add_argument(
"topic",
nargs="?",
default="quickstart",
choices=list_topics(),
metavar="TOPIC",
help="Documentation topic (default: quickstart)",
)
docs.add_argument(
"--list",
action="store_true",
help="List all available documentation topics",
)

docs = sub.add_parser("docs", help="Show user documentation in the terminal", description="Print built-in help topics. Full docs: https://github.com/RaghavOG/rag-python/tree/main/docs", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="topics: " + ", ".join(list_topics()))
docs.add_argument("topic", nargs="?", default="quickstart", choices=list_topics(), metavar="TOPIC", help="Documentation topic (default: quickstart)")
docs.add_argument("--list", action="store_true", help="List all available documentation topics")
return parser


def main(argv: list[str] | None = None) -> None:
parser = _make_parser()
args = parser.parse_args(argv)

if args.command == "docs":
if args.list:
print_topic_list()
else:
print_topic(args.topic)
return

if args.command == "ingest":
quiet = args.quiet
paths = args.paths
if not quiet:
count = len(paths)
noun = "path" if count == 1 else "paths"
preview = ", ".join(paths[:3]) + (" ..." if count > 3 else "")
print(f"Ingesting {count} {noun}: {preview}")
rag = _build_rag(args)
n = rag.ingest(args.paths, reindex=args.reindex)
print(f"Ingested {n} chunks.")
n = rag.ingest(paths, reindex=args.reindex)
if not quiet:
print(f"Ingested {n} chunks.")
return

if args.command == "query":
rag = _build_rag(args)
question = " ".join(args.question)
retriever = args.retriever
if retriever is None and args.no_multi_query:
retriever = "vector"
search = replace(
rag.config.search,
retriever=retriever or rag.config.search.retriever,
metadata_filter=args.metadata_filter or rag.config.search.metadata_filter,
)
search = replace(rag.config.search, retriever=retriever or rag.config.search.retriever, metadata_filter=args.metadata_filter or rag.config.search.metadata_filter)
if args.stream:
stream = rag.query_stream(question, search=search)
for token in stream:
Expand All @@ -251,7 +110,6 @@ def main(argv: list[str] | None = None) -> None:
for s in result.sources[:5]:
print(s.get("metadata", {}).get("source", ""), "score:", s.get("score"))
return

ans = rag.query(question, search=search)
print(ans.text)
if args.verbose:
Expand All @@ -260,7 +118,5 @@ def main(argv: list[str] | None = None) -> None:
print("\n--- sources ---")
for s in ans.sources[:5]:
print(s.get("metadata", {}).get("source", ""), "score:", s.get("score"))


if __name__ == "__main__":
main(sys.argv[1:])
Loading