← Back to all products
$29
RAG Pipeline Starter
Python RAG pipeline with document ingestion, text chunking, vector store, retrieval engine, and prompt assembly.
PythonMarkdown
📄 Product Preview
Try the interactive reader and demo tools below, or get the full product with all content unlocked.
📖 Interactive Reader (Free Preview) ⚙ Try Demo Tools 📦 Download Free Sample📁 File Structure 9 files
rag-pipeline-starter/
├── LICENSE
├── README.md
├── examples/
│ └── basic_usage.py
├── free-sample.zip
├── guide/
│ ├── 01_features.md
│ ├── 02_quick-start.md
│ └── 03_license.md
├── index.html
└── src/
└── rag_pipeline.py
📖 Documentation Preview README excerpt
RAG Pipeline Starter
Python RAG pipeline with document ingestion, text chunking, vector store, retrieval engine, and prompt assembly. Zero dependencies.
Part of the AI Toolkit collection by [CodeVault](https://ai-toolkit.codevault.dev).
Features
- Document loader — Ingest
.txt,.md,.py,.json,.csvfiles - Text chunker — Configurable chunk size and overlap
- Vector store — In-memory store with cosine similarity search
- Retrieval engine — Top-K retrieval with relevance scoring
- Prompt assembler — Template-based prompt construction with context injection
- Pipeline orchestrator — Single
RAGPipelineclass ties everything together - CLI + API — Use from terminal or import as a library
- Demo mode — Built-in sample docs to see it working instantly
Quick Start
# Run the built-in demo
python src/rag_pipeline.py --demo
# Ingest a directory and query
python src/rag_pipeline.py --ingest ./my-docs/ --query "How do I deploy?"
# Interactive mode
python src/rag_pipeline.py
License
MIT — use in personal, commercial, or client projects. No attribution required.
📄 Code Sample .py preview
examples/basic_usage.py#!/usr/bin/env python3
"""Basic usage of RAG Pipeline Starter."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
from rag_pipeline import RAGPipeline
def main() -> None:
pipeline = RAGPipeline(chunk_size=256, top_k=2)
# Ingest some knowledge
docs = [
"Our REST API supports JSON and XML responses. Set Accept header accordingly.",
"Rate limits: 100 req/min (free), 1000 req/min (pro). Returns 429 on exceed.",
"Authentication requires Bearer token in Authorization header.",
"WebSocket endpoint at wss://api.example.com/ws for real-time updates.",
"All timestamps are UTC in ISO 8601 format.",
]
for doc in docs:
pipeline.ingest_text(doc, source="api-docs")
print(f"Pipeline ready: {pipeline.stats()}\n")
# Query
question = "How do I authenticate?"
print(f"Q: {question}")
results = pipeline.retrieve(question)
for r in results:
print(f" [{r['score']:.3f}] {r['text'][:80]}")
# Get assembled prompt
print("\n--- Full Prompt ---")
print(pipeline.query(question))
if __name__ == "__main__":
main()