Ollama: تنسيق النماذج المتعددة

# Router pattern: small model classifies, large model handles complex
intent = small_model.classify(question)  # 1 second
if intent == "simple":
    answer = small_model.answer(question)  # 2 seconds
else:
    answer = large_model.answer(question)  # 5 seconds

import ollama
import json
from dataclasses import dataclass
from typing import Optional

@dataclass
class ModelRouter:
    classifier_model: str = "llama3.2:3b"
    small_model: str = "llama3.2:3b"
    large_model: str = "qwen2.5"

    INTENTS = {
        "faq": "simple",
        "order_status": "simple",
        "product_info": "medium",
        "comparison": "complex",
        "complaint": "complex"
    }

    def classify(self, question: str) -> dict:
        response = ollama.chat(
            model=self.classifier_model,
            messages=[{
                "role": "user",
                "content": f"""Classify this customer question.
Return JSON: {{"intent": "faq|order_status|product_info|comparison|complaint", "complexity": "simple|medium|complex"}}
Question: {question}"""
            }],
            format="json",
            stream=False,
            options={"temperature": 0.1}
        )
        return json.loads(response["message"]["content"])

    def route(self, question: str, system: str = "") -> str:
        intent_data = self.classify(question)
        complexity = intent_data.get("complexity", "simple")

        model = self.small_model if complexity == "simple" else self.large_model
        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": question})

        response = ollama.chat(model=model, messages=messages,
                               stream=False, options={"temperature": 0.3})
        return response["message"]["content"], model, intent_data

# Usage
router = ModelRouter()
questions = [
    "What is the return policy?",
    "Compare the 3 wireless headphones you sell",
    "Where is order #88765?"
]
for q in questions:
    answer, model_used, intent = router.route(q, system="You are SupportBot.")
    print(f"Q: {q}")
    print(f"Intent: {intent}, Model: {model_used}")
    print(f"A: {answer[:100]}...\n")

# Function defined successfully

import ollama

class CascadePipeline:
    def __init__(self):
        self.draft_model = "llama3.2:3b"
        self.refine_model = "qwen2.5"
        self.format_model = "llama3.2:3b"

    def generate(self, topic: str) -> dict:
        # Stage 1: Draft
        draft = ollama.chat(
            model=self.draft_model,
            messages=[{"role": "user",
                       "content": f"Write a brief draft about: {topic}"}],
            stream=False, options={"temperature": 0.5}
        )["message"]["content"]

        # Stage 2: Refine
        refined = ollama.chat(
            model=self.refine_model,
            messages=[
                {"role": "system", "content": "Improve the following text. Add details and fix errors."},
                {"role": "user", "content": draft}
            ],
            stream=False, options={"temperature": 0.3}
        )["message"]["content"]

        # Stage 3: Format
        formatted = ollama.chat(
            model=self.format_model,
            messages=[
                {"role": "system", "content": "Format this text as a professional product description with bullet points."},
                {"role": "user", "content": refined}
            ],
            stream=False, options={"temperature": 0.1}
        )["message"]["content"]

        return {"draft": draft, "refined": refined, "formatted": formatted}

pipeline = CascadePipeline()
result = pipeline.generate("wireless noise-cancelling headphones")
print("=== Final Output ===")
print(result["formatted"])

=== Final Output ===

import asyncio
import ollama
from dataclasses import dataclass

@dataclass
class ParallelVoter:
    models: list[str] = None

    def __post_init__(self):
        if self.models is None:
            self.models = ["llama3.2:3b", "qwen2.5", "mistral"]

    async def query_model(self, model: str, question: str) -> dict:
        client = ollama.AsyncClient()
        response = await client.chat(
            model=model,
            messages=[{"role": "user", "content": question}],
            stream=False,
            options={"temperature": 0.3}
        )
        return {"model": model, "answer": response["message"]["content"]}

    async def vote(self, question: str) -> dict:
        tasks = [self.query_model(m, question) for m in self.models]
        results = await asyncio.gather(*tasks)

        # Simple voting: check for consensus
        answers = [r["answer"] for r in results]
        return {
            "question": question,
            "results": results,
            "consensus": len(set(a[:50] for a in answers)) == 1
        }

# Usage
async def main():
    voter = ParallelVoter()
    result = await voter.vote("Is 2+2 equal to 4?")
    for r in result["results"]:
        print(f"{r['model']}: {r['answer'][:80]}")

