Слоистая архитектура для AI-приложений: как сделать код читаемым, надежным и расширяемым

Почему AI-проекты превращаются в спагетти-код через полгода

Открываешь проект через шесть месяцев после старта - и не узнаешь. В одном файле - логика вызова GPT-4.5 Turbo (да, в 2026 это уже базовая модель), в другом - обработка ошибок, в третьем - валидация данных. Изменяешь формат ответа API - ломаешь три модуля. Добавляешь кэширование - переписываешь половину кодовой базы.

Знакомо? Это классическая история AI-разработки, где скорость важнее архитектуры. Пока не становится слишком поздно.

Слоистая архитектура: не учебник по паттернам, а выживание в production

Слоистая архитектура для AI-приложений - это не про следование догмам. Это про возможность:

• Заменить модель (с GPT-4.5 на Claude 3.7 Sonnet) без переписывания бизнес-логики
• Добавить кэширование ответов, не трогая обработку данных
• Тестировать логику приложения без реальных вызовов к API (которые стоят денег и времени)
• Внедрить A/B-тестирование моделей за пару часов, а не дней

Основная идея проста: каждая часть приложения должна заниматься своим делом. И не знать о существовании других частей больше необходимого.

Три слоя, которые спасут ваш AI-проект

1 Слой представления (Presentation Layer): только HTTP и валидация

Этот слой знает только о HTTP-запросах, форматах данных и валидации. Его задача - принять запрос, проверить его корректность, и передать дальше. Никакой бизнес-логики, никаких вызовов моделей.

# ПЛОХО: все в одном месте
@app.post("/generate")
async def generate_text(request: TextRequest):
    # Валидация
    if len(request.text) > 1000:
        raise HTTPException(status_code=400, detail="Text too long")
    
    # Бизнес-логика
    if request.style == "formal":
        prompt = f"Rewrite formally: {request.text}"
    else:
        prompt = f"Rewrite casually: {request.text}"
    
    # Вызов модели
    response = openai.ChatCompletion.create(
        model="gpt-4.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    
    # Обработка ответа
    result = response.choices[0].message.content
    return {"result": result}

# ХОРОШО: слой представления делает только свою работу
from pydantic import BaseModel, Field
from typing import Optional

class TextGenerationRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=1000)
    style: str = Field(..., regex="^(formal|casual|creative)$")
    temperature: Optional[float] = Field(0.7, ge=0.0, le=2.0)

@app.post("/generate", response_model=TextGenerationResponse)
async def generate_text(request: TextGenerationRequest):
    """Только валидация и передача в сервисный слой"""
    try:
        result = await text_generation_service.generate(
            text=request.text,
            style=request.style,
            temperature=request.temperature
        )
        return TextGenerationResponse(result=result)
    except ServiceError as e:
        raise HTTPException(status_code=400, detail=str(e))

2 Сервисный слой (Service Layer): доменная логика без инфраструктуры

Здесь живет бизнес-логика вашего AI-приложения. Этот слой знает, КАК обрабатывать данные, но не знает, ГДЕ они хранятся или КАК вызываются модели.

class TextGenerationService:
    def __init__(self, model_client: ModelClient, cache_client: CacheClient):
        self.model_client = model_client
        self.cache_client = cache_client
    
    async def generate(self, text: str, style: str, temperature: float) -> str:
        """Бизнес-логика генерации текста"""
        
        # Проверка кэша
        cache_key = f"generation:{hash(text+style)}"
        cached = await self.cache_client.get(cache_key)
        if cached:
            return cached
        
        # Подготовка промпта (бизнес-логика)
        prompt = self._build_prompt(text, style)
        
        # Вызов модели через абстракцию
        response = await self.model_client.generate(
            prompt=prompt,
            temperature=temperature,
            max_tokens=1000
        )
        
        # Пост-обработка (бизнес-логика)
        result = self._post_process(response, style)
        
        # Сохранение в кэш
        await self.cache_client.set(cache_key, result, ttl=3600)
        
        return result
    
    def _build_prompt(self, text: str, style: str) -> str:
        """Логика построения промпта - чистая бизнес-логика"""
        style_prompts = {
            "formal": "Rewrite the following text in formal business English:",
            "casual": "Make this text sound casual and friendly:",
            "creative": "Rewrite this text creatively:"
        }
        return f"{style_prompts[style]}\n\n{text}"

3 Слой инфраструктуры (Infrastructure Layer): работа с внешним миром

Здесь находятся все детали реализации: вызовы API к OpenAI/Anthropic, работа с базой данных, кэширование, очереди сообщений. Этот слой не знает о бизнес-логике.

from abc import ABC, abstractmethod
from typing import Protocol

# Абстракция для работы с любой AI-моделью
class ModelClient(Protocol):
    async def generate(self, prompt: str, **kwargs) -> str:
        ...

