بايسبور: أداة بناء وكيل الذكاء الاصطناعي مفتوحة المصدر

ما هو Pyspur؟

Pyspur هو منصة مفتوحة المصدر مصممة لتسريع تطوير وكلاء الذكاء الاصطناعي من خلال توفير بيئة مرئية قائمة على العقد (nodes). تمكن المهندسين من بناء وتصحيح الأخطاء ونشر سير عمل الذكاء الاصطناعي المعقدة عن طريق ربط المكونات المعيارية على لوحة رسم قابلة للسحب والإفلات.

Drag & Drop to build AI Agents.

And deploy to API in one click with Nocode.

100% Opensource. pic.twitter.com/iPEJEzIHVI

— Shubham Saboo (@Saboo_Shubham_) December 19, 2024

المشكلة الأساسية التي يحلها Pyspur هي نقص الشفافية ودورة التكرار البطيئة الشائعة في تطوير الذكاء الاصطناعي. يعالج مشكلتي "جحيم المطالبات" و"نقاط العمى في سير العمل" من خلال السماح للمطورين بفحص مدخلات ومخرجات كل خطوة في منطق وكيلهم في الوقت الفعلي. تتضمن المنصة دعمًا مدمجًا لأنماط متقدمة مثل التوليد المعزز بالاسترجاع (RAG)، وتسمح بنقاط توقف "الإنسان في الحلقة" (human-in-the-loop)، ويمكنها نشر أي سير عمل كواجهة برمجة تطبيقات (API) جاهزة للإنتاج بنقرة واحدة. في النهاية، يساعد Pyspur المهندسين على بناء أنظمة ذكاء اصطناعي أكثر موثوقية وقابلة للتصحيح بشكل أسرع.

We’re thrilled to share that MCP support is coming soon to Apidog! 🚀

Apidog MCP Server lets you feed API docs directly to Agentic AI, supercharging your vibe coding experience! Whether you're using Cursor, Cline, or Windsurf - it'll make your dev process faster and smoother.… pic.twitter.com/ew8U38mU0K

— Apidog (@ApidogHQ) March 19, 2025

هيا بنا نبدأ!

1. إعداد البيئة

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

الخيار أ: التثبيت المحلي باستخدام pip

المتطلبات الأساسية: Python 3.11+

التثبيت من PyPI:

pip install pyspur

تهيئة مجلد المشروع: يقوم هذا الأمر بإنشاء هيكل مشروع، بما في ذلك ملف .env للتكوين.

pyspur init my-pyspur-project && cd my-pyspur-project

تشغيل الخادم: توجه العلامة --sqlite Pyspur لاستخدام قاعدة بيانات SQLite محلية، مما يزيل الاعتماد على خادم قاعدة بيانات منفصل.

pyspur serve --sqlite

الوصول إلى واجهة المستخدم: انتقل إلى http://localhost:6080 في متصفحك.

الخيار ب: إعداد Docker

المتطلبات الأساسية: محرك Docker

تنفيذ سكريبت الإعداد: يقوم هذا الأمر بتنزيل وتشغيل سكريبت شل يقوم باستنساخ مستودع Pyspur، وتكوين ملف docker-compose.dev.yml، وتشغيل مكدس التطبيق (الواجهة الأمامية، الواجهة الخلفية، قاعدة البيانات).

curl -fsSL https://raw.githubusercontent.com/PySpur-com/pyspur/main/start_pyspur_docker.sh | bash -s pyspur-project

الوصول إلى واجهة المستخدم: انتقل إلى http://localhost:6080.

2. هيا بنا نبني سير عمل باستخدام Pyspur

بدلاً من البناء من الصفر، سنقوم بتحميل وتحليل قالب Pyspur موجود. يوفر هذا النهج نظرة واقعية على سير عمل غير تافه.

تحميل القالب:

• في لوحة تحكم Pyspur، انقر على "New Spur".
• في النافذة المنبثقة التي تظهر، حدد علامة التبويب "Templates".
• اختر قالب "Joke Generator". سيتم ملء لوحة الرسم بسير عمل مبني مسبقًا.

تحليل سير العمل:
تم تصميم سير العمل هذا لتوليد نكتة ثم تحسينها. يستخدم عقدة BestOfNNode، وهي مكون متقدم يقوم بتشغيل مطالبة LLM عدد N من المرات، ويستخدم استدعاء LLM آخر لتقييم المخرجات N، ويختار الأفضل منها.

