PEV (Планировщик-Исполнитель-Верификатор)

Наш агент планирования хорошо работает, когда путь ясен: он составляет план и следует ему. Но есть одно скрытое предположение…

 

Что происходит, когда что-то идёт не так? Если инструмент даёт сбой, API недоступен или поиск выдаёт некорректные результаты, стандартный планировщик просто передаёт ошибку дальше, что приводит к сбою или бессмысленному результату.

 

PEV (Планировщик-Исполнитель-Верификатор) ​​— это простое, но мощное усовершенствование шаблона планирования, добавляющее критически важный уровень контроля качества и самокоррекции.

 

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

 

Вот как это работает…

 

Нажмите Enter или щелкните, чтобы просмотреть изображение в полном размере.

 

 

 

 

 

PEV (Создано)

Фарид Хан

)

План: Агент «Планировщик» создает последовательность шагов, как и раньше.

Выполнить: Агент «Исполнитель» берет на себя следующую задачу. шагу плана и вызывает инструмент.

Проверка: Это новый этап. Агент «Проверяющий» анализирует результаты работы инструмента. Он проверяет правильность, релевантность и наличие ошибок.

Маршрутизация и итерация: на основе решения верификатора:

Если этап пройден успешно , агент переходит к следующему этапу плана.

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

Если план выполнен, он доводится до конца.

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

 

def flaky_web_search(query: str) -> str: 

    "" "Инструмент веб-поиска, который намеренно выдает ошибку при выполнении определенного запроса." ""

     console. print (f "--- ИНСТРУМЕНТ: Поиск '{query}'... ---" ) 

    if "количество сотрудников" in query. lower (): 

        console. print ( "--- ИНСТРУМЕНТ: [жирным красным]Имитация сбоя API![/жирным красным] ---" ) 

        return "Ошибка: Не удалось получить данные. Конечная точка API в данный момент недоступна." 

    else : 

        return search_tool.invoke({ "query" : query})

Теперь перейдём к сути паттерна PEV: узел verifier_node. Единственная задача этого узла — проанализировать результат работы предыдущего инструмента и решить, был ли это успех или неудача.

 

class VerificationResult ( BaseModel ): 

    """Схема выходных данных верификатора."""

     is_successful: bool = Field(description= "True, если выполнение инструмента было успешным и данные действительны." ) 

    reasoning: str = Field(description= "Обоснование решения о проверке." ) 

 

def verifier_node ( state: PEVState ): 

    """Проверяет последний результат работы инструмента на наличие ошибок."""

     console. print ( "--- ПРОВЕРКА: Проверка результата последнего инструмента... ---" ) 

    verifier_llm = llm.with_structured_output(VerificationResult) 

    prompt = f"Проверьте, является ли следующий вывод инструмента успешным, допустимым результатом или сообщением об ошибке. Задача была ' {state[ 'user_request' ]} '.\n\nВывод инструмента: ' {state[ 'last_tool_result' ]} '"

     verification = verifier_llm.invoke(prompt) 

    console. print ( f"--- ПРОВЕРКА: Результат проверки: ' { 'Успех' , если verification.is_successful , иначе 'Неудача' } ' ---" ) 

    if verification.is_successful: 

        return { "intermediate_steps" : state[ "intermediate_steps" ] + [state[ 'last_tool_result' ]]} 

    else : 

        # Если проверка не удалась, добавляем причину сбоя и очищаем план, чтобы запустить перепланирование 

        return { "plan" : [], "intermediate_steps" : state[ "intermediate_steps" ] + [ f"Проверка не удалась: {state[ 'last_tool_result' ]} " ]}

Подготовив верификатор, мы можем подключить весь граф. Ключевым моментом здесь является логика маршрутизатора. После выполнения заданий verifier_node, если план внезапно оказывается пустым (потому что верификатор его очистил), наш маршрутизатор знает, что нужно отправить агента обратно, чтобы planner_nodeповторить попытку.

 

class PEVState ( TypedDict ): 

    user_request: str

     plan: Optional [ List [ str ]] 

    last_tool_result: Optional [ str ] 

    intermediate_steps: List [ str ] 

    final_answer: Optional [ str ] 

    retries: int 

 

# ... (определения planner_node, executor_node и synthesizer_node аналогичны предыдущим) ... 

def pev_router ( state: PEVState ): 

    """Выполнение маршрутов на основе проверки и статуса плана.""" 

    if not state[ "plan" ]: 

        # Проверяем, пуст ли план, потому что проверка не удалась 

        if state[ "intermediate_steps" ] and "Verification Failed" in state[ "intermediate_steps" ][- 1 ]: 

            console. print ( "--- МАРШРУТИЗАТОР: Проверка не удалась. Перепланирование... ---" ) 

            return "plan" 

        else : 

            console. print ( "--- МАРШРУТИЗАТОР: Планирование завершено. Переходим к синтезатору. ---" ) 

            return "synthesize" 

    else : 

        console. print ( "--- МАРШРУТИЗАТОР: План содержит еще шаги. Продолжение выполнения. ---" ) 

        return "execute"

 

 pev_graph_builder = StateGraph(PEVState) 

 

