Ollama: الممارسة الشاملة للمرحلة 3

# docker-compose.phase3.yml
version: "3.8"
services:
  ollama:
    image: ollama/ollama
    ports: ["11434:11434"]
    volumes: ["ollama_data:/root/.ollama"]
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: all
              capabilities: [gpu]
    healthcheck:
      test: ["CMD-SHELL", "curl -f http://localhost:11434/api/tags"]
      interval: 30s
      retries: 5

  chroma:
    image: chromadb/chroma
    ports: ["8001:8000"]
    volumes: ["chroma_data:/chroma/chroma"]
    environment: [ANONYMIZED_TELEMETRY=FALSE]

  webui:
    image: ghcr.io/open-webui/open-webui:main
    ports: ["3000:8080"]
    environment: [OLLAMA_BASE_URL=http://ollama:11434]
    volumes: ["webui_data:/app/backend/data"]
    depends_on:
      ollama: {condition: service_healthy}

volumes:
  ollama_data:
  chroma_data:
  webui_data:

# Deploy and verify
docker compose -f docker-compose.phase3.yml up -d

# Pull models
docker exec ollama ollama pull qwen2.5
docker exec ollama ollama pull nomic-embed-text
docker exec ollama ollama pull llama3.2:3b

# Verify all services
docker compose -f docker-compose.phase3.yml ps
curl http://localhost:11434/api/tags     # Ollama
curl http://localhost:8001/api/v1/heartbeat  # Chroma
curl http://localhost:3000               # WebUI

Full-stack deployment verified successfully, all services running normally

from langchain_ollama import OllamaEmbeddings
from langchain_community.vectorstores import Chroma
import numpy as np

def compare_embeddings(query: str, docs: list[str],
                       models: list[str]) -> dict:
    """Compare retrieval results across embedding models."""
    results = {}
    for model_name in models:
        embeddings = OllamaEmbeddings(model=model_name)
        vectorstore = Chroma.from_texts(
            texts=docs, embedding=embeddings,
            collection_name=f"test_{model_name.replace('-','_')}"
        )
        retrieved = vectorstore.similarity_search_with_score(query, k=3)
        results[model_name] = [
            {"text": doc.page_content[:80], "score": round(score, 4)}
            for doc, score in retrieved
        ]
    return results

# Test documents
docs = [
    "Return policy: 30 days from purchase, original condition required.",
    "Shipping: Free for orders over $50, 3-5 business days delivery.",
    "Warranty: 1-year manufacturer warranty for all electronics.",
    "International shipping available to 50+ countries with import duties.",
    "Refund process: 5-7 business days after return received.",
    "Product exchange: Available within 14 days for different size/color."
]

# Compare
results = compare_embeddings(
    query="How do I get my money back?",
    docs=docs,
    models=["nomic-embed-text", "all-minilm"]
)

for model, hits in results.items():
    print(f"\n{model}:")
    for hit in hits:
        print(f"  [{hit['score']}] {hit['text']}")

# Function defined successfully

import ollama
import json
from langchain_ollama import ChatOllama, OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

class SmartRouter:
    def __init__(self, chroma_path: str = "./chroma_kb"):
        self.classifier = ChatOllama(model="llama3.2:3b", temperature=0)
        self.small_model = ChatOllama(model="llama3.2:3b", temperature=0.3)
        self.large_model = ChatOllama(model="qwen2.5", temperature=0.4)
        self.embeddings = OllamaEmbeddings(model="nomic-embed-text")

        try:
            self.vectorstore = Chroma(
                persist_directory=chroma_path,
                embedding_function=self.embeddings
            )
            self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3})
        except Exception:
            self.retriever = None

    def classify(self, question: str) -> str:
        response = self.classifier.invoke(
            f"Classify as one word: faq, product, comparison, complaint, other.\nQuestion: {question}"
        )
        return response.content.strip().lower()

    def answer(self, question: str) -> dict:
        intent = self.classify(question)

        if intent in ("faq", "other") and self.retriever:
            # RAG pipeline for FAQ
            docs = self.retriever.invoke(question)
            context = "\n\n".join(d.page_content for d in docs)
            response = self.small_model.invoke(
                f"Context:\n{context}\n\nAnswer based on context: {question}"
            )
            model_used = "3B + RAG"
        elif intent == "complaint":
            response = self.large_model.invoke(
                f"You are an empathetic customer service agent. "
                f"Show understanding and offer solutions:\n{question}"
            )
            model_used = "8B (empathetic)"
        else:
            response = self.large_model.invoke(question)
            model_used = "8B"

        return {
            "intent": intent,
            "model": model_used,
            "answer": response.content
        }

