Repository navigation
Expand file tree
/
Copy pathazure_ai_agent_message_callback.py
More file actions
132 lines (108 loc) · 4.98 KB
/
Copy pathazure_ai_agent_message_callback.py
File metadata and controls
132 lines (108 loc) · 4.98 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from typing import Annotated
from azure.identity.aio import AzureCliCredential
from semantic_kernel.agents import AzureAIAgent, AzureAIAgentSettings, AzureAIAgentThread
from semantic_kernel.contents import FunctionCallContent, FunctionResultContent
from semantic_kernel.contents.chat_message_content import ChatMessageContent
from semantic_kernel.functions import kernel_function
"""
This sample demonstrates how to create an Azure AI Agent and invoke it using the non-streaming `invoke()` method.
While `invoke()` returns only the final assistant message, the agent can optionally emit intermediate messages
(e.g., function calls and results) via a callback by supplying `on_intermediate_message`.
In this example, the agent is configured with a plugin that provides menu specials and item pricing. As the user
asks about the menu, the agent performs tool calls mid-invocation, and those intermediate steps are surfaced
via the callback function while the invocation is still in progress.
"""
# Define a sample plugin for the sample
class MenuPlugin:
"""A sample Menu Plugin used for the concept sample."""
@kernel_function(description="Provides a list of specials from the menu.")
def get_specials(self) -> Annotated[str, "Returns the specials from the menu."]:
return """
Special Soup: Clam Chowder
Special Salad: Cobb Salad
Special Drink: Chai Tea
"""
@kernel_function(description="Provides the price of the requested menu item.")
def get_item_price(
self, menu_item: Annotated[str, "The name of the menu item."]
) -> Annotated[str, "Returns the price of the menu item."]:
return "$9.99"
# This callback function will be called for each intermediate message,
# which will allow one to handle FunctionCallContent and FunctionResultContent.
# If the callback is not provided, the agent will return the final response
# with no intermediate tool call steps.
async def handle_intermediate_steps(message: ChatMessageContent) -> None:
for item in message.items or []:
if isinstance(item, FunctionResultContent):
print(f"Function Result:> {item.result} for function: {item.name}")
elif isinstance(item, FunctionCallContent):
print(f"Function Call:> {item.name} with arguments: {item.arguments}")
else:
print(f"{item}")
async def main() -> None:
ai_agent_settings = AzureAIAgentSettings()
async with (
AzureCliCredential() as creds,
AzureAIAgent.create_client(credential=creds, endpoint=ai_agent_settings.endpoint) as client,
):
AGENT_NAME = "Host"
AGENT_INSTRUCTIONS = "Answer questions about the menu."
# Create agent definition
agent_definition = await client.agents.create_agent(
model=ai_agent_settings.model_deployment_name,
name=AGENT_NAME,
instructions=AGENT_INSTRUCTIONS,
)
# Create the AzureAI Agent
agent = AzureAIAgent(
client=client,
definition=agent_definition,
plugins=[MenuPlugin()], # add the sample plugin to the agent
)
# Create a thread for the agent
# If no thread is provided, a new thread will be
# created and returned with the initial response
thread: AzureAIAgentThread = None
user_inputs = [
"Hello",
"What is the special soup?",
"How much does that cost?",
"Thank you",
]
try:
for user_input in user_inputs:
print(f"# User: '{user_input}'")
async for response in agent.invoke(
messages=user_input,
thread=thread,
on_intermediate_message=handle_intermediate_steps,
):
print(f"# Agent: {response}")
thread = response.thread
finally:
# Cleanup: Delete the thread and agent
await thread.delete() if thread else None
await client.agents.delete_agent(agent.id)
"""
Sample Output:
# User: 'Hello'
# Agent: Hi there! How can I assist you today?
# User: 'What is the special soup?'
Function Call:> MenuPlugin-get_specials with arguments: {}
Function Result:>
Special Soup: Clam Chowder
Special Salad: Cobb Salad
Special Drink: Chai Tea
for function: MenuPlugin-get_specials
# Agent: The special soup is Clam Chowder. Would you like to know anything else about the menu?
# User: 'How much does that cost?'
Function Call:> MenuPlugin-get_item_price with arguments: {"menu_item":"Clam Chowder"}
Function Result:> $9.99 for function: MenuPlugin-get_item_price
# Agent: The Clam Chowder costs $9.99. Let me know if you'd like assistance with anything else!
# User: 'Thank you'
# Agent: You're welcome! Enjoy your meal! 😊
"""
if __name__ == "__main__":
asyncio.run(main())