Мой личный джуниор. Часть 2. Даём агенту немножечко мозгов

Привет, Хабр! Меня зовут Владимир и это продолжение статьи про разработку локального кодер-агента на LangGraph с доступом к MCP-инструментам.

В первой части мы создали инфраструктуру, написали простейшего агента и дали ему доступ к MCP-серверам для работы с файлами, кодом Git и документацией. В этой половинке мы добавим агенту немного мозгов. Но для начала немного истории, которая появилась при написании этой статьи.

if message.response_metadata.get('finish_reason') == 'tool_calls' and not message.tool_calls:    return {'messages': [HumanMessage(load_prompt_from_langfuse('mcp_agent_error_tool_call_prompt'))]}

А теперь вернёмся к разработке агента.

Архитектура

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

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

Наверное такая архитектура подойдёт для большой модели. С моей же (GigaChat на то время) началась магия — простейшее задание “Напиши функцию…” модель разбивала на 4-5 шагов, сама же потом путалась в плане. Иногда генерила код, а следующим шагом писала в файл заново нагенерированный код. Плюс вскрывшиеся вышеописанные проблемы с зацикливанием. В итоге было решено перейти на “полуавтономный” вариант — результат каждого шага требует вмешательства пользователя. Выглядеть это стало как-то так:

На основании этой архитектуры и будем строить агент.

Реализация агента

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

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

• узлы оценщики

• узел инъекции шага плана

• узел сжатия контекста

• узел финализации

Ну и Стейт также надо доработать. С него и начнём

AgentState

class AgentState(MessagesState):    user_request: str    user_input: str    plan: list[dict]    current_step: int    step_iteration: int    history: list[str]    phase: Literal['planning', 'executing', 'done']    is_approved: bool    trace_id: str

В стейт мы добавили:

• Запрос пользователя — чтобы всегда был под рукой

• Поле для ввода пользователя — поле для промежуточного общения

• Место для плана от планировщика

• Индекс текущего шага плана

• Счетчик неудачных попыток для защиты от зацикливания

• История выполнения (сжатая)

• Фаза работы агента — планирует, выполняет, закончил

• Подтверждение, что шаг принят пользователем.

• Идентификатор сессии (для организации памяти)

Для счетчика попыток в настройки агента также надо добавить поле — порог попыток

class ServiceConfig(BaseModel):    # старый код    max_step_iterations: int = 30

Перейдём к узлам

Узел Планировщик

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

from langchain.chat_models import BaseChatModelfrom langchain_core.output_parsers import PydanticOutputParserclass PlanerNode:    def __init__(            self,            llm: BaseChatModel,            prompt_name: str = 'mcp_agent_planer_prompt',            prompt_label: str = 'production'    ):        self.parser = PydanticOutputParser(pydantic_object=Workflow)        self.llm = llm        self.prompt = load_prompt_from_langfuse(            prompt_name=prompt_name, prompt_label=prompt_label        ).format(format_instructions=self.parser.get_format_instructions())

На выходе узла мы хотим получить список пунктов. Для достижения результата ма используем PydanticOutputParser — данный класс позволяет на основе Pydantic модели сформировать дополнение к промпту, которое подскажет модели, что мы хотим на выходе. А метод PydanticOutputParser.parse вернёт нам заданный Pydantic-объект. В итоге вместо ответа вида “Хорошо, вот что я могу предложить в качестве плана на твой запрос…” мы получим просто список пунктов. Для этого опишем простую модель:

from typing import Annotatedfrom pydantic import BaseModel, Fieldclass Workflow(BaseModel):    plan: Annotated[list[str], Field(description='Список шагов для выполнения задачи')]

Для загрузки промптов из LangFuse используется следующая функция:

def load_prompt_from_langfuse(prompt_name: str, prompt_label: str = 'production') -> str:    try:        return (            settings.langfuse.client            .get_prompt(prompt_name, label=prompt_label)            .get_langchain_prompt())    except Exception as e:        logger.error(f'Не удалось загрузить промпт "{prompt_name}" из Langfuse: {e}')        raise RuntimeError(f'Не удалось загрузить промпт "{prompt_name}" из Langfuse')

Промпт получаем по имени и тегу (на случай проверки разных вариантов) и следующим шагом преобразуем его в строку, совместимую с Langchain.PromptTemplate.

Дальше код самого узла:

class PlanerNode:    # предыдущий код    async def node(self, state: AgentState) -> dict:        user_input = state.get('user_input', '')        messages = [SystemMessage(content=self.prompt)] + state['messages']        if user_input:            messages += [HumanMessage(content=user_input)]        response = await self.llm.ainvoke(messages)        plan = self.parser.parse(response.content).plan        message = 'Проверьте план действий. Подтвердите план словом **"Продолжить"** либо внесите корректировки'        for i, item in enumerate(plan, start=1):            message += f'\n* Шаг {i}: {item}'        return {            'messages': [                HumanMessage(content=user_input), AIMessage(content=message)            ] if user_input else [AIMessage(content=message)],            'plan': plan, 'current_step': 0, 'phase': 'planning', 'is_approved': False,        }

