بناء وكلاء ذكاء اصطناعي قائم على الأحداث باستخدام UAgents و جوجل Gemini: دليل تنفيذ شامل بلغة بايثون

بناء وكلاء ذكاء اصطناعي قائم على الأحداث باستخدام UAgents وجوجل Gemini: دليل شامل بلغة بايثون

يُظهر هذا البرنامج التعليمي كيفية استخدام إطار عمل UAgents لبناء بنية وكيل ذكاء اصطناعي خفيف الوزن وقائم على الأحداث فوق واجهة برمجة تطبيقات جوجل Gemini. سنبدأ بتطبيق nest_asyncio لتمكين حلقات الأحداث المتداخلة، ثم نُهيئ مفتاح واجهة برمجة تطبيقات Gemini الخاص بك وننشئ عميل GenAI. بعد ذلك، سنُعرف عقود الاتصال الخاصة بنا، ونماذج Pydantic للأسئلة والأجوبة، ونُنشئ وكيلين UAgents: وكيل “gemini_agent” الذي يستمع إلى رسائل الأسئلة الواردة، ويستدعي نموذج Gemini “flash” لإنشاء استجابات، ويُصدر رسائل الإجابة؛ ووكيل “client_agent” الذي يُطلق استعلامًا عند بدء التشغيل ويعالج الإجابة الواردة. أخيرًا، سنتعلم كيفية تشغيل هذين الوكيلين بالتزامن باستخدام أداة معالجة متعددة العمليات في بايثون، وإغلاق حلقة الأحداث بشكل صحيح بمجرد اكتمال التبادل، مما يُوضح سهولة تنظيم UAgents لرسائل التبادل بين الوكلاء.

1. التثبيت والإعداد

سنبدأ بتثبيت إطار عمل UAgents ومكتبة عميل Google GenAI، مما يوفر الأدوات اللازمة لبناء وتشغيل وكلاء الذكاء الاصطناعي القائم على الأحداث مع Gemini:

pip install -q uagents google-genai

علم q يُشغّل التثبيت بهدوء، مما يحافظ على نظافة مخرجات دفتر الملاحظات.

بعد ذلك، سنقوم بإعداد بيئة بايثون الخاصة بنا من خلال استيراد الوحدات الأساسية، وأدوات النظام (os، time، multiprocessing، asyncio)، و nest_asyncio لتمكين حلقات الأحداث المتداخلة (ضروري في دفاتر الملاحظات)، وعميل Google GenAI، و Pydantic للتحقق من صحة المخططات، وفئات UAgents الأساسية. أخيرًا، nest_asyncio.apply() يُصلح حلقة الأحداث حتى تتمكن من تشغيل سير عمل UAgents غير المتزامن بسلاسة في البيئات التفاعلية:

import os, time, multiprocessing, asyncio
import nest_asyncio
from google import genai
from pydantic import BaseModel, Field
from uagents import Agent, Context

nest_asyncio.apply()

2. تهيئة مفتاح واجهة برمجة التطبيقات

نقوم بتعيين مفتاح واجهة برمجة تطبيقات Gemini الخاص بك في البيئة. تأكد من استبدال العنصر النائب بمفتاحك الفعلي، ثم قم بتهيئة عميل GenAI، والذي سيتعامل مع جميع الطلبات اللاحقة إلى نماذج Gemini من جوجل. تضمن هذه الخطوة حصول وكيلنا على إذن مُصادق عليه لإنشاء محتوى من خلال واجهة برمجة التطبيقات:

os.environ["GOOGLE_API_KEY"] = "Use Your Own API Key Here"
client = genai.Client()

3. تعريف نماذج الرسائل

سنُعرف نماذج Pydantic لتنسيق رسائل الأسئلة والأجوبة التي سيتبادلها وكلاؤنا:

class Question(BaseModel):
    question: str = Field(...)

class Answer(BaseModel):
    answer: str = Field(...)

يحتوي نموذج Question على حقل سلسلة نصية واحدة للسؤال، ويحتوي نموذج Answer على حقل سلسلة نصية واحدة للإجابة. باستخدام Pydantic، نحصل على التحقق التلقائي وتسلسل الرسائل الواردة والصادرة، مما يضمن أن كل وكيل يعمل دائمًا مع بيانات مُنسقة بشكل صحيح.

