Projektowanie systemów wielomodelowych: Kiedy jeden model to za mało
Wybierz najprostszy wzorzec, który działa.
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 | Skomplikowanie | Kiedy stosować | Kompromis |
|---|---|---|---|
| Pojedynczy model | Najniższy | Prototypowanie, proste zadania | Ograniczone możliwości |
| Sekwencyjny | Niski | Procesy wieloetapowe | Wyższa opóźnienie |
| Równoległy | Średni | Zadania niezależne | Wyższy koszt |
| Hierarchiczny | Wysoki | Skomplikowane wnioskowanie | Skomplikowana koordynacja |
| Ensemble | Najwyższy | Krytyczne decyzje | Najwyższy koszt |
Wybierz najprostszy z tych, które działają. Skomplikowanie jest realne i się kumuluje.
Architektura sekwencyjna
Przetwarzaj zadania poprzez łańcuch modeli, z których każdy specjalizuje się w określonym etapie.
Wzorzec 1: Rurociąg
Wzorzec rurociągu — wyjście każdego modelu podaje się jako wejście dla następnego:
class ModelPipeline:
def __init__(self):
self.models = [
{"model": "qwen2.5-1.5b", "task": "classify"},
{"model": "qwen2.5-7b", "task": "extract"},
{"model": "qwen2.5-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
Opóźnienia sumują się. Trzy modele w sekwencji oznaczają trzykrotnie dłuższe opóźnienie. Stosuj to tylko wtedy, gdy każdy etap naprawdę wymaga innego modelu.
Wzorzec 2: Router
Wzorzec routera — sklasyfikuj zadanie, przekieruj do specjalisty:
class ModelRouter:
def __init__(self):
self.classifier = "qwen2.5-1.5b"
self.specialists = {
"code": "qwen2.5-coder-7b",
"math": "qwen2.5-32b",
"creative": "claude-sonnet-4",
"general": "qwen2.5-7b",
}
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 najsłabszym ogniwem. Jeśli popełni błąd w klasyfikacji, prześlesz zapytanie 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: Rozszerzanie (Fan-Out)
Rozszerzanie — uruchom ten sam prompt w wielu modelach:
import asyncio
class ModelFanOut:
def __init__(self):
self.models = [
"qwen2.5-7b",
"qwen2.5-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. Jest drogie, ale wzrost jakości jest warty tego dla krytycznych decyzji.
Wzorzec 2: Głosowanie
Głosowanie — łączenie wyników poprzez konsensus:
class ModelVoting:
def __init__(self):
self.models = [
"qwen2.5-7b",
"qwen2.5-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 w przypadku klasyfikacji. W zadaniach generatywnych jest trudniej – potrzebujesz podobieństwa semantycznego, a nie dokładnych dopasowań.
Architektura hierarchiczna
Użyj modeli na różnych poziomach abstrakcji.
Wzorzec 1: Planista-Executor
Planista-Executor — silny model planuje, mniejsze modele wykonują:
class PlannerExecutor:
def __init__(self):
self.planner = "qwen2.5-32b"
self.executors = {
"code": "qwen2.5-coder-7b",
"search": "qwen2.5-7b",
"math": "qwen2.5-7b",
}
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"], "qwen2.5-7b")
result = self.call_model(executor, step["prompt"])
results.append(result)
return self.call_model(self.planner, f"Synthesize: {results}")
Planista wykonuje ciężką pracę. Executorzy zajmują się konkretnymi zadaniami. Ten wzorzec działa dobrze, gdy etap planowania jest kosztowny, a etapy wykonania są tanie.
Wzorzec 2: Nadzorujący-Pracownik
Nadzorujący-Pracownik – nadzorujący deleguje zadania i recenzuje:
class SupervisorWorker:
def __init__(self):
self.supervisor = "qwen2.5-32b"
self.workers = ["qwen2.5-7b", "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 recenzuje. Upewnij się, że jest wystarczająco szybki, w przeciwnym razie cały system zwolni.
Architektura Ensemble
Połącz wiele modeli dla krytycznych decyzji.
Wzorzec 1: Ważony Ensemble
Ważony ensemble — oceń wyjście każdego modelu, wybierz najwyższe:
class WeightedEnsemble:
def __init__(self):
self.models = {
"qwen2.5-32b": 0.5,
"claude-sonnet-4": 0.3,
"qwen2.5-7b": 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: Konsensusowy Ensemble
Konsensusowy ensemble – wymóg zgody, eskalacja w przypadku braku zgody:
class ConsensusEnsemble:
def __init__(self, threshold: float = 0.7):
self.threshold = threshold
self.models = [
"qwen2.5-32b",
"claude-sonnet-4",
"qwen2.5-7b",
]
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("qwen2.5-32b", prompt)
Próg kontroluje, jak surowy musi być konsensus. 0.7 oznacza zgodność 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 opóźnienia.
Nie mają sensu, gdy wszystkie zadania mają podobny poziom skomplikowania, prototypujesz lub prostota jest ważniejsza niż optymalizacja.
Zasada kciuka: zacznij od jednego modelu. Dodaj więcej, gdy natkniesz się na realne ograniczenie – koszt, opóźnienie lub jakość. Nie projektuj skomplikowania, dopóki go nie potrzebujesz.
Kompromisy
| Wzorzec | Koszt | Opóźnienie | Jakość | Skomplikowanie |
|---|---|---|---|---|
| Pojedynczy model | Najniższy | Najniższy | Zmienna | Najniższy |
| Sekwencyjny | Średni | Wysoki | Wysoka | Średni |
| Równoległy | Wysoki | Niski | Wysoka | Średni |
| Hierarchiczny | Wysoki | Wysoki | Najwyższa | Wysoki |
| Ensemble | Najwyższy | Średni | Najwyższa | Najwyższy |
Każdy wzorzec wymaga kompromisu. Wybierz ten, który pasuje do Twoich ograniczeń.
Powiązane
- Strategie routingu modeli — routing oparty na możliwościach, wrażliwy na koszty i opóźnienia
- Optymalizacja kosztów dla systemów LLM — budżetowanie tokenów, modele zapasowe, cache
- Barykady LLM w praktyce — walidacja wejścia, filtrowanie wyjścia, bezpieczeństwo
- Architektura LLM — filar projektowania systemów: routing, koszty, barykady i koordynacja
