DSPy 3: بناء خطوط أنابيب LLM ذاتية التحسين بلغة بايثون

مقدمة

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

أُصدر DSPy 3 من قبل مختبر Stanford NLP في مطلع 2026، ويأتي مع مُحسّن أسرع (MIPROv2)، ودعم أصلي للعمل غير المتزامن، والبث، ومحولات لـ OpenAI و Anthropic و Google و Groq وأي مزوّد متوافق مع LiteLLM. وهو الإطار الذي يُشغّل أنظمة استرجاع واستدلال إنتاجية في شركات مثل JetBlue و Databricks و Replit.

في هذا الدرس ستبني خط أنابيب للإجابة عن أسئلة معرفية يتضمن:

• استرجاع مقاطع من قاعدة معرفية صغيرة
• الاستدلال خطوة بخطوة عبر سلسلة الأفكار
• التحسين التلقائي مقابل مقاييس الدقة والكمون
• تحسناً قابلاً للقياس دون أن تلمس سلسلة موجّه واحدة

المتطلبات

قبل البدء، تأكد من توفر:

• بايثون 3.10 أو أحدث
• مفتاح API لـ OpenAI أو Anthropic (أو أي مزود متوافق مع LiteLLM)
• إلمام بـ pip والبيئات الافتراضية
• معرفة أساسية بمفاهيم LLM (الرموز، التضمينات، RAG)
• نحو 30 دقيقة من التركيز

ما الذي ستبنيه

نظام أسئلة وأجوبة ذاتي التحسين يجيب على أسئلة حول منظومة التكنولوجيا التونسية انطلاقاً من مجموعة نصوص صغيرة. في النهاية ستحصل على:

• خط أنابيب مكتوب بلغة بايثون نقية ومُحدد الأنواع
• دالة مقياس للتقييم الآلي
• برنامج مُجمَّع يتفوق على خط الأساس الصفري بأكثر من 15 نقطة دقة
• ملفات محفوظة يمكنك تحميلها وتقديمها عبر FastAPI

الخطوة 1: تثبيت DSPy وتهيئة نموذج اللغة

أنشئ مشروعاً جديداً وثبّت DSPy 3:

mkdir dspy-qa && cd dspy-qa
python -m venv .venv
source .venv/bin/activate
pip install "dspy-ai>=3.0.0" "litellm" "datasets" "fastapi" "uvicorn"

أنشئ ملف main.py وهيّئ DSPy بالنموذج الذي تفضله. تستخدم المكتبة LiteLLM في الخلفية، لذا الكود نفسه يعمل مع OpenAI و Anthropic و Groq و Ollama و vLLM.

import os
import dspy
 
lm = dspy.LM(
    "openai/gpt-4o-mini",
    api_key=os.environ["OPENAI_API_KEY"],
    temperature=0.0,
    max_tokens=512,
)
dspy.configure(lm=lm)

لتبديل المزود، يكفي تغيير سلسلة النموذج وحدها — وكل ما يأتي بعدها لا يعتمد على المزود.

# Anthropic
lm = dspy.LM("anthropic/claude-haiku-4-5", api_key=os.environ["ANTHROPIC_API_KEY"])
 
# Groq (استنتاج سريع)
lm = dspy.LM("groq/llama-3.3-70b-versatile", api_key=os.environ["GROQ_API_KEY"])
 
# Ollama محلي
lm = dspy.LM("ollama/llama3.1", api_base="http://localhost:11434")

الخطوة 2: تعريف توقيع

التوقيع هو إعلان مُحدد الأنواع للمهمة: المدخلات والمخرجات وسلسلة وصف توضّح النية. يستخدمه DSPy لبناء الموجه الفعلي وقت التشغيل.

class GenerateAnswer(dspy.Signature):
    """Answer questions about Tunisian tech companies and startups using the provided context."""
 
    context: list[str] = dspy.InputField(desc="Relevant passages from the knowledge base")
    question: str = dspy.InputField()
    answer: str = dspy.OutputField(desc="A concise factual answer, one to two sentences")

لاحظ أنك لن تكتب أبداً عبارات من قبيل "أنت مساعد مفيد" أو "أجب باختصار" داخل موجّه — وصف الـ docstring وحقول الإدخال والإخراج يكفي. مُحسّن DSPy سيُحسّن هذا لاحقاً إلى الموجّه الأفضل أداءً تلقائياً.

الخطوة 3: تركيب الوحدات

الوحدات تغلّف التوقيع باستراتيجية استدلال. الوحدات الثلاث الأكثر استخداماً هي:

• dspy.Predict — تنبؤ مباشر (استدعاء واحد للنموذج)
• dspy.ChainOfThought — تضيف حقل استدلال قبل المخرجة
• dspy.ReAct — تتبادل بين الاستدلال واستدعاءات الأدوات

ابنِ وحدة سؤال وجواب باستخدام سلسلة الأفكار:

class TunisianTechQA(dspy.Module):
    def __init__(self, num_passages: int = 3):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate_answer = dspy.ChainOfThought(GenerateAnswer)
 
    def forward(self, question: str):
        context = self.retrieve(question).passages
        prediction = self.generate_answer(context=context, question=question)
        return dspy.Prediction(context=context, answer=prediction.answer)

