27.09.2025 07:25
yakvenalex 0 508 Источник

В предыдущих частях мы создали умных агентов с памятью и мультимодельными системами. Но есть проблема — они всё ещё умные болтуны.

Критическое ограничение: агенты без рук

Наши агенты могут анализировать, классифицировать и синтезировать ответы, но НЕ МОГУТ:

  • Зайти в базу данных за информацией

  • Прочитать файл с диска

  • Сделать HTTP-запрос к API

  • Создать отчёт и сохранить его

  • Отправить email или выполнить git commit

Пользователь: "Проанализируй наши продажи за прошлый месяц"
Агент: "Пришлите мне CSV файл с данными"

Пользователь: "Создай отчёт и отправь менеджеру"  
Агент: "Я сгенерирую текст, а вы сами отправьте"

Умный, но беспомощный консультант.

Революция инструментов: когда ИИ получает руки

А теперь представьте:

Пользователь: "Проанализируй продажи за месяц"
Агент: ? Подключаюсь к БД → ? Анализирую → ? Создаю отчёт
      "Готово! Выручка +23%. Отчёт сохранён в sales_report.xlsx"

Пользователь: "Отправь CEO"  
Агент: ? Отправляю на ceo@company.com → ✅ "Отправлено!"

К концу статьи вы создадите таких агентов сами!

Что освоим:

  • Инструменты (Tools) — взаимодействие с внешними системами

  • Model Context Protocol (MCP) — универсальный стандарт инструментов

  • Файловые операции — чтение, анализ и создание документов

  • Базы данных — SQL через естественный язык

  • Безопасность — контроль действий агентов

Архитектурная эволюция

Если в предыдущих частях наша архитектура выглядела так:

Пользователь → [LangGraph] → [LLM] → Ответ

То теперь она станет такой:

Пользователь → [LangGraph] → [LLM] → [Выбор инструмента] → [Действие] → Результат
                    ↑                           ↓
                [Контекст] ←←←←←←←←← [Обратная связь]

Агент получает способность:

  • Воспринимать — анализировать задачу и контекст

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

  • Действовать — выполнять операции в реальном мире

  • Адаптироваться — корректировать план на основе результатов

Связь с предыдущими частями

Всё, что мы изучили ранее, остаётся актуальным и важным:

Из первой части нам понадобится:

  • Архитектура графов для создания сложных workflow с инструментами

  • Условные рёбра для маршрутизации между разными типами действий

  • Циклы для повторных попыток при ошибках операций

Из второй части мы будем использовать:

  • Структурированные JSON-ответы для получения параметров инструментов

  • Мультимодельные системы — разные LLM для разных типов операций

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

Кроме того, для работы нам понадобится доступ к любой нейросети. В качестве примера я буду использовать далее DeepSeek — он доступен без VPN и прокси, поэтому подойдет всем. Однако для более серьезных задач рациональнее выбирать ChatGPT или Claude.

Если же вы не хотите тратить время на настройку прокси под полноценные проекты с мощными моделями, удобный вариант выглядит так: на локальном компьютере можно работать через обычный VPN, а для деплоя использовать сервис Amvera Cloud. Это решение удобно и быстро, а главное — там уже встроена поддержка прокси для OpenAI и Claude, что избавляет от лишних трудностей с подключением.

Также через сервис Amvera можно оформить доступ к передовой модели GPT‑5. Это особенно удобно для тех, у кого нет иностранной карты.

Дорожная карта статьи

Мы пройдём путь от простого к сложному:

  1. Теория инструментальности — разберём, что отличает агента с инструментами от чистого болтуна

  2. Введение в MCP — изучим стандарт, который изменит мир ИИ-инструментов

  3. Практика: напишем ряд практических кейсов, демонстрирующих теоритические концепции

  4. Безопасность и ограничения — как не дать агентам разрушить ваш сервер

«Руки» для ИИ-агента: от теории к практике

Начать сразу хочу с того, что тему инструментов (tools) я достаточно подробно рассматривал в своей статье «Как научить нейросеть работать руками: создание полноценного ИИ-агента с MCP и LangGraph за час», но тогда данная тема была рассмотрена в отрыве от LangGraph.

Поэтому сейчас я коротко и в общих чертах напомню, как работают «руки» у ИИ-агентов, а если хочется более глубокого погружения — читайте ту статью и возвращайтесь сюда.

Что такое инструменты и зачем они нужны

Я говорю про «руки» не случайно. На более профессиональном языке данные «руки» называются инструментами (tools).

Технически инструмент — это простая функция, которая имеет специальные подсказки для нейросетей, позволяющие ИИ понимать, что эта функция написана специально для них и как её использовать.

Тут вопрос хорошо проиллюстрирует простой пример.

Простой пример: математика vs функция

Допустим, у нас есть математическая формула определения длины окружности по диаметру: C = πd, где C — длина окружности, d — диаметр окружности, π ≈ 3,14.

Мы можем предположить, что крупные языковые модели типа ChatGPT, LLaMA3, DeepSeek и прочие знают о данной формуле. И скорее всего, если вы попросите у них: «Братишка, у меня есть круг с диаметром 123 см, посчитай, пожалуйста, длину», — он, скорее всего, посчитает правильно.

Как работает ИИ БЕЗ инструментов:

❌ Пользователь: "Посчитай длину окружности при диаметре 123 см"
❌ ИИ: *лезет в математические знания*
    → *вспоминает формулу C = πd*
    → *вычисляет 3.14159 × 123*
    → *выдаёт результат ≈ 386.5 см*

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

Как работает ИИ С инструментами:

Но представьте ситуацию, при которой у вас уже есть готовая функция, способная рассчитывать эту длину окружности:

def calculate_circumference(diameter: float) -> float:
    """
    Вычисляет длину окружности по диаметру

    Args:
        diameter: Диаметр окружности в см
    Returns:
        Длина окружности в см
    """
    return 3.14159 * diameter

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

✅ Пользователь: "Посчитай длину окружности при диаметре 123 см"
✅ ИИ: *понимает задачу*
    → *видит доступный инструмент calculate_circumference*
    → *вызывает функцию calculate_circumference(123)*
    → *получает точный результат 386.5956*
    → *отвечает пользователю*

Почему инструменты лучше встроенных знаний

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

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

  • Точность: Ваша функция всегда вернёт одинаковый результат для одинаковых входных данных

  • Скорость: Питону нужны миллисекунды для выполнения расчёта, ИИ может «думать» секунды

  • Контроль: Вы точно знаете, какая логика выполняется

  • Сложность: Можете создать функцию с 10+ параметрами без объяснения математики ИИ

Как разработчики, мы можем легко создать функцию с десятком параметров, которую Python выполнит за миллисекунды — намного быстрее и точнее, чем ИИ будет вычислять "в голове".

Эволюция сложности: от калькулятора к enterprise-системам

С инструментами мы можем пойти намного дальше простой математики.

Уровень 1: Обработка данных

def analyze_sales_data(json_data: dict) -> str:
    """Анализирует данные продаж и создаёт CSV-отчёт"""
    # Ваша бизнес-логика
    return "sales_report.csv создан"

Уровень 2: Работа с базой данных

def execute_sql_query(query: str) -> list:
    """Выполняет безопасный SQL-запрос к базе данных"""
    # Подключение к БД, выполнение запроса
    return results

Уровень 3: Интеграция с внешними системами

def send_email_notification(recipient: str, subject: str, body: str) -> bool:
    """Отправляет email через корпоративную почтовую систему"""
    # Интеграция с Exchange/Gmail API
    return True

