عرض أداة الواجهة الأمامية باستخدام AG-UI

يتم الإعلان عن أدوات الواجهة الأمامية وتنفيذها بواسطة عميل AG-UI. يتلقى الخادم مخططاته حتى يتمكن النموذج من طلبها، ولكنه لا يتلقى تطبيقاته.

تسجيل أداة الواجهة الأمامية

إنشاء الأداة وتمريرها إلى العامل المدعوم من قبل AGUIChatClient:

using System.ComponentModel;
using AGUI.Client;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

[Description("Get the user's current location from the client device.")]
static string GetUserLocation() => "Amsterdam, Netherlands";

AITool locationTool = AIFunctionFactory.Create(
    GetUserLocation,
    name: "get_user_location");

using HttpClient httpClient = new() { BaseAddress = new Uri("http://localhost:8888") };
AGUIChatClient chatClient = new(new AGUIChatClientOptions(httpClient, "/"));
AIAgent agent = chatClient.AsAIAgent(tools: [locationTool]);

AGUIChatClient يعالج تدفق المتابعة:

1. إرسال إعلان أداة الواجهة الأمامية مع طلب التشغيل.
2. يتلقى استدعاء أداة النموذج من الخادم.
3. تنفيذ الدالة المطابقة محليا.
4. يرسل النتيجة مرة أخرى إلى الخادم.
5. متابعة التشغيل وبث الاستجابة النهائية.

Tip

راجع نموذج أدوات الواجهة الأمامية .NET لعميل وخادم كاملين.

تحذير

إعلانات الأداة والنتائج التي يوفرها عميل غير موثوق به هي مدخلات غير موثوق بها. تخويل أدوات العميل التي قد تؤثر على تنفيذ العامل من جانب الخادم، والتحقق من صحة النتائج قبل استخدامها للعمليات المتميزة.

للحصول على إرشادات عامة حول تأليف الأدوات، راجع استخدام أدوات الدالة مع عامل.

الخطوات التالية

استخدام الموافقة البشرية مع AG-UI

يوضح لك هذا البرنامج التعليمي كيفية إضافة أدوات وظيفة الواجهة الأمامية إلى عملاء AG-UI. أدوات الواجهة الأمامية هي وظائف يتم تنفيذها من جانب العميل، ما يسمح لعامل الذكاء الاصطناعي بالتفاعل مع البيئة المحلية للمستخدم، أو الوصول إلى البيانات الخاصة بالعميل، أو تنفيذ عمليات واجهة المستخدم.

المتطلبات الأساسية

قبل البدء، تأكد من إكمال البرنامج التعليمي الشروع في العمل ولديك:

• Python 3.10 أو أحدث
• httpx مثبت لوظائف عميل HTTP
• الفهم الأساسي لإعداد عميل AG-UI
• Azure تم تكوين خدمة OpenAI

ما هي أدوات الواجهة الأمامية؟

أدوات الواجهة الأمامية هي أدوات وظيفية:

• يتم تعريفها وتسجيلها على العميل
• التنفيذ في بيئة العميل (وليس على الخادم)
• السماح لعامل الذكاء الاصطناعي بالتفاعل مع الموارد الخاصة بالعميل
• توفير النتائج مرة أخرى إلى الخادم للعامل لتضمينها في الاستجابات

حالات الاستخدام الشائعة:

• قراءة بيانات المستشعر المحلي
• الوصول إلى التخزين من جانب العميل أو التفضيلات
• تنفيذ عمليات واجهة المستخدم
• التفاعل مع الميزات الخاصة بالجهاز

إنشاء أدوات الواجهة الأمامية

يتم تعريف أدوات الواجهة الأمامية في 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 باستخدام Frontend Tools

فيما يلي تنفيذ كامل للعميل باستخدام أدوات الواجهة الأمامية:

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

                            # Tool calls arrive as TOOL_CALL_START/ARGS/END events
                            # and results are streamed back as TOOL_CALL_RESULT events.
                            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")

            # In current Python AG-UI, frontend tool declarations are sent with
            # the run request. Tool-call lifecycle events are streamed back over SSE.
            print(f"Tool result for {tool_call_id}: {result}")

        except Exception as e:
            print(f"\033[91m[Tool Error: {e}]\033[0m")
            print(f"Tool error for {tool_call_id}: {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. تزامن الخادم: يقرر عامل الذكاء الاصطناعي متى يتصل بأدوات الواجهة الأمامية بناء على طلب المستخدم
3. أحداث استدعاء الأداة: تدفقات TOOL_CALL_STARTTOOL_CALL_ARGSالخادم والأحداث TOOL_CALL_END إلى العميل
4. تنفيذ العميل: ينفذ العميل الأداة محليا
5. أحداث النتيجة: يتم تمثيل نتائج الأداة كأحداث TOOL_CALL_RESULT في الدفق
6. معالجة العامل: يدمج الخادم النتيجة ويستمر في الاستجابة

الأحداث الرئيسية

• TOOL_CALL_START / TOOL_CALL_ARGS / TOOL_CALL_END: طلبات الخادم وتيارات تفاصيل استدعاء الأدوات
• TOOL_CALL_RESULT: حدث نتيجة تنفيذ الأداة

الإخراج المتوقع

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. تحويل النماذج إلى قوالب قبل التسلسل
3. معالجة تحويلات النوع بشكل صريح

الخطوات التالية

• عرض أداة الواجهة الخلفية: الجمع مع الأدوات من جانب الخادم

موارد إضافية

• نظرة عامة علىAG-UI
• البرنامج التعليمي لبدء الاستخدام
• وثائق إطار عمل العامل

انتقل AG-UI يمكن للخوادم ترك استدعاءات الأداة للواجهة الأمامية عن طريق تعطيل استدعاء الوظيفة التلقائية على العامل المستضاف.

a := foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
    Instructions: "You are a helpful assistant.",
    Config: agent.Config{
        Name:                "AGUIAssistant",
        DisableFuncAutoCall: true,
    },
})

mux := http.NewServeMux()
mux.Handle("/", aguiprovider.NewJSONHTTPHandler(a, aguiprovider.HandlerConfig{}))

Tip

راجع نموذج أدوات الواجهة الأماميةAG-UI للحصول على مثال كامل قابل للتشغيل.