# Usage
router = SmartRouter()
for q in ["What is the return policy?", "Compare 3 headphone models", "My order arrived damaged!"]:
    result = router.answer(q)
    print(f"[{result['intent']}|{result['model']}] {result['answer'][:80]}...")

# Function defined successfully

import ollama
import time

def quantization_benchmark(test_questions: list[str], models: list[str]):
    """Compare response quality across quantizations."""
    results = {}
    for model in models:
        model_results = []
        # Warm up
        ollama.chat(model=model, messages=[{"role": "user", "content": "hi"}],
                    stream=False)

        for q in test_questions:
            start = time.time()
            resp = ollama.chat(
                model=model,
                messages=[{"role": "user", "content": q}],
                stream=False,
                options={"temperature": 0.3}
            )
            elapsed = time.time() - start
            model_results.append({
                "question": q,
                "answer": resp["message"]["content"][:200],
                "latency_s": round(elapsed, 2)
            })
        results[model] = model_results

    return results

# Run if models are available
# questions = [
#     "What is the return policy for electronics?",
#     "Translate to French: Free shipping on orders over $50",
#     "Write a SQL query for top 5 customers"
# ]
# results = quantization_benchmark(questions, ["llama3.2:3b", "qwen2.5"])

# Function defined successfully

# ============================================
# Comprehensive: SupportBot final architecture
# Integrates all Phase 3 capabilities
# ============================================

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

@dataclass
class SupportBotArchitecture:
    """Final SupportBot architecture specification."""

    name: str = "SupportBot V4"
    version: str = "4.0"

    components: dict = field(default_factory=lambda: {
        "intent_classifier": {
            "model": "llama3.2:3b",
            "quantization": "Q4_K_M",
            "purpose": "Classify customer intent in <1s",
            "vram": "~2GB"
        },
        "faq_rag": {
            "model": "llama3.2:3b",
            "embedding": "nomic-embed-text",
            "vector_db": "Chroma",
            "purpose": "Answer FAQ from knowledge base",
            "vram": "~2.3GB shared"
        },
        "deep_answer": {
            "model": "qwen2.5:7b",
            "quantization": "Q4_K_M",
            "purpose": "Complex product questions and comparisons",
            "vram": "~5GB"
        },
        "complaint_handler": {
            "model": "qwen2.5:7b",
            "system_prompt": "empathetic mode",
            "purpose": "Handle complaints with empathy",
            "vram": "shared with deep_answer"
        },
        "multimodal": {
            "model": "llava",
            "purpose": "Damage report image analysis",
            "vram": "~5GB (on-demand)"
        }
    })

    deployment: dict = field(default_factory=lambda: {
        "platform": "Docker Compose",
        "services": ["ollama", "chroma", "fastapi", "nginx"],
        "vram_total": "8GB minimum, 12GB recommended",
        "persistence": "Named volumes for models and Chroma"
    })

    routing_rules: dict = field(default_factory=lambda: {
        "faq": "faq_rag (3B + RAG)",
        "order_status": "faq_rag (3B + RAG)",
        "shipping": "faq_rag (3B + RAG)",
        "product_info": "deep_answer (8B)",
        "comparison": "deep_answer (8B)",
        "complaint": "complaint_handler (8B empathetic)",
        "damage_report": "multimodal (llava)"
    })

    cost_analysis: dict = field(default_factory=lambda: {
        "before_gpt4": "2,000 USD/month",
        "after_local": "98 USD/month (hardware amortization + electricity)",
        "saving": "1,902 USD/month (95% reduction)",
        "payback_months": 1.1
    })

    def to_report(self) -> str:
        """Generate architecture report."""
        report = [f"# {self.name} v{self.version} Architecture\n"]
        report.append("## Components")
        for name, spec in self.components.items():
            report.append(f"- {name}: {spec['model']} - {spec['purpose']}")
        report.append("\n## Deployment")
        report.append(f"- Platform: {self.deployment['platform']}")
        report.append(f"- Services: {', '.join(self.deployment['services'])}")
        report.append(f"- VRAM: {self.deployment['vram_total']}")
        report.append("\n## Routing Rules")
        for intent, target in self.routing_rules.items():
            report.append(f"- {intent} → {target}")
        report.append("\n## Cost Analysis")
        for key, value in self.cost_analysis.items():
            report.append(f"- {key}: {value}")
        return "\n".join(report)

# Generate report
arch = SupportBotArchitecture()
print(arch.to_report())