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

Вместо простого:

• Пользователь: «Сколько будет 2+2?»

• Бот: «4»

Мы создадим агента, который может:

• Пользователь: «Привет! Меня зовут Алексей, я работаю Python‑разработчиком»

• Агент: «Приятно познакомиться, Алексей! Как дела в мире Python? Над какими проектами сейчас работаешь?»

• Пользователь: «Разрабатываю систему аналитики. Кстати, напомни мне через час позвонить заказчику»

• Агент: «Отличная задача для Python‑разработчика! Запомнил: поставлю напоминание Алексею на 15:30 — позвонить заказчику по проекту аналитики»

Звучит как научная фантастика? На самом деле, это уже реальность, доступная каждому разработчику благодаря LangGraph.

Добро пожаловать во вторую часть нашего путешествия в мир создания интеллектуальных агентов! Если в первой части мы заложили фундамент, разобравшись с архитектурой LangGraph, узлами, рёбрами и состояниями графа, то сейчас пришло время вдохнуть жизнь в наши конструкции.

От статических графов к живому интеллекту

Современные AI-агенты должны решать задачи, которые ещё недавно казались невозможными:

• Поддерживать многоходовые диалоги с сохранением контекста на протяжении всей беседы

• Адаптировать стиль общения в зависимости от собеседника и ситуации

• Интегрироваться с внешними системами, предоставляя структурированные ответы в формате JSON

• Работать с различными типами сообщений — от простого текста до сложных мультимодальных данных

Что нас ждёт в этой части

К концу сегодняшней публикации вы сможете:

• Создать чат-бота, который помнит имя пользователя и контекст через 100+ сообщений

• Построить агента, возвращающего только валидный JSON для интеграции с API

• Интегрировать несколько разных LLM в одном графе для специализированных задач

• Сохранять состояние агента между перезапусками приложения

В рамках практической работы мы разберём:

Интеграция нейросетей в графы

• Научимся подключать различные LLM к узлам наших графов, разберёмся с механизмами принятия решений и оптимизацией производительности.

Управление типами сообщений

• Изучим систему сообщений LangGraph, поймём разницу между HumanMessage, AIMessage и SystemMessage, а также их практическое применение.

Контекстная память агентов

• Разберёмся, как различные нейросети могут совместно работать с общим контекстом, обмениваться информацией и строить связные диалоги.

Гарантированное получение структурированных ответов

• Освоим техники получения валидного JSON от языковых моделей — критически важный навык для интеграции с backend-системами и создания production-ready приложений.

Персистентность состояний

• Рассмотрим способы сохранения памяти агентов между сессиями и организации долговременного хранения контекста.

Пришло время превратить теоретические знания в мощный практический инструментарий для создания по-настоящему умных AI-агентов!

Инициализация LLM: подготовка нейросетей для интеграции в графы

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

В экосистеме LangChain существует четыре основных подхода к инициализации нейросетей, каждый из которых имеет свои преимущества и области применения.

Подход 1: Универсальный метод init_chat_model

Самый простой способ быстро подключить популярную модель — использовать универсальный метод инициализации:

import os
from langchain.chat_models import init_chat_model

# Устанавливаем API-ключ в переменные окружения
os.environ["OPENAI_API_KEY"] = "sk-..."

# Современные модели 2025 года
llm = init_chat_model("openai:gpt-4o-2024-11-20")     # Последняя стабильная версия
# или новейшие reasoning модели:
llm = init_chat_model("openai:o1-preview")            # Модели с цепочками рассуждений
llm = init_chat_model("anthropic:claude-3-5-sonnet")  # Актуальный Claude
llm = init_chat_model("deepseek:deepseek-chat")       # Экономичная альтернатива

Преимущества:

• Минимальный код для запуска

• Автоматическое определение API-ключей из переменных окружения

• Поддержка всех популярных провайдеров

• Идеально для прототипирования и быстрых экспериментов

Ограничения:

• Ограниченные возможности тонкой настройки

• Меньший контроль над параметрами модели

• Не всегда подходит для production-решений с специфическими требованиями

Подход 2: Официальные специализированные пакеты

Для более глубокого контроля над поведением моделей рекомендуется использовать специализированные пакеты:

# Установка: pip install langchain-openai
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o-2024-11-20",
    temperature=0.7,          # Креативность ответов
    max_tokens=2000,          # Максимум токенов в ответе
    timeout=30,               # Таймаут запроса
    max_retries=3,            # Количество повторных попыток
    streaming=True            # Потоковая передача ответов
)

Актуальная таблица провайдеров и библиотек (2025)

Преимущества специализированных библиотек:

• Полный контроль параметров — temperature, max_tokens, stop_sequences и другие

• Расширенная обработка ошибок — настройка retry-логики и таймаутов

• Специфические возможности — функции, доступные только для конкретных провайдеров

• Production-готовность — оптимизированные для высоконагруженных систем

Подход 3: Неофициальные специализированные пакеты

Помимо огромного количества официальных пакетов для интеграции с топовыми нейросетями, LangChain предоставил полноценный инструментарий, который позволяет обернуть в их оболочку практически любой API-протокол, работающий с нейросетями.

Множество компаний использовало эти инструменты для создания интеграции с экосистемой LangChain. В результате мы получили готовые неофициальные библиотеки от таких провайдеров как Amvera Cloud (официальный доступ к моделям LLaMA и ChatGPT без VPN с пополнением через карты РФ), GigaChat (Сбер), YandexGPT и многих других.

Пример интеграции с Amvera

Amvera предоставляет доступ к современным нейросетям: Llama3.3 70B, Llama 3.1 8B, GPT-4.1, GPT-5 через единый API.

Установка:

pip install langchain langchain-amvera

Получение токена:

1. Регистрируемся на Amvera Cloud

2. Переходим в раздел LLM проектов

3. Выбираем нужную модель (каждая включает бесплатные токены для тестирования)

4. Копируем токен из документации выбранной модели

Код интеграции:

from langchain_amvera import AmveraLLM
from dotenv import load_dotenv
import os

load_dotenv()

# Поддерживаемые модели: llama8b, llama70b, gpt-4.1, gpt-5
llm = AmveraLLM(model="llama70b", api_token=os.getenv("AMVERA_API_TOKEN"))

response = llm.invoke("Объясни принципы работы нейросетей простым языком")
print(response.content)

Пример ответа:

Нейросети работают по принципу, схожему с человеческим мозгом. 
Представь сеть из взаимосвязанных узлов (нейронов), где каждый узел получает
информацию, обрабатывает её и передаёт результат дальше...

Пример интеграции с GigaChat (Сбер)

Установка:

pip install langchain-gigachat

Получение токена:

1. Входим на Сбер Developer (через Сбер ID)

2. Создаём проект

3. Получаем новый ключ в разделе "API ключи"

Код интеграции:

from langchain_gigachat.chat_models import GigaChat
from dotenv import load_dotenv
import os

load_dotenv()

llm = GigaChat(
    model="GigaChat-2-Max",
    credentials=os.getenv("GIGACHAT_CREDENTIALS"),
    verify_ssl_certs=False
)

response = llm.invoke("Расскажи о своих возможностях")
print(response.content)

Файл .env:

AMVERA_API_TOKEN=your_amvera_token_here
GIGACHAT_CREDENTIALS=your_gigachat_credentials_here

Подход 4: Прямая интеграция через API

Если вы хотите полный контроль над запросами или работаете с API, не имеющими готовых LangChain-интеграций, можете использовать прямые HTTP-запросы:

Простой HTTP-запрос (aiohttp)

import aiohttp
import asyncio


async def ask_amvera_llm(token: str, model_name: str, messages: list):
    url = f"https://kong-proxy.yc.amvera.ru/api/v1/models/gpt"
    headers = {
        "accept": "application/json",
        "Content-Type": "application/json",
        "X-Auth-Token": f"Bearer {token}",
    }
    data = {
        "model": model_name,
        "messages": messages
    }
    async with aiohttp.ClientSession() as session:
        async with session.post(url, headers=headers, json=data) as response:
            response.raise_for_status()
            result = await response.json()
            return result


# Пример вызова с сообщениями
async def main():
    token = "полученный токен"
    model = "gpt-5"
    messages = [
        {"role": "system", "text": "Ты полезный ассистент"},
        {"role": "user", "text": "Привет, как дела?"},
    ]
    response = await ask_amvera_llm(token, model, messages)
    print(response)


# Запуск примера
if __name__ == "__main__":
    asyncio.run(main())

Amvera Cloud не предоставляет нативной интеграции с OpenAI без использования неофициального адаптера. В приведённом выше примере показал, как выполнить прямой вызов. Далее остаётся лишь добавить функцию вызова в граф.

Через OpenAI SDK (для совместимых API)

from openai import OpenAI

client = OpenAI(
    api_key="your_openai_key"
    # base_url не указывается, если используете официальный сервис OpenAI
)

def llm_node(state):
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",  # или "gpt-4"
        messages=[
            {"role": "system", "content": "Ты полезный ассистент"},
            {"role": "user", "content": state["user_message"]}
        ]
    )
    return {"ai_response": response.choices[0].message.content}

Обратите внимание: на территории РФ без использования VPN или прокси недоступны нейросети вроде OpenAI (ChatGPT), Claude и Grok. В качестве альтернативы можно воспользоваться решениями Amvera или платформой OpenRouter, где собраны десятки моделей от различных разработчиков.

Важное предупреждение:

При использовании прямых API-вызовов вы теряете множество полезных возможностей LangChain: автоматический retry, кэширование, обработка ошибок, единообразие интерфейсов и интеграцию с инструментами мониторинга. Поэтому всегда рекомендую использовать LangChain-интеграции, когда они доступны.

Выбор правильного подхода

Что дальше?

В следующем разделе мы рассмотрим, как эти инициализированные модели встраиваются в архитектуру LangGraph и начинают принимать решения на основе состояния графа и входящих сообщений. Узнаем, как различные типы сообщений влияют на поведение агентов и как обеспечить бесшовную передачу контекста между узлами.

Готовы превратить статические узлы в интеллектуальных агентов? Тогда переходим к практической интеграции!

Сообщения и диалоговый контекст

Напоминаю, что сегодня мы не будем касаться темы инструментов (tools, MCP). Это позволит нам лучше сосредоточиться на других моментах. В частности, важнейшая часть взаимодействия с ИИ — это сообщения и сохранение диалогового контекста. В этом разделе с данным вопросом ознакомимся детально.

