Агентный RAG в Azure AI Search с циклами планирования, оценки и переписывания

Система агентного RAG в Azure представляет собой нечто большее, чем просто «поиск информации и последующий ответ». Промышленные системы планируют действия, выполняют поиск с использованием гибридных методов извлечения данных, оценивают достоверность информации, переписывают запросы, синтезируют результаты с учетом контекста, а также реализуют механизмы контроля доступа и защиты от галлюцинаций — обычно это осуществляется в виде цикла в стиле LangGraph через Azure AI Search.

Краткое описание структуры и потока обработки:

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

Архитектура и проектирование системы

Функции планирования, извлечения данных, оценки, переписывания запросов и синтеза разделены на отдельные узлы с общим состоянием. Azure AI Search хранит индексированный корпус с механизмами обеспечения безопасности. Механизм выполнения агента отвечает за повторные попытки и рефлексию; сервис поиска отвечает за определение релевантности и применение фильтров.

• Индекс + наборы функций / API для загрузки документов

Гибридные лексикальные и векторные запросы

Состояние агента для целей, подзапросов, доказательств и оценок

Ограничения для ACL и обеспечения основания

Телеметрия, охватывающая узлы

Azure AI Search: основа агентного поиска

Гибридный поиск и фильтры делают возможными агентные циклы: каждое рефлексирование может корректировать фильтры и текст запроса без необходимости пересоздания приложения. Семантические ранжировщики и векторные поля сосуществуют в одной определении индекса.

Настройка инфраструктуры: схема индекса (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.")

Продакшн-готовая глубокая реализация

Подключите клиентов, учетные данные и политики повторных попыток один раз; узлы не должны самостоятельно создавать механизмы аутентификации.

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

Настройка и схема состояния агента

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
)

Состояние отслеживает вопрос пользователя, разложенные подзапросы, результаты поиска, оценки качества, количество переписываний и черновики окончательного ответа.

Узел 1: планировщик и разборщик запросов

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
    }

Сложные вопросы преобразуются в отсортированные подзапросы, которые может обработать остальная часть графа.

Узел 2: выполнение гибридного поиска

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

Сочетаются каналы поиска на основе ключевых слов и векторов; применяются фильтры безопасности, основанные на контексте идентичности вызывающего пользователя.

Узел 3: оценщик и цикл рефлексии

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
    }

Оценивается достаточность и точность ответа. Если доказательства недостаточны, запрос направляется на переписывание вместо ответа.

Узел 4: адаптивный переписыватель запросов

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}

На основе обратной связи от оценщика формируется более точный запрос; количество попыток переписывания ограничивается для избежания циклов.

Узел 5: синтез основанного на фактах ответа

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}

Ответ генерируется только с использованием цитируемых фрагментов; отклоняется, если оценки не достигают установленных порогов.

Граф маршрутизации узлов и оркестрации

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

Соберите граф так, чтобы процесс шел от планирования к поиску и оценке ветвей с целью переписывания или синтеза, затем завершите работу.

Ограничения в процессе производства

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

1. Ограничения безопасности на уровне индекса (применение ACL)

Документы содержат поля ACL; каждый запрос включает группы пользователя. Агенты не должны обходить фильтры в узлах переписывания.

2. Проверка на галлюцинации и соответствие реальности

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

Перед публикацией проверяйте утверждения на соответствие полученной информации.

Наблюдаемость и телеметрия

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

Отслеживайте время работы узлов, количество операций переписывания, кардинальность фильтров и процент отклонений. Агентный RAG без телеметрии представляет собой недоступную для отладки систему.

Стандартный подход в корпорациях — это замкнутый цикл получения информации: планирование, поиск, оценка, переписывание, ответ — на основе индекса, который уже определяет, кто может просматривать какую информацию.

Почему важна рефлексия в Azure AI Search

Гибридный метод поиска всё ещё возвращает результаты с низкой точностью. Оценочный инструмент, позволяющий требовать переписывания текста, предотвращает превращение слабых результатов в убедительные тексты синтезатором. Ограничьте количество переписываний (например, до двух) и отображайте код refusal, когда доказательства недостаточны. Менеджерам продуктов следует так же тщательно отслеживать качество отказов, как и время ответа: честный отказ лучше, чем правильное, но ошибочное цитирование.

Распространение идентичности

Каждый узел поиска должен передавать информацию о тенанте и группе вызывающего пользователя в фильтры Azure AI Search. Инструменты переписывания не должны удалять фильтры при «улучшении» текста запроса. Записывайте хэши фильтров вместе с текстом запроса, чтобы при аудите можно было подтвердить соблюдение правил доступа для каждого вызова узла.

Тренировки по имитации сбоев

Нарушите конечную точку поиска, установите нулевые значения для векторных полей или удалите поля ACL из индекса тестирования. Убедитесь, что граф не содержит ошибок — нет некорректных ответов и нет утечек данных между тенантами. Циклы агентов усугубляют сбои; тренировочные сценарии предотвращают превращение этих усугублений в серьезные инциденты.

Советы по постепенному внедрению

Сначала выпустите компоненты planner+search+synthesize за один проход; добавьте функции оценки и переписывания только тогда, когда показатели точности работы в офлайн-режиме оправдают использование дополнительных токенов. Регулирующие механизмы и системы телеметрии должны быть внедрены до того, как циклы станут доступны для широкой аудитории пользователей.

Детали проектирования индекса, влияющие на циклы агентов

Выберите размеры векторов и лексические анализаторы, соответствующие языку корпуса. Храните идентификаторы родителей фрагментов, чтобы синтезатор мог указывать URL на уровне документа, а не только ключи фрагментов. Включите поле last_updated, чтобы агенты могли во время оценки отдавать предпочтение актуальному тексту правил. При переписывании запросов сохраняйте условия фильтрации, кодирующие эти поля — актуальность и ACL не являются дополнительными описательными элементами.

Контракты инструкций для синтезатора

Синтезатор, основанный на реальных данных, должен требовать встроенных цитат, привязанных к идентификаторам результатов поиска. Если модель не может включить цитату, она должна исключить соответствующее утверждение. Постобработчики могут удалить нecитированные предложения в качестве дополнительной меры защиты. Сочетайте это с узлом предотвращения галлюцинаций, чтобы применялись как дисциплинарные требования к инструкциям, так и программная верификация.

Мощность и стоимость

Каждое отражение увеличивает количество единиц поиска и расход токенов. Установите бюджеты для каждого уровня пользователей. Составляйте показатели среднего количества итераций на вопрос. Если среднее количество переписываний для группы превышает одно, инвестируйте в разбиение данных на части и улучшение качества ранжирования перед увеличением максимального количества итераций — в противном случае агенты будут тратить ресурсы впустую из-за плохого индекса.