Архитектура систем с множественными моделями: когда одной модели недостаточно
Выбирайте самый простой работающий вариант
Системы на основе одной модели просты. Многомодельные системы — мощны. Сложность заключается не в выборе моделей, а в проектировании архитектуры, которая их оркеструет.
Многомодельная система — это не просто наличие большего количества моделей. Это использование правильной модели для правильной задачи в правильное время.
Паттерны архитектуры
Пять паттернов покрывают большинство случаев использования:
| Паттерн | Сложность | Когда использовать | Компромисс |
|---|---|---|---|
| Одна модель | Наименьшая | Прототипирование, простые задачи | Ограниченные возможности |
| Последовательный | Низкая | Многоэтапные рабочие процессы | Высокая задержка |
| Параллельный | Средняя | Независимые задачи | Высокая стоимость |
| Иерархический | Высокая | Сложное рассуждение | Сложная оркестрация |
| Ансамблевый | Наивысшая | Критические решения | Наивысшая стоимость |
Выбирайте самый простой работающий вариант. Сложность реальна, и она накапливается.
Последовательная архитектура
Обработка задач через цепочку моделей, каждая из которых специализируется на определенном шаге.
Паттерн 1: Конвейер
Паттерн конвейера — вывод каждой модели поступает на вход следующей:
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
Задержка суммируется. Три модели в последовательности означают утроение задержки. Используйте этот подход только тогда, когда каждый шаг действительно требует другой модели.
Паттерн 2: Маршрутизатор
Паттерн маршрутизации — классифицируйте задачу и направьте её к специалисту:
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)
Классификатор — самое слабое звено. Если он ошибается в классификации, вы направляете запрос к неправильной модели и теряете качество. Используйте классификатор, который достаточно хорош — даже небольшая модель работает, если категории четкие.
Параллельная архитектура
Обработка независимых задач одновременно.
Паттерн 1: Веерное распространение (Fan-Out)
Веерное распространение — пропустите один и тот же промпт через несколько моделей:
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)
Полезно для сравнения, A/B тестирования или когда вы хотите выбрать лучший результат. Дорого, но прирост качества того стоит для критических решений.
Паттерн 2: Голосование
Голосование — объединение выводов через консенсус:
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]
Большинство голосов работает для задач классификации. Для задач генерации это сложнее — вам нужна семантическая сходность, а не точное совпадение.
Иерархическая архитектура
Использование моделей на разных уровнях абстракции.
Паттерн 1: Планировщик-Исполнитель
Планировщик-исполнитель — сильная модель планирует, а более мелкие модели выполняют:
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}")
Планировщик выполняет основную работу. Исполнители обрабатывают конкретные задачи. Этот паттерн хорошо работает, когда этап планирования дорог, а этапы выполнения — дешевы.
Паттерн 2: Супервайзер-Рабочий
Супервайзер-рабочий — супервайзер делегирует и проверяет:
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}")
Супервайзер — это узкое место. Он планирует, делегирует и проверяет. Убедитесь, что он достаточно быстр, иначе вся система замедлится.
Ансамблевая архитектура
Объединение нескольких моделей для критических решений.
Паттерн 1: Взвешенный ансамбль
Взвешенный ансамбль — оцените вывод каждой модели и выберите лучший:
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)
Веса отражают вашу уверенность в каждой модели. Настраивайте их на основе фактической производительности, а не бенчмарков.
Паттерн 2: Ансамбль консенсуса
Ансамбль консенсуса — требуйте согласия, эскалируйте, если его нет:
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)
Порог контролирует строгость консенсуса. 0.7 означает согласие двух третей. Понизьте его для более быстрых решений, повысьте для большей уверенности.
Когда многомодельные системы имеют смысл
Многомодельные системы имеют смысл, когда у вас смешанные нагрузки, требуется высокое качество для критических решений или вы оптимизируете по стоимости или задержке.
Они не имеют смысла, когда все задачи имеют схожую сложность, вы занимаетесь прототипированием или простота важнее оптимизации.
Практическое правило: начинайте с одной модели. Добавляйте больше, когда столкнетесь с реальной проблемой — стоимостью, задержкой или качеством. Не проектируйте сложность, пока она вам не нужна.
Компромиссы
| Паттерн | Стоимость | Задержка | Качество | Сложность |
|---|---|---|---|---|
| Одна модель | Наименьшая | Наименьшая | Переменная | Наименьшая |
| Последовательный | Средняя | Высокая | Высокое | Средняя |
| Параллельный | Высокая | Низкая | Высокое | Средняя |
| Иерархический | Высокая | Высокая | Наивысшее | Высокая |
| Ансамблевый | Наивысшая | Средняя | Наивысшее | Наивысшая |
Каждый паттерн предлагает свой компромисс. Выберите тот, который соответствует вашим ограничениям.
См. также
- Стратегии маршрутизации моделей — маршрутизация на основе возможностей, стоимости и задержки
- Оптимизация затрат для LLM-систем — бюджетирование токенов, резервные модели, кэширование
- Ограничения LLM на практике — валидация ввода, фильтрация вывода, безопасность
- Архитектура LLM — столп系统设计: маршрутизация, стоимость, ограничения и оркестрация
