بناء وكيل جوي بسيط باستخدام بروتوكول اتصال الوكلاء (ACP) مع بايثون

مقدمة إلى بروتوكول اتصال الوكلاء (ACP) وبناء وكيل للطقس باستخدام بايثون

يُعد بروتوكول اتصال الوكلاء (ACP) معيارًا مفتوحًا مصممًا لتمكين التواصل السلس بين وكلاء الذكاء الاصطناعي، والتطبيقات، والبشر. نظرًا لأن أنظمة الذكاء الاصطناعي غالبًا ما يتم تطويرها باستخدام أطر عمل وهياكل أساسية متنوعة، فقد تنتهي معزولة وغير متوافقة، مما يحد من قدرتها على التعاون. ويعالج ACP هذه التجزئة من خلال تقديم واجهة برمجة تطبيقات REST موحدة تسهل:

• الاتصال متعدد الوسائط: يدعم ACP طرقًا متعددة للتواصل، مما يسمح للوكلاء بالتفاعل باستخدام نصوص، صور، أصوات، وغيرها.
• الرسائل المتزامنة وغير المتزامنة: يمكن للوكلاء إرسال واستقبال الرسائل بشكل متزامن (انتظار الرد) أو غير متزامن (إرسال الرسالة والمضي قدمًا).
• البث في الوقت الفعلي: يدعم ACP البث في الوقت الفعلي للبيانات، مما يسمح بالتفاعلات الديناميكية.
• التفاعلات مع الوكلاء ذات الحالة والحالة عديمة الحالة: يمكن للوكلاء تخزين معلومات الحالة (مثل سجل المحادثات) أو العمل بدونها.
• اكتشاف الوكلاء: يسمح ACP باكتشاف الوكلاء المتصلين وغير المتصلين.
• تنفيذ المهام طويلة الأمد: يمكن للوكلاء التعامل مع المهام التي تتطلب وقتًا طويلًا للتنفيذ.

في هذا البرنامج التعليمي، سنبدأ خطواتنا الأولى مع ACP من خلال بناء خادم أساسي يوفر معلومات طقس لندن، وعميل بسيط يمكنه التفاعل معه.

إعداد التبعيات وتثبيت المكتبات

قبل البدء، نحتاج إلى تثبيت المكتبات اللازمة:

pip install acp acp-sdk beeai-framework httpx

إنشاء خادم ACP

سنبدأ بإعداد خادم ACP، بدءًا من إنشاء ملف agent.py. سنبدأ باستيراد المكتبات الضرورية. للحصول على بيانات طقس لندن، سنستخدم مكتبة httpx لإرسال طلب إلى واجهة برمجة التطبيقات Open-Meteo.

import asyncio
from collections.abc import AsyncGenerator
import httpx
from acp_sdk.models import Message, MessagePart
from acp_sdk.server import Context, RunYield, RunYieldResume, Server

server = Server()

دالة الحصول على طقس لندن

سنعرف دالة مساعدة غير متزامنة تسمى get_london_weather تسترد حالة الطقس الحالية في لندن باستخدام واجهة برمجة التطبيقات Open-Meteo. ترسل هذه الدالة طلبًا مع إحداثيات لندن وتعيد ملخصًا مُنسقًا للطقس يشمل درجة الحرارة، وسرعة الرياح، ورمز حالة الطقس.

async def get_london_weather() -> str:
    """استرجاع حالة طقس لندن الحالية من واجهة برمجة التطبيقات المجانية Open-Meteo."""
    params = {
        "latitude": 51.5072,  # إحداثيات لندن
        "longitude": -0.1276,
        "current_weather": True,
        "timezone": "Europe/London"
    }
    url = "https://api.open-meteo.com/v1/forecast"
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.get(url, params=params)
        resp.raise_for_status()
        cw = resp.json()["current_weather"]
        return (
            f"طقس لندن: {cw['temperature']} درجة مئوية، "
            f"الرياح {cw['windspeed']} كم/ساعة، الرمز {cw['weathercode']}."
        )

وكيل ACP

يُعرّف هذا الكود وكيلًا متوافقًا مع ACP باستخدام مُزيّن @server.agent(). تتعامل الدالة london_weather_agent مع الرسائل الواردة من خلال توليد رسالة تفكير أولاً، ثم استرجاع حالة الطقس الحالية في لندن بشكل غير متزامن باستخدام الدالة المساعدة get_london_weather(). ثم يتم إرجاع بيانات الطقس كرسالة نصية عادية. وأخيرًا، يبدأ server.run() خادم ACP ويجعل الوكيل متاحًا للتعامل مع الطلبات.

@server.agent()
async def london_weather_agent(input: list[Message], context: Context) -> AsyncGenerator[RunYield, RunYieldResume]:
    """إرجاع حالة طقس لندن الحالية."""
    for _ in input:
        yield {"thought": "جاري جلب طقس لندن..."}
    weather = await get_london_weather()
    yield Message(role="agent", parts=[MessagePart(content=weather, content_type="text/plain")])

server.run()

تشغيل الخادم

بعد ذلك، سنشغل ملف agent.py لبدء تشغيل الخادم. بمجرد التشغيل، سيكون وكيل ACP متاحًا للتعامل مع الطلبات على http://localhost:8000.

python agent.py

للتحقق من أن وكيلك قيد التشغيل، افتح محطة جديدة وقم بتنفيذ الأمر curl التالي:

curl http://localhost:8000/agents

إذا كان كل شيء يعمل بشكل صحيح، فستتلقى استجابة JSON تُدرج وكيلك، مما يؤكد أنه متاح وجاهز للتعامل مع الطلبات.

إنشاء عميل ACP

سنقوم الآن بإنشاء عميل ACP (client.py) للتفاعل مع خادمنا. يستخدم برنامج نصوص العميل هذا مجموعة أدوات ACP للاتصال بـ london_weather_agent الذي يعمل محليًا عبر خادم ACP على http://localhost:8000. يُرسل رسالة متزامنة يطلب فيها حالة الطقس باستخدام طريقة run_sync. بمجرد استجابة الوكيل، يطبع البرنامج النصي تفاصيل حالة الطقس التي تم إرجاعها.

import asyncio
from acp_sdk.client import Client
from acp_sdk.models import Message, MessagePart

async def call_london_weather_agent() -> None:
    async with Client(base_url="http://localhost:8000") as client:
        run = await client.run_sync(
            agent="london_weather_agent",
            input=[
                Message(
                    parts=[MessagePart(content="أخبرني عن حالة الطقس", content_type="text/plain")]
                )
            ],
        )
        print("استجابة من london_weather_agent:")
        for message in run.output:
            for part in message.parts:
                print("-", part.content)

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

تشغيل العميل

في محطة أخرى، قم بتشغيل الأمر التالي لإرسال طلب إلى خادم ACP الخاص بنا:

python client.py

يجب أن ترى استجابة من الخادم تحتوي على حالة الطقس الحالية في لندن، التي تم إرجاعها بواسطة london_weather_agent.

(ملاحظة: تمت إزالة الجمل الأخيرة من المقال الأصلي لأنها ترويجية وليست ذات صلة بالمحتوى التقني.)

المصدر: MarkTechPost