سير عمل النماذج المتعددة

تشغيل نماذج متعددة محلياً يمكّن سير عمل متطور: توجيه الاستعلامات لنماذج متخصصة، دمج الردود، وسلسلة النماذج.

لماذا نماذج متعددة؟

┌─────────────────────────────────────────────────────────────────┐
│                    استراتيجيات النماذج المتعددة                  │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  1. التوجيه: أرسل الاستعلامات لأفضل نموذج لكل مهمة             │
│     • أسئلة الكود → deepseek-coder                              │
│     • دردشة عامة → llama3.2                                     │
│     • متعدد اللغات → qwen2.5                                    │
│                                                                 │
│  2. الدمج: اجمع المخرجات من نماذج متعددة                        │
│     • ولّد مع 3 نماذج، صوّت على أفضل إجابة                      │
│     • يزيد الموثوقية للمهام الحرجة                              │
│                                                                 │
│  3. السلسلة: استخدم مخرجات نموذج كمدخل لآخر                     │
│     • نموذج سريع للمسودة، نموذج كبير للتحسين                    │
│     • مصنف → نموذج متخصص                                        │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

موجه النموذج

import ollama
from typing import Literal

class ModelRouter:
    """وجّه الاستعلامات للنموذج الأنسب."""

    def __init__(self):
        self.models = {
            "code": "deepseek-coder",
            "general": "llama3.2",
            "multilingual": "qwen2.5:7b",
            "fast": "phi3:mini"
        }
        self.classifier = "llama3.2"

    def classify_query(self, query: str) -> str:
        """صنف نوع الاستعلام."""
        response = ollama.chat(
            model=self.classifier,
            messages=[{
                "role": "user",
                "content": f"""Classify this query into ONE category:
- code (programming, debugging, code review)
- multilingual (non-English or translation)
- general (everything else)

Query: {query}

Reply with just the category name."""
            }]
        )
        category = response["message"]["content"].strip().lower()

        # خريطة لنوع النموذج
        if "code" in category:
            return "code"
        elif "multilingual" in category or "translation" in category:
            return "multilingual"
        return "general"

    def route(self, query: str) -> str:
        """وجّه الاستعلام للنموذج المناسب واحصل على الرد."""
        category = self.classify_query(query)
        model = self.models[category]

        print(f"[التوجيه إلى {model}]")

        response = ollama.chat(
            model=model,
            messages=[{"role": "user", "content": query}]
        )
        return response["message"]["content"]

# الاستخدام
router = ModelRouter()
print(router.route("Write a Python function to sort a list"))
print(router.route("Translate 'hello' to Japanese"))
print(router.route("What causes rain?"))

دمج الردود

import ollama
from collections import Counter

class EnsembleModel:
    """ادمج الردود من نماذج متعددة."""

    def __init__(self, models: list[str]):
        self.models = models

    def generate_all(self, query: str) -> list[str]:
        """احصل على ردود من جميع النماذج."""
        responses = []
        for model in self.models:
            response = ollama.chat(
                model=model,
                messages=[{"role": "user", "content": query}]
            )
            responses.append(response["message"]["content"])
        return responses

    def synthesize(self, query: str) -> str:
        """اصنع الردود في إجابة نهائية."""
        responses = self.generate_all(query)

        # استخدم نموذج آخر للتركيب
        synthesis_prompt = f"""You received these answers to the question: "{query}"

Answer 1: {responses[0]}

Answer 2: {responses[1]}

Answer 3: {responses[2]}

Synthesize these into a single, accurate response. If they disagree, go with the majority or most detailed answer."""

        final = ollama.chat(
            model="llama3.2",
            messages=[{"role": "user", "content": synthesis_prompt}]
        )
        return final["message"]["content"]

    def vote(self, query: str, options: list[str]) -> str:
        """اجعل النماذج تصوت على خيارات متعددة."""
        votes = []
        for model in self.models:
            response = ollama.chat(
                model=model,
                messages=[{
                    "role": "user",
                    "content": f"""Question: {query}
Options: {options}
Reply with just the letter of the correct answer."""
                }]
            )
            vote = response["message"]["content"].strip().upper()
            if vote in ["A", "B", "C", "D"]:
                votes.append(vote)

        # أرجع تصويت الأغلبية
        if votes:
            return Counter(votes).most_common(1)[0][0]
        return "Unable to determine"

