Введение
В предыдущей статье был рассмотрен способ интеграции ИИ-ассистентов в процесс торговли на бирже. Чтобы углубиться в проектирование ИИ‑приложений, я решил доработать проект. В данной части опишу, как реализовал механизм GuardRails, а именно технику Human‑In‑The‑Loop.
Примечание
На скриншотах вы встретите логотипы и фирменные цвета Т‑Банка. Это сделано намеренно, так как проект строится на официальном T-Invest API и их SDK. Все материалы приведены исключительно в демонстрационных целях. Статья не является рекламой, не содержит инвестиционных рекомендаций и не подразумевает официального партнёрства с банком.
Не давать агентам делать лишнего - GuardRails
Дословно GuardRails переводится как «Ограждения». Это набор техник и инструментов, которые контролируют входные запросы и выходные ответы LLM. Проще говоря, это механизмы управления рисками при работе с ИИ‑агентами.
Основное с чем борется GuardRails это:
Маскирование ПД при выполнении инструментов;
Отсечение запросов, которые не относятся к функционалу агента;
Детектирование и борьба с промпт инъекциями;
Борьба с галлюцинациями.
На текущем этапе самым важным действием моего агента остаётся покупка и продажа акций. При проведении тестов было выявлено, что при выполнении этих задачах модель редко, но ошибается. Помимо этого, сохраняя парадигму, что агент является помощником, а итоговое решение принимает пользователь, целесообразно сделать подтверждение действий агента.
Human‑In‑The‑Loop (HITL)
Для реализации HITL был совершен полный переход на проектирование агента в фреймворке LangGraph. В отличие от линейных Chain-пайплайнов, LangGraph позволяет строить ориентированные графы с циклами, явно управлять состоянием и, что важно для моей задачи, безопасно останавливать выполнение в любой точке. Ключевой механизм, который делает HITL возможным в LangGraph - это interrupt. По своей сути, это контрольная точка, в которой выполнение графа приостанавливается, текущее состояние сохраняется в Checkpointer, а управление передаётся внешнему приложению (веб-интерфейсу, Telegram-боту, CLI и т.д.). LangGraph автоматически вставляет хук в граф. Как только поток управления доходит до этого узла, выполнение прерывается. Подробнее можно почитать в документации, а на форуме ознакомиться с примерами.
Реализация HITL. Вывод на веб-интерфейс Chainlit
Данная техника реализована в ноде исполнения инструментов:
# Импорты библиотек from langgraph.types import interrupt ... async def tool_node(state: MessagesState): tool_call = state['messages'][-1].tool_calls[0] name = tool_call['name'] arg_ = tool_call['args'] id_ = tool_call['id'] print(f'Вызов инструмента {name} с аргументами {arg_}') if (name == "buy") or (name == "sale"): t2f = await tool_by_name['figi_to_ticker'].ainvoke({"figi_list": list(arg_['order_dict'].keys())}) t2f = literal_eval(t2f[0]['text']) t2q = {k: arg_['order_dict'][v] for k, v in t2f.items() if v in arg_['order_dict']} resume_payload = interrupt(value=(t2q, 'Подтвердите действия')) if resume_payload['action'] == 'timeout': return {'messages': [ToolMessage( content="Пользователь не подтвердил действие или истекло время ожидания. Покупка отменена.", tool_call_id=id_ )]} if resume_payload['action'] == 'confirm': changed_orders = {} msg = '' for order_dict in resume_payload['data']: ticker = order_dict['ticker'] quantity = order_dict['quant'] figi = t2f[ticker] if arg_['order_dict'][figi] != quantity: changed_orders[figi] = quantity arg_['order_dict'][figi] = quantity if changed_orders: msg = 'Пользователем были отредактированы следующие позиции:\n' for figi, quantity in changed_orders.items(): msg += f'FIGI {figi} - новое количество лотов: {quantity}\n' msg += 'В связи с этим были осуществлены операции:\n' result = await tool_by_name[name].ainvoke(arg_) return {'messages': [ToolMessage(content=msg+result[0]['text'], tool_call_id=id_)]} if resume_payload['action'] == 'cancel': return {'messages': [ToolMessage(content="Покупка отменена пользователем", tool_call_id=id_)]} result = await tool_by_name[name].ainvoke(arg_) return {'messages': [ToolMessage(content=result[0]['text'], tool_call_id=id_)]} ...
interrupt приостанавливает дальнейшее выполнение графа и ждет результата пользователя.
В функции, обернутой декоратором @cl.on_message, происходит следующее:
@cl.on_message async def main(message: cl.Message): config = {"configurable": {"thread_id": cl.context.session.id}} agent = cl.user_session.get("agent") input_message = {'messages': [HumanMessage(content=message.content)]} while True: interrupted = False mes = cl.Message(content='') async with cl.Step(name="Размышление", type="agent") as reasoning_step: stream = agent.astream( input_message, stream_mode=['messages', 'updates'], #Стримим и токены и состояния config=RunnableConfig(callbacks=[ChainlitToolTracer()], **config) ) async for mode, chunk in stream: if mode == "messages": # Стримим токены пользователю msg, _ = chunk if hasattr(msg, 'additional_kwargs') and 'reasoning_content' in msg.additional_kwargs: await reasoning_step.stream_token(str(msg.additional_kwargs['reasoning_content'])) if (hasattr(msg, 'content') and msg.content and not isinstance(msg, ToolMessage) and 'reasoning_content' not in getattr(msg, 'additional_kwargs', {})): await mes.stream_token(msg.content) if (hasattr(msg, 'content') and msg.content and isinstance(msg, ToolMessage)): mes.content = '' await mes.update() elif mode == "updates": if isinstance(chunk, dict) and INTERRUPT in chunk: #Ловим состояние await mes.send() await mes.update() t2q, msg = chunk[INTERRUPT][0].value # Создаем кастомный элемент HITL elements = create_table(t2q) # Отправляем пользователю approve_mes = await cl.AskElementMessage(content=msg, element=elements[0]).send() # Обрабатываем результат if approve_mes is None: # Пользователь не ответил в течение отведенного времени input_message = Command(resume={'action': 'timeout'}) else: # Отправляем решение пользователя input_message = Command(resume=approve_mes) interrupted = True await mes.remove() if not interrupted: await mes.send() reasoning_step.end = utc_now() break
В специальный объект Command записываются данные из элемента (json, где ключ - тикер, значение - количество лотов. Сам элемент предоставляет возможность пользователю изменять количество лотов.) и возвращается следующим сообщением в граф.
Насчет reasoning_content
Тут описал как сделать вывод размышлений reasoning моделей, которые возвращают ответы по стандарту OpenAI.
Демонстрация
Предположим, мы неправильно ввели количество акций. На данном этапе можно редактировать значение и подтвердить ордер:
Вывод
В этой части мы перевели агента из режима «автономного исполнителя» в режим «со-пилотирования», реализовав механизм Human-in-the-Loop на базе interrupt из LangGraph.
Что удалось реализовать:
Полноценный HITL‑контур: граф безопасно приостанавливается перед ордерами, сохраняет состояние в чекпоинтере и корректно обрабатывает пользовательские ответы.
Интерактивный UI на Chainlit: встроенный элемент позволяет просматривать и менять параметры позиций прямо в чате, а стриминг обеспечивает прозрачность промежуточных шагов агента.
Архитектурный переход на LangGraph: отказ от линейных Chain‑пайплайнов в пользу управляемого графа дал возможность внедрить прерывания, циклы и явное управление состоянием.
Что дальше:
Интеграция с Chainlit Data Layer для надёжного хранения истории, восстановления сессий и отладки сценариев.
Проработка логики подключения к удалённым MCP‑серверам.
Делитесь своим мнением и идеями усовершенствования!
