v0.5.0
Local-first observability, debugging and evaluation for RAG systems. The MLflow for RAG.
Unlike general LLM observability tools, RAGObserve focuses on the retrieval lifecycle:
documents → chunking → embedding → indexing → retrieval → fusion
→ reranking → context assembly → generation → grounding
Framework-agnostic. Provider-agnostic. Vector-DB-agnostic. Zero required config — defaults to a local SQLite file inside ./.ragobserve/ (like .git). Scale up to Postgres or any cloud storage when you're ready.
pip install ragobserve # core (SQLite, dashboard, all adapters)
pip install ragobserve[langchain] # + LangChain auto-instrumentation
pip install ragobserve[llamaindex] # + LlamaIndex auto-instrumentation
pip install ragobserve[postgres] # + PostgreSQL backend
pip install ragobserve[files] # + FileStore (S3, GCS, Azure, Drive, local)
pip install ragobserve[files] s3fs # FileStore → Amazon S3
pip install ragobserve[files] gcsfs # FileStore → Google Cloud Storage
pip install ragobserve[files] adlfs # FileStore → Azure Blob / ADLS
pip install ragobserve[files] gdrivefs # FileStore → Google Driveimport ragobserve
ragobserve.init(project="contract-rag")
# or: ragobserve.init(project="contract-rag", tracking_uri="http://localhost:5601")
with ragobserve.trace("query", query=question):
ragobserve.log_retrieval(question, results, retriever="qdrant", duration_ms=23)
ragobserve.log_rerank(before, after, model="bge-reranker")
ragobserve.log_context(final_prompt, system_prompt=sys, chunks=top_chunks, context_window=8192)
ragobserve.log_generation(model="gpt-4o", prompt=final_prompt, response=answer, cost=0.002)Async pipelines work identically — async with ragobserve.trace(...) and all log_* functions are safe to call from async code without blocking the event loop.
Then explore:
ragobserve ui # http://127.0.0.1:5601?key=<key>
ragobserve export --project my-rag --output traces.ndjson
ragobserve eval --project my-rag --api-key gsk_...
ragobserve providers
ragobserve versionOr start the dashboard from Python:
ragobserve.serve() # same as `ragobserve ui`
ragobserve.serve(port=8080) # custom portNo. Everything is created automatically.
| Backend | What happens on init() |
|---|---|
| SQLiteStore (default) | .ragobserve/ragobserve.db is created with the full schema |
| PostgresStore | all tables are created via CREATE TABLE IF NOT EXISTS on first connect |
| FileStore / S3 / GCS / Azure | directories/buckets are created on first write |
You never run migrations or create schemas manually.
RAGObserve ships three backends. Swap them via store= in init().
Zero config. Local file. Full dashboard.
ragobserve.init(project="dev") # default hidden path
ragobserve.init(project="dev", db_path="/data/store.db") # custom path
ragobserve.init(project="dev", store=ragobserve.SQLiteStore("/data/store.db"))Full read/write. Dashboard works. Best for team deployments and production.
ragobserve.init(
project="prod",
store=ragobserve.PostgresStore("postgresql://user:pass@host:5432/dbname"),
)Tables are auto-created on first connect. No migrations needed. Requires pip install ragobserve[postgres].
Write-only JSONL. Works with any fsspec-compatible target: S3, GCS, Azure Blob, Google Drive, SFTP, or local. No SQL queries — use with MultiStore for a dashboard, or query offline with DuckDB / Athena / BigQuery.
ragobserve.init(project="prod", store=ragobserve.FileStore("s3://my-bucket/rag-events/"))
ragobserve.init(project="prod", store=ragobserve.FileStore("gs://my-bucket/rag-events/"))
ragobserve.init(project="prod", store=ragobserve.FileStore("az://container/rag-events/"))
ragobserve.init(project="prod", store=ragobserve.FileStore("gdrive://My Drive/rag-events/"))
ragobserve.init(project="prod", store=ragobserve.FileStore("/local/archive/"))Fan-out writes to multiple backends. Reads come from the first backend that supports them (the primary). The canonical pattern: local dashboard + durable cloud archive.
store = ragobserve.MultiStore([
ragobserve.SQLiteStore(), # primary: dashboard reads
ragobserve.FileStore("s3://my-bucket/events/"), # sink: durable archive
])
ragobserve.init(project="prod", store=store)Any object that implements ingest_events(events), set_ground_truth(...), and close() is a valid store:
class MyStore:
def ingest_events(self, events): ...
def set_ground_truth(self, trace_id, project, ids): ...
def close(self): ...
ragobserve.init(project="prod", store=MyStore())- Query Explorer — every query with latency, cost, retriever, model, chunk count
- Trace waterfall — the full pipeline per query, stage by stage
- Retrieval Explorer — retrieved chunks with scores, ranks, metadata
- Hybrid Search Explorer — BM25 vs vector vs fused results
- Reranker Analytics — before/after with rank shifts and Kendall's τ
- Context Builder Viewer — exactly what was sent to the model, DevTools-style
- Chunk Explorer — most retrieved / never retrieved (dead) / duplicate chunks
- Metrics — Precision@k, Recall@k, MRR, nDCG over logged ground truth, plus chunk utilization
- Generations & cost — Langfuse-style cost tracing: per-model / per-day token & $ breakdowns, charts, and the context that produced each generation. Costs are auto-backfilled from a built-in price book when you don't pass
cost=.
# Docker Compose (recommended)
docker compose up
# → http://localhost:5601?key=<printed-key>
# Or plain Docker
docker build -t ragobserve .
docker run -p 5601:5601 -v ragobserve_data:/data \
-e RAGOBSERVE_API_KEY=mysecretkey ragobserveData persists in the ragobserve_data named volume. Pass GROQ_API_KEY to enable ragobserve eval inside the container.
Single worker: the container runs one uvicorn worker by default — required for the WebSocket live feed.
When running in server mode (ragobserve ui or ragobserve.serve()), the REST API and WebSocket are protected by an API key.
# Auto-generated on first start; printed in the console URL
ragobserve ui
# → Dashboard: http://127.0.0.1:5601?key=<key>
# Set your own key
RAGOBSERVE_API_KEY=mysecretkey ragobserve uiClients authenticate via:
Authorization: Bearer <key>
# or
X-Api-Key: <key>
The dashboard auto-reads the key from the ?key= URL param on first load and stores it in localStorage.
Rate your RAG system's answers with LLM-as-judge metrics powered by Groq (fast, free tier).
from ragobserve.eval import score_faithfulness, score_answer_relevance, evaluate_trace
# Single metrics
faith = score_faithfulness(answer="90 days.", context=["Notice period is 90 days."])
# → {"score": 0.97, "reason": "All claims directly supported by context."}
rel = score_answer_relevance(answer="90 days.", query="What is the notice period?")
# → {"score": 0.95, "reason": "Directly answers the query."}
# Score a full trace (loads answer, context, and query automatically)
result = evaluate_trace(trace_data, api_key="gsk_...")
# → {"faithfulness": {...}, "answer_relevance": {...}}Requires GROQ_API_KEY env var (or pass api_key= explicitly). Uses llama3-8b-8192 by default; pass model= to change.
# Batch-eval all traces in a project from the CLI
ragobserve eval --project my-rag --api-key gsk_...RAGObserve ships a zero-SDK, httpx-based provider layer covering 11 providers — Anthropic, OpenAI, Gemini, Groq, OpenRouter, Together, Mistral, DeepSeek, Fireworks, Perplexity, Ollama. From any trace's Generation / Context view you can replay the captured context against a live provider (when its API key is set) and the new generation is logged back into the trace with its cost.
ragobserve providers # list providers and which have keys configuredFull pipeline — ingest and query — is captured.
from ragobserve.adapters import (
RagObserveCallbackHandler,
instrument_loader, instrument_splitter, instrument_embeddings,
)
# query-time: retrieval + generation (+ model, token usage, cost) via the handler
chain.invoke(q, config={"callbacks": [RagObserveCallbackHandler()]})
# ingest-time: loaders/splitters/embeddings emit no callbacks, so wrap them
loader = instrument_loader(PyPDFLoader("contract.pdf"))
splitter = instrument_splitter(RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=50))
emb = instrument_embeddings(OpenAIEmbeddings()) # real Embeddings subclass — FAISS-safefrom ragobserve.adapters.llamaindex import register
register() # ONE call instruments the global dispatcher — ingest + query| Stage | LangChain | LlamaIndex |
|---|---|---|
| ingestion | instrument_loader |
(via pipeline) |
| chunking | instrument_splitter |
auto |
| embedding | instrument_embeddings |
auto |
| retrieval | auto (callback) | auto |
| reranking | instrument_compressor (or log_rerank) |
auto |
| context assembly | auto (handler) | auto |
| generation + cost | auto | auto |
import ragobserve
ragobserve.init(project="my-rag")
col = ragobserve.instrument_chroma(chroma_collection)
idx = ragobserve.instrument_pinecone(pinecone_index)
qc = ragobserve.instrument_qdrant(qdrant_client)
wv = ragobserve.instrument_weaviate(weaviate_collection)
mv = ragobserve.instrument_milvus(milvus_collection)
# pgvector — no client to proxy, pass the rows:
rows = cur.fetchall()
ragobserve.log_pgvector(query, rows)# Export all traces for a project to NDJSON (one trace+events per line)
ragobserve export --project my-rag --output traces.ndjson
# Works with Postgres too
ragobserve export --project my-rag \
--backend-store-uri postgresql://user:pass@host:5432/ragobs \
--output traces.ndjsonGET /health → {"status": "ok", "version": "0.4.0"}
No auth required — use for load balancer health checks and container readiness probes.
The dashboard Query Explorer auto-refreshes when new events arrive via WebSocket. You can also connect directly:
const ws = new WebSocket("ws://localhost:5601/ws/traces?key=<apikey>&project=my-rag");
ws.onmessage = e => {
const msg = JSON.parse(e.data);
if (msg.type === "event") console.log(msg.data);
// msg.type === "ping" every ~30s (keepalive)
};python examples/demo_rag.py
ragobserve uipip install -e .[dev]
pytest