Агентны RAG на Azure AI Search з сцэнарыямі планавання, адзыначэння і перапісву.

Enterprise-рангавы RAG на Azure — гэта не проста процедура «з’явіць інфармацыю, а потым даць адказ». Системы для працы ў прыемным режыме плануюць дзеяння, ведуць пошук за дапамою гібрыдных методаў адзыскання інфармацыі, атрыбутуюць значэння доказаў, перапісваюць запиты, синтэзуюць інфармацыю на адповідных падставах, а таксама рэалізуюць механізмы контролю ACL і запобежэння галлюцінацый — усё гэта зазвычай арганізуецца як цикл у стылі 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, калі доказы застаюцца недастатковымі. Менаджэрамі продуктав трэба столькі ж адзяйнаста старацца пры выкарыстоўванні параметра refusal, колькі і пры выкарыстоўванні параметра latency: правдзіва адмова лепшая, чым плавна напісаны некоректны цитат.

Распространэнне ідэнтычнасці

Кожны вузел пошуку должен перадаваць данні прыналежнасці тэнанта і групы вызывача ў фільтры Azure AI Search. Програмы для перапісваў не должны адключаць фільтры пад час “прадактывання” тексту запита. Зяўляйце хэшы фільтраў разам з текстам запита, каб аудыты моглі падтвердзіць сяброўства праваў ACL па кожнам вызыву вузла.

Трэніраванне на выклікі неудач

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

Прадакты распространення

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

Дзеянні дызайна індэкса, якіе вплываюць на циклы агентаў

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

Контракты прамптав синтэзатора

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

Магчымасці і косты

Кожна пераспектыва падае колькісць елементаў пошуку і выкарыстоўвання токенав. Задаце бюджэты па кожнай групе корыстувачаў. Адзначайце показнікі сярэдньяй колькасці петляў за адна запытанне. Якшчуль група корыстувачаў у сярэднім вырахунку адбывае больш чым адна перапісва, трэба інвеставаць у разбіўку задач на часткі і падвышэнне якосці ранжыравання, прычым не падвышаючы максимальную колькасць петляў — інакш агенты будуць выкарыстоўваць вялікую колькасць ресурсаў, ёсць чым прыкрасіць паўзлівы індэкс.