دعونا نفصل العقد الرئيسية كما هي محددة في joke_generator.json:

input_node (InputNode): تحدد هذه العقدة نقطة دخول سير العمل.

• config.output_schema: { "topic": "string", "audience": "string" }
• يحدد هذا أن سير العمل يتطلب مدخلين نصيين (string): topic النكتة وaudience المستهدف.

JokeDrafter (BestOfNNode): هذه هي المرحلة الأولى من إنشاء النكتة.

• config.system_message: "You are a stand-up comedian who uses dark humor like Ricky Gervais or Jimmy Carr..."
• config.user_message: "Your audience is: {{input_node.audience}}\nThe topic should be about {{input_node.topic}}"
• يقوم قالب Jinja2 هذا بإدراج البيانات ديناميكيًا من input_node.
• config.samples: 10
• يوجه هذا العقدة لتوليد 10 مسودات نكت.
• config.rating_prompt: "Rate the following joke on a scale from 0 to 10..."
• بعد توليد 10 نكت، يتم استخدام هذه المطالبة لجعل LLM يقيم كل واحدة. ثم تختار العقدة النكتة الأعلى تقييمًا.
• config.output_schema: { "initial_joke": "string" }
• مخرج العقدة هو نص واحد (string): أفضل نكتة من العينات العشر.

JokeRefiner (BestOfNNode): تأخذ هذه العقدة النكتة المسودة وتحسنها.

• config.system_message: "Your goal is to refine a joke to make it more vulgar and concise..."
• config.user_message: {{JokeDrafter.initial_joke}}
• الأهم من ذلك، مدخل هذه العقدة هو مخرج عقدة JokeDrafter.
• config.samples: 3
• تولد 3 نسخ محسنة من النكتة الأولية.
• تستخدم أيضًا مطالبة تقييم لاختيار الأفضل من التحسينات الثلاثة.
• config.output_schema: { "final_joke": "string" }

الروابط (Links): تحدد مصفوفة links في JSON تدفق البيانات:
input_node -> JokeDrafter -> JokeRefiner.

التنفيذ والفحص:

• في لوحة الاختبار على اليمين، استخدم المدخلات الاختبارية المعبأة مسبقًا أو أنشئ مدخلاتك الخاصة (على سبيل المثال، الموضوع: "مساعدو الذكاء الاصطناعي"، الجمهور: "المطورون الكبار").
• انقر على Run.
• أثناء تنفيذ سير العمل، انقر على عقدة JokeDrafter. يمكنك فحص run_data الخاصة بها، والتي ستعرض جميع العينات العشر التي تم توليدها وتقييماتها المقابلة، مما يمنحك رؤية شفافة لـ "عملية تفكير" الوكيل.
• ستكون النكتة النهائية المحسنة متاحة في مخرج عقدة JokeRefiner.

3. تطبيق مسار RAG

على الرغم من أنها ليست جزءًا من مولد النكت، إلا أن التوليد المعزز بالاسترجاع (Retrieval-Augmented Generation - RAG) هو قدرة حاسمة في Pyspur. فيما يلي العملية الفنية لإضافة المعرفة إلى وكيل:

1. استيعاب المستندات (Collection): انتقل إلى قسم RAG. عند إنشاء "مجموعة مستندات" (Document Collection) وتحميل ملف (مثل PDF)، يبدأ Pyspur عملية خلفية تقوم بتحليل المستند إلى نص، وتقسيمه إلى أجزاء قابلة للتكوين بناءً على طول الرمز (token length)، وتخزين هذه الأجزاء مع بيانات تعريف المصدر في قاعدة بياناته.
2. تحويل إلى متجه (Index): يؤدي إنشاء "فهرس متجه" (Vector Index) من مجموعة إلى تشغيل عملية أخرى. يقوم Pyspur بالمرور عبر كل جزء نصي، ويقوم باستدعاء واجهة برمجة تطبيقات لنموذج تضمين محدد (مثل text-embedding-ada-002 من OpenAI) للحصول على تمثيل متجه، وإدراج (upsert) هذه المتجهات في قاعدة بيانات متجهات مكونة (مثل ChromaDB، PGVector).
3. الاسترجاع (Workflow Node): في سير العمل، يتم تكوين عقدة الاسترجاع (Retriever Node) للإشارة إلى فهرس متجه محدد. في وقت التشغيل، يتم تضمين مدخلها query باستخدام نفس النموذج، ويتم إجراء بحث دلالي (أقرب جار تقريبي) مقابل قاعدة بيانات المتجهات لجلب أجزاء النص الأكثر صلة. ثم يتم تمرير هذه الأجزاء كسياق إلى LLM لاحق.

