Представьте себе 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 напрямую влияет на гибкость архитектуры, производительность и возможности кастомизации вашего AI-агента.
В экосистеме LangChain существует четыре основных подхода к инициализации нейросетей, каждый из которых имеет свои преимущества и области применения.
Самый простой способ быстро подключить популярную модель — использовать универсальный метод инициализации:
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-решений с специфическими требованиями
Для более глубокого контроля над поведением моделей рекомендуется использовать специализированные пакеты:
# Установка: 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 # Потоковая передача ответов
)
Провайдер | Библиотека |
---|---|
OpenAI |
|
Anthropic |
|
DeepSeek |
|
| |
Groq |
|
Ollama |
|
Преимущества специализированных библиотек:
Полный контроль параметров — temperature, max_tokens, stop_sequences и другие
Расширенная обработка ошибок — настройка retry-логики и таймаутов
Специфические возможности — функции, доступные только для конкретных провайдеров
Production-готовность — оптимизированные для высоконагруженных систем
Помимо огромного количества официальных пакетов для интеграции с топовыми нейросетями, LangChain предоставил полноценный инструментарий, который позволяет обернуть в их оболочку практически любой API-протокол, работающий с нейросетями.
Множество компаний использовало эти инструменты для создания интеграции с экосистемой LangChain. В результате мы получили готовые неофициальные библиотеки от таких провайдеров как Amvera Cloud (официальный доступ к моделям LLaMA и ChatGPT без VPN с пополнением через карты РФ), GigaChat (Сбер), YandexGPT и многих других.
Amvera предоставляет доступ к современным нейросетям: Llama3.3 70B, Llama 3.1 8B, GPT-4.1, GPT-5 через единый API.
Установка:
pip install langchain langchain-amvera
Получение токена:
Регистрируемся на Amvera Cloud
Переходим в раздел LLM проектов
Выбираем нужную модель (каждая включает бесплатные токены для тестирования)
Копируем токен из документации выбранной модели
Код интеграции:
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)
Пример ответа:
Нейросети работают по принципу, схожему с человеческим мозгом.
Представь сеть из взаимосвязанных узлов (нейронов), где каждый узел получает
информацию, обрабатывает её и передаёт результат дальше...
Установка:
pip install langchain-gigachat
Получение токена:
Входим на Сбер Developer (через Сбер ID)
Создаём проект
Получаем новый ключ в разделе "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
Если вы хотите полный контроль над запросами или работаете с API, не имеющими готовых LangChain-интеграций, можете использовать прямые HTTP-запросы:
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 без использования неофициального адаптера. В приведённом выше примере показал, как выполнить прямой вызов. Далее остаётся лишь добавить функцию вызова в граф.
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-интеграции, когда они доступны.
Ситуация | Рекомендуемый подход | Почему |
---|---|---|
Быстрый прототип |
| Минимум кода, максимум скорости |
Production-система | Специализированные пакеты | Полный контроль, надёжность |
Российские провайдеры | Неофициальные пакеты | Готовые решения для локальных API |
Кастомный API | Прямая интеграция | Когда нет готовых решений |
Локальные модели | Ollama или прямые запросы | Приватность данных, полный контроль |
В следующем разделе мы рассмотрим, как эти инициализированные модели встраиваются в архитектуру 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, и продолжит диалог с учётом этого "факта".
Можно комбинировать ответы разных нейросетей в одном диалоге:
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}")
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-эксперта
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 подстановка 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 # Возвращаем без изменений
Всегда используйте типизированные сообщения для диалогов длиннее одного обмена
SystemMessage задаёт тон — размещайте его в начале для настройки поведения
Сохраняйте историю в списке — порядок сообщений критически важен
Контролируйте длину контекста — избегайте превышения лимитов модели
Используйте метаданные — отслеживайте потребление токенов и производительность
Подстановка AIMessage — мощный инструмент для создания сложных диалоговых сценариев
Этот мощный механизм открывает безграничные возможности для тонкой настройки поведения ИИ-агентов и создания сложных мультимодельных систем!
В следующем разделе мы применим эти знания для создания первого полноценного диалогового агента в 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()
Логика работы графа:
START → user_input — начинаем с ввода пользователя
user_input → llm_response — передаём сообщение ИИ для ответа
llm_response → should_continue — проверяем, нужно ли продолжать
should_continue → user_input (если "continue") — новый цикл диалога
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) -> 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}
# Неправильно - мутируем существующий список
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}
# Неправильно - можем потерять 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 диалоговых
# Неправильно - не обрабатываем пустые сообщения
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-ответы для интеграции с внешними системами.
В реальных приложениях AI-агенты должны интегрироваться с базами данных, API и другими системами. Это означает, что нам нужны не красивые диалоги, а строго структурированные данные в предсказуемом формате. К сожалению, языковые модели по природе своей склонны к творчеству, даже когда мы просим их о сухих фактах.
Представьте, что вы создаете систему анализа отзывов клиентов. От агента требуется простая структура:
{
"sentiment": "positive",
"confidence": 0.85,
"key_topics": ["качество", "доставка"]
}
Но вместо этого получаете:
Конечно, я проанализирую отзыв! Вот результат моего анализа:
{
"sentiment": "positive",
"confidence": 0.85,
"key_topics": ["качество", "доставка"]
}
Как видите, отзыв довольно позитивный, особенно в части качества товара.
Надеюсь, это поможет в вашем анализе!
Проблемы такого ответа:
Невозможно распарсить JSON из-за лишнего текста
Нестабильный формат — иногда комментарии в начале, иногда в конце
Нарушение автоматизированных процессов обработки данных
Увеличение расходов на токены из-за "болтовни" модели
Для решения этой проблемы в LangChain есть три фундаментальные сущности, которые работают в связке:
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
Описания полей для лучшего понимания нейросетью
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".
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) # Ошибка!
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:
template — текст с плейсхолдерами в {}
input_variables — список переменных от пользователя
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. В следующем разделе мы:
Создадим граф с отдельными узлами для каждого этапа обработки
Добавим обработку ошибок и retry-логику на уровне узлов
Построим систему пакетной обработки отзывов
Интегрируем JSON-анализ в многоуровневые диалоговые агенты
Граф будет выглядеть так:
START → [Подготовка промпта] → [Вызов LLM] → [Парсинг JSON] → [Валидация] → END
↓ ↓ ↓ ↓
[Обработка ошибок] ←────┴──────────────┴──────────────┘
Каждый узел будет отвечать за свой этап, что обеспечит максимальную наблюдаемость, тестируемость и возможность точной настройки процесса получения структурированных данных от ИИ.
На данный момент мы уже умеем работать с графом, умеем подключать к графу LLM и разобрались с важной темой парсинга ответов в валидный JSON формат. А это значит, что мы готовы к более серьезной практической работе.
Суть задачи будет сводиться к следующему:
В интерактивном формате пользователь будет писать сообщения
Нейросеть должна будет определять — это отзыв или просто обычное сообщение (вопрос)
В случае если это отзыв — запускаем анализ с получением структурированного JSON
В случае если это вопрос — даем обычный ответ чат-бота
Тут смысл вот в чем. Я покажу вам на этом простом примере, что при грамотном применении инструментов мы будем получать тот функционал, который даже не закладывали изначально, а именно — интерактивный анализатор отзывов от пользователя. Когда перейдем к коду станет все более понятно.
Представьте граф, который работает как умный диспетчер:
START → [Ввод пользователя] → [Классификация ИИ]
↓
┌─── Отзыв? ───┐
↓ ↓
[Анализ отзыва] [Ответ на вопрос]
↓ ↓
[JSON результат] [Обычный чат]
↓ ↓
└─── [Продолжить] ──┘
↓
[Новый ввод] или END
Ключевая особенность: одна нейросеть принимает решение, какой путь выбрать, а затем система автоматически направляет данные в соответствующую ветку обработки.
Для нашей системы понадобятся две модели — одна для классификации, другая для анализа:
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 результаты анализа отзывов
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
}
Особенности:
Обрабатывает команды системы (выход
, стат
)
Показывает накопленную статистику по отзывам
Передает обычный ввод дальше по графу
# Создаем парсер и промпт для классификации
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 стратегия при ошибках
# Парсер и промпт для анализа
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 в консоль
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
Функциональность, которую мы не закладывали изначально:
Автоматическая классификация — система сама понимает тип сообщения
Накопление статистики — автоматически собирает данные по отзывам
Гибридный интерфейс — JSON анализ + обычный чат в одной системе
Контекстная память — чат-бот помнит предыдущие сообщения
Командный интерфейс — встроенные команды для управления
Ключевые принципы архитектуры:
Разделение ответственности — каждый узел решает одну задачу
Умная маршрутизация — граф сам выбирает путь обработки
Состояние как память — вся важная информация сохраняется между узлами
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 к диалогу
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": "Ошибка анализа кода"}
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": "Ошибка диалогового ответа"}
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}")
❓ Ваш вопрос: Как исправить ошибку "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]
============================================================
❓ Ваш вопрос: Как получить справку о доходах через Госуслуги?
🤔 Классифицирую задачу...
📋 Тип: local (0.92) - Вопрос про госуслуги России
🇷🇺 GigaChat анализирует локальный контекст...
✅ GigaChat: Для получения справки о доходах через Госуслуги нужно...
🔄 Синтезирую итоговый ответ...
============================================================
🎯 ИТОГОВЫЙ ОТВЕТ:
============================================================
Чтобы получить справку о доходах через портал Госуслуги, следуйте инструкции...
[Экспертная информация от GigaChat + понятное объяснение от Amvera]
============================================================
❓ Ваш вопрос: Расскажи о пользе чтения книг
🤔 Классифицирую задачу...
📋 Тип: 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
Четкое разделение ролей — каждая модель решает конкретный класс задач
Умная маршрутизация — правильное направление запросов к нужным моделям
Graceful fallback — запасные варианты при недоступности моделей
Экономическая оптимизация — использование дешевых моделей где это возможно
Мониторинг качества — отслеживание производительности каждой модели
Мультимодельный подход в LangGraph открывает возможности создания по-настоящему мощных и экономически эффективных ИИ-систем, где каждая модель работает в своей области экспертизы.
Во второй части мы превратили безжизненные узлы и рёбра в настоящих цифровых собеседников. Если в первой части мы заложили архитектурный фундамент LangGraph, то сейчас мы научили наши графы по-настоящему думать.
Интеграция языковых моделей
Подключение нейросетей к узлам графов
Работа с российскими провайдерами (Amvera, GigaChat, DeepSeek)
Выбор оптимального подхода под конкретные задачи
Диалоговая память и контекст
Система сообщений: SystemMessage, HumanMessage, AIMessage
Управление длиной контекста и оптимизация токенов
Создание агентов с памятью на сотни ходов диалога
Структурированные JSON-ответы
Pydantic модели для строгих схем данных
JsonOutputParser с автогенерацией инструкций
PromptTemplate для динамических промптов
Получение валидного JSON в 99.9% случаев
Интеллектуальная маршрутизация
ИИ-классификация типов сообщений
Автоматическое направление в нужные ветки обработки
Гибридные интерфейсы (JSON анализ + чат)
Мультимодельные системы
Специализация разных моделей под разные задачи
Экономическая оптимизация через правильный выбор модели
Синтез результатов от нескольких источников
Наши агенты умеют думать, анализировать, классифицировать, вести диалоги — но не могут действовать в реальном мире:
Отправлять email
Создавать файлы
Обращаться к базам данных
Делать HTTP-запросы
Управлять внешними сервисами
Это критическое ограничение для production-систем.
В третьей части мы совершим качественный скачок — научим наших агентов взаимодействовать с внешним миром через инструменты (tools) и MCP-серверы.
К концу третьей части мы создадим агентов, способных:
Автоматизировать рабочие процессы:
Мониторить почту и автоматически отвечать на типовые запросы
Анализировать логи сервера и отправлять уведомления при ошибках
Создавать отчеты в Google Sheets на основе данных из разных источников
Управлять инфраструктурой:
Деплоить приложения через Git hooks
Мониторить метрики системы и масштабировать ресурсы
Бэкапировать базы данных по расписанию
Интегрироваться с бизнес-системами:
Синхронизировать данные между CRM и учетными системами
Обрабатывать заказы и обновлять складские остатки
Анализировать обратную связь клиентов и создавать тикеты
Мы возьмем наши интеллектуальные диалоговые системы и превратим их в полноценных цифровых сотрудников, способных:
Принимать решения на основе анализа данных
Выполнять действия в реальных системах
Реагировать на события в режиме реального времени
Интегрироваться с любыми внешними сервисами
Работать автономно без постоянного присмотра человека
Если во второй части мы создали агентов, которые умеют думать, то в третьей части мы научим их делать.
Это будет финальный переход от демонстрационных примеров к production-ready системам, способным автоматизировать реальные бизнес-процессы.
Готовы превратить ваших агентов из цифровых собеседников в цифровых сотрудников?
P.S. Если эта статья была для вас полезной, поддержите автора — подпиской, комментарием или лайком. А если хотите найти больше эксклюзивного контента, которого нет на Хабре, присоединяйтесь к моему бесплатному Телеграм-каналу «Легкий путь в Python».