Пул-реквест называется «Add GPT-5.5 fallback». Дифф: 340 строк вместо двадцати — скопированные декораторы ретраев, захардкоженные строки моделей, пять почти идентичных вариантов call_llm_with_retry.
Вы в ревью: «Здесь нужна абстракция». Автор: «Какая именно абстракция?»
Эта статья отвечает на этот вопрос.
Шесть паттернов банды четырёх (Gang of Four), перенесённых в область LLM API: Factory для выбора модели, Strategy для промптов, Observer для стриминга, Decorator для ретраев и логирования, Chain of Responsibility для фолбэков, Template Method для циклов агентов — каждый с продакшн-кодом на Python и анти-паттерном, который он заменяет.
Паттерн 1: Factory — централизованная инстанциация моделей
Проблема
"gpt-5.5" в chat.py. "claude-sonnet-4-20250514" в summarizer.py. "deepseek-v4-flash" в classifier.py. Миграция модели означает поиск и замену по всей кодовой базе — и надежду, что вы не пропустили одно вхождение в конфиг-файле, который загружается только в продакшне.
Паттерн
ModelFactory с централизованным реестром. Типы задач, а не захардкоженные строки, определяют, какая модель используется. Переменные окружения обеспечивают канареечные развёртывания и откаты. Метаданные модели — capabilities, cost tier, context window — живут рядом с ID модели. Реализация использует OpenAI Python SDK — стандартную клиентскую библиотеку для API, совместимых с OpenAI, — чей клиент AsyncOpenAI питает Factory ниже.
from dataclasses import dataclass
from openai import AsyncOpenAI
@dataclass
class ModelSpec:
model_id: str
provider: str
capabilities: list[str] # ["chat", "vision", "tools", "json_mode"]
cost_tier: str # "cheap", "mid", "frontier"
context_window: int
class ModelFactory:
def __init__(self, base_url: str, api_key: str):
self.client = AsyncOpenAI(base_url=base_url, api_key=api_key)
self.registry: dict[str, ModelSpec] = {}
self._load_registry()
def create(self, task_type: str, requirements: list[str] = None) -> tuple[AsyncOpenAI, ModelSpec]:
model_id = os.getenv(f"MODEL_OVERRIDE_{task_type.upper()}", None)
if model_id:
spec = self.registry[model_id]
else:
spec = self._select_by_capability(task_type, requirements or [])
return self.client, spec
def _select_by_capability(self, task_type: str, requirements: list[str]) -> ModelSpec:
candidates = [
m for m in self.registry.values()
if all(req in m.capabilities for req in requirements)
]
tier_map = {"classification": "cheap", "generation": "mid", "review": "frontier"}
tier = tier_map.get(task_type, "mid")
return next((m for m in candidates if m.cost_tier == tier), candidates[0])Единый API-endpoint — один base_url для всех провайдеров — сокращает конфиг Factory с O(N провайдеров) до O(1 endpoint + N строк моделей). Один API-ключ. Один экземпляр клиента. Каждая модель реестра доступна через него. Настройка этой архитектуры с единой точкой входа начинается с аутентификации по API-ключу — одного учётного данного, управляющего доступом к каждой модели вашего реестра и устраняющего расползание N ключей × M провайдеров.
Анти-паттерн, который он заменяет
Строки моделей, захардкоженные в каждой точке вызова. Депрекация модели запускает поиск и замену по всей кодовой базе — и первый признак того, что вы что-то пропустили, это ошибка 404 в продакшне.
Паттерн 2: Strategy — подключаемые шаблоны промптов
Проблема
Строки промптов, инлайнутые в бизнес-логику. Изменение тона чекаута означает поиск каждого "You are a helpful shopping assistant...", разбросанного по коду чекаута, поддержки и онбординга. A/B-тестирование двух вариантов промптов означает if/else-спагетти в каждой точке вызова.
Паттерн
Интерфейс PromptStrategy. Конкретные реализации для каждого сценария использования или варианта эксперимента. Выбор во время выполнения через фиче-флаг или корзину A/B-теста. Каждая стратегия — версионируемый артефакт; ваш реестр промптов сопоставляет версии с классами стратегий.
from abc import ABC, abstractmethod
class PromptStrategy(ABC):
version: str
@abstractmethod
def build_messages(self, context: dict) -> list[dict]:
"""Build the messages array for this prompt strategy."""
class CheckoutV3(PromptStrategy):
version = "checkout_v3.2"
def build_messages(self, context: dict) -> list[dict]:
return [
{"role": "system", "content": CHECKOUT_SYSTEM_V3},
{"role": "user", "content": f"<cart>{context['cart']}</cart>"}
]
class PromptRouter:
def __init__(self, strategies: dict[str, PromptStrategy]):
self.strategies = strategies
def select(self, feature_flags: dict, task: str) -> PromptStrategy:
variant = feature_flags.get(f"prompt_{task}", "default")
return self.strategies[variant]Когда вы A/B-тестируете промпт чекаута V3 против V4, вы переключаете фиче-флаг. Ноль изменений кода. Набор для оценки (см. наше руководство по тестированию) измеряет, какой вариант побеждает.
Анти-паттерн, который он заменяет
Строки промптов, разбросанные по бизнес-логике. Изменение тона означает поиск каждой скопированной копии. Без трассировки логов развёртывания нет способа узнать, какую версию промпта получил пользователь.
Паттерн 3: Observer — слабосвязанные стриминговые потребители
Проблема
В вашем стриминговом цикле переплетены синтез TTS, рендеринг UI-чанков, отслеживание стоимости и логирование. Добавление нового потребителя — аналитика, оверлей перевода, запись аудита — означает изменение основного цикла генерации. После трёх добавлений цикл достигает 200 строк, и никто не хочет его трогать.
Паттерн
Интерфейс StreamObserver. Конкретные наблюдатели для каждого потребителя. Генератор уведомляет наблюдателей — но не знает, что они делают. Слабая связанность. Наблюдатели могут быть добавлены, удалены или заменены независимо.
class StreamObserver(ABC):
@abstractmethod
async def on_token(self, token: str, sequence: int): ...
@abstractmethod
async def on_complete(self, full_response: str, usage: dict): ...
@abstractmethod
async def on_error(self, error: Exception): ...
class StreamObservable:
def __init__(self, client: AsyncOpenAI):
self.client = client
self.observers: list[StreamObserver] = []
def attach(self, observer: StreamObserver): self.observers.append(observer)
async def stream(self, **kwargs):
stream = await self.client.chat.completions.create(stream=True, **kwargs)
full_response = ""
try:
async for chunk in stream:
token = chunk.choices[0].delta.content or ""
full_response += token
await asyncio.gather(*[
o.on_token(token, len(full_response)) for o in self.observers
])
await asyncio.gather(*[
o.on_complete(full_response, usage) for o in self.observers
])
except Exception as e:
await asyncio.gather(*[o.on_error(e) for o in self.observers])Сбой одного наблюдателя не убивает стрим — ошибки изолированы по каждому наблюдателю. Добавьте наблюдателя CostTracker. Добавьте наблюдателя TTSOutput. Ни один не знает о существовании другого.
Анти-паттерн, который он заменяет
Вся логика стриминговых потребителей инлайнута в цикл генерации. Добавление аналитической инструментации требует редактирования той же функции, которая обрабатывает TTS, — с риском регрессии в аудиовыходе из-за опечатки в имени переменной.
Паттерн 4: Decorator — операционные слои без захламления
Проблема
10-строчный вызов LLM, окружённый 60 строками логики ретраев, отслеживания стоимости, структурированного логирования и обработки ошибок. Скопирован с немного разными параметрами в восьми точках вызова.
Паттерн
Слоистые декораторы, оборачивающие основной вызов LLM. У каждого декоратора одна ответственность. Они компонуются в разные комбинации для разных точек вызова.
@with_retry(max_retries=3, backoff="exponential", retry_on=[429, 503])
@with_cost_tracking(budget_per_call=5.00)
@with_structured_logging(log_level="DEBUG")
async def core_llm_call(client, model_spec, messages):
return await client.chat.completions.create(
model=model_spec.model_id, messages=messages
)Декоратор ретраев обрабатывает временные ошибки с экспоненциальным отступлением (backoff) и джиттером; типы ошибок, которые не следует ретраить (400, 401, 403), пропускаются немедленно. Механика лимитов запросов и полная архитектура обработки 429 освещены в нашем руководстве по обработке лимитов запросов — этот паттерн инкапсулирует эту логику для последовательного применения во всех точках вызова. Декоратор отслеживания стоимости логирует gen_ai.usage и предупреждает, если стоимость одного вызова превышает бюджет. Ни один декоратор не знает о другом. Порядок стека важен: ретрай снаружи (чтобы неудачные ретраи тоже отслеживались по стоимости), логирование внутри (чтобы оно видело финальный ответ).
Этот паттерн показывает, как инкапсулировать логику обработки лимитов запросов, чтобы она применялась последовательно в каждой точке вызова. Тот же принцип инкапсуляции применим к кэшированию промптов — декоратор @with_cache перехватывает повторные или похожие запросы до того, как они повлекут API-вызов. Документация TokSpan по кэшированию промптов описывает механику кэширования на уровне API, которую оборачивает декоратор.
Анти-паттерн, который он заменяет
Операционный boilerplate, скопированный вокруг каждого вызова LLM. Несогласованные параметры ретраев. Отсутствующее отслеживание стоимости в трёх из восьми точек вызова. Никто не знает, какой формат логирования «правильный», потому что каждая точка вызова делает это немного по-своему.
Паттерн 5: Chain of Responsibility — фолбэк-пайплайны
Проблема
Фейловер модели захардкожен в вложенных блоках try/except. try gpt-5.5 —except: try claude-sonnet —except: try deepseek —except: return error. Добавление фолбэк-модели или изменение порядка цепочки означает переписывание всего блока. В каждой точке вызова своя, немного отличная цепочка.
Паттерн
Цепочка объектов ModelHandler. Каждый обработчик знает свою модель и как обработать запрос. Если он падает — нетранзиентная ошибка, таймаут, качество ниже порога — он передаёт следующему обработчику. Композиция цепочки живёт в конфиге, а не в коде.
class ModelHandler(ABC):
def __init__(self, model_spec: ModelSpec):
self.model_spec = model_spec
self._next: ModelHandler | None = None
def set_next(self, handler: "ModelHandler") -> "ModelHandler":
self._next = handler
return handler
async def handle(self, request: dict) -> dict | None:
try:
result = await self._call_model(request)
if self._quality_check(result):
return result
except NonRetryableError:
pass
if self._next:
return await self._next.handle(request)
return None
class FallbackChain:
def __init__(self):
self.head: ModelHandler | None = None
self.circuit_breaker: dict[str, int] = {} # model_id —consecutive failures
async def execute(self, request: dict) -> dict:
if not self.head:
raise RuntimeError("Empty fallback chain")
return await self.head.handle(request)Три последовательных сбоя у обработчика — circuit breaker временно убирает его из цепочки. Он возвращается после периода охлаждения с тестовым запросом. Наше руководство по мультимодельной архитектуре подробно освещает стратегии маршрутизации — этот паттерн предоставляет формализованную реализацию цепочки.
Анти-паттерн, который он заменяет
Вложенная try/except-логика фолбэка, скопированная по точкам вызова. Несогласованный порядок цепочки. Нет circuit breaker — деградированная модель на второй позиции добавляет задержку каждому фолбэку, так и не добиваясь успеха.
Паттерн 6: Template Method — стандартизированный цикл агента
Проблема
У каждого агента слегка разный цикл вызова инструментов. Кто-то использует while True. Кто-то — for i in range(max_iterations). Кто-то вообще забыл про лимит цикла. Несогласованное поведение между агентами. Риск неконтролируемого роста стоимости у агента, который может зацикливаться бесконечно.
Паттерн
Шаблонный метод AgentLoop с фиксированным скелетом: plan — выполнение инструмента — observe — решение о следующем шаге. Подклассы переопределяют хук-методы для собственного поведения. Скелет гарантирует, что каждый агент наследует одни и те же характеристики безопасности — лимит цикла, таймаут, потолок стоимости, структурированную обработку ошибок.
class AgentLoop(ABC):
def __init__(self, max_iterations: int = 15, timeout: float = 120.0, cost_cap: float = 5.00):
self.max_iterations = max_iterations
self.timeout = timeout
self.cost_cap = cost_cap
async def run(self, task: str) -> dict:
start = time.time()
total_cost = 0.0
for i in range(self.max_iterations):
if time.time() - start > self.timeout:
return {"status": "timeout", "partial_result": self._build_partial()}
if total_cost > self.cost_cap:
return {"status": "cost_cap_exceeded"}
plan = await self.plan(task) # Hook: override
tool = await self.select_tool(plan) # Hook: override
result = await self.execute(tool) # Hook: override
total_cost += result.get("cost", 0)
if await self.should_stop(i, result): # Hook: override
return await self.synthesize()
@abstractmethod
async def plan(self, task: str) -> dict: ...
@abstractmethod
async def select_tool(self, plan: dict) -> str: ...
@abstractmethod
async def execute(self, tool: str) -> dict: ...Наше руководство по одиночному агенту освещает основы цикла вызова инструментов. Этот паттерн даёт перспективу паттернов проектирования: формализованный шаблон, делающий гарантии безопасности структурными, а не декларативными.
Анти-паттерн, который он заменяет
Каждый агент реализует собственный цикл. Несогласованные защитные механизмы. Агент, который может зацикливаться вечно, потому что кто-то скопировал версию “while True” без проверки max_iterations.
Краткая справка: какой паттерн когда?
| У вас есть… | Используйте… |
|---|---|
| >3 строк моделей в вашей кодовой базе | Factory —централизуйте выбор модели |
| A/B-тесты промптов, реализованные через if/else | Strategy —инкапсулируйте варианты промптов |
| Потоковые потребители, связанные с кодом генерации | Observer —развяжите связь с помощью событийно-ориентированного дизайна |
| 60 строк шаблонного кода вокруг каждого вызова LLM | Decorator —наслаивайте эксплуатационные задачи |
| Вложенные try/except для отказоустойчивости моделей | Chain of Responsibility —настраиваемый запасной вариант |
| Несколько агентов с несогласованными циклами | Template Method —стандартизируйте с помощью защитных механизмов |
Порядок внедрения по размеру кодовой базы: Малая (менее 5K строк, 1–2 сценария) — начните с Decorator и Factory. Средняя (5–50K строк) — добавьте Strategy и Chain of Responsibility. Крупная (более 50K строк, несколько агентов) — добавьте Observer и Template Method.
Все шесть паттернов работают со стандартными SDK, совместимыми с OpenAI. Единый API-endpoint означает, что конфиг Factory — это один base_url и N строк моделей, а не N базовых URL × M провайдеров.
FAQ
Не будут ли эти паттерны избыточной инженерией для простого вызова API?
Если в вашей кодовой базе один вызов LLM и больше двух он не вырастет, да — 50 строк прямого client.chat.completions.create() — правильный ответ. Когда кодовая база достигает 10+ вызовов LLM, 3+ вариантов моделей и требований продакшн-надёжности, ROI этих паттернов материализуется на первом инциденте — на первой миграции модели, которая должна была стать изменением конфига, на первом неконтролируемом росте стоимости из-за отсутствующего лимита цикла, на первой регрессии промпта без пути отката.
Какой паттерн реализовать первым?
Decorator. Он наслаивается на существующие вызовы LLM без их изменения. Один стек декораторов — ретрай, логирование, отслеживание стоимости — применяется к каждой точке вызова. Немедленный выигрыш в продакшн-надёжности. Ноль рефакторинга существующего кода. Затем Factory — когда вам снова понадобится сменить модели, вы измените одно значение конфига вместо 15 файлов.
Работают ли эти паттерны с LangChain или LlamaIndex?
Они сосуществуют. Factory и Strategy работают чище вне LangChain — они предотвращают блокировку фреймворком выбора модели и управления промптами. Observer и Template Method могут жить внутри агентов LangChain — структура цикла и стриминговые потребители выигрывают от интеграции с фреймворком. Эти паттерны не заменяют LangChain. Они структурируют код вокруг выбранного вами фреймворка.
Как паттерны выдерживают модели, живущие за разными API провайдеров?
Паттерны становятся проще в реализации, а не сложнее. Factory: один экземпляр клиента покрывает каждую модель — ваш конфиг это один base_url и N строк моделей, а не N базовых URL × M провайдеров. Chain of Responsibility: фолбэк между провайдерами через одну точку интеграции. Decorator: согласованное отслеживание стоимости, потому что все вызовы проходят через один шлюз. Сами паттерны провайдер-агностичны. Унифицированный endpoint сокращает интеграционную поверхность, которую должен обслуживать каждый паттерн, — в этом весь смысл абстракции. За более широкой перспективой, почему архитектура с одним endpoint становится стандартом индустрии, наш анализ перехода к платформам агрегации AI API освещает операционные и стоимостные драйверы тренда.
Существуют ли LLM-специфичные паттерны помимо GoF?
Да. Semantic Router — маршрутизация по семантике запроса, а не по захардкоженным правилам. Guard — конвейер валидации входа/выхода, выполняющийся до и после каждого вызова LLM. Cache-Aside — слой семантического кэширования, проверяющий схожесть эмбеддингов перед API-вызовом. Это LLM-нативные паттерны, заслуживающие отдельной статьи. Шесть паттернов здесь выбраны намеренно: большинство инженерных команд уже знают паттерны GoF. Сопоставление их с LLM API сводит кривую обучения почти к нулю.
Паттерны проектирования не про изощрённость. Они про то, чтобы не иметь один и тот же баг в восьми местах, потому что код был скопирован, а не структурирован.
Начните с Decorator. Добавьте Factory. Следующая миграция модели займёт 30 секунд — а не утро поиска и замены и послеобеденное время на отладку точки вызова, которую вы пропустили.
Сохраните этот справочник в закладки. В следующий раз, когда поймаете себя на четвёртом копировании логики ретраев, вы будете знать, какой ящик открыть. За дополнительными продакшн-паттернами и руководствами по архитектуре LLM API, которые держат вашу кодовую базу структурированной по мере роста, подпишитесь на наш блог.