Простой способ общения с ИИ

Технически, LangChain позволяет отправлять сообщения ИИ даже в таком упрощённом формате:

llm.invoke("Кто тебя создал?")

В результате мы получим ответ от ИИ и сможем с ним работать. Давайте рассмотрим такой ответ, используя официальный адаптер OpenAi от LangChain.

Установка:

pip install langchain-openai

Настройка (файл .env):

OPENAI_API_KEY=sk-e7c13...

Пример кода:

from langchain.chat_models import ChatOpenAI

llm = ChatOpenAI(model_name="gpt-4")  # Инициализация модели OpenAI
response = llm.invoke([{"role": "user", "content": "Кто тебя создал?"}])

print(f"Тип ответа: {type(response)}")
print(f"Содержимое: {response[0].message.content}")

Результат:

Тип ответа: <class 'langchain_core.messages.ai.AIMessage'>
Содержимое: "Меня создала команда OpenAI, специализирующаяся на разработке искусственного интеллекта.

Что происходит под капотом

Обратите внимание — в данном примере мы неявно задействовали сразу два типа сообщений:

• HumanMessage — LangChain автоматически обернул наше сообщение в этот формат

• AIMessage — автоматически создался из ответа модели

Это не случайность. Типы сообщений нужны языковым моделям для понимания ролей участников диалога.

Три основных типа сообщений

SystemMessage — "Это твоя роль и инструкции"

• Определяет поведение и характер ИИ-агента

• Устанавливает контекст и правила работы

• Обычно размещается в начале диалога

HumanMessage — "Это говорит пользователь"

• Все сообщения от человека

• Вопросы, команды, информация от пользователя

• Основной способ ввода данных в систему

AIMessage — "Это твой предыдущий ответ"

• Ответы нейросети из истории диалога

• Позволяет модели "помнить" свои предыдущие высказывания

• Критично для поддержания последовательности

Работа с сообщениями явным образом

Для полного контроля над диалогом импортируем типы сообщений:

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage


messages = [
    SystemMessage(content="Ты полезный программист-консультант"),
    HumanMessage(content="Как написать цикл в Python?"),
    AIMessage(content="Используйте for или while. Пример: for i in range(10):"),
    HumanMessage(content="А что такое range?")
]


# Отправляем структурированную историю диалога
response = llm.invoke(messages)

Ответ:

`range()` — это встроенная функция Python, которая генерирует последовательность чисел.
Она очень полезна для создания циклов `for`.

**Основные способы использования:**

1. range(stop)...

Пример выше демонстрирует, как видит контекст общения нейросеть. Точнее, то как ей проще ориентироваться — и тут мы замечаем первую важнейшую особенность LangChain: возможность чёткого распределения ролей в сообщениях с целью высокого качества сохранения контекста.

Почему структура сообщений критически важна

Сравните два подхода:

# Плохо - всё в одной строке
bad_context = "Система: Ты помощник. Человек: Привет. ИИ: Привет! Человек: Как дела?"


# Хорошо - структурированные сообщения
good_context = [
    SystemMessage(content="Ты полезный помощник"),
    HumanMessage(content="Привет"),
    AIMessage(content="Привет! Как дела?"),
    HumanMessage(content="Как дела?")
]

Проблемы неструктурированного подхода:

• Нейросеть не понимает, где заканчивается одно сообщение и начинается другое

• Теряется информация о ролях участников диалога

• Контекст превращается в «кашу» из слов без чёткой логики

• Качество ответов резко снижается при длинных диалогах

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

• Чёткое разделение ролей и ответственности

• Сохранение логики диалога на протяжении всей беседы

• Возможность точного управления контекстом

• Высокое качество ответов даже в сложных сценариях

Практический пример: многоходовой диалог

Давайте создадим полноценный диалог с сохранением контекста:

from langchain.chat_models import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage


llm = ChatOpenAI(model_name="gpt-4")


def chat_with_context():
    # Инициализация диалога с системным сообщением
    messages = [
        SystemMessage(content="Ты дружелюбный помощник-программист. Запоминай информацию о пользователе.")
    ]

    # Первое сообщение пользователя
    user_input_1 = "Привет! Меня зовут Алексей, я изучаю Python"
    messages.append(HumanMessage(content=user_input_1))

    response_1 = llm.invoke(messages)
    messages.append(response_1)  # Добавляем ответ ИИ в историю
    print(f"ИИ: {response_1.content}")

    # Второе сообщение - проверяем память
    user_input_2 = "Как меня зовут и что я изучаю?"
    messages.append(HumanMessage(content=user_input_2))

    response_2 = llm.invoke(messages)
    messages.append(response_2)
    print(f"ИИ: {response_2.content}")

    # Третье сообщение - продолжение темы
    user_input_3 = "Посоветуй мне книгу по моей теме изучения"
    messages.append(HumanMessage(content=user_input_3))

    response_3 = llm.invoke(messages)
    print(f"ИИ: {response_3.content}")

    print(f"\nОбщее количество сообщений в истории: {len(messages)}")
    return messages


# Запуск диалога
history = chat_with_context()

Важный момент: Сейчас вы должны закрепить, что контекст диалога — это всего лишь набор системных, человеческих и ИИ-сообщений, объединённых в массиве. Для простых примеров достаточно в качестве такого массива использовать простой Python-список, в который вы будете помещать сообщения с метками о их типе.

Для более сложных структур стоит использовать базы данных (в том числе векторные) и разные фишки по "умному обрезанию контекста", но это уже тема другой большой беседы.

Подстановка собственных ответов: мощный трюк для управления диалогом

Более того, если вы внимательно посмотрите на структуру ответа, то заметите очень интересную возможность — вы можете создавать собственные AIMessage и подставлять их в контекст диалога. Это открывает множество продвинутых сценариев использования.

Создание "фиктивных" ответов ИИ

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage


# Создаём диалог с подставленным ответом
messages = [
    SystemMessage(content="Ты эксперт по Python"),
    HumanMessage(content="Что такое списки в Python?"),
    # Подставляем свой "ответ ИИ"
    AIMessage(content="Списки в Python — это упорядоченные коллекции элементов, которые можно изменять"),
    HumanMessage(content="Приведи пример работы со списками")
]


response = llm.invoke(messages)
print(response.content)

В данном примере модель будет считать, что она уже отвечала на вопрос о списках именно так, как мы указали в AIMessage, и продолжит диалог с учётом этого "факта".

Практические применения этого трюка

1. Мультимодельные диалоги

Можно комбинировать ответы разных нейросетей в одном диалоге:

from langchain_openai import ChatOpenAI
from langchain_amvera import AmveraLLM


gpt = ChatOpenAI(model="gpt-4o")
amvera = AmveraLLM(model="llama70b")

messages = [
    SystemMessage(content="Ты помощник по программированию"),
    HumanMessage(content="Объясни ООП в Python")
]

# Получаем ответ от DeepSeek
amvera_response = deepseek.invoke(messages)

# Добавляем его как AIMessage и продолжаем с GPT
messages.append(amvera_response)
messages.append(HumanMessage(content="Теперь покажи практический пример"))

# GPT отвечает, считая что предыдущий ответ дал он сам
gpt_response = gpt.invoke(messages)
print(f"Продолжение от GPT: {gpt_response.content}")

2. Создание экспертных персон

def create_expert_persona(expertise_area):
    """Создаём экспертную персону через подставленные ответы"""
    return [
        SystemMessage(content=f"Ты эксперт в области {expertise_area}"),
        HumanMessage(content="Расскажи о себе"),
        AIMessage(content=f"Я специализируюсь на {expertise_area} уже более 10 лет. "
                         f"Помогаю разработчикам решать сложные задачи и делюсь практическим опытом."),
        HumanMessage(content="Какой у тебя подход к обучению?"),
        AIMessage(content="Я предпочитаю объяснять сложные концепции через практические примеры "
                         "и реальные кейсы. Теория важна, но практика — ещё важнее!")
    ]

# Создаём эксперта по машинному обучению
ml_expert_context = create_expert_persona("машинное обучение")
ml_expert_context.append(HumanMessage(content="Объясни мне нейронные сети"))

response = llm.invoke(ml_expert_context)
print(response.content)  # Ответ будет в стиле опытного ML-эксперта

3. Контроль качества и коррекция ответов

def improve_response(original_response):
    """Улучшаем ответ ИИ перед добавлением в контекст"""
    if len(original_response.content) < 50:
        # Если ответ слишком короткий, заменяем на более развёрнутый
        return AIMessage(
            content=f"{original_response.content}\n\nПозвольте мне дать более подробное объяснение..."
        )
    return original_response

  
# Использование
messages = [HumanMessage(content="Что такое Python?")]
response = llm.invoke(messages)
improved = improve_response(response)
messages.append(improved)  # Добавляем улучшенную версию

Важные моменты при использовании

Осторожность с противоречиями:

# Плохо - создаём противоречивый контекст
messages = [
    HumanMessage(content="Сколько будет 2+2?"),
    AIMessage(content="2+2 = 5"),  # Неправильный "ответ ИИ"
    HumanMessage(content="А сколько будет 3+3?")
]
# Модель может продолжить давать неправильные ответы!

Хорошо - поддерживаем логичность:

messages = [
    HumanMessage(content="Объясни принцип DRY"),
    AIMessage(content="DRY (Don't Repeat Yourself) — принцип программирования, "
                     "согласно которому следует избегать дублирования кода"),
    HumanMessage(content="Как применить DRY на практике?")
]
# Логичное продолжение темы

Управление длиной контекста

При длинных диалогах возникает проблема ограничений контекста. У каждой модели есть лимит токенов:

• GPT-4o — до 128К токенов

• DeepSeek-V3 — до 64К токенов

• Claude-3.5 — до 200К токенов

Стратегии управления контекстом

def manage_context_length(messages, max_messages=20):
    """Простая стратегия: сохраняем системное сообщение + последние N сообщений"""
    if len(messages) <= max_messages:
        return messages

    # Выделяем системные сообщения
    system_messages = [msg for msg in messages if isinstance(msg, SystemMessage)]
    dialog_messages = [msg for msg in messages if not isinstance(msg, SystemMessage)]

    # Берём последние сообщения диалога
    recent_messages = dialog_messages[-(max_messages - len(system_messages)):]

    return system_messages + recent_messages

# Применение при каждом запросе
def smart_invoke(llm, messages):
    managed_messages = manage_context_length(messages)
    return llm.invoke(managed_messages)

