ragleap-rag

A fast, honest, self-hosted RAG engine. Hybrid dense+sparse retrieval, real streaming, automatic provider fallback, and actual token usage numbers — not estimates. Bring your own API keys; nothing is routed through us.

pip install ragleap-rag
# or
uv add ragleap-rag[gemini]

Quickstart

You'll need two things: a PostgreSQL database with the pgvector extension, and a free Gemini API key (used for embeddings).

from ragleap import RagLeap, ProviderConfig, EmbeddingConfig

rag = RagLeap(
    database_url="postgresql://user:pass@localhost/mydb",
    embedder=EmbeddingConfig(provider="gemini", api_key="your-gemini-key"),
    primary=ProviderConfig(provider="gemini", api_key="your-gemini-key"),
)
rag.init_schema()  # one-time, idempotent — safe to call every run

rag.ingest_text("handbook.txt", "Employees get unlimited PTO and a $500/year learning budget.")

answer = rag.ask("How much PTO do employees get?")
print(answer["answer"])

Employees get unlimited PTO. (Source 1)

That's the whole loop: ingest text (or a .txt/.pdf/.docx file via rag.ingest(filename, raw_bytes)), then ask questions grounded in it.

Vector backends

By default, RagLeap stores vectors in Postgres/pgvector - the database_url you already pass in. Conversation memory always uses this same Postgres connection regardless of which vector backend you choose, since memory (session history) is a separate concern from vector storage.

You can swap the vector backend via vector_backend=:

from ragleap import RagLeap, ProviderConfig, EmbeddingConfig
from ragleap.vectorstores import FAISSBackend

rag = RagLeap(
    database_url="postgresql://user:pass@localhost/mydb",  # still required, for conversation memory
    vector_backend=FAISSBackend(persist_directory="./my_faiss_data"),  # vectors go here instead
    embedder=EmbeddingConfig(provider="gemini", api_key="..."),
    primary=ProviderConfig(provider="gemini", api_key="..."),
)
rag.init_schema()

Currently available:

More backends (Pinecone, Weaviate, Qdrant, Milvus) are planned - see the project roadmap. Building your own is straightforward: implement the ragleap.vectorstores.VectorBackend abstract interface (init_schema, insert_document, insert_chunk, search_dense, list_documents, delete_document, get_document_filename, and optionally search_sparse/search_hybrid/supports_sparse if your backend can do keyword search).

The three things that matter

Retrieval is hybrid by default — dense (pgvector cosine similarity) and sparse (Postgres full-text search) results are combined via Reciprocal Rank Fusion, so both semantic matches and exact keyword/ identifier matches get found. Pass hybrid=False to rag.ask(...) for dense-only retrieval (cheaper — one query instead of two).

Generation accepts temperature, system_prompt, and max_tokens as real per-call arguments — build your own agent behavior on top of retrieval without forking the library:

answer = rag.ask(
    "Summarize the handbook",
    temperature=0.1,
    system_prompt="Answer in exactly one sentence.",
    max_tokens=100,
)

Reliability — configure a fallback chain so a rate limit, outage, or bad key on your primary provider doesn't mean a failed request:

rag = RagLeap(
    database_url="...",
    embedder=EmbeddingConfig(provider="gemini", api_key="..."),
    primary=ProviderConfig(provider="gemini", api_key="..."),
    fallbacks=[ProviderConfig(provider="groq", api_key="...", model="llama-3.3-70b-versatile")],
)

Every ask() response tells you which provider actually answered (answer["provider_used"]) and exactly how many tokens it cost (answer["usage"]) — real numbers pulled from the provider's own response, not an estimate.

Streaming

for piece in rag.ask_stream("What SDKs are supported?"):
    print(piece, end="", flush=True)

Real per-provider streaming — Gemini, Anthropic, and any OpenAI-compatible endpoint each have different streaming APIs; all three are implemented properly, not stubbed.

