وكيل ذكاء اصطناعي هيكلي قائم على الرسم البياني مع جيميني: تنفيذ كامل بالشفرة

في هذا البرنامج التعليمي، سننفذ وكيل ذكاء اصطناعي متقدم قائم على الرسم البياني باستخدام إطار عمل GraphAgent ونموذج Gemini 1.5 Flash. سنُعرّف رسمًا بيانيًا موجهًا للعُقد، حيث يكون كل عقد مسؤولاً عن وظيفة محددة: مُخطط لتقسيم المهمة، ومُوجّه للتحكم في التدفق، وعقد بحث ورياضيات لتوفير أدلة وحسابات خارجية، وكاتِب لتركيب الإجابة، وناقد للتحقق من صحة الإخراج وتنقيحه.

دمج جيميني والأدوات المحلية

ندمج جيميني من خلال غلاف (Wrapper) يعالج مطالبات JSON مُهيكلة، بينما تعمل الدوال المحلية في Python كأدوات لتقييم الرياضيات الآمن والبحث في المستندات. من خلال تنفيذ هذا الخطّ الأنبوبي من البداية إلى النهاية، سنُبيّن كيف يتم تجميع الاستدلال، والاسترجاع، والتحقق من الصحة داخل نظام متماسك واحد.

استيراد المكتبات الأساسية

نبدأ باستيراد مكتبات Python الأساسية لمعالجة البيانات، والتوقيت، والتقييم الآمن، بالإضافة إلى dataclasses وtyping للمساعدة في هيكلة حالتنا. نقوم أيضًا بتحميل عميل google.generativeai للوصول إلى جيميني، واختياريًا، NetworkX لعرض الرسم البياني.

import os, json, time, ast, math, getpass
from dataclasses import dataclass, field
from typing import Dict, List, Callable, Any
import google.generativeai as genai
try:
    import networkx as nx
except ImportError:
    nx = None

تهيئة نموذج جيميني ودالة الاتصال

نعرف مُساعدًا لتهيئة نموذج جيميني وإعادته مع تعليمات نظام مخصصة، ودالة أخرى تستدعي LLM مع مطالبة مع التحكم في درجة الحرارة. نستخدم هذا الإعداد لضمان حصول وكيلنا على نتائج مُهيكلة ومُوجزة باستمرار.

def make_model(api_key: str, model_name: str = "gemini-1.5-flash"):
    genai.configure(api_key=api_key)
    return genai.GenerativeModel(model_name, system_instruction=(
        "You are GraphAgent, a principled planner-executor. "
        "Prefer structured, concise outputs; use provided tools when asked."
    ))

def call_llm(model, prompt: str, temperature=0.2) -> str:
    r = model.generate_content(prompt, generation_config={"temperature": temperature})
    return (r.text or "").strip()

أدوات البحث والحساب الآمن

نقوم بتنفيذ أداة حساب آمنة تحلّل وتتحقق من التعبيرات الحسابية باستخدام ast قبل التنفيذ، وأداة بحث بسيطة تسترجع المقاطع الأكثر صلة من مجموعة بيانات صغيرة في الذاكرة. نستخدم هذه الأدوات لمنح الوكيل قدرات حسابية واسترجاعية موثوقة دون تبعيات خارجية.

def safe_eval_math(expr: str) -> str:
    # ... (code remains the same) ...

DOCS = [
    "Solar panels convert sunlight to electricity; capacity factor ~20%.",
    "Wind turbines harvest kinetic energy; onshore capacity factor ~35%.",
    "RAG = retrieval-augmented generation joins search with prompting.",
    "LangGraph enables cyclic graphs of agents; good for tool orchestration.",
]

def search_docs(q: str, k: int = 3) -> List[str]:
    # ... (code remains the same) ...

عُقد الرسم البياني ووظائفها

نعرف dataclass من النوع State للحفاظ على المهمة، والخطة، والأدلة، والملاحظات الأولية، وأعلام التحكم أثناء تنفيذ الرسم البياني. نقوم بتنفيذ وظائف العقد: مُخطط، ومُوجّه، وبحث، ورياضيات، وكاتِب، وناقد. تُغيّر هذه الوظائف الحالة وتُعيد تسمية العقدة التالية. ثم نقوم بتسجيلها في NODES والتكرار في run_graph حتى الانتهاء.

عقدة التخطيط (node_plan)

تُنشئ هذه العقدة خطة فرعية للمهمة.

@dataclass
class State:
    # ... (code remains the same) ...

def node_plan(state: State, model) -> str:
    # ... (code remains the same) ...

عقدة التوجيه (node_route)

تقوم هذه العقدة بتحديد العقدة التالية بناءً على سياق المهمة.

def node_route(state: State, model) -> str:
    # ... (code remains the same) ...

عُقد البحث، الرياضيات، الكتابة، والنقاش

نفس الشيء ينطبق على باقي العقد.

def node_research(state: State, model) -> str:
    # ... (code remains the same) ...

def node_math(state: State, model) -> str:
    # ... (code remains the same) ...

def node_write(state: State, model) -> str:
    # ... (code remains the same) ...

def node_critic(state: State, model) -> str:
    # ... (code remains the same) ...

وظيفة تشغيل الرسم البياني (run_graph)

تدير هذه الوظيفة تدفق التنفيذ بين العقد.

NODES: Dict[str, Callable[[State, Any], str]] = {
    "plan": node_plan, "route": node_route, "research": node_research, "math": node_math, "write": node_write, "critic": node_critic
}

def run_graph(task: str, api_key: str) -> State:
    # ... (code remains the same) ...

عرض الرسم البياني

def ascii_graph():
    return """
    START -> plan -> route -> (research <-> route) & (math <-> route) -> write -> critic -> END
    """

نقطة دخول البرنامج

نقوم بقراءة مفتاح API الخاص بـ Gemini بأمان، ونأخذ مهمة كمدخل، ثم نقوم بتشغيل الرسم البياني من خلال run_graph. نقيس وقت التنفيذ، ونطبع الرسم البياني ASCII لسير العمل، ونعرض النتيجة النهائية، كما نخرج الأدلة الداعمة وأخر بضع ملاحظات أولية من أجل الشفافية.

if __name__ == "__main__":
    # ... (code remains the same) ...

الخلاصة

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

المصدر: MarkTechPost