Уровень 4: DevOps автоматизация

def deploy_application(branch: str, environment: str) -> dict:
    """Деплоит приложение в указанное окружение"""
    # Git pull, Docker build, Kubernetes deploy
    return {"status": "success", "url": "https://app.company.com"}

А что если у нас будет такая функция, которая на вход примет большой JSON-массив (в Python — словарь) и на основании этого массива сгенерирует CSV-отчёт или сделает запись в SQL-базе данных?

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

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

Реальный пример: ИИ-помощник офис-менеджера

Представьте диалог с агентом, у которого есть набор инструментов:

Пользователь: "Проанализируй продажи за декабрь и отправь отчёт директору"

ИИ: ? Подключаюсь к базе данных...
    ? Выполняю запрос: SELECT * FROM sales WHERE month = 'December'
    ? Анализирую 1,247 записей...
    ? Создаю отчёт в формате Excel...
    ? Отправляю на director@company.com...

    ✅ Готово! Отчёт показывает рост продаж на 23%.
       Детальный анализ отправлен директору.

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

  1. Инструмент БД: get_sales_data(month="December")

  2. Инструмент анализа: analyze_sales(data)

  3. Инструмент создания файлов: create_excel_report(analysis)

  4. Инструмент email: send_email(recipient, file, subject)

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

Видите паттерн? ИИ становится оркестром ваших функций:

  • Чтение данных: Файлы, БД, API → структурированная информация

  • Обработка: Анализ, вычисления, трансформация данных

  • Создание контента: Отчёты, документы, визуализации

  • Коммуникация: Email, Slack, создание задач в Jira

  • Автоматизация: Деплой, мониторинг, резервное копирование

Архитектурный сдвиг

Если раньше наша архитектура выглядела так:

Пользователь → [ИИ] → Текстовый ответ

То теперь она становится такой:

Пользователь → [ИИ] → [Выбор инструмента] → [Действие] → Результат в реальном мире
                ↑                                            ↓
            [Контекст] ←←←←←←←←← [Обратная связь о результате]

Проблема стандартизации

Звучит круто, но как технически это реализовать? Как нейросеть узнаёт, какие функции у неё есть и как их вызывать?

До недавнего времени каждый разработчик решал это по-своему:

  • OpenAI использовали свой формат Function Calling

  • Anthropic имели свой подход к Tool Use

  • Local модели требовали кастомных адаптеров

Результат? Хаос совместимости.

Решение: LangChain Tools

В LangChain проблема решается элегантно — декоратор @tool превращает любую Python-функцию в инструмент для ИИ. Агент автоматически получает описание функции и может её вызывать.

Создаем первые инструменты

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

def get_quote():
    response = requests.get(
        "https://api.forismatic.com/api/1.0/",
        params={
            "method": "getQuote",
            "format": "json",
            "lang": "ru"
        }
    )
    return response.json()

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

{
  "quoteText": "Не ищите злой умысел там, где все можно объяснить глупостью.",
  "quoteAuthor": "Наполеон Бонапарт",
  "senderName": "",
  "senderLink": "",
  "quoteLink": "http://forismatic.com/ru/b7802a6b89/"
}

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

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

Давайте инициируем модель. Делаем на своё усмотрение, но я вызову официальный адаптер DeepSeek:

from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv

load_dotenv()

llm = ChatDeepSeek(model="deepseek-chat")

Обновлённое состояние с add_messages

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

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


class AgentState(TypedDict):
    """Состояние агента, содержащее последовательность сообщений."""
    messages: List[BaseMessage]

Сейчас же сделаем наш пример более профессиональным.

Прежде чем перейти к узлам, обновим состояние агента. В новой версии используем более продвинутый подход к управлению сообщениями (забегая вперед, скажу что именно так выглядит состояние в реактивных агентах):

from langgraph.graph.message import add_messages


class AgentState(TypedDict):
    """Состояние агента, содержащее историю сообщений"""
    messages: Annotated[Sequence[BaseMessage], add_messages]

Зачем Sequence вместо List?

  • Sequence[BaseMessage] — более общий тип, который включает списки, кортежи и другие последовательности

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

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

Что такое add_messages?

add_messages — это встроенная reducer-функция LangGraph, которая:

# Без add_messages (ручное управление)
def manual_node(state: AgentState) -> dict:
    new_message = AIMessage(content="Ответ")
    return {"messages": state["messages"] + [new_message]}  # Ручное слияние


# С add_messages (автоматическое управление)
def auto_node(state: AgentState) -> dict:
    new_message = AIMessage(content="Ответ")
    return {"messages": [new_message]}  # add_messages сам добавит к существующим

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

  • Автоматическое слияние новых сообщений с существующими

  • Предотвращение дублирования сообщений

  • Оптимизированная работа с памятью

  • Упрощение логики узлов графа

Новый тип сообщений: ToolMessage

В прошлой части мы подробно говорили о типах сообщений. Напоминаю, что мы рассматривали SystemMessage, AIMessage и HumanMessage. Сегодня к данному списку добавятся ToolMessage — это сообщения, которые по сути являются ответом наших инструментов.

В нашем случае тип сообщения ToolMessage нам будет важен для определения, стоит ли продолжать. Так как если тип сообщения ToolMessage последний, то после ответа от нейросети на базе сообщения инструмента мы должны будем завершить работу графа (далее будет понятнее).

from langchain_core.messages import (
    BaseMessage,
    SystemMessage,
    HumanMessage,
    AIMessage,
    ToolMessage,
)

Создание профессионального инструмента

Теперь нам нужно трансформировать нашу функцию в инструмент. Для этого нужно:

  1. Добавить документирование внутри функции (подробное описание что делает наша функция)

  2. Добавить аннотацию исходящего типа (нужно показать что будет возвращать наша функция)

  3. САМОЕ ВАЖНОЕ: Необходимо повесить специальный декоратор, который послужит подсказкой нейросетям, что это не просто функция, а инструмент

Выполним импорт декоратора:

from langchain_core.tools import tool

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

Сначала создадим модель для структурированного возврата данных:

from pydantic import BaseModel


class Quote(BaseModel):
    quote: str
    author: str

Трансформируем функцию в асинхронный инструмент

import aiohttp


@tool
async def get_quote() -> Quote:
    """Получить случайную мотивационную цитату в формате {"quote": "цитата", "author": "автор"}."""
    try:
        async with aiohttp.ClientSession() as session:
            params = {
                "method": "getQuote",
                "format": "json",
                "lang": "ru"
            }

            async with session.get(
                "https://api.forismatic.com/api/1.0/",
                params=params,
                timeout=aiohttp.ClientTimeout(total=5)
            ) as response:
                # Пробуем декодировать как JSON
                data = await response.json()
                print(data)
                quote = data.get("quoteText", "").strip()
                author = data.get("quoteAuthor", "").strip()

                if quote:
                    return {"quote": quote, "author": author}
                else:
                    return {"quote": "Работа не волк. Никто не волк. Только волк — волк.",
                            "author": "Джейсон Стетхем"}
    except Exception as e:
        print(f"Ошибка при получении цитаты: {e}")
        return {"quote": "Если закрыть глаза, становится темно.", "author": "Джейсон Стетхем"}

      
tools = [get_quote]

Что изменилось:

  • Асинхронность: Функция теперь async для неблокирующих HTTP-запросов (можно было оставить синхронной, асинхронность добавил просто для примера)

  • Структурированный возврат: Используем Pydantic модель Quote

  • Обработка ошибок: Fallback-цитаты от Джейсона Стетхема при сбоях

  • Таймауты: Защита от зависания запросов

