In continuation of previous post, you can either upgrade your LLM from text-to-text, multimodal or reasoning models or you can employ specific agents in an agentic retrieval to get better analysis for your drone images. The agentic retrieval is easier to diversify and scale but the model is harder to upskill. The flip side is that the more agents use LLMs, the number and cost for tokens used increases proportionally while the sample data can be enhanced to help fine-tune or improve chain of thought for the reasoning model with zero cost or change to the infrastructure. The following section walks us through a sample of using an agent retrieval.
#! /usr/bin/python
"""
requirements.txt:
azure-identity
openai
aiohttp
ipykernel
dotenv
requests
azure-search-documents==11.6.0b12
"""
from azure.ai.agents.models import FunctionTool, ToolSet, ListSortOrder
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)
thread = project_client.agents.threads.create()
retrieval_results = {}
def agentic_retrieval() -> str:
"""
Searches the drone images and other curated metadata and facts.
The returned string is in a JSON format that contains the reference id.
Using the same format as in agent's response
References are cited by zero-based id number
"""
# Take the last 5 messages in the conversation
messages = project_client.agents.messages.list(thread.id, limit=5, order=ListSortOrder.DESCENDING)
# Reverse the order so the most recent message is last
messages = list(messages)
messages.reverse()
retrieval_result = agent_client.retrieve(
retrieval_request=KnowledgeAgentRetrievalRequest(
messages=[KnowledgeAgentMessage(role=msg["role"], content=[KnowledgeAgentMessageTextContent(text=msg.content[0].text)]) for msg in messages if msg["role"] != "system"],
target_index_params=[KnowledgeAgentIndexParams(index_name=index_name, reranker_threshold=2.5)]
)
)
# Associate the retrieval results with the last message in the conversation
last_message = messages[-1]
retrieval_results[last_message.id] = retrieval_result
# Return the grounding response to the agent
return retrieval_result.response[0].content[0].text
# https://learn.microsoft.com/en-us/azure/ai-services/agents/how-to/tools/function-calling
functions = FunctionTool({ agentic_retrieval })
toolset = ToolSet()
toolset.add(functions)
project_client.agents.enable_auto_function_calls(toolset)
# start a chat
from azure.ai.agents.models import AgentsNamedToolChoice, AgentsNamedToolChoiceType, FunctionName
message = project_client.agents.messages.create(
thread_id=thread.id,
role="user",
content="""
Which landmarks are responsible for intersections that are more prone to vehicle and pedestrian traffic conflicts or activities? Which hours of the day are quiet for pedestrian traffic at those intersections?
"""
)
run = project_client.agents.runs.create_and_process(
thread_id=thread.id,
agent_id=agent.id,
tool_choice=AgentsNamedToolChoice(type=AgentsNamedToolChoiceType.FUNCTION, function=FunctionName(name="agentic_retrieval")),
toolset=toolset)
if run.status == "failed":
raise RuntimeError(f"Run failed: {run.last_error}")
output = project_client.agents.messages.get_last_message_text_by_role(thread_id=thread.id, role="assistant").text.value
print("Agent response:", output.replace(".", "\n"))
