Agent-basiertes RAG in Azure AI Search mit Schleifen für Planung, Bewertung und Überarbeitung

Enterprise-agiles RAG auf Azure ist mehr als nur „Daten abrufen und dann antworten“. Produktivsysteme planen, suchen mit hybriden Abfragemethoden, bewerten Beweismittel, überarbeiten Anfragen, synthetisieren Inhalte unter Berücksichtigung des Kontexts und setzen ACL-Regeln sowie Schutzmechanismen gegen Halluzinationen um – in der Regel als LangGraph-ähnlicher Kreislauf über Azure AI Search orchestriert.

Kurze Übersicht der Architekturkomponenten und Abläufe:

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

Architektur und Systemdesign

Planung, Datenabruf, Bewertung, Überarbeitung und Synthese werden in separate Knoten aufgeteilt, die einen gemeinsamen Zustand teilen. Azure AI Search speichert das indizierte Korpus unter Berücksichtigung von Sicherheitsmaßnahmen. Der Agent-Runtime ist für Wiederholungsversuche und Reflexion verantwortlich; der Suchdienst kümmert sich um Relevanz und Filter.

Kernarchitekturkomponenten

• Index + Skillsets / Push-API für Dokumente

Hybride lexikalische + vektorbasierte Abfragen

Agentenzustand für Ziele, Unterabfragen, Beweise und Bewertungsscores

Schutzmechanismen für ACL und Grundlegung der Informationen

Telemetrie, die über alle Knoten reicht

Azure AI Search: Grundlage der agierenden Informationsabrufung

Hybride Suchfunktionen und Filter machen agierende Schleifen möglich: Jede Reflexion kann Filter sowie Abfragetext anpassen, ohne die Anwendung neu aufbauen zu müssen. Semantische Rangierer und Vektorfelder existieren in derselben Indexdefinition nebeneinander.

Infrastruktur-Einrichtung: 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.")

Produktionsreife tiefgehende Implementierung

Kundenverbindungen, Anmeldeinformationen sowie Wiederholungsrichtlinien müssen nur einmal eingestellt werden; die Knoten sollten nicht jeweils eigene Authentifizierungsmechanismen entwickeln.

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

Einrichtung und Schema des Agentenzustands

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
)

Der Zustand verfolgt die Benutzerfrage, die aufgeteilten Unterabfragen, Treffer, Bewertungsscores, die Anzahl der Überarbeitungen sowie Entwürfe der endgültigen Antwort.

Node 1: Planer und Abfragenteiler

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
    }

Komplexe Fragen werden in rangierte Unterabfragen umgewandelt, die der Rest des Graphen ausführen kann.

Node 2: Hybride Suchausführung

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

Kombinieren Sie Schlüsselwort- und Vektorkanäle; wenden Sie Sicherheitsfilter aus dem Identitätskontext des Aufrufers an.

Node 3: Bewertungsmechanismus und Reflexionsschleife

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
    }

Bewerten Sie Genügsamkeit und Treue der Ergebnisse. Wenn die Beweise schwach sind, leiten Sie die Anfrage zur Neuformulierung statt zur Beantwortung um.

Node 4: Anpassungsfähiger Abfragenumformulator

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}

Erstellen Sie auf Basis der Rückmeldungen des Bewertungsmechanismus präzisere Abfragen; begrenzen Sie die Anzahl der Neuformulierungsversuche, um Schleifen zu vermeiden.

Node 5: Synthese fundierter Antworten

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}

Erstellen Sie Antworten ausschließlich mit zitierten Passagen; lehnen Sie sie ab, wenn die Bewertungswerte unter den Schwellenwerten liegen.

Node-Verkehr und Orchestrierungsgraph

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

Kompilieren Sie das Graphen, sodass der Planer → die Suche → die Bewertung der Äste zur Überarbeitung oder Synthese erfolgt, anschließend END.

Produktions-Schutzmaßnahmen

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

1. Sicherheitsfilterung auf Indexebene (Durchsetzung von ACL)

Dokumente enthalten ACL-Felder; jede Anfrage übermittelt die Gruppen des Benutzers. Agenten dürfen die Filter in den Überarbeitungsknoten nicht umgehen.

2. Überprüfung auf Halluzinationen und Grundlagen

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

Überprüfen Sie die Angaben vor der Freigabe anhand der abgerufenen Datenabschnitte.

Beobachtbarkeit und Telemetrie

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

