بناء وكلاء الذكاء الاصطناعي خطوة بخطوة باستخدام إطار عمل Agent-Lightning من مايكروسوفت

دليل شامل لبناء وكلاء الذكاء الاصطناعي باستخدام إطار عمل Agent-Lightning من مايكروسوفت

يوضح هذا الدليل طريقة إعداد وكيل ذكاء اصطناعي متقدم باستخدام إطار عمل Agent-Lightning من مايكروسوفت. سنقوم بتشغيل كل شيء مباشرةً داخل Google Colab، مما يعني أنه يمكننا تجربة كل من مكونات الخادم والعميل في مكان واحد. من خلال تعريف وكيل أسئلة وأجوبة صغير، وربطه بخادم Agent-Lightning محلي، ثم تدريبه باستخدام العديد من مطالبات النظام، يمكننا ملاحظة كيف يدعم الإطار تحديثات الموارد، ووضع المهام في قائمة الانتظار، والتقييم التلقائي. يمكنك الاطلاع على الكود الكامل هنا [رابط للكود].

1. التثبيت والاستيراد وإعداد مفتاح API

نبدأ بتثبيت المكتبات المطلوبة واستيراد جميع الوحدات الأساسية التي نحتاجها لـ Agent-Lightning. نقوم أيضًا بإعداد مفتاح API الخاص بـ OpenAI بشكل آمن وتحديد النموذج الذي سنستخدمه في هذا البرنامج التعليمي.

pip -q install agentlightning openai nest_asyncio python-dotenv > /dev/null

import os, threading, time, asyncio, nest_asyncio, random
from getpass import getpass
from agentlightning.litagent import LitAgent
from agentlightning.trainer import Trainer
from agentlightning.server import AgentLightningServer
from agentlightning.types import PromptTemplate
import openai

if not os.getenv("OPENAI_API_KEY"):
    try:
        os.environ["OPENAI_API_KEY"] = getpass("أدخل مفتاح API الخاص بـ OpenAI (اتركه فارغًا إذا كنت تستخدم قاعدة بيانات محلية/وكيل): ") or ""
    except Exception:
        pass

MODEL = os.getenv("MODEL", "gpt-4o-mini")

2. تعريف وكيل الأسئلة والأجوبة (QAAgent)

نقوم بتعريف وكيل QAAgent بسيط عن طريق توسيع LitAgent، حيث نقوم بمعالجة كل عملية تدريب من خلال إرسال مطالبة المستخدم إلى نموذج اللغة الكبير (LLM)، وجمع الاستجابة، وتقييمها مقابل الإجابة الصحيحة. قمنا بتصميم دالة المكافأة للتحقق من الصحة، وتداخل الرموز، والإيجاز، مما يسمح للوكيل بالتعلم وإنتاج مخرجات موجزة ودقيقة.

class QAAgent(LitAgent):
    def training_rollout(self, task, rollout_id, resources):
        """مع إعطاء مهمة {'prompt':..., 'answer':...}, اطرح سؤالًا على LLM باستخدام مطالبة النظام التي يوفرها الخادم وأرجع مكافأة في [0,1]."""
        sys_prompt = resources["system_prompt"].template
        user = task["prompt"]; gold = task.get("answer","").strip().lower()
        try:
            r = openai.chat.completions.create(
                model=MODEL,
                messages=[{"role":"system","content":sys_prompt}, {"role":"user","content":user}],
                temperature=0.2,
            )
            pred = r.choices[0].message.content.strip()
        except Exception as e:
            pred = f"[error]{e}"

        def score(pred, gold):
            P = pred.lower()
            base = 1.0 if gold and gold in P else 0.0
            gt = set(gold.split()); pr = set(P.split()); inter = len(gt & pr); denom = (len(gt)+len(pr)) or 1
            overlap = 2*inter/denom
            brevity = 0.2 if base==1.0 and len(P.split())<=8 else 0.0
            return max(0.0, min(1.0, 0.7*base + 0.25*overlap + brevity))
        return float(score(pred, gold))

