Vykreslování nástrojů front-endu pomocí AG-UI

V tomto kurzu se dozvíte, jak do klientů AG-UI přidat nástroje front-endových funkcí. Front-endové nástroje jsou funkce, které se spouští na straně klienta a umožňují agentovi AI pracovat s místním prostředím uživatele, přistupovat k datům specifických pro klienta nebo provádět operace uživatelského rozhraní. Server orchestruje, kdy tyto nástroje volat, ale provedení zcela probíhá na klientovi.

Požadavky

Než začnete, ujistěte se, že jste dokončili kurz Začínáme a měli:

• .NET 8.0 nebo novější
• Microsoft.Agents.AI.AGUI nainstalovaný balíček
• Microsoft.Agents.AI nainstalovaný balíček
• Základní znalost nastavení klienta AG-UI

Co jsou front-endové nástroje?

Front-endové nástroje jsou nástroje funkcí, které:

• Jsou definovány a registrovány v klientovi.
• Spuštění v prostředí klienta (ne na serveru)
• Povolit agentu AI interakci s prostředky specifickými pro klienta
• Zadejte výsledky zpět na server, aby agent začlenil do odpovědí.
• Umožnění personalizovaných a kontextově uvědomělých zážitků

Běžné případy použití:

• Čtení dat místního senzoru (GPS, teplota atd.)
• Přístup k úložišti na straně klienta nebo předvoleb
• Provádění operací uživatelského rozhraní (změna motivů, zobrazování oznámení)
• Interakce s funkcemi specifickými pro zařízení (kamera, mikrofon)

Registrace front-endových nástrojů v klientovi

Hlavní rozdíl oproti kurzu Začínáme spočívá v registraci nástrojů v klientském agentu. Co se změní:

// Define a frontend function tool
[Description("Get the user's current location from GPS.")]
static string GetUserLocation()
{
    // Access client-side GPS
    return "Amsterdam, Netherlands (52.37°N, 4.90°E)";
}

// Create frontend tools
AITool[] frontendTools = [AIFunctionFactory.Create(GetUserLocation)];

// Pass tools when creating the agent
AIAgent agent = chatClient.AsAIAgent(
    name: "agui-client",
    description: "AG-UI Client Agent",
    tools: frontendTools);

Zbývající kód klienta zůstane stejný jako v kurzu Začínáme.

Jak se nástroje odesílají na server

Když zaregistrujete nástroje pomocí AsAIAgent(), AGUIChatClient automaticky:

1. Zaznamenává definice nástrojů (názvy, popisy, schémata parametrů).
2. Odešle nástroje s každou žádostí agentu serveru, který je mapuje na ChatAgentRunOptions.ChatOptions.Tools

Server obdrží deklarace klientského nástroje a model AI se může rozhodnout, kdy je volat.

Kontrola a úpravy nástrojů pomocí middlewaru

Ke kontrole nebo úpravě spuštění agenta můžete použít middleware agenta, včetně přístupu k nástrojům:

// Create agent with middleware that inspects tools
AIAgent inspectableAgent = baseAgent
    .AsBuilder()
    .Use(runFunc: null, runStreamingFunc: InspectToolsMiddleware)
    .Build();

static async IAsyncEnumerable<AgentResponseUpdate> InspectToolsMiddleware(
    IEnumerable<ChatMessage> messages,
    AgentSession? session,
    AgentRunOptions? options,
    AIAgent innerAgent,
    CancellationToken cancellationToken)
{
    // Access the tools from ChatClientAgentRunOptions
    if (options is ChatClientAgentRunOptions chatOptions)
    {
        IList<AITool>? tools = chatOptions.ChatOptions?.Tools;
        if (tools != null)
        {
            Console.WriteLine($"Tools available for this run: {tools.Count}");
            foreach (AITool tool in tools)
            {
                if (tool is AIFunction function)
                {
                    Console.WriteLine($"  - {function.Metadata.Name}: {function.Metadata.Description}");
                }
            }
        }
    }

    await foreach (AgentResponseUpdate update in innerAgent.RunStreamingAsync(messages, session, options, cancellationToken))
    {
        yield return update;
    }
}

Tento model middlewaru umožňuje:

• Před spuštěním ověřte definice nástrojů.

Klíčové koncepty

Následují nové koncepty pro front-endové nástroje:

• Registrace na straně klienta: Nástroje jsou v klientovi registrovány pomocí AIFunctionFactory.Create() a předány do AsAIAgent()
• Automatické zachytávání: Nástroje se automaticky zachytávají a odesílají prostřednictvím ChatAgentRunOptions.ChatOptions.Tools

Jak fungují front-endové nástroje

Server-Side proces

Server nezná podrobnosti implementace front-endových nástrojů. Ví to jen:

1. Názvy a popisy nástrojů (z registrace klienta)
2. Schémata parametrů
3. Kdy požádat o spuštění nástroje

Když se agent AI rozhodne volat frontendový nástroj:

1. Server odešle klientovi žádost o volání nástroje přes SSE.
2. Server čeká na spuštění nástroje klientem a vrátí výsledky.
3. Server zahrnuje výsledky do kontextu agenta.
4. Agent pokračuje ve zpracování na základě výsledků nástroje.

Klientský proces

Klient zpracovává spuštění front-endového nástroje:

1. FunctionCallContent Přijímá ze serveru označující požadavek na volání nástroje
2. Název nástroje se přiřadí k místně registrované funkci.
3. Deserializuje parametry z požadavku.
4. Spustí funkci místně.
5. Serializuje výsledek.
6. FunctionResultContent Odešle zpět na server.
7. Pokračuje v přijímání odpovědí agenta.

Očekávaný výstup s nástroji front-endu

Když agent volá nástroje uživatelského rozhraní, zobrazí se volání nástrojů a výsledek ve streamovaném výstupu.

User (:q or quit to exit): Where am I located?

[Client Tool Call - Name: GetUserLocation]
[Client Tool Result: Amsterdam, Netherlands (52.37°N, 4.90°E)]

You are currently in Amsterdam, Netherlands, at coordinates 52.37°N, 4.90°E.

Nastavení serveru pro front-endové nástroje

Server nepotřebuje speciální konfiguraci pro podporu front-endových nástrojů. Použijte standardní AG-UI server z kurzu Začínáme – automaticky:

• Přijímá deklarace nástrojů front-endu během připojení klienta.
• Žádosti o spuštění nástroje, když je agent AI potřebuje
• Čeká na výsledky od klienta.
• Začlení výsledky do rozhodování agenta.

Další kroky

Teď, když rozumíte front-endovým nástrojům, můžete:

• Kombinování s back-endovými nástroji: Společně používejte front-endové i back-endové nástroje.

Další zdroje

• Přehled AG-UI
• Úvodní kurz
• Vykreslování back-endových nástrojů
• Dokumentace k rozhraní Agent Framework

V tomto kurzu se dozvíte, jak do klientů AG-UI přidat nástroje front-endových funkcí. Front-endové nástroje jsou funkce, které se spouští na straně klienta a umožňují agentovi AI pracovat s místním prostředím uživatele, přistupovat k datům specifických pro klienta nebo provádět operace uživatelského rozhraní.

Požadavky

Než začnete, ujistěte se, že jste dokončili kurz Začínáme a měli:

• Python 3.10 nebo novější
• httpx nainstalovaná pro funkce klienta HTTP
• Základní znalost nastavení klienta AG-UI
• Nakonfigurovaná služba Azure OpenAI

Co jsou front-endové nástroje?

Front-endové nástroje jsou nástroje funkcí, které:

• Jsou definovány a registrovány v klientovi.
• Spuštění v prostředí klienta (ne na serveru)
• Povolit agentu AI interakci s prostředky specifickými pro klienta
• Zadejte výsledky zpět na server, aby agent začlenil do odpovědí.

Běžné případy použití:

• Čtení dat místního senzoru
• Přístup k úložišti na straně klienta nebo předvoleb
• Provádění operací uživatelského rozhraní
• Interakce s funkcemi specifickými pro zařízení

Vytváření front-endových nástrojů

Front-endové nástroje v Pythonu se definují podobně jako back-endové nástroje, ale jsou zaregistrované v klientovi:

from typing import Annotated
from pydantic import BaseModel, Field


class SensorReading(BaseModel):
    """Sensor reading from client device."""
    temperature: float
    humidity: float
    air_quality_index: int


def read_climate_sensors(
    include_temperature: Annotated[bool, Field(description="Include temperature reading")] = True,
    include_humidity: Annotated[bool, Field(description="Include humidity reading")] = True,
) -> SensorReading:
    """Read climate sensor data from the client device."""
    # Simulate reading from local sensors
    return SensorReading(
        temperature=22.5 if include_temperature else 0.0,
        humidity=45.0 if include_humidity else 0.0,
        air_quality_index=75,
    )