تجمع الدالة forward بين الاسترجاع والتوليد. ولا توجد أي موجهات مكتوبة يدوياً في أي مكان.

الخطوة 4: إعداد الاسترجاع

يأتي DSPy مع مسترجعات لـ ChromaDB و Pinecone و Weaviate و Qdrant، إضافة إلى مخزن متجهات داخلي في الذاكرة اسمه Embeddings. في هذا الدرس ستستخدم النسخة الداخلية مع مجموعة نصوص صغيرة.

corpus = [
    "InstaDeep is a Tunisian AI startup founded in 2014, acquired by BioNTech in 2023 for around 562 million euros.",
    "Expensya, founded in Tunis in 2014, was acquired by Medius in 2023.",
    "Tunisia hosts the annual Maghreb Emerging conference focused on AI and deep tech investment.",
    "Noqta is a Tunisian software studio focused on AI-driven web platforms and developer tooling.",
    "Datavora, founded in Tunis, was acquired by Semrush in 2020.",
    "The Smart Tunisia program subsidizes salaries for tech graduates working in export-oriented companies.",
]
 
embedder = dspy.Embedder("openai/text-embedding-3-small")
retriever = dspy.retrievers.Embeddings(embedder=embedder, corpus=corpus, k=3)
dspy.configure(rm=retriever)

والآن أي وحدة تستخدم dspy.Retrieve ستسحب من هذه المجموعة. وفي الإنتاج تستبدل هذا بقاعدة متجهات مُدارة.

الخطوة 5: تجهيز بيانات التدريب والتقييم

يحتاج التحسين إلى بيانات. تكفيك 20 إلى 50 مثالاً لرؤية تحسن ملموس — يتكفل DSPy بالعمل الثقيل عبر التمهيد التلقائي.

from dspy import Example
 
trainset = [
    Example(question="Who acquired InstaDeep and when?",
            answer="BioNTech acquired InstaDeep in 2023.").with_inputs("question"),
    Example(question="What does Noqta build?",
            answer="Noqta builds AI-driven web platforms and developer tooling.").with_inputs("question"),
    Example(question="Which company acquired Expensya?",
            answer="Medius acquired Expensya in 2023.").with_inputs("question"),
    Example(question="What is Smart Tunisia?",
            answer="A program subsidizing salaries for tech graduates in export-oriented companies.").with_inputs("question"),
    # ... أضف 16 مثالاً آخر ليصبح المجموع 20
]
 
devset = [
    Example(question="In what year was Datavora acquired?",
            answer="2020.").with_inputs("question"),
    Example(question="What sector is InstaDeep in?",
            answer="Artificial intelligence.").with_inputs("question"),
    # ... أضف 8 أمثلة أخرى ليصبح المجموع 10
]

لاحظ استخدام .with_inputs("question") — فهو يخبر DSPy أيّ الحقول مدخلات وأيّها مخرجات مرجوة.

الخطوة 6: تعريف مقياس

المقياس هو دالة بايثون تأخذ مثالاً ذهبياً وتنبؤاً وتُرجع درجة. للأسئلة الواقعية، التطابق التام صارم جداً؛ استخدم مقياساً قائماً على التشابه الدلالي أو دع نموذج لغة يحكم.

def answer_correctness(example, prediction, trace=None) -> float:
    """LLM-as-judge metric. Returns 1.0 if the prediction is factually equivalent to the gold answer."""
    judge = dspy.Predict("question, gold_answer, predicted_answer -> is_correct: bool")
    result = judge(
        question=example.question,
        gold_answer=example.answer,
        predicted_answer=prediction.answer,
    )
    return 1.0 if result.is_correct else 0.0

هذا المقياس نفسه برنامج DSPy. مرحباً بك في عالم التكرار الذاتي.

الخطوة 7: التجميع باستخدام MIPROv2

هنا يستحق DSPy اسمه. يبحث المُحسّن في فضاء تعليمات الموجّه والأمثلة قليلة الإطلاق لتعظيم مقياسك على مجموعة التدريب.

from dspy.teleprompt import MIPROv2
 
qa = TunisianTechQA()
 
# خط الأساس قبل التحسين
baseline_score = sum(answer_correctness(ex, qa(question=ex.question)) for ex in devset) / len(devset)
print(f"Zero-shot baseline: {baseline_score:.2%}")
 
optimizer = MIPROv2(
    metric=answer_correctness,
    auto="medium",  # "light" أو "medium" أو "heavy"
    num_threads=8,
)
 
compiled_qa = optimizer.compile(
    qa,
    trainset=trainset,
    requires_permission_to_run=False,
    minibatch=True,
)
 
# الدرجة بعد التحسين
optimized_score = sum(answer_correctness(ex, compiled_qa(question=ex.question)) for ex in devset) / len(devset)
print(f"Optimized: {optimized_score:.2%}")

