Урок 3: Интеграция российских моделей с агентами

Чему вы научитесь

• Создавать universal wrapper для GigaChat и YandexGPT
• Реализовывать fallback-логику (если одна модель недоступна → вторая)
• Интегрировать российские LLM в платформы автоматизации (n8n, Dify)
• Работать с LangChain (если поддерживается)
• Проектировать системы с резервированием
• Корректно обрабатывать ошибки и таймауты

Содержание

Пример 1: n8n + GigaChat — Автоматизация обработки текстов

Сценарий: Автоматически обрабатывать входящие письма через GigaChat

n8n — это no-code/low-code платформа для автоматизации. Она имеет встроенные узлы (nodes) для HTTP запросов.

Шаги в n8n (visual workflow):

[Email Trigger]
    ↓
[HTTP Request: Get GigaChat Token]
    ↓
[HTTP Request: Call GigaChat API]
    ↓
[Extract Text from Response]
    ↓
[Send Email with Result]

HTTP Node для получения токена:

Method: POST
URL: https://ngw.devices.sberbank.ru:9443/api/v2/oauth

Headers:
- Content-Type: application/x-www-form-urlencoded
- Authorization: Basic {{ base64($env.GIGACHAT_CLIENT_ID + ':' + $env.GIGACHAT_CLIENT_SECRET) }}
- RQ-UUID: {{ $json.uuid }}

Body (form):
grant_type: client_credentials
scope: GIGACHAT_API_PERS

Output path in node: $.access_token

HTTP Node для вызова GigaChat:

Method: POST
URL: https://gigachat.devices.sberbank.ru/api/v1/chat/completions

Headers:
- Authorization: Bearer {{ $node["HTTP Request - Token"].json.access_token }}
- Content-Type: application/json
- RQ-UUID: {{ $json.uuid }}

Body:
{
  "model": "GigaChat-2",
  "messages": [
    {
      "role": "user",
      "content": "Классифицируй это письмо по темам: {{ $node.Email.json.body }}"
    }
  ],
  "temperature": 0.7,
  "max_tokens": 512
}

Раздел будет дополнен после получения материалов от партнёров (скриншоты и точные конфиги).

Пример 2: Dify + YandexGPT — Q&A бот на русском

Сценарий: Создать Q&A бот, который отвечает на вопросы на основе вашей базы знаний (RAG)

Dify — платформа для создания LLM-приложений. Она поддерживает custom API интеграции.

Архитектура:

User Question
    ↓
[Vector Store: Find Relevant Docs]
    ↓
[YandexGPT Embedding API: Encode Question]
    ↓
[Find K nearest neighbors]
    ↓
[YandexGPT Complete: Generate Answer with Context]
    ↓
Response to User

Конфигурация в Dify:

1. Перейти Integrations → Custom LLM
2. Добавить новый провайдер:

Provider Name: YandexGPT
Model ID: yandexgpt
API Type: REST

Request Format:
{
  "modelUri": "gpt://{{ folder_id }}/yandexgpt",
  "completionOptions": {
    "temperature": {{ temperature }},
    "maxTokens": {{ max_tokens }}
  },
  "messages": [
    {
      "role": "user",
      "text": "{{ prompt }}"
    }
  ]
}

Response Parsing:
result.alternatives[0].message.text

3. Создать Knowledge Base:

   • Загрузить документы (PDF, TXT)
   • Выбрать embedding модель: YandexGPT Text Embedding
   • Сохранить в Vector Store

4. Создать Chat Node:

   • Система: “Ты русскоязычный помощник”
   • Context от Knowledge Base
   • Model: YandexGPT Pro

Пример 3: Python-скрипт с fallback логикой

Идея: Если GigaChat недоступна → переключиться на YandexGPT → если и она не работает → Claude

sequenceDiagram
  participant C as Client
  participant G as GigaChat
  participant Y as YandexGPT

  C->>G: запрос
  G-->>C: ошибка 503
  C->>Y: fallback
  Y-->>C: успех

import requests
import uuid
import base64
import os
import json
from typing import Optional
from enum import Enum
from dotenv import load_dotenv

