Agentic RAG on Azure AI Search with plan, evaluate, and rewrite loops

Enterprise agentic RAG on Azure is more than “retrieve then answer.” Production systems plan, search with hybrid retrieval, evaluate evidence, rewrite queries, synthesize with grounding, and enforce ACL and hallucination guards—typically orchestrated as a LangGraph-style loop over Azure AI Search.

Opening sketches of the stack and flow:

NAIVE RAG + TOOLS (Linear Tool Calling)
User Query ──► LLM Router ──► [ Call Search Tool ] ──► Return Documents ──► LLM Answer

PRODUCTION AGENTIC RAG (Stateful Cognitive Loop)
User Query ──► Query Decomposer ──► State Graph Orchestrator ◄──► Azure AI Search
                                          │
                                   Evaluator / Critic
                                          │
                    ┌─────────────────────┴─────────────────────┐
             [ Gap Detected ]                           [ Context Sufficient ]
                    │                                           │
          Query Rewriter / Sub-Query                   Grounded Synthesis

Architecture and system design

Separate planning, retrieval, evaluation, rewriting, and synthesis into nodes with shared state. Azure AI Search holds the indexed corpus with security trimming. The agent runtime owns retries and reflection; the search service owns relevance and filters.

Core architectural components

• Index + skillsets / push API for documents
• Hybrid lexical + vector queries
• Agent state for goals, subqueries, evidence, scores
• Guardrails for ACL and grounding
• Telemetry spanning nodes

Azure AI Search: foundation of agentic retrieval

Hybrid search and filters make agentic loops feasible: each reflection can adjust filters and query text without rebuilding the app. Semantic rankers and vector fields coexist in one index definition.

Infrastructure setup: index schema (Python SDK)

import os
from azure.core.credentials import AzureKeyCredential
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex,
    SimpleField,
    SearchField,
    SearchFieldDataType,
    SearchableField,
    VectorSearch,
    HnswAlgorithmConfiguration,
    VectorSearchProfile,
    AzureOpenAIVectorizer,
    AzureOpenAIVectorizerParameters,
    SemanticConfiguration,
    SemanticPrioritizedFields,
    SemanticField,
    SemanticSearch
)
# Initialize Service Credentials
endpoint = os.environ["AZURE_SEARCH_ENDPOINT"]
credential = AzureKeyCredential(os.environ["AZURE_SEARCH_KEY"])
client = SearchIndexClient(endpoint=endpoint, credential=credential)
index_name = "enterprise-agentic-kb"
# Define Vectorizer & HNSW Profile
vector_search = VectorSearch(
    algorithms=[
        HnswAlgorithmConfiguration(name="my-hnsw-config")
    ],
    profiles=[
        VectorSearchProfile(
            name="my-vector-profile",
            algorithm_configuration_name="my-hnsw-config",
            vectorizer="my-openai-vectorizer"
        )
    ],
    vectorizers=[
        AzureOpenAIVectorizer(
            name="my-openai-vectorizer",
            parameters=AzureOpenAIVectorizerParameters(
                resource_url=os.environ["AZURE_OPENAI_ENDPOINT"],
                deployment_name="text-embedding-3-large",
                model_name="text-embedding-3-large",
                api_key=os.environ["AZURE_OPENAI_KEY"]
            )
        )
    ]
)
# Define Semantic Ranking Configuration
semantic_search = SemanticSearch(
    configurations=[
        SemanticConfiguration(
            name="agentic-semantic-config",
            prioritized_fields=SemanticPrioritizedFields(
                title_field=SemanticField(field_name="title"),
                content_fields=[SemanticField(field_name="content")],
                keywords_fields=[SemanticField(field_name="category")]
            )
        )
    ]
)
# Build Schema
fields = [
    SimpleField(name="id", type=SearchFieldDataType.String, key=True),
    SearchableField(name="title", type=SearchFieldDataType.String, analyzer_name="en.microsoft"),
    SearchableField(name="content", type=SearchFieldDataType.String, analyzer_name="en.microsoft"),
    SearchableField(name="category", type=SearchFieldDataType.String, filterable=True, facetable=True),
    SearchableField(name="security_group", type=SearchFieldDataType.String, filterable=True),
    SearchField(
        name="content_vector",
        type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
        searchable=True,
        vector_search_dimensions=3072,
        vector_search_profile_name="my-vector-profile"
    )
]
index = SearchIndex(
    name=index_name,
    fields=fields,
    vector_search=vector_search,
    semantic_search=semantic_search
)
client.create_or_update_index(index)
print(f"Index {index_name} successfully created/updated.")