Анализ метаданных сообщений

AIMessage содержит полезную техническую информацию:

response = llm.invoke("Расскажи о языке Python")

print(f"Содержимое: {response.content[:100]}...")
print(f"ID сообщения: {response.id}")

# Метаданные о генерации
metadata = response.response_metadata
print(f"Использовано токенов: {metadata.get('token_usage', {})}")
print(f"Модель: {metadata.get('model_name')}")
print(f"Причина завершения: {metadata.get('finish_reason')}")

# Информация о токенах для оптимизации
usage = response.usage_metadata
print(f"Входящие токены: {usage.get('input_tokens')}")
print(f"Исходящие токены: {usage.get('output_tokens')}")

Техническая реализация в LangGraph

В контексте LangGraph подстановка AIMessage особенно полезна для создания узлов-фильтров:

def response_filter_node(state):
    """Узел-фильтр для коррекции ответов"""
    last_message = state["messages"][-1]

    if isinstance(last_message, AIMessage):
        # Проверяем и корректируем ответ
        if "извините" in last_message.content.lower():
            # Заменяем на более уверенный ответ
            corrected = AIMessage(
                content=last_message.content.replace("Извините", "Позвольте уточнить")
            )
            # Заменяем последнее сообщение
            new_messages = state["messages"][:-1] + [corrected]
            return {"messages": new_messages}

    return state  # Возвращаем без изменений

Ключевые принципы работы с контекстом

1. Всегда используйте типизированные сообщения для диалогов длиннее одного обмена

2. SystemMessage задаёт тон — размещайте его в начале для настройки поведения

3. Сохраняйте историю в списке — порядок сообщений критически важен

4. Контролируйте длину контекста — избегайте превышения лимитов модели

5. Используйте метаданные — отслеживайте потребление токенов и производительность

6. Подстановка AIMessage — мощный инструмент для создания сложных диалоговых сценариев

Этот мощный механизм открывает безграничные возможности для тонкой настройки поведения ИИ-агентов и создания сложных мультимодельных систем!

В следующем разделе мы применим эти знания для создания первого полноценного диалогового агента в LangGraph, который сможет вести осмысленные беседы с сохранением контекста на любое количество ходов.

Интеграция в LangGraph: создание первого диалогового агента

Мини-курс, всё таки, про LangGraph, поэтому пора переходить к графам. Далее я буду считать, что вы ознакомились с первой частью данного мини-курса. В частности, у вас должно быть базовое понимание работы с состояниями в LangGraph, узлами, рёбрами и условными узлами. Сейчас эти навыки нам понадобятся.

Далее рассмотрим несколько примеров разработки диалогов с ИИ, начиная от простых примеров для более мягкого погружения и заканчивая более сложными.

Напоминаю, что полный код из этой статьи, а также эксклюзивный контент, который я не публикую на Хабре, доступен в моем бесплатном телеграм-канале "Легкий путь в Python". В сообществе уже более 4600 участников.

Архитектура простого диалогового агента

Прежде чем погрузиться в код, давайте разберёмся с архитектурой нашего первого агента:

START → [Ввод пользователя] → [Ответ ИИ] → [Проверка продолжения]
                                ↑                      ↓
                                └─── Продолжить ←──────┘

                                     Завершить → END

Наш граф состоит из трёх ключевых компонентов:

• Узел ввода — получает сообщения от пользователя и проверяет команды выхода

• Узел ИИ — генерирует ответ на основе полного контекста диалога

• Условное ребро — принимает решение о продолжении или завершении беседы

Практическая реализация: чат с сохранением контекста

Рассмотрим первый простой пример: чат с ИИ с сохранением контекста и с выходом из диалога, когда пользователь сам решит прервать его. В качестве примера использую адаптер от Amvera Cloud.

Подготовка импортов и окружения

from dotenv import load_dotenv
from langchain_amvera import AmveraLLM
from langchain_core.messages import SystemMessage, HumanMessage, BaseMessage, AIMessage
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, List

# Выгружаем переменные окружения
load_dotenv()

Из того, что мы ранее не рассматривали — вы можете заметить BaseMessage. Это базовый класс, на котором основаны все классы сообщений в LangChain. Чуть позже вы увидите, как он используется.

Определение состояния диалога

class ChatState(TypedDict):
    messages: List[BaseMessage]
    should_continue: bool

Данный класс содержит 2 переменные:

• messages — список любых сообщений LangChain (SystemMessage, HumanMessage, AIMessage)

• should_continue — булевая переменная, которая указывает на продолжение или остановку диалога

Почему именно List[BaseMessage]?

Использование базового типа даёт нам гибкость — мы можем хранить любые типы сообщений в одном списке, не ограничиваясь конкретными классами.

Инициализация нейросети

llm = AmveraLLM(model="llama70b")

Узловые функции

Узел пользовательского ввода

def user_input_node(state: ChatState) -> dict:
    """Узел для получения ввода пользователя"""
    user_input = input("Вы: ")

    # Проверяем команды выхода
    if user_input.lower() in ["выход", "quit", "exit", "пока", "bye"]:
        return {"should_continue": False}

    # Добавляем сообщение пользователя
    new_messages = state["messages"] + [HumanMessage(content=user_input)]
    return {"messages": new_messages, "should_continue": True}

Достаточно простая функция. На входе будет принимать сообщение от пользователя, и если оно будет содержать «стоп-слова», то будет менять переменную продолжения на False, иначе True.

Важная деталь: Мы не мутируем существующий список сообщений, а создаём новый. Это соответствует принципам функционального программирования и предотвращает неожиданные побочные эффекты.

Решил не усложнять данный пример. Всему своё время. В реальной практике решение об остановке диалога вполне может принимать нейросеть. Вопрос в правильной настройке.

Узел ответа ИИ

def llm_response_node(state: ChatState) -> dict:
    """Узел для генерации ответа ИИ"""
    # Получаем ответ от LLM, передавая весь контекст
    response = llm.invoke(state["messages"])
    msg_content = response.content

    # Выводим ответ
    print(f"ИИ: {msg_content}")

    # Добавляем ответ в историю как AIMessage
    new_messages = state["messages"] + [AIMessage(content=msg_content)]
    return {"messages": new_messages}

Теперь добавим функцию, которая будет вызывать нейросеть. Посмотрите на код внимательно.

Мы уже ранее вызывали нейросеть, но теперь вместо простой передачи сообщения мы каждый раз достаём весь контекст (все сообщения). По этому принципу работают большие чат-модели, как Claude или ChatGPT. То есть, это наглядный пример «памяти» нейросетей.

После того как ответ получен, мы извлекаем из него только текст и помещаем его в массив сообщений с конкретной пометкой.

Плюсы такого подхода:

• Чистый контекст без технических метаданных

• Экономия токенов (метаданные тоже считаются!)

• Явная демонстрация создания AIMessage

• Контроль над тем, что попадает в историю

Выше я указал простой и лаконичный пример хранения сообщений от ИИ, но в реальных системах стоит сохранять полный AIMessage объект вместо извлечения только текста. Дело в том, что response содержит важные метаданные: информацию о потраченных токенах, времени выполнения запроса и, что критически важно для будущих инструментов, данные о вызовах внешних функций. Для учебных примеров текущий подход идеален, но в production лучше использовать new_messages = state["messages"] + [response] — это поможет при отладке и мониторинге.

Условная функция продолжения

def should_continue(state: ChatState) -> str:
    """Условная функция для определения продолжения диалога"""
    return "continue" if state.get("should_continue", True) else "end"

Тут уже всё просто. Если should_continue на момент вызова функции True, возвращаем строку "continue", иначе "end".

Создание и компиляция графа

# Создаём граф
graph = StateGraph(ChatState)

# Добавляем узлы
graph.add_node("user_input", user_input_node)
graph.add_node("llm_response", llm_response_node)

# Создаём рёбра
graph.add_edge(START, "user_input")
graph.add_edge("user_input", "llm_response")

# Условное ребро для проверки продолжения
graph.add_conditional_edges(
    "llm_response",
    should_continue,
    {
        "continue": "user_input",  # Возвращаемся к вводу пользователя
        "end": END                 # Завершаем диалог
    }
)

# Компиляция графа
app = graph.compile()

Логика работы графа:

1. START → user_input — начинаем с ввода пользователя

2. user_input → llm_response — передаём сообщение ИИ для ответа

3. llm_response → should_continue — проверяем, нужно ли продолжать

4. should_continue → user_input (если "continue") — новый цикл диалога

5. should_continue → END (если "end") — завершение работы

Запуск диалогового агента

if __name__ == "__main__":
    print("Добро пожаловать в чат с ИИ!")
    print("Для выхода введите: выход, quit, exit, пока, или bye")
    print("-" * 50)

    # Начальное состояние с системным сообщением
    initial_state = {
        "messages": [
            SystemMessage(
                content="Ты дружелюбный помощник. Отвечай коротко и по делу."
            )
        ],
        "should_continue": True
    }

    try:
        # Запуск чата
        final_state = app.invoke(initial_state)

        print("-" * 50)
        print("Чат завершён. До свидания!")
        print(f"Всего сообщений в диалоге: {len(final_state['messages'])}")

    except KeyboardInterrupt:
        print("\n\nЧат прерван пользователем (Ctrl+C)")
    except Exception as e:
        print(f"\nОшибка в работе чата: {e}")

Пример работы агента

Добро пожаловать в чат с ИИ!
Для выхода введите: выход, quit, exit, пока, или bye
--------------------------------------------------
Вы: Привет! Как дела?
ИИ: Привет! Дела хорошо, спасибо! Как у тебя дела? Чем могу помочь?

Вы: Расскажи про Python
ИИ: Python — популярный язык программирования, известный простотой синтаксиса и мощными возможностями. Используется в веб-разработке, анализе данных, машинном обучении и автоматизации. Что именно интересует?

Вы: А какие у него недостатки?
ИИ: Основные недостатки Python:
• Медленная скорость выполнения по сравнению с C++ или Java
• Высокое потребление памяти
• Слабая поддержка многопоточности (GIL)
• Не подходит для мобильной разработки

Вы: пока
--------------------------------------------------
Чат завершён. До свидания!
Всего сообщений в диалоге: 7

Обратите внимание, как ИИ помнит контекст диалога — в третьем ответе он понимает, что недостатки нужно рассказать именно про Python, хотя в последнем сообщении язык программирования явно не упоминался.