load_dotenv()

class LLMProvider(Enum):
    GIGACHAT = "gigachat"
    YANDEXGPT = "yandexgpt"
    CLAUDE = "claude"

class UniversalLLMClient:
    """
    Универсальный клиент для работы с несколькими LLM.
    Поддерживает fallback: если основной провайдер не работает → резервный.
    """

    def __init__(self, primary_provider=LLMProvider.GIGACHAT):
        self.primary = primary_provider
        self.fallback_order = self._get_fallback_order(primary_provider)

        # Инициализация конфигов
        self.gigachat_config = {
            "client_id": os.getenv("GIGACHAT_CLIENT_ID"),
            "client_secret": os.getenv("GIGACHAT_CLIENT_SECRET"),
            "token": None
        }

        self.yandex_config = {
            "access_token": os.getenv("YANDEX_ACCESS_TOKEN"),
            "folder_id": os.getenv("YANDEX_FOLDER_ID")
        }

        self.claude_config = {
            "api_key": os.getenv("ANTHROPIC_API_KEY")
        }

    def _get_fallback_order(self, primary):
        """Определить порядок fallback"""
        order = {
            LLMProvider.GIGACHAT: [
                LLMProvider.YANDEXGPT,
                LLMProvider.CLAUDE
            ],
            LLMProvider.YANDEXGPT: [
                LLMProvider.GIGACHAT,
                LLMProvider.CLAUDE
            ],
            LLMProvider.CLAUDE: [
                LLMProvider.GIGACHAT,
                LLMProvider.YANDEXGPT
            ]
        }
        return order[primary]

    def chat(self, message: str, model_preference: Optional[LLMProvider] = None) -> dict:
        """
        Отправить сообщение. Автоматически пробует fallback.

        Returns:
            {
                "text": "ответ",
                "provider": "gigachat",
                "tokens": 150,
                "error": None
            }
        """

        providers_to_try = [model_preference or self.primary] + self.fallback_order

        for provider in providers_to_try:
            try:
                if provider == LLMProvider.GIGACHAT:
                    return self._call_gigachat(message)
                elif provider == LLMProvider.YANDEXGPT:
                    return self._call_yandexgpt(message)
                elif provider == LLMProvider.CLAUDE:
                    return self._call_claude(message)
            except Exception as e:
                print(f"[!]  {provider.value} ошибка: {str(e)}")
                continue

        # Если все провайдеры не сработали
        return {
            "text": None,
            "provider": None,
            "error": "Все провайдеры недоступны"
        }

    def _call_gigachat(self, message: str) -> dict:
        """Вызвать GigaChat"""

        # Получить токен
        token = self._get_gigachat_token()

        # Отправить запрос
        url = "https://gigachat.devices.sberbank.ru/api/v1/chat/completions"
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "RQ-UUID": str(uuid.uuid4())
        }

        payload = {
            "model": "GigaChat-2",
            "messages": [{"role": "user", "content": message}],
            "temperature": 0.7,
            "max_tokens": 512
        }

        response = requests.post(url, headers=headers, json=payload, timeout=30)
        response.raise_for_status()

        result = response.json()
        return {
            "text": result["choices"][0]["message"]["content"],
            "provider": "gigachat",
            "tokens": result.get("usage", {}).get("total_tokens"),
            "error": None
        }

    def _get_gigachat_token(self) -> str:
        """Получить access token для GigaChat"""

        if self.gigachat_config["token"]:
            return self.gigachat_config["token"]

        credentials = f"{self.gigachat_config['client_id']}:{self.gigachat_config['client_secret']}"
        encoded = base64.b64encode(credentials.encode()).decode()

        headers = {
            "Authorization": f"Basic {encoded}",
            "RQ-UUID": str(uuid.uuid4()),
            "Content-Type": "application/x-www-form-urlencoded"
        }

        data = {
            "grant_type": "client_credentials",
            "scope": "GIGACHAT_API_PERS"
        }

        response = requests.post(
            "https://ngw.devices.sberbank.ru:9443/api/v2/oauth",
            headers=headers,
            data=data,
            timeout=10
        )
        response.raise_for_status()

        token = response.json()["access_token"]
        self.gigachat_config["token"] = token
        return token

    def _call_yandexgpt(self, message: str) -> dict:
        """Вызвать YandexGPT"""

        url = "https://llm.api.cloud.yandex.net/foundationModels/v1/completion"

        headers = {
            "Authorization": f"Bearer {self.yandex_config['access_token']}",
            "Content-Type": "application/json"
        }

        payload = {
            "modelUri": f"gpt://{self.yandex_config['folder_id']}/yandexgpt",
            "completionOptions": {
                "stream": False,
                "temperature": 0.7,
                "maxTokens": "512"
            },
            "messages": [
                {
                    "role": "user",
                    "text": message
                }
            ]
        }

        response = requests.post(url, headers=headers, json=payload, timeout=30)
        response.raise_for_status()

        result = response.json()
        return {
            "text": result["result"]["alternatives"][0]["message"]["text"],
            "provider": "yandexgpt",
            "tokens": result.get("result", {}).get("usage", {}).get("total_tokens"),
            "error": None
        }

    def _call_claude(self, message: str) -> dict:
        """Вызвать Claude API (как fallback)"""

        try:
            import anthropic
        except ImportError:
            raise ImportError("Установите anthropic: pip install anthropic")

        client = anthropic.Anthropic(api_key=self.claude_config["api_key"])

        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=512,
            messages=[
                {"role": "user", "content": message}
            ]
        )

        return {
            "text": response.content[0].text,
            "provider": "claude",
            "tokens": response.usage.output_tokens,
            "error": None
        }