Production-grade deep implementation

Wire clients, credentials, and retry policies once; nodes should not each invent authentication.

[ START ]
           │
           ▼
   ┌───────────────┐
   │ Plan & Split  │
   └───────┬───────┘
           │
           ▼
   ┌───────────────┐
   │ Hybrid Search │
   └───────┬───────┘
           │
           ▼
   ┌───────────────┐
   │  Evaluator    │◄───────────┐ (Loop Limit: 3)
   └───────┬───────┘            │
           │                    │
     [ Sufficient? ] ──NO──►[ Rewrite ]
           │
          YES
           │
           ▼
   ┌───────────────┐
   │ Grounded Syn  │
   └───────┬───────┘
           │
           ▼
        [ END ]

Setup and agent state schema

import json
from typing import List, TypedDict, Annotated
import operator
from pydantic import BaseModel, Field
from azure.search.documents import SearchClient
from azure.search.documents.models import VectorizableTextQuery
from langchain_openai import AzureChatOpenAI
# 1. State Object
class AgenticRAGState(TypedDict):
    original_query: str
    sub_queries: List[str]
    retrieved_documents: Annotated[List[dict], operator.add]
    evaluation_score: float
    is_sufficient: bool
    iterations: int
    final_response: str
    security_context: str  # For Document-Level ACL Security Trimming
# Initialize LLM
llm = AzureChatOpenAI(
    azure_deployment="gpt-4o",
    api_version="2024-08-01-preview",
    temperature=0.0
)

State tracks the user question, decomposed subqueries, hits, evaluation scores, rewrite count, and final answer drafts.

Node 1: planner and query decomposer

class QueryDecompositionPlan(BaseModel):
    sub_queries: List[str] = Field(description="List of targeted sub-queries to retrieve complete context.")
planner_system_prompt = """You are an expert retrieval planner.
Analyze the given user input and break it down into 1 to 3 distinct sub-queries optimized for a corporate knowledge base.
If the query is straightforward, return a list containing only the original query."""
def plan_and_decompose_node(state: AgenticRAGState) -> dict:
    structured_llm = llm.with_structured_output(QueryDecompositionPlan)
    messages = [
        {"role": "system", "content": planner_system_prompt},
        {"role": "user", "content": state["original_query"]}
    ]
    response = structured_llm.invoke(messages)
    return {
        "sub_queries": response.sub_queries,
        "iterations": state.get("iterations", 0) + 1
    }

Complex questions become ranked subqueries the rest of the graph can execute.

Node 2: hybrid search execution

search_client = SearchClient(
    endpoint=endpoint,
    index_name=index_name,
    credential=credential
)
def retrieve_hybrid_node(state: AgenticRAGState) -> dict:
    retrieved_docs = []

    for sub_query in state["sub_queries"]:
        vector_query = VectorizableTextQuery(
            text=sub_query,
            k_nearest_neighbors=5,
            fields="content_vector"
        )

        # Security Trimming Filter
        security_filter = f"security_group eq '{state['security_context']}'"

        results = search_client.search(
            search_text=sub_query,
            vector_queries=[vector_query],
            filter=security_filter,
            query_type="semantic",
            semantic_configuration_name="agentic-semantic-config",
            top=3
        )

        for doc in results:
            retrieved_docs.append({
                "id": doc["id"],
                "title": doc["title"],
                "content": doc["content"],
                "reranker_score": doc.get("@search.reranker_score", 0.0),
                "query_used": sub_query
            })

    # Deduplicate by document ID
    unique_docs = {doc['id']: doc for doc in retrieved_docs}.values()
    return {"retrieved_documents": list(unique_docs)}

