NeuralTrust | Platform for Agent Security.

AzureKnowledgeBase reads documents from an Azure AI Search index. Import it from the connector submodule — it is not re-exported from trusttest.knowledge_base.

from trusttest.knowledge_base.azure_search import AzureKnowledgeBase

The constructor takes credentials (TokenCredential or AzureKeyCredential), not key=.

Installation

uv add "trusttest[rag-azure]"

trusttest[azure] covers Azure identity / Azure OpenAI only. Use rag-azure for this connector.

Constructor

Parameter	Type	Description
`credentials`	`TokenCredential \| AzureKeyCredential`	Required. API key or Azure AD credential
`service_endpoint`	`str`	Or `AZURE_SEARCH_SERVICE_ENDPOINT`
`index_name`	`str`	Or `AZURE_SEARCH_INDEX_NAME`
`fields_mapping`	`dict[str, str]`	Defaults to `{"id": "id", "content": "content"}`
`language`	`LanguageType`	Optional; otherwise detected from documents
`embeddings_model`	`EmbeddingsModel`	From `set_config` if omitted
`llm_client`	`LLMClient`	Topic summarizer
`seed_topics`	`list[str]`	Skip auto-clustering when set
`max_doc_count`	`int`	Cap on retrieved documents

Example

import os

from azure.core.credentials import AzureKeyCredential
from dotenv import load_dotenv

from trusttest.evaluation_scenarios import EvaluationScenario
from trusttest.evaluator_suite import EvaluatorSuite
from trusttest.evaluators import AnswerRelevanceEvaluator
from trusttest.knowledge_base.azure_search import AzureKnowledgeBase
from trusttest.probes.rag import RAGProbe, BenignQuestion
from trusttest.targets.testing import DummyTarget

load_dotenv(override=True)

knowledge_base = AzureKnowledgeBase(
    credentials=AzureKeyCredential(os.getenv("AZURE_SEARCH_KEY")),
    service_endpoint=os.getenv("AZURE_SEARCH_SERVICE_ENDPOINT"),
    index_name=os.getenv("AZURE_SEARCH_INDEX_NAME"),
    fields_mapping={"content": "chunk", "id": "chunk_id"},
    language="Spanish",
)

probe = RAGProbe(
    target=DummyTarget(),
    knowledge_base=knowledge_base,
    num_questions=2,
    question_types=[BenignQuestion.SIMPLE],
)

scenario = EvaluationScenario(
    name="RAG Functional",
    evaluator_suite=EvaluatorSuite(
        evaluators=[AnswerRelevanceEvaluator()],
        criteria="any_fail",
    ),
)

test_set = probe.get_test_set()
results = scenario.evaluate(test_set)
results.display_summary()