In continuation of the previous three articles for 1. vectorizing and analyzing drone images 2. performing semantic retrieval from vector store for drone sensing applications and 3. errors and resolutions encountered, this article takes it a step forward from one-shot RAG based response to automatic query decomposition and multiple query executions using agentic-retrieval which has a 40% improvement in f-score.
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
KnowledgeAgent,
KnowledgeAgentAzureOpenAIModel,
KnowledgeAgentRequestLimits,
KnowledgeAgentTargetIndex
)
from dotenv import load_dotenv
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
import os
load_dotenv(override=True)
answer_model = os.getenv("ANSWER_MODEL", "gpt-4o")
endpoint = os.environ["AZURE_SEARCH_ENDPOINT"]
credential = DefaultAzureCredential()
token_provider = get_bearer_token_provider(credential, "https://search.azure.com/.default")
index_name = os.getenv("AZURE_SEARCH_INDEX", "index01")
azure_openai_endpoint = os.environ["AZURE_OPENAI_ENDPOINT"]
azure_openai_gpt_deployment = os.getenv("AZURE_OPENAI_GPT_DEPLOYMENT", "gpt-4o")
azure_openai_gpt_model = os.getenv("AZURE_OPENAI_GPT_MODEL", "gpt-4o")
azure_openai_api_version = os.getenv("AZURE_OPENAI_API_VERSION", "2025-03-01-preview")
azure_openai_embedding_deployment = os.getenv("AZURE_OPENAI_EMBEDDING_DEPLOYMENT", "text-embedding-3-large")
azure_openai_embedding_model = os.getenv("AZURE_OPENAI_EMBEDDING_MODEL", "text-embedding-3-large")
agent_name = os.getenv("AZURE_SEARCH_AGENT_NAME", "image-search-agent")
api_version = "2025-05-01-Preview"
AZURE_OPENAI_ENDPOINT=https://<openai-resource-name>.openai.azure.com
AZURE_OPENAI_GPT_DEPLOYMENT=gpt-4o-mini
AZURE_SEARCH_ENDPOINT=https://<search-resource-name>.search.windows.net
AZURE_SEARCH_INDEX_NAME=agentic-retrieval-drone-images
agent=KnowledgeAgent(
name=agent_name,
target_indexes=[
KnowledgeAgentTargetIndex(
index_name=index_name, default_include_reference_source_data=True,
default_reranker_threshold=2.5
)
],
models=[
KnowledgeAgentAzureOpenAIModel(
azure_open_ai_parameters=AzureOpenAIVectorizerParameters(
resource_url=azure_openai_endpoint,
deployment_name=azure_openai_gpt_deployment,
model_name=azure_openai_gpt_model,
)
)
],
request_limits=KnowledgeAgentRequestLimits(
max_output_size=agent_max_output_tokens
)
)
index_client = SearchIndexClient(endpoint=endpoint, credential=credential)
index_client.create_or_update_agent(agent)
instructions = """
A Q&A agent that can answer questions about the drone images stored in Azure AI Search.
Sources have a JSON description and vector format with a ref_id that must be cited in the answer.
If you do not have the answer, respond with "I don't know".
"""
messages = [
{
"role": "system",
"content": instructions
}
]
from azure.search.documents.agent import KnowledgeAgentRetrievalClient
from azure.search.documents.agent.models import KnowledgeAgentRetrievalRequest, KnowledgeAgentMessage, KnowledgeAgentMessageTextContent, KnowledgeAgentIndexParams
agent_client = KnowledgeAgentRetrievalClient(endpoint=endpoint, agent_name=agent_name, credential=credential)
messages.append({
"role": "user",
"content":
"""
How many red cars could be found?
"""
})
# retrieval_result = agent_client.knowledge_retrieval.retrieve(
# messages[KnowledgeAgentMessage(
# role=msgp["role"],
# content=[KnowledgeAgentMessageTextContent(text=msg["content"])])
# for msg in messages if msg["role"] != "system"],
# Target_index_params=[KnowedgeAgentIndexParams(index_name=index_name, reranker_threshold=3, include_reference_source_data=True)],
# )
# )
retrieval_result = agent_client.retrieve(
retrieval_request=KnowledgeAgentRetrievalRequest(
messages=[KnowledgeAgentMessage(role=msg["role"], content=[KnowledgeAgentMessageTextContent(text=msg["content"])]) for msg in messages if msg["role"] != "system"],
target_index_params=[KnowledgeAgentIndexParams(index_name=index_name, reranker_threshold=2.5)]
)
)
messages.append({
"role": "assistant",
"content": retrieval_result.response[0].content[0].text
})
print(messages)
import textwrap
print("Response")
print(textwrap.fill(retrieval_result.response[0].content[0].text, width=120))
import json
print("Activity")
print(json.dumps([a.as_dict() for a in retrieval_result.activity], indent=2))
print("Results")
print(json.dumps([r.as_dict() for r in retrieval_result.references], indent=2))
