A sales rep asked for "forklifts in Texas with 10,000+ lb capacity under active warranty," and my own architecture started arguing with itself.
Three databases. Three answers, each one correct in isolation, each one wrong in production for a different reason. Azure AI Search could do synonym expansion and semantic reranking over my CRM-shaped records. Postgres + pgvector could do fast similarity over Zoom transcripts without duplicating data into yet another index. And the CRM, the messy real-time source, was sometimes the only place the newest truth existed.
This is Part 8 of my series "How to Architect an Enterprise AI System (And Why the Engineer Still Matters)". Part 7 was about building a reliable sync engine against a CRM API with undocumented behaviors. This one is what I built on top of that: a search router that decides, per query, which backend deserves the job.
The key insight (and why the naive approach fails)
The naive approach is seductive. Pick one search system, shove everything into it, call it a day.
I had two obvious "one tool" options:
- Put everything into Azure AI Search. It's great for structured indexes and has features I actually want for CRM data: faceted search, semantic reranking, geo-distance scoring, synonym maps.
- Put everything into Postgres + pgvector. My transcripts already live in Postgres, and pgvector gives me semantic retrieval without maintaining a parallel indexing universe.
Neither of those fails in the obvious way. The failure mode is operational dishonesty:
- If I duplicate transcripts into Azure AI Search, I'm signing up for index duplication and two copies that can disagree.
- If I force CRM-shaped search into pgvector-only semantics, I lose the structured search affordances that make CRM data usable.
- If I pretend sync is perfectly real-time, I'll return stale answers the moment a CRM record changes and the sync hasn't caught up.
So I made a different bet: I route queries to the right backend based on what the user is asking for, and I keep a real-time CRM fallback when the index can't be trusted.
That routing layer is the whole trick. Unglamorous, and also the difference between a demo and a system.
What I actually built: UnifiedSearchService
The core decision in this post is explicit:
- Azure AI Search for structured CRM data
- pgvector for Zoom transcripts (because the data is already in Postgres)
- The CRM API as a real-time fallback
I implemented that as a single service: UnifiedSearchService. It's a dispatcher.
It looks at the query, and sometimes at the target domain (assets vs contracts vs companies vs transcripts). It picks a backend. Then it shapes the request into that backend's native strengths and returns results in a unified format, so the rest of the system doesn't have to care which one answered.
One piece of pragmatism I'll come back to: during migration I kept two Azure AI Search implementations, SDK and HTTP, and a feature flag decides which one is active.
Architecture at a glance
The non-obvious part is that I'm not trying to hide the differences between these systems. I want to exploit them without letting them leak into the product.
How keyword-based routing works (and why it's intentionally "dumb")
The routing logic is keyword-based dispatch. That's deliberate.
I've built enough AI systems to know that routing made "smart" in the wrong way turns flaky. I want something deterministic and easy to debug. When a query smells like transcript search, it goes to pgvector. When it smells like CRM structured retrieval, it goes to Azure AI Search. When freshness matters, I have a CRM escape hatch.
import asyncio
from dataclasses import dataclass, field
from typing import Any, Optional
@dataclass
class SearchResult:
id: str
score: float
source: str # "azure", "pgvector", "crm"
title: str
snippet: str
metadata: dict = field(default_factory=dict)
# Keyword triggers for each backend
ASSET_KEYWORDS = {"asset", "equipment", "lease", "fleet", "vehicle", "forklift"}
CONTRACT_KEYWORDS = {"contract", "deal", "proposal", "renewal", "closed", "won"}
TRANSCRIPT_KEYWORDS = {"meeting", "call", "transcript", "discussion", "said", "mentioned"}
class UnifiedSearchService:
def __init__(self, azure_client, pg_pool, crm_client, embedder):
self.azure = azure_client
self.pg = pg_pool
self.crm = crm_client
self.embedder = embedder
async def search(
self,
query: str,
force_live: bool = False,
top: int = 20,
) -> list[SearchResult]:
"""
Route a query to the right backend based on keyword dispatch.
Returns unified results regardless of which backend answered.
"""
if force_live:
return await self._search_crm_live(query, top)
query_lower = query.lower()
tasks = []
# Dispatch to backends based on keyword triggers
if any(w in query_lower for w in ASSET_KEYWORDS | CONTRACT_KEYWORDS):
tasks.append(self._search_azure(query, top))
if any(w in query_lower for w in TRANSCRIPT_KEYWORDS):
tasks.append(self._search_transcripts(query, top))
# Default: if no keywords matched, try Azure (structured is the safe default)
if not tasks:
tasks.append(self._search_azure(query, top))
# Run all matched backends concurrently
results_lists = await asyncio.gather(*tasks, return_exceptions=True)
# Merge results, skip backends that errored
merged: list[SearchResult] = []
for result in results_lists:
if isinstance(result, Exception):
continue # log it, don't crash
merged.extend(result)
return merged
What surprised me is how often the dumb version is more reliable than the clever one. A misroute is a keyword or a rule I can go change. The clever version leaves me reverse-engineering why a model picked the wrong path, which is a much worse evening.
Azure AI Search: multi-vector scoring with 4 embedding fields
For CRM data I use Azure AI Search, and the implementation leans on a multi-vector representation:
profile_vectorexperience_vectorspecs_vectorgeneral_vector
Those four vectors come from my embedding pipeline. The embedding agent generates exactly those four parallel embeddings per record.
The idea is simple. An equipment asset might match a query because its specs are perfect even if its general description text is noisy, or because its service history aligns even if the specs list is sparse. One vector can lie. Four give me multiple angles on the same record.
The Azure side then does weighted fusion across those vectors using RRF (Reciprocal Rank Fusion).
from azure.search.documents import SearchClient
from azure.search.documents.models import (
VectorizableTextQuery,
QueryType,
)
from typing import Any
async def search_azure_multi_vector(
client: SearchClient,
query: str,
query_embedding: list[float],
filters: Optional[dict[str, Any]] = None,
top: int = 20,
) -> list[SearchResult]:
"""
Multi-vector search across 4 embedding fields with semantic reranking.
Oversampling retrieves 2x candidates before reranking narrows to top-k.
"""
# Build vector queries: one per embedding field
vector_queries = [
VectorizableTextQuery(
text=query,
k_nearest_neighbors=top * 2, # oversample for better reranking
fields=field_name,
)
for field_name in [
"profile_vector",
"experience_vector",
"specs_vector",
"general_vector",
]
]
# Build OData filter from structured constraints
odata_filter = _build_odata_filter(filters) if filters else None
results = client.search(
search_text=query,
vector_queries=vector_queries,
filter=odata_filter,
query_type=QueryType.SEMANTIC,
semantic_configuration_name="default",
top=top,
)
return [
SearchResult(
id=r["id"],
score=r.get("@search.reranker_score", r.get("@search.score", 0)),
source="azure",
title=r.get("title", ""),
snippet=r.get("description", "")[:200],
metadata={k: v for k, v in r.items() if not k.startswith("@")},
)
for r in results
]
def _build_odata_filter(filters: dict[str, Any]) -> str:
"""
Convert a parameter dict into an Azure AI Search OData filter.
Handles booleans, numerics, strings, and collection fields.
"""
clauses = []
for field_name, value in filters.items():
if isinstance(value, bool):
clauses.append(f"{field_name} eq {str(value).lower()}")
elif isinstance(value, (int, float)):
clauses.append(f"{field_name} ge {value}")
elif isinstance(value, list):
# OR across multiple values
sub = " or ".join(f"{field_name} eq '{v}'" for v in value)
clauses.append(f"({sub})")
else:
clauses.append(f"{field_name} eq '{value}'")
return " and ".join(clauses)
I learned the hard way that "just do one embedding similarity" is how you get results that look plausible and then fall apart at the edges. The multi-vector setup gives me a way to recover when one representation is incomplete.
Semantic reranking (and automatic fallback)
Azure AI Search semantic reranking is a feature I use when it's available, and I designed the system to fall back automatically when it isn't.
That fallback matters because availability is part of correctness. If reranking is intermittently unavailable and your system depends on it, your relevance will flap.
So I try not to get romantic about reranking. It's one tool among several.
pgvector: transcripts stay in Postgres
For Zoom transcripts, I route to Postgres + pgvector.
The decision here is almost dull, which is exactly why it's right:
- The transcript data already lives in Postgres.
- Duplicating it into Azure AI Search would create a second copy that can wander away from the first.
- pgvector gives me semantic retrieval where the data already is.
import asyncpg
from openai import AsyncOpenAI
openai = AsyncOpenAI()
async def search_transcripts(
pool: asyncpg.Pool,
query: str,
top: int = 20,
) -> list[SearchResult]:
"""
Semantic search over Zoom transcripts stored in Postgres with pgvector.
Generates a query embedding, then finds nearest neighbors by cosine distance.
Falls back to full-text search if embedding generation fails.
"""
try:
# Generate query embedding
resp = await openai.embeddings.create(
input=query,
model="text-embedding-3-large",
)
query_embedding = resp.data[0].embedding
# Cosine similarity search via pgvector
async with pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT
id,
title,
transcript_text,
1 - (transcript_embedding <=> $1::vector) AS similarity
FROM zoom_transcripts
WHERE transcript_embedding IS NOT NULL
ORDER BY transcript_embedding <=> $1::vector
LIMIT $2
""",
str(query_embedding),
top,
)
except Exception:
# Fallback: full-text search when embeddings are unavailable
async with pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT
id,
title,
transcript_text,
ts_rank(to_tsvector('english', transcript_text),
plainto_tsquery('english', $1)) AS similarity
FROM zoom_transcripts
WHERE to_tsvector('english', transcript_text)
@@ plainto_tsquery('english', $1)
ORDER BY similarity DESC
LIMIT $2
""",
query,
top,
)
return [
SearchResult(
id=str(row["id"]),
score=float(row["similarity"]),
source="pgvector",
title=row["title"],
snippet=row["transcript_text"][:200],
)
for row in rows
]
The subtle win is operational. I'm not maintaining a second indexing pipeline for transcripts. Start one of those and the work stops being search and becomes reconciliation.
The CRM API as the real-time fallback
Even with a good sync engine (Part 7), there are moments where the index is not the source of truth.
So the router keeps a CRM API real-time fallback.
I didn't add that path for performance. I added it for correctness. When the system suspects that the indexed view might be stale for a query that needs freshness, it can ask the CRM directly.
Being able to ask the CRM directly is what keeps the system truthful. It's also what lets me make stronger caching and indexing decisions elsewhere, because I'm not pretending the index is always perfect.
The migration pragmatism: USE_AZURE_SEARCH_SDK
I maintained both an SDK-based and an HTTP-based Azure AI Search implementation during a migration, and I controlled it with a feature flag: USE_AZURE_SEARCH_SDK.
Unsexy. Also one of those decisions that makes everything else possible:
- It lets me ship improvements without betting the entire system on a big-bang switch.
- It gives me a controlled rollback path.
- It makes it possible to compare behavior between implementations.
I'm calling this out because it generalizes. When you're changing a foundational backend, you want a kill switch and a way to run both paths.
A concrete example: one query, three tiers of truth
Back to the query I use to explain the system:
"forklifts in Texas with 10,000+ lb capacity under active warranty"
The router treats that as a structured CRM-style query:
- "forklifts" is an equipment type constraint.
- "Texas" is a geo constraint.
- "10,000+ lb capacity" is a numeric spec constraint.
- "active warranty" is a status filter.
Tier 1: Azure AI Search (structured CRM indexes)
Azure AI Search is the natural home for this query because it's built for:
- structured constraints (equipment type, numeric thresholds like capacity)
- geo filtering/scoring
- semantic reranking
- synonym maps for equipment domain language
This is also where my equipment domain synonym maps earn their keep. Different teams say "forklift" and "lift truck" and "material handler" for the same machine, and synonym maps are one of the few scaling tools that doesn't require model calls.
Tier 2: pgvector (what it would look like)
If I forced this query into pgvector-only semantics, I could retrieve assets whose text mentions forklifts and Texas and capacity. Then I'd be rebuilding structured filtering logic in the application layer.
Doable, sure. It's still the wrong tool for the data shape.
Tier 3: The CRM fallback (when freshness wins)
If the query is run in a context where the newest CRM edits matter (or the index is suspected stale), the router can choose the CRM API path.
The CRM is not where I want to do heavy semantic retrieval. I go there when I need the newest truth.
The tradeoff: you own a router now
The limitation of this approach is real: you're accepting that there isn't one perfect search backend.
That costs you:
- a unified result shape
- consistent logging, so you can see which backend answered
- routing rules that stay understandable to a human
What you get back is each backend doing what it's good at:
- Azure AI Search handles CRM-shaped retrieval with semantic reranking and synonym maps.
- pgvector handles transcript similarity where the data already lives.
- The CRM provides the escape hatch when the index can't be trusted.
My one analogy for this is a workshop. I don't use a table saw to tighten a bolt. I keep the bolt near the wrench, the wood near the saw, and after enough repetitions, reaching for the right tool stops being a decision.
Closing
Searching three backends is the least interesting thing UnifiedSearchService does. The win is that it lets me stop arguing with myself about which backend is "the one true system." I built a router that respects the shape of the data, keeps transcripts where they already live, and still has the humility to ask the CRM when real-time truth matters.
No search framework made that decision for me. It came from understanding the data well enough to know that "unified" doesn't have to mean "identical"... which is the same judgment the next piece needs: dual-channel streaming with interruptible UX, in Part 9.