Skip to main content

Search Example

Complete examples for searching RAG collections.

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]}...")
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")
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")
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")
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

  1. Query specificity - More specific queries = better results
  2. Use filters - Narrow results with metadata
  3. Adjust threshold - Balance precision and recall
  4. Multi-collection - Search across related collections
  5. Error handling - Handle network failures gracefully
  6. Performance - Monitor search response times

Next Steps