ask_stream() supports rerank= and metadata_filter=, the same as ask() - these were missing from the streaming variant until 0.5.7, a genuine API inconsistency now fixed.

Async support

async equivalents exist for every method that touches the database or an API: aingest, aingest_text, aask, aask_stream. Use these inside an async web server (FastAPI, etc.) so a slow embedding call or LLM response does not block the event loop.

result = await rag.aingest_text("handbook.txt", "...")
answer = await rag.aask("How much PTO do employees get?")
async for piece in rag.aask_stream("What SDKs are supported?"):
    print(piece, end="", flush=True)

Honest note: these wrap the existing, tested sync implementation in a worker thread (asyncio.to_thread) rather than using natively async database/HTTP clients end to end. This still avoids blocking the event loop and works correctly under concurrent load - confirmed via a live test running 3 aask() calls concurrently - but it is not the same as a from-scratch async rewrite using asyncpg and async HTTP clients throughout. A fully native async implementation may follow in a future release if there is real demand for it.

Conversation memory

Pass session_id to ask() or ask_stream() to get persistent, multi-turn memory. Prior turns in that session are automatically injected as context. Omit it and every call is fully stateless, exactly as before (no breaking change).

session = "support-chat-42"

rag.ask("What is the CEO name", session_id=session)
rag.ask("What country is he based in", session_id=session)

Memory is Postgres-backed (its own conversations/conversation_messages tables, created by init_schema()). It survives restarts and works across processes, not just in-memory for a single script run.

rag.get_history(session)
rag.clear_session(session)

By default the last 10 messages are included per call (max_history_messages, no token-aware trimming yet).

Reranking

Pass rerank=True to ask() for cross-encoder reranking. The initial hybrid search retrieves a wider candidate pool, then a cross-encoder scores each (query, chunk) pair jointly, reordering results by genuine relevance rather than the initial retrieval score alone. Off by default (extra latency, extra dependency).

answer = rag.ask("What is the exact pricing?", rerank=True)

Requires the rerank extra:

pip install ragleap-rag[rerank]

The reranker runs on ONNX Runtime (CPU only) rather than torch/sentence-transformers - a quantized cross-encoder model (~23MB) downloads once on the first rerank=True call and is cached locally by huggingface_hub, not at RagLeap construction time. This avoids the 2GB+ torch+CUDA install that sentence-transformers pulled in even for CPU-only use in versions before 0.5.8.

Not currently available on ask_stream().

Document lifecycle

list_documents(), delete_document(), and update_document() manage previously ingested content.

docs = rag.list_documents(limit=20)
for d in docs:
    print(d["filename"], d["chunk_count"], "chunks")

rag.delete_document(docs[0]["document_id"])

# update_document is delete + re-ingest under the hood - the document
# gets a new document_id, old chunks and embeddings are not preserved
result = rag.update_document(some_document_id, "new content here")

Metadata & filtering

Attach arbitrary JSON metadata to a document at ingest time, then restrict retrieval to matching chunks with metadata_filter on ask() - the basis for multi-tenant isolation, date-range filtering, or any other tagging scheme.

rag.ingest_text("acme_handbook.txt", "...", metadata={"tenant": "acme"})
rag.ingest_text("globex_handbook.txt", "...", metadata={"tenant": "globex"})

# Only retrieves chunks whose metadata contains {"tenant": "acme"}
answer = rag.ask("What is our PTO policy?", metadata_filter={"tenant": "acme"})

Filtering uses Postgres JSONB containment (metadata @> filter), backed by a GIN index - so it scales, and metadata can hold any JSON-serializable structure, not just flat tenant IDs. metadata is also returned by list_documents().

Known limitation: update_document() does not currently preserve the original document's metadata - re-ingesting content via update_document() resets metadata to empty unless you pass it again yourself. Worth fixing in a follow-up if this trips anyone up.

Content sanitization