Combine keyword and vector channels; apply security filters from the caller’s identity context.

Node 3: evaluator and reflection loop

class ContextEvaluation(BaseModel):
    is_sufficient: bool = Field(description="True if context is enough to completely and accurately answer the prompt.")
    missing_aspects: str = Field(description="Description of what details are still missing.")
    score: float = Field(description="Confidence score between 0.0 and 1.0.")
evaluator_prompt = """You are a rigorous context critic evaluating retrieved document snippets.
Evaluate if the provided documents contain enough factual evidence to answer the user query completely.
User Query: {query}
Retrieved Documents:
{docs}
"""
def evaluate_context_node(state: AgenticRAGState) -> dict:
    structured_llm = llm.with_structured_output(ContextEvaluation)

    formatted_docs = "\n\n".join([
        f"Doc ID: {d['id']}\nTitle: {d['title']}\nContent: {d['content']}"
        for d in state["retrieved_documents"]
    ])

    prompt = evaluator_prompt.format(
        query=state["original_query"],
        docs=formatted_docs
    )

    eval_result = structured_llm.invoke(prompt)

    return {
        "is_sufficient": eval_result.is_sufficient,
        "evaluation_score": eval_result.score
    }

Score sufficiency and faithfulness. If evidence is weak, route to rewrite instead of answering.

Node 4: adaptive query rewriter

class QueryRewrite(BaseModel):
    new_sub_queries: List[str] = Field(description="Refined queries targeting missing information.")
rewrite_prompt = """The previous retrieval step failed to gather complete context.
Original Query: {query}
Current Retrieved Docs Count: {doc_count}
Generate 2 targeted search queries to fill in missing gaps."""
def adaptive_rewrite_node(state: AgenticRAGState) -> dict:
    structured_llm = llm.with_structured_output(QueryRewrite)

    prompt = rewrite_prompt.format(
        query=state["original_query"],
        doc_count=len(state["retrieved_documents"])
    )

    result = structured_llm.invoke(prompt)
    return {"sub_queries": result.new_sub_queries}

Produce a tighter query from evaluator feedback; bound rewrite attempts to avoid loops.

Node 5: grounded answer synthesis

synthesis_prompt = """You are an enterprise AI assistant. Answer the user query strictly using the grounded documentation provided below.
If a detail is missing, state what you cannot answer based on the evidence. Always cite document IDs in square brackets (e.g. [doc_id]).User Query: {query}
Documents Context:
{context}"""
def synthesize_response_node(state: AgenticRAGState) -> dict:
    formatted_context = "\n\n".join([
        f"[{d['id']}] {d['title']}: {d['content']}"
        for d in state["retrieved_documents"]
    ])

    prompt = synthesis_prompt.format(
        query=state["original_query"],
        context=formatted_context
    )

    response = llm.invoke(prompt)
    return {"final_response": response.content}

Generate only with cited passages; refuse when scores fail thresholds.

Node routing and orchestration graph

from langgraph.graph import StateGraph, END
def decide_next_step(state: AgenticRAGState) -> str:
    # Circuit breaker: stop if sufficient OR if max iterations reached (e.g., 3 loops)
    if state["is_sufficient"] or state["iterations"] >= 3:
        return "synthesize"
    return "rewrite"
# Build Graph
builder = StateGraph(AgenticRAGState)
builder.add_node("planner", plan_and_decompose_node)
builder.add_node("retriever", retrieve_hybrid_node)
builder.add_node("evaluator", evaluate_context_node)
builder.add_node("rewrite", adaptive_rewrite_node)
builder.add_node("synthesize", synthesize_response_node)
# Connect Edges
builder.set_entry_point("planner")
builder.add_edge("planner", "retriever")
builder.add_edge("retriever", "evaluator")
builder.add_conditional_edges(
    "evaluator",
    decide_next_step,
    {
        "synthesize": "synthesize",
        "rewrite": "rewrite"
    }
)
builder.add_edge("rewrite", "retriever")
builder.add_edge("synthesize", END)
# Compile Graph
agentic_rag_app = builder.compile()