على مجموعة تدريب من 20 مثالاً، توقع خطوط أساس بين 60 و70 بالمئة ودرجات مُحسَّنة فوق 85 بالمئة. يكتمل المُحسّن عادةً خلال دقيقتين إلى خمس دقائق حسب حجم البيانات وإعداد auto.

الخطوة 8: استكشاف ما تعلمه

DSPy شفاف — يمكنك أن ترى الموجّه الفعلي الذي أنتجه المُحسّن.

compiled_qa.generate_answer.demos  # الأمثلة قليلة الإطلاق التي اختارها
compiled_qa.generate_answer.signature.instructions  # التعليمات المُعاد كتابتها

هذا لا يُقدّر بثمن في التنقيح، وأيضاً لنقل الموجه المُجمَّع إلى أنظمة أخرى إذا قررت يوماً مغادرة DSPy.

الخطوة 9: الحفظ والتحميل

التجميع مكلف، لذا احفظ النتيجة.

compiled_qa.save("./compiled_qa.json")
 
# لاحقاً، في الإنتاج
qa = TunisianTechQA()
qa.load("./compiled_qa.json")
answer = qa(question="Who acquired InstaDeep?").answer

الملف المحفوظ ملف JSON صغير يحتوي تعليمات وأمثلة — لا أوزان نموذج ولا حالة احتكارية.

الخطوة 10: التقديم عبر FastAPI

غلّف البرنامج المُجمَّع داخل واجهة API صغيرة.

from fastapi import FastAPI
from pydantic import BaseModel
 
app = FastAPI()
 
qa = TunisianTechQA()
qa.load("./compiled_qa.json")
 
class Query(BaseModel):
    question: str
 
@app.post("/ask")
async def ask(q: Query):
    result = await qa.acall(question=q.question)
    return {"answer": result.answer, "context": result.context}

شغّله:

uvicorn main:app --reload

أضاف DSPy 3 دعماً أصلياً للعمل غير المتزامن، لذا فإن acall غير حاجبة وتتكامل بسلاسة مع FastAPI و Starlette وأي حزمة غير متزامنة.

اختبار التطبيق

فحص سريع للسلامة قبل النشر:

test_questions = [
    "Who founded InstaDeep?",
    "When was Expensya acquired?",
    "What does the Smart Tunisia program do?",
]
 
for q in test_questions:
    result = qa(question=q)
    print(f"Q: {q}\nA: {result.answer}\n")

إذا كانت الإجابات تستشهد بحقائق غير موجودة في المجموعة، فإن الاسترجاع لديك ضيق جداً — ارفع قيمة k في dspy.Retrieve أو وسّع المجموعة.

استكشاف الأخطاء وحلها

المُحسّن بطيء جداً. اضبط auto="light" لتشغيل سريع أو قلّل num_threads إذا كنت محدوداً بحدود طلبات الـ API.

المقياس يعطي درجات متساوية. نموذج الحكم لديك صغير على الأرجح. استخدم نموذجاً أقوى للتقييم فقط عبر تمرير lm= إلى استدعاء Predict للحكم.

التنبؤات تتجاهل السياق. تأكد أن docstring في توقيعك يطلب من النموذج استخدام السياق، وزِد عدد المقاطع المسترجعة.

حدود طلبات OpenAI أثناء التجميع. قد يطلق MIPROv2 مع auto="heavy" آلاف الاستدعاءات. ابدأ بـ auto="light" أو شغّل نموذج Ollama محلياً أثناء النماذج الأولية.

خطأ ModuleNotFoundError: dspy.retrievers. أنت على نسخة قديمة. شغّل pip install -U "dspy-ai>=3.0.0".

الخطوات التالية

وسّع خط الأنابيب هذا بـ:

• استخدام الأدوات: استبدل ChainOfThought بـ dspy.ReAct وأعطه أداة حاسبة أو بحث في الويب
• الاستدلال متعدد القفزات: سلسل وحدتي dspy.ChainOfThought ليقوم النموذج بتفكيك الأسئلة المعقدة
• استرجاع إنتاجي: صل بقاعدة متجهات مُدارة. راجع درس البحث المتجهي مع Pinecone لجانب الفهرسة، أو دليل المراقبة مع LangFuse لمراقبة الكمون والجودة في الإنتاج
• تقييم أقوى: استخدم dspy.evaluate.SemanticF1 المضمن أو اربطه بـ Promptfoo
• تحسين مستمر: أعد تشغيل المُحسّن أسبوعياً على بيانات متنامية وانشر الملف الجديد عبر CI

الخاتمة

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

يعمم هذا النمط إلى ما هو أبعد بكثير من الأسئلة والأجوبة. التصنيف، التلخيص، الاستخراج المُهيكل، توجيه الوكلاء المتعددين — أيّ شيء كنت ستهندس له موجهاً يدوياً، يمكن التعبير عنه كتوقيع وتجميعه. وحين تستبدل النماذج من GPT-4o-mini إلى Claude Haiku أو إلى نموذج Llama محلي، يعيد المُحسّن ضبط كل شيء بدلاً من إجبارك على إعادة الكتابة يدوياً.

تلك هي رهانة DSPy: توقف عن ضبط الموجهات، وابدأ بالبرمجة.