Что такое ИИ-агент с ожиданием: локальный граф на LangGraph без облака за 2 часа

Руководство является третьей частью серии. В первой части автор поднял контейнер с моделью, настроил Langfuse (инструмент трассировки, то есть записи каждого шага работы агента) и подключил MCP-инструменты (Model Context Protocol, протокол, через который агент вызывает внешние функции). Во второй добавил планировщик, оценщики и защиту от зацикливания. Теперь всё это собирается в работающий граф, где агент умеет останавливаться и ждать подтверждения от человека.

Что понадобится?

• Знакомство с первой и второй частями серии на Хабре (ссылки в оригинале)
• Python 3.10+ и установленные библиотеки: LangGraph, LangChain, Langfuse
• Локально развёрнутая языковая модель (контейнер из первой части)
• Настроенный Langfuse для трассировки
• MCP-инструменты, подключённые по протоколу из первой части
• Примерно 2 часа на сборку и отладку графа

Пошаговая инструкция

1. Добавьте чекпоинтер в конструктор агента.

Чекпоинтер (checkpointer) сохраняет состояние графа после каждого узла. Если агент остановился и ждёт ответа пользователя, состояние не теряется. Дополнительно формируется session_id на основе даты и времени, чтобы трассировки в Langfuse не смешивались между сессиями.

class MCPAgent:
    def __init__(self):
        self.llm_with_tools: ChatOpenAI | None = None
        self.tools: list[BaseTool] | None = None
        self.graph = None
        self.checkpointer = InMemorySaver()
        self.lf_handler = CallbackHandler(
            public_key=settings.langfuse.public_key
        )
        self.init_time = datetime.now().strftime('%Y%m%d_%H%M%S')

2. Инициализируйте граф: подключите инструменты и привяжите их к модели.

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

async def init_graph(self):
    try:
        self.tools = await init_tools()
    except Exception as e:
        logger.error(f'Ошибка инициализации инструментов: {e}')
        raise
    self.llm_with_tools = settings.llm.chat.llm.bind_tools(
        self.tools, parallel_tool_calls=False
    )
    self.graph = self._compile_graph()

3. Настройте условные рёбра графа.

Условные рёбра определяют, куда двигается агент после каждого узла. Два ключевых маршрутизатора проверяют флаг is_approved в состоянии.

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

@staticmethod
def need_adjust_plan_router(state: AgentState) -> Literal['injector', 'planer']:
    if state.get('is_approved', False):
        return 'injector'
    return 'planer'

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

@staticmethod
def need_modify_step_router(state: AgentState) -> Literal['agent_node', 'compressor']:
    if state.get('is_approved', False):
        return 'compressor'
    return 'agent_node'

4. Добавьте маршрутизатор перехода по шагам плана.

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

@staticmethod
def next_step_router(state: AgentState) -> Literal['injector', 'finalizer']:
    if state['current_step'] < len(state['plan']):
        return 'injector'
    return 'finalizer'

5. Соберите и скомпилируйте граф.

Здесь все узлы соединяются рёбрами. Два ключевых параметра компиляции: interrupt_after=['planer'] останавливает граф после формирования плана и ждёт подтверждения, interrupt_before=['agent_solver'] останавливает перед оценкой результата шага.

