本教程介绍如何将函数工具添加到 AG-UI 代理。 函数工具是自定义 C# 方法，代理可以调用这些方法来执行特定任务，例如检索数据、执行计算或与外部系统交互。 使用 AG-UI，这些工具在后端执行，其结果会自动流式传输到客户端。
先决条件
在开始之前，请确保已完成 入门 教程并具有：
- .NET 8.0 或更高版本
-
Microsoft.Agents.AI.Hosting.AGUI.AspNetCore已安装的包 - 已配置 Azure OpenAI 服务
- 基本了解 AG-UI 服务器和客户端设置
什么是后端工具渲染？
后端工具呈现意味着：
- 函数工具在服务器上定义
- AI 代理决定何时调用这些工具
- 在服务器端（后端）运行的工具
- 工具调用事件和结果实时流式传输到客户端
- 客户端接收有关工具执行进度的更新
使用函数工具创建 AG-UI 服务器
下面是演示如何使用复杂参数类型注册工具的完整服务器实现：
// Copyright (c) Microsoft. All rights reserved.
using System.ComponentModel;
using System.Text.Json.Serialization;
using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AGUI.AspNetCore;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.Options;
using OpenAI.Chat;
WebApplicationBuilder builder = WebApplication.CreateBuilder(args);
builder.Services.AddHttpClient().AddLogging();
builder.Services.ConfigureHttpJsonOptions(options =>
options.SerializerOptions.TypeInfoResolverChain.Add(SampleJsonSerializerContext.Default));
builder.Services.AddAGUI();
WebApplication app = builder.Build();
string endpoint = builder.Configuration["AZURE_OPENAI_ENDPOINT"]
?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
string deploymentName = builder.Configuration["AZURE_OPENAI_DEPLOYMENT_NAME"]
?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT_NAME is not set.");
// Define request/response types for the tool
internal sealed class RestaurantSearchRequest
{
public string Location { get; set; } = string.Empty;
public string Cuisine { get; set; } = "any";
}
internal sealed class RestaurantSearchResponse
{
public string Location { get; set; } = string.Empty;
public string Cuisine { get; set; } = string.Empty;
public RestaurantInfo[] Results { get; set; } = [];
}
internal sealed class RestaurantInfo
{
public string Name { get; set; } = string.Empty;
public string Cuisine { get; set; } = string.Empty;
public double Rating { get; set; }
public string Address { get; set; } = string.Empty;
}
// JSON serialization context for source generation
[JsonSerializable(typeof(RestaurantSearchRequest))]
[JsonSerializable(typeof(RestaurantSearchResponse))]
internal sealed partial class SampleJsonSerializerContext : JsonSerializerContext;
// Define the function tool
[Description("Search for restaurants in a location.")]
static RestaurantSearchResponse SearchRestaurants(
[Description("The restaurant search request")] RestaurantSearchRequest request)
{
// Simulated restaurant data
string cuisine = request.Cuisine == "any" ? "Italian" : request.Cuisine;
return new RestaurantSearchResponse
{
Location = request.Location,
Cuisine = request.Cuisine,
Results =
[
new RestaurantInfo
{
Name = "The Golden Fork",
Cuisine = cuisine,
Rating = 4.5,
Address = $"123 Main St, {request.Location}"
},
new RestaurantInfo
{
Name = "Spice Haven",
Cuisine = cuisine == "Italian" ? "Indian" : cuisine,
Rating = 4.7,
Address = $"456 Oak Ave, {request.Location}"
},
new RestaurantInfo
{
Name = "Green Leaf",
Cuisine = "Vegetarian",
Rating = 4.3,
Address = $"789 Elm Rd, {request.Location}"
}
]
};
}
// Get JsonSerializerOptions from the configured HTTP JSON options
Microsoft.AspNetCore.Http.Json.JsonOptions jsonOptions = app.Services.GetRequiredService<IOptions<Microsoft.AspNetCore.Http.Json.JsonOptions>>().Value;
// Create tool with serializer options
AITool[] tools =
[
AIFunctionFactory.Create(
SearchRestaurants,
serializerOptions: jsonOptions.SerializerOptions)
];
// Create the AI agent with tools
ChatClient chatClient = new AzureOpenAIClient(
new Uri(endpoint),
new DefaultAzureCredential())
.GetChatClient(deploymentName);
ChatClientAgent agent = chatClient.AsIChatClient().AsAIAgent(
name: "AGUIAssistant",
instructions: "You are a helpful assistant with access to restaurant information.",
tools: tools);
// Map the AG-UI agent endpoint
app.MapAGUI("/", agent);
await app.RunAsync();
关键概念
- 服务器端执行：工具在服务器进程中执行
- 自动流式处理：工具调用和结果实时流式传输到客户端
重要
使用复杂参数类型（对象、数组等）创建工具时，必须将 serializerOptions 参数提供给 AIFunctionFactory.Create()。 应通过JsonOptions应用程序配置的IOptions<Microsoft.AspNetCore.Http.Json.JsonOptions>获取序列化程序选项，以确保与应用程序的其他 JSON 序列化保持一致性。
运行服务器
设置环境变量并运行：
export AZURE_OPENAI_ENDPOINT="https://your-resource.openai.azure.com/"
export AZURE_OPENAI_DEPLOYMENT_NAME="gpt-4o-mini"
dotnet run --urls http://localhost:8888
监视客户端中的工具调用
入门教程中的基本客户端显示代理的最终文本响应。 但是，可以扩展它以观察从服务器以流方式传输的工具调用和结果。
显示工具执行详细信息
若要查看工具调用和实时结果，请扩展客户端的流式处理循环来处理 FunctionCallContent 和 FunctionResultContent：
// Inside the streaming loop from getting-started.md
await foreach (AgentResponseUpdate update in agent.RunStreamingAsync(messages, session))
{
ChatResponseUpdate chatUpdate = update.AsChatResponseUpdate();
// ... existing run started code ...
// Display streaming content
foreach (AIContent content in update.Contents)
{
switch (content)
{
case TextContent textContent:
Console.ForegroundColor = ConsoleColor.Cyan;
Console.Write(textContent.Text);
Console.ResetColor();
break;
case FunctionCallContent functionCallContent:
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine($"\n[Function Call - Name: {functionCallContent.Name}]");
// Display individual parameters
if (functionCallContent.Arguments != null)
{
foreach (var kvp in functionCallContent.Arguments)
{
Console.WriteLine($" Parameter: {kvp.Key} = {kvp.Value}");
}
}
Console.ResetColor();
break;
case FunctionResultContent functionResultContent:
Console.ForegroundColor = ConsoleColor.Magenta;
Console.WriteLine($"\n[Function Result - CallId: {functionResultContent.CallId}]");
if (functionResultContent.Exception != null)
{
Console.WriteLine($" Exception: {functionResultContent.Exception}");
}
else
{
Console.WriteLine($" Result: {functionResultContent.Result}");
}
Console.ResetColor();
break;
case ErrorContent errorContent:
Console.ForegroundColor = ConsoleColor.Red;
Console.WriteLine($"\n[Error: {errorContent.Message}]");
Console.ResetColor();
break;
}
}
}
工具调用的预期输出
当代理调用后端工具时，你将看到：
User (:q or quit to exit): What's the weather like in Amsterdam?
[Run Started - Thread: thread_abc123, Run: run_xyz789]
[Function Call - Name: SearchRestaurants]
Parameter: Location = Amsterdam
Parameter: Cuisine = any
[Function Result - CallId: call_def456]
Result: {"Location":"Amsterdam","Cuisine":"any","Results":[...]}
The weather in Amsterdam is sunny with a temperature of 22°C. Here are some
great restaurants in the area: The Golden Fork (Italian, 4.5 stars)...
[Run Finished - Thread: thread_abc123]
关键概念
-
FunctionCallContent：表示通过使用其Name和Arguments（参数键值对）来调用的工具 ：包含工具的 或 ，由 标识
后续步骤
您现在可以添加函数工具，接下来可以：
- 前端工具：添加前端工具。
- 使用 Dojo 进行测试：使用 AG-UI 的 Dojo 应用测试代理
其他资源
本教程介绍如何将函数工具添加到 AG-UI 代理。 函数工具是代理可以调用的自定义 Python 函数来执行特定任务，例如检索数据、执行计算或与外部系统交互。 使用 AG-UI，这些工具在后端执行，其结果会自动流式传输到客户端。
先决条件
在开始之前，请确保已完成 入门 教程并具有：
- Python 3.10 或更高版本
-
agent-framework-ag-ui安装 - 已配置 Azure OpenAI 服务
- 基本了解 AG-UI 服务器和客户端设置
注释
这些示例使用 DefaultAzureCredential 进行身份验证。 确保通过 Azure 进行身份验证（例如，通过 az login）。 有关详细信息，请参阅 Azure 标识文档。
什么是后端工具渲染？
后端工具呈现意味着：
- 函数工具在服务器上定义
- AI 代理决定何时调用这些工具
- 在服务器端（后端）运行的工具
- 工具调用事件和结果实时流式传输到客户端
- 客户端接收有关工具执行进度的更新
此方法提供：
- 安全性：敏感操作会在服务器上执行
- 一致性：所有客户端都使用相同的工具实现
- 透明度：客户端可以显示工具执行进度
- 灵活性：在不更改客户端代码的情况下更新工具
创建函数工具
基本功能工具
可以使用修饰器将任何 Python 函数转换为工具 @tool ：
from typing import Annotated
from pydantic import Field
from agent_framework import tool
@tool
def get_weather(
location: Annotated[str, Field(description="The city")],
) -> str:
"""Get the current weather for a location."""
# In a real application, you would call a weather API
return f"The weather in {location} is sunny with a temperature of 22°C."
关键概念
-
@tool修饰器：将函数标记为可用于代理 - 类型注释：提供参数的类型信息
-
Annotated和Field：添加说明以帮助代理了解参数 - Docstring：描述函数的作用（帮助代理决定何时使用它）
- 返回值：结果返回到代理（并流式传输到客户端）
多功能工具
可以提供多个工具，为代理提供更多功能：
from typing import Any
from agent_framework import tool
@tool
def get_weather(
location: Annotated[str, Field(description="The city.")],
) -> str:
"""Get the current weather for a location."""
return f"The weather in {location} is sunny with a temperature of 22°C."
@tool
def get_forecast(
location: Annotated[str, Field(description="The city.")],
days: Annotated[int, Field(description="Number of days to forecast")] = 3,
) -> dict[str, Any]:
"""Get the weather forecast for a location."""
return {
"location": location,
"days": days,
"forecast": [
{"day": 1, "weather": "Sunny", "high": 24, "low": 18},
{"day": 2, "weather": "Partly cloudy", "high": 22, "low": 17},
{"day": 3, "weather": "Rainy", "high": 19, "low": 15},
],
}
使用函数工具创建 AG-UI 服务器
下面是使用函数工具的完整服务器实现：
"""AG-UI server with backend tool rendering."""
import os
from typing import Annotated, Any
from agent_framework import Agent, tool
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework_ag_ui import add_agent_framework_fastapi_endpoint
from azure.identity import AzureCliCredential
from fastapi import FastAPI
from pydantic import Field
# Define function tools
@tool
def get_weather(
location: Annotated[str, Field(description="The city")],
) -> str:
"""Get the current weather for a location."""
# Simulated weather data
return f"The weather in {location} is sunny with a temperature of 22°C."
@tool
def search_restaurants(
location: Annotated[str, Field(description="The city to search in")],
cuisine: Annotated[str, Field(description="Type of cuisine")] = "any",
) -> dict[str, Any]:
"""Search for restaurants in a location."""
# Simulated restaurant data
return {
"location": location,
"cuisine": cuisine,
"results": [
{"name": "The Golden Fork", "rating": 4.5, "price": "$$"},
{"name": "Bella Italia", "rating": 4.2, "price": "$$$"},
{"name": "Spice Garden", "rating": 4.7, "price": "$$"},
],
}
# Read required configuration
endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT")
deployment_name = os.environ.get("AZURE_OPENAI_DEPLOYMENT_NAME")
if not endpoint:
raise ValueError("AZURE_OPENAI_ENDPOINT environment variable is required")
if not deployment_name:
raise ValueError("AZURE_OPENAI_DEPLOYMENT_NAME environment variable is required")
chat_client = AzureOpenAIChatClient(
credential=AzureCliCredential(),
endpoint=endpoint,
deployment_name=deployment_name,
)
# Create agent with tools
agent = Agent(
name="TravelAssistant",
instructions="You are a helpful travel assistant. Use the available tools to help users plan their trips.",
chat_client=chat_client,
tools=[get_weather, search_restaurants],
)
# Create FastAPI app
app = FastAPI(title="AG-UI Travel Assistant")
add_agent_framework_fastapi_endpoint(app, agent, "/")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8888)
了解工具事件
代理调用工具时，客户端会收到多个事件：
工具调用事件
# 1. TOOL_CALL_START - Tool execution begins
{
"type": "TOOL_CALL_START",
"toolCallId": "call_abc123",
"toolCallName": "get_weather"
}
# 2. TOOL_CALL_ARGS - Tool arguments (may stream in chunks)
{
"type": "TOOL_CALL_ARGS",
"toolCallId": "call_abc123",
"delta": "{\"location\": \"Paris, France\"}"
}
# 3. TOOL_CALL_END - Arguments complete
{
"type": "TOOL_CALL_END",
"toolCallId": "call_abc123"
}
# 4. TOOL_CALL_RESULT - Tool execution result
{
"type": "TOOL_CALL_RESULT",
"toolCallId": "call_abc123",
"content": "The weather in Paris, France is sunny with a temperature of 22°C."
}
增强的工具事件客户端
这里有一个使用 AGUIChatClient 来显示工具执行的增强客户端：
"""AG-UI client with tool event handling."""
import asyncio
import os
from agent_framework import Agent, ToolCallContent, ToolResultContent
from agent_framework_ag_ui import AGUIChatClient
async def main():
"""Main client loop with tool event display."""
server_url = os.environ.get("AGUI_SERVER_URL", "http://127.0.0.1:8888/")
print(f"Connecting to AG-UI server at: {server_url}\n")
# Create AG-UI chat client
chat_client = AGUIChatClient(server_url=server_url)
# Create agent with the chat client
agent = Agent(
name="ClientAgent",
chat_client=chat_client,
instructions="You are a helpful assistant.",
)
# Get a thread for conversation continuity
thread = agent.create_session()
try:
while True:
message = input("\nUser (:q or quit to exit): ")
if not message.strip():
continue
if message.lower() in (":q", "quit"):
break
print("\nAssistant: ", end="", flush=True)
async for update in agent.run(message, session=thread, stream=True):
# Display text content
if update.text:
print(f"\033[96m{update.text}\033[0m", end="", flush=True)
# Display tool calls and results
for content in update.contents:
if isinstance(content, ToolCallContent):
print(f"\n\033[95m[Calling tool: {content.name}]\033[0m")
elif isinstance(content, ToolResultContent):
result_text = content.result if isinstance(content.result, str) else str(content.result)
print(f"\033[94m[Tool result: {result_text}]\033[0m")
print("\n")
except KeyboardInterrupt:
print("\n\nExiting...")
except Exception as e:
print(f"\n\033[91mError: {e}\033[0m")
if __name__ == "__main__":
asyncio.run(main())
示例交互
在增强的服务器和客户端正在运行的情况下：
User (:q or quit to exit): What's the weather like in Paris and suggest some Italian restaurants?
[Run Started]
[Tool Call: get_weather]
[Tool Result: The weather in Paris, France is sunny with a temperature of 22°C.]
[Tool Call: search_restaurants]
[Tool Result: {"location": "Paris", "cuisine": "Italian", "results": [...]}]
Based on the current weather in Paris (sunny, 22°C) and your interest in Italian cuisine,
I'd recommend visiting Bella Italia, which has a 4.2 rating. The weather is perfect for
outdoor dining!
[Run Finished]
工具实现最佳做法
错误处理
在工具中优雅地处理错误：
@tool
def get_weather(
location: Annotated[str, Field(description="The city.")],
) -> str:
"""Get the current weather for a location."""
try:
# Call weather API
result = call_weather_api(location)
return f"The weather in {location} is {result['condition']} with temperature {result['temp']}°C."
except Exception as e:
return f"Unable to retrieve weather for {location}. Error: {str(e)}"
丰富的返回类型
适当时返回结构化数据：
@tool
def analyze_sentiment(
text: Annotated[str, Field(description="The text to analyze")],
) -> dict[str, Any]:
"""Analyze the sentiment of text."""
# Perform sentiment analysis
return {
"text": text,
"sentiment": "positive",
"confidence": 0.87,
"scores": {
"positive": 0.87,
"neutral": 0.10,
"negative": 0.03,
},
}
描述性文档
提供明确的说明，以帮助代理了解何时使用工具：
@tool
def book_flight(
origin: Annotated[str, Field(description="Departure city and airport code, e.g., 'New York, JFK'")],
destination: Annotated[str, Field(description="Arrival city and airport code, e.g., 'London, LHR'")],
date: Annotated[str, Field(description="Departure date in YYYY-MM-DD format")],
passengers: Annotated[int, Field(description="Number of passengers")] = 1,
) -> dict[str, Any]:
"""
Book a flight for specified passengers from origin to destination.
This tool should be used when the user wants to book or reserve airline tickets.
Do not use this for searching flights - use search_flights instead.
"""
# Implementation
pass
具有类的工具组织
对于相关工具，请在类中组织它们：
from agent_framework import tool
class WeatherTools:
"""Collection of weather-related tools."""
def __init__(self, api_key: str):
self.api_key = api_key
@tool
def get_current_weather(
self,
location: Annotated[str, Field(description="The city.")],
) -> str:
"""Get current weather for a location."""
# Use self.api_key to call API
return f"Current weather in {location}: Sunny, 22°C"
@tool
def get_forecast(
self,
location: Annotated[str, Field(description="The city.")],
days: Annotated[int, Field(description="Number of days")] = 3,
) -> dict[str, Any]:
"""Get weather forecast for a location."""
# Use self.api_key to call API
return {"location": location, "forecast": [...]}
# Create tools instance
weather_tools = WeatherTools(api_key="your-api-key")
# Create agent with class-based tools
agent = Agent(
name="WeatherAgent",
instructions="You are a weather assistant.",
chat_client=AzureOpenAIChatClient(...),
tools=[
weather_tools.get_current_weather,
weather_tools.get_forecast,
],
)
后续步骤
现在你了解后端工具的渲染后，你可以：
- 创建高级工具：详细了解如何使用 Agent Framework 创建函数工具