Основа — в промпте. В нём надо добиться, чтобы LLM разбила запрос пользователя на отдельные операции (типа написать код, сохранить файл, добавить докстринг и т.п.) и выдать их в виде списка. Конкретные промпты приводить тут не буду, но к репозиторию приложу свой дамп с LangFuse.

Дальше, с помощью PydanticOutputParser, парсим ответ модели и формируем ответ для пользователя — просто перечисляем шаги сформированного плана.

Поясню за user_input — это поле специально отведено под уточняющие сообщения пользователя (как — в разделе про интерфейс). Если в поле что-то есть, значит пользователь просит что-то уточнить, и мы добавляем это в историю сообщений.

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

Инъектор

Данный узел — способ сообщить текущий шаг агенту. Одно время в узле чистил поле state['messages'] и заполнял суммаризацией. Но на модели Qwen3.6 от Nvidia суммаризация только всё портила, поэтому оставил только добавление шага плана:

class StepInjectorNode:    def node(self, state: AgentState) -> dict:        step_text = state['plan'][state['current_step']]        return {'messages': HumanMessage(f'ТЕКУЩИЙ ШАГ, КОТОРЫЙ НУЖНО ВЫПОЛНИТЬ: {step_text}'),}

Узел-агент

Конструктор узла стандартный (для моих узлов):

class AgentNode:    def __init__(            self,            llm: BaseChatModel,            prompt_name: str = 'mcp_agent_prompt',            prompt_label: str = 'production'):        self.llm = llm        self.prompt = load_prompt_from_langfuse(prompt_name=prompt_name, prompt_label=prompt_label)

Дальше сам узел. И начинается он с проверки зацикливания модели:

from langchain_core.messages import AIMessage, RemoveMessagefrom langgraph.graph.message import REMOVE_ALL_MESSAGESclass AgentNode:    # предыдущий код    async def node(self, state: AgentState) -> dict:        if state.get('step_iteration', 0) > settings.service.max_step_iterations:            msg = 'КРИТИЧЕСКАЯ ОШИБКА! Агент зациклился и был сброшен в начальное состояние!'            logger.error(msg)            return {                'messages': [RemoveMessage(id=REMOVE_ALL_MESSAGES), AIMessage(msg)],                'history': [], 'step_iteration': 0, 'current_step': 0,}

В случае превышения порога повторов чистим историю сообщений и оставляем одно единственное сообщение об ошибке.

Очистить историю нельзя просто передав [] — она объявлена как messages: Annotated[list[AnyMessage], add_messages], и пустой список просто добавится к истории. Для очистки надо использовать специальный тип сообщения RemoveMessage. В поле id можно передать идентификатор конкретного сообщения либо специальную константу langgraph.graph.message.REMOVE_ALL_MESSAGES для очистки всей истории.

Исполнительный код узла чем-то напоминает планер — формируем системный промпт, добавляем историю, сообщение пользователя (при наличии):

class AgentNode:    # предыдущий код    async def node(self, state: AgentState) -> dict:        # предыдущий код        messages = [SystemMessage(self.prompt)] + state['messages']        if state['user_input']:            messages += [HumanMessage(state['user_input'])]        response = await self.llm.ainvoke(messages)        return {            'messages': [HumanMessage(state['user_request']), response] if state['user_input'] else [response],            'step_iteration': state.get('step_iteration', 0) + 1, 'phase': 'executing', 'is_approved': False, }

Ответ никак не форматирую — квену оказалось достаточно фразы в промпте “После выполнения текущего шага сообщи, что шаг выполнен, и укажи результат”. В случае чего — можно попросить уточнить ответ и он это сделает. Возвращает узел, кроме сообщений, инкремент шага, фазу “Исполнение” и сброшенный флаг подтверждения.

Узлы Оценщики

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

class BaseSolver:    @staticmethod    async def node(state: AgentState) -> dict:        if 'продолжить' in state['user_input'].lower():            return {'is_approved': True, 'user_input': ''}        return {'is_approved': False}class PlanSolverNode(BaseSolver):    passclass AgentSolverNode(BaseSolver):    pass

Узел Суммаризатор

Данный узел изначально использовался для уменьшения длины контекста. Но, как говорится, “Что русскому хорошо — немцу смерть”. В нашем случае с Qwen, читая суммаризованные сообщения модель начала давать ответ в таком-же формате. По-этому узел есть, но толку от него в данный момент нет — используется только при подготовке финального ответа.

