다음을 통해 공유


AG-UI 사용하여 프런트 엔드 도구 렌더링

이 자습서에서는 AG-UI 클라이언트에 프런트 엔드 함수 도구를 추가하는 방법을 보여 줍니다. 프런트 엔드 도구는 클라이언트 쪽에서 실행되는 함수로, AI 에이전트가 사용자의 로컬 환경과 상호 작용하거나 클라이언트별 데이터에 액세스하거나 UI 작업을 수행할 수 있도록 합니다. 서버는 이러한 도구를 호출할 시기를 오케스트레이션하지만 실행은 전적으로 클라이언트에서 수행됩니다.

필수 조건

시작하기 전에 시작 자습서를 완료하고 다음을 수행했는지 확인합니다.

  • .NET 8.0 이상
  • Microsoft.Agents.AI.AGUI 패키지 설치됨
  • Microsoft.Agents.AI 패키지 설치됨
  • AG-UI 클라이언트 설정에 대한 기본 이해

프런트 엔드 도구란?

프런트 엔드 도구는 다음과 같은 함수 도구입니다.

  • 클라이언트에 정의 및 등록됨
  • 서버가 아닌 클라이언트의 환경에서 실행
  • AI 에이전트가 클라이언트별 리소스와 상호 작용하도록 허용
  • 에이전트가 응답에 통합할 수 있도록 결과를 서버에 다시 제공합니다.
  • 개인화된 컨텍스트 인식 경험을 활성화하십시오

일반적인 사용 사례:

  • 로컬 센서 데이터 읽기(GPS, 온도 등)
  • 클라이언트 쪽 스토리지 또는 기본 설정에 액세스
  • UI 작업 수행(테마 변경, 알림 표시)
  • 디바이스별 기능과 상호 작용(카메라, 마이크)

클라이언트에 프런트 엔드 도구 등록

시작 자습서의 주요 차이점은 클라이언트 에이전트에 도구를 등록하는 것입니다. 변경 내용은 다음과 같습니다.

// 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);

나머지 클라이언트 코드는 시작 자습서와 동일하게 유지됩니다.

도구가 서버로 전송되는 방법

도구를 AsAIAgent()에 등록하면, AGUIChatClient가 자동으로 다음을 수행합니다.

  1. 도구 정의(이름, 설명, 매개 변수 스키마)를 캡처합니다.
  2. 각 요청과 함께 도구를 서버 에이전트에 보내서 이를 매핑합니다. ChatAgentRunOptions.ChatOptions.Tools

서버는 클라이언트 도구 선언을 수신하고 AI 모델은 호출 시기를 결정할 수 있습니다.

미들웨어를 사용하여 도구 검사 및 수정

에이전트 미들웨어를 사용하여 도구 액세스를 포함하여 에이전트 실행을 검사하거나 수정할 수 있습니다.

// 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;
    }
}

이 미들웨어 패턴을 사용하면 다음을 수행할 수 있습니다.

  • 실행하기 전에 도구 정의 유효성 검사

주요 개념

다음은 프런트 엔드 도구의 새로운 개념입니다.

  • 클라이언트 측 등록: 도구가 AIFunctionFactory.Create() 클라이언트에 등록되며 AsAIAgent()로 전달됩니다.
  • 자동 캡처: 도구가 자동으로 캡처되어 ChatAgentRunOptions.ChatOptions.Tools을 통해 전송됩니다.

프런트 엔드 도구의 작동 방식

서버 측 처리 흐름

서버는 프런트 엔드 도구의 구현 세부 정보를 모릅니다. 그것은 단지 알고있다 :

  1. 도구 이름 및 설명(클라이언트 등록에서)
  2. 매개 변수 스키마
  3. 도구 실행을 요청하는 경우

AI 에이전트가 프런트 엔드 도구를 호출하기로 결정한 경우:

  1. 서버는 SSE를 통해 클라이언트에 도구 호출 요청을 보냅니다.
  2. 서버는 클라이언트가 도구를 실행하고 결과를 반환할 때까지 기다립니다.
  3. 서버는 결과를 에이전트의 컨텍스트에 통합합니다.
  4. 에이전트가 도구 결과를 사용하여 계속 처리

클라이언트 측 프로세스

클라이언트는 프런트 엔드 도구 실행을 처리합니다.

  1. FunctionCallContent 도구 호출 요청을 나타내는 서버에서 수신
  2. 도구 이름을 로컬로 등록된 함수와 일치
  3. 요청에서 매개 변수 역직렬화
  4. 함수를 로컬로 실행합니다.
  5. 결과 직렬화
  6. FunctionResultContent 서버로 다시 보내기
  7. 에이전트 응답을 계속 받습니다.