Compile the graph so planner → search → evaluate branches to rewrite or synthesize, then END.

Production guardrails

Incoming Request ──► [ Input Guardrail ] ──► System Processing ──► [ Grounding & Output Guardrail ] ──► Response Output
                    (PII / Injection)                             (Faithfulness Check)

1. Security trimming at index level (ACL enforcement)

Documents carry ACL fields; every query injects the principal’s groups. Agents must not bypass filters in rewrite nodes.

2. Hallucination and grounding verification

from pydantic import BaseModel, Field
class GroundingCheck(BaseModel):
    is_grounded: bool = Field(description="True if every claim in the answer is supported by the context.")
    hallucinated_claims: List[str] = Field(description="List of claims made without source backing.")
def validate_grounding(response_text: str, context_docs: List[dict]) -> bool:
    checker_llm = llm.with_structured_output(GroundingCheck)

    docs_blob = "\n".join([d["content"] for d in context_docs])
    prompt = f"Verify if all statements in this answer are factually supported by the context.\nContext: {docs_blob}\nAnswer: {response_text}"

    res = checker_llm.invoke(prompt)
    return res.is_grounded

Check claims against retrieved spans before release.

Observability and telemetry

from openinference.instrumentation.langchain import LangChainInstrumentor
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
# Configure OpenTelemetry Tracing for Azure Monitor Application Insights
provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="https://your-appinsights-otlp-endpoint"))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
# Auto-instrument LangChain & LangGraph execution traces
LangChainInstrumentor().instrument()

Trace node timings, rewrite counts, filter cardinality, and refusal rates. Agentic RAG without telemetry is an un-debuggable loop.

The enterprise pattern is closed-loop retrieval: plan, search, judge, rewrite, answer—on an index that already knows who may see what.

Why reflection matters on Azure AI Search

Hybrid retrieval still returns near-miss chunks. An evaluator that can demand a rewrite prevents the synthesizer from laundering weak hits into confident prose. Cap rewrites (for example at two) and surface refusal when evidence remains thin. Product managers should track refusal quality as carefully as answer latency: a truthful refusal beats a fluent wrong citation.

Identity propagation

Every search node must forward the caller’s tenant and group claims into Azure AI Search filters. Rewriters must not strip filters while “improving” the query text. Log filter hashes alongside query text so audits can prove ACL enforcement per node invocation.

Failure injection drills

Break the search endpoint, zero out vector fields, or remove ACL fields in a staging index. Confirm the graph fails closed—no ungrounded answers, no cross-tenant leakage. Agentic loops amplify outages; drills keep the amplification from becoming a breach narrative.

Rollout advice

Ship planner+search+synthesize first with a single pass; add evaluator/rewrite once offline faithfulness metrics justify the extra tokens. Guardrails and telemetry should land before opening the loop to wide user populations.

Index design details that affect agent loops

Choose vector dimensions and lexical analyzers to match the corpus language. Store chunk parent ids so synthesis can cite document-level URLs, not only fragment keys. Include a last_updated field to allow agents to prefer fresh policy text during evaluation. When rewriting queries, preserve filter clauses that encode those fields—freshness and ACL are not optional flavor text.

Synthesizer prompt contracts

The grounded synthesizer should require inline citations keyed to search result ids. If the model cannot place a citation, it must drop the claim. Post-processors can strip uncited sentences as a second line of defense. Pair this with the hallucination guardrail node so both prompt discipline and programmatic verification apply.

Capacity and cost

Each reflection multiplies search units and token spend. Set budgets per user tier. Emit metrics for average loops per question. If a cohort averages more than one rewrite, invest in chunking and ranking quality before raising max loops—otherwise agents spend their way around a bad index.