# Использование
client = UniversalLLMClient(primary_provider=LLMProvider.GIGACHAT)

# Простой запрос
result = client.chat("Объясни, что такое API, очень просто")
print(f"[+] Провайдер: {result['provider']}")
print(f"Ответ: {result['text']}")
print(f"Токенов: {result['tokens']}")

# С явным выбором провайдера
result = client.chat(
    "Напиши SQL запрос",
    model_preference=LLMProvider.YANDEXGPT
)

Пример 4: Универсальная обёртка с логированием

Для production-систем важно отслеживать, какой провайдер использовался и сколько стоил запрос:

import json
from datetime import datetime
from dataclasses import dataclass, asdict

@dataclass
class LLMCost:
    """Стоимость запроса"""
    input_tokens: int
    output_tokens: int
    provider: str

    @property
    def total_price_rub(self) -> float:
        """Рассчитать стоимость в рублях"""

        rates = {
            "gigachat": 0.40 / 1000,  # 0.40 ₽ за 1K токенов (GigaChat-2 Lite)
            "yandexgpt": 0.40 / 1000,  # 0.40 ₽ за 1K токенов (YandexGPT 5.1 Pro)
        }

        rate = rates.get(self.provider, 0)
        return (self.input_tokens + self.output_tokens) * rate

class LLMLogger:
    """Логирование всех LLM запросов"""

    def __init__(self, log_file="llm_calls.json"):
        self.log_file = log_file

    def log_call(self, prompt: str, response: str, provider: str, tokens: int):
        """Логировать запрос"""

        entry = {
            "timestamp": datetime.now().isoformat(),
            "provider": provider,
            "prompt_length": len(prompt),
            "response_length": len(response),
            "tokens": tokens,
            "cost_rub": LLMCost(
                input_tokens=len(prompt.split()),
                output_tokens=tokens or 0,
                provider=provider
            ).total_price_rub
        }

        with open(self.log_file, "a") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

        print(f"Стоимость запроса: {entry['cost_rub']:.4f} ₽")

# Использование
logger = LLMLogger()
result = client.chat("Привет")
logger.log_call(
    prompt="Привет",
    response=result["text"],
    provider=result["provider"],
    tokens=result["tokens"]
)

Архитектура с fallback (Mermaid)