4. النشر كواجهة برمجة تطبيقات (API) جاهزة للإنتاج

عند الانتهاء من سير العمل الخاص بك، يمكنك كشفه كنقطة نهاية HTTP آمنة.

بدء النشر: انقر على زر "Deploy" في شريط التنقل العلوي.

تحديد نوع استدعاء API:

• Blocking (متزامن): لسير العمل سريع التنفيذ. يتلقى العميل النتيجة في نفس استجابة HTTP.
• نقطة النهاية (Endpoint): POST /api/wf/{workflow_id}/run/?run_type=blocking
• Non-Blocking (غير متزامن): لسير العمل طويل الأمد (مثل مولد النكت لدينا مع 10+ استدعاءات LLM). يتلقى العميل فورًا run_id ويجب عليه استقصاء نقطة نهاية منفصلة للحصول على النتيجة. هذا يمنع انتهاء مهلة العميل.
• نقطة نهاية البدء (Start Endpoint): POST /api/wf/{workflow_id}/start_run/?run_type=non_blocking
• نقطة نهاية الحالة (Status Endpoint): GET /api/runs/{run_id}/status/
• التكامل مع تطبيقك:
  تقوم النافذة المنبثقة للنشر بتوليد كود العميل. يجب أن يكون جسم طلب POST كائن JSON حيث يحتوي المفتاح initial_inputs على كائن تتطابق مفاتيحه مع title عقد الإدخال الخاصة بك.

مثال على عميل Python (لمولد النكت):

import requests
import json
import time

PYSUR_HOST = "http://localhost:6080"
WORKFLOW_ID = "your_workflow_id_here" # Get this from the deploy modal

# The keys in this dict must match the 'output_schema' of the input_node
payload = {
    "initial_inputs": {
        "input_node": {
            "topic": "Python decorators",
            "audience": "Senior Software Engineers"
        }
    }
}

# 1. Start the non-blocking run
start_url = f"{PYSUR_HOST}/api/wf/{WORKFLOW_ID}/start_run/?run_type=non_blocking"
start_resp = requests.post(start_url, json=payload)
run_id = start_resp.json()['id']
print(f"Workflow started with run_id: {run_id}")

# 2. Poll for the result
status_url = f"{PYSUR_HOST}/api/runs/{run_id}/status/"
while True:
    status_resp = requests.get(status_url)
    data = status_resp.json()
    status = data.get("status")
    print(f"Current status: {status}")
    if status in ["COMPLETED", "FAILED"]:
        print("Final Output:")
        print(json.dumps(data.get("outputs"), indent=2))
        break
    time.sleep(2)

الخلاصة

يوفر Pyspur بيئة قوية وشفافة ومبنية على أسس تقنية لتطوير وكلاء الذكاء الاصطناعي. من خلال تجريد العمليات الفردية إلى عقد معيارية وتوفير لوحة رسم مرئية، فإنه يسمح للمهندسين بالتركيز على المنطق عالي المستوى لوكلائهم. تكمن القوة الحقيقية للمنصة في قابليتها للفحص العميق - مما يسمح لك بتتبع تدفق البيانات وتصحيح مدخلات/مخرجات كل مكون - وانتقالها السلس من النموذج الأولي المرئي إلى واجهة برمجة تطبيقات جاهزة للإنتاج. الانتقال من وكلاء بسيطين ذوي استدعاء واحد إلى سير عمل معقد متعدد المراحل يستخدم أنماطًا مثل Best-of-N أو RAG ليس ممكنًا فحسب، بل هو بديهي أيضًا. من خلال البناء باستخدام Pyspur، أنت لا تقوم فقط بربط الصناديق؛ بل تقوم بهندسة أنظمة ذكاء اصطناعي موثوقة وقابلة للتصحيح وقابلة للتوسع.