Создание узлов графа

Привязка инструментов к модели

# Привязываем инструменты к модели сразу
llm = ChatDeepSeek(model="deepseek-chat").bind_tools(tools)

Асинхронный узел модели

async def model_call(state: AgentState) -> AgentState:
    system_prompt = SystemMessage(
        content="Ты моя система. Ответь на мой вопрос исходя из доступных для тебя инструментов"
    )
    messages = [system_prompt] + list(state["messages"])
    response = await llm.ainvoke(messages)
    return {"messages": [response]}

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

async def should_continue(state: AgentState) -> str:
    """Проверяет, нужно ли продолжить выполнение или закончить."""
    messages = state["messages"]
    last_message = messages[-1]

    # Если последнее сообщение от AI и содержит вызовы инструментов - продолжаем
    if isinstance(last_message, AIMessage) and last_message.tool_calls:
        return "continue"

    # Иначе заканчиваем
    return "end"

Сборка и запуск графа

import asyncio


async def main():
    # Создание графа
    graph = StateGraph(AgentState)
    graph.add_node("our_agent", model_call)
    tool_node = ToolNode(tools=tools)
    graph.add_node("tools", tool_node)

    # Настройка потока
    graph.add_edge(START, "our_agent")
    graph.add_conditional_edges(
        "our_agent", should_continue, {"continue": "tools", "end": END}
    )
    graph.add_edge("tools", "our_agent")

    # Компиляция и запуск
    app = graph.compile()
    result = await app.ainvoke(
        {
            "messages": [
                HumanMessage(
                    content="Дай мне мотивационную цитату и расскажи пару слов о авторе, если он известен"
                )
            ]
        }
    )

    # Показываем результат
    print("=== Полная история сообщений ===")
    for i, msg in enumerate(result["messages"]):
        print(f"{i+1}. {type(msg).__name__}: {getattr(msg, 'content', None)}")
        if hasattr(msg, "tool_calls") and msg.tool_calls:
            print(f"   Tool calls: {msg.tool_calls}")

if __name__ == "__main__":
    asyncio.run(main())

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

Поток выполнения:

  1. START → our_agent: Модель получает запрос пользователя

  2. our_agent анализирует, что нужна цитата, генерирует tool_call

  3. should_continue возвращает "continue" — есть вызов инструмента

  4. tools: ToolNode выполняет get_quote(), возвращает структурированный результат

  5. tools → our_agent: Модель получает результат и формулирует финальный ответ

  6. should_continue возвращает "end" — завершаем работу

Подробнее рассматривал архитектуру графовых конструкций в первой части курса.

Ожидаемый результат:

=== Полная история сообщений ===
1. HumanMessage: Дай мне мотивационную цитату и расскажи пару слов о авторе, если он известен
2. AIMessage: вызов инструмента get_quote&;
3. ToolMessage: {"quote": "Не ищите злой умысел там, где все можно объяснить глупостью.", "author": "Наполеон Бонапарт"}
4. AIMessage: Вот мотивационная цитата: "Не ищите злой умысел там, где все можно объяснить глупостью." — Наполеон Бонапарт. Это высказывание принадлежит великому французскому полководцу и императору...

Этот пример демонстрирует основы работы с инструментами в LangChain, но у такого подхода есть ограничения при работе с внешними системами и готовыми решениями.

В следующем разделе мы изучим Model Context Protocol (MCP) — универсальный стандарт, который открывает доступ к экосистеме готовых инструментов для любых задач.

MCP или когда не хочется писать свои инструменты

Теме MCP у меня уже была посвящена статья «Как создать MCP-сервер и научить ИИ работать с любым кодом и инструментами через LangGraph», в рамках которой я не только детально разобрал тему MCP, но и наглядно продемонстрировал, как создавать и поднимать свой собственный MCP-сервер. Поэтому сейчас коротко освежим знания, а для тех, кому нужно более глубокое погружение — предлагаю ознакомиться с этой статьей и после двигаться далее.

Что такое MCP-сервер простыми словами

Если говорить совсем простыми словами, то MCP-сервер — это совокупность инструментов (тулсов), объединённых какой-то общей задачей. В моём примере такой задачей была математика, к другим примерам можно отнести:

  • Работу с базами данных (под каждую базу данных свой MCP-сервер)

  • Работу с информацией из сети

  • Работу с файловой системой

  • Автоматизацию браузера и многое другое

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

  • Отображение списка файлов в указанной директории

  • Создание файла

  • Редактирование

  • Удаление файла

То есть, если вы самостоятельно разработали ряд функций по работе с файловой системой, которые можете вызывать руками, то при определённых манипуляциях вы сможете сделать так, что ИИ-агент, когда того требует задача, самостоятельно вызывал инструменты данного MCP-сервера.

Интеграция MCP с LangGraph

В случае с LangGraph и LangChain утверждение про «совокупность инструментов» максимально точное, так как вам ничего не помешает интегрировать в свой граф как собственные инструменты (функции), которые вы написали самостоятельно, так и набор инструментов из существующего MCP-сервера.

Для подключения MCP-серверов в свои графы LangChain предоставляет полезный адаптер — MultiServerMCPClient, который импортируется следующим образом:

from langchain_mcp_adapters.client import MultiServerMCPClient

Технически его роль сводится к двум задачам:

  1. Подключиться к существующему MCP-серверу

  2. Извлечь из него все инструменты

Протоколы подключения

Подключение к MCP-серверу может быть по двум основным протоколам:

1. stdio — это когда MCP-клиент запускается локально на вашей машине
2. streamable_http — та же логика, но подключение к MCP происходит удалённо по HTTP-протоколу

Простота интеграции

Если вы разобрались с биндом обычных тулзов, то и вопросов бинда тулзов от MCP-сервера у вас тоже возникнуть не должно. Всё сводится к следующему:

  1. Подключаемся к MCP-серверу

  2. Извлекаем все инструменты в обычный список

  3. Биндим (подключаем) инструменты к ИИ-агенту, как показано в примере выше

Практический пример: объединение своих и MCP-инструментов

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

Добавим функцию, которая объединит наш инструмент с инструментами от существующего MCP-сервера:

async def get_all_tools():
    """Получение всех инструментов: ваших + MCP"""
    # Настройка MCP клиента
    mcp_client = MultiServerMCPClient(
        {
            "filesystem": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-filesystem", "."],
                "transport": "stdio",
            }
        }
    )

    # Получаем MCP инструменты
    mcp_tools = await mcp_client.get_tools()

    # Объединяем ваши инструменты с MCP инструментами
    return [get_quote] + mcp_tools

Обратите внимание: MultiServerMCPClient принимает вложенный словарь. То есть теоретически вы сможете подключить сколько угодно MCP-серверов и извлечь из них все инструменты.

Более того, при необходимости вы сможете не включать в финальный список все инструменты (подробнее об этом далее, в разделе про безопасность), но в данном примере я получил все инструменты от MCP-сервера server-filesystem и объединил их с нашей функцией с цитатами.

Важные моменты установки

Важный момент по поводу локальных MCP (транспорт stdio). Для того чтобы они работали, часто требуется локальная установка. В случае с server-filesystem MCP установка будет иметь следующий вид:

npm install -g @modelcontextprotocol/server-filesystem

Также, в зависимости от команды, возможно, вам необходимо будет установить дополнительный софт. Например, Python с библиотекой uv, Node.js последней версии, npm и так далее.

