Ollama: تمارين المرحلة الثانية الشاملة

# SupportBot V2 يستفيد من جميع مهارات المرحلة الثانية:
# - Modelfile مخصص لتخصيص الأدوار
# - تسريع GPU للسرعة
# - واجهة API متوافقة مع OpenAI لسهولة التكامل
# - متعدد الوسائط لتحليل الصور

# supportbot.Modelfile
FROM qwen2.5:7b
SYSTEM You are SupportBot for GlobalShop e-commerce. Be polite, concise (2-3 sentences). For order queries ask for order number. If unsure, say 'Let me connect you with a human agent.'
PARAMETER temperature 0.4
PARAMETER num_ctx 4096
PARAMETER num_predict 256
MESSAGE user What is your return policy?
MESSAGE assistant Our return policy allows returns within 30 days in original condition. Free return shipping included.

import ollama

# Create model (run once)
# ollama.create(model='supportbot', from_='./supportbot.Modelfile')

# Use the custom model - no system prompt needed!
def test_supportbot():
    questions = [
        "I want to return order #12345",
        "Do you ship to Germany?",
        "What's the price of the iPhone 15?"
    ]
    for q in questions:
        response = ollama.chat(
            model='supportbot',
            messages=[{'role': 'user', 'content': q}],
            stream=False
        )
        print(f"Q: {q}")
        print(f"A: {response['message']['content']}\n")

test_supportbot()

import ollama
import time
from statistics import mean

def benchmark(model: str, prompt: str, runs: int = 5) -> dict:
    """Benchmark model inference speed."""
    latencies = []
    token_speeds = []

    # تحمية (الاستدعاء الأول يتضمن تحميل النموذج)
    ollama.chat(model=model, messages=[{'role': 'user', 'content': 'warm up'}],
                stream=False)

    for _ in range(runs):
        start = time.time()
        response = ollama.chat(
            model=model,
            messages=[{'role': 'user', 'content': prompt}],
            stream=False
        )
        elapsed = time.time() - start

        # تقدير الرموز (تقريبي: ~4 أحرف لكل رمز بالإنجليزية)
        content = response['message']['content']
        token_count = len(content) // 4

        latencies.append(elapsed)
        token_speeds.append(token_count / elapsed if elapsed > 0 else 0)

    return {
        "model": model,
        "avg_latency_s": round(mean(latencies), 2),
        "avg_tokens_per_s": round(mean(token_speeds), 1),
        "runs": runs
    }

# تشغيل القياس
print("=== CPU vs GPU Benchmark ===")
result = benchmark("qwen2.5", "Explain machine learning in 100 words")
print(f"Model: {result['model']}")
print(f"Avg latency: {result['avg_latency_s']}s")
print(f"Avg speed: {result['avg_tokens_per_s']} tokens/s")

# مقارنة نماذج مختلفة
for model in ["llama3.2:3b", "qwen2.5", "mistral"]:
    result = benchmark(model, "Explain AI in 50 words", runs=3)
    print(f"{model}: {result['avg_tokens_per_s']} tok/s, {result['avg_latency_s']}s")

=== CPU vs GPU Benchmark ===

from openai import OpenAI
import json

# قبل الترحيل
# client = OpenAI(api_key="sk-xxx")
# model = "gpt-4"

# بعد الترحيل - فقط هذه الأسطر الثلاثة تغيرت
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
model = "qwen2.5"
embed_model = "nomic-embed-text"

# اختبار 1: المحادثة
print("=== Test 1: Chat ===")
response = client.chat.completions.create(
    model=model,
    messages=[{"role": "user", "content": "Hello, who are you?"}],
    temperature=0.3
)
print(response.choices[0].message.content)

# اختبار 2: المتدفقة
print("\n=== Test 2: Streaming ===")
stream = client.chat.completions.create(
    model=model,
    messages=[{"role": "user", "content": "Count from 1 to 5"}],
    stream=True
)
for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="")
print()

# اختبار 3: وضع JSON
print("\n=== Test 3: JSON Mode ===")
response = client.chat.completions.create(
    model=model,
    messages=[{"role": "user", "content": "List 3 colors as JSON array"}],
    response_format={"type": "json_object"},
    temperature=0.1
)
print(response.choices[0].message.content)

# اختبار 4: التضمينات
print("\n=== Test 4: Embeddings ===")
response = client.embeddings.create(model=embed_model, input="Hello world")
print(f"Embedding dimension: {len(response.data[0].embedding)}")

print("\n=== All tests passed! ===")

=== Test 1: Chat ===
\n=== Test 2: Streaming ===
\n=== Test 3: JSON Mode ===
\n=== Test 4: Embeddings ===
\n=== All tests passed! ===

from fastapi import FastAPI, UploadFile, File, Form
from fastapi.responses import JSONResponse
import ollama
import base64

app = FastAPI(title="Image Description Service")

@app.post("/describe")
async def describe_image(
    file: UploadFile = File(...),
    prompt: str = Form("Describe this image in detail")
):
    """Describe an uploaded image using llava."""
    image_data = base64.b64encode(await file.read()).decode("utf-8")

    response = ollama.chat(
        model="llava",
        messages=[{
            "role": "user",
            "content": prompt,
            "images": [image_data]
        }],
        stream=False,
        options={"temperature": 0.3}
    )

    return JSONResponse({
        "description": response["message"]["content"],
        "model": "llava",
        "filename": file.filename
    })