Оптимизация и улучшения

Добавление обработки ошибок

def llm_response_node_with_retry(state: ChatState) -> dict:
    """Узел с обработкой ошибок и повторными попытками"""
    max_retries = 3

    for attempt in range(max_retries):
        try:
            response = llm.invoke(state["messages"])
            msg_content = response.content
            print(f"ИИ: {msg_content}")

            new_messages = state["messages"] + [AIMessage(content=msg_content)]
            return {"messages": new_messages}

        except Exception as e:
            if attempt == max_retries - 1:
                # Последняя попытка — возвращаем ошибку пользователю
                error_msg = "Извините, произошла ошибка. Попробуйте ещё раз."
                print(f"ИИ: {error_msg}")
                new_messages = state["messages"] + [AIMessage(content=error_msg)]
                return {"messages": new_messages}
            else:
                print(f"Попытка {attempt + 1} неудачна, повторяю...")
                continue

Контроль длины контекста

def trim_context_if_needed(messages: List[BaseMessage], max_messages: int = 20) -> List[BaseMessage]:
    """Обрезаем контекст, если он становится слишком длинным"""
    if len(messages) <= max_messages:
        return messages

    # Сохраняем системные сообщения + последние сообщения диалога
    system_msgs = [msg for msg in messages if isinstance(msg, SystemMessage)]
    dialog_msgs = [msg for msg in messages if not isinstance(msg, SystemMessage)]

    recent_msgs = dialog_msgs[-(max_messages - len(system_msgs)):]
    return system_msgs + recent_msgs

  
def optimized_llm_response_node(state: ChatState) -&gt; dict:
    """Оптимизированный узел с контролем длины контекста"""
    # Обрезаем контекст при необходимости
    trimmed_messages = trim_context_if_needed(state["messages"])

    response = llm.invoke(trimmed_messages)
    msg_content = response.content
    print(f"ИИ: {msg_content}")

    new_messages = state["messages"] + [AIMessage(content=msg_content)]
    return {"messages": new_messages}

Что может пойти не так: типичные ошибки

Ошибка 1: Мутация состояния

# Неправильно - мутируем существующий список
def bad_user_input_node(state: ChatState) -> dict:
    user_input = input("Вы: ")
    state["messages"].append(HumanMessage(content=user_input))  # Мутация!
    return state

# Правильно - создаём новый список
def good_user_input_node(state: ChatState) -> dict:
    user_input = input("Вы: ")
    new_messages = state["messages"] + [HumanMessage(content=user_input)]
    return {"messages": new_messages}

Ошибка 2: Потеря системного контекста

# Неправильно - можем потерять SystemMessage
def bad_trim_context(messages: List[BaseMessage]) -> List[BaseMessage]:
    return messages[-10:]  # Просто берём последние 10

# Правильно - сохраняем системные сообщения
def good_trim_context(messages: List[BaseMessage]) -> List[BaseMessage]:
    system_msgs = [msg for msg in messages if isinstance(msg, SystemMessage)]
    dialog_msgs = [msg for msg in messages if not isinstance(msg, SystemMessage)]
    return system_msgs + dialog_msgs[-8:]  # Система + последние 8 диалоговых

Ошибка 3: Неправильная обработка пустого ввода

# Неправильно - не обрабатываем пустые сообщения
def bad_user_input_node(state: ChatState) -> dict:
    user_input = input("Вы: ")
    new_messages = state["messages"] + [HumanMessage(content=user_input)]
    return {"messages": new_messages, "should_continue": True}

# Правильно - проверяем пустой ввод
def good_user_input_node(state: ChatState) -> dict:
    user_input = input("Вы: ").strip()

    if not user_input:  # Пустое сообщение
        print("Пожалуйста, введите сообщение.")
        return state  # Возвращаем состояние без изменений

    if user_input.lower() in ["выход", "quit", "exit", "пока", "bye"]:
        return {"should_continue": False}

    new_messages = state["messages"] + [HumanMessage(content=user_input)]
    return {"messages": new_messages, "should_continue": True}

Альтернативные подходы к управлению диалогом

ИИ принимает решение о завершении

def ai_controlled_continuation_node(state: ChatState) -> dict:
    """ИИ сам решает, нужно ли завершить диалог"""

    # Добавляем специальный промпт для принятия решения
    decision_messages = state["messages"] + [
        HumanMessage(
            content="Проанализируй диалог. Если пользователь явно хочет завершить беседу "
                   "или диалог исчерпан, ответь ТОЛЬКО словом 'ЗАВЕРШИТЬ'. "
                   "Иначе продолжи обычный разговор."
        )
    ]

    response = llm.invoke(decision_messages)

    if "ЗАВЕРШИТЬ" in response.content.upper():
        print("ИИ: Было приятно пообщаться! До свидания!")
        return {"should_continue": False}
    else:
        # Обычный ответ
        print(f"ИИ: {response.content}")
        new_messages = state["messages"] + [AIMessage(content=response.content)]
        return {"messages": new_messages, "should_continue": True}

Мы создали первый полноценный диалоговый агент в LangGraph, который:

• Сохраняет контекст диалога между сообщениями

• Корректно завершается по команде пользователя

• Использует типизированные состояния для надёжной работы

• Демонстрирует циклическую логику графа с условными переходами

Ключевые принципы, которые мы изучили:

• Неизменяемость состояний — создаём новые объекты вместо мутации существующих

• Правильная типизация — используем TypedDict для чёткой структуры состояний

• Контроль потока — управляем выполнением через условные рёбра

• Обработка ошибок — предусматриваем сценарии сбоев и восстановления

В следующей главе мы усложним задачу — создадим агента, который может работать с различными типами запросов и возвращать структурированные JSON-ответы для интеграции с внешними системами.

Структурированные JSON-ответы: как всегда получать то, что ждешь

В реальных приложениях AI-агенты должны интегрироваться с базами данных, API и другими системами. Это означает, что нам нужны не красивые диалоги, а строго структурированные данные в предсказуемом формате. К сожалению, языковые модели по природе своей склонны к творчеству, даже когда мы просим их о сухих фактах.

Проблема: когда ИИ слишком "умный"

Представьте, что вы создаете систему анализа отзывов клиентов. От агента требуется простая структура:

{
  "sentiment": "positive",
  "confidence": 0.85,
  "key_topics": ["качество", "доставка"]
}

Но вместо этого получаете:

Конечно, я проанализирую отзыв! Вот результат моего анализа:

{
  "sentiment": "positive", 
  "confidence": 0.85,
  "key_topics": ["качество", "доставка"]
}

Как видите, отзыв довольно позитивный, особенно в части качества товара. 
Надеюсь, это поможет в вашем анализе!

Проблемы такого ответа:

• Невозможно распарсить JSON из-за лишнего текста

• Нестабильный формат — иногда комментарии в начале, иногда в конце

• Нарушение автоматизированных процессов обработки данных

• Увеличение расходов на токены из-за "болтовни" модели

Решение: три ключевые сущности LangChain

Для решения этой проблемы в LangChain есть три фундаментальные сущности, которые работают в связке:

1. Pydantic модель — строгая схема данных

Pydantic — это библиотека для валидации данных в Python. В контексте LangChain она определяет, какую именно структуру JSON мы хотим получить от нейросети.

На Хабре у меня есть подробная статья о данной библиотеке: Pydantic 2: Полное руководство для Python-разработчиков — от основ до продвинутых техник. Рекомендую прочитать, если вы еще не работали с этим инструментом.

from pydantic import BaseModel, Field
from typing import List, Literal


class SentimentAnalysis(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"] = Field(
        description="Тональность отзыва: положительная, отрицательная или нейтральная"
    )
    confidence: float = Field(
        description="Уверенность в анализе от 0.0 до 1.0",
        ge=0.0,  # больше или равно 0
        le=1.0   # меньше или равно 1
    )
    key_topics: List[str] = Field(
        description="Ключевые темы, упомянутые в отзыве",
        max_items=5
    )
    summary: str = Field(
        description="Краткое резюме отзыва в одном предложении",
        max_length=200
    )

Возможности Pydantic для ИИ:

• Ограничение значений через Literal["positive", "negative", "neutral"]

• Валидация диапазонов через ge=0.0, le=1.0

• Ограничение размеров через max_items=5, max_length=200

• Описания полей для лучшего понимания нейросетью

2. JsonOutputParser — переводчик между ИИ и JSON

JsonOutputParser берет Pydantic модель и умеет:

• Генерировать детальные инструкции для нейросети

• Парсить ответ нейросети в валидный Python dict

• Валидировать результат по заданной схеме

from langchain_core.output_parsers import JsonOutputParser

# Создаем парсер на основе нашей модели
parser = JsonOutputParser(pydantic_object=SentimentAnalysis)

print("Что генерирует парсер:")
print(parser.get_format_instructions())

Что генерирует get_format_instructions():

The output should be formatted as a JSON instance that conforms to the JSON schema below.

As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]}
the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema.

Here is the output schema:
{
  "properties": {
    "sentiment": {
      "description": "Тональность отзыва: положительная, отрицательная или нейтральная",
      "enum": ["positive", "negative", "neutral"],
      "title": "Sentiment",
      "type": "string"
    },
    "confidence": {
      "description": "Уверенность в анализе от 0.0 до 1.0",
      "maximum": 1.0,
      "minimum": 0.0,
      "title": "Confidence",
      "type": "number"
    },
    // ... остальные поля
  },
  "required": ["sentiment", "confidence", "key_topics", "summary"]
}

Эти инструкции нейросеть понимает намного лучше, чем наши человеческие объяснения типа "верни JSON".

3. PromptTemplate — умный шаблон промптов

PromptTemplate решает проблему динамической подстановки данных в промпты:

Проблема простых строк:

# Неудобно и не масштабируется
def create_prompt(review, format_instructions):
    return f"""Проанализируй отзыв: {review}
    
{format_instructions}

ТОЛЬКО JSON!"""

# При каждом использовании нужно помнить порядок параметров
prompt1 = create_prompt(review_text, instructions)  # Правильно
prompt2 = create_prompt(instructions, review_text)  # Ошибка!

Решение через PromptTemplate:

from langchain_core.prompts import PromptTemplate

prompt_template = PromptTemplate(
    template="""Проанализируй отзыв: {review}

{format_instructions}

ТОЛЬКО JSON!""",
    input_variables=["review"], # Что должен предоставить пользователь
    partial_variables={         # Что заполняется автоматически
        "format_instructions": parser.get_format_instructions()
    }
)

