Repository navigation
Expand file tree
/
Copy pathazure_ai_agent_declarative_function_calling_from_file.py
More file actions
90 lines (74 loc) · 3.14 KB
/
Copy pathazure_ai_agent_declarative_function_calling_from_file.py
File metadata and controls
90 lines (74 loc) · 3.14 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
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import Annotated
from azure.identity.aio import AzureCliCredential
from semantic_kernel.agents import AgentRegistry, AzureAIAgent, AzureAIAgentSettings, AzureAIAgentThread
from semantic_kernel.functions.kernel_function_decorator import kernel_function
"""
The following sample demonstrates how to create an Azure AI agent that answers
user questions. The sample shows how to load a declarative spec from a file.
The plugins/functions must already exist in the kernel.
They are not created declaratively via the spec.
"""
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"
async def main():
async with (
AzureCliCredential() as creds,
AzureAIAgent.create_client(credential=creds) as client,
):
try:
# Define the YAML file path for the sample
file_path = os.path.join(
os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))),
"resources",
"declarative_spec",
"spec.yaml",
)
# Create the AzureAI Agent from the YAML spec
agent: AzureAIAgent = await AgentRegistry.create_from_file(
file_path,
plugins=[MenuPlugin()],
client=client,
settings=AzureAIAgentSettings(), # The Spec's ChatModelId & Endpoint come from .env/env vars
)
# Create the agent
user_inputs = [
"Hello",
"What is the special soup?",
"How much does that cost?",
"Thank you",
]
# 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 = None
for user_input in user_inputs:
print(f"# User: '{user_input}'")
# Invoke the agent for the specified task
async for response in agent.invoke(
messages=user_input,
thread=thread,
):
print(f"# {response.name}: {response}")
# Store the thread for the next iteration
thread = response.thread
finally:
# Cleanup: Delete the thread and agent
await client.agents.delete_agent(agent.id) if agent else None
await thread.delete() if thread else None
if __name__ == "__main__":
asyncio.run(main())