# Реализация для OpenAI (на 27.01.2026 актуальна версия 2.0+)
class OpenAIClient:
    def __init__(self, api_key: str, base_url: str = "https://api.openai.com/v1"):
        self.client = AsyncOpenAI(api_key=api_key, base_url=base_url)
    
    async def generate(self, prompt: str, **kwargs) -> str:
        """Конкретная реализация вызова OpenAI"""
        try:
            response = await self.client.chat.completions.create(
                model=kwargs.get("model", "gpt-4.5-turbo"),
                messages=[{"role": "user", "content": prompt}],
                temperature=kwargs.get("temperature", 0.7),
                max_tokens=kwargs.get("max_tokens", 1000)
            )
            return response.choices[0].message.content
        except APIConnectionError as e:
            raise ModelConnectionError(f"OpenAI connection failed: {e}")
        except APIError as e:
            raise ModelError(f"OpenAI API error: {e}")

# Реализация для Claude (Anthropic обновил API в 2025)
class ClaudeClient:
    def __init__(self, api_key: str):
        self.client = anthropic.AsyncAnthropic(api_key=api_key)
    
    async def generate(self, prompt: str, **kwargs) -> str:
        """Конкретная реализация вызова Claude 3.7"""
        response = await self.client.messages.create(
            model="claude-3-7-sonnet-20250226",
            max_tokens=kwargs.get("max_tokens", 1000),
            temperature=kwargs.get("temperature", 0.7),
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text

Вертикальные срезы: когда слоев недостаточно

Слоистая архитектура решает проблему разделения ответственности, но создает другую: раскидывание одной фичи по десятку файлов. Решение - вертикальные срезы (vertical slices).

# Структура проекта с вертикальными срезами
project/
├── features/
│   ├── text_generation/          # Весь код для генерации текста
│   │   ├── __init__.py
│   │   ├── models.py            # Pydantic модели
│   │   ├── service.py           # Сервисный слой
│   │   ├── clients.py           # Клиенты для моделей
│   │   ├── routers.py           # FastAPI роутеры
│   │   └── tests/               # Тесты для этой фичи
│   ├── image_classification/    # Классификация изображений
│   └── chat_bot/               # Чат-бот логика
├── core/                       # Общая инфраструктура
│   ├── caching/
│   ├── database/
│   └── logging/
└── shared/                     # Общие утилиты

Преимущества такого подхода:

• Новая фича = новая директория. Не нужно искать код по всему проекту
• Можно удалить фичу, удалив одну директорию (и обновив зависимости)
• Разные команды работают над разными фичами, не мешая друг другу
• Тестирование изолировано по фичам

Dependency Injection: секретное оружие тестируемости

Без dependency injection (DI) слоистая архитектура превращается в костыли. DI позволяет:

# Без DI - жесткая связь
class TextGenerationService:
    def __init__(self):
        # Прямое создание зависимостей
        self.model_client = OpenAIClient(api_key=os.getenv("OPENAI_API_KEY"))
        self.cache = RedisCache(host="localhost")
    
# С DI - гибкость и тестируемость
class TextGenerationService:
    def __init__(self, model_client: ModelClient, cache: CacheClient):
        # Зависимости передаются извне
        self.model_client = model_client
        self.cache = cache

# В production
service = TextGenerationService(
    model_client=OpenAIClient(api_key=os.getenv("OPENAI_API_KEY")),
    cache=RedisCache(host="redis.prod")
)

# В тестах
mock_client = Mock(spec=ModelClient)
mock_client.generate.return_value = "Mocked response"

service = TextGenerationService(
    model_client=mock_client,
    cache=Mock(spec=CacheClient)
)
# Тестируем бизнес-логику без реальных вызовов API

Для FastAPI (актуально на 2026 год) используем Depends:

from fastapi import Depends

# Фабрики зависимостей
def get_model_client() -> ModelClient:
    """Возвращает клиент для работы с AI-моделями"""
    provider = os.getenv("AI_PROVIDER", "openai")
    
    if provider == "openai":
        return OpenAIClient(api_key=os.getenv("OPENAI_API_KEY"))
    elif provider == "anthropic":
        return ClaudeClient(api_key=os.getenv("ANTHROPIC_API_KEY"))
    elif provider == "local":
        return LocalLLMClient(model_path="/models/llama-3.2-3b")
    else:
        raise ValueError(f"Unknown provider: {provider}")

@app.post("/generate")
async def generate_text(
    request: TextGenerationRequest,
    model_client: ModelClient = Depends(get_model_client),
    cache: CacheClient = Depends(get_cache_client)
):
    service = TextGenerationService(
        model_client=model_client,
        cache=cache
    )
    return await service.generate(**request.dict())

Тестирование AI-приложений: как не разориться на API-вызовах

Тестирование AI-приложений имеет свою специфику. Реальные вызовы к GPT-4.5 или Claude 3.7 дорогие и медленные. Решение - многоуровневое тестирование.

# Пример юнит-теста для бизнес-логики
import pytest
from unittest.mock import Mock, AsyncMock

@pytest.mark.asyncio
async def test_text_generation_service_with_cache():
    """Тестируем логику кэширования без реальных вызовов API"""
    
    # Моки зависимостей
    mock_model = AsyncMock(spec=ModelClient)
    mock_cache = AsyncMock(spec=CacheClient)
    
    # Настраиваем поведение моков
    mock_cache.get.return_value = None  # Кэш пустой
    mock_model.generate.return_value = "Generated text"
    
    # Создаем сервис с моками
    service = TextGenerationService(
        model_client=mock_model,
        cache_client=mock_cache
    )
    
    # Выполняем тест
    result = await service.generate("test", "formal")
    
    # Проверяем, что логика работает правильно
    assert result == "Generated text"
    mock_cache.get.assert_called_once()  # Проверили кэш
    mock_model.generate.assert_called_once()  # Вызвали модель
    mock_cache.set.assert_called_once()  # Сохранили в кэш

# Контрактный тест с записью ответов
@pytest.mark.vcr  # Используем VCR.py для записи/воспроизведения HTTP
async def test_openai_client_integration():
    """Тест реальной интеграции с OpenAI (записывается один раз)"""
    client = OpenAIClient(api_key="test_key")
    
    # При первом запуске делает реальный вызов и записывает ответ
    # При последующих - использует записанный ответ
    result = await client.generate("Hello, world!")
    
    assert isinstance(result, str)
    assert len(result) > 0

Типичные ошибки и как их избежать

Решение: Выносите промпты в конфигурацию или базу данных:

# Плохо
class Service:
    def generate(self, text):
        prompt = f"Rewrite this text: {text}"  # Хардкод

# Хорошо
class PromptTemplate:
    def __init__(self, template: str):
        self.template = template
    
    def render(self, **kwargs) -> str:
        return self.template.format(**kwargs)

# В конфигурации или БД
prompt_templates = {
    "text_rewrite": PromptTemplate(
        "Rewrite the following text in {style} style: {text}"
    ),
    "text_summarize": PromptTemplate(
        "Summarize this text in {length} words: {text}"
    )
}

Решение: Паттерн Circuit Breaker и retry с экспоненциальной задержкой:

from tenacity import retry, stop_after_attempt, wait_exponential
from circuitbreaker import circuit

class ResilientModelClient:
    def __init__(self, client: ModelClient):
        self.client = client
    
    @circuit(failure_threshold=5, expected_exception=ModelError)
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
        retry_error_callback=lambda retry_state: None
    )
    async def generate(self, prompt: str, **kwargs) -> str:
        """Устойчивый вызов с повторными попытками и circuit breaker"""
        try:
            return await self.client.generate(prompt, **kwargs)
        except (APIConnectionError, RateLimitError) as e:
            # Логируем, но не паникуем
            logger.warning(f"Model call failed: {e}")
            
            # Можно вернуть fallback или бросить кастомную ошибку
            if kwargs.get("use_fallback"):
                return self._fallback_response(prompt)
            raise ModelTemporaryError(f"Service unavailable: {e}")