Анатомия PromptTemplate:

1. template — текст с плейсхолдерами в {}

2. input_variables — список переменных от пользователя

3. partial_variables — переменные с предустановленными значениями

Способы использования:

# Способ 1: format() — возвращает обычную строку
formatted_text = prompt_template.format(review="Отличный товар!")

# Способ 2: invoke() — возвращает специальный PromptValue объект
prompt_value = prompt_template.invoke({"review": "Отличный товар!"})

# Способ 3: в цепочке (самый элегантный)
chain = prompt_template | llm | parser

Почему invoke() лучше format():

• Валидация параметров

• Поддержка всех типов данных

• Лучшая интеграция с LangChain компонентами

Практический пример: собираем все вместе

from langchain_amvera import AmveraLLM
from pydantic import BaseModel, Field
from typing import List, Literal
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate
from dotenv import load_dotenv

load_dotenv()

# Определяем структуру данных
class SentimentAnalysis(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"] = Field(
        description="Тональность отзыва: положительная, отрицательная или нейтральная"
    )
    confidence: float = Field(
        description="Уверенность в анализе от 0.0 до 1.0",
        ge=0.0, le=1.0
    )
    key_topics: List[str] = Field(
        description="Ключевые темы, упомянутые в отзыве",
        max_items=5
    )
    summary: str = Field(
        description="Краткое резюме отзыва в одном предложении",
        max_length=200
    )

# Создаем парсер
parser = JsonOutputParser(pydantic_object=SentimentAnalysis)

# Создаем умный шаблон
prompt_template = PromptTemplate(
    template="""Проанализируй отзыв: {review}

{format_instructions}

ТОЛЬКО JSON!""",
    input_variables=["review"],
    partial_variables={
        "format_instructions": parser.get_format_instructions()  # Автомагия!
    }
)

# Инициализируем нейросеть
llm = AmveraLLM(model="llama70b", temperature=0.0)

Тестируем пошагово:

# Тестовый отзыв
review = "Товар отличный, быстрая доставка! Очень доволен покупкой."

print("=== ПОШАГОВОЕ ВЫПОЛНЕНИЕ ===")

# Шаг 1: Применяем шаблон
print("Применяем PromptTemplate")
prompt_value = prompt_template.invoke({"review": review})
print(f"Тип: {type(prompt_value)}")

# Посмотрим на готовый промпт
prompt_text = prompt_value.to_string()
print("Готовый промпт:")
print(prompt_text[:200] + "...")  # Первые 200 символов
print()

# Шаг 2: Отправляем в нейросеть
print("Отправляем в нейросеть")
llm_response = llm.invoke(prompt_value)
print(f"Тип ответа: {type(llm_response)}")
print(f"Ответ: {llm_response.content}")
print()

# Шаг 3: Парсим JSON
print("Парсим JSON")
parsed_result = parser.invoke(llm_response)
print(f"Тип результата: {type(parsed_result)}")
print("Структурированные данные:")
for key, value in parsed_result.items():
    print(f"  {key}: {value}")

Результат:

=== ПОШАГОВОЕ ВЫПОЛНЕНИЕ ===
1️⃣ Применяем PromptTemplate
Тип: 
Готовый промпт:
Проанализируй отзыв: Товар отличный, быстрая доставка! Очень доволен покупкой.

The output should be formatted as a JSON instance...

2️⃣ Отправляем в нейросеть
Тип ответа: 
Ответ: {"sentiment": "positive", "confidence": 0.95, "key_topics": ["качество", "доставка"], "summary": "Положительный отзыв о качестве товара и быстрой доставке."}

3️⃣ Парсим JSON
Тип результата: 
Структурированные данные:
  sentiment: positive
  confidence: 0.95
  key_topics: ['качество', 'доставка']
  summary: Положительный отзыв о качестве товара и быстрой доставке.

Лаконичный способ через цепочку:

# Все в одну строку
analysis_chain = prompt_template | llm | parser
result = analysis_chain.invoke({"review": review})

print("=== ЧЕРЕЗ ЦЕПОЧКУ ===")
print(f"Результат: {result}")

Результат тот же:

=== ЧЕРЕЗ ЦЕПОЧКУ ===
Результат: {'sentiment': 'positive', 'confidence': 0.95, 'key_topics': ['качество', 'доставка'], 'summary': 'Положительный отзыв о качестве товара и быстрой доставке.'}

Ключевые принципы работы

Последовательность компонентов:

Pydantic модель → JsonOutputParser → PromptTemplate → LLM → JsonOutputParser
     ↓                ↓                  ↓           ↓           ↓
  Схема JSON    Инструкции для ИИ   Полный промпт  Ответ ИИ   Валидный dict

Важные детали:

• JsonOutputParser используется дважды: для генерации инструкций и для парсинга ответа

• PromptTemplate автоматически подставляет инструкции через partial_variables

• temperature=0.0 обеспечивает максимальную предсказуемость

• Pydantic валидация гарантирует соответствие схеме

Что дальше: интеграция в LangGraph

Теперь, когда мы разобрали основные компоненты по отдельности, пора интегрировать их в архитектуру LangGraph. В следующем разделе мы:

• Создадим граф с отдельными узлами для каждого этапа обработки

• Добавим обработку ошибок и retry-логику на уровне узлов

• Построим систему пакетной обработки отзывов

• Интегрируем JSON-анализ в многоуровневые диалоговые агенты

Граф будет выглядеть так:

START → [Подготовка промпта] → [Вызов LLM] → [Парсинг JSON] → [Валидация] → END
                ↓                 ↓              ↓              ↓
           [Обработка ошибок] ←────┴──────────────┴──────────────┘

Каждый узел будет отвечать за свой этап, что обеспечит максимальную наблюдаемость, тестируемость и возможность точной настройки процесса получения структурированных данных от ИИ.

Закрепляем на практике: умная система анализа

На данный момент мы уже умеем работать с графом, умеем подключать к графу LLM и разобрались с важной темой парсинга ответов в валидный JSON формат. А это значит, что мы готовы к более серьезной практической работе.

Суть задачи будет сводиться к следующему:

1. В интерактивном формате пользователь будет писать сообщения

2. Нейросеть должна будет определять — это отзыв или просто обычное сообщение (вопрос)

3. В случае если это отзыв — запускаем анализ с получением структурированного JSON

4. В случае если это вопрос — даем обычный ответ чат-бота

Тут смысл вот в чем. Я покажу вам на этом простом примере, что при грамотном применении инструментов мы будем получать тот функционал, который даже не закладывали изначально, а именно — интерактивный анализатор отзывов от пользователя. Когда перейдем к коду станет все более понятно.

Архитектура системы: два пути обработки

Представьте граф, который работает как умный диспетчер:

START → [Ввод пользователя] → [Классификация ИИ]
                                      ↓
                              ┌─── Отзыв? ───┐
                              ↓              ↓
                    [Анализ отзыва]    [Ответ на вопрос]
                         ↓                   ↓
                    [JSON результат]   [Обычный чат]
                         ↓                   ↓
                         └─── [Продолжить] ──┘
                                      ↓
                                [Новый ввод] или END

Ключевая особенность: одна нейросеть принимает решение, какой путь выбрать, а затем система автоматически направляет данные в соответствующую ветку обработки.

Pydantic модели: определяем структуры данных

Для нашей системы понадобятся две модели — одна для классификации, другая для анализа:

from pydantic import BaseModel, Field
from typing import List, Literal


# Модель для классификации сообщения
class MessageClassification(BaseModel):
    message_type: Literal["review", "question"] = Field(
        description="Тип сообщения: отзыв или вопрос"
    )
    confidence: float = Field(
        description="Уверенность в классификации от 0.0 до 1.0",
        ge=0.0, le=1.0
    )

    
# Модель для анализа отзыва
class ReviewAnalysis(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"] = Field(
        description="Тональность отзыва"
    )
    confidence: float = Field(
        description="Уверенность в анализе от 0.0 до 1.0",
        ge=0.0, le=1.0
    )
    key_topics: List[str] = Field(
        description="Ключевые темы из отзыва",
        max_items=5
    )
    summary: str = Field(
        description="Краткое резюме в одном предложении",
        max_length=150
    )

Почему две модели?

• MessageClassification — простая задача: отзыв или вопрос?

• ReviewAnalysis — сложная задача: детальный анализ отзыва

Это позволяет нейросети лучше сосредоточиться на каждой конкретной задаче.

Состояние системы: что храним между узлами

from langchain_core.messages import BaseMessage
from typing import TypedDict, List


class SystemState(TypedDict):
    messages: List[BaseMessage]       # История диалога
    current_user_input: str           # Текущее сообщение пользователя
    message_type: str                 # Результат классификации
    should_continue: bool             # Продолжать работу?
    analysis_results: List[dict]      # Накопленные результаты анализа

Логика состояния:

• messages — сохраняет контекст для чат-бота

• current_user_input — передает данные между узлами

• message_type — результат классификации для маршрутизации

• analysis_results — накапливает JSON результаты анализа отзывов

Узлы системы: пошаговая обработка

Узел 1: Получение пользовательского ввода

def user_input_node(state: SystemState) -> dict:
    """Узел получения пользовательского ввода"""
    user_input = input("\n? Вы: ").strip()
    
    # Команды выхода
    if user_input.lower() in ["выход", "quit", "exit", "пока", "bye"]:
        return {"should_continue": False}
    
    # Команда статистики
    if user_input.lower() in ["стат", "статистика", "results"]:
        analysis_results = state.get("analysis_results", [])
        if analysis_results:
            print(f"\n? Проанализировано отзывов: {len(analysis_results)}")
            # Подсчет тональности
            sentiments = [r["analysis"]["sentiment"] for r in analysis_results]
            pos = sentiments.count("positive")
            neg = sentiments.count("negative") 
            neu = sentiments.count("neutral")
            print(f"Положительные: {pos}, Отрицательные: {neg}, Нейтральные: {neu}")
        else:
            print("? Пока нет проанализированных отзывов")
        return {"should_continue": True}  # Остаемся в том же узле
    
    return {
        "current_user_input": user_input,
        "should_continue": True
    }

Особенности:

• Обрабатывает команды системы (выход, стат)

• Показывает накопленную статистику по отзывам

• Передает обычный ввод дальше по графу

Узел 2: Классификация сообщения

