Как построить серверный AI Gateway на AWS AppSync: архитектура и пример кода

Почему все ненавидят прямой доступ к LLM из фронтенда

Вы пишете приложение, которое общается с GPT-5, Claude 3.7 или каким-нибудь свежим Qwen2.5. Ключ API лежит в переменных окружения фронтенда. Все работает, пока ваш инженер случайно не закоммитит его в публичный репозиторий. Или пока один пользователь не начнет слать 100 запросов в секунду, сжирая весь бюджет. Или пока не понадобится логировать запросы, ограничивать токены, добавлять кэширование.

Прямой вызов LLM из клиента - это технический долг, который придет коллекторами через месяц. Нужен шлюз. Не просто прокси, а полноценный AI Gateway. И AWS AppSync в 2026 году - один из самых элегантных способов его построить. Почему? Потому что он из коробки дает GraphQL, подписки через Websocket, авторизацию и автоматическое масштабирование. Вы платите только за реальное использование, а не за постоянно крутящиеся инстансы.

AppSync Events: ваш билет в мир управляемых AI-запросов

В 2026 году AWS серьезно прокачал AppSync. Появилась прямая интеграция с EventBridge, поддержка кастомных резолверов на Rust, и что важнее - встроенная работа с длительными операциями и WebSockets для стриминга. Это идеально для LLM, которые могут генерировать ответ минутами.

Смысл AI Gateway прост: все запросы к разным моделям (OpenAI, Anthropic, локальные модели на SageMaker) проходят через единый GraphQL-интерфейс. Вы добавляете логику: ограничение запросов, модификацию промптов, логирование, кэширование похожих запросов. Клиент подключается по WebSocket и получает ответы потоком, как в ChatGPT. А вы спите спокойно, зная, что никто не украдет ключи и не обанкротит вас DDoS-ом.

Архитектура, которая не сломается при первом же запросе

Давайте посмотрим на компоненты. Это не монолит, а набор сервисов, где каждый отвечает за свое. Если один компонент падает, система продолжает работать.

Поток данных выглядит так: клиент → GraphQL-запрос в AppSync → Lambda-резолвер (проверяет лимиты, логирует) → вызов внешнего LLM API или SageMaker → потоковый ответ через WebSocket подписку. Для асинхронных задач (генерация длинного текста, обработка файлов) резолвер запускает EventBridge rule, который запускает отдельную Lambda, а та отправляет прогресс обратно через AppSync Subscriptions. Это сложнее, чем кажется, но работает как часы. Если вы проектируете сложные AI-системы, эта архитектура станет основой.

1 Шаг 1: Настраиваем AppSync API с поддержкой WebSocket

Откройте консоль AWS или возьмите CloudFormation. Создайте AppSync API с типом "GraphQL и подписки". В 2026 году можно выбрать схему авторизации - лучше использовать API_KEY для быстрого старта, но в продакшне перейти на AWS_IAM или OIDC. Включите логирование в CloudWatch - без логов вы слепы.

# cloudformation-template.yaml фрагмент
Resources:
  AiGatewayApi:
    Type: AWS::AppSync::GraphQLApi
    Properties:
      Name: ai-gateway-2026
      AuthenticationType: API_KEY
      AdditionalAuthenticationProviders:
        - AuthenticationType: AWS_IAM
      LogConfig:
        CloudWatchLogsRoleArn: !GetAtt AppSyncLogsRole.Arn
        FieldLogLevel: ALL
      XrayEnabled: true

Не забудьте про квоты. По умолчанию AppSync имеет лимит на количество подписок на соединение. Если вы планируете тысячи одновременных пользователей, откройте тикет в поддержку AWS заранее. И да, WebSocket-соединения таймаутят через неактивность - клиент должен отправлять ping-сообщения.

2 Шаг 2: Пишем GraphQL схему, которая не будет меняться каждую неделю

Схема - это контракт между фронтендом и бэкендом. Сделайте ее гибкой, но стабильной. Добавьте union-типы для разных моделей и streaming-полей для прогресса.

# schema.graphql
type Query {
  health: String
}

type Mutation {
  # Основной запрос к LLM
  generateCompletion(prompt: String!, model: AIModel!, stream: Boolean = false): CompletionResult
  # Асинхронная задача
  startLongTask(input: TaskInput!): TaskStatus
}

type Subscription {
  # Поток токенов для streaming-ответов
  onCompletionStream(taskId: ID!): CompletionStream
  # Прогресс длительной задачи
  onTaskProgress(taskId: ID!): TaskProgress
}

enum AIModel {
  GPT_5
  CLAUDE_3_7
  QWEN2_5_72B
  ANTHROPIC_CLAUDE_3_5_SONNET_2025
}

union CompletionResult = CompletionSuccess | CompletionError | CompletionStreamInit