Verfolgen Sie die Laufzeiten der Knoten, die Anzahl der Überarbeitungen, die Kardinalität der Filter sowie die Ablehnungsraten. Ein agentbasiertes RAG ohne Telemetrie ist ein nicht zu debuggender Schleifenprozess.

Das Unternehmensmuster ist die geschlossene Schleife der Datenabfrage: Planen, Suchen, Beurteilen, Überarbeiten, Antworten – auf einem Index, der bereits kennt, wer was sehen darf.

Warum Reflexion bei Azure AI Search wichtig ist

Die hybride Suchmethode liefert weiterhin nur nahe liegende Ergebnisse. Ein Evaluator, der eine Überarbeitung erzwingen kann, verhindert, dass der Synthesizer schwache Treffer als zuverlässige Texte darstellt. Begrenzen Sie die Anzahl der Überarbeitungen (zum Beispiel auf zwei) und geben Sie Abweisung an, wenn die Beweise unzureichend sind. Produktmanager sollten die Qualität der Abweisungen genauso sorgfältig überwachen wie die Antwortzeit: Eine ehrliche Abweisung ist besser als eine flüssige, falsche Zitierung.

Identitätsverbreitung

Jeder Suchknoten muss die Angaben zum Tenant und zur Gruppe des Aufrufers in die Filter von Azure AI Search weiterleiten. Überarbeitungstools dürfen die Filter nicht entfernen, während sie den Suchtext „verbessern“. Protokollieren Sie die Hash-Werte der Filter zusammen mit dem Suchtext, damit Audits nachweisen können, dass die ACL-Vorschriften bei jeder Knotenaufrufung eingehalten werden.

Fehlerinjektionsübungen

Stören Sie den Suchendpunkt, setzen Sie die Vektorfelder auf Null oder entfernen Sie die ACL-Felder in einem Staging-Index. Überprüfen Sie, dass das Graph nicht fehlerhaft arbeitet – es dürfen keine unbegründeten Antworten oder Datenlecks zwischen verschiedenen Nutzern auftreten. Agentenbasierte Schleifen verstärken Ausfälle; Tests verhindern, dass diese Verstärkung zu einer Katastrophenbeschreibung wird.

Ratschläge zur Einführung

Führen Sie zunächst Planner+Search+Synthesize in einer einzigen Durchlaufphase ein; fügen Sie Evaluator/Rewrite erst hinzu, wenn Metriken zur Offline-Genauigkeit die Verwendung zusätzlicher Token rechtfertigen. Schutzmechanismen und Telemetriedaten sollten vor der Freigabe für eine breite Nutzerbasis implementiert werden.

Details des Indexdesigns, die Agentenschleifen beeinflussen

Wählen Sie Vektordimensionen sowie lexikalische Analysewerkzeuge, die zur Sprache des Korpus passen. Speichern Sie die Eltern-ID der Blöcke, damit die Synthese Dokument-URLs statt nur Fragment-Schlüssel angeben kann. Fügen Sie ein last_updated-Feld hinzu, damit Agenten bei der Bewertung bevorzugt aktuelle Richtlinientexte verwenden können. Beim Umformulieren von Abfragen sollten die Filterklauseln beibehalten werden, die diese Felder kodieren – Aktualität und ACL sind keine optionalen Zusatzangaben.

Synthesizer-Prompt-Verträge

Der auf der Realität basierende Synthesizer sollte inline-Zitate erfordern, die an Suchergebnis-IDs gebunden sind. Wenn das Modell kein Zitat einfügen kann, muss es die Aussage streichen. Post-Processor können als zweite Verteidigungslinie unzitierte Sätze entfernen. Kombinieren Sie dies mit dem Halluzinationsschutz-Node, damit sowohl Prompt-Disziplin als auch programmatische Überprüfung angewandt werden.

Kapazität und Kosten

Jede Reflexion erhöht die Anzahl der Sucheinheiten sowie den Tokenverbrauch. Legen Sie Budgets für jede Benutzerstufe fest. Erstellen Sie Metriken zur durchschnittlichen Anzahl der Durchläufe pro Frage. Wenn eine Gruppe im Durchschnitt mehr als einen Überarbeitungsvorgang benötigt, investieren Sie in die Aufteilung der Aufgaben in kleinere Teile sowie in eine bessere Ranking-Qualität, bevor Sie die maximale Anzahl der Durchläufe erhöhen – andernfalls umgehen die Agenten einen schlechten Index einfach durch hohen Verbrauch.