بناء وكيل ذكاء اصطناعي للتفكير الهرمي مستوحى من الدماغ باستخدام نماذج Hugging Face

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

تثبيت المكتبات وتحميل النموذج

نبدأ بتثبيت المكتبات المطلوبة وتحميل نموذج Qwen2.5-1.5B-Instruct من Hugging Face. نقوم بتعيين نوع البيانات بناءً على توفر وحدة معالجة الرسومات لضمان تنفيذ النموذج بكفاءة:

pip -q install -U transformers accelerate bitsandbytes rich

ثم نقوم بتحميل المُعالج (Tokenizer) والنموذج، وتكوينه للعمل بكفاءة عالية، ولف كل شيء في خط أنابيب توليد النصوص:

import os, re, json, textwrap, traceback
from typing import Dict, Any, List
from rich import print as rprint
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline

MODEL_NAME = "Qwen/Qwen2.5-1.5B-Instruct"
DTYPE = torch.bfloat16 if torch.cuda.is_available() else torch.float32

tok = AutoTokenizer.from_pretrained(MODEL_NAME, use_fast=True)
model = AutoModelForCausalLM.from_pretrained(MODEL_NAME, device_map="auto", torch_dtype=DTYPE, load_in_4bit=True)
gen = pipeline("text-generation", model=model, tokenizer=tok, return_full_text=False)

تعريف دوال مساعدة

نقوم بتعريف دوال مساعدة: دالة chat تسمح لنا بإرسال مطالبات إلى النموذج مع تعليمات نظام اختيارية وضوابط أخذ العينات، بينما تساعدنا دالة extract_json على تحليل مخرجات JSON المنظمة من النموذج بشكل موثوق، حتى لو كانت الاستجابة تتضمن أسوارًا رمزية أو نصًا إضافيًا:

def chat(prompt: str, system: str = "", max_new_tokens: int = 512, temperature: float = 0.3) -> str:
    # ... (code remains the same)

def extract_json(txt: str) -> Dict[str, Any]:
    # ... (code remains the same)

استخراج الكود وتنفيذه

نقوم بتعريف دالتين مهمتين: دالة extract_code تستخرج أجزاء البايثون من مخرجات النموذج، ودالة run_python تنفذ هذه الأجزاء بأمان وتلتقط نتائجها:

def extract_code(txt: str) -> str:
    # ... (code remains the same)

def run_python(code: str, env: Dict[str, Any] | None = None) -> Dict[str, Any]:
    # ... (code remains the same)

تعليمات نظامية

نضيف تعليمات نظامية مهمة توجه النموذج لتقسيم المهام إلى أهداف فرعية، وحلها باستخدام الكود، والتحقق من الصحة، وإنتاج إجابة نهائية واضحة:

PLANNER_SYS = """You are the HRM Planner. Decompose the TASK into 2–4 atomic, code-solvable subgoals. Return compact JSON only: {"subgoals":[...], "final_format":"<one-line answer format>"}."""
SOLVER_SYS = """You are the HRM Solver. Given SUBGOAL and CONTEXT vars, output a single Python snippet. Rules: - Compute deterministically. - Set a variable RESULT to the answer. - Keep code short; stdlib only. Return only a Python code block."""
CRITIC_SYS = """You are the HRM Critic. Given TASK and LOGS (subgoal results), decide if final answer is ready. Return JSON only: {"action":"submit"|"revise","critique":"...", "fix_hint":"<if revise>"}."""
SYNTH_SYS = """You are the HRM Synthesizer. Given TASK, LOGS, and final_format, output only the final answer (no steps). Follow final_format exactly."""

تنفيذ حلقة HRM

نقوم بتنفيذ حلقة HRM الكاملة: نقوم بالتخطيط للأهداف الفرعية، وحل كل منها من خلال توليد وتشغيل بايثون (التقاط RESULT)، ثم نقوم بالانتقاد، وتنقية الخطة اختيارياً، وتوليف إجابة نهائية واضحة. نقوم بتنسيق هذه الجولات في hrm_agent، ونحمل النتائج الوسيطة كسياق حتى نقوم بتحسينها بشكل متكرر ونتوقف بمجرد أن يقول الناقد “إرسال”:

def plan(task: str) -> Dict[str, Any]:
    # ... (code remains the same)

def solve_subgoal(subgoal: str, context: Dict[str, Any]) -> Dict[str, Any]:
    # ... (code remains the same)

def critic(task: str, logs: List[Dict[str, Any]]) -> Dict[str, Any]:
    # ... (code remains the same)

def refine(task: str, logs: List[Dict[str, Any]]) -> Dict[str, Any]:
    # ... (code remains the same)

def synthesize(task: str, logs: List[Dict[str, Any]], final_format: str) -> str:
    # ... (code remains the same)

def hrm_agent(task: str, context: Dict[str, Any] | None = None, budget: int = 2) -> Dict[str, Any]:
    # ... (code remains the same)

تجربة الوكيل

نقوم بتشغيل عرضين توضيحيين للتحقق من صحة الوكيل: مهمة على طراز ARC حيث نستنتج تحويلًا من أزواج التدريب ونطبقه على شبكة اختبار، ومشكلة حسابية كلامية تتحقق من المنطق العددي. نستدعي hrm_agent مع كل مهمة، ونطبع الإجابات النهائية، ونعرض أيضًا عدد جولات الاستنتاج التي تستغرقها عملية تشغيل ARC:

ARC_TASK = textwrap.dedent(""" Infer the transformation rule from train examples and apply to test. Return exactly: "Answer: <grid>", where <grid> is a Python list of lists of ints. """).strip()
ARC_DATA = { "train": [ {"inp": [[0,0],[1,0]], "out": [[1,1],[0,1]]}, {"inp": [[0,1],[0,0]], "out": [[1,0],[1,1]]} ], "test": [[0,0],[0,1]] }
res1 = hrm_agent(ARC_TASK, context={"TRAIN": ARC_DATA["train"], "TEST": ARC_DATA["test"]}, budget=2)
rprint("n[bold]Demo 1 — ARC-like Toy[/bold]")
rprint(res1["final"])

WM_TASK = "A tank holds 1200 L. It leaks 2% per hour for 3 hours, then is refilled by 150 L. Return exactly: 'Answer: <liters>'."
res2 = hrm_agent(WM_TASK, context={}, budget=2)
rprint("n[bold]Demo 2 — Word Math[/bold]")
rprint(res2["final"])
rprint("n[dim]Rounds executed (Demo 1):[/dim]", len(res1["trace"]))

الخاتمة

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

المصدر: MarkTechPost