RAG agentiel sur Azure AI Search avec des boucles de planification, d’évaluation et de réécriture

Le RAG agentic d’entreprise sur Azure va au-delà du simple « récupérer puis répondre ». Les systèmes de production planifient, effectuent des recherches via une récupération hybride, évaluent les preuves, réécrivent les requêtes, synthétisent en s’appuyant sur des données fiables, et appliquent des mécanismes de contrôle d’accès ACL ainsi que des protections contre les hallucinations — généralement organisés sous forme de boucle de type LangGraph via Azure AI Search.

Aperçu des composants et du flux du système :

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 et conception du système

Les étapes de planification, récupération, évaluation, réécriture et synthèse sont séparées en nœuds partageant un état commun. Azure AI Search stocke le corpus indexé avec des mesures de sécurité. Le moteur d’exécution de l’agent gère les tentatives répétées et la réflexion ; le service de recherche s’occupe de la pertinence et des filtres.

Composants architecturaux principaux

• Index + ensembles de compétences / API de mise à jour pour les documents

Requêtes hybrides lexicales + vectorielles

État de l’agent pour les objectifs, sous-requêtes, preuves et scores

Règles de contrôle pour l’ACL et le grounding

Télémétrie couvrant les nœuds

Azure AI Search : fondement du récupération agente

La recherche et les filtres hybrides rendent possibles les boucles agentes : chaque réflexion peut ajuster les filtres et le texte de la requête sans devoir reconstruire l’application. Les classeurs sémantiques et les champs vectoriels coexistent dans une même définition d’index.

Mise en place de l’infrastructure : schéma d’index (SDK Python)

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.")

Mise en œuvre avancée adaptée à la production

Connectez les clients, les identifiants et les politiques de tentative plusieurs fois ; chaque nœud ne doit pas inventer sa propre méthode d’authentification.

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

Mise en place et schéma de l’état de l’agent

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
)

L’état suit la question de l’utilisateur, les sous-requêtes décomposées, les résultats trouvés, les scores d’évaluation, le nombre de réécritures et les brouillons de réponse finale.

Nœud 1 : planificateur et décomposeur de requêtes

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
    }

Les questions complexes sont transformées en sous-requêtes classées que le reste du graphe peut exécuter.

Nœud 2 : exécution de recherche hybride

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)}

Combiner les canaux de mots-clés et vectoriels ; appliquer des filtres de sécurité issus du contexte d’identité de l’appelant.

Nœud 3 : évaluateur et boucle de réflexion

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
    }

Évaluer la suffisance et la fidélité des résultats. Si les preuves sont insuffisantes, rediriger vers une réécriture plutôt que de répondre.

Nœud 4 : réécriveur adaptatif de requêtes

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}

Produire une requête plus ciblée à partir des retours de l’évaluateur ; limiter les tentatives de réécriture pour éviter les boucles.

Nœud 5 : synthèse de réponses fondées

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}

Générer uniquement avec des extraits cités ; refuser de répondre lorsque les scores ne dépassent pas les seuils fixés.

Graphique de routage et d’orchestration des nœuds

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()

Compilez le graphe afin que le planificateur → la recherche → l’évaluation des branches permettent de réécrire ou de synthétiser, puis terminez.

Règles de sécurité en production

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

1. Élagage de la sécurité au niveau de l’index (application des ACL)

Les documents contiennent des champs ACL ; chaque requête injecte les groupes du principal. Les agents ne doivent pas contourner les filtres dans les nœuds de réécriture.

2. Vérification contre les hallucinations et le ancrage dans la réalité

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

Vérifiez les affirmations par rapport aux extraits récupérés avant de les publier.

Observabilité et télémétrie

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()

Suivez les temps de traitement des nœuds, le nombre de réécritures, la cardinalité des filtres et les taux de refus. Un RAG agentiel sans télémétrie constitue un boucle impossible à déboguer.

Le modèle d’entreprise repose sur une récupération en boucle fermée : planifier, rechercher, évaluer, réécrire, répondre — sur un index qui sait déjà qui peut voir quoi.

Pourquoi la réflexion est importante dans Azure AI Search

La récupération hybride renvoie toujours des fragments proches de la bonne réponse. Un évaluateur capable d’exiger une réécriture empêche le synthétiseur de transformer des résultats faibles en textes convaincants. Limitez le nombre de réécritures (par exemple à deux) et affichez un refus lorsque les preuves restent insuffisantes. Les responsables de produit doivent suivre la qualité des refus avec autant d’attention que le délai de réponse : un refus honnête vaut mieux qu’une citation erronée mais fluide.

Propagation de l’identité

Chaque nœud de recherche doit transmettre les identifiants du client ainsi que ses attributs de groupe aux filtres d’Azure AI Search. Les outils de réécriture ne doivent pas supprimer les filtres lorsqu’ils « améliorent » le texte de la requête. Enregistrez les hachages des filtres en même temps que le texte de la requête afin que les audits puissent vérifier l’application des règles d’accès par chaque appel au nœud.

Exercices d’injection de pannes

Brisez le point d’entrée de recherche, mettez à zéro les champs vectoriels ou supprimez les champs ACL dans un index de préparation. Vérifiez que le graphe échoue de manière fermée : aucune réponse non fondée, aucune fuite entre locataires. Les boucles d’agents amplifient les pannes ; les tests permettent d’éviter que cette amplification ne devienne une narration de violation de sécurité.

Conseils pour le déploiement

Déployez d’abord planner+search+synthesize en une seule étape ; ajoutez l’évaluateur/la réécriture uniquement lorsque les métriques de fidélité hors ligne justifient l’usage supplémentaire de tokens. Les garde-fous et la télémétrie doivent être mis en place avant d’ouvrir le cycle à un large public d’utilisateurs.

Détails de la conception de l’index affectant les boucles d’agents

Choisissez les dimensions vectorielles et les analyseurs lexicaux en fonction de la langue du corpus. Stockez les identifiants des éléments parent afin que la synthèse puisse citer des URL au niveau du document, et non seulement des clés de fragments. Incluez un champ last_updated pour permettre aux agents de privilégier des textes de politique à jour lors de l’évaluation. Lors de la réécriture des requêtes, conservez les clauses de filtrage qui encodent ces champs — la fraîcheur et l’ACL ne sont pas de simples éléments optionnels.

Contrats de prompt pour le synthétiseur

Le synthétiseur basé sur des données concrètes doit exiger des citations en ligne liées aux identifiants des résultats de recherche. Si le modèle ne peut pas insérer une citation, il doit supprimer l’affirmation correspondante. Les post-traiteurs peuvent éliminer les phrases non citées en tant que deuxième couche de protection. Associez cela au nœud de contrôle des hallucinations afin que tant la discipline des prompts que la vérification programmée soient appliquées.

Capacité et coût

Chaque réflexion multiplie les unités de recherche et la consommation de tokens. Définissez des budgets par niveau d’utilisateur. Produisez des métriques relatives au nombre moyen de boucles par question. Si une cohorte nécessite en moyenne plus d’une réécriture, investissez dans la segmentation des données et l’amélioration de la qualité du classement avant d’augmenter le nombre maximal de boucles — sinon, les agents contourneront un index défectueux en consommant davantage de ressources.