При подключении инструментов MCP‑сервера в LangGraph мы описываем структуру клиента, где задаем произвольное название инструмента, указываем команду запуска (условный бинарь или npm‑пакет), выбираем транспорт обмена данными и определяем аргументы.

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

Чтобы использовать такие возможности полноценно, важно внимательно читать документацию конкретного MCP‑сервера, так как именно там описано, какие аргументы поддерживаются и как правильно их конфигурировать.

Полный код интеграции

# Получаем все инструменты (свои + MCP)
tools = asyncio.run(get_all_tools())
llm = ChatDeepSeek(model="deepseek-chat").bind_tools(tools)

async def model_call(state: AgentState) -> AgentState:
    system_prompt = SystemMessage(
        content="В твоём распоряжении есть инструменты для работы с файловой системой и инструмент для получения мотивационных цитат на русском языке."
    )
    messages = [system_prompt] + list(state["messages"])
    response = await llm.ainvoke(messages)
    return {"messages": [response]}

В целом это всё, что требуется. Теперь давайте вызовем наш граф с просьбой сохранить полученный результат в файл quote.txt:

result = await app.ainvoke(
    {
        "messages": [
            HumanMessage(
                content="Найди мотивационную цитату на русском языке и сохрани её в файл quote.txt со всей информацией, которую ты найдёшь по автору цитаты."
            )
        ]
    }
)

Результат работы

Консольный вывод:

Secure MCP Filesystem Server running on stdio
Allowed directories: [ '/home/yakvenalex/PythonProjects/ai_agent_project' ]
{'quoteText': 'Умение остановиться, поразмышлять и удивиться — одна из характеристик, которыми обладают мудрейшие люди планеты. Попробуй обучиться этому навыку. Польза от него тебе будет огромна.', 'quoteAuthor': '', 'senderName': '', 'senderLink': '', 'quoteLink': 'http://forismatic.com/ru/78259ab202/'}

=== Полная история сообщений ===
1. HumanMessage: Найди мотивационную цитату на русском языке и сохрани её в файл quote.txt...
2. AIMessage: [Tool calls: get_quote]
3. ToolMessage: {"quote": "Умение остановиться...", "author": ""}
4. AIMessage: Я получил мотивационную цитату, но автор не указан. Давайте сохраним эту цитату в файл quote.txt:
   [Tool calls: write_file с содержимым файла]
5. ToolMessage: Successfully wrote to quote.txt
6. AIMessage: Готово! Я нашёл мотивационную цитату на русском языке и сохранил её в файл quote.txt...

Содержимое файла quote.txt:

Умение остановиться, поразмышлять и удивиться — одна из характеристик, которыми обладают мудрейшие люди планеты. Попробуй обучиться этому навыку. Польза от него тебе будет огромна.

Автор: неизвестен

Дата сохранения: текущая дата
Тип: мотивационная цитата
Тема: мудрость, размышление, саморазвитие

Интеллектуальное поведение агента

Что примечательно: Агент самостоятельно выбрал формат записи в файл — я не указывал структуру данных. Более того, в графе не было отдельных узлов для двух действий, но агент сам понял, что нужно:

  1. Вызвать функцию с цитатами

  2. Вызвать инструменты для создания и редактирования файловой системы

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

Упрощение с ReAct агентом

Создание графа вручную дает полный контроль, но требует много кода. А что если есть более простой способ?

Встречайте ReAct агентов — готовое решение, которое автоматически обрабатывает логику принятия решений и вызова инструментов.

Что такое ReAct агент?

ReAct (Reasoning + Acting) — это встроенный паттерн в LangGraph, который автоматически:

  • Анализирует задачу (Reasoning)

  • Выбирает подходящие инструменты

  • Выполняет действия (Acting)

  • Повторяет цикл до решения задачи

Код с ReAct агентом (значительно проще!)

from langgraph.prebuilt import create_react_agent


async def main_react():
    # Получаем все инструменты
    tools = await get_all_tools()

    # Создаём ReAct агента (вся логика графа уже встроена!)
    agent = create_react_agent(
        model=ChatDeepSeek(model="deepseek-chat"),
        tools=tools,
        state_modifier="В твоём распоряжении есть инструменты для работы с файловой системой и получения мотивационных цитат."
    )

    # Запускаем задачу
    result = await agent.ainvoke({
        "messages": [
            HumanMessage(
                content="Найди мотивационную цитату и сохрани её в файл quote_react.txt с подробной информацией об авторе"
            )
        ]
    })

    print("=== Результат ReAct агента ===")
    print(result["messages"][-1].content)

if __name__ == "__main__":
    asyncio.run(main_react())

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

Ручная графовая цепочка:

  • Полный контроль над логикой

  • Кастомизация каждого шага

  • Больше кода

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

ReAct агент:

  • Минимум кода (3 строки vs 50+)

  • Встроенная логика рассуждений

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

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

Экосистема готовых MCP-серверов

Красота MCP в том, что нам не нужно изобретать велосипед — уже существует богатая экосистема готовых решений для самых разных задач:

Популярные MCP-серверы:

  • @modelcontextprotocol/server-filesystem — работа с файлами (пример выше)

  • @modelcontextprotocol/server-postgres — интеграция с PostgreSQL

  • @modelcontextprotocol/server-git — Git операции

  • @modelcontextprotocol/server-fetch — HTTP-запросы

  • @modelcontextprotocol/server-brave-search — поиск в интернете

Преимущества экосистемы:

  • Быстрый старт — не нужно писать инструменты с нуля

  • Проверенные решения — серверы тестируются сообществом

  • Стандартизация — единый интерфейс для всех инструментов

  • Обновления — автоматические улучшения от разработчиков

Резюме: MCP как ускоритель разработки

MCP превращает создание агентов из недель программирования в часы конфигурации. Вместо написания десятков инструментов с нуля, вы:

  1. Находите подходящий MCP-сервер

  2. Подключаете через MultiServerMCPClient

  3. Комбинируете с собственными инструментами

  4. Получаете готового агента

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

Магия памяти и интерактивный режим

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

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

  1. Создание текстового документа / подключение к существующему текстовому документу

  2. Чтение документа

  3. Отображение текущей версии документа в консоли

  4. Редактирование документа

  5. При необходимости создание нового документа и переключение между ними

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

Я же сейчас не буду тратить время на такое решение задачи и спрошу вас: а что если мы сделаем так, что агент сам будет решать не только о том какой инструмент выбирать (этим нас уже не удивить), а еще и самостоятельно будет управлять состоянием? Легко!

В LangGraph, кроме уже знакомого нам состояния графа, присутствует такая штука как Checkpointer (чекпойнтер) и слой персистентности.

Что такое чекпойнтер

Начну с немного занудного определения, но после постараюсь пояснить на более простом языке.

Чекпойнтер — это встроенный механизм персистентности LangGraph. Он сохраняет "снимки" состояния графа на каждом супер-шаге в рамках конкретного thread (нити). Это дает:

  • "память" между обращениями (по thread_id)

  • просмотр истории/"time-travel"

  • fault-tolerance и human-in-the-loop

То есть, в LangGraph есть механизм позволяющий, буквально, сохранять снимок сессии и далее, при помощи встроенных инструментов и зная айдишник сессии у вас появляется возможность вернуться к состоянию.

Данная штука очень полезна в классических графах и, буквально, незаменима в реактивных агентах!

Незаменима в реактивных агентах она потому что там мы работаем с неким "черным ящиком". То есть в рамках реактивного агента мы не делаем ручное управление состоянием, но при этом мы просто обязаны где-то хранить это самое состояние.