class ContextCompressorNode:    @staticmethod    def _extract_text(content) -> str:        if isinstance(content, str):            return content.strip()        if isinstance(content, list):            parts = []            for block in content:                if isinstance(block, dict) and block.get('type') == 'text':                    parts.append(block.get('text', ''))                elif isinstance(block, str):                    parts.append(block)            return '\n'.join(p for p in parts if p).strip()        return str(content).strip()

class ContextCompressorNode:    def _build_steps_summary(self, messages: list) -> str:        tool_index = {            m.tool_call_id: m            for m in messages            if getattr(m, 'type', None) == 'tool' and getattr(m, 'tool_call_id', None)        }        lines = []        for msg in messages:            if getattr(msg, 'type', None) != 'ai':                continue            text = self._extract_text(msg.content)            if text:                lines.append(f'[ОТВЕТ АГЕНТА]: {text}')                continue            tool_calls = getattr(msg, 'tool_calls', None) or []            for tc in tool_calls:                tool_msg = tool_index.get(tc.get('id'))                if tool_msg is not None:                    result = self._extract_text(tool_msg.content)                    lines.append(f'[ИНСТРУМЕНТ] "{tc.get("name", "unknown")}". Результат: {result}')        return '\n\n'.join(lines) if lines else '(история выполнения пуста)'

class ContextCompressorNode:    def node(self, state: AgentState) -> dict:        steps_summary = self._build_steps_summary(state['messages'])        step = state['current_step']        history = state.get('history', []) + [            f"[РЕЗУЛЬТАТ ВЫПОЛНЕНИЯ ШАГА {step + 1}] {state['plan'][step]}.\n\n{steps_summary}"]        return {'history': history, 'current_step': step + 1}

Узел Финализатор

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

class FinalizerNode:    def __init__(            self,            llm: BaseChatModel,            prompt_name: str = 'mcp_agent_finalize_prompt',            prompt_label: str = 'production'):        self.llm = llm        self.prompt = load_prompt_from_langfuse(prompt_name=prompt_name, prompt_label=prompt_label)    async def node(self, state: AgentState) -> dict:        plan = '\n'.join(f"{i}. {s}" for i, s in enumerate(state.get('plan', []), start=1))        steps_summary = '\n\n'.join(state.get('history', []))        msg = (            f'План: \n{plan}\n\n'            f'Результаты выполнения:\n{steps_summary}\n\n'            f'Сформируй итоговый ответ для пользователя.')        response = await self.llm.ainvoke([SystemMessage(content=self.prompt), HumanMessage(content=msg),])        message = f'Выполнение задачи завершено:\n\n{response.content}'        return {'messages': [message], 'phase': 'done', 'is_approved': False}

С узлами закончили. Но перед тем как перейти к сборке графа, рассмотрим две сущности LangGraph, которые позволят реализовать общение с пользователем: чекпоинтер и прерывания.

Чекпоинтер

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

config = {'configurable': {'thread_id': 'my-thread'}}graph.invoke(inputs, config)

Чекпоинтер можно сделать локально, а можно на базе Postgres (нужна библиотека langgraph-checkpoint-postgres). Для прода LangGraph рекомендует именно Postgres вариант, но нам хватит и langgraph.checkpoint.memory.InMemorySaver

Использовать чекпоинтер достаточно просто — создаём экземпляр и подключаем его к графу на этапе компиляции.

checkpointer = InMemorySaver()graph = workflow.compile(checkpointer=checkpointer)

Прерывания

Прерывания нужны для прерывания работы графа

Могут быть до, внутри и после узла. Первый и третий тип задаются на этапе компиляции графа, а второй реализуется в теле узла с помощью оператора langgraph.types.interrupt.

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

graph = workflow.compile(    checkpointer=checkpointer,    interrupt_before=['before_node'],    interrupt_after=['after_node'],)

Для второго типа реализуем что-то подобное:

def ask_user(state: State):    answer = interrupt('Как вас зовут?')  # Граф останавливается здесь и ждет ответа    return {'user_answer': answer}  # Когда пользователь ответит, выполнение продолжится

В любом случае, для возобновления работы графа надо передать в граф специальное сообщение типа langgraph.types.Command с параметром update для обновления стейта или resume для возврата значения. Без этого граф начнет своё выполнение заново.

Важное уточнение: не понял, баг это или фича, но если после узла идёт условное ребро, то сначала выполнится оно, а уже потом будет останов. Таким образом можно намертво зациклить свой граф, если ребро решает задачу “Повтор/Продолжить”

Разрыв лонгрида

У нас готовы все компоненты агента:

• Планировщик — разбивает задачу на шаги

• Инжектор — сообщает агенту текущий шаг

• Исполнитель — выполняет шаги

• Оценщики — проверяют подтверждение пользователя

• Суммаризатор — сжимает историю выполнения

• Финализатор — подводит итоги

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

В следующей части мы превратим этот набор узлов в работающий граф.