Привет, Хабр! Меня зовут Владимир и это продолжение статьи про разработку локального кодер-агента.
В первой части мы создали инфраструктуру: подняли контейнер с моделью, настроили Langfuse для трассировки и написали простейшего агента с доступом к MCP-инструментам.
Во второй части мы добавили агенту мозги: планировщик, оценщики, защиту от зацикливания и суммаризацию контекста.
Сегодня мы превратим этот набор в работающий граф. Добавим чекпоинтеры, прерывания и доработаем интерфейс под обновлённую архитектуру.
Сборка графа
Доработку графа начнем с конструктора - в него необходимо добавить чекпоинтер. Дополнительно, для упорядочивания трассировок добавим 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')Инициализация графа не менялась - инициализируем инструменты, биндим на LLM и собираем граф (код далее)
class MCPAgent:
# Предыдущий код
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()Перейдём к условным рёбрам. Начнём с рёбер маршрутиризации подтверждения плана и выполнения шага. В них проверяем флаг is_approved. Если подтверждение есть, то переходим к исполнению плана или суммаризации соответственно. Если нет, то возвращаем на доработку.
class MCPAgent:
# Предыдущий код
@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'Теперь маршрутизатор перехода по шагам. В нём мы должны проверить, что план ещё в процессе выполнения. Проверяем сравнением номера шага и длины плана
class MCPAgent:
# Предыдущий код
@staticmethod
def next_step_router(state: AgentState) -> Literal['injector', 'finalizer']:
if state['current_step'] < len(state['plan']):
return 'injector'
return 'finalizer'Соберём граф. Инициализируем узлы и соединим их согласно схеме из второй части статьи:
class MCPAgent:
# Предыдущий код
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В компилятор передаём два останова - после формирования плана и до "решателя" узла-агента.
Для того, чтобы граф работал с остановами, нам надо два метода - один для первоначального запуска графа, второй - для продолжения общения. Начнем с запуска:
class MCPAgent:
# Предыдущий код
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')Служебные методы разберу позже, пока сам алгоритм. Для начала проверяем, что не забыли инициализировать граф. Далее формируем trace_id для Langfuse средствами Langfuse. Без передачи trace_id Langfuse будет создавать отдельные трейсы для каждого возврата в граф. Ну и вызываем служебный метод вызова LLM.
Теперь метод возобновления общения:
class MCPAgent:
# Предыдущий код
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')Основные отличия - номер трейса в Langfuse мы берём из стейта и наше сообщение пользователя оборачиваем в Command. Переходим к служебным методам.
Первый метод проверяет наличие агента:
class MCPAgent:
# Предыдущий код
def _check_graph_available(self):
if self.graph is None:
raise RuntimeError(
'Агент не инициализирован. Запустите `initialize()`.')Второй нужен для извлечения trace_id из сохраненного стейта:
class MCPAgent:
# Предыдущий код
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)Про thread_id писал в части про чекпоинтер. Он нужен для того, чтобы граф знал, стейт какой именно сессии надо извлекать из памяти. Формируется на стороне веб-интерфейса. Теперь метод вызова LLM:
class MCPAgent:
# Предыдущий код
async def _ainvoke_with_tracing(
self, data: dict[str, Any] | Command, request_id: str, trace_id: str, span_name: str
) -> dict[str, Any]:
with settings.langfuse.client.start_as_current_observation(
as_type='span',
name=span_name,
trace_context={'trace_id': trace_id},
) as span:
result = await self.graph.ainvoke(data, config=self._create_config(request_id))
return result['messages']Первым делом создаём кастомное наблюдение типа спан и передаём в него параметром наш trace_id. Langfuse сам сгруппирует все наблюдения по trace_id. Далее вызываем метод ainvoke графа, передав в него конфиг:
class MCPAgent:
# Предыдущий код
def _create_config(self, request_id: str) -> dict[str, Any]:
return {
'callbacks': [self.lf_handler],
'metadata': {
'langfuse_session_id': f'docker_session_{self.init_time}',
},
'configurable': {'thread_id': request_id}
}В конфиге мы прописываем Langfuse CallbackHandler для организации наблюдения, в метаданных передаём параметр langfuse_session_id, который в дальнейшем можно использовать для фильтрации трейсов и thread_id для сохранения стейта в чекпоинтере.
Граф готов. Осталось доработать Gradio интерфейс под новые функции и можно релизить)
Интерфейс пользователя
При изменении архитектуры агента (добавление подтверждения пользователем) я пошел на один компромис - подтверждение простым словом “Продолжить”. И чтобы каждый раз его не писать, я решил добавить в интерфейс кнопку “Продолжить”. Для этого пришлось отказаться от gr.ChatInterface и переписать интерфейс на gr.Chatbot:
class MCPCodingAgentApp:
def build_interface(self):
with gr.Blocks(title='MCP Coding Agent', fill_height=True) as self.demo:
gr.Markdown('# MCP Coding Agent')
gr.Markdown('Помощник разработчика с доступом к файлам, Git и документации')
chatbot = gr.Chatbot(label='Чат с агентом', height=700)
request_id_state = gr.State('')
with gr.Row():
msg = gr.Textbox(
label='Ваше сообщение',
placeholder='Введите сообщение или нажмите "Продолжить"',
scale=8,
container=False
)
submit_btn = gr.Button('Отправить', variant='primary')
continue_btn = gr.Button('▶ Продолжить', variant='secondary')
submit_btn.click(
fn=self._respond,
inputs=[msg, chatbot, request_id_state],
outputs=[chatbot, msg, request_id_state])
msg.submit(
fn=self._respond,
inputs=[msg, chatbot, request_id_state],
outputs=[chatbot, msg, request_id_state])
continue_btn.click(
fn=self._continue,
inputs=[chatbot, request_id_state],
outputs=[chatbot, request_id_state])После поля чата создаём строку с полем для ввода сообщения и двумя кнопками - “Отправить” и “Продолжить”. Далее назначаем функции-обработчики для наших элементов (для поля ввода тоже, чтобы была отправка по нажатию Enter). Параметры inputs и outputs связывают входы-выходы функции-обработчика с объектами Gradio.
Объект gr.State('') нужен для хранения request_id в рамках сессии.
Так как чат у нас теперь самодельный, то и управлять всем в обработчиках мы должны руками:
class MCPCodingAgentApp:
# Предыдущий код
async def _respond(self, message: str, history: list, request_id: str):
if not request_id:
request_id = str(uuid.uuid4())
if not message or not message.strip():
return history, '', request_id
phase = self.agent.get_phase(request_id)
if phase is None or phase == 'done':
result = await self.agent.run(user_messages=message, request_id=request_id)
else:
result = await self.agent.resume(user_messages=message, request_id=request_id)
history.append({'role': 'user', 'content': message})
agent_response = result[-1].content if result else 'Нет ответа'
history.append({'role': 'assistant', 'content': agent_response})
return history, '', request_idВ обработчике кнопки “Отправить” и нажатия Enter мы должны предусмотреть, что пользователь решил напечатать слово “Продолжить” вместо нажатия отдельной кнопки. Для этого мы анализируем фазу выполнения агента и, в зависимости от неё, вызываем соответствующий метод агента. Дополнительно создаём первичный request_id, защищаемся от пустого сообщения (как в запросе, так и в ответе) и наполняем историю.
Обработчик продолжения общения немного покороче:
class MCPCodingAgentApp:
# Предыдущий код
async def _continue(self, history: list, request_id: str):
phase = self.agent.get_phase(request_id)
if phase is None or phase == 'done':
history.append({
'role': 'assistant',
'content': 'Нет активных задач для продолжения. Задайте новый вопрос.'
})
return history, request_id
result = await self.agent.resume(user_messages='Продолжить', request_id=request_id)
history.append({'role': 'user', 'content': '▶ Продолжить'})
agent_response = result[-1].content if result else 'Нет ответа'
history.append({'role': 'assistant', 'content': agent_response})
return history, request_idУбеждаемся по фазе, что агент в процессе работы, после чего направляем ему сообщение “Продолжить”, иначе сообщаем пользователю, что вы не в цикле. Также сохраняем историю.
Код инициализации и запуска остаётся без изменений.
С кодом всё.
Проверка работоспособности
Запускаем make up.
Теперь можно расслабиться и откинуться на спинку кресла (есть тут те, кто ставил Windows98?😄)
Ждём закачки образов, модели. Для скачивания модели можно использовать uvx (ИМХО в разы быстрее):
HF_TOKEN=<ТОКЕН> HF_HOME=<ПАПКА ДЛЯ МОДЕЛИ> uvx hf download nvidia/Qwen3.6-35B-A3B-NVFP4В HF_HOME нужно указать туже папку, которую указываем как том для vLLM.
Простой вопрос
Проверку начал с задания попроще, но в несколько шагов:
Напиши функцию проверки на четность. Сохрани её в файл и сделай коммит
Приемлемо. Глянем трейс задачи
Всё как надо - два шага выполнения, три вызова инструментов (write file, git add, git commit).
Вопрос посложнее
Тут нашел в интернете какую-то задачу на знание FastAPI, помеченную как средней сложности
Создай асинхронный dependency get_db, который открывает транзакцию.
Реализуй модель для эндпоинта User с id:uuid и nickname: str
Реализуй эндпоинт GET /users/{user_id}/posts, который возвращает пользователя и его посты в одном запросе к БД (без N+1 проблемы).
Сохрани результаты в отдельные файлы
Сделай коммит изменений
На запрос получил следующий план действий:
Глянем схему трейса
Красота) При выполнении агент даже вызывал инструменты для просмотра содержимого директории в поисках файла model.py. Ну и результат работы агента в рабочей директории:
Итоги
Финальным экспериментом я решил проверить, сможет ли агент написать документацию для самого себя. Дал ему доступ к собственной папке и попросил:
Проанализируй файлы в рабочей папке. Необходимо составить файл readme.md. Файл должен содержать краткое описание проекта и способ запуска
Получилось неожиданно хорошо - агент описал структуру проекта, перечислил основные компоненты и добавил команды для запуска через Docker. Результат можно посмотреть в репозитории: ссылка на README.md. Ручного в файле только про параметр WS.
Код проекта доступен тут
Первая часть с базой и простым ReAct агентом с MCP инструментами тут
Вторая часть в которой показана реализация узлов тут