# الاستخدام
ensemble = EnsembleModel(["llama3.2", "mistral", "phi3:medium"])
answer = ensemble.synthesize("What are the benefits of exercise?")
print(answer)

سلسلة النماذج

import ollama

class ModelChain:
    """سلسل النماذج للمعالجة متعددة الخطوات."""

    def draft_and_refine(self, query: str) -> str:
        """استخدم نموذج سريع للمسودة، نموذج كبير للتحسين."""
        # الخطوة 1: مسودة سريعة
        draft = ollama.chat(
            model="phi3:mini",
            messages=[{"role": "user", "content": query}]
        )["message"]["content"]

        # الخطوة 2: حسّن مع نموذج أكبر
        refined = ollama.chat(
            model="llama3.2",
            messages=[{
                "role": "user",
                "content": f"""Improve this response for clarity and accuracy:

Original response: {draft}

Original question was: {query}

Provide an improved version:"""
            }]
        )
        return refined["message"]["content"]

    def translate_then_answer(self, query: str, source_lang: str) -> str:
        """ترجم استعلام غير إنجليزي، أجب، ترجم للخلف."""
        # الخطوة 1: ترجم للإنجليزية
        english_query = ollama.chat(
            model="qwen2.5:7b",
            messages=[{
                "role": "user",
                "content": f"Translate to English: {query}"
            }]
        )["message"]["content"]

        # الخطوة 2: أجب بالإنجليزية (أفضل نموذج)
        answer = ollama.chat(
            model="llama3.2",
            messages=[{"role": "user", "content": english_query}]
        )["message"]["content"]

        # الخطوة 3: ترجم الإجابة للخلف
        final = ollama.chat(
            model="qwen2.5:7b",
            messages=[{
                "role": "user",
                "content": f"Translate to {source_lang}: {answer}"
            }]
        )
        return final["message"]["content"]

    def code_review_chain(self, code: str) -> dict:
        """مراجعة كود متعددة النماذج."""
        # مراجعة الأمان
        security = ollama.chat(
            model="deepseek-coder",
            messages=[{
                "role": "user",
                "content": f"Review for security issues only:\n```\n{code}\n```"
            }]
        )["message"]["content"]

        # مراجعة الأداء
        performance = ollama.chat(
            model="deepseek-coder",
            messages=[{
                "role": "user",
                "content": f"Review for performance issues only:\n```\n{code}\n```"
            }]
        )["message"]["content"]

        # ملخص نهائي
        summary = ollama.chat(
            model="llama3.2",
            messages=[{
                "role": "user",
                "content": f"""Summarize this code review:

Security findings: {security}

Performance findings: {performance}

Provide a brief executive summary."""
            }]
        )["message"]["content"]

        return {
            "security": security,
            "performance": performance,
            "summary": summary
        }

# الاستخدام
chain = ModelChain()
result = chain.draft_and_refine("Explain quantum computing")
print(result)

المعالجة المتوازية

import ollama
import asyncio
from concurrent.futures import ThreadPoolExecutor

async def parallel_models(query: str, models: list[str]) -> dict[str, str]:
    """استعلم من نماذج متعددة بالتوازي."""
    loop = asyncio.get_event_loop()

    def query_model(model: str) -> tuple[str, str]:
        response = ollama.chat(
            model=model,
            messages=[{"role": "user", "content": query}]
        )
        return model, response["message"]["content"]

    # شغّل بالتوازي باستخدام مجمع الخيوط
    with ThreadPoolExecutor(max_workers=len(models)) as executor:
        futures = [
            loop.run_in_executor(executor, query_model, model)
            for model in models
        ]
        results = await asyncio.gather(*futures)

    return dict(results)

# الاستخدام
async def main():
    results = await parallel_models(
        "What is the capital of France?",
        ["llama3.2", "mistral", "phi3:mini"]
    )
    for model, response in results.items():
        print(f"{model}: {response[:100]}...")

asyncio.run(main())

مصفوفة اختيار النموذج

المهمة	النموذج الموصى	لماذا
توليد الكود	deepseek-coder	متخصص للكود
ردود سريعة	phi3:mini	الأصغر، الأسرع
استدلال معقد	llama3.2:70b	الأكثر قدرة
متعدد اللغات	qwen2.5:7b	أفضل تغطية لغوية
دردشة عامة	llama3.2:8b	شامل جيد

سير عمل النماذج المتعددة يعظم نقاط قوة كل نموذج مع العمل بالكامل محلياً. في الوحدة التالية، سنحسن الأداء. :::