pev_graph_builder.add_node( "plan" , pev_planner_node) 

pev_graph_builder.add_node( "execute" , pev_executor_node) 

pev_graph_builder.add_node( "verify" , verifier_node) 

pev_graph_builder.add_node( "synthesize" , synthesizer_node) 

 

pev_graph_builder.set_entry_point( "plan" ) 

pev_graph_builder.add_edge( "plan" , "execute" ) 

pev_graph_builder.add_edge( "execute" , "verify" ) 

pev_graph_builder.add_conditional_edges( "verify" , pev_router) 

 

pev_graph_builder.add_edge( "synthesize" , END) 

pev_agent_app = pev_graph_builder.компилировать ()

 

 

 

 

Архитектура PEV (создано компанией...)

Фарид Хан

)

Теперь перейдём к решающему тесту. Мы дадим нашему агенту PEV задачу, которая потребует от него вызова нашего flaky_web_searchинструмента с запросом, который, как мы знаем, завершится ошибкой. Простой агент Planner-Executor здесь бы сломался.

 

flaky_query = "Каковы были расходы Apple на НИОКР в прошлом финансовом году и каково было общее количество сотрудников? Рассчитайте расходы на НИОКР на одного сотрудника."

 

 console. print ( f"[bold green]Тестирование агента PEV на нестабильном запросе:[/bold green]\n' {flaky_query} '\n" ) 

initial_pev_input = { "user_request" : flaky_query, "intermediate_steps" : [], "retries" : 0 } 

 

final_pev_output = pev_agent_app.invoke(initial_pev_input) 

 

console. print ( "\n--- [bold green]Итоговый вывод от агента PEV[/bold green] ---" ) 

console. print (Markdown(final_pev_output[ 'final_answer' ]))

Тестирование агента PEV на том же нестабильном запросе: 

«Каковы были расходы Apple на НИОКР в прошлом финансовом году и какова была общая численность сотрудников? Рассчитайте 

расходы на НИОКР на одного сотрудника». 

 

--- (PEV) ПЛАНИРОВЩИК: Создание/редактирование плана (повторов 0)... --- 

--- (PEV) ИСПОЛНИТЕЛЬ: Выполнение следующего шага... --- 

--- ИНСТРУМЕНТ: Поиск «расходы Apple на НИОКР в прошлом финансовом году»... --- 

--- ПРОВЕРЯЮЩИЙ: Проверка последнего результата работы инструмента... --- 

--- ПРОВЕРЯЮЩИЙ: Оценка «Успех» --- 

 

...

План 1: Агент составляет план: ["Apple R&D spend...", "Apple total employee count"].

Выполнение и сбой: Проект получает финансирование на НИОКР, но employee countпоиск обращается к нашему нестабильному инструменту и возвращает ошибку.

Проверка и перехват: система verifier_nodeполучает сообщение об ошибке, корректно определяет её как сбой и сбрасывает план.

Маршрутизатор и перепланирование: Маршрутизатор видит пустой план и сообщение об ошибке и отправляет агента обратно на planner_node.

План 2: Планировщик, теперь понимая, что расчет общего числа сотрудников Apple провалился, создает новый, более продуманный план, возможно, пытаясь web_search('Apple number of employees worldwide')...

Выполнение и успех: Этот новый план работает, и агент получает все необходимые данные.

В итоге получился правильный результат вычислений. Агент не просто сдался; он обнаружил проблему, придумал новый подход и добился успеха.

 

Для формализации этого вопроса нашему магистру права в качестве судьи необходимо оценить его надежность.

 

class RobustnessEvaluation ( BaseModel ): 

    """Схема для оценки устойчивости агента и обработки ошибок."""

     task_completion_score: int = Field(description= "Оценка от 1 до 10 за выполнение задачи." ) 

    error_handling_score: int = Field(description= "Оценка от 1 до 10 за способность агента обнаруживать ошибки и восстанавливаться после них." ) 

    justification: str = Field(description= "Краткое обоснование оценок." )

Стандартный агент «Планировщик-Исполнитель» получил бы ужасный результат error_handling_score. Наш же агент PEV, напротив, превосходен.

 

--- Оценка устойчивости агента PEV ---

 { 

    'task_completion_score' : 8 , 

    'error_handling_score' : 10 , 

    'justification' : "Агент продемонстрировал идеальную устойчивость. Он успешно выявил сбой инструмента с помощью своего верификатора, запустил цикл перепланирования и сформулировал новый запрос для обхода проблемы. Это образцовый пример восстановления после ошибки."

 }

Как видите, архитектура PEV — это не просто получение правильного ответа, когда всё идёт хорошо…

 

Речь идёт о том, чтобы не дать неправильный ответ, когда что-то пойдёт не так.