Building a search system for AI applications is fundamentally different from general-purpose search. You need to handle both keyword and semantic queries, scale to thousands of concurrent requests, and tune relevance in ways that improve downstream RAG pipeline quality. This article walks through the patterns I’ve used to build search infrastructure that powers production AI applications.
Why FastAPI and Elasticsearch Together
FastAPI gives you async-first Python with minimal boilerplate. Elasticsearch gives you a battle-tested distributed search engine that handles indexing, querying, and relevance ranking out of the box. Together, they let you move fast without sacrificing production reliability.
You could build search from scratch with vector databases alone, which works well for pure semantic search. However, you’d be managing relevance tuning, scaling, and operational complexity yourself. Elasticsearch handles the infrastructure. FastAPI handles the API layer cleanly.
For AI applications specifically, this combination lets you combine BM25 keyword search with vector embeddings in a single query, which is critical for RAG quality. A pure vector store forces you to choose one or the other.
Core Architecture: Indexing and Query Flow
Start with a simple mental model. You have documents (articles, product descriptions, knowledge base entries). You index them into Elasticsearch with both raw text and vector embeddings. When a query comes in, you search using both BM25 (keyword relevance) and vector similarity, combine the scores, and return ranked results.
Here’s the indexing setup:
from elasticsearch import Elasticsearch
from sentence_transformers import SentenceTransformer
import json
es_client = Elasticsearch(["http://localhost:9200"])
model = SentenceTransformer('all-MiniLM-L6-v2')
# Define index mapping with vector field
index_mapping = {
"settings": {
"number_of_shards": 2,
"number_of_replicas": 1,
"analysis": {
"analyzer": {
"custom_analyzer": {
"type": "standard",
"stopwords": "_english_"
}
}
}
},
"mappings": {
"properties": {
"id": {"type": "keyword"},
"title": {
"type": "text",
"analyzer": "custom_analyzer",
"fields": {"keyword": {"type": "keyword"}}
},
"content": {
"type": "text",
"analyzer": "custom_analyzer"
},
"embedding": {
"type": "dense_vector",
"dims": 384,
"index": true,
"similarity": "cosine"
},
"metadata": {"type": "object"},
"tenant_id": {"type": "keyword"}
}
}
}
es_client.indices.create(index="documents", body=index_mapping, ignore=400)
The key decisions here: two shards for horizontal scaling, a custom analyzer that removes stop words, and a dense_vector field for embeddings. The tenant_id field enables multi-tenancy isolation at the index level, which I’ll cover later.
Now index documents with their embeddings:
def index_document(doc_id, title, content, tenant_id, metadata=None):
embedding = model.encode(f"{title} {content}").tolist()
doc_body = {
"id": doc_id,
"title": title,
"content": content,
"embedding": embedding,
"tenant_id": tenant_id,
"metadata": metadata or {}
}
es_client.index(index="documents", id=doc_id, body=doc_body)
# Bulk indexing for performance
def bulk_index_documents(documents, tenant_id):
actions = []
for doc in documents:
embedding = model.encode(f"{doc['title']} {doc['content']}").tolist()
action = {
"_op_type": "index",
"_index": "documents",
"_id": doc["id"],
"id": doc["id"],
"title": doc["title"],
"content": doc["content"],
"embedding": embedding,
"tenant_id": tenant_id,
"metadata": doc.get("metadata", {})
}
actions.append(action)
from elasticsearch.helpers import bulk
bulk(es_client, actions, chunk_size=500)
For production, always use bulk indexing. Single-document indexing creates I/O bottlenecks. Batch documents in chunks of 500 to 1000 depending on document size and memory constraints.
Hybrid Search: Combining BM25 and Vector Similarity
This is where the stack shines. A single query can leverage both keyword matching and semantic understanding. Elasticsearch 8.9+ provides RRF (Reciprocal Rank Fusion) as the recommended approach for combining BM25 and vector results.
RRF works by ranking documents based on their position in each result set, not raw scores. This makes it robust and requires almost no tuning. Here’s how to implement it:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
class SearchRequest(BaseModel):
query: str
tenant_id: str
limit: int = 10
def hybrid_search_rrf(query: str, tenant_id: str, limit: int = 10):
query_embedding = model.encode(query).tolist()
search_body = {
"size": limit,
"query": {
"bool": {
"must": [
{"term": {"tenant_id": tenant_id}}
]
}
},
"retriever": {
"standard": {
"rerank": [
{
"rrf": {
"rank_window_size": 50,
"rank_constant": 20
}
}
]
}
}
}
results = es_client.search(index="documents", body=search_body)
return [
{
"id": hit["_source"]["id"],
"title": hit["_source"]["title"],
"content": hit["_source"]["content"],
"score": hit["_score"],
"metadata": hit["_source"].get("metadata", {})
}
for hit in results["hits"]["hits"]
]
@app.post("/search")
async def search(request: SearchRequest):
try:
results = hybrid_search_rrf(
query=request.query,
tenant_id=request.tenant_id,
limit=request.limit
)
return {"results": results, "count": len(results)}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
If you need more control over the balance between BM25 and vector search, use a script_score approach with explicit weighting:
def hybrid_search_weighted(query: str, tenant_id: str, limit: int = 10,
bm25_weight: float = 0.5, vector_weight: float = 0.5):
query_embedding = model.encode(query).tolist()
search_body = {
"size": limit,
"query": {
"bool": {
"must": [
{"term": {"tenant_id": tenant_id}}
],
"should": [
{
"multi_match": {
"query": query,
"fields": ["title^2", "content"],
"type": "best_fields",
"operator": "or"
}
}
]
}
},
"rescore": {
"window_size": 50,
"query": {
"rescore_query": {
"script_score": {
"query": {"match_all": {}},
"script": {
"source": "_score * params.bm25_weight + (1 - Math.abs(params.vector_weight - cosineSimilarity(params.query_vector, 'embedding'))) * params.vector_weight",
"params": {
"query_vector": query_embedding,
"bm25_weight": bm25_weight,
"vector_weight": vector_weight
}
}
}
},
"query_weight": 0,
"rescore_query_weight": 1
}
}
}
results = es_client.search(index="documents", body=search_body)
return [
{
"id": hit["_source"]["id"],
"title": hit["_source"]["title"],
"content": hit["_source"]["content"],
"score": hit["_score"],
"metadata": hit["_source"].get("metadata", {})
}
for hit in results["hits"]["hits"]
]
For pure semantic search when keywords don’t matter, use vector similarity alone:
def vector_search(query: str, tenant_id: str, limit: int = 10):
query_embedding = model.encode(query).tolist()
search_body = {
"size": limit,
"query": {
"bool": {
"must": [
{"term": {"tenant_id": tenant_id}},
{
"script_score": {
"query": {"match_all": {}},
"script": {
"source": "cosineSimilarity(params.query_vector, 'embedding') + 1.0",
"params": {"query_vector": query_embedding}
}
}
}
]
}
}
}
results = es_client.search(index="documents", body=search_body)
return [hit["_source"] for hit in results["hits"]["hits"]]
Relevance Tuning Through Query Analysis
Default BM25 parameters rarely work for specialized domains. AI applications often need tuning for technical terminology, acronyms, and domain-specific synonyms.
Customize the analyzer for your domain:
# Update index settings with domain-specific analyzer
analyzer_config = {
"analysis": {
"analyzer": {
"ai_analyzer": {
"type": "custom",
"tokenizer": "standard",
"filter": [
"lowercase",
"stop",
"snowball",
"ai_synonyms"
]
}
},
"filter": {
"ai_synonyms": {
"type": "synonym",
"synonyms": [
"ml,machine learning",
"llm,large language model",
"rag,retrieval augmented generation",
"nlp,natural language processing"
]
}
}
}
}
# Close and reopen index with new settings
es_client.indices.close(index="documents")
es_client.indices.put_settings(index="documents", body={"settings": analyzer_config["analysis"]})
es_client.indices.open(index="documents")
For BM25 tuning, adjust k1 and b parameters based on your document characteristics:
# k1: controls term frequency saturation (default 1.2)
# b: controls field length normalization (default 0.75)
# Higher k1 = more weight to term frequency
# Higher b = more penalty for longer documents
bm25_settings = {
"settings": {
"index": {
"similarity": {
"custom_bm25": {
"type": "BM25",
"k1": 1.5,
"b": 0.5
}
}
}
},
"mappings": {
"properties": {
"content": {
"type": "text",
"similarity": "custom_bm25"
}
}
}
}
Test your tuning against real queries. Build a small evaluation set of queries with expected top results, then measure precision and recall as you adjust parameters.
Real-Time Indexing Patterns
For AI applications that ingest new data continuously, you need efficient update patterns. Elasticsearch supports several approaches.
For document updates with version control:
from datetime import datetime
def update_document_with_versioning(doc_id: str, new_content: str, tenant_id: str):
# Fetch current document
try:
current = es_client.get(index="documents", id=doc_id)
current_source = current["_source"]
current_version = current["_version"]
except:
current_source = {}
current_version = 0
# Generate new embedding
new_embedding = model.encode(new_content).tolist()
# Update with version tracking
updated_doc = {
**current_source,
"content": new_content,
"embedding": new_embedding,
"updated_at": datetime.utcnow().isoformat(),
"version": current_version + 1
}
es_client.index(index="documents", id=doc_id, body=updated_doc)
For high-volume ingestion, use a message queue to decouple indexing from the main application:
import asyncio
from typing import List
class IndexingQueue:
def __init__(self, batch_size: int = 100, flush_interval: int = 5):
self.batch_size = batch_size
self.flush_interval = flush_interval
self.queue: List[dict] = []
self.lock = asyncio.Lock()
async def add_document(self, doc: dict):
async with self.lock:
# Generate embedding
doc["embedding"] = model.encode(f"{doc['title']} {doc['content']}").tolist()
self.queue.append(doc)
if len(self.queue) >= self.batch_size:
await self.flush()
async def flush(self):
if not self.queue:
return
async with self.lock:
docs_to_index = self.queue[:]
self.queue = []
# Index in background
actions = [
{
"_op_type": "index",
"_index": "documents",
"_id": doc["id"],
**doc
}
for doc in docs_to_index
]
from elasticsearch.helpers import bulk
bulk(es_client, actions)
indexing_queue = IndexingQueue(batch_size=100, flush_interval=5)
@app.post("/ingest")
async def ingest_document(doc: dict):
await indexing_queue.add_document(doc)
return {"status": "queued"}
Multi-Tenancy and Data Isolation
For SaaS applications, isolate tenant data at query time using filters. The tenant_id field in every document enables this.
from fastapi import Depends, Header
from fastapi.security import HTTPBearer
security = HTTPBearer()
async def get_tenant_id(authorization = Header(None)) -> str:
# Extract tenant from JWT or header
# This is a placeholder; use your auth system
if not authorization:
raise HTTPException(status_code=401, detail="Unauthorized")
# Decode and extract tenant_id
tenant_id = extract_tenant_from_token(authorization)
return tenant_id
@app.post("/search")
async def search(request: SearchRequest, tenant_id: str = Depends(get_tenant_id)):
# Ensure tenant_id from request matches authenticated tenant
if request.tenant_id != tenant_id:
raise HTTPException(status_code=403, detail="Forbidden")
results = hybrid_search_rrf(
query=request.query,
tenant_id=tenant_id,
limit=request.limit
)
return {"results": results}
The bool query’s must clause ensures every search is filtered by tenant_id before scoring. This prevents data leakage and keeps queries efficient.
Integration with RAG Pipelines
For retrieval-augmented generation, you need fast, accurate retrieval that feeds context into your LLM. The hybrid search approach works well here because it balances relevance with diversity.
from openai import OpenAI
llm_client = OpenAI()
@app.post("/rag-query")
async def rag_query(question: str, tenant_id: str):
# Retrieve relevant documents
search_results = hybrid_search_weighted(
query=question,
tenant_id=tenant_id,
limit=5,
bm25_weight=0.4,
vector_weight=0.6
)
# Build context from top results
context = "
".join([
f"Document {i+1}: {result['title']}
{result['content']}"
for i, result in enumerate(search_results)
])
# Generate response with LLM
response = llm_client.chat.completions.create(
model="gpt-4",
messages=[
{
"role": "system",
"content": "You are a helpful assistant. Answer based on the provided context."
},
{
"role": "user",
"content": f"Context:
{context}
Question: {question}"
}
]
)
return {
"question": question,
"context_documents": search_results,
"answer": response.choices[0].message.content
}
For RAG, vector_weight higher than bm25_weight often works better because semantic relevance matters more than keyword matching for context quality.
Handling Concurrency and Performance
FastAPI’s async nature handles concurrent requests efficiently, but you need to manage Elasticsearch connection pooling and query timeouts.
from elasticsearch import Elasticsearch
from elasticsearch.connection_pool import ConnectionPool
# Configure connection pool
es_client = Elasticsearch(
["http://localhost:9200"],
max_retries=3,
retry_on_timeout=True,
timeout=30
)
# Add request timeout to queries
def search_with_timeout(query_body: dict, timeout: str = "5s"):
return es_client.search(
index="documents",
body=query_body,
timeout=timeout
)
For production, implement circuit breakers and fallback search strategies:
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def resilient_search(query: str, tenant_id: str):
try:
return await asyncio.to_thread(
hybrid_search_rrf,
query=query,
tenant_id=tenant_id
)
except Exception as e:
# Fallback to keyword-only search if vector search fails
return await asyncio.to_thread(
keyword_only_search,
query=query,
tenant_id=tenant_id
)
def keyword_only_search(query: str, tenant_id: str, limit: int = 10):
search_body = {
"size": limit,
"query": {
"bool": {
"must": [{"term": {"tenant_id": tenant_id}}],
"should": [{"multi_match": {"query": query, "fields": ["title^2", "content"]}}]
}
}
}
results = es_client.search(index="documents", body=search_body)
return [hit["_source"] for hit in results["hits"]["hits"]]
Deployment on Azure Container Apps
For production on Azure, containerize your FastAPI application and deploy alongside a managed Elasticsearch instance.
Dockerfile:
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
requirements.txt:
fastapi==0.104.1
uvicorn==0.24.0
elasticsearch==8.10.0
sentence-transformers==2.2.2
pydantic==2.5.0
python-dotenv==1.0.0
tenacity==8.2.3
openai==1.3.0
Deploy to Azure Container Apps using the Azure CLI:
az containerapp create \
--name search-api \
--resource-group my-rg \
--environment my-env \
--image myregistry.azurecr.io/search-api:latest \
--target-port 8000 \
--ingress external \
--env-vars ELASTICSEARCH_HOST=my-es.azure.example.com \
--env-vars ELASTICSEARCH_PORT=9200
For Elasticsearch, use Azure Cognitive Search or deploy Elasticsearch on Azure VMs. Cognitive Search is a managed service with built-in Azure integration. Elasticsearch on VMs gives you full control over indexing and scoring strategies. Choose based on your operational preferences and customization needs.
Monitoring and Observability
Add logging and metrics to track search quality and system health:
import logging
from prometheus_client import Counter, Histogram, start_http_server
import time
logger = logging.getLogger(__name__)
# Prometheus metrics
search_duration = Histogram('search_duration_seconds', 'Search query duration')
search_count = Counter('search_total', 'Total searches', ['tenant_id'])
search_errors = Counter('search_errors_total', 'Search errors', ['error_type'])
@app.post("/search")
async def search(request: SearchRequest):
start_time = time.time()
try:
results = hybrid_search_rrf(
query=request.query,
tenant_id=request.tenant_id,
limit=request.limit
)
duration = time.time() - start_time
search_duration.observe(duration)
search_count.labels(tenant_id=request.tenant_id).inc()
logger.info(
f"Search completed",
extra={
"query": request.query,
"tenant_id": request.tenant_id,
"result_count": len(results),
"duration_ms": duration * 1000
}
)
return {"results": results, "count": len(results)}
except Exception as e:
search_errors.labels(error_type=type(e).__name__).inc()
logger.error(f"Search failed: {str(e)}", exc_info=True)
raise HTTPException(status_code=500, detail="Search failed")
# Start metrics server
start_http_server(8001)
Set up alerts for high error rates, slow queries, and indexing lag. In production, aim for p99 search latency under 200ms for interactive applications, though this varies based on document size and query complexity.
Putting It Together
A production search system combines these pieces: hybrid query scoring that balances keywords and semantics, real-time indexing that keeps data fresh, multi-tenant isolation that protects data, and resilient deployment that handles failures gracefully.
Start simple with basic BM25 search, then add vector embeddings once you understand your relevance requirements. Tune parameters based on real query performance, not assumptions. Use metrics and logging to catch issues before they affect users.
The stack of FastAPI and Elasticsearch scales from MVP to millions of queries per day without fundamental architecture changes. You adjust shard count, tune scoring parameters, and add caching as load increases. That’s the practical advantage of building on battle-tested infrastructure rather than rolling your own.
Should I use BM25 or vector search for my AI application?
Use both. Hybrid search that combines BM25 and vector similarity gives you keyword precision when it matters (exact matches, technical terms) and semantic understanding for concept-based queries. For RAG pipelines specifically, hybrid search often produces better context because it balances relevance with diversity. Pure vector search works well if your domain has consistent semantic patterns, but most production systems benefit from the combination.
How do I handle real-time document updates without breaking search consistency?
Elasticsearch handles versioning automatically. For high-volume updates, queue them asynchronously and batch index every few seconds rather than indexing one document at a time. This reduces I/O overhead. For critical updates that must be immediately searchable, index synchronously but set a reasonable timeout (5 to 10 seconds). Use monitoring to catch indexing lag, which indicates your queue is overwhelmed.
What embedding model should I use for production search?
Start with sentence-transformers models like all-MiniLM-L6-v2 (384 dimensions, fast, good quality) or all-mpnet-base-v2 (768 dimensions, higher quality, slower). For specialized domains (medical, legal, technical), fine-tune on your own data or use domain-specific models. Larger models (1024 plus dimensions) improve relevance but increase storage and latency. Measure trade-offs with real queries before committing to production.
How do I scale Elasticsearch for high-concurrency search?
Increase shard count to distribute queries across nodes, add replicas for read throughput, and use query caching for repeated searches. Monitor query latency with Prometheus. If p99 latency exceeds your target, add more nodes rather than tuning queries. Elasticsearch scales horizontally well. Also implement circuit breakers and fallback search strategies so one slow query does not cascade failures.
Is Azure Cognitive Search a good alternative to Elasticsearch?
Both are valid choices. Azure Cognitive Search is fully managed and integrates well with Azure services. Elasticsearch gives you more control over indexing, scoring, and relevance tuning. If you need deep customization for AI workflows, Elasticsearch is a strong option. If you want minimal operations overhead and are already on Azure, Cognitive Search is a reasonable choice. Evaluate both against your specific requirements.
