Retrieval-Augmented Generation (RAG) unlocks powerful Enterprise search over unstructured document repositories. However, standard RAG implementations often face security challenges. In thousands of organisations, a security gap exists. Once documents are ingested into a unified vector store, vector similarity search operates indiscriminately across security boundaries. An employee querying human resources policies might inadvertently retrieve vector chunks originating from executive compensation plans or restricted financial forecasts.
To make RAG enterprise-ready, identity aware document retrieval is mandatory. Security filters must execute directly within the retrieval query engine, ensuring users only retrieve chunks they are authorised to access under Active Directory / Entra ID policies.
- Public HR policy documents (Group: Everyone)
- Regional manager operational guidelines (Group: Regional_Managers)
- Sensitive salary metadata (Group: Compensation_HR)
from azure.core.credentials import AzureKeyCredential
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
SearchIndex,
SimpleField,
SearchableField,
SearchField,
SearchFieldDataType,
VectorSearch,
HnswAlgorithmConfiguration,
VectorSearchProfile
)
def create_rlac_vector_index(endpoint: str, credential: AzureKeyCredential, index_name: str):
client = SearchIndexClient(endpoint=endpoint, credential=credential)
fields = [
SimpleField(name="id", type=SearchFieldDataType.String, key=True),
SearchableField(name="content", type=SearchFieldDataType.String),
SearchField(
name="content_vector",
type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
searchable=True,
vector_search_dimensions=1536,
vector_search_profile_name="myHnswProfile"
),
# Security Filter Attributes
SimpleField(
name="allowed_groups",
type=SearchFieldDataType.Collection(SearchFieldDataType.String),
filterable=True
),
SimpleField(
name="owner_oid",
type=SearchFieldDataType.String,
filterable=True
)
]
vector_search = VectorSearch(
algorithms=[HnswAlgorithmConfiguration(name="myHnsw")],
profiles=[VectorSearchProfile(name="myHnswProfile", algorithm_configuration_name="myHnsw")]
)
index = SearchIndex(name=index_name, fields=fields, vector_search=vector_search)
client.create_or_update_index(index)
Python
import os
from azure.identity import DefaultAzureCredential
from azure.search.documents import SearchClient
from azure.search.documents.models import VectorizedQuery
from azure.ai.evaluation import AzureAIProjectClient
def query_vector_store_with_rlac(
search_endpoint: str,
index_name: str,
query_text: str,
query_vector: list[float],
user_oid: str,
user_groups: list[str]
):
credential = DefaultAzureCredential()
search_client = SearchClient(endpoint=search_endpoint, index_name=index_name, credential=credential)
# Format OData security filter expression
groups_filter = ", ".join([f"'{g}'" for g in user_groups])
odata_filter = f"owner_oid eq '{user_oid}' or allowed_groups/any(g: search.in(g, '{groups_filter}'))"
vector_query = VectorizedQuery(
vector=query_vector,
k_nearest_neighbors=5,
fields="content_vector"
)
results = search_client.search(
search_text=query_text,
vector_queries=[vector_query],
filter=odata_filter,
select=["id", "content"]
)
retrieved_chunks = [doc["content"] for doc in results]
return retrieved_chunks

No comments:
Post a Comment