Ollama: Orquestração Multi-Modelo

Orquestração multi-modelo é trabalho em equipe de IA — modelos pequenos lideram a carga, modelos grandes dão respaldo, cada um no seu papel.

💡 Dica: O núcleo da estratégia de roteamento multi-modelo é "atribuir por tarefa" — FAQs simples usam um modelo pequeno 3B (resposta em 1 segundo, baixo consumo de GPU), raciocínio complexo usa um modelo grande 8B (resposta em 5 segundos, alta precisão). Você também pode "atribuir por carga" — usar modelos pequenos durante picos para lidar com mais requisições, e mudar para modelos grandes durante períodos de baixo tráfego para maior qualidade. Rotear 80% das requisições para modelos pequenos pode reduzir a latência média em 50%.

📋 Pré-requisitos: Você deve dominar o seguinte primeiro

Aula 13: Integração com LangChain

1. O Que Você Vai Aprender

Padrões de arquitetura multi-modelo: Router / Cascade / Parallel Voting
Padrão Router: modelo pequeno classifica → modelo grande responde em profundidade
Padrão Cascade: rascunho → refinamento → pipeline de formatação
asyncio Python para chamadas concorrentes a múltiplas instâncias Ollama
SupportBot V4 da Alice: divisão de tráfego baseada em Router

2. Uma História Real de Uma Empreendedora SaaS

ℹ️ Info: Na orquestração multi-modelo, cada modelo precisa ser independentemente carregado na VRAM. Dois modelos (3B + 8B) residindo simultaneamente requerem cerca de 10GB VRAM. Se VRAM for insuficiente, o Ollama automaticamente descarrega modelos inativos, mas a troca incorre em um atraso de recarregamento de 5-30 segundos.

(1) O Problema: Modelos Grandes Desperdiçam Recursos em Perguntas Simples

Alice descobriu que 80% das perguntas do SupportBot ("Qual é a política de devolução?" "Como verificar o frete?") poderiam ser respondidas por um modelo pequeno, mas rotear tudo por um modelo grande desperdiça recursos de GPU e é mais lento.

(2) A Solução: Padrão Router para Divisão de Tráfego

Um modelo pequeno (3B) classifica a intenção em 1 segundo; perguntas simples recebem respostas diretas, perguntas complexas são roteadas para um modelo grande (8B):

PYTHON

# 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

3. Três Padrões de Orquestração

⚠️ Nota: O padrão Parallel Voting tem a maior confiabilidade mas também o maior custo — cada requisição requer chamar 3-5 modelos, multiplicando a computação de GPU. Apenas use para cenários de decisão crítica (por exemplo, conselho médico, julgamento legal); o padrão Router é suficiente para atendimento ao cliente diário.

💡 Dica: O padrão Router é o método de orquestração mais custo-benefício — 80% das requisições simples usam um modelo 3B (resposta em 1 segundo), e apenas 20% das requisições complexas são roteadas para um modelo 8B. Latência média geral cai de 5 segundos para 2 segundos, com 60% menos utilização de GPU.

(1) Comparação de Padrões

Padrão	Fluxo	Caso de Uso	Latência	Custo
Router	Classificar → Despachar	Roteamento de atendimento, classificação de tarefas	Baixa	Baixo
Cascade	Rascunho → Refinar → Formatar	Geração de código, criação de conteúdo	Média	Médio
Parallel Voting	Multi-modelo → Votar/Mesclar	Decisões de alta confiabilidade	Alta	Alto

flowchart TD
    subgraph Router
        R1[Entrada] --> R2[Classificador<br/>Modelo 3B]
        R2 -->|Simples| R3[Resposta Modelo Pequeno]
        R2 -->|Complexo| R4[Resposta Modelo Grande]
    end

    subgraph Cascade
        C1[Entrada] --> C2[Rascunho<br/>Modelo 3B]
        C2 --> C3[Refinar<br/>Modelo 8B]
        C3 --> C4[Formatar<br/>Modelo 3B]
    end

    subgraph Voting
        V1[Entrada] --> V2[Modelo A]
        V1 --> V3[Modelo B]
        V1 --> V4[Modelo C]
        V2 --> V5[Votação Majoritária]
        V3 --> V5
        V4 --> V5
    end

4. Padrão Router

(1) Arquitetura Router

sequenceDiagram
    participant U as Usuário
    participant C as Classificador (3B)
    participant S as Modelo Pequeno (3B)
    participant L as Modelo Grande (8B)
    U->>C: "Quero um reembolso"
    C-->>C: Intenção: reembolso (simples)
    C->>S: Roteado para modelo pequeno
    S-->>U: Resposta da política de reembolso
    U->>C: "Compare termos de garantia de 3 produtos"
    C-->>C: Intenção: comparação (complexo)
    C->>L: Roteado para modelo grande
    L-->>U: Comparação detalhada

(2) Design de Rótulos de Classificação de Intenção

