Basic RAG — single query, single index, single response — works for many scenarios. But production systems often require more sophisticated patterns: searching across multiple data sources, maintaining conversation context, enforcing security boundaries, and evaluating quality. This lesson covers advanced RAG patterns for enterprise-grade applications.
Real-world applications often need to search across multiple data sources simultaneously — product docs, support tickets, internal wikis, and knowledge bases.
User Query
│
├──▶ Index 1: Product Docs ──┐
├──▶ Index 2: Support Tickets ──┼──▶ Merge & Rerank ──▶ LLM ──▶ Response
└──▶ Index 3: Internal Wiki ──┘
from azure.search.documents import SearchClient
from azure.identity import DefaultAzureCredential
credential = DefaultAzureCredential()
endpoint = "https://my-search.search.windows.net"
# Search multiple indexes in parallel
indexes = ["product-docs", "support-tickets", "internal-wiki"]
all_results = []
for index_name in indexes:
client = SearchClient(endpoint=endpoint, index_name=index_name,
credential=credential)
results = client.search(
search_text=query,
vector_queries=[VectorizableTextQuery(
text=query, k_nearest_neighbors=3, fields="content_vector"
)],
query_type="semantic",
semantic_configuration_name="default",
top=3
)
for r in results:
all_results.append({
"content": r["content"],
"source": f"{index_name}/{r['source_file']}",
"score": r.get("@search.reranker_score", r["@search.score"])
})
# Sort by score and take top results
all_results.sort(key=lambda x: x["score"], reverse=True)
top_context = all_results[:5]
In multi-turn conversations, you need to reformulate the user’s latest question in the context of chat history before searching.
User: "What models does Azure OpenAI support?"
AI: "Azure OpenAI supports GPT-4o, GPT-4.1, GPT-3.5 Turbo..."
User: "Which one is cheapest?" ← Ambiguous without context
from openai import AzureOpenAI
client = AzureOpenAI(
azure_endpoint="https://my-openai.openai.azure.com/",
api_key="<key>",
api_version="2024-10-21"
)
def rewrite_query(chat_history: list, latest_query: str) -> str:
"""Rewrite the user's query using chat history for context."""
messages = [
{"role": "system", "content": (
"Rewrite the user's latest question as a standalone search query. "
"Use the chat history for context. Return ONLY the rewritten query."
)}
]
messages.extend(chat_history)
messages.append({"role": "user", "content": latest_query})
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0,
max_tokens=200
)
return response.choices[0].message.content
# Example usage
history = [
{"role": "user", "content": "What models does Azure OpenAI support?"},
{"role": "assistant", "content": "Azure OpenAI supports GPT-4o, GPT-4.1..."}
]
rewritten = rewrite_query(history, "Which one is cheapest?")
# Result: "Which Azure OpenAI model is the cheapest: GPT-4o, GPT-4.1, or GPT-3.5 Turbo?"
📖 Reference: Chat with your data using Azure OpenAI
In agentic RAG, an AI agent decides when, what, and how to retrieve — rather than retrieving on every query.
| Scenario | Approach |
|---|---|
| Simple factual Q&A | Standard RAG (always retrieve) |
| Mixed queries (some need docs, some don’t) | Agentic RAG (agent decides) |
| Multi-step research tasks | Agentic RAG with iterative retrieval |
| Tool-augmented workflows | Agentic RAG with function calling |
tools = [
{
"type": "function",
"function": {
"name": "search_knowledge_base",
"description": "Search the company knowledge base for relevant information",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query"
},
"category": {
"type": "string",
"enum": ["product", "support", "policy", "technical"],
"description": "Category to filter results"
}
},
"required": ["query"]
}
}
}
]
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "You are a helpful assistant. Use the search "
"tool when the user asks about company products, policies, or technical docs."},
{"role": "user", "content": user_query}
],
tools=tools,
tool_choice="auto" # Model decides whether to call the tool
)
Enterprise RAG must enforce access control — users should only retrieve documents they’re authorized to see.
def search_with_security(query: str, user_groups: list[str]):
"""Search with document-level security filtering."""
# Build OData filter for user's security groups
group_filters = " or ".join(
[f"allowed_groups/any(g: g eq '{group}')" for group in user_groups]
)
security_filter = f"({group_filters})"
results = search_client.search(
search_text=query,
vector_queries=[VectorizableTextQuery(
text=query, k_nearest_neighbors=5, fields="content_vector"
)],
filter=security_filter, # Only return authorized documents
query_type="semantic",
semantic_configuration_name="default",
top=5
)
return list(results)
# Usage: user only sees docs their groups can access
results = search_with_security(
query="What is the Q3 revenue forecast?",
user_groups=["finance-team", "executives"]
)
💡 Tip: Implement document-level security from day one in enterprise scenarios. Retrofitting access control into an existing RAG system is significantly harder than building it in from the start.
📖 Reference: Security filters in Azure AI Search
Caching reduces latency and cost for frequently asked questions.
import hashlib
import json
class RAGCache:
def __init__(self, ttl_seconds=3600):
self.cache = {}
self.ttl = ttl_seconds
def _cache_key(self, query: str, filters: dict = None) -> str:
raw = json.dumps({"query": query.lower().strip(), "filters": filters},
sort_keys=True)
return hashlib.sha256(raw.encode()).hexdigest()
def get(self, query: str, filters: dict = None):
key = self._cache_key(query, filters)
entry = self.cache.get(key)
if entry and (time.time() - entry["timestamp"]) < self.ttl:
return entry["results"]
return None
def set(self, query: str, results: list, filters: dict = None):
key = self._cache_key(query, filters)
self.cache[key] = {"results": results, "timestamp": time.time()}
For production, use Azure Cache for Redis instead of in-memory caching.
Evaluation is essential to know whether your RAG system is actually working well.
| Metric | What It Measures | Target |
|---|---|---|
| Groundedness | Is the answer supported by retrieved context? | > 4.0 / 5.0 |
| Relevance | Are retrieved documents relevant to the query? | > 4.0 / 5.0 |
| Faithfulness | Does the answer accurately represent the sources? | > 4.0 / 5.0 |
| Answer completeness | Does the answer fully address the question? | > 3.5 / 5.0 |
| Retrieval precision | What fraction of retrieved docs are relevant? | > 0.7 |
from azure.ai.evaluation import GroundednessEvaluator, RelevanceEvaluator
groundedness_eval = GroundednessEvaluator(model_config={
"azure_endpoint": "https://my-openai.openai.azure.com/",
"azure_deployment": "gpt-4o",
"api_version": "2024-10-21"
})
result = groundedness_eval(
query="What is the return policy?",
context="Our return policy allows returns within 30 days of purchase...",
response="You can return items within 30 days of purchase for a full refund."
)
print(f"Groundedness score: {result['groundedness']}")
📖 Reference: Evaluate your generative AI app
| ← Lesson 2 | Next: Knowledge Check → |