Representación de herramientas de back-end con AG-UI

Las herramientas de backend usan el flujo de herramientas normal de MAF. AG-UI agrega eventos de transporte para que un cliente pueda observar la llamada y el resultado; no introduce una abstracción de herramientas independiente.

Añadir una herramienta de servidor

Defina y registre la herramienta como lo haría para cualquier agente de MAF:

using System.ComponentModel;
using Microsoft.Extensions.AI;

[Description("Get the weather for a location.")]
static string GetWeather(
    [Description("The city to look up.")] string location) =>
    $"The weather in {location} is sunny.";

AITool getWeather = AIFunctionFactory.Create(GetWeather, name: "get_weather");
AIAgent agent = chatClient.AsAIAgent(tools: [getWeather]);

app.MapAGUIServer("/", agent);

Para los tipos de solicitud o respuesta complejos, configure lo mismo JsonSerializerOptions para ASP.NET Core y AIFunctionFactory.Create.

Para conocer los esquemas de herramientas, la inserción de dependencias, el control de errores y el diseño general de herramientas, consulte Uso de herramientas de funciones con un agente.

asignación de eventos de AG-UI

Cuando el agente llama a la herramienta:

• FunctionCallContent se emite como eventos AG-UI TOOL_CALL_START, TOOL_CALL_ARGSy TOOL_CALL_END .
• FunctionResultContent se emite como evento TOOL_CALL_RESULT.
• El texto y el resto del contenido del agente continúan fluyendo con normalidad.

Un cliente .NET recibe el contenido traducido como FunctionCallContent y FunctionResultContent:

await foreach (AgentResponseUpdate update in agent.RunStreamingAsync(messages, session))
{
    foreach (AIContent content in update.Contents)
    {
        if (content is FunctionCallContent call)
        {
            Console.WriteLine($"Calling {call.Name}");
        }
        else if (content is FunctionResultContent result)
        {
            Console.WriteLine($"Result: {result.Result}");
        }
    }
}

Los resultados de la herramienta son valores orientados al modelo que AG-UI también expone al cliente. Para emitir el estado compartido de la interfaz de usuario, además del resultado de una herramienta, use las correspondencias explícitas descritas en Administración del estado.

Pasos siguientes

En este tutorial se muestra cómo agregar herramientas de función a los agentes de AG-UI. Las herramientas de funciones son funciones personalizadas de Python a las que el agente puede llamar para realizar tareas específicas, como recuperar datos, realizar cálculos o interactuar con sistemas externos. Con AG-UI, estas herramientas se ejecutan en el back-end y sus resultados se transmiten automáticamente al cliente.

Prerequisites

Antes de comenzar, asegúrese de que ha completado el tutorial de introducción y que tiene:

• Python 3.10 o posterior
• agent-framework-ag-ui Instalado
• Servicio OpenAI de Azure configurado
• Conocimientos básicos de la configuración de cliente y servidor de AG-UI

¿Qué es el renderizado de herramientas de backend?

La representación de herramientas de backend se refiere a:

• Las herramientas de función se definen en el servidor
• El agente de IA decide cuándo llamar a estas herramientas
• Las herramientas se ejecutan en el back-end (servidor)
• Los eventos y los resultados de la llamada a la herramienta se transmiten al cliente en tiempo real
• El cliente recibe actualizaciones sobre el progreso de la ejecución de la herramienta.

Este enfoque proporciona:

• Seguridad: las operaciones confidenciales permanecen en el servidor
• Coherencia: todos los clientes usan las mismas implementaciones de herramientas
• Transparencia: los clientes pueden mostrar el progreso de la ejecución de la herramienta
• Flexibilidad: actualización de herramientas sin cambiar el código de cliente

Creación de herramientas de funciones

Herramienta de función básica

Puede convertir cualquier función de Python en una herramienta mediante el @tool decorador:

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."

Conceptos clave

• @tool decorador: marca una función como disponible para el agente
• Anotaciones de tipo: proporcionar información de tipo para los parámetros.
• Annotated y Field: se agregan descripciones para ayudar al agente a comprender los parámetros.
• Docstring: describe lo que hace la función (ayuda al agente a decidir cuándo usarlo)
• Valor devuelto: el resultado devuelto al agente (y transmitido al cliente)

Herramientas multifuncionales

Puede proporcionar varias herramientas para proporcionar al agente más funcionalidades:

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},
        ],
    }

Creación de un servidor de AG-UI con herramientas de funciones

Esta es una implementación completa del servidor con herramientas de función:

"""AG-UI server with backend tool rendering."""

import os
from typing import Annotated, Any

from agent_framework import Agent, tool
from agent_framework.openai import OpenAIChatCompletionClient
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_CHAT_COMPLETION_MODEL")

if not endpoint:
    raise ValueError("AZURE_OPENAI_ENDPOINT environment variable is required")
if not deployment_name:
    raise ValueError("AZURE_OPENAI_CHAT_COMPLETION_MODEL environment variable is required")

chat_client = OpenAIChatCompletionClient(
    model=deployment_name,
    azure_endpoint=endpoint,
    api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
    credential=AzureCliCredential(),
)

# 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.",
    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)

Comprensión de eventos de herramientas

Cuando el agente llama a una herramienta, el cliente recibe varios eventos:

Eventos de llamada de herramienta

# 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."
}

Cliente mejorado para eventos de herramientas

Este es un cliente mejorado que utiliza AGUIChatClient para mostrar la ejecución de la herramienta.

"""AG-UI client with tool event handling."""

import asyncio
import os

from agent_framework import Agent
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(endpoint=server_url)

    # Create agent with the chat client
    agent = Agent(
        name="ClientAgent",
        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 content.type == "function_call":
                        print(f"\n\033[95m[Calling tool: {content.name}]\033[0m")
                    elif content.type == "function_result":
                        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())

Interacción de ejemplo

Con el servidor mejorado y el cliente en ejecución:

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]

Procedimientos recomendados de implementación de herramientas

Tratamiento de errores

Maneje los errores de manera eficiente en las herramientas:

@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)}"

Tipos de retorno enriquecidos

Devolver datos estructurados cuando corresponda:

@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,
        },
    }

Documentación descriptiva

Proporcione descripciones claras para ayudar al agente a comprender cuándo usar herramientas:

@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

Organización de herramientas con clases

En el caso de las herramientas relacionadas, organícelas en una clase:

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.",
    client=OpenAIChatCompletionClient(...),
    tools=[
        weather_tools.get_current_weather,
        weather_tools.get_forecast,
    ],
)

Pasos siguientes

Ahora que comprende la representación de herramientas de back-end, puede hacer lo siguiente:

• Creación de herramientas avanzadas: Más información sobre la creación de herramientas de funciones con Agent Framework

Recursos adicionales

• Información general deAG-UI
• Introducción a AG-UI
• Tutorial de herramientas funcionales

Los servidores AG-UI de Go pueden exponer las herramientas de función estándar de Agent Framework. Cree herramientas con tool/functool, adjúntelas al agente alojado e implemente el agente con aguiprovider.NewJSONHTTPHandler.

searchRestaurants := functool.MustNew(functool.Config{
    Name:        "search_restaurants",
    Description: "Search for restaurants in a location.",
}, func(ctx context.Context, in restaurantSearchRequest) (restaurantSearchResponse, error) {
    return restaurantSearchResponse{
        Location: in.Location,
        Cuisine:  in.Cuisine,
        Results:  []restaurantInfo{{Name: "The Golden Fork", Cuisine: in.Cuisine}},
    }, nil
})

a := foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
    Config: agent.Config{
        Tools: []tool.Tool{searchRestaurants},
    },
})