5 min read
Hybrid Search Patterns in Azure Cognitive Search
Hybrid search combines traditional keyword search with vector similarity search to get the best of both worlds. This approach handles both exact matches and semantic similarity.
Why Hybrid Search
- Keyword Search: Great for exact matches, filters, and facets
- Vector Search: Excellent for semantic similarity and concept matching
- Hybrid: Combines precision of keywords with recall of vectors
Basic Hybrid Search
from azure.search.documents import SearchClient
from azure.search.documents.models import Vector
from azure.core.credentials import AzureKeyCredential
import openai
endpoint = "https://mysearchservice.search.windows.net"
credential = AzureKeyCredential("your-admin-key")
search_client = SearchClient(
endpoint=endpoint,
index_name="hybrid-index",
credential=credential
)
def hybrid_search(query, k=10):
"""Perform hybrid search combining keyword and vector"""
# Generate query embedding
query_embedding = get_embedding(query)
results = search_client.search(
search_text=query, # Keyword component
vectors=[
Vector(
value=query_embedding,
k=k,
fields="content_vector"
)
],
select=["id", "title", "content", "category"],
top=k
)
return list(results)
# Execute hybrid search
results = hybrid_search("machine learning model deployment best practices")
for r in results:
print(f"{r['title']}: {r['@search.score']:.4f}")Weighted Hybrid Search
class WeightedHybridSearch:
"""Hybrid search with configurable weights"""
def __init__(self, search_client, embedding_func):
self.search_client = search_client
self.get_embedding = embedding_func
def search(self, query, keyword_weight=0.5, vector_weight=0.5, k=10):
"""Search with weighted combination of keyword and vector scores"""
# Get keyword results
keyword_results = self._keyword_search(query, k * 2)
# Get vector results
vector_results = self._vector_search(query, k * 2)
# Combine and rerank
combined = self._combine_results(
keyword_results,
vector_results,
keyword_weight,
vector_weight
)
return combined[:k]
def _keyword_search(self, query, k):
results = self.search_client.search(
search_text=query,
select=["id", "title", "content"],
top=k
)
return {r["id"]: {"doc": r, "score": r["@search.score"]} for r in results}
def _vector_search(self, query, k):
embedding = self.get_embedding(query)
results = self.search_client.search(
search_text=None,
vectors=[Vector(value=embedding, k=k, fields="content_vector")],
select=["id", "title", "content"]
)
return {r["id"]: {"doc": r, "score": r["@search.score"]} for r in results}
def _combine_results(self, keyword_results, vector_results, kw_weight, vec_weight):
"""Combine results using Reciprocal Rank Fusion or weighted scoring"""
all_ids = set(keyword_results.keys()) | set(vector_results.keys())
combined = []
for doc_id in all_ids:
kw_score = keyword_results.get(doc_id, {}).get("score", 0)
vec_score = vector_results.get(doc_id, {}).get("score", 0)
# Normalize scores (assuming max score of 1 for simplicity)
normalized_kw = kw_score / max(r["score"] for r in keyword_results.values()) if keyword_results else 0
normalized_vec = vec_score / max(r["score"] for r in vector_results.values()) if vector_results else 0
combined_score = kw_weight * normalized_kw + vec_weight * normalized_vec
doc = keyword_results.get(doc_id, vector_results.get(doc_id))["doc"]
combined.append({
**doc,
"combined_score": combined_score,
"keyword_score": kw_score,
"vector_score": vec_score
})
return sorted(combined, key=lambda x: x["combined_score"], reverse=True)
# Usage
hybrid = WeightedHybridSearch(search_client, get_embedding)
results = hybrid.search("ML deployment", keyword_weight=0.3, vector_weight=0.7)Reciprocal Rank Fusion
def reciprocal_rank_fusion(result_lists, k=60):
"""Combine multiple result lists using RRF"""
scores = {}
for results in result_lists:
for rank, doc in enumerate(results, 1):
doc_id = doc["id"]
if doc_id not in scores:
scores[doc_id] = {"doc": doc, "score": 0}
# RRF formula: 1 / (k + rank)
scores[doc_id]["score"] += 1 / (k + rank)
# Sort by combined RRF score
combined = sorted(scores.values(), key=lambda x: x["score"], reverse=True)
return [{"doc": item["doc"], "rrf_score": item["score"]} for item in combined]
# Apply RRF to hybrid search
keyword_results = list(search_client.search(search_text="machine learning", top=50))
vector_results = list(search_client.search(
search_text=None,
vectors=[Vector(value=get_embedding("machine learning"), k=50, fields="content_vector")]
))
fused_results = reciprocal_rank_fusion([keyword_results, vector_results])Hybrid Search with Semantic Ranking
from azure.search.documents.models import QueryType, QueryCaptionType
def semantic_hybrid_search(query, k=10):
"""Hybrid search with semantic reranking"""
query_embedding = get_embedding(query)
results = search_client.search(
search_text=query,
query_type=QueryType.SEMANTIC,
semantic_configuration_name="my-semantic-config",
vectors=[
Vector(
value=query_embedding,
k=50, # Over-retrieve for reranking
fields="content_vector"
)
],
query_caption=QueryCaptionType.EXTRACTIVE,
top=k
)
return list(results)
# Results are semantically reranked
results = semantic_hybrid_search("How do I deploy ML models to production?")
for r in results:
print(f"{r['title']}")
print(f" Reranker Score: {r.get('@search.reranker_score', 'N/A')}")
captions = r.get("@search.captions", [])
if captions:
print(f" Caption: {captions[0].text}")Query Expansion for Hybrid Search
def expanded_hybrid_search(query, k=10):
"""Hybrid search with query expansion"""
# Generate expanded queries using embeddings
query_embedding = get_embedding(query)
# Also search for synonyms/related terms
related_queries = generate_related_queries(query) # Using LLM or thesaurus
# Combine all keyword searches
all_results = []
# Original query search
results = search_client.search(search_text=query, top=k*2)
all_results.append(list(results))
# Related queries
for related in related_queries[:3]:
results = search_client.search(search_text=related, top=k)
all_results.append(list(results))
# Vector search
results = search_client.search(
search_text=None,
vectors=[Vector(value=query_embedding, k=k*2, fields="content_vector")]
)
all_results.append(list(results))
# Combine using RRF
return reciprocal_rank_fusion(all_results)[:k]
def generate_related_queries(query):
"""Generate related queries using OpenAI"""
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "Generate 3 related search queries."},
{"role": "user", "content": f"Original query: {query}"}
]
)
return response.choices[0].message.content.strip().split("\n")Adaptive Hybrid Search
class AdaptiveHybridSearch:
"""Automatically adjust weights based on query characteristics"""
def __init__(self, search_client, embedding_func):
self.search_client = search_client
self.get_embedding = embedding_func
def search(self, query, k=10):
"""Search with adaptive weighting"""
query_type = self._classify_query(query)
weights = self._get_weights(query_type)
return self._hybrid_search(query, weights, k)
def _classify_query(self, query):
"""Classify query type"""
# Short, specific queries -> favor keyword
if len(query.split()) <= 3:
return "keyword_focused"
# Questions -> favor semantic/vector
if query.strip().endswith("?") or query.lower().startswith(("how", "what", "why", "when")):
return "semantic_focused"
# Quotes indicate exact match need
if '"' in query:
return "exact_match"
return "balanced"
def _get_weights(self, query_type):
"""Get weights based on query type"""
weights = {
"keyword_focused": {"keyword": 0.7, "vector": 0.3},
"semantic_focused": {"keyword": 0.3, "vector": 0.7},
"exact_match": {"keyword": 0.9, "vector": 0.1},
"balanced": {"keyword": 0.5, "vector": 0.5}
}
return weights.get(query_type, weights["balanced"])
def _hybrid_search(self, query, weights, k):
"""Execute hybrid search with given weights"""
embedding = self.get_embedding(query)
results = self.search_client.search(
search_text=query,
vectors=[Vector(value=embedding, k=k, fields="content_vector")],
top=k
)
return list(results)
# Usage
adaptive = AdaptiveHybridSearch(search_client, get_embedding)
results = adaptive.search("How to implement neural networks?")Hybrid search patterns enable building search systems that handle diverse query types effectively.