type CompletionSuccess {
  id: ID!
  content: String!
  tokensUsed: Int!
  model: String!
}

type CompletionStreamInit {
  taskId: ID!
  message: String!
}

# ... остальные типы

Видите union-типы? Это хитрость. Вместо того чтобы возвращать поле `error` в каждом объекте, вы разделяете успех и ошибку на уровне GraphQL. Клиентский код становится чище. А streaming-ответы идут через подписки - это ключевая фишка для современных AI-приложений, которые хотят показывать ответ по мере генерации, как в ChatGPT.

3 Шаг 3: Lambda-резолверы, где живет вся логика

Резолверы в AppSync могут быть прямыми (VTL) или на Lambda. VTL - это боль и страдание. Используйте Lambda на Python или Node.js. Вот как выглядит основной резолвер для `generateCompletion`.

# lambda_resolver.py (Python 3.13)
import os
import json
import boto3
from ai_providers import OpenAIClient, AnthropicClient, SageMakerClient
from rate_limiter import RateLimiter
from aws_lambda_powertools import Logger, Tracer

logger = Logger()
tracer = Tracer()
secrets = boto3.client('secretsmanager')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('ai-requests-2026')

def handler(event, context):
    # event содержит GraphQL-аргументы
    prompt = event['arguments']['prompt']
    model = event['arguments']['model']
    stream = event['arguments'].get('stream', False)
    user_id = event['identity']['sub']  # из IAM или Cognito
    
    # 1. Проверяем лимиты
    limiter = RateLimiter(user_id)
    if not limiter.check_limit('daily_tokens', max_tokens=1000000):
        return {
            '__typename': 'CompletionError',
            'errorCode': 'RATE_LIMIT_EXCEEDED',
            'message': 'Daily token limit exceeded'
        }
    
    # 2. Логируем запрос в DynamoDB
    request_id = context.aws_request_id
    table.put_item(Item={
        'requestId': request_id,
        'userId': user_id,
        'prompt': prompt,
        'model': model,
        'timestamp': context.get_remaining_time_in_millis(),
        'ttl': int(time.time()) + 30*24*3600  # храним 30 дней
    })
    
    # 3. Выбираем провайдера
    provider = get_provider(model)
    
    # 4. Если streaming, запускаем асинхронную задачу и возвращаем taskId
    if stream:
        task_id = start_streaming_task(prompt, model, user_id, request_id)
        return {
            '__typename': 'CompletionStreamInit',
            'taskId': task_id,
            'message': 'Streaming started. Subscribe to onCompletionStream.'
        }
    
    # 5. Синхронный вызов
    try:
        response = provider.generate_completion(prompt, max_tokens=2000)
        return {
            '__typename': 'CompletionSuccess',
            'id': request_id,
            'content': response.text,
            'tokensUsed': response.usage.total_tokens,
            'model': model
        }
    except Exception as e:
        logger.error(f'LLM call failed: {e}')
        return {
            '__typename': 'CompletionError',
            'errorCode': 'PROVIDER_ERROR',
            'message': str(e)
        }

def get_provider(model):
    # Получаем ключи из Secrets Manager
    secret_name = f'llm-keys-{model}'
    secret = secrets.get_secret_value(SecretId=secret_name)
    api_key = json.loads(secret['SecretString'])['api_key']
    
    if 'GPT' in model:
        return OpenAIClient(api_key, model='gpt-5-2026-01')
    elif 'CLAUDE' in model:
        return AnthropicClient(api_key, model='claude-3-7-sonnet-2025-12')
    else:
        return SageMakerClient(endpoint_name='qwen2-5-72b-endpoint')

Код кажется простым? Это иллюзия. В реальности нужно добавить ретраи с экспоненциальной задержкой, кэширование идентичных промптов (чтобы не платить дважды за один и тот же вопрос), и валидацию промптов на инъекции. Для enterprise-проектов это критично, как описано в статье про интеграцию AI в Java/Kotlin.

4 Шаг 4: Streaming через подписки и EventBridge

Стриминг - самая сложная часть. Lambda имеет лимит 15 минут выполнения, а генерация длинного текста может занимать больше. Поэтому используем EventBridge и отдельную Lambda для длительных задач.

# lambda_streamer.py
import json
import boto3
from websocket import create_connection

appsync = boto3.client('appsync')