# Создаем парсер и промпт для классификации
classification_parser = JsonOutputParser(pydantic_object=MessageClassification)
classification_prompt = PromptTemplate(
    template="""Определи, является ли это сообщение отзывом о товаре/услуге или обычным вопросом.

ОТЗЫВ - это мнение о товаре, услуге, опыте использования, оценка качества.
ВОПРОС - это запрос информации, общение, просьба о помощи.

Сообщение: {user_input}

{format_instructions}

Верни ТОЛЬКО JSON!""",
    input_variables=["user_input"],
    partial_variables={"format_instructions": classification_parser.get_format_instructions()}
)

def classify_message_node(state: SystemState) -> dict:
    """Узел классификации сообщения"""
    user_input = state["current_user_input"]
    
    try:
        print("? Определяю тип сообщения...")
        
        # Создаем цепочку классификации
        classification_chain = classification_prompt | llm | classification_parser
        result = classification_chain.invoke({"user_input": user_input})
        
        message_type = result["message_type"]
        confidence = result["confidence"]
        
        print(f"? Тип: {message_type} (уверенность: {confidence:.2f})")
        
        return {"message_type": message_type}
        
    except Exception as e:
        print(f"❌ Ошибка классификации: {e}")
        # По умолчанию считаем вопросом
        return {"message_type": "question"}

Ключевая логика:

• Одна нейросеть решает: отзыв это или вопрос

• Четкие критерии в промпте помогают точной классификации

• Fallback стратегия при ошибках

Узел 3: Анализ отзыва (JSON путь)

# Парсер и промпт для анализа
review_parser = JsonOutputParser(pydantic_object=ReviewAnalysis)
review_analysis_prompt = PromptTemplate(
    template="""Проанализируй этот отзыв клиента:

Отзыв: {review}

{format_instructions}

Верни ТОЛЬКО JSON без дополнительных комментариев!""",
    input_variables=["review"],
    partial_variables={"format_instructions": review_parser.get_format_instructions()}
)

def analyze_review_node(state: SystemState) -> dict:
    """Узел анализа отзыва"""
    user_input = state["current_user_input"]
    
    try:
        print("? Анализирую отзыв...")
        
        # Анализируем отзыв
        analysis_chain = review_analysis_prompt | llm | review_parser
        analysis_result = analysis_chain.invoke({"review": user_input})
        
        # Создаем полный результат
        full_result = {
            "original_review": user_input,
            "analysis": analysis_result
        }
        
        # Добавляем в накопленные результаты
        analysis_results = state.get("analysis_results", [])
        new_analysis_results = analysis_results + [full_result]
        
        # Красивый вывод JSON
        print("\n" + "="*60)
        print("? АНАЛИЗ ОТЗЫВА (JSON):")
        print("="*60)
        print(json.dumps(full_result, ensure_ascii=False, indent=2))
        print("="*60)
        
        # Добавляем в контекст диалога
        messages = state["messages"]
        new_messages = messages + [
            HumanMessage(content=user_input),
            AIMessage(content=f"Отзыв проанализирован: {analysis_result['sentiment']} тональность с уверенностью {analysis_result['confidence']:.2f}")
        ]
        
        return {
            "messages": new_messages,
            "analysis_results": new_analysis_results
        }
        
    except Exception as e:
        print(f"❌ Ошибка анализа отзыва: {e}")
        
        # Fallback: добавляем в диалог сообщение об ошибке
        messages = state["messages"]
        new_messages = messages + [
            HumanMessage(content=user_input),
            AIMessage(content="Извините, произошла ошибка при анализе отзыва.")
        ]
        
        return {"messages": new_messages}

Что происходит:

• Полный JSON анализ отзыва

• Результат сохраняется в analysis_results для статистики

• Краткая информация добавляется в диалоговый контекст

• Красивый вывод JSON в консоль

Узел 4: Ответ на вопрос (чат путь)

def answer_question_node(state: SystemState) -> dict:
    """Узел ответа на вопрос"""
    user_input = state["current_user_input"]
    
    try:
        print("? Отвечаю на вопрос...")
        
        # Добавляем вопрос в контекст
        messages = state["messages"] + [HumanMessage(content=user_input)]
        
        # Получаем ответ от LLM
        response = llm.invoke(messages)
        ai_response = response.content
        
        print(f"? ИИ: {ai_response}")
        
        # Добавляем ответ в контекст
        new_messages = messages + [AIMessage(content=ai_response)]
        
        return {"messages": new_messages}
        
    except Exception as e:
        print(f"❌ Ошибка при ответе: {e}")
        
        messages = state["messages"] + [
            HumanMessage(content=user_input),
            AIMessage(content="Извините, произошла ошибка при обработке вашего вопроса.")
        ]
        
        return {"messages": messages}

Простая логика чат-бота:

• Добавляем вопрос в контекст диалога

• LLM отвечает на основе всей истории сообщений

• Сохраняем ответ в контекст для следующих вопросов

Функции маршрутизации: как граф принимает решения

Маршрутизация после ввода

def route_after_input(state: SystemState) -> str:
    """Маршрутизация после ввода пользователя"""
    if not state.get("should_continue", True):
        return "end"
    
    if state.get("current_user_input"):
        return "classify"
    
    return "get_input"  # Если пустой ввод, запрашиваем заново

Маршрутизация после классификации

def route_after_classification(state: SystemState) -> str:
    """Маршрутизация после классификации"""
    message_type = state.get("message_type", "question")
    
    if message_type == "review":
        return "analyze_review"  # → JSON анализ
    else:
        return "answer_question"  # → обычный чат

Здесь происходит магия: одно решение нейросети определяет весь дальнейший путь обработки.

Маршрутизация продолжения

def route_continue(state: SystemState) -> str:
    """Проверка продолжения работы"""
    return "get_input" if state.get("should_continue", True) else "end"

Сборка графа: связываем все узлы

from langgraph.graph import StateGraph, START, END

# Создание графа
graph = StateGraph(SystemState)

# Добавляем узлы
graph.add_node("get_input", user_input_node)
graph.add_node("classify", classify_message_node)
graph.add_node("analyze_review", analyze_review_node)
graph.add_node("answer_question", answer_question_node)

# Создаем рёбра
graph.add_edge(START, "get_input")

# Условные рёбра для маршрутизации
graph.add_conditional_edges(
    "get_input",
    route_after_input,
    {
        "classify": "classify",
        "get_input": "get_input",  # Цикл при пустом вводе
        "end": END
    }
)

graph.add_conditional_edges(
    "classify",
    route_after_classification,
    {
        "analyze_review": "analyze_review",  # → JSON путь
        "answer_question": "answer_question"  # → чат путь
    }
)

graph.add_conditional_edges(
    "analyze_review",
    route_continue,
    {
        "get_input": "get_input",  # Возврат к вводу
        "end": END
    }
)

graph.add_conditional_edges(
    "answer_question", 
    route_continue,
    {
        "get_input": "get_input",  # Возврат к вводу
        "end": END
    }
)

# Компиляция
app = graph.compile()

Запуск и тестирование системы

if __name__ == "__main__":
    print("? Умная система: Анализ отзывов + Чат-бот")
    print("Введите отзыв - получите JSON анализ")
    print("Задайте вопрос - получите ответ")
    print("Команды: 'стат' - статистика, 'выход' - завершить")
    print("-" * 60)
    
    # Начальное состояние
    initial_state = {
        "messages": [
            SystemMessage(content="Ты дружелюбный помощник. Отвечай коротко и по делу на вопросы пользователя.")
        ],
        "current_user_input": "",
        "message_type": "",
        "should_continue": True,
        "analysis_results": []
    }
    
    try:
        final_state = app.invoke(initial_state)
        print("\n✅ Работа завершена!")
        print(f"? Всего сообщений: {len(final_state.get('messages', []))}")
        print(f"? Проанализировано отзывов: {len(final_state.get('analysis_results', []))}")
        
    except KeyboardInterrupt:
        print("\n\n⚠️ Работа прервана (Ctrl+C)")
    except Exception as e:
        print(f"\n❌ Ошибка системы: {e}")

Пример работы системы

? Умная система: Анализ отзывов + Чат-бот
Введите отзыв - получите JSON анализ
Задайте вопрос - получите ответ
Команды: 'стат' - статистика, 'выход' - завершить
------------------------------------------------------------

? Вы: Отличный товар, быстрая доставка!
? Определяю тип сообщения...
? Тип: review (уверенность: 0.95)
? Анализирую отзыв...

============================================================
? АНАЛИЗ ОТЗЫВА (JSON):
============================================================
{
  "original_review": "Отличный товар, быстрая доставка!",
  "analysis": {
    "sentiment": "positive",
    "confidence": 0.92,
    "key_topics": ["качество", "доставка"],
    "summary": "Положительный отзыв о качестве товара и скорости доставки."
  }
}
============================================================

? Вы: А как работает ваша доставка?
? Определяю тип сообщения...
? Тип: question (уверенность: 0.88)
? Отвечаю на вопрос...
? ИИ: Я не представляю конкретную компанию, но обычно доставка работает через курьерские службы или пункты выдачи. Уточните, о какой доставке вы спрашиваете?

? Вы: стат
? Проанализировано отзывов: 1
Положительные: 1, Отрицательные: 0, Нейтральные: 0

? Вы: выход
✅ Работа завершена!
? Всего сообщений: 5
? Проанализировано отзывов: 1

Что мы получили в итоге

Функциональность, которую мы не закладывали изначально:

1. Автоматическая классификация — система сама понимает тип сообщения

2. Накопление статистики — автоматически собирает данные по отзывам

3. Гибридный интерфейс — JSON анализ + обычный чат в одной системе

4. Контекстная память — чат-бот помнит предыдущие сообщения

5. Командный интерфейс — встроенные команды для управления

Ключевые принципы архитектуры:

• Разделение ответственности — каждый узел решает одну задачу

• Умная маршрутизация — граф сам выбирает путь обработки

• Состояние как память — вся важная информация сохраняется между узлами

• Graceful degradation — система работает даже при ошибках отдельных компонентов

Это демонстрирует мощь LangGraph: правильно спроектированная архитектура дает функциональность, которая превышает сумму отдельных компонентов!

Мультимодельные системы: когда одной нейросети недостаточно

До сих пор мы использовали одну нейросеть для решения всех задач в наших графах. Но в реальных проектах часто возникают ситуации, когда разные модели лучше справляются с разными типами задач. Представьте систему, где:

• DeepSeek анализирует код и технические документы

• Amvera (LLaMA) ведет естественные диалоги с пользователями

• GigaChat работает с русскоязычным контентом и локальными реалиями

Каждая модель имеет свои сильные стороны, и LangGraph позволяет элегантно объединить их в единую систему.

Зачем нужны мультимодельные системы?

Специализация моделей

Разные модели — разные таланты:

• Кодовые модели (DeepSeek-Coder) лучше понимают программирование

• Диалоговые модели (GPT-4, Claude) лучше ведут беседы

• Локальные модели (GigaChat, YandexGPT) лучше знают местные реалии

• Мультимодальные (GPT-4V, Gemini Vision) работают с изображениями

Оптимизация затрат

Экономическая выгода:

Простая классификация → дешевая модель (DeepSeek)
Сложный анализ → мощная модель (GPT-4)
Локальный контекст → региональная модель (GigaChat)

Отказоустойчивость

Резервирование:

• Основная модель недоступна → переключение на backup

• Разные провайдеры → снижение рисков блокировок

• Географическая распределенность → стабильность сервиса

Архитектура мультимодельной системы

Представим граф, где разные узлы используют разные модели:

START → [Определение задачи] → [Маршрутизация]
                                    ↓
           ┌──── Код? ────┐    ┌── Диалог? ──┐    ┌── Локальный контекст? ──┐
           ↓              ↓    ↓             ↓    ↓                        ↓
    [DeepSeek Coder]  [Анализ]  [Amvera]  [Беседа]  [GigaChat]  [Местные реалии]
           ↓              ↓    ↓             ↓    ↓                        ↓
           └──────── [Объединение результатов] ──────────────────────────┘
                              ↓
                         [Финальный ответ] → END

Практический пример: техническая поддержка с ИИ

Создадим систему техподдержки, где:

• DeepSeek анализирует код и технические вопросы

• Amvera ведет общий диалог и объясняет решения

• GigaChat отвечает на вопросы про российские особенности

Подготовка моделей

from dotenv import load_dotenv
from langchain_deepseek import ChatDeepSeek
from langchain_amvera import AmveraLLM
from langchain_gigachat.chat_models import GigaChat
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, List, Literal
from pydantic import BaseModel, Field

load_dotenv()

# Инициализация трех разных моделей
deepseek_model = ChatDeepSeek(
    model="deepseek-chat",
    temperature=0.1  # Низкая температура для технических задач
)

amvera_model = AmveraLLM(
    model="llama70b",
    temperature=0.7  # Умеренная температура для диалогов
)

gigachat_model = GigaChat(
    model="GigaChat-2-Max",
    temperature=0.3,  # Средняя температура
    verify_ssl_certs=False
)

Модель для классификации задач

class TaskClassification(BaseModel):
    task_type: Literal["code", "dialog", "local"] = Field(
        description="Тип задачи: code - программирование, dialog - общение, local - российские реалии"
    )
    confidence: float = Field(
        description="Уверенность в классификации от 0.0 до 1.0",
        ge=0.0, le=1.0
    )
    reasoning: str = Field(
        description="Краткое объяснение выбора",
        max_length=100
    )

Состояние системы

class MultiModelState(TypedDict):
    user_question: str          # Вопрос пользователя
    task_type: str              # Результат классификации
    code_analysis: str          # Результат от DeepSeek
    dialog_response: str        # Результат от Amvera
    local_context: str          # Результат от GigaChat
    final_answer: str           # Итоговый ответ
    should_continue: bool       # Продолжать работу

Узел классификации задач

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate

# Настройка классификатора (используем DeepSeek как быструю модель)
classification_parser = JsonOutputParser(pydantic_object=TaskClassification)
classification_prompt = PromptTemplate(
    template="""Определи тип задачи пользователя:

CODE - вопросы про программирование, отладку, код, алгоритмы, технологии
DIALOG - обычные вопросы, просьбы о помощи, общение, объяснения
LOCAL - вопросы про Россию, российские законы, локальные особенности, госуслуги

Вопрос: {question}

{format_instructions}

Верни ТОЛЬКО JSON!""",
    input_variables=["question"],
    partial_variables={"format_instructions": classification_parser.get_format_instructions()}
)

def classify_task_node(state: MultiModelState) -> dict:
    """Узел классификации задачи - используем DeepSeek"""
    question = state["user_question"]
    
    try:
        print(f"? Классифицирую задачу...")
        
        classification_chain = classification_prompt | deepseek_model | classification_parser
        result = classification_chain.invoke({"question": question})
        
        task_type = result["task_type"]
        confidence = result["confidence"]
        reasoning = result["reasoning"]
        
        print(f"? Тип: {task_type} ({confidence:.2f}) - {reasoning}")
        
        return {"task_type": task_type}
        
    except Exception as e:
        print(f"❌ Ошибка классификации: {e}")
        return {"task_type": "dialog"}  # Fallback к диалогу

Узел анализа кода (DeepSeek)

def code_analysis_node(state: MultiModelState) -> dict:
    """Узел анализа кода - специализация DeepSeek"""
    question = state["user_question"]
    
    try:
        print("? DeepSeek анализирует код...")
        
        code_messages = [
            SystemMessage(content="""Ты эксперт-программист. Анализируй код, находи ошибки, 
                         предлагай оптимизации. Отвечай технично и точно."""),
            HumanMessage(content=question)
        ]
        
        response = deepseek_model.invoke(code_messages)
        analysis = response.content
        
        print(f"✅ DeepSeek: {analysis[:100]}...")
        
        return {"code_analysis": analysis}
        
    except Exception as e:
        print(f"❌ Ошибка DeepSeek: {e}")
        return {"code_analysis": "Ошибка анализа кода"}

Узел диалогового общения (Amvera)

def dialog_response_node(state: MultiModelState) -> dict:
    """Узел диалогового общения - сила Amvera LLaMA"""
    question = state["user_question"]
    
    try:
        print("? Amvera ведет диалог...")
        
        dialog_messages = [
            SystemMessage(content="""Ты дружелюбный помощник. Отвечай развернуто, 
                         объясняй простым языком, будь полезным и понимающим."""),
            HumanMessage(content=question)
        ]
        
        response = amvera_model.invoke(dialog_messages)
        dialog_answer = response.content
        
        print(f"✅ Amvera: {dialog_answer[:100]}...")
        
        return {"dialog_response": dialog_answer}
        
    except Exception as e:
        print(f"❌ Ошибка Amvera: {e}")
        return {"dialog_response": "Ошибка диалогового ответа"}

Узел локального контекста (GigaChat)

def local_context_node(state: MultiModelState) -> dict:
    """Узел локального контекста - экспертиза GigaChat"""
    question = state["user_question"]
    
    try:
        print("?? GigaChat анализирует локальный контекст...")
        
        local_messages = [
            SystemMessage(content="""Ты эксперт по России: законы, традиции, особенности, 
                         госуслуги, местная специфика. Давай точную информацию о российских реалиях."""),
            HumanMessage(content=question)
        ]
        
        response = gigachat_model.invoke(local_messages)
        local_info = response.content
        
        print(f"✅ GigaChat: {local_info[:100]}...")
        
        return {"local_context": local_info}
        
    except Exception as e:
        print(f"❌ Ошибка GigaChat: {e}")
        return {"local_context": "Ошибка анализа локального контекста"}

Узел получения пользовательского ввода

def user_input_node(state: MultiModelState) -> dict:
    """Узел получения вопроса от пользователя"""
    question = input("\n❓ Ваш вопрос: ").strip()
    
    if question.lower() in ["выход", "quit", "exit", "bye"]:
        return {"should_continue": False}
    
    return {
        "user_question": question,
        "should_continue": True
    }

Узел синтеза финального ответа

def synthesize_answer_node(state: MultiModelState) -> dict:
    """Узел синтеза итогового ответа - используем Amvera для объединения"""
    task_type = state["task_type"]
    question = state["user_question"]
    
    # Собираем доступные результаты
    results = []
    
    if state.get("code_analysis"):
        results.append(f"Технический анализ: {state['code_analysis']}")
    
    if state.get("dialog_response"):
        results.append(f"Общий ответ: {state['dialog_response']}")
        
    if state.get("local_context"):
        results.append(f"Локальная информация: {state['local_context']}")
    
    if not results:
        return {"final_answer": "Не удалось получить ответ от моделей"}
    
    try:
        print("? Синтезирую итоговый ответ...")
        
        synthesis_prompt = f"""На основе результатов от разных ИИ-моделей дай пользователю единый полезный ответ.

Вопрос пользователя: {question}
Тип задачи: {task_type}

Результаты от моделей:
{chr(10).join(results)}

Создай связный, полезный ответ, объединив лучшее из каждого источника."""

        synthesis_messages = [
            SystemMessage(content="Ты синтезируешь ответы от разных ИИ в единый полезный ответ."),
            HumanMessage(content=synthesis_prompt)
        ]
        
        response = amvera_model.invoke(synthesis_messages)
        final_answer = response.content
        
        print("="*60)
        print("? ИТОГОВЫЙ ОТВЕТ:")
        print("="*60) 
        print(final_answer)
        print("="*60)
        
        return {"final_answer": final_answer}
        
    except Exception as e:
        print(f"❌ Ошибка синтеза: {e}")
        return {"final_answer": "Ошибка при создании итогового ответа"}

Функции маршрутизации

def route_after_input(state: MultiModelState) -> str:
    """Маршрутизация после ввода"""
    if not state.get("should_continue", True):
        return "end"
    return "classify"

def route_after_classification(state: MultiModelState) -> str:
    """Маршрутизация по типу задачи"""
    task_type = state.get("task_type", "dialog")
    
    if task_type == "code":
        return "analyze_code"
    elif task_type == "local":
        return "local_context"
    else:
        return "dialog_response"

def route_to_synthesis(state: MultiModelState) -> str:
    """Маршрутизация к синтезу ответа"""
    return "synthesize"

def route_continue(state: MultiModelState) -> str:
    """Проверка продолжения"""
    return "get_input" if state.get("should_continue", True) else "end"

Сборка мультимодельного графа

# Создание графа
graph = StateGraph(MultiModelState)

# Добавляем узлы
graph.add_node("get_input", user_input_node)
graph.add_node("classify", classify_task_node)
graph.add_node("analyze_code", code_analysis_node)
graph.add_node("dialog_response", dialog_response_node)
graph.add_node("local_context", local_context_node)
graph.add_node("synthesize", synthesize_answer_node)