Миграция существующего проекта: не с нуля, а шаг за шагом

Не нужно переписывать весь проект сразу. Начните с самого болезненного места:

1. Выделите самый грязный модуль - тот, где смешана логика, вызовы API и работа с БД
2. Создайте для него интерфейсы (Protocol) - определите, что этот модуль ДОЛЖЕН делать
3. Выделите бизнес-логику в отдельный класс, принимающий зависимости через __init__
4. Заменяйте прямой вызов на вызов через интерфейс в одном месте
5. Пишите тесты для новой реализации
6. Повторяйте для следующего модуля

Через месяц у вас будет архитектура, а не хаос. Через два - вы сможете заменить LLM-провайдера за день, а не за неделю.

Инструменты 2026 года, которые упростят жизнь

• FastAPI 1.0+ (наконец-то стабильный релиз) с встроенной поддержкой dependency injection
• Pydantic 3.0+ с валидацией на уровне типов и генерацией схем из Python-аннотаций
• Litellm - унифицированный интерфейс для 100+ LLM провайдеров (актуально на 2026)
• LangChain 1.0+ (после полного рефакторинга) для сложных цепочек вызовов
• Dagger для управления зависимостями в больших проектах
• AI-ассистенты для рефакторинга - они научились понимать архитектуру, а не просто переписывать код

Архитектура AI-приложений в 2026 - это не про сложность. Это про простоту. Про то, чтобы через полгода не бояться открывать свой же код. Про то, чтобы добавлять фичи за дни, а не недели. Про то, чтобы тестировать логику, а не надеяться, что "в продакшене сработает".

Самый важный принцип: если изменение в одном месте требует изменений в трех других - архитектура сломана. Исправляйте это сразу, не откладывая. Потому что завтра таких мест будет десять. Послезавтра - сто. А потом проект умрет под грузом технического долга.

Начните с одного модуля. Сегодня.