asyncio.run(main())

# Function defined successfully

import ollama

def weighted_ensemble(question: str, models: list[dict]) -> str:
    """Query multiple models and select the best response."""
    responses = []
    for m in models:
        resp = ollama.chat(
            model=m["name"],
            messages=[{"role": "user", "content": question}],
            stream=False,
            options={"temperature": m.get("temperature", 0.3)}
        )
        responses.append({
            "model": m["name"],
            "answer": resp["message"]["content"],
            "weight": m.get("weight", 1.0),
            "length": len(resp["message"]["content"])
        })

    # Select longest response from highest-weight model
    responses.sort(key=lambda x: x["weight"] * x["length"], reverse=True)
    return responses[0]["answer"]

models = [
    {"name": "qwen2.5", "weight": 1.5, "temperature": 0.3},
    {"name": "llama3.2:3b", "weight": 0.8, "temperature": 0.5},
]

result = weighted_ensemble("Explain the benefits of local AI", models)
print(result)

# Function defined successfully

# ============================================
# Comprehensive: SupportBot V4 with Router
# Multi-model routing for e-commerce support
# ============================================

import ollama
import json
import time
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class SupportBotV4:
    classifier: str = "llama3.2:3b"
    simple_model: str = "llama3.2:3b"
    complex_model: str = "qwen2.5"
    system_prompt: str = "You are SupportBot for GlobalShop e-commerce. Be polite and concise."

    INTENT_MAP: dict = field(default_factory=lambda: {
        "faq": "simple",
        "order_status": "simple",
        "shipping": "simple",
        "product_info": "complex",
        "comparison": "complex",
        "complaint": "complex",
        "refund": "complex"
    })

    def classify_intent(self, question: str) -> tuple[str, str]:
        """Classify intent and determine complexity."""
        response = ollama.chat(
            model=self.classifier,
            messages=[{
                "role": "user",
                "content": f"""Classify intent (one word): {question}
Categories: faq, order_status, shipping, product_info, comparison, complaint, refund
Reply with just the category word."""
            }],
            stream=False,
            options={"temperature": 0.0}
        )
        intent = response["message"]["content"].strip().lower()
        for key in self.INTENT_MAP:
            if key in intent:
                return key, self.INTENT_MAP[key]
        return "faq", "simple"

    def answer(self, question: str) -> dict:
        """Route question and generate answer."""
        start = time.time()
        intent, complexity = self.classify_intent(question)

        model = self.simple_model if complexity == "simple" else self.complex_model
        response = ollama.chat(
            model=model,
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": question}
            ],
            stream=False,
            options={"temperature": 0.3 if complexity == "simple" else 0.4}
        )
        elapsed = time.time() - start

        return {
            "question": question,
            "intent": intent,
            "complexity": complexity,
            "model_used": model,
            "answer": response["message"]["content"],
            "latency_s": round(elapsed, 2)
        }

    def batch_process(self, questions: list[str]) -> list[dict]:
        """Process multiple questions and show routing stats."""
        results = [self.answer(q) for q in questions]

        simple_count = sum(1 for r in results if r["complexity"] == "simple")
        avg_simple = sum(r["latency_s"] for r in results if r["complexity"] == "simple") / max(simple_count, 1)
        complex_count = len(results) - simple_count
        avg_complex = sum(r["latency_s"] for r in results if r["complexity"] == "complex") / max(complex_count, 1)

        print(f"=== Routing Stats ===")
        print(f"Simple: {simple_count}/{len(results)} (avg {avg_simple:.1f}s)")
        print(f"Complex: {complex_count}/{len(results)} (avg {avg_complex:.1f}s)")
        print(f"Cost saving: ~{simple_count * 60}% of queries use small model")

        return results

# Usage
if __name__ == "__main__":
    bot = SupportBotV4()
    questions = [
        "What is the return policy?",
        "Where is my order #88765?",
        "Do you ship internationally?",
        "Compare the 3 headphone models in detail",
        "I received a completely wrong item, this is unacceptable!",
        "What are the warranty terms for electronics?",
        "Analyze which laptop is best for a graphic designer under $1000",
    ]
    results = bot.batch_process(questions)
    for r in results:
        print(f"\n[{r['complexity'].upper()}|{r['model_used']}] {r['intent']}")
        print(f"  Q: {r['question']}")
        print(f"  A: {r['answer'][:100]}... ({r['latency_s']}s)")