探索语义内核
重要
此功能处于实验阶段。 此阶段的功能正在开发中,在升级到预览阶段或候选发布阶段之前可能会更改。
功能当前在 Java 中不可用。
什么是 AzureAIAgent
?
AzureAIAgent
是语义内核框架中的专用代理,旨在通过无缝工具集成提供高级聊天功能。 它自动执行工具调用,无需手动分析和调用。 代理还使用线程安全地管理会话历史记录,从而减少维护状态的开销。 此外,AzureAIAgent
还支持各种内置工具,包括通过必应、Azure AI 搜索、Azure Functions 和 OpenAPI 进行文件检索、代码执行和数据交互。
若要使用 AzureAIAgent
,必须使用 Azure AI Foundry 项目。 以下文章概述了 Azure AI Foundry、如何创建和配置项目以及代理服务:
准备您的开发环境
若要继续开发 AzureAIAgent
,请使用相应的包配置开发环境。
将 Microsoft.SemanticKernel.Agents.AzureAI
包添加到项目:
dotnet add package Microsoft.SemanticKernel.Agents.AzureAI --prerelease
您可能还希望包括 Azure.Identity
包:
dotnet add package Azure.Identity
安装 semantic-kernel
包:
pip install semantic-kernel
功能当前在 Java 中不可用。
配置 AI 项目客户端
首先访问AzureAIAgent
需要创建一个为特定Foundry项目配置的客户端,通常通过提供项目终结点(Azure AI Foundry SDK:项目入门)来实现。
PersistentAgentsClient client = AzureAIAgent.CreateAgentsClient("<your endpoint>", new AzureCliCredential());
修改根目录中 .env
文件以包括:
AZURE_AI_AGENT_ENDPOINT = "<example-endpoint>"
AZURE_AI_AGENT_MODEL_DEPLOYMENT_NAME = "<example-model-deployment-name>"
定义配置后,可以创建客户端:
from semantic_kernel.agents import AzureAIAgent
async with (
DefaultAzureCredential() as creds,
AzureAIAgent.create_client(credential=creds) as client,
):
# Your operational code here
如果已配置,Pydantic 设置将选取endpoint
的底层元素。 否则,您可以将其显式传给 create_client()
方法:
from semantic_kernel.agents import AzureAIAgent
async with (
DefaultAzureCredential() as creds,
AzureAIAgent.create_client(credential=creds, endpoint="<your-endpoint>") as client,
):
# Your operational code here
功能当前在 Java 中不可用。
创建一个 AzureAIAgent
要创建 AzureAIAgent
,首先通过 Azure Agent 服务来配置和初始化 Foundry 项目,然后将其与语义内核集成。
PersistentAgentsClient client = AzureAIAgent.CreateAgentsClient("<your endpoint>", new AzureCliCredential());
// 1. Define an agent on the Azure AI agent service
PersistentAgent definition = await agentsClient.Administration.CreateAgentAsync(
"<name of the the model used by the agent>",
name: "<agent name>",
description: "<agent description>",
instructions: "<agent instructions>");
// 2. Create a Semantic Kernel agent based on the agent definition
AzureAIAgent agent = new(definition, agentsClient);
from azure.identity.aio import DefaultAzureCredential
from semantic_kernel.agents import AzureAIAgent, AzureAIAgentSettings
async with (
DefaultAzureCredential() as creds,
AzureAIAgent.create_client(credential=creds) as client,
):
# 1. Define an agent on the Azure AI agent service
agent_definition = await client.agents.create_agent(
model=AzureAIAgentSettings().model_deployment_name,
name="<name>",
instructions="<instructions>",
)
# 2. Create a Semantic Kernel agent based on the agent definition
agent = AzureAIAgent(
client=client,
definition=agent_definition,
)
功能当前在 Java 中不可用。
与 AzureAIAgent
进行互动
与 AzureAIAgent
的交互非常简单。 代理使用线程自动维护会话历史记录。
Azure AI 代理线程 的具体细节通过 Microsoft.SemanticKernel.Agents.AzureAI.AzureAIAgentThread
类抽象化,这是 Microsoft.SemanticKernel.Agents.AgentThread
的实现。
重要
请注意,Azure AI 代理 SDK 具有 PersistentAgentThread
该类。 它不应与Microsoft.SemanticKernel.Agents.AgentThread
混淆,Microsoft.SemanticKernel.Agents.AgentThread
是针对所有线程类型的通用语义内核代理抽象。
AzureAIAgent
目前仅支持类型为 AzureAIAgentThread
的线程。
AzureAIAgentThread agentThread = new(agent.Client);
try
{
ChatMessageContent message = new(AuthorRole.User, "<your user input>");
await foreach (ChatMessageContent response in agent.InvokeAsync(message, agentThread))
{
Console.WriteLine(response.Content);
}
}
finally
{
await agentThread.DeleteAsync();
await agent.Client.DeleteAgentAsync(agent.Id);
}
Azure AI 代理线程 的具体细节通过 AzureAIAgentThread
类抽象化,这是 AgentThread
的实现。
USER_INPUTS = ["Hello", "What's your name?"]
thread: AzureAIAgentThread = AzureAIAgentThread()
try:
for user_input in USER_INPUTS:
response = await agent.get_response(messages=user_inputs, thread=thread)
print(response)
thread = response.thread
finally:
await thread.delete() if thread else None
(可选)可以调用代理,如下所示:
for user_input in USER_INPUTS:
async for content in agent.invoke(messages=user_input, thread=thread):
print(content.content)
thread = response.thread
还可以将消息列表传递给 get_response(...)
、invoke(...)
或 invoke_stream(...)
方法:
USER_INPUTS = ["Hello", "What's your name?"]
thread: AzureAIAgentThread = AzureAIAgentThread()
try:
for user_input in USER_INPUTS:
response = await agent.get_response(messages=USER_INPUTS, thread=thread)
print(response)
thread = response.thread
finally:
await thread.delete() if thread else None
代理还可以生成流式响应:
ChatMessageContent message = new(AuthorRole.User, "<your user input>");
await foreach (StreamingChatMessageContent response in agent.InvokeStreamingAsync(message, agentThread))
{
Console.Write(response.Content);
}
for user_input in USER_INPUTS:
await agent.add_chat_message(thread_id=thread.id, message=user_input)
async for content in agent.invoke_stream(thread_id=thread.id):
print(content.content, end="", flush=True)
功能当前在 Java 中不可用。
将插件与 AzureAIAgent
配合使用
语义内核支持使用自定义插件扩展 AzureAIAgent
以提高功能:
KernelPlugin plugin = KernelPluginFactory.CreateFromType<YourPlugin>();
PersistentAgentsClient client = AzureAIAgent.CreateAgentsClient("<your endpoint>", new AzureCliCredential());
PersistentAgent definition = await agentsClient.Administration.CreateAgentAsync(
"<name of the the model used by the agent>",
name: "<agent name>",
description: "<agent description>",
instructions: "<agent instructions>");
AzureAIAgent agent = new(definition, agentsClient, plugins: [plugin]);
from semantic_kernel.functions import kernel_function
class SamplePlugin:
@kernel_function(description="Provides sample data.")
def get_data(self) -> str:
return "Sample data"
async with (
DefaultAzureCredential() as creds,
AzureAIAgent.create_client(credential=creds) as client,
):
agent_definition = await client.agents.create_agent(
model=AzureAIAgentSettings().model_deployment_name,
)
agent = AzureAIAgent(
client=client,
definition=agent_definition,
plugins=[SamplePlugin()]
)
功能当前在 Java 中不可用。
高级功能
AzureAIAgent
可以利用高级工具,例如:
代码解释器
代码解释器允许代理在沙盒执行环境中编写和运行 Python 代码(Azure AI 代理服务代码解释器)。
PersistentAgentsClient client = AzureAIAgent.CreateAgentsClient("<your endpoint>", new AzureCliCredential());
PersistentAgent definition = await agentsClient.CreateAgentAsync(
"<name of the the model used by the agent>",
name: "<agent name>",
description: "<agent description>",
instructions: "<agent instructions>",
tools: [new CodeInterpreterToolDefinition()],
toolResources:
new()
{
CodeInterpreter = new()
{
FileIds = { ... },
}
}));
AzureAIAgent agent = new(definition, agentsClient);
from azure.ai.agents.models import CodeInterpreterTool
async with (
DefaultAzureCredential() as creds,
AzureAIAgent.create_client(credential=creds) as client,
):
code_interpreter = CodeInterpreterTool()
agent_definition = await client.agents.create_agent(
model=ai_agent_settings.model_deployment_name,
tools=code_interpreter.definitions,
tool_resources=code_interpreter.resources,
)
功能当前在 Java 中不可用。
文件搜索
文件搜索利用其模型之外的知识提升代理的能力(Azure AI 代理服务文件搜索工具)。
PersistentAgentsClient client = AzureAIAgent.CreateAgentsClient("<your endpoint>", new AzureCliCredential());
PersistentAgent definition = await agentsClient.CreateAgentAsync(
"<name of the the model used by the agent>",
name: "<agent name>",
description: "<agent description>",
instructions: "<agent instructions>",
tools: [new FileSearchToolDefinition()],
toolResources:
new()
{
FileSearch = new()
{
VectorStoreIds = { ... },
}
});
AzureAIAgent agent = new(definition, agentsClient);
from azure.ai.agents.models import FileSearchTool
async with (
DefaultAzureCredential() as creds,
AzureAIAgent.create_client(credential=creds) as client,
):
file_search = FileSearchTool(vector_store_ids=[vector_store.id])
agent_definition = await client.agents.create_agent(
model=ai_agent_settings.model_deployment_name,
tools=file_search.definitions,
tool_resources=file_search.resources,
)
功能当前在 Java 中不可用。
OpenAPI 集成
将代理连接到外部 API(如何将 Azure AI 代理服务与 OpenAPI 指定工具配合使用)。
PersistentAgentsClient client = AzureAIAgent.CreateAgentsClient("<your endpoint>", new AzureCliCredential());
string apiJsonSpecification = ...; // An Open API JSON specification
PersistentAgent definition = await agentsClient.CreateAgentAsync(
"<name of the the model used by the agent>",
name: "<agent name>",
description: "<agent description>",
instructions: "<agent instructions>",
tools: [
new OpenApiToolDefinition(
"<api name>",
"<api description>",
BinaryData.FromString(apiJsonSpecification),
new OpenApiAnonymousAuthDetails())
]
);
AzureAIAgent agent = new(definition, agentsClient);
from azure.ai.agents.models import OpenApiTool, OpenApiAnonymousAuthDetails
async with (
DefaultAzureCredential() as creds,
AzureAIAgent.create_client(credential=creds) as client,
):
openapi_spec_file_path = "sample/filepath/..."
with open(os.path.join(openapi_spec_file_path, "spec_one.json")) as file_one:
openapi_spec_one = json.loads(file_one.read())
with open(os.path.join(openapi_spec_file_path, "spec_two.json")) as file_two:
openapi_spec_two = json.loads(file_two.read())
# Note that connection or managed identity auth setup requires additional setup in Azure
auth = OpenApiAnonymousAuthDetails()
openapi_tool_one = OpenApiTool(
name="<name>",
spec=openapi_spec_one,
description="<description>",
auth=auth,
)
openapi_tool_two = OpenApiTool(
name="<name>",
spec=openapi_spec_two,
description="<description>",
auth=auth,
)
agent_definition = await client.agents.create_agent(
model=ai_agent_settings.model_deployment_name,
tools=openapi_tool_one.definitions + openapi_tool_two.definitions,
)
功能当前在 Java 中不可用。
AzureAI 搜索集成
将现有的 Azure AI 搜索索引与代理配合使用(使用现有的 AI 搜索索引)。
PersistentAgentsClient client = AzureAIAgent.CreateAgentsClient("<your endpoint>", new AzureCliCredential());
PersistentAgent definition = await agentsClient.CreateAgentAsync(
"<name of the the model used by the agent>",
name: "<agent name>",
description: "<agent description>",
instructions: "<agent instructions>",
tools: [new AzureAISearchToolDefinition()],
toolResources: new()
{
AzureAISearch = new()
{
IndexList = { new AISearchIndexResource("<your connection id>", "<your index name>") }
}
});
AzureAIAgent agent = new(definition, agentsClient);
from azure.ai.agents.models import AzureAISearchTool, ConnectionType
async with (
DefaultAzureCredential() as creds,
AzureAIAgent.create_client(credential=creds) as client,
):
conn_list = await client.connections.list()
ai_search_conn_id = ""
for conn in conn_list:
if conn.connection_type == ConnectionType.AZURE_AI_SEARCH:
ai_search_conn_id = conn.id
break
ai_search = AzureAISearchTool(
index_connection_id=ai_search_conn_id,
index_name=AZURE_AI_SEARCH_INDEX_NAME,
)
agent_definition = await client.agents.create_agent(
model=ai_agent_settings.model_deployment_name,
instructions="Answer questions using your index.",
tools=ai_search.definitions,
tool_resources=ai_search.resources,
headers={"x-ms-enable-preview": "true"},
)
功能当前在 Java 中不可用。
必应地面
即将推出示例。
from azure.ai.agents.models import BingGroundingTool
from azure.identity.aio import DefaultAzureCredential
from semantic_kernel.agents import AzureAIAgent, AzureAIAgentSettings
async with (
DefaultAzureCredential() as creds,
AzureAIAgent.create_client(credential=creds) as client,
):
# 1. Enter your Bing Grounding Connection Name
bing_connection = await client.connections.get(connection_name="<your-bing-grounding-connection-name>")
conn_id = bing_connection.id
# 2. Initialize agent bing tool and add the connection id
bing_grounding = BingGroundingTool(connection_id=conn_id)
# 3. Create an agent with Bing grounding on the Azure AI agent service
agent_definition = await client.agents.create_agent(
name="BingGroundingAgent",
instructions="Use the Bing grounding tool to answer the user's question.",
model=AzureAIAgentSettings().model_deployment_name,
tools=bing_grounding.definitions,
)
# 4. Create a Semantic Kernel agent for the Azure AI agent
agent = AzureAIAgent(
client=client,
definition=agent_definition,
)
使用 Bing Grounding 工具时,传递到 FunctionCallContent
回调的 on_intermediate_message
的函数名称将设置为 "bing_grounding"
。 运行完成后,ChatMessageContent.items
列表将包括 AnnotationContent
或 StreamingAnnotationContent
, 具体取决于调用是标准还是流式处理。 这些批注项包含代理在响应过程中访问的链接信息,类似于FunctionCallContent
中提供的信息。
有关详细信息,请参阅以下概念示例:
功能当前在 Java 中不可用。
检索现有 AzureAIAgent
可以通过指定现有的助理ID来检索并重复使用某个助理:
PersistentAgent definition = await agentsClient.Administration.GetAgentAsync("<your agent id>");
AzureAIAgent agent = new(definition, agentsClient);
agent_definition = await client.agents.get_agent(assistant_id="your-agent-id")
agent = AzureAIAgent(client=client, definition=agent_definition)
功能当前在 Java 中不可用。
删除 AzureAIAgent
不再需要代理及其关联的线程时,可以删除:
await agentThread.DeleteAsync();
await agentsClient.Administration.DeleteAgentAsync(agent.Id);
await client.agents.delete_thread(thread.id)
await client.agents.delete_agent(agent.id)
如果使用矢量存储或文件,它们也可能被删除:
await agentsClient.VectorStores.DeleteVectorStoreAsync("<your store id>");
await agentsClient.Files.DeleteFileAsync("<your file id>");
await client.agents.files.delete(file_id=file.id)
await client.agents.vector_stores.delete(vector_store_id=vector_store.id)
功能当前在 Java 中不可用。
有关 文件搜索 工具的详细信息,请参阅 Azure AI 代理服务文件搜索工具 一文。
操作说明
有关使用 AzureAIAgent
的实际示例,请参阅 GitHub 上的代码示例:
功能当前在 Java 中不可用。
使用 AzureAIAgent
处理中间消息
语义内核 AzureAIAgent
旨在调用满足用户查询或问题的代理。 在调用期间,代理可以执行工具来派生最终答案。 若要访问在此过程中生成的中间消息,调用方可以提供一个回调函数来处理FunctionCallContent
或FunctionResultContent
的实例。
即将推出
AzureAIAgent
回调文档。
在on_intermediate_message
或agent.invoke(...)
中配置agent.invoke_stream(...)
回调函数允许调用方在制定代理最终响应的过程中接收生成的中间消息。
import asyncio
from typing import Annotated
from azure.identity.aio import DefaultAzureCredential
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
# 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 (
DefaultAzureCredential() 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.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)
if __name__ == "__main__":
asyncio.run(main())
下面演示了代理调用过程的示例输出:
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 would like assistance with anything else!
User: 'Thank you'
Agent: You're welcome! Enjoy your meal! 😊
功能当前在 Java 中不可用。
声明性规范
即将推出有关使用声明式规格的文档。
重要
此功能处于实验阶段。 此阶段的功能正在开发中,在升级到预览阶段或候选发布阶段之前可能会更改。
AzureAIAgent
支持从 YAML 声明性规范中进行实例化。 声明性方法允许在单个可审核文档中定义代理的属性、说明、模型配置、工具和其他选项。 这使得代理组合可移植且易于跨环境进行管理。
注释
声明性 YAML 中列出的任何工具、函数或插件都必须在构造时可供代理使用。 对于基于内核的插件,这意味着必须在内核中注册它们。 对于必应 Grounding 工具、文件搜索工具或 OpenAPI 工具等内置工具,必须提供正确的配置和凭据。 代理加载程序不会从头开始创建函数。 如果缺少所需的组件,代理创建将失败。
如何使用声明性规范
本部分概述了关键原则,并提供概念示例的链接,这些示例显示了每种工具类型的完整代码,而不是枚举每个可能的 YAML 配置。 请参阅以下概念示例,了解具有声明性规范的 AzureAIAgent
端到端实现:
示例:从 YAML 创建 AzureAIAgent
最小 YAML 声明性规范可能如下所示:
type: foundry_agent
name: MyAgent
instructions: Respond politely to the user's questions.
model:
id: ${AzureAI:ChatModelId}
tools:
- id: MenuPlugin.get_specials
type: function
- id: MenuPlugin.get_item_price
type: function
有关如何连接代理的详细信息,请参阅上面的完整代码示例。
要点
- 声明性规范允许在 YAML 中定义代理结构、工具和行为。
- 必须在运行时注册或访问所有引用的工具和插件。
- 必应、文件搜索和代码解释器等内置工具需要适当的配置和凭据(通常通过环境变量或显式参数)。
- 有关全面的示例,请参阅提供的示例链接,这些链接演示了实际方案,包括插件注册、Azure 标识配置和高级工具使用。
此功能不可用。