ingest_text() sanitizes and screens content by default (sanitize=True, warn_on_injection_risk=True) - both can be disabled per call if you are already sanitizing upstream.

# Default: sanitizes control characters and warns on suspicious patterns
rag.ingest_text("doc.txt", text)

# Opt out if you handle this yourself
rag.ingest_text("doc.txt", text, sanitize=False, warn_on_injection_risk=False)

Sanitization strips null bytes, control characters, and invisible/zero-width Unicode characters - a documented technique for hiding instructions inside text that looks normal to a human reviewer.

Injection-risk detection is heuristic pattern matching against a fixed list of common trigger phrases ("ignore previous instructions", "reveal your system prompt", etc). It logs a warning and does NOT block ingestion. Honest limitation: this is pattern matching, not semantic understanding - prompt injection via retrieved content is an open research problem, and a sufficiently motivated attacker can rephrase around any fixed pattern list. Treat a warning as a signal to review, not a guarantee of safety, and treat the absence of a warning as "nothing matched", not "this content is safe."

Guardrails

Beyond the built-in sanitization above, input_guardrails=/output_guardrails= let you plug in your own validation callbacks - a PII filter, a profanity check, a brand-voice check, whatever your use case needs. Each guardrail is a function (text: str) -> str that returns the (possibly modified) text, or raises GuardrailViolation to reject it outright.

from ragleap import RagLeap, ProviderConfig, EmbeddingConfig
from ragleap.guardrails import GuardrailViolation

def reject_pii(text: str) -> str:
    if "ssn:" in text.lower():
        raise GuardrailViolation("Document appears to contain an SSN")
    return text

def enforce_brand_voice(text: str) -> str:
    return text.replace("gonna", "going to")

rag = RagLeap(
    database_url="...",
    embedder=EmbeddingConfig(...),
    primary=ProviderConfig(...),
    input_guardrails=[reject_pii],       # runs during ingest_text(), after sanitization
    output_guardrails=[enforce_brand_voice],  # runs on ask()/ask_stream()'s answer
)

On ask(), a raised GuardrailViolation replaces the answer with a refusal message and sets answer["guardrail_blocked"] = True (the key is only present when guardrails are configured). On ingest_text(), a violation aborts ingestion entirely - nothing is stored.

Honest limitation for ask_stream(): output guardrails can only run after the full answer is assembled, but by then individual tokens have already been yielded to the caller - streaming can't retroactively un-send content. A violation during streaming is logged as a warning, not enforced. If blocking bad output before the user sees any of it matters for your use case, use ask() instead of ask_stream().

Evaluation

rag.evaluate(test_cases) runs a labeled test set through ask() and reports deterministic quality signals - it is not an LLM-as-judge framework (no faithfulness/relevancy scoring like Ragas). It measures three things that don't require an LLM call themselves:

• Retrieval hit rate - did the expected document actually show up in sources?
• Keyword coverage - what fraction of expected keywords appear in the generated answer?
• Citation groundedness - of the keywords found in the answer, how many also appear in the chunks the answer actually cited? A low score here is a real (if heuristic) hallucination signal.

result = rag.evaluate([
    {"query": "What's the refund policy?", "expected_document": "policy.pdf", "expected_keywords": ["30 days", "receipt"]},
    {"query": "How do I reset my password?", "expected_document": "faq.pdf", "expected_keywords": ["settings", "email link"]},
])

print(result["retrieval_hit_rate"])       # 1.0
print(result["keyword_coverage_rate"])    # 0.75
print(result["groundedness_rate"])        # 0.9
print(result["results"][0])               # per-case detail: query, answer, sources, hits

Any extra keyword arguments (top_k=, rerank=, hybrid=, metadata_filter=, etc.) are passed through to every ask() call the evaluation makes. This is a fast, free, repeatable sanity check for catching regressions in your own retrieval/generation setup - not a substitute for human review, and not a claim of measuring "truthfulness." A full LLM-as-judge evaluation framework is planned as a separate, dedicated tool (see the project roadmap).