graph TD
    A[User Request] --> B{Try GigaChat?}
    B -->|Success| C["[+] GigaChat Response"]
    B -->|Fail| D{Try YandexGPT?}
    D -->|Success| E["[+] YandexGPT Response"]
    D -->|Fail| F{Try Claude?}
    F -->|Success| G["[+] Claude Response"]
    F -->|Fail| H["[-] All providers failed"]

    C --> I[Log & Return]
    E --> I
    G --> I
    H --> I

    I --> J["Response to User"]

    classDef success fill:#90EE90
    classDef fail fill:#FF6B6B
    classDef neutral fill:#87CEEB

    class C,E,G success
    class H fail
    class A,J neutral

Пример 5: LangChain с российскими моделями

Обе модели официально поддерживаются в LangChain и LlamaIndex.

GigaChat + LangChain — официальный партнёрский пакет:

pip install langchain-gigachat

from langchain_gigachat import GigaChat

llm = GigaChat(
    credentials="<Client_ID>:<Client_Secret>",  # Base64 не нужен — SDK делает сам
    scope="GIGACHAT_API_PERS",
    model="GigaChat-2",
    verify_ssl_certs=False  # для dev; в prod используйте сертификат
)

response = llm.invoke("Объясни, что такое REST API")
print(response.content)

YandexGPT + LangChain:

pip install langchain-community yandexcloud

from langchain_community.llms import YandexGPT

llm = YandexGPT(
    api_key="<API-ключ сервисного аккаунта>",
    folder_id="<folder_id>",
    model_uri="gpt://<folder_id>/yandexgpt/latest"
)

response = llm.invoke("Расскажи про машинное обучение")
print(response)

Также поддерживается embeddings через YandexGPTEmbeddings из langchain_community.embeddings.yandex.

Попробуйте сами

1. Создайте .env файл:

GIGACHAT_CLIENT_ID=your_id
GIGACHAT_CLIENT_SECRET=your_secret
YANDEX_ACCESS_TOKEN=your_token
YANDEX_FOLDER_ID=your_folder
ANTHROPIC_API_KEY=your_key

2. Установите зависимости:

pip install requests anthropic python-dotenv

3. Скопируйте класс UniversalLLMClient из примера 3

4. Тестируйте fallback:

from fallback_client import UniversalLLMClient, LLMProvider

# Сценарий 1: обычный запрос (с автоматическим fallback)
client = UniversalLLMClient(primary_provider=LLMProvider.GIGACHAT)
result = client.chat("Привет!")
print(result)

# Сценарий 2: явное использование YandexGPT
result = client.chat(
    "Напиши функцию на Python для сортировки",
    model_preference=LLMProvider.YANDEXGPT
)
print(result)

# Сценарий 3: логирование всех запросов
logger = LLMLogger()
for i in range(5):
    result = client.chat(f"Вопрос {i}")
    logger.log_call(
        f"Вопрос {i}",
        result["text"],
        result["provider"],
        result["tokens"]
    )

Задачи для упражнения:

1. Создайте UniversalLLMClient и убедитесь, что fallback работает (отключите основной провайдер намеренно)
2. Добавьте логирование всех запросов в JSON файл
3. Рассчитайте общую стоимость за 10 запросов к разным провайдерам
4. Создайте функцию benchmark(), которая отправляет одинаковый вопрос всем трём провайдерам и сравнивает время и качество

Ключевые выводы

[+] Запомните:

1. Fallback-логика — критична для production систем (провайдеры могут быть недоступны)
2. n8n + Dify — no-code решения для автоматизации; позволяют использовать LLM без написания кода
3. Универсальный wrapper упрощает переключение между провайдерами
4. Логирование — важно для контроля затрат и отладки
5. gRPC быстрее, но REST проще интегрировать в существующие системы

Дополнительные материалы:

• n8n Docs
• Dify Docs
• LangChain Custom LLM

Следующий урок

В Уроке 4: Когда выбирать российские модели мы разберём критерии выбора между GigaChat, YandexGPT, западными моделями и Claude.

Узнаем, когда российские модели — обязательны (закон ФЗ-152), а когда это просто удобный выбор.