Типы чекпойнтеров

Существуют разные типы чекпойнтеров в зависимости от способа хранения состояния:

InMemorySaver

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

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

Персистентные бэкенды

Официальная линия такая: в самом langgraph встроен только in-memory; персистентные — отдельными пакетами.

  • SqliteSaver — легкая локальная персистентность (пакет langgraph-checkpoint-sqlite). Удобен для разработки, CI, небольшая нагрузка. Данные сохраняются в файл базы данных и переживают перезапуск приложения.

  • PostgresSaver — продакшен-решение (отдельный пакет langgraph-checkpoint-postgres). Дает асинхронные/синхронные классы PostgresSaver/AsyncPostgresSaver для высоконагруженных систем.

То есть, для тестов и прототипов - InMemorySaver, для локальной разработки с персистентностью - SqliteSaver, для продакшена - PostgresSaver.

Как это матчится со StateGraph (без prebuilt-агента)

Чекпойнтеры работают одинаково как с prebuilt-агентами, так и с "классическим" StateGraph:

  1. Определяем состояние (TypedDict/MessagesState и т.п.), строим граф.

  2. Компилируем с чекпойнтером:

    graph = StateGraph(...).compile(checkpointer=checkpointer)

  3. Запускаем с config={"configurable": {"thread_id": "<ид нити>"}} через invoke/ainvoke или stream/astream.

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

После записи чекпойнтеров в нить мы сможем:

  • читать текущее состояние нити: graph.get_state(config) / aget_state(...)

  • получать историю чекпойнтов: get_state_history(thread_id) и "перематывать"/форкать выполнение (time-travel)

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

Checkpointer vs Store: разбираем путаницу

Важно не путать два различных слоя работы с данными в LangGraph:

Checkpointer - "Журнал выполнения"

  • Назначение: хранение истории выполнения графа и состояния сессии

  • Что хранит: снимки state графа, метаданные шагов, историю сообщений в рамках thread_id

  • Жизненный цикл: привязан к конкретной сессии/нити

  • API: get_state(), get_state_history()

Store - "База знаний"

  • Назначение: долговременное хранение фактов, знаний и данных

  • Что хранит: документы, настройки пользователей, векторные эмбеддинги, справочную информацию

  • Жизненный цикл: независим от сессий, доступен глобально

  • API: put(namespace, key, value), get(namespace, key), search()

Практическое различие: Checkpointer помнит "как проходил диалог", Store знает "что пользователь предпочитает". Часто используют вместе: checkpointer для краткосрочной памяти выполнения, Store — для долговременной памяти знаний.

Рекомендуемый стек под различные сценарии

  • Разработка/локально: InMemorySaver или SqliteSaver (минимум настроек, быстрый старт)

  • Продакшен: PostgresSaver (надежность, масштабирование, совместим с LangGraph Platform)

  • Опционально с Store: InMemoryStore для разработки, PostgresStore для продакшена

Частые ошибки

  • Нет thread_id — "память" не цепляется к сессии и кажется, что чекпойнтер "не помнит"

  • Ожидание "магической памяти" без Store — чекпойнтер хранит хронологию выполнения, но не заменяет долгосрочную БД знаний (используй Store для LTM)

  • Импорты SqliteSaver: SqliteSaver находится в отдельном пакете (langgraph-checkpoint-sqlite), а не в основном langgraph

  • Путаница с терминологией: "сервер памяти" — это просто место, где чекпойнтер хранит данные (память процесса, SQLite файл, PostgreSQL), а не отдельный сервис

От теории к практике: создаем интерактивного помощника

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

Напоминаю, что полный исходный код этого помощника, а также все примеры из статьи в полном виде вы можете найти в моем телеграм‑канале «Легкий путь в Python».

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

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

Архитектура решения

Наш агент будет состоять из:

  • ReAct-паттерн как основа для принятия решений

  • MCP-инструменты для работы с файловой системой

  • Кастомные инструменты для управления сессией

  • InMemorySaver для персистентного хранения состояния

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

Инструменты управления сессией

Сначала создадим инструменты, которые дадут агенту возможность управлять жизненным циклом сессии:

from langchain_core.tools import tool

@tool
async def session_status() -> str:
    """Показывает статус текущей сессии и доступные файлы."""
    return "Сессия активна. Используйте filesystem инструменты для работы с документами."

@tool
async def end_session(reason: str = "Пользователь завершил работу") -> str:
    """Завершает текущую сессию работы с документами."""
    print(f"\nЗавершение сессии: {reason}")
    return f"Сессия завершена. {reason}"

Эти простые функции превращаются в мощные инструменты самоуправления агента. Он сможет проверять свой статус и принимать решения о завершении работы.

Интеграция с MCP

from langgraph_mcp import MultiServerMCPClient


async def get_all_tools():
    """Получение всех инструментов: MCP + управление сессией."""
    custom_tools = [session_status, end_session]

    try:
        mcp_client = MultiServerMCPClient({
            "filesystem": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-filesystem", "."],
                "transport": "stdio",
            }
        })
        mcp_tools = await mcp_client.get_tools()
        print(f"Подключено {len(mcp_tools)} инструментов filesystem")
        return custom_tools + mcp_tools
    except Exception as e:
        print(f"MCP недоступен, используем базовые инструменты: {e}")
        return custom_tools

Здесь мы объединяем собственные инструменты с готовыми MCP-инструментами. Если MCP-сервер недоступен — агент продолжит работу с базовым функционалом.

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

А вот здесь происходит магия:

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.messages import SystemMessage


async def create_agent():
    """Создание агента с инструментами."""
    tools = await get_all_tools()
    model = ChatDeepSeek(model="deepseek-chat", temperature=0.3)

    system_prompt = """Ты профессиональный помощник по созданию и редактированию писем и документов.

У тебя есть полный доступ к файловой системе через MCP инструменты:
- read_file, write_file - чтение и запись файлов
- list_directory - просмотр содержимого папок
- create_directory - создание папок
- move_file, copy_file - операции с файлами

ПРИНЦИПЫ РАБОТЫ:
1. Помогай пользователю создавать качественные письма и документы
2. Всегда сохраняй результаты работы в файлы
3. Предлагай улучшения и редактирование
4. Поддерживай контекст сессии - помни о созданных файлах и документах

ЗАВЕРШЕНИЕ СЕССИИ:
Используй end_session() когда:
- Пользователь явно просит завершить ("закончить", "выйти", "хватит")
- Работа полностью выполнена и пользователь доволен результатом
- После фраз типа "спасибо", "готово", "всё хорошо"

ВАЖНО: Будь полезным, дружелюбным и профессиональным!"""

    checkpointer = InMemorySaver()
    agent = create_react_agent(
        model=model,
        tools=tools,
        checkpointer=checkpointer,
        prompt=SystemMessage(content=system_prompt)
    )

    print("Агент инициализирован с персистентной памятью")
    return agent

Ключевая строка — checkpointer=InMemorySaver(). Именно она превращает обычного ReAct-агента в помощника с памятью.

Интерактивная сессия

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

from langchain_core.messages import HumanMessage, AIMessage


async def print_session_stats(agent, config):
    """Вывод статистики сессии."""
    try:
        state = await agent.aget_state(config)
        if state and "messages" in state.values:
            messages = state.values["messages"]
            human_messages = [m for m in messages if isinstance(m, HumanMessage)]
            ai_messages = [m for m in messages if isinstance(m, AIMessage)]

            print(f"Сообщений пользователя: {len(human_messages)}")
            print(f"Ответов агента: {len(ai_messages)}")
            print(f"Всего сообщений в сессии: {len(messages)}")
        else:
            print("Статистика недоступна")
    except Exception as e:
        print(f"Ошибка получения статистики: {e}")

        