Observability hooks

on_ingest=/on_query=/on_answer= are lightweight, fire-and-forget event emission points - not a dashboard or storage layer, just an instrumentation seam for logging, metrics, tracing, or a future observability tool to plug into. Each is a list of callables (event: dict) -> None.

import logging
logger = logging.getLogger("ragleap.events")

rag = RagLeap(
    database_url="...",
    embedder=EmbeddingConfig(...),
    primary=ProviderConfig(...),
    on_ingest=[lambda e: logger.info(f"Ingested {e['filename']}: {e['chunks_stored']} chunks")],
    on_query=[lambda e: logger.info(f"Query: {e['query']!r} (hybrid={e['hybrid']}, streaming={e['streaming']})")],
    on_answer=[lambda e: logger.info(f"Answered via {e.get('provider_used')}, usage={e.get('usage')}")],
)

A hook that raises an exception is caught, logged as a warning, and swallowed - it never breaks the actual ingest/ask call. This is fire-and-forget, not a delivery guarantee: if a hook is slow or fails, that's on the hook, not on your RAG pipeline.

on_answer fires on both ask() (with provider_used, usage, chunks_sent, guardrail_blocked) and ask_stream() (with answer_length, since usage isn't available for streaming - see the Streaming section). Every event includes streaming: bool so one handler can distinguish the two if needed.

Citations

Every ask() response includes a citations field - a structured, chunk-level breakdown that resolves a real ambiguity: a citation like "(Source 1)" in an answer could mean a whole document or one specific passage within it. It always means the latter.

answer = rag.ask("What is our refund policy?")
print(answer["answer"])

for c in answer["citations"]:
    print(c["source_number"], c["document_name"], "chunk", c["chunk_index"], "-", c["text_preview"])

Each citation includes source_number (matching the [Source N] label the model was given in its prompt), document_name, document_id, chunk_id, chunk_index, and a text_preview of that specific chunk - enough to verify exactly which passage backs a claim, which matters for audit or compliance use cases. The existing sources field (a deduped list of document names) is unchanged for backward compatibility.

URL ingestion

ingest_url() fetches a web page and extracts clean, readable text - stripping navigation, ads, and other boilerplate via trafilatura - rather than ingesting raw HTML markup. Requires the web extra.

pip install ragleap-rag[web]

result = rag.ingest_url("https://example.com/blog/some-article")
answer = rag.ask("What does the article say about X?")

The URL itself is stored as the document_name, so citations point back to the original page. Metadata works the same as ingest_text() - pass metadata= to tag the ingested content.

Supported file formats

rag.ingest(filename, raw_bytes) supports 28 formats via file extension, dispatched automatically. Core formats (txt, pdf, docx, md) work with the base install; everything else requires the formats extra.

pip install ragleap-rag[formats]

Office & documents: pdf, docx, pptx, odt, ods, odp, rtf Spreadsheets & tabular: xlsx, xls, csv, tsv, parquet Structured data: json, yaml, xml, xsl, xslt Markup & web: html, htm, md Archives & email: zip (recurses into supported files inside), eml Books & media metadata: epub, vtt, srt (subtitle cue numbers and timestamps are stripped, spoken text kept) Plain text: txt, sql

Not supported: legacy binary .doc and .ppt (pre-2007 Office formats) - no reliable pure-Python parser exists for these. Convert to the modern equivalent first, e.g. via LibreOffice headless: soffice --headless --convert-to docx yourfile.doc

with open("report.xlsx", "rb") as f:
    rag.ingest("report.xlsx", f.read())

Image ingestion

ingest_image(filename, raw_bytes, mode=...) supports two different techniques for two different kinds of images.

mode="ocr" (default) reads literal visible text - scanned documents, screenshots, photos of text. Requires the ocr extra AND the Tesseract binary installed on the system (not pip-installable - e.g. apt install tesseract-ocr on Debian/Ubuntu).

mode="caption" describes an image's contents using a vision-capable model instead - for photos, diagrams, or charts with no readable text. Currently requires Gemini configured as the primary or a fallback provider; no extra install needed since it reuses the existing generation client.

pip install ragleap-rag[ocr]

# Scanned document or screenshot
rag.ingest_image("receipt.png", raw_bytes, mode="ocr")

# Photo, chart, or diagram with no text
rag.ingest_image("product_photo.jpg", raw_bytes, mode="caption")

Audio ingestion

ingest_audio(filename, raw_bytes, transcriber=None) transcribes audio and ingests the result. Defaults to OpenAI's hosted Whisper API using OPENAI_API_KEY from the environment; pass a TranscriptionConfig to choose a different provider or add options.

from ragleap import TranscriptionConfig

# Default: Whisper via OpenAI
rag.ingest_audio("meeting.mp3", raw_bytes)

# Explicit config, with a vocabulary hint and language
config = TranscriptionConfig(provider="whisper", language="en", prompt="RagLeap, pgvector, Gemini")
rag.ingest_audio("meeting.mp3", raw_bytes, transcriber=config)

# Deepgram instead
config = TranscriptionConfig(provider="deepgram", api_key="...")
rag.ingest_audio("meeting.mp3", raw_bytes, transcriber=config)

Honest limitation: transcription quality is only as good as the underlying provider. Whisper (the default) is a strong general-purpose baseline, but has no built-in denoising - quiet or noisy audio genuinely degrades accuracy - and no domain-vocabulary biasing by default, so brand names and jargon commonly get mangled unless you pass a prompt hint. Accuracy also varies meaningfully by language. Use provider="deepgram" or pass a prompt hint if these matter for your use case.

Both providers use hosted APIs - no local model weights, no torch/CUDA dependency, consistent with keeping the base install light (the same reasoning behind reranking's optional [rerank] extra). Local/offline Whisper is not currently supported.

Not limited to Whisper and Deepgram - provider="custom" accepts any transcription function you supply, so you can use AssemblyAI, Speechmatics, Azure Speech, AWS Transcribe, or anything else without waiting on ragleap-rag to add native support.

def my_transcriber(filename: str, audio_bytes: bytes) -> str:
    # call whatever provider/SDK you prefer
    return transcript_text

config = TranscriptionConfig(provider="custom", transcribe_fn=my_transcriber)
rag.ingest_audio("call.mp3", raw_bytes, transcriber=config)

Verified live: a real Deepgram API call against synthesized speech correctly transcribed the audio and produced an accurate, grounded answer referencing what was actually said. Whisper's API shape was verified via a mocked call (no OpenAI key was available in this session), confirming the correct request structure without a live network round-trip.

Video ingestion

ingest_video(filename, raw_bytes, transcriber=None) extracts the audio track from a video file (via ffmpeg) and transcribes it - the same transcriber= options and honest limitations as ingest_audio() apply, since this is audio ingestion plus an extraction step, not separate video-specific logic. Requires the ffmpeg binary installed on the system (not pip-installable - e.g. apt install ffmpeg on Debian/Ubuntu).

rag.ingest_video("webinar.mp4", raw_bytes)

If the video already has a matching subtitle file (.vtt/.srt), ingesting that directly via ingest() is cheaper and more accurate than re-transcribing the audio - see Supported file formats.

Verification note: both the ffmpeg audio-extraction step and the transcription step are now fully verified live. ffprobe independently confirmed a real 3-second test video is extracted to a valid, playable audio stream of the correct duration. Separately, a real Deepgram API call against synthesized speech (via espeak) correctly transcribed the audio and produced an accurate, grounded answer referencing what was actually said - closing the gap noted in Audio ingestion, where live-provider testing was initially unavailable.

Batch ingestion

rag.ingest_batch(items) ingests a list of mixed-type items concurrently, and returns a per-item result rather than raising on the first failure - one bad item never blocks or rolls back the others.

results = await rag.ingest_batch([
    {"type": "file", "filename": "notes.txt", "raw_bytes": b"..."},
    {"type": "url", "url": "https://example.com/article"},
    {"type": "image", "filename": "scan.png", "raw_bytes": b"...", "mode": "ocr"},
    {"type": "audio", "filename": "call.mp3", "raw_bytes": b"..."},
])

for r in results:
    if r["success"]:
        print(r["result"])
    else:
        print("failed:", r["error"])

Each item dict needs a "type" key ("file", "url", "image", "audio", or "video") plus whatever kwargs that type's ingest_* method normally needs. Verification note: tested with a real 4-item mixed batch (a .txt file, a live URL fetch, a deliberately corrupt fake PDF, and a .json file) - 3 of 4 succeeded exactly as expected, and the corrupt PDF failed cleanly with a clear logged error rather than crashing the batch or silently affecting the other results.

Performance

Database connections are pooled internally (min 1, max 10 by default) rather than opened fresh on every call. Previously every ingest, ask, and memory operation opened a brand-new Postgres connection and closed it afterward - real, avoidable latency, especially under concurrent load (e.g. a web server handling multiple requests at once). This is automatic and requires no configuration.

Query embeddings are also cached in memory (LRU, 1000 entries by default) - repeated identical questions skip a redundant embedding call. This caches embeddings only, never full answers, since with conversation memory the same question can legitimately produce different answers depending on session history. Check cache effectiveness with rag.cache_stats(), or disable with cache_enabled=False.

For multi-process deployments (multiple Gunicorn or Celery workers), the in-memory cache doesn't help across processes - each worker has its own separate cache, so a hit in one worker is invisible to the others. Setting cache_backend="redis" (with the [redis] extra and a redis_url) shares the query embedding cache across all worker processes via Redis, with a configurable cache_ttl_seconds (default 86400 = 24h).

rag = RagLeap(
    database_url="...",
    embedder=EmbeddingConfig(...),
    primary=ProviderConfig(...),
    cache_backend="redis",
    redis_url="redis://localhost:6379/0",
    cache_ttl_seconds=86400,
)

Verification note: proven with two separate RagLeap instances (simulating two separate worker processes) pointed at the same Redis - the first instance's cache miss and resulting embedding write was correctly picked up as a cache hit on the second, completely separate instance's first call, confirming the embedding is genuinely shared across process boundaries and not just cached within one Python object.

Celery integration

Running ingestion or asking as background tasks (so a web request doesn't block on an LLM call) is a common pattern. The one thing that matters: RagLeap's connection pool is per-instance and not fork-safe across processes. Create one global RagLeap object before Celery's default "prefork" pool forks workers, and every worker inherits the same pool and file descriptors - this shows up later as hung queries or connection errors under concurrent load, not as an obvious startup crash.

The fix: each worker process builds its own RagLeap instance, once, after it forks - not at module import time.

from celery import Celery
from celery.signals import worker_process_init
from ragleap import RagLeap, ProviderConfig, EmbeddingConfig

app = Celery("ragleap_tasks", broker="redis://localhost:6379/0", backend="redis://localhost:6379/0")

_rag_instance = None

def get_rag() -> RagLeap:
    global _rag_instance
    if _rag_instance is None:
        _rag_instance = RagLeap(
            database_url="postgresql://...",
            embedder=EmbeddingConfig(provider="gemini", api_key="..."),
            primary=ProviderConfig(provider="gemini", api_key="..."),
            # Same Redis server as the broker above is fine - a different
            # db index keeps the query cache and the task queue from colliding.
            cache_backend="redis",
            redis_url="redis://localhost:6379/1",
        )
        _rag_instance.init_schema()
    return _rag_instance

@worker_process_init.connect
def init_worker(**kwargs):
    get_rag()  # built right after fork, not before

@app.task(name="ragleap.ingest_text")
def ingest_text_task(filename: str, text: str, metadata: dict | None = None):
    result = get_rag().ingest_text(filename=filename, text=text, metadata=metadata)
    return {"document_id": result.document_id, "chunks_stored": result.chunks_stored}

@app.task(name="ragleap.ask")
def ask_task(query: str, session_id: str | None = None):
    answer = get_rag().ask(query, session_id=session_id)
    return {"answer": answer["answer"], "sources": answer["sources"]}

Architecture:

                    +-----------+
  Web request  ---> |   Redis   | ---> worker process 1 (own RagLeap + pool)
  (non-blocking)    |  broker   | ---> worker process 2 (own RagLeap + pool)
                    +-----------+                |
                                                  v
                                          +---------------+
                                          |   PostgreSQL   |
                                          |   + pgvector   |
                                          +---------------+
                                                  ^
                                                  |
                                   (optional) Redis query cache,
                                   shared across all worker processes

Each worker process talks to the same Postgres database and, optionally, shares the Redis query cache (a separate concern from the Celery broker/backend, even if it's the same Redis server). Run celery -A your_module worker --loglevel=info to start a worker, then call .delay(...) from your web app to enqueue tasks without blocking the request.

See examples/05_celery_background_tasks.py for the full runnable version.

How it fits together

         +------------------+
         |   Your text or   |
         |  .txt/.pdf/.docx |
         +--------+---------+
                  |
         +--------v---------+
         |  rag.ingest(...)  |   chunk -> embed -> store
         +--------+---------+
                  |
         +--------v---------+
         |  PostgreSQL +     |
         |  pgvector         |
         +--------+---------+
                  |
         +--------v---------+
         |   rag.ask(...)    |   hybrid retrieve (dense + sparse, RRF)
         +--------+---------+          |
                  |                     v
         +--------v---------+   +---------------+
         |   Generation      |-->| Fallback chain |
         |  (temp/prompt/    |   | (if primary    |
         |   max_tokens)     |   |  fails)        |
         +--------+---------+   +---------------+
                  |
         +--------v---------+
         |  Conversation     |   optional: session_id ->
         |  memory (Postgres)|   prior turns injected as context
         +-------------------+

Supported LLM providers

Gemini, Anthropic, and any OpenAI-compatible endpoint: OpenAI, Groq, Mistral, Together, OpenRouter, Ollama, DeepSeek, xAI, Cohere, Perplexity, or a custom endpoint (provider="custom" + base_url=...). Install extras as needed: pip install ragleap-rag[anthropic], [openai], or [all].

More examples

See examples/ in the source repo:

• 01_basic_ingest_and_ask.py — the loop above, runnable as-is
• 02_streaming.py — streaming responses
• 03_fallback_and_hybrid_search.py — provider fallback + hybrid toggle
• 04_flask_web_api.py — drop-in web API (works identically in FastAPI)
• 05_celery_background_tasks.py — background ingestion/asking via Celery + Redis

Why this exists

Most RAG libraries give you a toolkit and leave production concerns (retrieval quality, provider reliability, cost visibility) as an exercise for you. ragleap-rag treats hybrid search, fallback, and real token usage reporting as defaults, not add-ons — because a RAG engine that silently fails on a rate limit, or that you can't verify the actual cost of, isn't production-ready no matter how good its retrieval is.

ragleap-rag is the foundation layer of ragleap-core, a larger open-source, self-hosted AI platform (channels, knowledge graph, language detection, business integrations). Companion packages (ragleap-graph, ragleap-integrations) are in progress.

Status

Young, actively developed. Verified end-to-end: built, published to PyPI, and independently confirmed working via pip, uv, and Google Colab, in a genuinely separate environment from the development machine.

License

MIT