# Создаем рёбра
graph.add_edge(START, "get_input")

# Условные рёбра
graph.add_conditional_edges(
    "get_input",
    route_after_input,
    {
        "classify": "classify",
        "end": END
    }
)

graph.add_conditional_edges(
    "classify",
    route_after_classification,
    {
        "analyze_code": "analyze_code",
        "dialog_response": "dialog_response",
        "local_context": "local_context"
    }
)

# Все специализированные узлы ведут к синтезу
graph.add_conditional_edges(
    "analyze_code",
    route_to_synthesis,
    {"synthesize": "synthesize"}
)

graph.add_conditional_edges(
    "dialog_response", 
    route_to_synthesis,
    {"synthesize": "synthesize"}
)

graph.add_conditional_edges(
    "local_context",
    route_to_synthesis, 
    {"synthesize": "synthesize"}
)

graph.add_conditional_edges(
    "synthesize",
    route_continue,
    {
        "get_input": "get_input",
        "end": END
    }
)

# Компиляция
multi_model_app = graph.compile()

Запуск системы

if __name__ == "__main__":
    print("? Мультимодельная система техподдержки")
    print("DeepSeek - код | Amvera - диалоги | GigaChat - локальный контекст")
    print("Команда 'выход' для завершения")
    print("-" * 70)
    
    initial_state = {
        "user_question": "",
        "task_type": "",
        "code_analysis": "",
        "dialog_response": "",
        "local_context": "", 
        "final_answer": "",
        "should_continue": True
    }
    
    try:
        final_state = multi_model_app.invoke(initial_state)
        print("\n✅ Система завершена!")
        
    except KeyboardInterrupt:
        print("\n\n⚠️ Работа прервана (Ctrl+C)")
    except Exception as e:
        print(f"\n❌ Ошибка системы: {e}")

Пример работы системы

Сценарий 1: Вопрос про код

❓ Ваш вопрос: Как исправить ошибку "list index out of range" в Python?

? Классифицирую задачу...
? Тип: code (0.95) - Вопрос про отладку Python

? DeepSeek анализирует код...
✅ DeepSeek: Ошибка "list index out of range" возникает при попытке...

? Синтезирую итоговый ответ...
============================================================
? ИТОГОВЫЙ ОТВЕТ:
============================================================
Ошибка "list index out of range" в Python возникает, когда вы пытаетесь 
обратиться к элементу списка по индексу, которого не существует...

[Технический анализ от DeepSeek + объяснение от Amvera]
============================================================

Сценарий 2: Вопрос про российские реалии

❓ Ваш вопрос: Как получить справку о доходах через Госуслуги?

? Классифицирую задачу...
? Тип: local (0.92) - Вопрос про госуслуги России

?? GigaChat анализирует локальный контекст...
✅ GigaChat: Для получения справки о доходах через Госуслуги нужно...

? Синтезирую итоговый ответ...
============================================================
? ИТОГОВЫЙ ОТВЕТ:
============================================================
Чтобы получить справку о доходах через портал Госуслуги, следуйте инструкции...

[Экспертная информация от GigaChat + понятное объяснение от Amvera]
============================================================

Сценарий 3: Обычный диалог

❓ Ваш вопрос: Расскажи о пользе чтения книг

? Классифицирую задачу...
? Тип: dialog (0.88) - Общий вопрос для обсуждения

? Amvera ведет диалог...
✅ Amvera: Чтение книг приносит множество пользы...

? Синтезирую итоговый ответ...
============================================================
? ИТОГОВЫЙ ОТВЕТ:
============================================================
Чтение книг - это одна из самых полезных привычек...

[Развернутый ответ от Amvera]
============================================================

Преимущества мультимодельного подхода

Специализация и качество

Каждая модель делает то, что умеет лучше всего:

• DeepSeek дает точные технические ответы

• Amvera ведет живые диалоги и синтезирует информацию

• GigaChat предоставляет актуальную локальную информацию

Экономическая эффективность

Оптимизация затрат:

• Простая классификация через быструю модель (DeepSeek)

• Сложные задачи направляются к специализированным моделям

• Нет переплаты за неиспользуемые возможности

Отказоустойчивость

Резервирование на уровне архитектуры:

def fallback_node(state: MultiModelState) -> dict:
    """Узел-fallback при недоступности основных моделей"""
    try:
        # Пробуем запасную модель
        backup_response = backup_model.invoke(state["user_question"])
        return {"final_answer": backup_response.content}
    except:
        return {"final_answer": "Все модели временно недоступны"}

Паттерны использования мультимодельных систем

Паттерн "Специалист-Генералист"

Классификация → Специалист → Генералист (синтез)

• Специалист решает узкую задачу (код, локальная информация)

• Генералист объединяет результаты в понятный ответ

Паттерн "Консилиум экспертов"

def expert_consensus_node(state: MultiModelState) -> dict:
    """Получаем мнения от всех моделей и выбираем лучший ответ"""
    
    results = []
    
    # Спрашиваем у всех моделей
    for model_name, model in [("DeepSeek", deepseek_model), 
                             ("Amvera", amvera_model), 
                             ("GigaChat", gigachat_model)]:
        try:
            response = model.invoke(state["user_question"])
            results.append(f"{model_name}: {response.content}")
        except:
            continue
    
    # Метамодель выбирает лучший ответ
    best_answer = choose_best_response(results)
    return {"final_answer": best_answer}

Паттерн "Конвейер обработки"

Модель 1 (предобработка) → Модель 2 (анализ) → Модель 3 (финализация)

Управление версиями и конфигурациями

class ModelConfig:
    def __init__(self):
        self.models = {
            "classifier": deepseek_model,
            "coder": deepseek_model,
            "dialog": amvera_model,
            "local": gigachat_model,
            "synthesizer": amvera_model
        }
    
    def get_model(self, role: str):
        """Получить модель по роли с возможностью A/B тестирования"""
        if role in self.models:
            return self.models[role]
        return self.models["dialog"]  # fallback
    
    def switch_model(self, role: str, new_model):
        """Горячая замена модели"""
        self.models[role] = new_model

Мониторинг и аналитика

def monitor_model_performance(state: MultiModelState) -> dict:
    """Отслеживание производительности моделей"""
    
    metrics = {
        "classification_confidence": state.get("classification_confidence", 0),
        "response_time": time.time() - state.get("start_time", 0),
        "model_used": state.get("task_type", "unknown"),
        "success": bool(state.get("final_answer"))
    }
    
    # Логирование метрик
    log_metrics(metrics)
    
    return state

Ключевые принципы мультимодельных систем

1. Четкое разделение ролей — каждая модель решает конкретный класс задач

2. Умная маршрутизация — правильное направление запросов к нужным моделям

3. Graceful fallback — запасные варианты при недоступности моделей

4. Экономическая оптимизация — использование дешевых моделей где это возможно

5. Мониторинг качества — отслеживание производительности каждой модели

Мультимодельный подход в LangGraph открывает возможности создания по-настоящему мощных и экономически эффективных ИИ-систем, где каждая модель работает в своей области экспертизы.

Итоги второй части: от статических схем к интеллектуальным собеседникам

Во второй части мы превратили безжизненные узлы и рёбра в настоящих цифровых собеседников. Если в первой части мы заложили архитектурный фундамент LangGraph, то сейчас мы научили наши графы по-настоящему думать.

Что мы освоили

Интеграция языковых моделей

• Подключение нейросетей к узлам графов

• Работа с российскими провайдерами (Amvera, GigaChat, DeepSeek)

• Выбор оптимального подхода под конкретные задачи

Диалоговая память и контекст

• Система сообщений: SystemMessage, HumanMessage, AIMessage

• Управление длиной контекста и оптимизация токенов

• Создание агентов с памятью на сотни ходов диалога

Структурированные JSON-ответы

• Pydantic модели для строгих схем данных

• JsonOutputParser с автогенерацией инструкций

• PromptTemplate для динамических промптов

• Получение валидного JSON в 99.9% случаев

Интеллектуальная маршрутизация

• ИИ-классификация типов сообщений

• Автоматическое направление в нужные ветки обработки

• Гибридные интерфейсы (JSON анализ + чат)

Мультимодельные системы

• Специализация разных моделей под разные задачи

• Экономическая оптимизация через правильный выбор модели

• Синтез результатов от нескольких источников

Текущие ограничения

Наши агенты умеют думать, анализировать, классифицировать, вести диалоги — но не могут действовать в реальном мире:

• Отправлять email

• Создавать файлы

• Обращаться к базам данных

• Делать HTTP-запросы

• Управлять внешними сервисами

Это критическое ограничение для production-систем.

Переход к реактивным агентам: что нас ждет в третьей части

Часть 3: Реактивные агенты — от слов к реальным действиям

В третьей части мы совершим качественный скачок — научим наших агентов взаимодействовать с внешним миром через инструменты (tools) и MCP-серверы.

К концу третьей части мы создадим агентов, способных:

Автоматизировать рабочие процессы:

• Мониторить почту и автоматически отвечать на типовые запросы

• Анализировать логи сервера и отправлять уведомления при ошибках

• Создавать отчеты в Google Sheets на основе данных из разных источников

Управлять инфраструктурой:

• Деплоить приложения через Git hooks

• Мониторить метрики системы и масштабировать ресурсы

• Бэкапировать базы данных по расписанию

Интегрироваться с бизнес-системами:

• Синхронизировать данные между CRM и учетными системами

• Обрабатывать заказы и обновлять складские остатки

• Анализировать обратную связь клиентов и создавать тикеты

В следующей части...

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

• Принимать решения на основе анализа данных

• Выполнять действия в реальных системах

• Реагировать на события в режиме реального времени

• Интегрироваться с любыми внешними сервисами

• Работать автономно без постоянного присмотра человека

Если во второй части мы создали агентов, которые умеют думать, то в третьей части мы научим их делать.

Это будет финальный переход от демонстрационных примеров к production-ready системам, способным автоматизировать реальные бизнес-процессы.

Готовы превратить ваших агентов из цифровых собеседников в цифровых сотрудников?

P.S. Если эта статья была для вас полезной, поддержите автора — подпиской, комментарием или лайком. А если хотите найти больше эксклюзивного контента, которого нет на Хабре, присоединяйтесь к моему бесплатному Телеграм-каналу «Легкий путь в Python».