async def run_interactive_session():
    """Запуск интерактивной сессии."""
    print("ИНТЕРАКТИВНЫЙ ПОМОЩНИК ПО СОЗДАНИЮ ПИСЕМ")
    print("Команды: 'выход', 'quit', 'стоп' - для завершения")
    print("Просто опишите что нужно создать или отредактировать")

    agent = await create_agent()
    config = {"configurable": {"thread_id": "document-session"}}

    try:
        while True:
            try:
                user_input = input("\nВаш запрос: ").strip()

                if not user_input:
                    continue

                if user_input.lower() in ['выход', 'quit', 'exit', 'стоп', 'stop']:
                    print("\nДо свидания!")
                    break

                print("\nОбрабатываю запрос...")

                user_message = HumanMessage(content=user_input)
                response_printed = False
                session_ended = False

                async for chunk in agent.astream(
                    {"messages": [user_message]},
                    config=config,
                    stream_mode="values"
                ):
                    if "messages" in chunk and chunk["messages"]:
                        last_msg = chunk["messages"][-1]

                        if isinstance(last_msg, AIMessage) and not response_printed:
                            print(f"\n{last_msg.content}")
                            response_printed = True

                        if hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
                            for tool_call in last_msg.tool_calls:
                                if tool_call["name"] == "end_session":
                                    session_ended = True

                if session_ended:
                    print("Агент завершил сессию")
                    break

            except KeyboardInterrupt:
                print("\n\nПолучен сигнал прерывания")
                break
            except Exception as e:
                print(f"\nОшибка при обработке запроса: {e}")
                continue

    finally:
        print("\n" + "=" * 60)
        print("СТАТИСТИКА СЕССИИ")
        await print_session_stats(agent, config)
        print("Сессия завершена")

        
# Запуск
if __name__ == "__main__":
    import asyncio
    asyncio.run(run_interactive_session())

Критически важная деталь: thread_id

Обратите внимание на строку:

config = {"configurable": {"thread_id": "document-session"}}

thread_id — это ключ к памяти агента. Без него чекпойнтер не знает, где сохранять и откуда восстанавливать состояние. Каждый уникальный thread_id создает отдельную "сессию" с независимой памятью.

Магия в действии

Что происходит под капотом при каждом вызове агента:

  1. Загрузка состояния: Чекпойнтер восстанавливает предыдущие сообщения и состояние по thread_id

  2. Добавление нового сообщения: Пользовательский ввод добавляется к истории

  3. Выполнение ReAct-цикла: Агент анализирует ситуацию и принимает решения

  4. Сохранение обновленного состояния: После каждого супер-шага состояние автоматически сохраняется

Результат

В итоге получается агент, который:

  • Помнит всю историю диалога и действий

  • Знает, какие файлы создавал ранее

  • Может продолжить работу над документами с предыдущих обращений в рамках сессии

  • Самостоятельно принимает решение о завершении работы

  • Предоставляет статистику по завершении

Это уже не просто "умный болтун", а полноценный рабочий помощник с персистентной памятью сессии. В следующих разделах мы рассмотрим еще более продвинутые кейсы использования инструментов в реальных enterprise-сценариях.

astream vs ainvoke

В примере выше я сознательно использовал astream вместо привычного ainvoke. Коротко разберёмся, что это такое и когда что выбирать.

В двух словах

  • ainvoke(...) — один вызов → один готовый результат после завершения графа/агента. Отлично, когда вам не нужно показывать прогресс: получили ответ и отдали клиенту.

  • astream(...) — поток обновлений по мере выполнения: шаги рассуждения, вызовы инструментов, промежуточные сообщения, финал. Нужен для «живого» UI/CLI и прод-API со стримингом (SSE/WebSocket).

Что именно стримится

astream умеет разные режимы (указываются параметром stream_mode):

  • "values" — слепки всего состояния после каждого шага (то, что я использовал). Удобно логировать и строить прогресс-бар.

  • "updates" — дельты по шагам (LLM → Tool → LLM). Удобно «оживлять» консоль/дешборд.

  • "messages" — поток контента сообщений (включая токены от модели, если бэкенд поддерживает токен-стриминг). Это как привычный чат — «текст набирается» по кусочкам.

Практика: для UI/консоли чаще хватает "updates" или "values". Для «чат-опыта» — "messages".

Когда что использовать в сервисе (FastAPI)

  • Просто ответить быстро (синхронный REST, без прогресса) → ainvoke.

  • Показать пользователю ход работы (длинные инструменты, файловые операции, запросы в БД/API) → astream + SSE/WebSocket.

Мини-скелеты:

ainvoke (быстрый ответ)

res = await agent.ainvoke({"messages": [HumanMessage(content=text)]}, config=config)
return {"reply": res["messages"][-1].content}

astream + SSE (прогресс в реальном времени)

async def stream():
    async for chunk in agent.astream({"messages": [HumanMessage(content=text)]},
                                     config=config,
                                     stream_mode="updates"):
        # сериализуете chunk и шлёте в клиент (SSE/WebSocket)
        yield format_sse(chunk)

return EventSourceResponse(stream())  # sse-starlette / fastapi-sse

Плюсы/минусы

  • ainvoke:

    • просто; + минимум кода; − нет прогресса; − «молчаливые» долгие операции.

  • astream:

    • UX как в чатах/IDE; + видно шаги и инструменты; − сложнее API (стрим-протокол); − нужно думать про backpressure/таймауты.

Нюансы и грабли

  • Не путайте уровни стриминга. astream стримит шаги/сообщения графа. Стрим токенов модели появляется только при stream_mode="messages" и если LLM поддерживает токен-стрим.

  • Фильтруйте шум. В "values" прилетает весь state — отдавайте клиенту только то, что имеет смысл (последний шаг, статус инструмента, кусок текста).

  • Таймауты и отмена. Для длинных тулзов ставьте таймауты и обрабатывайте отмену клиентского соединения (закрыли вкладку — прекратите генератор).

  • Безопасные логи. Не стримьте секреты и сырые payload’ы инструментов наружу — делайте санитайз.

Итого: если нужен «готовый ответ» — ainvoke. Если нужен «живой» прогресс и/или чат-опыт — astream с подходящим stream_mode.

Безопасность: даём агенту руки, но держим поводок

Как только ваш агент получает инструменты (файловая система, БД, git, HTTP-запросы) — он становится не просто болтуном, а исполнителем. А значит, открываются и риски.

Основные угрозы

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

  • База данных — SQL-инъекции, случайное DROP TABLE, массовые апдейты.

  • Сеть — утечки токенов, SSRF-атаки, злоупотребление API.

  • DevOps-интеграции — деплой не в то окружение, остановка продакшена.