def _compile_graph(self):
    workflow = StateGraph(AgentState)
    workflow.add_node('agent_node', AgentNode(llm=self.llm_with_tools).node)
    workflow.add_node('compressor', ContextCompressorNode().node)
    workflow.add_node('finalizer', FinalizerNode(llm=settings.llm.chat.llm).node)
    workflow.add_node('planer', PlanerNode(llm=settings.llm.chat.llm).node)
    workflow.add_node('plan_solver', PlanSolverNode().node)
    workflow.add_node('agent_solver', AgentSolverNode().node)
    workflow.add_node('injector', StepInjectorNode().node)
    workflow.add_node('tools', ToolNode(self.tools))

    workflow.set_entry_point(key='planer')
    workflow.add_edge(start_key='planer', end_key='plan_solver')
    workflow.add_conditional_edges(source='plan_solver', path=self.need_adjust_plan_router)
    workflow.add_edge(start_key='injector', end_key='agent_node')
    workflow.add_conditional_edges(source='agent_node', path=self.agent_router)
    workflow.add_edge(start_key='tools', end_key='agent_node')
    workflow.add_conditional_edges(source='agent_solver', path=self.need_modify_step_router)
    workflow.add_conditional_edges(source='compressor', path=self.next_step_router)
    workflow.set_finish_point(key='finalizer')

    graph = workflow.compile(
        checkpointer=self.checkpointer,
        interrupt_after=['planer'],
        interrupt_before=['agent_solver'],
    )
    return graph

6. Реализуйте запуск и возобновление диалога.

Метод run запускает граф с нуля. Метод resume продолжает работу после паузы, оборачивая пользовательский ввод в Command и извлекая trace_id из сохранённого состояния, чтобы Langfuse не плодил отдельные трейсы на каждый возврат.

async def run(self, user_messages: str, request_id: str | None = None) -> dict[str, Any]:
    self._check_graph_available()
    trace_id = Langfuse.create_trace_id()
    return await self._ainvoke_with_tracing(
        data={'user_request': user_messages, 'user_input': user_messages, 'trace_id': trace_id},
        request_id=request_id, trace_id=trace_id, span_name='agent_run')

async def resume(self, user_messages: str, request_id: str | None = None) -> dict[str, Any]:
    self._check_graph_available()
    trace_id = self._get_trace_id(request_id)
    return await self._ainvoke_with_tracing(
        data=Command(update={'user_input': user_messages}),
        request_id=request_id, trace_id=trace_id, span_name='agent_resume')

7. Добавьте служебные методы: проверку инициализации и извлечение trace_id.

Первый метод не даёт вызвать агента до init_graph. Второй достаёт trace_id из сохранённого состояния по thread_id, чтобы вся сессия попала в один трейс.

def _check_graph_available(self):
    if self.graph is None:
        raise RuntimeError('Агент не инициализирован. Запустите `initialize()`.')

def _get_trace_id(self, request_id: str) -> str | None:
    return self.graph.get_state(
        {'configurable': {'thread_id': request_id}}
    ).values.get('trace_id', None)

Что такое ИИ-агент в контексте этого графа?

Когда говорят «что такое ИИ-агент», обычно подразумевают программу, которая сама выбирает инструменты и решает, когда остановиться. Но в реальной сборке агент без чекпоинтера и прерываний превращается в чёрный ящик: он либо молча делает всё подряд, либо теряет контекст при любой паузе. Именно механизм interrupt_after и interrupt_before превращает набор узлов в управляемого агента, который спрашивает, прежде чем действовать.

Кому это пригодится и что делать прямо сейчас?

Разработчику, который собирает бота или автоматизацию. Весь стек работает локально: модель в контейнере, Langfuse для трассировки, MCP для инструментов. Данные не уходят в облако, что критично для корпоративных проектов в РФ. Возьмите код из статьи и начните с InMemorySaver, затем замените на персистентный чекпоинтер под свою инфраструктуру.

Автору Дзена или контент-маркетологу. Сам код вам не нужен, но механика «агент делает, останавливается, спрашивает» это то, к чему идут все пользовательские ИИ-продукты. Понимание, что такое ИИ-агент на уровне архитектуры, поможет точнее писать о технологиях и не путать «чат-бот» с «агентом».

Предпринимателю. Если ваша команда обсуждает внедрение ИИ-агентов, покажите им эту серию статей как пример реализации без зависимости от OpenAI, Anthropic и других зарубежных API. Локальная модель плюс LangGraph плюс MCP равно полный контроль над данными и предсказуемыми затратами.

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

По материалам Habr AI