3. تحديد المهام ومطالبات النظام

نقوم بتعريف مقياس أداء صغير بثلاث مهام QA ونسرد العديد من مطالبات النظام المرشحة للتحسين. ثم نطبق nest_asyncio ونحدد مضيف الخادم المحلي ومنفذه، مما يسمح لنا بتشغيل خادم Agent-Lightning والعملاء ضمن بيئة تشغيل Colab واحدة.

TASKS = [
    {"prompt":"عاصمة فرنسا؟","answer":"باريس"},
    {"prompt":"من كتب كبرياء وتحامل؟","answer":"جين أوستن"},
    {"prompt":"2+2 = ?","answer":"4"},
]
PROMPTS = [
    "أنت خبير موجز. أجب فقط بالواقع النهائي، بدون جمل.",
    "أنت ذكاء اصطناعي مفيد، ومتعلم. فضّل الإجابات الدقيقة والموجزة.",
    "أجب كمحقق دقيق؛ أرجع فقط الحقيقة الأساسية.",
    "كن معلمًا ودودًا. أعط الإجابة بكلمة واحدة إذا كانت واضحة."
]
nest_asyncio.apply()
HOST, PORT = "127.0.0.1", 9997

4. تشغيل الخادم وتقييم المطالبات

نبدأ تشغيل خادم Agent-Lightning ونتجول في مطالبات النظام المرشحة، ونحدث system_prompt المشتركة قبل وضع كل مهمة تدريب في قائمة الانتظار. ثم نستطلع عمليات الانتشار المكتملة، ونحسب متوسط المكافآت لكل مطالبة، ونبلغ عن أفضل مطالبة أداءً، ونوقف الخادم بلطف.

async def run_server_and_search():
    server = AgentLightningServer(host=HOST, port=PORT)
    await server.start()
    print("تم بدء تشغيل الخادم")
    await asyncio.sleep(1.5)
    results = []
    for sp in PROMPTS:
        await server.update_resources({"system_prompt": PromptTemplate(template=sp, engine="f-string")})
        scores = []
        for t in TASKS:
            tid = await server.queue_task(sample=t, mode="train")
            rollout = await server.poll_completed_rollout(tid, timeout=40) # ينتظر عاملًا
            if rollout is None:
                print("انتهى وقت الانتظار لعملية الانتشار؛ المتابعة...")
                continue
            scores.append(float(getattr(rollout, "final_reward", 0.0)))
        avg = sum(scores)/len(scores) if scores else 0.0
        print(f"متوسط المطالبة: {avg:.3f} | {sp}")
        results.append((sp, avg))
    best = max(results, key=lambda x: x[1]) if results else ("<none>",0)
    print("nأفضل مطالبة:", best[0], " | النتيجة:", f"{best[1]:.3f}")
    await server.stop()

5. تشغيل العميل

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

def run_client_in_thread():
    agent = QAAgent()
    trainer = Trainer(n_workers=2)
    trainer.fit(agent, backend=f"http://{HOST}:{PORT}")

client_thr = threading.Thread(target=run_client_in_thread, daemon=True)
client_thr.start()
asyncio.run(run_server_and_search())

في الختام، سنرى كيف يسمح لنا Agent-Lightning بإنشاء خط أنابيب وكيل مرن ببضع أسطر من التعليمات البرمجية فقط. يمكننا بدء تشغيل خادم، وتشغيل عمال عملاء متوازيين، وتقييم مطالبات نظام مختلفة، وقياس الأداء تلقائيًا، كل ذلك ضمن بيئة Colab واحدة. هذا يوضح كيف يبسط الإطار عملية بناء واختبار وتحسين وكلاء الذكاء الاصطناعي بطريقة منظمة. يمكنك الاطلاع على الكود الكامل هنا [رابط للكود]. تابعونا على [رابط تويتر] و [رابط ريديت] واشترك في [رابط النشرة البريدية].

المصدر: MarkTechPost