Средний

5 мин

8.1K

Введение

В предыдущей статье был рассмотрен способ интеграции ИИ-ассистентов в процесс торговли на бирже. Чтобы углубиться в проектирование ИИ‑приложений, я решил доработать проект. В данной части опишу, как реализовал механизм GuardRails, а именно технику Human‑In‑The‑Loop.

Не давать агентам делать лишнего - GuardRails

Дословно GuardRails переводится как «Ограждения». Это набор техник и инструментов, которые контролируют входные запросы и выходные ответы LLM. Проще говоря, это механизмы управления рисками при работе с ИИ‑агентами.

Основное с чем борется GuardRails это:

1. Маскирование ПД при выполнении инструментов;

2. Отсечение запросов, которые не относятся к функционалу агента;

3. Детектирование и борьба с промпт инъекциями;

4. Борьба с галлюцинациями.

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

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, где ключ - тикер, значение - количество лотов. Сам элемент предоставляет возможность пользователю изменять количество лотов.) и возвращается следующим сообщением в граф.

Демонстрация

Предположим, мы неправильно ввели количество акций. На данном этапе можно редактировать значение и подтвердить ордер:

Вывод

В этой части мы перевели агента из режима «автономного исполнителя» в режим «со-пилотирования», реализовав механизм Human-in-the-Loop на базе interrupt из LangGraph.

Что удалось реализовать:

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

• Интерактивный UI на Chainlit: встроенный элемент позволяет просматривать и менять параметры позиций прямо в чате, а стриминг обеспечивает прозрачность промежуточных шагов агента.

• Архитектурный переход на LangGraph: отказ от линейных Chain‑пайплайнов в пользу управляемого графа дал возможность внедрить прерывания, циклы и явное управление состоянием.

Что дальше:

• Интеграция с Chainlit Data Layer для надёжного хранения истории, восстановления сессий и отладки сценариев.

• Проработка логики подключения к удалённым MCP‑серверам.

Делитесь своим мнением и идеями усовершенствования!