Простые практики защиты

  1. Принцип наименьших прав
    Давайте агенту ровно тот набор инструментов, который реально нужен.
    Нужен только read_file → не подключайте delete_file.

  2. Песочница
    Ограничьте рабочую директорию (chroot/alloweddirs) или схему БД. _Filesystem MCP умеет задавать список разрешённых путей.

  3. Валидация ввода
    Проверяйте параметры инструментов: whitelist таблиц, проверка формата файлов, лимиты на размер.

  4. Интерактивное подтверждение (human-in-the-loop)
    Для опасных действий (удаление, деплой) делайте шаг подтверждения: агент предлагает → человек жмёт «ОК».

  5. Лимиты и таймауты
    Ограничивайте длину запросов, количество вызовов, время выполнения инструментов.

  6. Логирование и аудит
    Логируйте все вызовы инструментов и сохраняйте историю в чекпойнтер. Это важно для «разбора полётов».

  7. Разделение окружений
    Отдельные ключи и базы для теста/прода. Никогда не гоняйте агента «с прод-ключами» на локальной разработке.

Практическое правило

Чем больше у агента «рук», тем короче должен быть «поводок».

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

Реальная мультиагентная система

Мы прошли путь от простых инструментов до ReAct-агентов. Теперь настало время для финального уровня — создания настоящей мультиагентной системы.

Представьте задачу: нам нужно исследовать тему, собрать данные из интернета, структурировать их и создать профессиональный отчёт. Одиночный агент справится, но результат будет посредственным. А что если разделить задачу между специалистами?

Давайте создадим команду агентов, где каждый эксперт в своей области.

Для простоты используем собственные функции вместо MCP-серверов — так вы лучше поймете принципы. В реальном проекте легко заменить их на готовые MCP-инструменты.

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

Наша система состоит из трёх специализированных агентов, каждый из которых выполняет свою роль:

  1. Исследователь — ищет информацию в интернете и анализирует данные

  2. Дата-инженер — структурирует данные, создаёт CSV и работает с SQLite

  3. Редактор — создаёт итоговые отчёты в формате Markdown

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

Инструментарий системы

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

Web-инструменты (для Исследователя):

@tool
async def web_search(query: str) -> List[str]:
    """Простейший web-поиск: возвращает список ссылок."""
    try:
        url = "https://duckduckgo.com/html/"
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.post(url, data={"q": query})
            # Упрощенный парсинг ссылок
            hrefs = []
            for line in r.text.splitlines():
                if 'result__a' in line and 'href=' in line:
                    start = line.find('href="') + 6
                    end = line.find('"', start)
                    link = line[start:end]
                    if link.startswith("http"):
                        hrefs.append(link)
                if len(hrefs) >= 5:
                    break
        return hrefs or ["https://example.com"]
    except Exception as e:
        return [f"ERROR:{e}"]

@tool
async def fetch_url(url: str) -> str:
    """Скачивает HTML/текст по URL."""
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.get(url)
            return r.text[:50_000]  # ограничим объём
    except Exception as e:
        return f"ERROR:{e}"

Файловые инструменты (для всех агентов):

@tool
async def fs_write_text(path: str, content: str) -> str:
    """Пишет текст в файл внутри рабочей директории."""
    full = (WORKDIR / Path(path)).resolve()
    if not str(full).startswith(str(WORKDIR)):
        return "ERROR: path outside sandbox"
    full.parent.mkdir(parents=True, exist_ok=True)
    full.write_text(content, encoding="utf-8")
    return f"OK: wrote {full}"

  