4. إنشاء وكيل Gemini

ننشئ وكيل UAgents “gemini_agent” مع اسم فريد، وعبارة البذرة (لهوية حتمية)، ومنفذ استماع، ونقطة نهاية HTTP لإرسال الرسائل. ثم نقوم بتسجيل مُعالِج حدث بدء التشغيل الذي يسجل عندما يكون الوكيل جاهزًا، مما يضمن الرؤية في دورة حياته. تقوم الدالة المساعدة ask_gemini المُزامنة بتغليف مكالمة عميل GenAI إلى نموذج Gemini “flash”. في الوقت نفسه، يقوم مُعالِج @ai_agent.on_message بفك تسلسل رسائل Question الواردة، ويستدعي ask_gemini، ويرسل بشكل غير متزامن حمولة Answer مُعتمدة إلى المُرسِل الأصلي:

ai_agent = Agent(
    name="gemini_agent",
    seed="agent_seed_phrase",
    port=8000,
    endpoint=["http://127.0.0.1:8000/submit"]
)

@ai_agent.on_event("startup")
async def ai_startup(ctx: Context):
    ctx.logger.info(f"{ai_agent.name} listening on {ai_agent.address}")

def ask_gemini(q: str) -> str:
    resp = client.models.generate_content(model="gemini-2.0-flash", contents=f"Answer the question: {q}")
    return resp.text

@ai_agent.on_message(model=Question, replies=Answer)
async def handle_question(ctx: Context, sender: str, msg: Question):
    ans = ask_gemini(msg.question)
    await ctx.send(sender, Answer(answer=ans))

5. إنشاء وكيل العميل

ننشئ وكيل “client_agent” الذي يرسل عند بدء التشغيل سؤالًا إلى gemini_agent يسأل عن عاصمة فرنسا، ثم يستمع لإجابة، ويطبع الاستجابة المُستلمة، ويُغلق حلقة الأحداث بشكل صحيح بعد تأخير قصير:

client_agent = Agent(
    name="client_agent",
    seed="client_seed_phrase",
    port=8001,
    endpoint=["http://127.0.0.1:8001/submit"]
)

@client_agent.on_event("startup")
async def ask_on_start(ctx: Context):
    await ctx.send(ai_agent.address, Question(question="What is the capital of France?"))

@client_agent.on_message(model=Answer)
async def handle_answer(ctx: Context, sender: str, msg: Answer):
    print("Answer from Gemini:", msg.answer)
    # Use a more graceful shutdown
    asyncio.create_task(shutdown_loop())

async def shutdown_loop():
    await asyncio.sleep(1)  # Give time for cleanup
    loop = asyncio.get_event_loop()
    loop.stop()

6. تشغيل الوكلاء

نُعرّف دالة مساعدة run_agent التي تستدعي agent.run(), ثم نستخدم معالجة متعددة العمليات في بايثون لإطلاق gemini_agent في عمليته الخاصة. بعد إعطائه لحظة للبدء، فإنه يُشغّل client_agent في العملية الرئيسية، ويحظر حتى اكتمال تبادل الإجابة، ثم ينضم إلى العملية الخلفية لضمان إغلاق نظيف:

def run_agent(agent):
    agent.run()

if __name__ == "__main__":
    p = multiprocessing.Process(target=run_agent, args=(ai_agent,))
    p.start()
    time.sleep(2)
    client_agent.run()
    p.join()

في الختام، مع هذا البرنامج التعليمي الذي يركز على UAgents، أصبح لدينا الآن مخططًا واضحًا لإنشاء خدمات ذكاء اصطناعي مُدمجة تتواصل عبر نقاط اتصال حدث ورسائل مُحددة جيدًا. لقد رأيت كيف يُبسط UAgents إدارة دورة حياة الوكيل، وتسجيل أحداث بدء التشغيل، ومعالجة الرسائل الواردة، وإرسال ردود مُنسقة، وكل ذلك بدون تعقيدات برمجة الشبكات. من هنا، يمكنك توسيع إعداد UAgents الخاص بك لتضمين سير عمل محادثة أكثر تطورًا، وأنواع رسائل متعددة، واكتشاف وكيل ديناميكي.

المصدر: MarkTechPost