def change_background_color(color: Annotated[str, Field(description="Color name")] = "blue") -> str:
    """Change the console background color."""
    # Simulate UI change
    print(f"\n🎨 Background color changed to {color}")
    return f"Background changed to {color}"

Vytvoření klienta AG-UI pomocí nástrojů front-endu

Tady je kompletní implementace klienta s front-endovými nástroji:

"""AG-UI client with frontend tools."""

import asyncio
import json
import os
from typing import Annotated, AsyncIterator

import httpx
from pydantic import BaseModel, Field


class SensorReading(BaseModel):
    """Sensor reading from client device."""
    temperature: float
    humidity: float
    air_quality_index: int


# Define frontend tools
def read_climate_sensors(
    include_temperature: Annotated[bool, Field(description="Include temperature")] = True,
    include_humidity: Annotated[bool, Field(description="Include humidity")] = True,
) -> SensorReading:
    """Read climate sensor data from the client device."""
    return SensorReading(
        temperature=22.5 if include_temperature else 0.0,
        humidity=45.0 if include_humidity else 0.0,
        air_quality_index=75,
    )


def get_user_location() -> dict:
    """Get the user's current GPS location."""
    # Simulate GPS reading
    return {
        "latitude": 52.3676,
        "longitude": 4.9041,
        "accuracy": 10.0,
        "city": "Amsterdam",
    }


# Tool registry maps tool names to functions
FRONTEND_TOOLS = {
    "read_climate_sensors": read_climate_sensors,
    "get_user_location": get_user_location,
}


class AGUIClientWithTools:
    """AG-UI client with frontend tool support."""

    def __init__(self, server_url: str, tools: dict):
        self.server_url = server_url
        self.tools = tools
        self.thread_id: str | None = None

    async def send_message(self, message: str) -> AsyncIterator[dict]:
        """Send a message and handle streaming response with tool execution."""
        # Prepare tool declarations for the server
        tool_declarations = []
        for name, func in self.tools.items():
            tool_declarations.append({
                "name": name,
                "description": func.__doc__ or "",
                # Add parameter schema from function signature
            })

        request_data = {
            "messages": [
                {"role": "system", "content": "You are a helpful assistant with access to client tools."},
                {"role": "user", "content": message},
            ],
            "tools": tool_declarations,  # Send tool declarations to server
        }

        if self.thread_id:
            request_data["thread_id"] = self.thread_id

        async with httpx.AsyncClient(timeout=60.0) as client:
            async with client.stream(
                "POST",
                self.server_url,
                json=request_data,
                headers={"Accept": "text/event-stream"},
            ) as response:
                response.raise_for_status()

                async for line in response.aiter_lines():
                    if line.startswith("data: "):
                        data = line[6:]
                        try:
                            event = json.loads(data)

                            # Handle tool call requests from server
                            if event.get("type") == "TOOL_CALL_REQUEST":
                                await self._handle_tool_call(event, client)
                            else:
                                yield event

                            # Capture thread_id
                            if event.get("type") == "RUN_STARTED" and not self.thread_id:
                                self.thread_id = event.get("threadId")

                        except json.JSONDecodeError:
                            continue

    async def _handle_tool_call(self, event: dict, client: httpx.AsyncClient):
        """Execute frontend tool and send result back to server."""
        tool_name = event.get("toolName")
        tool_call_id = event.get("toolCallId")
        arguments = event.get("arguments", {})

        print(f"\n\033[95m[Client Tool Call: {tool_name}]\033[0m")
        print(f"  Arguments: {arguments}")

        try:
            # Execute the tool
            tool_func = self.tools.get(tool_name)
            if not tool_func:
                raise ValueError(f"Unknown tool: {tool_name}")

            result = tool_func(**arguments)

            # Convert Pydantic models to dict
            if hasattr(result, "model_dump"):
                result = result.model_dump()

            print(f"\033[94m[Client Tool Result: {result}]\033[0m")

            # Send result back to server
            await client.post(
                f"{self.server_url}/tool_result",
                json={
                    "tool_call_id": tool_call_id,
                    "result": result,
                },
            )

        except Exception as e:
            print(f"\033[91m[Tool Error: {e}]\033[0m")
            # Send error back to server
            await client.post(
                f"{self.server_url}/tool_result",
                json={
                    "tool_call_id": tool_call_id,
                    "error": str(e),
                },
            )


