Use when building Retrieval-Augmented Generation systems - covers document ingestion, hybrid search retrieval, reranking results, and prompt augmentation for accurate LLM responses grounded in your knowledge base
uv add llmemory
# For reranking support
uv add "llmemory[reranker-local]" # Local cross-encoder models
# or configure OpenAI reranking (no extra install needed)
Retrieval-Augmented Generation (RAG) combines llmemory's document retrieval with LLM generation for accurate, grounded responses.
RAG Pipeline:
When to use RAG:
from llmemory import LLMemory, SearchType, DocumentType
from openai import AsyncOpenAI
async def rag_system():
# Initialize
memory = LLMemory(
connection_string="postgresql://localhost/mydb",
openai_api_key="sk-..."
)
await memory.initialize()
# 1. Ingest documents
await memory.add_document(
owner_id="workspace-1",
id_at_origin="kb",
document_name="product_guide.md",
document_type=DocumentType.MARKDOWN,
content="Your product documentation..."
)
# 2. Retrieve with reranking
results = await memory.search(
owner_id="workspace-1",
query_text="how to reset password",
search_type=SearchType.HYBRID,
query_expansion=True, # Better retrieval
rerank=True, # Better ranking
rerank_top_k=50, # Rerank top 50 candidates
rerank_return_k=10, # Prefer 10 best after reranking
limit=5 # Final result count (max of limit and rerank_return_k)
)
# 3. Build prompt with context
context = "\n\n".join([
f"Source: {r.metadata.get('source', 'unknown')}\n{r.content}"
for r in results
])
prompt = f"""Answer the question using only the provided context.
Context:
{context}
Question: how to reset password
Answer:"""
# 4. Generate response
client = AsyncOpenAI()
response = await client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
print(response.choices[0].message.content)
await memory.close()
import asyncio
asyncio.run(rag_system())
Production RAG systems should detect when queries cannot be answered from available documents.
When to use query routing:
Example:
from llmemory import LLMemory
async with LLMemory(connection_string="...") as memory:
# Search with automatic routing
result = await memory.search_with_routing(
owner_id="workspace-1",
query_text="What's the current weather in Paris?",
enable_routing=True,
limit=5
)
if result["route"] == "retrieval":
# Answer from documents
return generate_answer(result["results"])
elif result["route"] == "web_search":
# Route to web search
return fetch_from_web(query)
elif result["route"] == "unanswerable":
# Honest response
return "I don't have information to answer that question."
else: # clarification
return "Could you please provide more details?"
API Reference:
Route queries intelligently before searching.
Signature:
async def search_with_routing(
owner_id: str,
query_text: str,
enable_routing: bool = True,
routing_threshold: float = 0.7,
**search_kwargs
) -> Dict[str, Any]
Parameters:
owner_id (str): Owner identifierquery_text (str): Search queryenable_routing (bool, default: True): Enable automatic routingrouting_threshold (float, default: 0.7): Confidence threshold**search_kwargs: Additional arguments passed to search()Returns: Dict with:
route (str): "retrieval", "web_search", "unanswerable", or "clarification"confidence (float): 0-1 confidence in routing decisionresults (List[SearchResult]): If route="retrieval"message (str): If route != "retrieval"reason (str): Explanation of routing decisionExample:
result = await memory.search_with_routing(
owner_id="support",
query_text="How do I reset my password?",
routing_threshold=0.8
)
if result["route"] == "retrieval":
answer = generate_rag_response(result["results"])
else:
answer = result["message"] # Pre-formatted response
from llmemory import LLMemory, DocumentType, ChunkingConfig, LLMemoryConfig
async def ingest_knowledge_base(owner_id: str):
"""Ingest documents into RAG system."""
# Configure chunking for RAG (smaller chunks for precise retrieval)
chunking_config = ChunkingConfig(
chunk_size=300, # Tokens per chunk (smaller for RAG)
chunk_overlap=50, # Overlap for context preservation
strategy="hierarchical", # Chunking strategy
min_chunk_size=100, # Minimum chunk size
max_chunk_size=500 # Maximum chunk size
)
# Enable chunk summaries via LLMemoryConfig
config = LLMemoryConfig()
config.chunking.enable_chunk_summaries = True
config.chunking.summary_max_tokens = 80
memory = LLMemory(
connection_string="postgresql://localhost/mydb",
config=config
)
await memory.initialize()
documents = [
{
"name": "product_guide.md",
"type": DocumentType.MARKDOWN,
"content": "...",
"metadata": {"category": "guide", "version": "2.0"}
},
{
"name": "faq.md",
"type": DocumentType.MARKDOWN,
"content": "...",
"metadata": {"category": "faq"}
},
{
"name": "api_docs.md",
"type": DocumentType.TECHNICAL_DOC,
"content": "...",
"metadata": {"category": "api", "language": "python"}
}
]
for doc in documents:
result = await memory.add_document(
owner_id=owner_id,
id_at_origin="knowledge_base",
document_name=doc["name"],
document_type=doc["type"],
content=doc["content"],
metadata=doc["metadata"],
chunking_config=chunking_config,
generate_embeddings=True
)
print(f"Ingested {doc['name']}: {result.chunks_created} chunks")
async def retrieve_for_rag(
memory: LLMemory,
owner_id: str,
query: str,
top_k: int = 5
) -> List[SearchResult]:
"""Retrieve relevant chunks for RAG."""
results = await memory.search(
owner_id=owner_id,
query_text=query,
# Hybrid search for best quality
search_type=SearchType.HYBRID,
alpha=0.6, # Slight favor to semantic search
# Query expansion for better recall
query_expansion=True,
max_query_variants=3,
# Reranking for precision
rerank=True,
rerank_top_k=20, # Consider top 20 candidates
rerank_return_k=top_k, # Prefer top_k after reranking
# Final limit (actual count = max(limit, rerank_return_k))
limit=top_k
)
return results
llmemory supports multiple reranking methods:
# Configure via environment
LLMEMORY_RERANK_PROVIDER=openai
LLMEMORY_RERANK_MODEL=gpt-4.1-mini
LLMEMORY_RERANK_TOP_K=30
LLMEMORY_RERANK_RETURN_K=10
# Or programmatically
from llmemory import LLMemoryConfig
config = LLMemoryConfig()
config.search.enable_rerank = True
config.search.rerank_provider = "openai"
config.search.default_rerank_model = "gpt-4.1-mini"
config.search.rerank_top_k = 30
config.search.rerank_return_k = 10
memory = LLMemory(
connection_string="postgresql://localhost/mydb",
config=config
)
# Install local reranker dependencies
uv add "llmemory[reranker-local]"
# Configure
config = LLMemoryConfig()
config.search.enable_rerank = True
config.search.default_rerank_model = "cross-encoder/ms-marco-MiniLM-L6-v2"
config.search.rerank_device = "cpu" # or "cuda"
config.search.rerank_batch_size = 16
# Automatic fallback when no reranker configured
# Uses token overlap scoring
results = await memory.search(
owner_id="workspace-1",
query_text="query",
rerank=True # Uses lexical reranking
)
Local cross-encoder model for reranking search results without API calls.
Constructor:
CrossEncoderReranker(
model_name: str = "cross-encoder/ms-marco-MiniLM-L6-v2",
device: Optional[str] = None,
batch_size: int = 16
)
Parameters:
model_name (str, default: "cross-encoder/ms-marco-MiniLM-L6-v2"): Hugging Face cross-encoder model name
device (Optional[str]): Device to run on: "cpu", "cuda", or None (auto-detect)batch_size (int, default: 16): Batch size for inferenceMethods:
Score query-document pairs for relevance.
Signature:
async def score(
query_text: str,
results: Sequence[SearchResult]
) -> Sequence[float]
Parameters:
query_text (str): Search queryresults (Sequence[SearchResult]): Search results to scoreReturns:
Sequence[float]: Relevance scores (same length as results)Example:
from llmemory import CrossEncoderReranker
# Initialize reranker
reranker = CrossEncoderReranker(
model_name="cross-encoder/ms-marco-MiniLM-L6-v2",
device="cpu",
batch_size=32
)
# Get initial search results
results = await memory.search(
owner_id="workspace-1",
query_text="machine learning",
limit=50,
rerank=False # Get unranked results
)
# Rerank with cross-encoder
scores = await reranker.score("machine learning", results)
# Sort by new scores
scored_results = list(zip(scores, results))
scored_results.sort(key=lambda x: x[0], reverse=True)
top_results = [r for _, r in scored_results[:10]]
Installation:
# Requires sentence-transformers
uv add "llmemory[reranker-local]"
Use OpenAI GPT models for intelligent reranking with natural language understanding.
Constructor:
OpenAIResponsesReranker(
model: str = "gpt-4.1-mini",
max_candidates: int = 30,
temperature: float = 0.0
)
Parameters:
model (str, default: "gpt-4.1-mini"): OpenAI model name
max_candidates (int, default: 30): Maximum candidates to send to APItemperature (float, default: 0.0): Model temperature (0 = deterministic)Methods:
Score query-document pairs using OpenAI API.
Signature:
async def score(
query_text: str,
results: Sequence[SearchResult]
) -> Sequence[float]
Parameters:
query_text (str): Search queryresults (Sequence[SearchResult]): Search results to scoreReturns:
Sequence[float]: Relevance scores between 0 and 1Example:
from llmemory import OpenAIResponsesReranker
import os
# Initialize reranker (uses OPENAI_API_KEY from env)
reranker = OpenAIResponsesReranker(
model="gpt-4.1-mini",
max_candidates=20,
temperature=0.0
)
# Get initial search results
results = await memory.search(
owner_id="workspace-1",
query_text="customer retention strategies",
limit=50,
rerank=False
)
# Rerank with OpenAI
scores = await reranker.score("customer retention strategies", results)
# Sort by scores
scored_results = list(zip(scores, results))
scored_results.sort(key=lambda x: x[0], reverse=True)
top_results = [r for _, r in scored_results[:10]]
print(f"Top result score: {scores[0]:.3f}")
Cost Considerations:
When to use:
Internal service that wraps reranker implementations (rarely used directly).
Usage: Automatically created by LLMemory when reranking is enabled via configuration. Generally not instantiated directly by users.
Search results contain multiple score fields depending on the search configuration. Understanding these fields helps optimize RAG retrieval quality.
chunk_id (UUID)
document_id (UUID)
content (str)
metadata (Dict[str, Any])
score (float)
similarity (Optional[float])
text_rank (Optional[float])
rrf_score (Optional[float])
rerank_score (Optional[float])
summary (Optional[str])
text = result.summary or result.content# Example: Analyzing search result scores
results = await memory.search(
owner_id="workspace-1",
query_text="machine learning algorithms",
search_type=SearchType.HYBRID,
query_expansion=True,
rerank=True,
limit=5
)
for result in results:
print(f"Chunk ID: {result.chunk_id}")
print(f" Final score: {result.score:.3f}")
# Vector component (if hybrid/vector search)
if result.similarity is not None:
print(f" Vector similarity: {result.similarity:.3f}")
# Text component (if hybrid/text search)
if result.text_rank is not None:
print(f" BM25 rank: {result.text_rank:.3f}")
# Multi-query fusion (if query_expansion=True)
if result.rrf_score is not None:
print(f" RRF score: {result.rrf_score:.3f}")
# Reranking (if rerank=True)
if result.rerank_score is not None:
print(f" Rerank score: {result.rerank_score:.3f}")
# Summary (if enabled during ingestion)
if result.summary:
print(f" Summary: {result.summary[:100]}...")
rerank_top_k (int, default: 50)
rerank_return_k (int, default: 15)
max(limit, rerank_return_k)limit (int, default: 10)
final_count = max(limit, rerank_return_k)# Example: Reranking parameter interactions
results = await memory.search(
owner_id="workspace-1",
query_text="database optimization",
search_type=SearchType.HYBRID,
rerank=True,
rerank_top_k=50, # Consider top 50 from base search
rerank_return_k=10, # Prefer 10 best after reranking
limit=5 # But return max(5, 10) = 10 results
)
# Returns 10 results (max of limit and rerank_return_k)
assert len(results) == 10
results = await memory.search(
owner_id="workspace-1",
query_text="database optimization",
search_type=SearchType.HYBRID,
rerank=True,
rerank_top_k=50, # Consider top 50 from base search
rerank_return_k=5, # Prefer 5 best after reranking
limit=20 # But return max(20, 5) = 20 results
)
# Returns 20 results (max of limit and rerank_return_k)
assert len(results) == 20
def build_rag_prompt(
query: str,
results: List[SearchResult],
system_instructions: str = "Answer based only on the provided context."
) -> str:
"""Build RAG prompt with retrieved context."""
# Format context from search results
context_parts = []
for i, result in enumerate(results, 1):
# Include source information
source = result.metadata.get("source", "Unknown")
doc_name = result.metadata.get("document_name", "")
# Use summary if available (more concise for prompts)
# Summary is populated when ChunkingConfig.enable_chunk_summaries=True
text = result.summary or result.content
context_parts.append(
f"[Source {i}: {doc_name or source}]\n{text}"
)
context = "\n\n".join(context_parts)
# Build final prompt
prompt = f"""{system_instructions}
Context:
{context}
Question: {query}
Answer:"""
return prompt
With Citation Requirements:
def build_prompt_with_citations(query: str, results: List[SearchResult]) -> str:
context_parts = []
for i, result in enumerate(results, 1):
source = result.metadata.get("document_name", f"Source {i}")
# Use summary if enabled (see enabling summaries section below)
text = result.summary or result.content
context_parts.append(f"[{i}] {source}: {text}")
context = "\n\n".join(context_parts)
prompt = f"""Answer the question using the provided context. Cite sources using [number] format.
Context:
{context}
Question: {query}
Answer (with citations):"""
return prompt
With Metadata Filtering:
async def rag_with_filters(
memory: LLMemory,
owner_id: str,
query: str,
category: str
):
"""RAG with metadata filtering."""
results = await memory.search(
owner_id=owner_id,
query_text=query,
search_type=SearchType.HYBRID,
metadata_filter={"category": category}, # Filter by category
rerank=True,
limit=5
)
return build_rag_prompt(query, results)
from openai import AsyncOpenAI
async def generate_rag_response(
query: str,
results: List[SearchResult],
model: str = "gpt-4"
) -> dict:
"""Generate LLM response with RAG context."""
# Build prompt
prompt = build_rag_prompt(query, results)
# Generate with OpenAI
client = AsyncOpenAI()
response = await client.chat.completions.create(
model=model,
messages=[
{
"role": "system",
"content": "You are a helpful assistant that answers questions based on provided context."
},
{
"role": "user",
"content": prompt
}
],
temperature=0.3, # Lower temperature for factual answers
max_tokens=500
)
# Extract response
answer = response.choices[0].message.content
return {
"answer": answer,
"sources": [
{
"content": r.content[:200] + "...",
"score": r.score,
"metadata": r.metadata
}
for r in results
],
"model": model
}
from llmemory import LLMemory, SearchType, DocumentType
from openai import AsyncOpenAI
from typing import List, Dict, Any
class RAGSystem:
"""Complete RAG system with llmemory."""
def __init__(self, connection_string: str, openai_api_key: str):
self.memory = LLMemory(
connection_string=connection_string,
openai_api_key=openai_api_key
)
self.client = AsyncOpenAI(api_key=openai_api_key)
self.initialized = False
async def initialize(self):
"""Initialize the RAG system."""
await self.memory.initialize()
self.initialized = True
async def ingest_document(
self,
owner_id: str,
document_name: str,
content: str,
document_type: DocumentType = DocumentType.TEXT,
metadata: Dict[str, Any] = None
):
"""Add a document to the knowledge base."""
result = await self.memory.add_document(
owner_id=owner_id,
id_at_origin="rag_kb",
document_name=document_name,
document_type=document_type,
content=content,
metadata=metadata or {},
generate_embeddings=True
)
return {
"document_id": str(result.document.document_id),
"chunks_created": result.chunks_created
}
async def answer_question(
self,
owner_id: str,
question: str,
top_k: int = 5,
model: str = "gpt-4"
) -> Dict[str, Any]:
"""Answer a question using RAG."""
# Retrieve relevant chunks
results = await self.memory.search(
owner_id=owner_id,
query_text=question,
search_type=SearchType.HYBRID,
query_expansion=True,
max_query_variants=3,
rerank=True,
rerank_top_k=20,
rerank_return_k=top_k,
limit=top_k
)
if not results:
return {
"answer": "I don't have enough information to answer this question.",
"sources": [],
"confidence": "low"
}
# Build prompt
context = "\n\n".join([
f"[Source: {r.metadata.get('document_name', 'Unknown')}]\n{r.summary or r.content}"
for r in results
])
prompt = f"""Answer the question using only the provided context. If the answer cannot be found in the contex
<!-- Content truncated for initial SEO render. Open the source file tab for the full file. -->
Search for places (restaurants, cafes, etc.) via Google Places API proxy on localhost.
Interact with GitHub using the `gh` CLI. Use `gh issue`, `gh pr`, `gh run`, and `gh api` for issues, PRs, CI runs, and advanced queries.
Create or update AgentSkills. Use when designing, structuring, or packaging skills with scripts, references, and assets.
Start voice calls via the OpenClaw voice-call plugin.
Notion API for creating and managing pages, databases, and blocks.
Gemini CLI for one-shot Q&A, summaries, and generation.
Category:developer