Projektowanie systemów wielomodelowych: kiedy jeden model to za mało
Wybierz najprostszy działający wzorzec.
Systemy oparte na jednym modelu są proste. Systemy wielomodelowe są potężne. Wyzwanie nie polega na wyborze modeli – chodzi o zaprojektowanie architektury, która je koordynuje.
System wielomodelowy nie oznacza posiadania większej liczby modeli. Oznacza to posiadanie odpowiedniego modelu do odpowiedniego zadania w odpowiednim czasie.
Wzorce architektoniczne
Pięć wzorców pokrywa większość przypadków użycia:
| Wzorzec | Złożoność | Kiedy stosować | Zależności |
|---|---|---|---|
| Pojedynczy model | Najniższa | Prototypowanie, proste zadania | Ograniczone możliwości |
| Sequentny | Niska | Procesy wieloetapowe | Wyższa latencja |
| Równoległy | Średnia | Zadania niezależne | Wyższy koszt |
| Hierarchiczny | Wysoka | Skomplikowane wnioskowanie | Skomplikowana orkiestracja |
| Ensemble (Zespół) | Najwyższa | Krytyczne decyzje | Najwyższy koszt |
Wybierz najprostszy z tych, które działają. Złożoność jest realna i się kumuluje.
Architektura sekwencyjna
Przetwarzaj zadania przez łańcuch modeli, z których każdy specjalizuje się w innym etapie.
Wzorzec 1: Rurociąg (Pipeline)
Wzorzec rurociąg – wyjście każdego modelu jest wejściem dla następnego:
class ModelPipeline:
def __init__(self):
self.models = [
{"model": "qwen3-1.7b", "task": "classify"},
{"model": "qwen3-8b", "task": "extract"},
{"model": "qwen3-32b", "task": "reason"},
]
def process(self, input: str) -> str:
current = input
for model_config in self.models:
current = self.call_model(
model_config["model"],
self.create_prompt(model_config["task"], current)
)
return current
Latencje się sumują. Trzy modele w sekwencji oznaczają trzykrotnie większą latencję. Stosuj to tylko wtedy, gdy każdy etap faktycznie wymaga innego modelu.
Wzorzec 2: Router (Przekierowujący)
Wzorzec router – sklasyfikuj zadanie, przekieruj do specjalisty:
class ModelRouter:
def __init__(self):
self.classifier = "qwen3-1.7b"
self.specialists = {
"code": "qwen2.5-coder-7b",
"math": "qwen3-32b",
"creative": "claude-sonnet-4",
"general": "qwen3-8b",
}
def route(self, prompt: str) -> str:
task_type = self.classify(prompt)
model = self.specialists.get(task_type, self.specialists["general"])
return self.call_model(model, prompt)
Klasyfikator jest ogniwaem najsłabszym. Jeśli błędnie sklasyfikuje, przekierujesz do niewłaściwego modelu i stracisz na jakości. Użyj klasyfikatora, który jest wystarczająco dobry – nawet mały model zadziała, jeśli kategorie są jasno zdefiniowane.
Architektura równoległa
Przetwarzaj niezależne zadania jednocześnie.
Wzorzec 1: Rozwielianie (Fan-Out)
Rozwielianie – uruchom ten sam prompt przez wiele modeli:
import asyncio
class ModelFanOut:
def __init__(self):
self.models = [
"qwen3-8b",
"qwen3-32b",
"claude-sonnet-4",
]
async def process(self, prompt: str) -> list[str]:
tasks = [self.call_model(model, prompt) for model in self.models]
return await asyncio.gather(*tasks)
Przydatne do porównań, testów A/B lub gdy chcesz wybrać najlepszy wynik. Kosztowne, ale zysk na jakości jest wart tego w przypadku krytycznych decyzji.
Wzorzec 2: Głosowanie (Voting)
Głosowanie – łączenie wyników poprzez konsensus:
class ModelVoting:
def __init__(self):
self.models = [
"qwen3-8b",
"qwen3-32b",
"claude-sonnet-4",
]
def vote(self, prompt: str) -> str:
responses = [self.call_model(model, prompt) for model in self.models]
from collections import Counter
votes = Counter(responses)
return votes.most_common(1)[0][0]
Głosowanie większościowe działa przy klasyfikacji. Przy zadaniach generowania jest trudniej – potrzebujesz podobieństwa semantycznego, a nie dokładnych dopasowań.
Architektura hierarchiczna
Używaj modeli na różnych poziomach abstrakcji.
Wzorzec 1: Planista-Executor (Planista-Wykonawca)
Planista-wykonawca – silny model planuje, mniejsze modele wykonują:
class PlannerExecutor:
def __init__(self):
self.planner = "qwen3-32b"
self.executors = {
"code": "qwen2.5-coder-7b",
"search": "qwen3-8b",
"math": "qwen3-8b",
}
def process(self, task: str) -> str:
plan = self.call_model(self.planner, f"Plan: {task}")
results = []
for step in self.parse_plan(plan):
executor = self.executors.get(step["type"], "qwen3-8b")
result = self.call_model(executor, step["prompt"])
results.append(result)
return self.call_model(self.planner, f"Synthesize: {results}")
Planista wykonuje ciężką pracę. Wykonawcy obsługują konkretne zadania. Ten wzorzec działa dobrze, gdy etap planowania jest kosztowny, a etapy wykonania są tanie.
Wzorzec 2: Nadzorujący-Pracownik (Supervisor-Worker)
Nadzorujący-pracownik – nadzorujący deleguje i przegląda:
class SupervisorWorker:
def __init__(self):
self.supervisor = "qwen3-32b"
self.workers = ["qwen3-8b", "qwen2.5-coder-7b"]
def process(self, task: str) -> str:
assignments = self.call_model(self.supervisor, f"Assign: {task}")
results = []
for assignment in self.parse_assignments(assignments):
result = self.call_model(
assignment["worker"], assignment["task"]
)
results.append(result)
return self.call_model(self.supervisor, f"Review: {results}")
Nadzorujący jest wąskim gardłem. Planuje, deleguje i przegląda. Upewnij się, że jest wystarczająco szybki, inaczej cały system zwolni.
Architektura Ensemble (Zespołowa)
Łącz wiele modeli dla krytycznych decyzji.
Wzorzec 1: Ensemble ważony
Ensemble ważony – oceń wyjście każdego modelu, wybierz najwyższe:
class WeightedEnsemble:
def __init__(self):
self.models = {
"qwen3-32b": 0.5,
"claude-sonnet-4": 0.3,
"qwen3-8b": 0.2,
}
def decide(self, prompt: str) -> str:
responses = {
model: self.call_model(model, prompt)
for model in self.models
}
scores = {}
for model, response in responses.items():
score = self.evaluate(response) * self.models[model]
scores[response] = scores.get(response, 0) + score
return max(scores, key=scores.get)
Wagi odzwierciedlają Twoje zaufanie do każdego modelu. Dostosuj je na podstawie rzeczywistej wydajności, a nie benchmarków.
Wzorzec 2: Ensemble konsensusowy
Ensemble konsensusowy – wymóg zgody, eskalacja w przypadku braku zgody:
class ConsensusEnsemble:
def __init__(self, threshold: float = 0.7):
self.threshold = threshold
self.models = [
"qwen3-32b",
"claude-sonnet-4",
"qwen3-8b",
]
def decide(self, prompt: str) -> str:
responses = [
self.call_model(model, prompt)
for model in self.models
]
from collections import Counter
votes = Counter(responses)
max_votes = max(votes.values())
if max_votes / len(self.models) >= self.threshold:
return votes.most_common(1)[0][0]
return self.call_model("qwen3-32b", prompt)
Próg kontroluje, jak rygorystyczny musi być konsensus. 0.7 oznacza zgodę dwóch trzecich. Obniż go dla szybszych decyzji, podwyższ dla większego zaufania.
Kiedy systemy wielomodelowe mają sens
Systemy wielomodelowe mają sens, gdy masz mieszane obciążenia, potrzebujesz wysokiej jakości dla krytycznych decyzji lub optymalizujesz pod kątem kosztu lub latencji.
Nie mają sensu, gdy wszystkie zadania mają podobną złożoność, prototypujesz lub gdy prostota ma większe znaczenie niż optymalizacja.
Zasada kciuka: zacznij od jednego modelu. Dodawaj więcej, gdy natkniesz się na realne ograniczenie – koszt, latencję lub jakość. Nie projektuj złożoności, dopóki jej nie potrzebujesz.
Zależności (Tradeoffs)
| Wzorzec | Koszt | Latencja | Jakość | Złożoność |
|---|---|---|---|---|
| Pojedynczy model | Najniższy | Najniższa | Zmienna | Najniższa |
| Sequentny | Średni | Wysoka | Wysoka | Średnia |
| Równoległy | Wysoki | Niska | Wysoka | Średnia |
| Hierarchiczny | Wysoki | Wysoka | Najwyższa | Wysoka |
| Ensemble | Najwyższy | Średnia | Najwyższa | Najwyższa |
Każdy wzorzec coś poświęca. Wybierz ten, który pasuje do Twoich ograniczeń.
Powiązane
- Strategie routingu modeli — routowanie oparte na możliwościach, kosztach i latencji
- Optymalizacja kosztów dla systemów LLM — budżetowanie tokenów, modele zapasowe, buforowanie
- Barierki LLM w praktyce — walidacja wejścia, filtrowanie wyjścia, bezpieczeństwo
- Architektura LLM — filar projektowania systemów: routowanie, koszt, bariery i orkiestracja