async def main():
    """Main client loop with frontend tools."""
    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")

    client = AGUIClientWithTools(server_url, FRONTEND_TOOLS)

    try:
        while True:
            message = input("\nUser (:q or quit to exit): ")
            if not message.strip():
                continue

            if message.lower() in (":q", "quit"):
                break

            print()
            async for event in client.send_message(message):
                event_type = event.get("type", "")

                if event_type == "RUN_STARTED":
                    print(f"\033[93m[Run Started]\033[0m")

                elif event_type == "TEXT_MESSAGE_CONTENT":
                    print(f"\033[96m{event.get('delta', '')}\033[0m", end="", flush=True)

                elif event_type == "RUN_FINISHED":
                    print(f"\n\033[92m[Run Finished]\033[0m")

                elif event_type == "RUN_ERROR":
                    error_msg = event.get("message", "Unknown error")
                    print(f"\n\033[91m[Error: {error_msg}]\033[0m")

            print()

    except KeyboardInterrupt:
        print("\n\nExiting...")
    except Exception as e:
        print(f"\n\033[91mError: {e}\033[0m")


if __name__ == "__main__":
    asyncio.run(main())

Jak fungují front-endové nástroje

Tok protokolu

1. Registrace klienta: Klient odesílá deklarace nástrojů (názvy, popisy, parametry) na server
2. Orchestrace serveru: Agent AI se rozhodne, kdy volat front-endové nástroje na základě požadavku uživatele.
3. Žádost o volání nástroje: Server odesílá TOOL_CALL_REQUEST událost klientovi přes SSE.
4. Spuštění klienta: Klient spustí nástroj místně.
5. Odeslání výsledku: Klient odešle výsledek zpět na server prostřednictvím požadavku POST.
6. Zpracování agenta: Server zahrnuje výsledek a pokračuje v odpovědi.

Klíčové události

• TOOL_CALL_REQUEST: Server požaduje spuštění front-endového nástroje
• TOOL_CALL_RESULT: Klient odešle výsledek spuštění (přes HTTP POST).

Očekávaný výstup

User (:q or quit to exit): What's the temperature reading from my sensors?

[Run Started]

[Client Tool Call: read_climate_sensors]
  Arguments: {'include_temperature': True, 'include_humidity': True}
[Client Tool Result: {'temperature': 22.5, 'humidity': 45.0, 'air_quality_index': 75}]

Based on your sensor readings, the current temperature is 22.5°C and the 
humidity is at 45%. These are comfortable conditions!
[Run Finished]

Nastavení serveru

Standardní AG-UI server z kurzu Začínáme automaticky podporuje front-endové nástroje. Na straně serveru nejsou potřeba žádné změny – zpracovává orchestraci nástrojů automaticky.

Osvědčené postupy

Zabezpečení

def access_sensitive_data() -> str:
    """Access user's sensitive data."""
    # Always check permissions first
    if not has_permission():
        return "Error: Permission denied"

    try:
        # Access data
        return "Data retrieved"
    except Exception as e:
        # Don't expose internal errors
        return "Unable to access data"

Zpracování chyb

def read_file(path: str) -> str:
    """Read a local file."""
    try:
        with open(path, "r") as f:
            return f.read()
    except FileNotFoundError:
        return f"Error: File not found: {path}"
    except PermissionError:
        return f"Error: Permission denied: {path}"
    except Exception as e:
        return f"Error reading file: {str(e)}"

Asynchronní operace

async def capture_photo() -> str:
    """Capture a photo from device camera."""
    # Simulate camera access
    await asyncio.sleep(1)
    return "photo_12345.jpg"

Řešení problémů

Nástroje, které se nevolají

1. Ujistěte se, že se deklarace nástrojů odesílají na server.
2. Ověřte, že popisy nástrojů jasně uvádějí účel
3. Kontrola registrace nástroje v protokolech serveru

Chyby spuštění

1. Přidání komplexního zpracování chyb
2. Ověření parametrů před zpracováním
3. Vrácení uživatelsky přívětivých chybových zpráv
4. Chyby protokolu pro ladění

Problémy s typem

1. Použití Pydantických modelů pro komplexní typy
2. Převést modely na slovníky před serializací
3. Použijte explicitně převody typů

Další kroky

• Vykreslování back-endových nástrojů: Kombinování s nástroji na straně serveru

Další zdroje

• Přehled AG-UI
• Úvodní kurz
• Dokumentace k rozhraní Agent Framework