RAG Pipeline¶
FlowgentraAI includes a complete Retrieval-Augmented Generation (RAG) pipeline: text splitting, embeddings, vector storage, and retrieval.
Text Splitting¶
from flowgentra_ai import chunk_text, estimate_tokens, chunk_text_by_tokens
# Split by character count
chunks = chunk_text("long text...", chunk_size=500, overlap=50)
# Split by token count
chunks = chunk_text_by_tokens("long text...", max_tokens=200, overlap_tokens=20)
# Estimate tokens
count = estimate_tokens("some text")
Specialized Splitters¶
from flowgentra_ai import (
RecursiveCharacterTextSplitter,
MarkdownTextSplitter,
HTMLTextSplitter,
TokenTextSplitter,
CodeTextSplitter,
)
PDF Extraction¶
from flowgentra_ai import extract_text, extract_pdf, extract_and_chunk
# Extract full text
text = extract_text("document.pdf")
# Extract as PdfDocument object
doc = extract_pdf("document.pdf")
print(doc.source) # "document.pdf"
print(doc.page_count) # number of pages
print(doc.text) # full extracted text
# Extract and chunk in one step
chunks = extract_and_chunk("document.pdf", chunk_size=1000, overlap=200)
# Returns list of (chunk_id, chunk_text) tuples
Document Loaders¶
from flowgentra_ai import load_document, load_directory
# Load a single file
doc = load_document("data/report.pdf")
# Load an entire directory
docs = load_directory("data/")
Embeddings¶
from flowgentra_ai import Embeddings
# OpenAI (recommended)
emb = Embeddings.openai("sk-...", "text-embedding-3-small")
# OpenAI with custom dimension
emb = Embeddings.openai_with_dimension("sk-...", "text-embedding-3-small", 256)
# OpenAI with caching
emb = Embeddings.openai_cached("sk-...")
# Ollama (local, free)
emb = Embeddings.ollama("nomic-embed-text")
# Mistral
emb = Embeddings.mistral("api-key-...")
# HuggingFace
emb = Embeddings.huggingface("sentence-transformers/all-MiniLM-L6-v2", "hf-token")
# Mock (for testing, no API needed)
emb = Embeddings.mock(128)
# Generate embeddings
vector = emb.embed("Hello world")
vectors = emb.embed_batch(["Hello", "World"])
dim = emb.get_dimension()
Vector Store¶
from flowgentra_ai import Document, InMemoryVectorStore
store = InMemoryVectorStore()
# Index documents
doc = Document("doc-1", "Rust is a systems programming language.")
embedding = emb.embed(doc.text)
store.index(doc, embedding)
# Search by embedding
query_emb = emb.embed("What is Rust?")
results = store.search(query_emb, top_k=5)
for r in results:
print(f"{r.id}: {r.text} (score: {r.score:.3f})")
# Search with metadata filter
results = store.search(query_emb, top_k=5, filter={"source": "wiki"})
# Manage documents
doc = store.get("doc-1")
all_docs = store.list()
store.delete("doc-1")
store.clear()
ChromaDB Store¶
Retriever¶
The Retriever combines embedding, search, hybrid scoring, and deduplication into a single pipeline:
from flowgentra_ai import Retriever, RetrievalConfig
# Semantic search
config = RetrievalConfig.semantic(top_k=5, threshold=0.7)
retriever = Retriever(store, emb, config)
results = retriever.retrieve("What is Rust?")
# Hybrid search (semantic + keyword)
config = RetrievalConfig.hybrid(keyword_weight=0.3, top_k=10, threshold=0.5)
retriever = Retriever(store, emb, config)
# With deduplication
retriever.with_dedup(threshold=0.85)
results = retriever.retrieve("query")
Hybrid Search Utilities¶
from flowgentra_ai import bm25_score, hybrid_merge, dedup_by_id, dedup_by_similarity
# BM25 keyword scoring
scores = bm25_score("rust language", ["Rust is fast", "Python is easy"])
# Merge semantic results with keyword scores
merged = hybrid_merge(semantic_results, "query", keyword_weight=0.3)
# Deduplication
unique = dedup_by_id(results)
unique = dedup_by_similarity(results, threshold=0.85)
Query Expansion¶
from flowgentra_ai import decompose_query
# Split compound queries into sub-queries
queries = decompose_query("Rust safety and performance", max_depth=2)
# ["Rust safety and performance", "Rust safety", "performance"]