@app.post("/extract-text")
async def extract_text(file: UploadFile = File(...)):
    """OCR-like text extraction from image."""
    image_data = base64.b64encode(await file.read()).decode("utf-8")

    response = ollama.chat(
        model="llava",
        messages=[{
            "role": "user",
            "content": "Extract all visible text from this image verbatim.",
            "images": [image_data]
        }],
        stream=False,
        options={"temperature": 0.1}
    )

    return JSONResponse({
        "text": response["message"]["content"],
        "model": "llava"
    })

# Run: uvicorn service:app --host 0.0.0.0 --port 8000

# Function defined successfully

# مخطط بنية SupportBot V2

SUPPORTBOT_V2_SPEC = {
    "name": "SupportBot V2",
    "components": {
        "custom_model": {
            "tool": "Modelfile",
            "purpose": "Role-specialized customer service model",
            "config": "qwen2.5:7b + custom SYSTEM + PARAMETER"
        },
        "api_gateway": {
            "tool": "OpenAI compatible API",
            "purpose": "Easy integration with existing tools",
            "config": "/v1/chat/completions endpoint"
        },
        "image_service": {
            "tool": "llava multimodal",
            "purpose": "Damage report analysis, product photo description",
            "config": "FastAPI + llava + Base64 encoding"
        },
        "embedding_service": {
            "tool": "nomic-embed-text",
            "purpose": "Product knowledge base (RAG ready for Phase 3)",
            "config": "/v1/embeddings endpoint"
        },
        "gpu_acceleration": {
            "tool": "CUDA / Metal",
            "purpose": "Fast inference for real-time responses",
            "config": "Auto-detected by Ollama"
        }
    },
    "cost_comparison": {
        "before": "GPT-4 API: ~2,000 USD/month",
        "after": "Local Ollama: ~100 USD/month (hardware + electricity)",
        "saving": "1,900 USD/month (95% reduction)"
    }
}

print("SupportBot V2 Architecture:")
for name, spec in SUPPORTBOT_V2_SPEC["components"].items():
    print(f"  {name}: {spec['tool']} - {spec['purpose']}")

SupportBot V2 Architecture:

# ============================================
# شامل: التحقق من مهارات المرحلة الثانية الكاملة
# Modelfile + SDK + GPU + OpenAI compat + Multimodal
# ============================================

import ollama
from openai import OpenAI
import base64
import time
import json
from pathlib import Path

def phase2_verify():
    results = {}

    # اختبار 1: نموذج Modelfile مخصص
    print("=== Test 1: Custom Modelfile Model ===")
    try:
        resp = ollama.chat(
            model="supportbot",
            messages=[{"role": "user", "content": "I want a refund for order #12345"}],
            stream=False
        )
        results["modelfile"] = "PASS"
        print(f"  Response: {resp['message']['content'][:80]}...")
    except Exception as e:
        results["modelfile"] = f"FAIL: {e}"

    # اختبار 2: تسريع GPU
    print("\n=== Test 2: GPU Acceleration ===")
    start = time.time()
    resp = ollama.chat(
        model="qwen2.5",
        messages=[{"role": "user", "content": "Hello"}],
        stream=False
    )
    elapsed = time.time() - start
    results["gpu_speed"] = f"{elapsed:.2f}s"
    print(f"  Response time: {elapsed:.2f}s ({'GPU' if elapsed < 3 else 'CPU'})")

    # اختبار 3: واجهة API المتوافقة مع OpenAI
    print("\n=== Test 3: OpenAI Compatible API ===")
    try:
        client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
        resp = client.chat.completions.create(
            model="qwen2.5",
            messages=[{"role": "user", "content": "Say OK"}],
            temperature=0.1
        )
        results["openai_compat"] = "PASS"
        print(f"  Response: {resp.choices[0].message.content}")
    except Exception as e:
        results["openai_compat"] = f"FAIL: {e}"

    # اختبار 4: التضمينات
    print("\n=== Test 4: Embeddings ===")
    try:
        resp = client.embeddings.create(model="nomic-embed-text", input="test")
        dim = len(resp.data[0].embedding)
        results["embeddings"] = f"PASS ({dim}d)"
        print(f"  Embedding dimension: {dim}")
    except Exception as e:
        results["embeddings"] = f"FAIL: {e}"

    # اختبار 5: المتدفقة
    print("\n=== Test 5: Streaming ===")
    try:
        stream = client.chat.completions.create(
            model="qwen2.5",
            messages=[{"role": "user", "content": "Count 1 to 3"}],
            stream=True
        )
        chunks = 0
        for chunk in stream:
            if chunk.choices[0].delta.content:
                chunks += 1
        results["streaming"] = f"PASS ({chunks} chunks)"
        print(f"  Received {chunks} chunks")
    except Exception as e:
        results["streaming"] = f"FAIL: {e}"

    # الملخص
    print("\n" + "=" * 40)
    print("Phase 2 Verification Results:")
    for test, status in results.items():
        print(f"  {test}: {status}")

    passed = sum(1 for v in results.values() if "PASS" in str(v))
    print(f"\n  Total: {passed}/{len(results)} passed")
    if passed >= 4:
        print("  ✅ Ready for Phase 3!")
    else:
        print("  ⚠️ Review failed tests.")

if __name__ == "__main__":
    phase2_verify()