Rótulo	Complexidade	Alvo de Roteamento	Exemplo de Pergunta
faq	Simples	Modelo pequeno	"Qual é a política de devolução?"
order_status	Simples	Modelo pequeno	"Onde está o pedido #12345?"
product_info	Médio	Modelo médio	"Este fone suporta Bluetooth 5.3?"
comparison	Complexo	Modelo grande	"Compare os 3 modelos de fone"
complaint	Complexo	Modelo grande	"O item que recebi é completamente diferente da descrição"

▶ Exemplo 1: Implementação do Padrão Router

PYTHON

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")

Saída：

TEXT

# Function defined successfully

5. Padrão Cascade

(1) Detalhes da Arquitetura Cascade

Estágio	Modelo	Tarefa	Parâmetros
Rascunho	3B	Rascunho rápido	temperature=0.5
Refinar	8B	Refinamento profundo	temperature=0.3
Formatar	3B	Formatar saída	temperature=0.1

▶ Exemplo 2: Geração de Conteúdo em Cascade

PYTHON

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"])

Saída：

TEXT

=== Final Output ===

6. Parallel Voting e Concorrência asyncio

(1) Arquitetura Parallel Voting

Padrão	Contagem de Modelos	Método de Decisão	Caso de Uso
Votação majoritária	3+	Tomar resposta de consenso majoritário	Julgamento factual
Média ponderada	2+	Ponderar pela qualidade do modelo	Tarefas de pontuação
Melhor seleção	2+	Escolher mais detalhada/crível	Perguntas abertas

▶ Exemplo 3: Invocação Concorrente com asyncio

PYTHON

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())

Saída：

TEXT

# Function defined successfully

▶ Exemplo 4: Ensemble Ponderado

PYTHON

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)

Saída：

TEXT

# Function defined successfully

7. Exemplo Abrangente: Sistema Router do SupportBot V4

PYTHON

# ============================================
# 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)")

❓ Perguntas Frequentes

P: E se a classificação do router for imprecisa? R: Use um classificador melhor. Modelos pequenos (3B) têm cerca de 85% de precisão de classificação, modelos 8B cerca de 92%. Você também pode pré-filtrar com correspondência de palavras-chave, deixando o modelo lidar apenas com casos ambíguos.

P: O padrão cascade é melhor que usar um modelo grande diretamente? R: Depende do cenário. Latência total do cascade = soma de todos os estágios, mas cada estágio é mais rápido. Adequado para cenários que precisam de rascunho + refinamento (escrita, código). Q&A simples não precisa de cascata.

P: O voting paralelo consome 3x recursos de GPU? R: Sim. 3 modelos rodando Inferência simultaneamente precisam de 3x VRAM. Você pode chamar 3 modelos sequencialmente para evitar OOM, mas perde a vantagem paralela.

P: Há grande diferença de velocidade entre concorrência asyncio e chamadas sequenciais? R: Depende da GPU. Com uma única GPU, o Ollama enfileira internamente, então concorrência não ajuda. Com múltiplas GPUs ou mistura CPU+GPU, asyncio pode rodar modelos diferentes concorrentemente.

P: Como avalio a eficácia da divisão de tráfego do router? R: Rastreie a proporção de roteamento, latência por categoria e satisfação do usuário. Meta: 80% das requisições para modelos pequenos, 50%+ de redução de latência média, sem queda na satisfação.

P: Modelos pequenos e grandes devem usar o mesmo System Prompt? R: Recomendado manter consistente para uma experiência de usuário unificada. Mas você pode ajustar para diferentes níveis de complexidade — o Prompt do modelo grande pode ser mais detalhado.

📖 Resumo

Três padrões de orquestração: Router (divisão de tráfego), Cascade (pipeline), Parallel Voting (alta confiabilidade)
Padrão Router: classificação 3B + respostas profundas 8B, 80% das requisições para modelo pequeno, latência média reduzida em 50%
Padrão Cascade: Rascunho(3B) → Refinar(8B) → Formatar(3B), adequado para criação de conteúdo
asyncio para chamadas multi-modelo concorrentes; note enfileiramento em GPU única
SupportBot V4 usa divisão baseada em router: 2s para perguntas simples, 5s para complexas
Avalie eficácia do router por proporção de roteamento, melhoria de latência e satisfação do usuário

📝 Exercícios

Básico (⭐): Implemente um router simples de duas vias — FAQ vai para modelo pequeno, todo o resto para modelo grande. Teste com 5 perguntas.
Intermediário (⭐⭐): Implemente um pipeline cascade de três estágios — rascunho → refinar → formatar. Compare qualidade com saída direta de modelo único.
Avançado (⭐⭐⭐): Implemente um sistema router completo para o SupportBot V4, incluindo classificação de intenção, estatísticas de roteamento, monitoramento de latência, e produza um relatório de eficácia de roteamento.

Equipe Técnica Web-Tutorial

Uma plataforma de tutoriais mantida por diversos desenvolvedores. Cada tutorial é escrito e revisado por profissionais da área correspondente. Trabalhamos para manter nosso conteúdo preciso e confiável — se encontrar algum problema, avise-nos.

Sobre Nós