def handler(event, context):
    task_id = event['detail']['taskId']
    prompt = event['detail']['prompt']
    model = event['detail']['model']
    
    # Подключаемся к LLM с поддержкой streaming (например, OpenAI streaming API)
    # Получаем поток токенов
    for chunk in stream_from_llm(prompt, model):
        # Отправляем каждый кусок через AppSync Subscription
        appsync.post(
            apiId=os.environ['APPSYNC_API_ID'],
            query=f'''
                mutation PublishStream($taskId: ID!, $chunk: String!) {{
                    publishStream(taskId: $taskId, chunk: $chunk) {{
                        taskId
                    }}
                }}
            ''',
            variables={
                'taskId': task_id,
                'chunk': chunk
            }
        )
    
    # Отправляем финальное событие
    appsync.post(
        apiId=os.environ['APPSYNC_API_ID'],
        query='''
            mutation CompleteStream($taskId: ID!) {
                completeStream(taskId: $taskId) {
                    taskId
                }
            }
        ''',
        variables={'taskId': task_id}
    )

Как это работает? Резолвер `generateCompletion` при `stream=true` создает задачу в EventBridge. EventBridge запускает эту Lambda, которая открывает долгое соединение с LLM и через мутации в AppSync отправляет данные всем подписанным клиентам. Клиент подписывается на подписку `onCompletionStream(taskId: $taskId)` и получает куски в реальном времени. Это мощно, но требует аккуратной обработки разрывов соединения и очистки ресурсов.

Где все падает: типичные ошибки и как их избежать

Я видел десятки таких реализаций. Вот что ломается чаще всего.

• Слишком много подписок на одно соединение. AppSync имеет лимит (обычно 100). Если у вас многопользовательское приложение, создавайте отдельное соединение для каждого потока или используйте фильтры в подписках.
• Таймауты WebSocket. Неактивные соединения разрываются через 10 минут. Клиент должен отправлять keep-alive сообщения. В 2026 году многие библиотеки делают это автоматически, но проверьте.
• Лимиты DynamoDB на запись. Если вы логируете каждый запрос, убедитесь, что таблица имеет достаточную пропускную способность или используйте on-demand режим.
• Стоимость Secrets Manager. Каждый секрет стоит $0.40 в месяц. Если у вас 100 моделей, это $40 просто за хранение ключей. Рассмотрите возможность хранения зашифрованных ключей в DynamoDB или Parameter Store для не-critical данных.
• Холодный старт Lambda при streaming. Если Lambda для стриминга запускается редко, ее инициализация может занять несколько секунд. Используйте Provisioned Concurrency или перепишите резолвер на контейнеры (если время критично).

А что с безопасностью? Это же AI, там свои приколы

Безопасность в AI Gateway делится на три слоя.

1. Аутентификация и авторизация. Используйте AWS_IAM с Cognito или OIDC-провайдером. Каждый пользователь имеет свой IAM-роль с политикой, ограничивающей количество запросов в день. Никаких общих API-ключей.
2. Защита промптов. Промпт-инъекции - реальная угроза. Валидируйте входные данные, обрезайте длинные промпты, используйте moderation API от OpenAI или AWS Bedrock, чтобы отфильтровать вредоносный контент.
3. Шифрование данных. Все ключи в Secrets Manager, все запросы в DynamoDB шифруются на стороне AWS (KMS). Если вы работаете в регулируемых отраслях, активируйте шифрование на клиентской стороне перед отправкой промпта.

Помните про централизованный контроль доступа к данным. Ваш AI Gateway должен знать, к каким данным пользователь имеет доступ, и не передавать в промпте лишнего. Это особенно важно, когда вы используете суб-агентов, которые сами делают запросы к базам данных.

Когда это все не нужно

Сложная архитектура на AppSync оправдана, если у вас десятки тысяч запросов в день, нужен streaming и tight integration с AWS-экосистемой. Если вы делаете MVP или внутренний инструмент на пять пользователей, возможно, проще использовать готовые решения вроде LiteLLM или OpenRouter, как описано в сравнении AI Gateway против кастомных решений.

Но если вы растите продукт, который должен масштабироваться до миллионов запросов, иметь сложную логику тарификации и работать в режиме реального времени - кастомный AI Gateway на AppSync окупится через полгода. Вы получаете полный контроль над каждым запросом, можете A/B тестировать разные модели и стоите на плечах гигантов AWS.

Что дальше? AI Gateway становится умнее

В 2026 году тренд - это AI Gateway с встроенной аналитикой и автоматической оптимизацией. Ваш шлюз может сам решать, какую модель использовать для конкретного промпта (дешевую или точную), кэшировать ответы и предсказывать стоимость запроса до его отправки. Добавьте сюда автономного QA-агента, который тестирует ваши промпты на регрессии, и получите систему, которая почти не требует ручного вмешательства.

Код из статьи - это каркас. Допишите под себя. Начните с простой схемы без streaming, добавьте одну модель, потом вторую. Через месяц у вас будет продакшн-готовый AI Gateway, который не стыдно показать инвесторам. И да, не забудьте настроить алерты в CloudWatch на аномальное потребление токенов - однажды это спасет ваш бюджет от студента, который решил прогнать через ваше API всю Википедию.