@tool
async def csv_write_rows(path: str, rows: List[List[str]]) -> str:
    """Создаёт/перезаписывает CSV с переданными строками."""
    full = (WORKDIR / Path(path)).resolve()
    if not str(full).startswith(str(WORKDIR)):
        return "ERROR: path outside sandbox"
    with open(full, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerows(rows)
    return f"OK: wrote {full}"

Базы данных (для Дата-инженера):

@tool
async def sqlite_execute(db_path: str, sql: str, params: Optional[List] = None) -> str:
    """Выполняет SQL (DDL/DML). Возвращает 'rows affected'."""
    full = (WORKDIR / Path(db_path)).resolve()
    if not str(full).startswith(str(WORKDIR)):
        return "ERROR: path outside sandbox"
    full.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(full)
    try:
        cur = conn.cursor()
        cur.execute(sql, params or [])
        conn.commit()
        return f"OK: {cur.rowcount} rows affected"
    finally:
        conn.close()

@tool
async def sqlite_query(db_path: str, sql: str, params: Optional[List] = None) -> List[List[str]]:
    """SELECT-запрос, возвращает строки как списки строк."""
    full = (WORKDIR / Path(db_path)).resolve()
    if not str(full).startswith(str(WORKDIR)) or not full.exists():
        return [["ERROR", "db not found or outside sandbox"]]
    conn = sqlite3.connect(full)
    try:
        cur = conn.cursor()
        cur.execute(sql, params or [])
        rows = cur.fetchall()
        return [[str(x) for x in row] for row in rows]
    finally:
        conn.close()

Состояние оркестратора

Все агенты работают в рамках единого состояния, которое передаётся между этапами:

class OrchestratorState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    topic: str             # Тема исследования
    urls: List[str]        # Найденные ссылки
    db_path: str           # Путь к SQLite базе
    csv_path: str          # Путь к CSV файлу
    report_path: str       # Путь к итоговому отчёту

Специализированные агенты

Агент-Исследователь

def mk_researcher():
    model = ChatDeepSeek(model="deepseek-chat", temperature=0.2)
    sys = SystemMessage(content=(
        "Ты Исследователь. Твоя задача: найти 3–5 релевантных ссылок, "
        "при необходимости коротко скачать содержимое по 1–2 из них и выдать сжатую сводку. "
        "Всегда используй инструменты web_search и fetch_url при необходимости."
    ))
    tools = [web_search, fetch_url]
    return create_react_agent(model=model, tools=tools, prompt=sys, checkpointer=MemorySaver())

Задачи Исследователя:

  • Получает тему исследования из состояния

  • Ищет релевантные ссылки через web_search

  • При необходимости загружает контент ключевых страниц

  • Формирует сжатую сводку с фактами и инсайтами

Агент Дата-инженер

def mk_data_engineer():
    model = ChatDeepSeek(model="deepseek-chat", temperature=0.1)
    sys = SystemMessage(content=(
        "Ты Дата-инженер. Тебе дают сводку/факты. "
        "Сформируй таблицу CSV (заголовки + строки), запиши её, "
        "создай/обнови таблицу в SQLite и вставь данные. "
        "Всегда используй csv_write_rows, sqlite_execute, sqlite_query по необходимости."
    ))
    tools = [csv_write_rows, sqlite_execute, sqlite_query]
    return create_react_agent(model=model, tools=tools, prompt=sys, checkpointer=MemorySaver())

Задачи Дата-инженера:

  • Получает сводку от Исследователя

  • Структурирует данные в табличный формат (source, insight)

  • Создаёт CSV файл с результатами

  • Настраивает SQLite таблицу и загружает данные

  • Проверяет корректность загрузки через COUNT(*)

Агент-Редактор

def mk_writer():
    model = ChatDeepSeek(model="deepseek-chat", temperature=0.3)
    sys = SystemMessage(content=(
        "Ты Редактор. Получив данные (сводку, CSV, SQL-выборки), создай читабельный отчёт в Markdown. "
        "Сохрани результат на диск через fs_write_text и верни краткое резюме."
    ))
    tools = [fs_write_text, csv_read_rows, sqlite_query]
    return create_react_agent(model=model, tools=tools, prompt=sys, checkpointer=MemorySaver())

Задачи Редактора:

  • Читает данные из CSV и SQLite

  • Формирует красивый отчёт в Markdown

  • Добавляет аналитические выводы и рекомендации

  • Сохраняет итоговый отчёт в файл

Граф оркестратора

Логика координации агентов реализована через StateGraph:

# Узлы оркестратора
async def node_research(state: OrchestratorState):
    msgs = [
        HumanMessage(content=f"Тема исследования: {state['topic']}. "
                             f"Найди 3–5 ссылок и сделай короткую сводку.")
    ]
    res = await researcher.ainvoke({"messages": msgs}, config={"configurable": {"thread_id": "research"}})
    # Извлекаем найденные ссылки из ответа
    text = res["messages"][-1].content
    urls = [u for u in text.split() if u.startswith("http")][:5]
    return {"messages": res["messages"], "urls": urls}

async def node_data(state: OrchestratorState):
    msgs = [
        HumanMessage(content=(
            "На основе предыдущей сводки/фактов сформируй таблицу с колонками "
            "[source, insight] и 3–8 строк. "
            f"Запиши CSV в {state['csv_path']}. "
            f"Далее создай таблицу sales_insights(source TEXT, insight TEXT) в БД {state['db_path']} "
            "и вставь все строки. Верни выборку COUNT(*) для проверки."
        ))
    ]
    res = await data_eng.ainvoke({"messages": msgs}, config={"configurable": {"thread_id": "data"}})
    return {"messages": res["messages"]}

async def node_write(state: OrchestratorState):
    msgs = [
        HumanMessage(content=(
            f"Собери читабельный отчёт в Markdown по теме '{state['topic']}'. "
            f"Используй данные из CSV {state['csv_path']} и выборку из SQLite {state['db_path']} "
            f"(сделай запрос COUNT(*) из sales_insights). "
            f"Сохрани отчёт в {state['report_path']} через fs_write_text. "
            "Верни короткое резюме и путь к файлу."
        ))
    ]
    res = await writer.ainvoke({"messages": msgs}, config={"configurable": {"thread_id": "write"}})
    return {"messages": res["messages"]}

# Сборка графа
graph = StateGraph(OrchestratorState)
graph.add_node("research", node_research)
graph.add_node("data", node_data)
graph.add_node("write", node_write)

graph.set_entry_point("research")
graph.add_edge("research", "data")
graph.add_edge("data", "write")

app = graph.compile(checkpointer=MemorySaver())

Запуск и результаты

Система запускается с минимальной конфигурацией:

async def main():
    # начальное состояние оркестратора
    init = {
        "messages": [HumanMessage(content="Старт")],
        "topic": "_dyn_: влияние погодных условий на продажи кофе в Нидерландах",
        "urls": [],
        "db_path": "data.sqlite",
        "csv_path": "dataset.csv",
        "report_path": "report.md",
    }
    config = {"configurable": {"thread_id": "orchestrator-demo"}}

    result = await app.ainvoke(init, config=config)

    print("\n==== ФИНАЛЬНЫЕ СООБЩЕНИЯ ====")
    for m in result["messages"][-6:]:
        role = type(m).__name__.replace("Message", "")
        print(f"[{role}] {getattr(m,'content','')[:300]}")

    print("\nФайлы в рабочей папке:")
    for p in sorted(WORKDIR.glob("*")):
        print(" -", p.name)

Результат работы системы:

  1. Исследователь находит 3-5 ссылок по теме и создаёт сводку фактов о влиянии погоды на кофейный бизнес

  2. Дата-инженер структурирует данные в CSV (source, insight) и загружает в SQLite таблицу sales_insights

  3. Редактор создаёт профессиональный Markdown-отчёт со всеми выводами и рекомендациями

В рабочей папке появляются файлы:

  • dataset.csv — структурированные данные исследования

  • data.sqlite — база данных с таблицей sales_insights

  • report.md — итоговый отчёт в Markdown формате

Ключевые преимущества архитектуры

Специализация агентов:

  • Каждый агент оптимизирован под свои задачи (разные temperature, prompt, инструменты)

  • Чёткое разделение ответственности снижает сложность и повышает качество

Персистентная память:

  • Каждый агент имеет свой thread_id для независимой памяти

  • Оркестратор передаёт контекст через единое состояние

  • MemorySaver обеспечивает сохранение истории между вызовами

Безопасность sandbox:

  • Все операции ограничены WORKDIR-директорией

  • Проверка путей файлов против выхода за пределы рабочей папки

  • Ограничения на объём скачиваемого контента (50KB)

Масштабируемость:

  • Легко добавить новых агентов (например, агент валидации или агент уведомлений)

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

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

Эволюция поведения

Что особенно впечатляет в этой системе — эмерджентное поведение. Агенты самостоятельно принимают интеллектуальные решения:

  • Исследователь может решить, стоит ли загружать контент конкретной страницы

  • Дата-инженер самостоятельно выбирает структуру таблицы и названия колонок

  • Редактор создаёт уникальный стиль отчёта, адаптированный под тему

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

Практическое применение

Эта архитектура подходит для решения сложных задач, требующих:

  • Исследования и анализа — market research, competitive analysis, аналитические отчёты

  • Обработки данных — ETL процессы, создание отчётности, автоматизация аналитики

  • Документооборота — автоматическое создание технической документации, сводок, презентаций

  • Бизнес-аналитики — сбор данных → анализ → презентация результатов в удобном формате

Дальнейшие улучшения

Базовую систему можно развивать в нескольких направлениях:

Интеграция с внешними системами:

  • Замена простого web_search на полноценные API (Google Search, Bing)

  • Подключение к корпоративным базам данных (PostgreSQL, MongoDB)

  • Интеграция с облачными хранилищами (S3, Google Drive)

Улучшение качества:

  • Добавление агента-валидатора для проверки качества данных

  • Система retry для обработки временных сбоев

  • A/B тестирование разных промптов и стратегий

Продуктивность:

  • Параллельное выполнение независимых задач

  • Кэширование результатов для повторных запросов

  • Конфигурируемые pipeline для разных типов исследований

Итоги: от болтунов к цифровым помощникам

Сегодня мы совершили революционный скачок в развитии наших агентов. Помните, с чего мы начинали? Наши «умные болтуны» могли только анализировать и отвечать, но были совершенно беспомощны в реальном мире.

Теперь ваши агенты умеют:

  • Работать с инструментами — файловая система, базы данных, веб-поиск, API

  • Самостоятельно принимать решения — какой инструмент использовать, в какой последовательности

  • Выстраивать цепочки действий — от исследования до создания финального отчёта

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

  • Работать безопасно — в изолированной среде с контролируемым доступом

Мы освоили три ключевых паттерна создания функциональных агентов:

  1. Ручные графы с инструментами — максимальный контроль, кастомная логика

  2. ReAct агенты — простота использования, встроенная логика рассуждений

  3. Мультиагентные системы — специализация и оркестрация для сложных задач

Что дальше? От демо к продакшену

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

На момент публикации этой статьи (27 сентября 2025 года) в моем телеграм‑канале проходит опрос на тему «Какой проект на базе ИИ‑агентов мы будем реализовывать». Если хотите повлиять на выбор идеи для большого проекта, который я подробно разберу в следующей части, приглашаю присоединиться к каналу. Атмосфера там уютная, а сообщество уже насчитывает почти 5000 единомышленников.

Увидимся в финальной части, где теория окончательно превратится в практику!

Amvera Cloud – облако для простого запуска проектов со встроенным CI/CD (деплой идёт через Git или загрузку файлов в интерфейсе), https-доменами, и встроенным проксированием до ведущих LLM и управляемым инференсом моделей LLaMA и GPT. Вам не нужно думать о настройке NGINX, виртуальных машин и другой инфраструктуры. Зарегистрируйтесь и получите 111 рублей на тест. 

Комментарии (0)