프런트 엔드 도구를 사용하는 예상 출력

에이전트가 프런트 엔드 도구를 호출하면 도구 호출이 표시되고 스트리밍 출력이 생성됩니다.

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.

프런트 엔드 도구에 대한 서버 설정

프런트 엔드 도구를 지원하기 위해 서버에 특별한 구성이 필요하지 않습니다. 시작 자습서에서 표준 AG-UI 서버를 사용합니다. 이 서버는 자동으로 다음을 수행합니다.

  • 클라이언트 연결 중에 프런트 엔드 도구 선언을 받습니다.
  • AI 에이전트에 필요한 경우 도구 실행을 요청합니다.
  • 클라이언트의 결과를 기다립니다.
  • 에이전트의 의사 결정에 결과 통합

다음 단계

프런트 엔드 도구를 이해했으므로 다음을 수행할 수 있습니다.

추가 리소스

이 자습서에서는 AG-UI 클라이언트에 프런트 엔드 함수 도구를 추가하는 방법을 보여 줍니다. 프런트 엔드 도구는 클라이언트 쪽에서 실행되는 함수로, AI 에이전트가 사용자의 로컬 환경과 상호 작용하거나 클라이언트별 데이터에 액세스하거나 UI 작업을 수행할 수 있도록 합니다.

필수 조건

시작하기 전에 시작 자습서를 완료하고 다음을 수행했는지 확인합니다.

  • Python 3.10 이상
  • httpx HTTP 클라이언트 기능을 위해 설치됨
  • AG-UI 클라이언트 설정에 대한 기본 이해
  • 구성된 Azure OpenAI 서비스

프런트 엔드 도구란?

프런트 엔드 도구는 다음과 같은 함수 도구입니다.

  • 클라이언트에 정의 및 등록됨
  • 서버가 아닌 클라이언트의 환경에서 실행
  • AI 에이전트가 클라이언트별 리소스와 상호 작용하도록 허용
  • 에이전트가 응답에 통합할 수 있도록 결과를 서버에 다시 제공합니다.

일반적인 사용 사례:

  • 로컬 센서 데이터 읽기
  • 클라이언트 쪽 스토리지 또는 기본 설정에 액세스
  • UI 작업 수행
  • 디바이스별 기능과 상호 작용

프런트 엔드 도구 만들기

Python의 프런트 엔드 도구는 백 엔드 도구와 유사하게 정의되지만 클라이언트에 등록됩니다.

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

프런트 엔드 도구를 사용하여 AG-UI 클라이언트 만들기

프런트 엔드 도구를 사용하는 전체 클라이언트 구현은 다음과 같습니다.

"""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())

프런트 엔드 도구의 작동 방식

프로토콜 흐름

  1. 클라이언트 등록: 클라이언트가 도구 선언(이름, 설명, 매개 변수)을 서버에 보냅니다.
  2. 서버 오케스트레이션: AI 에이전트가 사용자 요청에 따라 프런트 엔드 도구를 호출할 시기를 결정합니다.
  3. 도구 호출 요청: 서버가 SSE를 TOOL_CALL_REQUEST 통해 클라이언트에 이벤트를 보냅니다.
  4. 클라이언트 실행: 클라이언트가 로컬로 도구를 실행합니다.
  5. 결과 제출: 클라이언트가 POST 요청을 통해 결과를 서버로 다시 보냅니다.
  6. 에이전트 처리: 서버는 결과를 통합하고 응답을 계속합니다.

주요 이벤트

  • TOOL_CALL_REQUEST: 서버에서 프런트 엔드 도구 실행을 요청합니다.
  • TOOL_CALL_RESULT: 클라이언트가 실행 결과를 제출합니다(HTTP POST를 통해).

예상 출력

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]

서버 설정

시작 자습서의 표준 AG-UI 서버는 프런트 엔드 도구를 자동으로 지원합니다. 서버 쪽에서 변경이 필요하지 않습니다. 도구 오케스트레이션을 자동으로 처리합니다.

모범 사례

Security

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"

오류 처리

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

비동기 작업

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

Troubleshooting

도구가 호출되지 않음

  1. 도구 선언이 서버로 전송되는지 확인
  2. 도구 설명이 용도를 명확하게 나타내는지 확인
  3. 서버 로그에서 도구 등록 확인

실행 오류

  1. 포괄적인 오류 처리 추가
  2. 처리하기 전에 매개 변수 유효성 검사
  3. 사용자에게 친숙한 오류 메시지 반환
  4. 디버깅에 대한 로그 오류

타입 문제

  1. 복합 형식에 대해 Pydantic 모델 사용
  2. serialization 전에 모델을 딕셔너리로 변환
  3. 형식 변환을 명시적으로 처리

다음 단계

추가 리소스