Search Example
Complete examples for searching RAG collections.
Basic Search
import requests
def search_documents(collection_name, query, top_k=5):
response = requests.post(
"http://localhost/api/v1/rag/search",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"collection_name": collection_name,
"query": query,
"top_k": top_k
}
)
return response.json()
# Search for information
result = search_documents("documentation", "How do I reset my password?")
print(f"Found {len(result['results'])} results:\n")
for i, doc in enumerate(result['results'], 1):
print(f"{i}. Score: {doc['score']:.4f}")
print(f" Content: {doc['content'][:150]}...")
print(f" Document: {doc['metadata']['document_id']}\n")
Search with Filters
def search_with_filters(collection_name, query, filters=None, top_k=5):
payload = {
"collection_name": collection_name,
"query": query,
"top_k": top_k
}
if filters:
payload["filters"] = filters
response = requests.post(
"http://localhost/api/v1/rag/search",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json=payload
)
return response.json()
# Search with metadata filters
result = search_with_filters(
"policies",
"Data privacy guidelines",
filters={
"category": "legal",
"version": "2.0"
},
top_k=5
)
for doc in result["results"]:
print(f"Score: {doc['score']:.4f}")
print(f"Category: {doc['metadata'].get('category', 'N/A')}")
print(f"Content: {doc['content'][:100]}...\n")
Search with Similarity Threshold
def precise_search(collection_name, query, threshold=0.8, top_k=10):
response = requests.post(
"http://localhost/api/v1/rag/search",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"collection_name": collection_name,
"query": query,
"similarity_threshold": threshold,
"top_k": top_k
}
)
return response.json()
# Only highly relevant results (90%+ similarity)
result = precise_search("docs", "API authentication", threshold=0.9)
print(f"Found {len(result['results'])} highly relevant results:")
for doc in result["results"]:
print(f" Score: {doc['score']:.2f} - {doc['content'][:80]}...")
Multi-Collection Search
def search_multiple_collections(collections, query, top_k=3):
all_results = []
for collection in collections:
response = requests.post(
"http://localhost/api/v1/rag/search",
headers={"Authorization": "Bearer YOUR_API_KEY"},
json={
"collection_name": collection,
"query": query,
"top_k": top_k
}
)
results = response.json()["results"]
for r in results:
r["collection"] = collection
all_results.extend(results)
# Sort by score
all_results.sort(key=lambda x: x["score"], reverse=True)
return all_results
# Search across multiple collections
results = search_multiple_collections(
["docs", "support", "faq"],
"How do I upgrade my plan?"
)
print(f"Total results: {len(results)}\n")
for i, doc in enumerate(results[:10], 1):
print(f"{i}. [{doc['collection']}] Score: {doc['score']:.2f}")
print(f" {doc['content'][:100]}...\n")
JavaScript Search
class RAGSearcher {
constructor(apiKey) {
this.apiKey = apiKey;
this.baseUrl = 'http://localhost/api/v1';
}
async search(collectionName, query, options = {}) {
const payload = {
collection_name: collectionName,
query: query,
top_k: options.top_k || 5
};
if (options.similarityThreshold !== undefined) {
payload.similarity_threshold = options.similarityThreshold;
}
if (options.filters) {
payload.filters = options.filters;
}
const response = await fetch(`${this.baseUrl}/rag/search`, {
method: 'POST',
headers: {
'Authorization': `Bearer ${this.apiKey}`,
'Content-Type': 'application/json'
},
body: JSON.stringify(payload)
});
if (!response.ok) {
throw new Error('Search failed');
}
return await response.json();
}
async searchMultiple(collections, query, options = {}) {
const results = [];
for (const collection of collections) {
const data = await this.search(collection, query, options);
data.results.forEach(r => {
r.collection = collection;
});
results.push(...data.results);
}
results.sort((a, b) => b.score - a.score);
return results;
}
}
// Usage
const searcher = new RAGSearcher('YOUR_API_KEY');
// Basic search
const results = await searcher.search('documentation', 'How to configure API keys?');
console.log(`Found ${results.results.length} results`);
// Search with filters
const filtered = await searcher.search('policies', 'Privacy', {
filters: { category: 'legal', version: '2.0' }
});
// Multi-collection search
const multiResults = await searcher.searchMultiple(
['docs', 'support'],
'Upgrade plan'
);
React Search Component
import React, { useState } from 'react';
function RAGSearch() {
const [query, setQuery] = useState('');
const [results, setResults] = useState([]);
const [loading, setLoading] = useState(false);
const [collection] = useState('documentation');
const search = async () => {
if (!query.trim()) return;
setLoading(true);
try {
const response = await fetch('http://localhost/api/v1/rag/search', {
method: 'POST',
headers: {
'Authorization': 'Bearer YOUR_API_KEY',
'Content-Type': 'application/json'
},
body: JSON.stringify({
collection_name: collection,
query,
top_k: 5
})
});
const data = await response.json();
setResults(data.results || []);
} catch (error) {
console.error('Search failed:', error);
} finally {
setLoading(false);
}
};
return (
<div className="p-6">
<h2 className="text-2xl font-bold mb-4">RAG Search</h2>
<div className="mb-4">
<input
type="text"
value={query}
onChange={(e) => setQuery(e.target.value)}
onKeyPress={(e) => e.key === 'Enter' && search()}
className="w-full px-4 py-2 border rounded"
placeholder="Search documents..."
disabled={loading}
/>
<button
onClick={search}
disabled={loading || !query}
className="mt-2 px-4 py-2 bg-blue-500 text-white rounded"
>
{loading ? 'Searching...' : 'Search'}
</button>
</div>
{results.length > 0 && (
<div>
<h3 className="font-bold mb-2">Results</h3>
{results.map((doc, i) => (
<div key={i} className="mb-4 p-4 border rounded">
<div className="flex justify-between">
<span className="font-bold">
Score: {doc.score.toFixed(4)}
</span>
<span className="text-sm text-gray-600">
{doc.metadata.document_id}
</span>
</div>
<p className="mt-2">
{doc.content.substring(0, 200)}...
</p>
</div>
))}
</div>
)}
</div>
);
}
export default RAGSearch;
Advanced Search Patterns
Semantic Variations
semantic_variations = [
"How to configure API keys?",
"API key setup instructions",
"Where do I put my API key?",
"Configure authentication for API",
"API key configuration steps"
]
for query in semantic_variations:
result = search_documents("docs", query, top_k=3)
print(f"Query: {query}")
print(f"Results: {len(result['results'])}\n")
Hierarchical Search
def hierarchical_search(collection_name, query):
# First try precise search
precise = search_documents(collection_name, query, top_k=3)
# If few results, try broader search
if len(precise['results']) < 2:
broader = search_documents(collection_name, query, top_k=10)
return broader
return precise
result = hierarchical_search("docs", "configuration")
Context-Aware Search
def contextual_search(collection_name, query, context_keywords):
all_results = []
# Search main query
main_result = search_documents(collection_name, query, top_k=5)
all_results.extend(main_result['results'])
# Search context keywords
for keyword in context_keywords:
context_result = search_documents(collection_name, keyword, top_k=2)
all_results.extend(context_result['results'])
# Deduplicate and sort
unique = {r['metadata']['document_id']: r for r in all_results}
sorted_results = sorted(unique.values(), key=lambda x: x['score'], reverse=True)
return sorted_results[:10]
# Search with context
result = contextual_search(
"docs",
"password reset",
["security", "authentication", "account"]
)
Search Analytics
import time
def measure_search_performance(collection_name, queries):
metrics = {
"total_queries": len(queries),
"total_time": 0,
"total_results": 0,
"queries": []
}
for query in queries:
start = time.time()
result = search_documents(collection_name, query, top_k=5)
elapsed = time.time() - start
metrics["total_time"] += elapsed
metrics["total_results"] += len(result["results"])
metrics["queries"].append({
"query": query,
"time": elapsed,
"results": len(result["results"]),
"avg_score": sum(r["score"] for r in result["results"]) / len(result["results"]) if result["results"] else 0
})
metrics["avg_time"] = metrics["total_time"] / metrics["total_queries"]
metrics["avg_results"] = metrics["total_results"] / metrics["total_queries"]
return metrics
# Measure performance
queries = [
"API key configuration",
"Password reset",
"Plan upgrade",
"Account settings"
]
metrics = measure_search_performance("documentation", queries)
print(f"Total Queries: {metrics['total_queries']}")
print(f"Average Time: {metrics['avg_time']:.3f}s")
print(f"Average Results: {metrics['avg_results']:.1f}")
Best Practices
- Query specificity - More specific queries = better results
- Use filters - Narrow results with metadata
- Adjust threshold - Balance precision and recall
- Multi-collection - Search across related collections
- Error handling - Handle network failures gracefully
- Performance - Monitor search response times
Next Steps
- RAG Chat Example - Combine search with AI chat
- Upload Example - Add more documents
- Managing Example - Manage your documents